Loki Write: Internal Labels

This change is to add logic so that labels that start with `__` are
filtered out, unless they are `externalLabels`. Labels that start with
`__` are used internally for things such as metadata. If a user want to
preserve the contents of one of these internal labels, then can do so
using relabelling to remove the `__`.

This change also adds tests which check various different scenarios

Author: matt-gp

CHANGELOG.md

@@ -53,6 +53,8 @@ Main (unreleased)
 
 - Only log EOF errors for syslog port investigations in `loki.source.syslog` as Debug, not Warn. (@dehaansa)
 
+- Remove entry labels in `loki.write` that start with `__` used for metadata.  @matt-gp
+
 v1.11.2
 -----------------
 

internal/component/loki/write/write.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"path/filepath"
+	"strings"
 	"sync"
 	"time"
 
@@ -228,10 +229,19 @@ func (c *Component) Update(args component.Arguments) error {
 
 func newEntryHandler(handler loki.EntryHandler, externalLabels model.LabelSet) loki.EntryHandler {
 	return loki.NewEntryMutatorHandler(handler, func(e loki.Entry) loki.Entry {
-		if len(externalLabels) == 0 {
-			return e
+		// Merge external labels with entry labels first (entry labels take precedence)
+		mergedLabels := externalLabels.Merge(e.Labels)
+
+		// Filter out labels starting with "__" in a single pass
+		filteredLabels := make(model.LabelSet)
+		for k, v := range mergedLabels {
+			labelName := string(k)
+			if !strings.HasPrefix(labelName, "__") {
+				filteredLabels[k] = v
+			}
 		}
-		e.Labels = externalLabels.Merge(e.Labels)
+
+		e.Labels = filteredLabels
 		return e
 	})
 }

internal/component/loki/write/write_test.go

@@ -408,3 +408,189 @@ func benchSingleEndpoint(b *testing.B, tc testCase, alterConfig func(arguments *
 		}, time.Minute, time.Second, "haven't seen expected number of lines")
 	}
 }
+
+func TestNewEntryHandler(t *testing.T) {
+	testCases := []struct {
+		name           string
+		externalLabels model.LabelSet
+		entryLabels    model.LabelSet
+		expectedLabels model.LabelSet
+	}{
+		{
+			name:           "filter meta labels without external labels",
+			externalLabels: nil,
+			entryLabels: model.LabelSet{
+				"normal_label":   "value1",
+				"__meta_label":   "value2",
+				"__another_meta": "value3",
+				"regular_label":  "value4",
+			},
+			expectedLabels: model.LabelSet{
+				"normal_label":  "value1",
+				"regular_label": "value4",
+			},
+		},
+		{
+			name: "filter meta labels from both external and entry labels",
+			externalLabels: model.LabelSet{
+				"__external_meta": "external_value", // should be filtered
+				"external_label":  "external_normal",
+			},
+			entryLabels: model.LabelSet{
+				"normal_label":  "value1",
+				"__meta_label":  "value2", // should be filtered
+				"regular_label": "value4",
+			},
+			expectedLabels: model.LabelSet{
+				"external_label": "external_normal", // external normal label preserved
+				"normal_label":   "value1",          // entry normal label preserved
+				"regular_label":  "value4",          // entry normal label preserved
+			},
+		},
+		{
+			name: "entry labels override external labels",
+			externalLabels: model.LabelSet{
+				"common_label": "external_value",
+			},
+			entryLabels: model.LabelSet{
+				"common_label": "entry_value", // should override external
+				"entry_only":   "value1",
+			},
+			expectedLabels: model.LabelSet{
+				"common_label": "entry_value", // entry takes precedence
+				"entry_only":   "value1",
+			},
+		},
+		{
+			name:           "empty labels",
+			externalLabels: nil,
+			entryLabels:    model.LabelSet{},
+			expectedLabels: model.LabelSet{},
+		},
+		{
+			name:           "only meta labels",
+			externalLabels: nil,
+			entryLabels: model.LabelSet{
+				"__meta_only":    "value1",
+				"__another_meta": "value2",
+			},
+			expectedLabels: model.LabelSet{},
+		},
+		{
+			name: "external meta labels with no entry labels",
+			externalLabels: model.LabelSet{
+				"__external_meta": "external_value", // should now be filtered
+				"external_normal": "normal_value",
+			},
+			entryLabels: model.LabelSet{},
+			expectedLabels: model.LabelSet{
+				"external_normal": "normal_value", // only normal external label preserved
+			},
+		},
+		{
+			name: "mixed scenario with overrides and filtering",
+			externalLabels: model.LabelSet{
+				"__external_meta": "external_value", // should now be filtered
+				"shared_label":    "external_shared",
+			},
+			entryLabels: model.LabelSet{
+				"normal_label":   "normal_value",
+				"__meta_label":   "meta_value",   // should be filtered
+				"shared_label":   "entry_shared", // should override external
+				"another_normal": "another_value",
+			},
+			expectedLabels: model.LabelSet{
+				"shared_label":   "entry_shared",  // entry takes precedence (original behavior)
+				"normal_label":   "normal_value",  // entry normal preserved
+				"another_normal": "another_value", // entry normal preserved
+			},
+		},
+		{
+			name:           "labels starting with single underscore should be preserved",
+			externalLabels: nil,
+			entryLabels: model.LabelSet{
+				"_single_underscore":  "value1",
+				"__double_underscore": "value2", // should be filtered
+				"normal_label":        "value3",
+			},
+			expectedLabels: model.LabelSet{
+				"_single_underscore": "value1",
+				"normal_label":       "value3",
+			},
+		},
+		{
+			name: "external labels with empty values",
+			externalLabels: model.LabelSet{
+				"empty_external":  "",
+				"normal_external": "external_value",
+			},
+			entryLabels: model.LabelSet{
+				"entry_label": "entry_value",
+			},
+			expectedLabels: model.LabelSet{
+				"empty_external":  "",
+				"normal_external": "external_value",
+				"entry_label":     "entry_value",
+			},
+		},
+		{
+			name: "only external labels",
+			externalLabels: model.LabelSet{
+				"external_only":   "external_value",
+				"__external_meta": "meta_value", // should now be filtered
+			},
+			entryLabels: nil,
+			expectedLabels: model.LabelSet{
+				"external_only": "external_value", // only normal external label preserved
+			},
+		},
+		{
+			name: "all meta labels filtered regardless of source",
+			externalLabels: model.LabelSet{
+				"__external_meta1": "external_value1", // should be filtered
+				"__external_meta2": "external_value2", // should be filtered
+				"external_normal":  "external_normal_value",
+			},
+			entryLabels: model.LabelSet{
+				"__entry_meta1": "entry_value1", // should be filtered
+				"__entry_meta2": "entry_value2", // should be filtered
+				"entry_normal":  "entry_normal_value",
+			},
+			expectedLabels: model.LabelSet{
+				"external_normal": "external_normal_value",
+				"entry_normal":    "entry_normal_value",
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			// Create a channel to capture the entry
+			captureChannel := make(chan loki.Entry, 1)
+			mockHandler := loki.NewEntryHandler(captureChannel, func() {})
+
+			handler := newEntryHandler(mockHandler, tc.externalLabels)
+
+			testEntry := loki.Entry{
+				Labels: tc.entryLabels,
+				Entry: push.Entry{
+					Timestamp: time.Now(),
+					Line:      "test log line",
+				},
+			}
+
+			go func() {
+				handler.Chan() <- testEntry
+			}()
+
+			select {
+			case capturedEntry := <-captureChannel:
+				require.Equal(t, tc.expectedLabels, capturedEntry.Labels)
+			case <-time.After(time.Second):
+				t.Fatalf("timeout waiting for entry in test case: %s", tc.name)
+			}
+
+			handler.Stop()
+		})
+	}
+}