feat(lambda-promtail): add relabeling support for log entries (#15600)

Author: cyriltovena

docs/sources/send-data/lambda-promtail/_index.md

@@ -1,13 +1,13 @@
 ---
-title: Lambda Promtail client 
+title: Lambda Promtail client
 menuTitle:  Lambda Promtail
 description: Configuring the Lambda Promtail client to send logs to Loki.
-aliases: 
+aliases:
 - ../clients/lambda-promtail/
 weight:  700
 ---
 
-# Lambda Promtail client 
+# Lambda Promtail client
 
 Grafana Loki includes [Terraform](https://www.terraform.io/) and [CloudFormation](https://aws.amazon.com/cloudformation/) for shipping Cloudwatch, Cloudtrail, VPC Flow Logs and loadbalancer logs to Loki via a [lambda function](https://aws.amazon.com/lambda/). This is done via [lambda-promtail](https://github.com/grafana/loki/blob/main/tools/lambda-promtail) which processes cloudwatch events and propagates them to Loki (or a Promtail instance) via the push-api [scrape config]({{< relref "../../send-data/promtail/configuration#loki_push_api" >}}).
 
@@ -161,6 +161,113 @@ Incoming logs can have seven special labels assigned to them which can be used i
 - `__aws_s3_log_lb`: The name of the loadbalancer.
 - `__aws_s3_log_lb_owner`: The Account ID of the loadbalancer owner.
 
+## Relabeling Configuration
+
+Lambda-promtail supports Prometheus-style relabeling through the `RELABEL_CONFIGS` environment variable. This allows you to modify, keep, or drop labels before sending logs to Loki. The configuration is provided as a JSON array of relabel configurations. The relabeling functionality follows the same principles as Prometheus relabeling - for a detailed explanation of how relabeling works, see [How relabeling in Prometheus works](https://grafana.com/blog/2022/03/21/how-relabeling-in-prometheus-works/).
+
+Example configurations:
+
+1. Rename a label and capture regex groups:
+```json
+{
+  "RELABEL_CONFIGS": [
+    {
+      "source_labels": ["__aws_log_type"],
+      "target_label": "log_type",
+      "action": "replace",
+      "regex": "(.*)",
+      "replacement": "${1}"
+    }
+  ]
+}
+```
+
+2. Keep only specific log types (useful for filtering):
+```json
+{
+  "RELABEL_CONFIGS": [
+    {
+      "source_labels": ["__aws_log_type"],
+      "regex": "s3_.*",
+      "action": "keep"
+    }
+  ]
+}
+```
+
+3. Drop internal AWS labels (cleanup):
+```json
+{
+  "RELABEL_CONFIGS": [
+    {
+      "regex": "__aws_.*",
+      "action": "labeldrop"
+    }
+  ]
+}
+```
+
+4. Multiple relabeling rules (combining different actions):
+```json
+{
+  "RELABEL_CONFIGS": [
+    {
+      "source_labels": ["__aws_log_type"],
+      "target_label": "log_type",
+      "action": "replace",
+      "regex": "(.*)",
+      "replacement": "${1}"
+    },
+    {
+      "source_labels": ["__aws_s3_log_lb"],
+      "target_label": "loadbalancer",
+      "action": "replace"
+    },
+    {
+      "regex": "__aws_.*",
+      "action": "labeldrop"
+    }
+  ]
+}
+```
+
+### Supported Actions
+
+The following actions are supported, matching Prometheus relabeling capabilities:
+
+- `replace`: Replace a label value with a new value using regex capture groups
+- `keep`: Keep entries where labels match the regex (useful for filtering)
+- `drop`: Drop entries where labels match the regex (useful for excluding)
+- `hashmod`: Set a label to the modulus of a hash of labels (useful for sharding)
+- `labelmap`: Copy labels to other labels based on regex matching
+- `labeldrop`: Remove labels matching the regex pattern
+- `labelkeep`: Keep only labels matching the regex pattern
+- `lowercase`: Convert label values to lowercase
+- `uppercase`: Convert label values to uppercase
+
+### Configuration Fields
+
+Each relabel configuration supports these fields (all fields are optional except for `action`):
+
+- `source_labels`: List of label names to use as input for the action
+- `separator`: String to join source label values (default: ";")
+- `target_label`: Label to modify (required for replace and hashmod actions)
+- `regex`: Regular expression to match against (defaults to "(.+)" for most actions)
+- `replacement`: Replacement pattern for matched regex, supports ${1}, ${2}, etc. for capture groups
+- `modulus`: Modulus for hashmod action
+- `action`: One of the supported actions listed above
+
+### Important Notes
+
+1. Relabeling is applied after merging extra labels and dropping labels specified by `DROP_LABELS`.
+2. If all labels are removed after relabeling, the log entry will be dropped entirely.
+3. The relabeling configuration follows the same format as Prometheus's relabel_configs, making it familiar for users of Prometheus.
+4. Relabeling rules are processed in order, and each rule can affect the input of subsequent rules.
+5. Regular expressions in the `regex` field support full RE2 syntax.
+6. For the `replace` action, if the `regex` doesn't match, the target label remains unchanged.
+
+For more details about how relabeling works and advanced use cases, refer to the [Prometheus relabeling blog post](https://grafana.com/blog/2022/03/21/how-relabeling-in-prometheus-works/).
+
 ## Limitations
 
 ### Promtail labels

tools/lambda-promtail/lambda-promtail/eventbridge.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+
 	"github.com/aws/aws-lambda-go/events"
 	"github.com/go-kit/log"
 )

tools/lambda-promtail/lambda-promtail/eventbridge_test.go

@@ -3,11 +3,12 @@ package main
 import (
 	"context"
 	"encoding/json"
+	"os"
+	"testing"
+
 	"github.com/aws/aws-lambda-go/events"
 	"github.com/go-kit/log"
 	"github.com/stretchr/testify/require"
-	"os"
-	"testing"
 )
 
 type testPromtailClient struct{}

tools/lambda-promtail/lambda-promtail/main.go

@@ -13,6 +13,8 @@ import (
 	"github.com/go-kit/log/level"
 	"github.com/grafana/dskit/backoff"
 	"github.com/prometheus/common/model"
+	prommodel "github.com/prometheus/prometheus/model/labels"
+	"github.com/prometheus/prometheus/model/relabel"
 
 	"github.com/aws/aws-lambda-go/events"
 	"github.com/aws/aws-lambda-go/lambda"
@@ -38,6 +40,7 @@ var (
 	dropLabels                                                               []model.LabelName
 	skipTlsVerify                                                            bool
 	printLogLine                                                             bool
+	relabelConfigs                                                           []*relabel.Config
 )
 
 func setupArguments() {
@@ -106,14 +109,21 @@ func setupArguments() {
 		printLogLine = false
 	}
 	s3Clients = make(map[string]*s3.Client)
+
+	// Parse relabel configs from environment variable
+	if relabelConfigsRaw := os.Getenv("RELABEL_CONFIGS"); relabelConfigsRaw != "" {
+		if err := json.Unmarshal([]byte(relabelConfigsRaw), &relabelConfigs); err != nil {
+			panic(fmt.Errorf("failed to parse RELABEL_CONFIGS: %v", err))
+		}
+	}
 }
 
 func parseExtraLabels(extraLabelsRaw string, omitPrefix bool) (model.LabelSet, error) {
 	prefix := "__extra_"
 	if omitPrefix {
 		prefix = ""
 	}
-	var extractedLabels = model.LabelSet{}
+	extractedLabels := model.LabelSet{}
 	extraLabelsSplit := strings.Split(extraLabelsRaw, ",")
 
 	if len(extraLabelsRaw) < 1 {
@@ -151,13 +161,53 @@ func getDropLabels() ([]model.LabelName, error) {
 	return result, nil
 }
 
+func applyRelabelConfigs(labels model.LabelSet) model.LabelSet {
+	if len(relabelConfigs) == 0 {
+		return labels
+	}
+
+	// Convert model.LabelSet to prommodel.Labels
+	promLabels := make([]prommodel.Label, 0, len(labels))
+	for name, value := range labels {
+		promLabels = append(promLabels, prommodel.Label{
+			Name:  string(name),
+			Value: string(value),
+		})
+	}
+
+	// Sort labels as required by Process
+	promLabels = prommodel.New(promLabels...)
+
+	// Apply relabeling
+	processedLabels, keep := relabel.Process(promLabels, relabelConfigs...)
+	if !keep {
+		return model.LabelSet{}
+	}
+
+	// Convert back to model.LabelSet
+	result := make(model.LabelSet)
+	for _, l := range processedLabels {
+		result[model.LabelName(l.Name)] = model.LabelValue(l.Value)
+	}
+
+	return result
+}
+
 func applyLabels(labels model.LabelSet) model.LabelSet {
 	finalLabels := labels.Merge(extraLabels)
 
 	for _, dropLabel := range dropLabels {
 		delete(finalLabels, dropLabel)
 	}
 
+	// Apply relabeling after merging extra labels and dropping labels
+	finalLabels = applyRelabelConfigs(finalLabels)
+
+	// Skip entries with no labels after relabeling
+	if len(finalLabels) == 0 {
+		return nil
+	}
+
 	return finalLabels
 }
 

tools/lambda-promtail/lambda-promtail/promtail.go

@@ -58,6 +58,11 @@ func newBatch(ctx context.Context, pClient Client, entries ...entry) (*batch, er
 }
 
 func (b *batch) add(ctx context.Context, e entry) error {
+	// Skip entries with no labels (filtered out by relabeling)
+	if e.labels == nil {
+		return nil
+	}
+
 	labels := labelsMapToString(e.labels, reservedLabelTenantID)
 	stream, ok := b.streams[labels]
 	if !ok {

tools/lambda-promtail/lambda-promtail/s3.go

@@ -290,8 +290,8 @@ func processS3Event(ctx context.Context, ev *events.S3Event, pc Client, log *log
 		}
 		obj, err := s3Client.GetObject(ctx,
 			&s3.GetObjectInput{
-				Bucket:              aws.String(labels["bucket"]),
-				Key:                 aws.String(labels["key"]),
+				Bucket: aws.String(labels["bucket"]),
+				Key:    aws.String(labels["key"]),
 			})
 		if err != nil {
 			return fmt.Errorf("failed to get object %s from bucket %s, %s", labels["key"], labels["bucket"], err)