20250117 implement label join

internal/component/pyroscope/receive_http/ip_match.go

@@ -0,0 +1,70 @@
+package receive_http
+
+import (
+	"net/netip"
+
+	"github.com/go-kit/log"
+
+	"github.com/grafana/alloy/internal/component/discovery"
+	"github.com/grafana/alloy/internal/runtime/logging/level"
+)
+
+const (
+	labelMetaDockerNetworkIP = "__meta_docker_network_ip"
+	labelMetaKubernetesPodIP = "__meta_kubernetes_pod_ip"
+)
+
+// buildIPLookupMap builds a map of IP addresses to discovery targets. When there are targets with the same IP address, only labels that have the same value will be kept.
+func buildIPLookupMap(logger log.Logger, targets []discovery.Target) map[netip.Addr]discovery.Target {
+	result := make(map[netip.Addr]discovery.Target)
+	for _, t := range targets {
+		var addr netip.Addr
+		for _, key := range []string{labelMetaDockerNetworkIP, labelMetaKubernetesPodIP} {
+			ip, ok := t[key]
+			if !ok {
+				continue
+			}
+
+			if a, err := netip.ParseAddr(ip); err != nil {
+				level.Warn(logger).Log("msg", "Unable to parse IP address", "ip", ip)
+				continue
+			} else {
+				addr = a
+			}
+		}
+
+		if !addr.IsValid() {
+			continue
+		}
+
+		// add the discovery target into the resultkey
+		for k, v := range t {
+			if _, ok := result[addr]; !ok {
+				result[addr] = make(discovery.Target)
+			}
+
+			// check if the label already exists, if not add it and exit
+			vExisting, ok := result[addr][k]
+			if !ok {
+				result[addr][k] = v
+				continue
+			}
+
+			// check if existing element is matching, if not set it to the empty string
+			if vExisting != v {
+				result[addr][k] = ""
+			}
+		}
+	}
+
+	// go through all targets again and remove empty value labels
+	for keyIP := range result {
+		for name, value := range result[keyIP] {
+			if value == "" {
+				delete(result[keyIP], name)
+			}
+		}
+	}
+
+	return result
+}

internal/component/pyroscope/receive_http/ip_match_test.go

@@ -0,0 +1,93 @@
+package receive_http
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/grafana/alloy/internal/component/discovery"
+	"github.com/grafana/alloy/internal/util"
+)
+
+func toJson(t testing.TB, o any) string {
+	t.Helper()
+	x, err := json.Marshal(o)
+	require.NoError(t, err)
+	return string(x)
+}
+
+func TestBuildIPLookupMap(t *testing.T) {
+	logger := util.TestAlloyLogger(t)
+
+	for _, tc := range []struct {
+		name    string
+		targets []discovery.Target
+		result  string
+	}{
+		{name: "empty targets"},
+		{
+			name: "valid ips, no overlap",
+			targets: []discovery.Target{
+				map[string]string{
+					labelMetaDockerNetworkIP: "1.2.3.4",
+					"my-label":               "value",
+					"my-label2":              "value2",
+				},
+				map[string]string{
+					labelMetaKubernetesPodIP: "1.2.3.5",
+					"my-pod":                 "pod2",
+					"my-namespace":           "namespace2",
+				},
+			},
+			result: `{
+            "1.2.3.4":{"` + labelMetaDockerNetworkIP + `":"1.2.3.4","my-label":"value","my-label2":"value2"}, 
+            "1.2.3.5":{"` + labelMetaKubernetesPodIP + `":"1.2.3.5","my-pod":"pod2","my-namespace":"namespace2"}
+            }`,
+		},
+		{
+			name: "valid overlapping ips, pod label overlaps",
+			targets: []discovery.Target{
+				map[string]string{
+					labelMetaKubernetesPodIP: "1.2.3.4",
+					"my-pod":                 "pod1",
+					"my-namespace":           "namespace1",
+				},
+				map[string]string{
+					labelMetaKubernetesPodIP: "1.2.3.4",
+					"my-pod":                 "pod2",
+					"my-namespace":           "namespace1",
+				},
+			},
+			result: `{
+            "1.2.3.4":{"` + labelMetaKubernetesPodIP + `":"1.2.3.4","my-namespace":"namespace1"}
+            }`,
+		},
+		{
+			name: "valid overlapping ipv6s, pod label overlaps",
+			targets: []discovery.Target{
+				map[string]string{
+					labelMetaKubernetesPodIP: "cafe::",
+					"my-pod":                 "pod1",
+					"my-namespace":           "namespace1",
+				},
+				map[string]string{
+					labelMetaKubernetesPodIP: "cafe::0", // note: string is not overlapping
+					"my-pod":                 "pod2",
+					"my-namespace":           "namespace1",
+				},
+			},
+			result: `{
+            "cafe::":{"my-namespace":"namespace1"}
+            }`,
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			if tc.result == "" {
+				tc.result = "{}"
+			}
+			got := buildIPLookupMap(logger, tc.targets)
+			require.JSONEq(t, tc.result, toJson(t, got))
+		})
+	}
+}