feat(loki.enrich): Add multi-label matching

docs/sources/reference/components/loki/loki.enrich.md

@@ -15,25 +15,33 @@ description: The loki.enrich component enriches logs with labels from service di
 {{< docs/shared lookup="stability/experimental.md" source="alloy" version="<ALLOY_VERSION>" >}}
 
 The `loki.enrich` component enriches logs with additional labels from service discovery targets.
-It matches a label from incoming logs against a label from discovered targets, and copies specified labels from the matched target to the log entry.
+It matches labels from incoming logs against labels from discovered targets, and copies specified labels from the matched target to the log entry.
+If no match occurs, the component forwards the log entry unchanged.
+
+Use the `target_to_log_match` argument to specify which target labels correspond to which log labels.
+The map keys are target label names and the values are the corresponding log label names.
+All labels in the map must match for enrichment to occur.
+
+{{< admonition type="warning" >}}
+The `target_match_label` and `logs_match_label` arguments are deprecated in favor of `target_to_log_match`.
+If `target_to_log_match` is set, it takes precedence.
+Replace `target_match_label = "hostname"` with `target_to_log_match = {"hostname" = "hostname"}`.
+These deprecated arguments will be removed in a future release.
+{{< /admonition >}}
 
 ## Usage
 
 ```alloy
 loki.enrich "<LABEL>" {
-  // List of targets from a discovery component
   targets = <DISCOVERY_COMPONENT>.targets
-
-  // Which label from discovered targets to match against
-  target_match_label = "<LABEL>"
-
-  // Which label from incoming logs to match against
-  logs_match_label = "<LABEL>"
-
-  // List of labels to copy from discovered targets to logs
+
+  target_to_log_match = {
+    "<TARGET_LABEL_1>" = "<LOG_LABEL_1>",
+    "<TARGET_LABEL_2>" = "<LOG_LABEL_2>",
+  }
+
   labels_to_copy = ["<LABEL>", ...]
-
-  // Where to send enriched logs
+
   forward_to = [<RECEIVER_LIST>]
 }
 ```
@@ -42,15 +50,14 @@ loki.enrich "<LABEL>" {
 
 You can use the following arguments with `loki.enrich`:
 
-| Name                 | Type                  | Description                                                                                      | Default                | Required |
-| -------------------- | --------------------- | ------------------------------------------------------------------------------------------------ | ---------------------- | -------- |
-| `forward_to`         | `[]loki.LogsReceiver` | List of receivers to send enriched logs to.                                                      |                        | yes      |
-| `target_match_label` | `string`              | The label from discovered targets to match against, for example, `"__inventory_consul_service"`. |                        | yes      |
-| `targets`            | `[]discovery.Target`  | List of targets from a discovery component.                                                      |                        | yes      |
-| `labels_to_copy`     | `[]string`            | List of labels to copy from discovered targets to logs. If empty, all labels will be copied.     |                        | no       |
-| `logs_match_label`   | `string`              | The label from incoming logs to match against discovered targets, for example `"service_name"`.  |                        | no       |
-
-If not provided, the `logs_match_label` attribute will default to the value of `target_match_label`.
+| Name                  | Type                 | Description                                                                                                   | Default | Required |
+| --------------------- | -------------------- | ------------------------------------------------------------------------------------------------------------- | ------- | -------- |
+| `forward_to`          | `list(LogsReceiver)` | List of receivers to send log entries to.                                                                     |         | yes      |
+| `targets`             | `list(map(string))`  | List of targets from a discovery component.                                                                   |         | yes      |
+| `labels_to_copy`      | `list(string)`       | List of labels to copy from discovered targets to logs. If empty, all labels are copied.                      |         | no       |
+| `logs_match_label`    | `string`             | (Deprecated) The label from incoming logs to match against discovered targets, for example, `"service_name"`. |         | no       |
+| `target_match_label`  | `string`             | (Deprecated) The label from discovered targets to match against, for example, `"hostname"`.                   |         | no       |
+| `target_to_log_match` | `map(string)`        | Map of target label names to log label names. All entries must match for enrichment.                          |         | no       |
 
 ## Blocks
 
@@ -66,6 +73,9 @@ The following values are exported:
 
 ## Example
 
+The following example enriches syslog entries with labels from an HTTP service discovery target.
+It matches the connection IP address and tenant label before it forwards entries to Loki.
+
 ```alloy
 // Configure HTTP discovery
 discovery.http "default" {
@@ -101,15 +111,25 @@ discovery.relabel "default" {
     }
 }
 
+loki.relabel "syslog" {
+    rule {
+        source_labels = ["__syslog_connection_ip_address"]
+        target_label  = "source_ip"
+    }
+}
+
 // Receive syslog messages
 loki.source.syslog "incoming" {
     listener {
-        address = ":514"
+        address  = ":514"
         protocol = "tcp"
         labels = {
-            job = "syslog"
+            job    = "syslog"
+            tenant = "production"
         }
     }
+
+    relabel_rules = loki.relabel.syslog.rules
     forward_to = [loki.enrich.default.receiver]
 }
 
@@ -118,19 +138,27 @@ loki.enrich "default" {
     // Use targets from HTTP discovery (after relabeling)
     targets = discovery.relabel.default.output
 
-    // Match hostname from logs to DNS name
-    target_match_label = "primary_ip"
+    target_to_log_match = {
+        "primary_ip" = "source_ip"
+        "tenant"     = "tenant"
+    }
 
     forward_to = [loki.write.default.receiver]
 }
+
+loki.write "default" {
+    endpoint {
+        url = "http://loki:3100/loki/api/v1/push"
+    }
+}
 ```
 
-## Component Behavior
+## Component behavior
 
 The component matches logs to discovered targets and enriches them with additional labels:
 
-1. For each log entry, it looks up the value of `logs_match_label` from the log's labels or `target_match_label` if `logs_match_label` is not specified.
-1. It matches this value against the `target_match_label` in discovered targets.
+1. For each log entry, it looks up the log labels specified by `target_to_log_match`.
+1. It matches those values against the corresponding target labels. All label pairs must match the same target.
 1. If a match is found, it copies the requested `labels_to_copy` from the discovered target to the log entry. If `labels_to_copy` is empty, all labels are copied.
 1. The log entry, enriched or unchanged, is forwarded to the configured receivers.
 
@@ -158,4 +186,4 @@ Connecting some components may not be sensible or components may require further
 Refer to the linked documentation for more details.
 {{< /admonition >}}
 
-<!-- END GENERATED COMPATIBLE COMPONENTS -->
+<!-- END GENERATED COMPATIBLE COMPONENTS -->

internal/component/common/enrich/matcher.go

@@ -0,0 +1,88 @@
+// Package enrich provides shared target matching for enrichment components.
+package enrich
+
+import (
+	"sort"
+
+	"github.com/cespare/xxhash/v2"
+	"github.com/prometheus/common/model"
+
+	"github.com/grafana/alloy/internal/component/discovery"
+)
+
+// Matcher looks up discovery targets using matching label values. It is safe for
+// concurrent lookups. To change the targets or matching strategy, create a new Matcher.
+type Matcher struct {
+	labelNames []string
+	targets    map[uint64]model.LabelSet
+}
+
+// NewMatcher builds a lookup from targets. targetToLabel maps target label names
+// to incoming label names. Targets with a missing or empty matching label are
+// skipped. If multiple targets have the same matching values, the last one wins.
+func NewMatcher(targets []discovery.Target, targetToLabel map[string]string) *Matcher {
+	targetNames := make([]string, 0, len(targetToLabel))
+	for name := range targetToLabel {
+		targetNames = append(targetNames, name)
+	}
+	sort.Strings(targetNames)
+
+	m := &Matcher{
+		labelNames: make([]string, 0, len(targetNames)),
+		targets:    make(map[uint64]model.LabelSet),
+	}
+	for _, name := range targetNames {
+		m.labelNames = append(m.labelNames, targetToLabel[name])
+	}
+	for _, target := range targets {
+		h, ok := hashValues(func(name string) string {
+			value, _ := target.Get(name)
+			return value
+		}, targetNames)
+		if !ok {
+			continue
+		}
+		labelSet := make(model.LabelSet, target.Len())
+		target.ForEachLabel(func(name, value string) bool {
+			labelSet[model.LabelName(name)] = model.LabelValue(value)
+			return true
+		})
+		m.targets[h] = labelSet
+	}
+	return m
+}
+
+// Match returns the matching target's labels, or nil if no target matches.
+// get must return an empty string for missing labels. The returned label set
+// belongs to the Matcher and must not be modified.
+func (m *Matcher) Match(get func(string) string) model.LabelSet {
+	h, ok := hashValues(get, m.labelNames)
+	if !ok {
+		return nil
+	}
+	return m.targets[h]
+}
+
+// Len returns the number of distinct matching targets in the lookup.
+func (m *Matcher) Len() int {
+	return len(m.targets)
+}
+
+var sep = []byte{0xff} // separates UTF-8 label values to preserve value boundaries
+
+// hashValues returns false if no names are configured or any value is empty.
+func hashValues(get func(string) string, names []string) (uint64, bool) {
+	if len(names) == 0 {
+		return 0, false
+	}
+	h := xxhash.New()
+	for _, name := range names {
+		value := get(name)
+		if value == "" {
+			return 0, false
+		}
+		_, _ = h.WriteString(value)
+		_, _ = h.Write(sep)
+	}
+	return h.Sum64(), true
+}

internal/component/common/enrich/matcher_test.go

@@ -0,0 +1,158 @@
+package enrich_test
+
+import (
+	"testing"
+
+	"github.com/prometheus/common/model"
+	"github.com/prometheus/prometheus/model/labels"
+	"github.com/stretchr/testify/require"
+
+	"github.com/grafana/alloy/internal/component/common/enrich"
+	"github.com/grafana/alloy/internal/component/discovery"
+)
+
+func TestMatcher(t *testing.T) {
+	tests := []struct {
+		name     string
+		strategy map[string]string
+		targets  []model.LabelSet
+		input    model.LabelSet
+		want     model.LabelSet
+		wantLen  int
+	}{
+		{
+			name:     "single label",
+			strategy: map[string]string{"service": "service_name"},
+			targets:  []model.LabelSet{{"service": "api", "env": "prod"}},
+			input:    model.LabelSet{"service_name": "api"},
+			want:     model.LabelSet{"service": "api", "env": "prod"},
+			wantLen:  1,
+		},
+		{
+			name:     "all pairs match the same target regardless of label name order",
+			strategy: map[string]string{"namespace": "z_namespace", "pod": "a_pod"},
+			targets: []model.LabelSet{
+				{"namespace": "prod", "pod": "api", "env": "production"},
+				{"namespace": "stage", "pod": "api", "env": "staging"},
+			},
+			input:   model.LabelSet{"z_namespace": "stage", "a_pod": "api", "extra": "ignored"},
+			want:    model.LabelSet{"namespace": "stage", "pod": "api", "env": "staging"},
+			wantLen: 2,
+		},
+		{
+			name:     "pairs cannot match different targets",
+			strategy: map[string]string{"namespace": "namespace", "pod": "pod"},
+			targets: []model.LabelSet{
+				{"namespace": "prod", "pod": "api"},
+				{"namespace": "stage", "pod": "worker"},
+			},
+			input:   model.LabelSet{"namespace": "prod", "pod": "worker"},
+			wantLen: 2,
+		},
+		{
+			name:     "missing incoming label",
+			strategy: map[string]string{"namespace": "namespace", "pod": "pod"},
+			targets:  []model.LabelSet{{"namespace": "prod", "pod": "api"}},
+			input:    model.LabelSet{"namespace": "prod"},
+			wantLen:  1,
+		},
+		{
+			name:     "empty incoming label",
+			strategy: map[string]string{"namespace": "namespace", "pod": "pod"},
+			targets:  []model.LabelSet{{"namespace": "prod", "pod": "api"}},
+			input:    model.LabelSet{"namespace": "prod", "pod": ""},
+			wantLen:  1,
+		},
+		{
+			name:     "missing and empty target labels are skipped",
+			strategy: map[string]string{"namespace": "namespace", "pod": "pod"},
+			targets: []model.LabelSet{
+				{"namespace": "prod", "pod": "api"},
+				{"namespace": "prod"},
+				{"namespace": "prod", "pod": ""},
+			},
+			input:   model.LabelSet{"namespace": "prod", "pod": "api"},
+			want:    model.LabelSet{"namespace": "prod", "pod": "api"},
+			wantLen: 1,
+		},
+		{
+			name:    "empty strategy does not match",
+			targets: []model.LabelSet{{"service": "api"}},
+			input:   model.LabelSet{"service": "api"},
+		},
+		{
+			name:     "no targets",
+			strategy: map[string]string{"service": "service"},
+			input:    model.LabelSet{"service": "api"},
+		},
+		{
+			name:     "last duplicate target wins",
+			strategy: map[string]string{"service": "service"},
+			targets: []model.LabelSet{
+				{"service": "api", "env": "old"},
+				{"service": "api", "env": "new"},
+			},
+			input:   model.LabelSet{"service": "api"},
+			want:    model.LabelSet{"service": "api", "env": "new"},
+			wantLen: 1,
+		},
+		{
+			name:     "value boundaries are preserved",
+			strategy: map[string]string{"a": "a", "b": "b"},
+			targets: []model.LabelSet{
+				{"a": "ab", "b": "c"},
+				{"a": "a", "b": "bc"},
+			},
+			input:   model.LabelSet{"a": "ab", "b": "c"},
+			want:    model.LabelSet{"a": "ab", "b": "c"},
+			wantLen: 2,
+		},
+		{
+			name:     "multiple target labels map to the same incoming label",
+			strategy: map[string]string{"a": "service", "b": "service"},
+			targets:  []model.LabelSet{{"a": "api", "b": "api"}},
+			input:    model.LabelSet{"service": "api"},
+			want:     model.LabelSet{"a": "api", "b": "api"},
+			wantLen:  1,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			targets := make([]discovery.Target, 0, len(tt.targets))
+			for _, target := range tt.targets {
+				targets = append(targets, discovery.NewTargetFromLabelSet(target))
+			}
+			matcher := enrich.NewMatcher(targets, tt.strategy)
+			require.Equal(t, tt.wantLen, matcher.Len())
+
+			// Exercise both label representations used by the components.
+			require.Equal(t, tt.want, matcher.Match(func(name string) string {
+				return string(tt.input[model.LabelName(name)])
+			}))
+			builder := labels.NewScratchBuilder(len(tt.input))
+			for name, value := range tt.input {
+				builder.Add(string(name), string(value))
+			}
+			builder.Sort()
+			require.Equal(t, tt.want, matcher.Match(builder.Labels().Get))
+		})
+	}
+}
+
+func TestMatcherSnapshotsTargetsAndStrategy(t *testing.T) {
+	strategy := map[string]string{"service": "service_name"}
+	own := model.LabelSet{"service": "api", "env": "prod"}
+	group := model.LabelSet{"service": "default", "region": "eu"}
+	target := discovery.NewTargetFromSpecificAndBaseLabelSet(own, group)
+	matcher := enrich.NewMatcher([]discovery.Target{target}, strategy)
+
+	strategy["service"] = "other"
+	own["service"] = "worker"
+	own["env"] = "stage"
+	group["region"] = "us"
+
+	require.Equal(t, model.LabelSet{"service": "api", "env": "prod", "region": "eu"},
+		matcher.Match(labels.FromStrings("service_name", "api").Get))
+	require.Nil(t, matcher.Match(labels.FromStrings("service_name", "worker").Get))
+}