Add loki.enrich component

CHANGELOG.md

@@ -13,6 +13,7 @@ Main (unreleased)
 ### Features
 
 - Add `otelcol.receiver.awscloudwatch` component to receive logs from AWS CloudWatch and forward them to other `otelcol.*` components. (@wildum)
+- Add `loki.enrich` component to enrich logs using labels from `discovery.*` components. (@v-zhuravlev)
 
 ### Enhancements
 

docs/sources/reference/compatibility/_index.md

@@ -135,6 +135,7 @@ The following components, grouped by namespace, _consume_ Targets.
 {{< /collapse >}}
 
 {{< collapse title="loki" >}}
+- [loki.enrich](../components/loki/loki.enrich)
 - [loki.source.docker](../components/loki/loki.source.docker)
 - [loki.source.file](../components/loki/loki.source.file)
 - [loki.source.kubernetes](../components/loki/loki.source.kubernetes)
@@ -221,6 +222,7 @@ The following components, grouped by namespace, _export_ Loki `LogsReceiver`.
 
 {{< collapse title="loki" >}}
 - [loki.echo](../components/loki/loki.echo)
+- [loki.enrich](../components/loki/loki.enrich)
 - [loki.process](../components/loki/loki.process)
 - [loki.relabel](../components/loki/loki.relabel)
 - [loki.secretfilter](../components/loki/loki.secretfilter)
@@ -249,6 +251,7 @@ The following components, grouped by namespace, _consume_ Loki `LogsReceiver`.
 {{< /collapse >}}
 
 {{< collapse title="loki" >}}
+- [loki.enrich](../components/loki/loki.enrich)
 - [loki.process](../components/loki/loki.process)
 - [loki.relabel](../components/loki/loki.relabel)
 - [loki.secretfilter](../components/loki/loki.secretfilter)

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

@@ -0,0 +1,153 @@
+---
+aliases:
+- /docs/alloy/latest/reference/components/loki/loki.enrich/
+canonical: /docs/alloy/latest/reference/components/loki/loki.enrich/
+title: loki.enrich
+labels:
+  stage: experimental
+description: The loki.enrich component enriches logs with labels from service discovery.
+---
+
+# loki.enrich
+
+{{< 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.
+
+## Usage
+
+```alloy
+loki.enrich "LABEL" {
+  // List of targets from a discovery component
+  targets = DISCOVERY_COMPONENT.targets
+
+  // Which label from discovered targets to match against
+  match_label = "LABEL"
+
+  // Which label from incoming logs to match against
+  source_label = "LABEL"
+
+  // List of labels to copy from discovered targets to logs
+  target_labels = ["LABEL", ...]
+
+  // Where to send enriched logs
+  forward_to = [RECEIVER_LIST]
+}
+```
+
+## Arguments
+
+The following arguments are supported:
+
+Name | Type | Description | Default | Required
+---- | ---- | ----------- | ------- | --------
+`targets` | `[]discovery.Target` | List of targets from a discovery component. | | yes
+`target_match_label` | `string` | Which label from discovered targets to match against (e.g., "__inventory_consul_service"). | | yes
+`logs_match_label` | `string` | Which label from incoming logs to match against discovered targets (e.g., "service_name"). If not specified, `target_match_label` will be used. | `target_match_label` | no
+`target_labels` | `[]string` | List of labels to copy from discovered targets to logs. If empty, all labels will be copied. | | no
+`forward_to` | `[]loki.LogsReceiver` | List of receivers to send enriched logs to. | | yes
+
+## Exports
+
+The following values are exported:
+
+Name | Type | Description
+---- | ---- | -----------
+`receiver` | `loki.LogsReceiver` | A receiver that can be used to send logs to this component.
+
+## Example
+
+```alloy
+// Configure HTTP discovery
+discovery.http "default" {
+    url = "http://network-inventory.example.com/prometheus_sd"
+}
+
+discovery.relabel "default" {
+    targets = discovery.http.default.targets
+    rule {
+        action        = "replace"
+        source_labels = ["__inventory_rack"]
+        target_label  = "rack"
+    }
+    rule {
+        action        = "replace"
+        source_labels = ["__inventory_datacenter"]
+        target_label  = "datacenter"
+    }
+    rule {
+        action        = "replace"
+        source_labels = ["__inventory_environment"]
+        target_label  = "environment"
+    }
+    rule {
+        action        = "replace"
+        source_labels = ["__inventory_tenant"]
+        target_label  = "tenant"
+    }
+    rule {
+        action        = "replace"
+        source_labels = ["__inventory_primary_ip"]
+        target_label  = "primary_ip"
+    }
+}
+
+// Receive syslog messages
+loki.source.syslog "incoming" {
+    listener {
+        address = ":514"
+        protocol = "tcp"
+        labels = {
+            job = "syslog"
+        }
+    }
+    forward_to = [loki.enrich.default.receiver]
+}
+
+// Enrich logs using HTTP discovery
+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"
+
+
+
+    forward_to = [loki.write.default.receiver]
+}
+```
+
+## 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)
+2. It matches this value against the `target_match_label` in discovered targets
+3. If a match is found, it copies the requested `target_labels` from the discovered target to the log entry (if `target_labels` is empty, all labels are copied)
+4. The log entry (enriched or unchanged) is forwarded to the configured receivers
+
+## See also
+
+* [loki.source.syslog](../loki.source.syslog/)
+* [loki.source.api](../loki.source.api/)
+* [discovery.relabel](../discovery/discovery.relabel/)
+* [discovery.http](../discovery/discovery.http/) <!-- START GENERATED COMPATIBLE COMPONENTS -->
+
+## Compatible components
+
+`loki.enrich` can accept arguments from the following components:
+
+- Components that export [Targets](../../../compatibility/#targets-exporters)
+- Components that export [Loki `LogsReceiver`](../../../compatibility/#loki-logsreceiver-exporters)
+
+`loki.enrich` has exports that can be consumed by the following components:
+
+- Components that consume [Loki `LogsReceiver`](../../../compatibility/#loki-logsreceiver-consumers)
+
+{{< admonition type="note" >}}
+Connecting some components may not be sensible or components may require further configuration to make the connection work correctly.
+Refer to the linked documentation for more details.
+{{< /admonition >}}
+
+<!-- END GENERATED COMPATIBLE COMPONENTS -->

internal/cmd/integration-tests/common/common.go

@@ -1,7 +1,7 @@
 package common
 
 import (
-	"errors"
+	"fmt"
 	"io"
 	"net/http"
 	"time"
@@ -21,14 +21,14 @@ func FetchDataFromURL(url string, target Unmarshaler) error {
 	}
 	defer resp.Body.Close()
 
-	if resp.StatusCode != http.StatusOK {
-		return errors.New("Non-OK HTTP status: " + resp.Status)
-	}
-
 	bodyBytes, err := io.ReadAll(resp.Body)
 	if err != nil {
 		return err
 	}
 
+	if resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("Non-OK HTTP status: %s, body: %s, url: %s", resp.Status, string(bodyBytes), url)
+	}
+
 	return target.Unmarshal(bodyBytes)
 }

internal/cmd/integration-tests/common/log.go

@@ -2,6 +2,7 @@ package common
 
 import "encoding/json"
 
+// Query response types
 type LogResponse struct {
 	Status string `json:"status"`
 	Data   struct {
@@ -18,3 +19,8 @@ type LogData struct {
 func (m *LogResponse) Unmarshal(data []byte) error {
 	return json.Unmarshal(data, m)
 }
+
+// Push request types
+type PushRequest struct {
+	Streams []LogData `json:"streams"`
+}

internal/cmd/integration-tests/common/logs_assert.go

@@ -0,0 +1,54 @@
+package common
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const lokiURL = "http://localhost:3100/loki/api/v1/"
+
+// LogQuery returns a formatted Loki query with the given test_name label
+func LogQuery(testName string) string {
+	return fmt.Sprintf("%squery_range?query=%%7Btest_name%%3D%%22%s%%22%%7D", lokiURL, testName)
+}
+
+// AssertLogsPresent checks that logs are present in Loki and match expected labels
+func AssertLogsPresent(t *testing.T, testName string, expectedLabels map[string]string, expectedCount int) {
+	var logResponse LogResponse
+	require.EventuallyWithT(t, func(c *assert.CollectT) {
+		err := FetchDataFromURL(LogQuery(testName), &logResponse)
+		assert.NoError(c, err)
+		if len(logResponse.Data.Result) == 0 {
+			return
+		}
+
+		// Verify we got all logs
+		result := logResponse.Data.Result[0]
+		assert.Equal(c, expectedCount, len(result.Values), "should have %d log entries", expectedCount)
+
+		// Verify labels were enriched
+		for k, v := range expectedLabels {
+			assert.Equal(c, v, result.Stream[k], "label %s should be %s", k, v)
+		}
+	}, DefaultTimeout, DefaultRetryInterval)
+}
+
+// AssertLogsMissing checks that logs with specific labels are not present in Loki
+func AssertLogsMissing(t *testing.T, testName string, labels ...string) {
+	var logResponse LogResponse
+	require.EventuallyWithT(t, func(c *assert.CollectT) {
+		err := FetchDataFromURL(LogQuery(testName), &logResponse)
+		assert.NoError(c, err)
+		if len(logResponse.Data.Result) == 0 {
+			return
+		}
+
+		result := logResponse.Data.Result[0]
+		for _, label := range labels {
+			assert.NotContains(c, result.Stream, label, "label %s should not be present", label)
+		}
+	}, DefaultTimeout, DefaultRetryInterval)
+}

internal/cmd/integration-tests/tests/loki-enrich/config.alloy

@@ -0,0 +1,58 @@
+logging {
+	level = "debug"
+}
+// Discover device metadata from file
+discovery.file "network_devices" {
+  files = ["./devices.json"]
+}
+
+// Rename hidden labels so they don't get dropped
+discovery.relabel "network_devices" {
+  targets = discovery.file.network_devices.targets
+
+  rule {
+    action        = "replace"
+    source_labels = ["__meta_rack"]
+    target_label  = "rack"
+  }
+}
+
+// Receive logs via HTTP API
+loki.source.api "network_device" {
+  http {
+    listen_address = "0.0.0.0"
+    listen_port = 1514
+  }
+  labels = {
+    job = "network_device_logs",
+  }
+  forward_to = [loki.enrich.enricher.receiver]
+}
+
+// Enrich logs with device metadata
+loki.enrich "enricher" {
+
+  targets = discovery.relabel.network_devices.output
+  // Labels to copy from targets
+  target_labels = [
+    "environment",
+    "datacenter",
+    "role",
+    "rack",
+  ]
+  // Match on hostname/IP from logs
+  target_match_label = "hostname"
+  logs_match_label = "host"
+
+  forward_to = [loki.write.enriched.receiver]
+}
+
+// Write enriched logs to Loki
+loki.write "enriched" {
+  endpoint {
+    url = "http://127.0.0.1:3100/loki/api/v1/push"
+  }
+  external_labels = {
+    test_name = "network_device_enriched",
+  }
+}

internal/cmd/integration-tests/tests/loki-enrich/devices.json

@@ -0,0 +1,22 @@
+[
+  {
+    "targets": ["router1.example.com"],
+    "labels": {
+      "hostname": "router1.example.com",
+      "environment": "production",
+      "datacenter": "us-east",
+      "role": "core-router",
+      "__meta_rack": "rack1"
+    }
+  },
+  {
+    "targets": ["router2.example.com"],
+    "labels": {
+      "hostname": "router2.example.com", 
+      "environment": "production",
+      "datacenter": "us-west",
+      "role": "edge-router",
+      "__meta_rack": "rack2"
+    }
+  }
+]