[exporter/loki] Use record attributes as log labels

CHANGELOG.md

@@ -5,6 +5,7 @@
 ## 💡 Enhancements 💡
 
 - `hostreceiver/memoryscraper`: Migrate the scraper to the mdatagen metrics builder (#7312)
+- `lokiexporter`: Use record attributes as log labels (#7569)
 - `routingprocessor`: Do not err on failure to build exporters (#7423)
 - `apachereceiver`: Update to mdatagen v2 (#7573)
 - `datadogexporter`: Don't send host metadata if hostname is empty (#7426)

exporter/lokiexporter/README.md

@@ -21,6 +21,11 @@ The following settings are required:
   [Loki label best practices](https://grafana.com/docs/loki/latest/best-practices/) page for 
   additional details on the types of labels you may want to associate with log streams.
 
+- `labels.record` (no default): A map of record attributes to valid Loki label names (must match 
+  "^[a-zA-Z_][a-zA-Z0-9_]*$") allowed to be added as labels to Loki log streams.
+  Record attributes can be: `traceID`, `spanID`, `severity`, `severityN`. These attributes will be added as log labels 
+  and will be removed from the log body.
+
 The following settings can be optionally configured:
 
 - `tenant_id` (no default): The tenant ID used to identify the tenant the logs are associated to. This will set the 
@@ -62,7 +67,10 @@ loki:
       # Allowing 'severity' attribute and not providing a mapping, since the attribute name is a valid Loki label name.
       severity: ""
       http.status_code: "http_status_code" 
-
+    record:
+      # Adds 'traceID' as a log label, seen as 'traceid' in Loki.
+      traceID: "traceid"
+
   headers:
     "X-Custom-Header": "loki_rocks"
 ```

exporter/lokiexporter/config.go

@@ -55,11 +55,15 @@ type LabelsConfig struct {
 
 	// ResourceAttributes are the resource attributes that are allowed to be added as labels on a log stream.
 	ResourceAttributes map[string]string `mapstructure:"resource"`
+
+	// RecordAttributes are the attributes from the record that are allowed to be added as labels on a log stream. Possible keys:
+	// traceID, spanID, severity, severityN.
+	RecordAttributes map[string]string `mapstructure:"record"`
 }
 
 func (c *LabelsConfig) validate() error {
-	if len(c.Attributes) == 0 && len(c.ResourceAttributes) == 0 {
-		return fmt.Errorf("\"labels.attributes\" or \"labels.resource\" must be configured with at least one attribute")
+	if len(c.Attributes) == 0 && len(c.ResourceAttributes) == 0 && len(c.RecordAttributes) == 0 {
+		return fmt.Errorf("\"labels.attributes\", \"labels.resource\", or \"labels.record\" must be configured with at least one attribute")
 	}
 
 	logRecordNameInvalidErr := "the label `%s` in \"labels.attributes\" is not a valid label name. Label names must match " + model.LabelNameRE.String()
@@ -80,6 +84,17 @@ func (c *LabelsConfig) validate() error {
 		}
 	}
 
+	possibleRecordAttributes := map[string]bool{
+		"traceID":   true,
+		"spanID":    true,
+		"severity":  true,
+		"severityN": true,
+	}
+	for k := range c.RecordAttributes {
+		if _, found := possibleRecordAttributes[k]; !found {
+			return fmt.Errorf("record attribute %q not recognized, possible values: traceID, spanID, severity, severityN", k)
+		}
+	}
 	return nil
 }
 

exporter/lokiexporter/config_test.go

@@ -86,6 +86,9 @@ func TestLoadConfig(t *testing.T) {
 				"resource.name": "resource_name",
 				"severity":      "severity",
 			},
+			RecordAttributes: map[string]string{
+				"traceID": "traceid",
+			},
 		},
 		Format: "body",
 	}
@@ -202,7 +205,7 @@ func TestConfig_validate(t *testing.T) {
 					ResourceAttributes: nil,
 				},
 			},
-			errorMessage: "\"labels.attributes\" or \"labels.resource\" must be configured with at least one attribute",
+			errorMessage: "\"labels.attributes\", \"labels.resource\", or \"labels.record\" must be configured with at least one attribute",
 			shouldError:  true,
 		},
 		{
@@ -221,6 +224,33 @@ func TestConfig_validate(t *testing.T) {
 			},
 			shouldError: false,
 		},
+		{
+			name: "with valid `labels.record`",
+			fields: fields{
+				Endpoint: validEndpoint,
+				Labels: LabelsConfig{
+					RecordAttributes: map[string]string{
+						"traceID":   "traceID",
+						"spanID":    "spanID",
+						"severity":  "severity",
+						"severityN": "severityN",
+					},
+				},
+			},
+			shouldError: false,
+		},
+		{
+			name: "with invalid `labels.record`",
+			fields: fields{
+				Endpoint: validEndpoint,
+				Labels: LabelsConfig{
+					RecordAttributes: map[string]string{
+						"invalid": "Invalid",
+					},
+				},
+			},
+			shouldError: true,
+		},
 	}
 
 	for _, tt := range tests {
@@ -260,7 +290,7 @@ func TestLabelsConfig_validate(t *testing.T) {
 				Attributes:         map[string]string{},
 				ResourceAttributes: map[string]string{},
 			},
-			errorMessage: "\"labels.attributes\" or \"labels.resource\" must be configured with at least one attribute",
+			errorMessage: "\"labels.attributes\", \"labels.resource\", or \"labels.record\" must be configured with at least one attribute",
 			shouldError:  true,
 		},
 		{

exporter/lokiexporter/exporter.go

@@ -158,6 +158,11 @@ func (l *lokiExporter) logDataToLoki(ld pdata.Logs) (pr *logproto.PushRequest, n
 					numDroppedLogs++
 					continue
 				}
+
+				// now merge the labels based on the record attributes
+				recordLabels := l.convertRecordAttributesToLabels(log)
+				mergedLabels = mergedLabels.Merge(recordLabels)
+
 				labels := mergedLabels.String()
 				var entry *logproto.Entry
 				var err error
@@ -240,25 +245,47 @@ func (l *lokiExporter) convertAttributesToLabels(attributes pdata.AttributeMap,
 	return ls
 }
 
+func (l *lokiExporter) convertRecordAttributesToLabels(log pdata.LogRecord) model.LabelSet {
+	ls := model.LabelSet{}
+
+	if val, ok := l.config.Labels.RecordAttributes["traceID"]; ok {
+		ls[model.LabelName(val)] = model.LabelValue(log.TraceID().HexString())
+	}
+
+	if val, ok := l.config.Labels.RecordAttributes["spanID"]; ok {
+		ls[model.LabelName(val)] = model.LabelValue(log.SpanID().HexString())
+	}
+
+	if val, ok := l.config.Labels.RecordAttributes["severity"]; ok {
+		ls[model.LabelName(val)] = model.LabelValue(log.SeverityText())
+	}
+
+	if val, ok := l.config.Labels.RecordAttributes["severityN"]; ok {
+		ls[model.LabelName(val)] = model.LabelValue(log.SeverityNumber().String())
+	}
+
+	return ls
+}
+
 func (l *lokiExporter) convertLogBodyToEntry(lr pdata.LogRecord, res pdata.Resource) (*logproto.Entry, error) {
 	var b strings.Builder
 
-	if len(lr.SeverityText()) > 0 {
+	if _, ok := l.config.Labels.RecordAttributes["severity"]; !ok && len(lr.SeverityText()) > 0 {
 		b.WriteString("severity=")
 		b.WriteString(lr.SeverityText())
 		b.WriteRune(' ')
 	}
-	if lr.SeverityNumber() > 0 {
+	if _, ok := l.config.Labels.RecordAttributes["severityN"]; !ok && lr.SeverityNumber() > 0 {
 		b.WriteString("severityN=")
 		b.WriteString(strconv.Itoa(int(lr.SeverityNumber())))
 		b.WriteRune(' ')
 	}
-	if !lr.TraceID().IsEmpty() {
+	if _, ok := l.config.Labels.RecordAttributes["traceID"]; !ok && !lr.TraceID().IsEmpty() {
 		b.WriteString("traceID=")
 		b.WriteString(lr.TraceID().HexString())
 		b.WriteRune(' ')
 	}
-	if !lr.SpanID().IsEmpty() {
+	if _, ok := l.config.Labels.RecordAttributes["spanID"]; !ok && !lr.SpanID().IsEmpty() {
 		b.WriteString("spanID=")
 		b.WriteString(lr.SpanID().HexString())
 		b.WriteRune(' ')

exporter/lokiexporter/exporter_test.go

@@ -610,3 +610,79 @@ func TestExporter_convertLogtoJSONEntry(t *testing.T) {
 	require.NotNil(t, entry)
 	require.Equal(t, expEntry, entry)
 }
+
+func TestConvertRecordAttributesToLabels(t *testing.T) {
+	testCases := []struct {
+		desc     string
+		lr       pdata.LogRecord
+		expected model.LabelSet
+	}{
+		{
+			desc: "traceID",
+			lr: func() pdata.LogRecord {
+				lr := pdata.NewLogRecord()
+				lr.SetTraceID(pdata.NewTraceID([16]byte{1, 2, 3, 4}))
+				return lr
+			}(),
+			expected: func() model.LabelSet {
+				ls := model.LabelSet{}
+				ls[model.LabelName("traceID")] = model.LabelValue("01020304000000000000000000000000")
+				return ls
+			}(),
+		},
+		{
+			desc: "spanID",
+			lr: func() pdata.LogRecord {
+				lr := pdata.NewLogRecord()
+				lr.SetSpanID(pdata.NewSpanID([8]byte{1, 2, 3, 4}))
+				return lr
+			}(),
+			expected: func() model.LabelSet {
+				ls := model.LabelSet{}
+				ls[model.LabelName("spanID")] = model.LabelValue("0102030400000000")
+				return ls
+			}(),
+		},
+		{
+			desc: "severity",
+			lr: func() pdata.LogRecord {
+				lr := pdata.NewLogRecord()
+				lr.SetSeverityText("DEBUG")
+				return lr
+			}(),
+			expected: func() model.LabelSet {
+				ls := model.LabelSet{}
+				ls[model.LabelName("severity")] = model.LabelValue("DEBUG")
+				return ls
+			}(),
+		},
+		{
+			desc: "severityN",
+			lr: func() pdata.LogRecord {
+				lr := pdata.NewLogRecord()
+				lr.SetSeverityNumber(pdata.SeverityNumberDEBUG)
+				return lr
+			}(),
+			expected: func() model.LabelSet {
+				ls := model.LabelSet{}
+				ls[model.LabelName("severityN")] = model.LabelValue(pdata.SeverityNumberDEBUG.String())
+				return ls
+			}(),
+		},
+	}
+	for _, tC := range testCases {
+		t.Run(tC.desc, func(t *testing.T) {
+			exp := newExporter(&Config{
+				Labels: LabelsConfig{
+					RecordAttributes: map[string]string{
+						tC.desc: tC.desc,
+					},
+				},
+			}, componenttest.NewNopTelemetrySettings())
+
+			ls := exp.convertRecordAttributesToLabels(tC.lr)
+
+			assert.Equal(t, tC.expected, ls)
+		})
+	}
+}

exporter/lokiexporter/testdata/config.yaml

@@ -40,6 +40,8 @@ exporters:
     headers:
       "X-Custom-Header": "loki_rocks"
     labels:
+      record:
+        traceID: traceid
       attributes:
         container.name: "container_name"
         k8s.cluster.name: "k8s_cluster_name"