fix(promtail): Fix API level node filtering of Kubernetes SD to only fetch pod metadata for pods of the current node (#14267)

Co-authored-by: Christian Haudum <christian.haudum@gmail.com>

Author: LashaJini

clients/pkg/promtail/targets/file/filetargetmanager.go

@@ -264,13 +264,13 @@ func (tm *FileTargetManager) fulfillKubePodSelector(selectors []kubernetes.Selec
 		return []kubernetes.SelectorConfig{{Role: kubernetes.RolePod, Field: nodeSelector}}
 	}
 
-	for _, selector := range selectors {
-		if selector.Field == "" {
-			selector.Field = nodeSelector
-		} else if !strings.Contains(selector.Field, nodeSelector) {
-			selector.Field += "," + nodeSelector
+	for i := range selectors {
+		if selectors[i].Field == "" {
+			selectors[i].Field = nodeSelector
+		} else if !strings.Contains(selectors[i].Field, nodeSelector) {
+			selectors[i].Field += "," + nodeSelector
 		}
-		selector.Role = kubernetes.RolePod
+		selectors[i].Role = kubernetes.RolePod
 	}
 
 	return selectors

clients/pkg/promtail/targets/file/filetargetmanager_test.go

@@ -14,6 +14,7 @@ import (
 	"github.com/go-kit/log"
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/discovery"
+	"github.com/prometheus/prometheus/discovery/kubernetes"
 	"github.com/prometheus/prometheus/discovery/targetgroup"
 
 	"github.com/grafana/loki/v3/clients/pkg/promtail/api"
@@ -594,3 +595,70 @@ func TestLabelSetUpdate(t *testing.T) {
 	require.Equal(t, 0, len(syncer.fileEventWatchers))
 
 }
+
+func TestFulfillKubePodSelector(t *testing.T) {
+	w := log.NewSyncWriter(os.Stderr)
+	logger := log.NewLogfmtLogger(w)
+	logDirName := newTestLogDirectories(t)
+
+	positionsFileName := filepath.Join(logDirName, "positions.yml")
+	ps, err := newTestPositions(logger, positionsFileName)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	client := fake.New(func() {})
+	defer client.Stop()
+
+	ftm, err := newTestFileTargetManager(logger, client, ps, logDirName+"/*")
+	assert.NoError(t, err)
+
+	host := "test-host"
+
+	// empty selectors
+	selectors := []kubernetes.SelectorConfig{}
+	expected := []kubernetes.SelectorConfig{
+		{Role: kubernetes.RolePod, Field: fmt.Sprintf("%s=%s", kubernetesPodNodeField, host)},
+	}
+
+	result := ftm.fulfillKubePodSelector(selectors, host)
+	require.Equal(t, expected, result)
+
+	// non-empty selectors with empty field
+	selectors = []kubernetes.SelectorConfig{
+		{Role: kubernetes.RolePod, Field: ""},
+	}
+	expected = []kubernetes.SelectorConfig{
+		{Role: kubernetes.RolePod, Field: fmt.Sprintf("%s=%s", kubernetesPodNodeField, host)},
+	}
+
+	result = ftm.fulfillKubePodSelector(selectors, host)
+	require.Equal(t, expected, result)
+
+	// non-empty selectors with existing field without nodeSelector
+	selectors = []kubernetes.SelectorConfig{
+		{Role: kubernetes.RolePod, Field: "app=frontend"},
+	}
+	expectedField := "app=frontend," + fmt.Sprintf("%s=%s", kubernetesPodNodeField, host)
+	expected = []kubernetes.SelectorConfig{
+		{Role: kubernetes.RolePod, Field: expectedField},
+	}
+
+	result = ftm.fulfillKubePodSelector(selectors, host)
+	require.Equal(t, expected, result)
+
+	// non-empty selectors with existing Field containing nodeSelector
+	nodeSelector := fmt.Sprintf("%s=%s", kubernetesPodNodeField, host)
+	selectors = []kubernetes.SelectorConfig{
+		{Role: kubernetes.RolePod, Field: nodeSelector},
+	}
+	expected = []kubernetes.SelectorConfig{
+		{Role: kubernetes.RolePod, Field: nodeSelector},
+	}
+
+	result = ftm.fulfillKubePodSelector(selectors, host)
+	require.Equal(t, expected, result)
+
+	ftm.Stop()
+	ps.Stop()
+}