fix: preserve meta labels

Author: kalleep

internal/component/loki/source/kubernetes/kubernetes.go

@@ -200,7 +200,7 @@ func (c *Component) resyncTargets(targets []discovery.Target) {
 			level.Error(c.log).Log("msg", "failed to process input target", "target", lset.String(), "err", err)
 			continue
 		}
-		tailTargets = append(tailTargets, kubetail.NewTarget(lset, processed))
+		tailTargets = append(tailTargets, kubetail.NewTarget(lset, processed, false))
 	}
 
 	// This will never fail because it only fails if the context gets canceled.

internal/component/loki/source/kubernetes/kubetail/tailer_test.go

@@ -90,17 +90,17 @@ func Test_parseKubernetesLog(t *testing.T) {
 	}
 }
 
-func Test_processLogStream_duplicateTimestamps(t *testing.T) {
+func Test_processLogStream(t *testing.T) {
 	baseTime := time.Date(2023, time.January, 23, 17, 0, 10, 0, time.UTC)
 
 	tt := []struct {
-		name         string
-		logLines     []string
-		lastReadTime time.Time
-		expectLines  []string
+		name               string
+		preserveMetaLabels bool
+		logLines           []string
+		lastReadTime       time.Time
+		expectLines        []string
 	}{
-		{
-			name: "duplicate timestamps are not discarded",
+		{name: "duplicate timestamps are not discarded",
 			logLines: []string{
 				"2023-01-23T17:00:10Z line1\n",
 				"2023-01-23T17:00:10Z line2\n",
@@ -150,22 +150,27 @@ func Test_processLogStream_duplicateTimestamps(t *testing.T) {
 				"2023-01-23T17:00:10Z line3\n",
 				"2023-01-23T17:00:10Z line4\n",
 			},
-			lastReadTime: baseTime,
-			expectLines:  []string{"line1\n", "line2\n", "line3\n", "line4\n"},
+			lastReadTime:       baseTime,
+			expectLines:        []string{"line1\n", "line2\n", "line3\n", "line4\n"},
+			preserveMetaLabels: true,
 		},
 	}
 
 	for _, tc := range tt {
 		t.Run(tc.name, func(t *testing.T) {
 			// Create a mock tailer with minimal setup
 			lset := labels.FromStrings(
-				LabelPodNamespace, "default",
-				LabelPodName, "test-pod",
-				LabelPodContainerName, "test-container",
-				LabelPodUID, "test-uid-123",
+				kubePodName, "test-pod",
+				kubePodNamespace, "default",
+				kubePodContainerName, "test-container",
+				kubePodUID, "test-uid-123",
 				"test", "value",
 			)
-			target := NewTarget(lset, lset)
+
+			lset, err := PrepareLabelsWithMetaPreservation(lset, "test", tc.preserveMetaLabels)
+			require.NoError(t, err)
+
+			target := NewTarget(lset, lset, tc.preserveMetaLabels)
 			opts := &Options{
 				Positions: &mockPositions{},
 			}
@@ -215,6 +220,22 @@ func Test_processLogStream_duplicateTimestamps(t *testing.T) {
 			}
 
 			require.Equal(t, tc.expectLines, receivedLines, "received lines should match expected lines")
+
+			if tc.preserveMetaLabels {
+				lbls := target.Labels()
+				require.Equal(t, "test", lbls.Get("job"))
+				require.Equal(t, "value", lbls.Get("test"))
+				require.Equal(t, "default/test-pod:test-container", lbls.Get("instance"))
+				require.Equal(t, "test-pod", lbls.Get(kubePodName))
+				require.Equal(t, "default", lbls.Get(kubePodNamespace))
+				require.Equal(t, "test-container", lbls.Get(kubePodContainerName))
+				require.Equal(t, "test-uid-123", lbls.Get(kubePodUID))
+			} else {
+				lbls := target.Labels()
+				require.Equal(t, "test", lbls.Get("job"))
+				require.Equal(t, "value", lbls.Get("test"))
+				require.Equal(t, "default/test-pod:test-container", lbls.Get("instance"))
+			}
 		})
 	}
 }

internal/component/loki/source/kubernetes/kubetail/target.go

@@ -57,7 +57,7 @@ type Target struct {
 }
 
 // NewTarget creates a new Target which can be passed to a tailer.
-func NewTarget(origLabels labels.Labels, lset labels.Labels) *Target {
+func NewTarget(origLabels labels.Labels, lset labels.Labels, preserveMetaLabels bool) *Target {
 	// Precompute some values based on labels so we don't have to continually
 	// search them.
 	var (
@@ -70,7 +70,7 @@ func NewTarget(origLabels labels.Labels, lset labels.Labels) *Target {
 		uid           = lset.Get(LabelPodUID)
 
 		id           = fmt.Sprintf("%s:%s", namespacedName, containerName)
-		publicLabels = publicLabels(lset)
+		publicLabels = publicLabels(lset, preserveMetaLabels)
 	)
 
 	// Precompute the hash of the target from the public labels and the ID of the
@@ -93,10 +93,14 @@ func NewTarget(origLabels labels.Labels, lset labels.Labels) *Target {
 	}
 }
 
-func publicLabels(lset labels.Labels) labels.Labels {
+func publicLabels(lset labels.Labels, preserveMetaLabels bool) labels.Labels {
 	lb := labels.NewBuilder(lset)
 
 	lset.Range(func(l labels.Label) {
+		if preserveMetaLabels && strings.HasPrefix(l.Name, model.MetaLabelPrefix) {
+			return
+		}
+
 		if strings.HasPrefix(l.Name, model.ReservedLabelPrefix) {
 			lb.Del(l.Name)
 		}

internal/component/loki/source/podlogs/reconciler.go

@@ -351,7 +351,7 @@ func (r *reconciler) reconcilePodLogs(ctx context.Context, cli client.Client, po
 				return
 			}
 
-			target := kubetail.NewTarget(targetLabels.Copy(), finalLabels)
+			target := kubetail.NewTarget(targetLabels.Copy(), finalLabels, preserveMetaLabels)
 			if processedLabels.Len() != 0 {
 				targets = append(targets, target)
 			}