feat(loki/src/k8s): configurable default instance label for scraped pod logs

CHANGELOG.md

@@ -47,6 +47,8 @@ v0.41.0 (2024-05-31)
 
 - Propagate request metadata for `faro.receiver` to downstream components. (@hainenber)
 
+- Configurable default `instance` label for logs scraped by `loki.source.kubernetes`. (@hainenber)
+
 ### Features
 
 - A new `loki.rules.kubernetes` component that discovers `PrometheusRule` Kubernetes resources and loads them into a Loki Ruler instance. (@EStork09)

docs/sources/flow/reference/components/loki.source.kubernetes.md

@@ -52,6 +52,7 @@ Name | Type | Description | Default | Required
 `targets` | `list(map(string))` | List of files to read from. | | yes
 `forward_to` | `list(LogsReceiver)` | List of receivers to send log entries to. | | yes
 
+
 Each target in `targets` must have the following labels:
 
 * `__meta_kubernetes_namespace` or `__pod_namespace__` to specify the namespace
@@ -95,6 +96,7 @@ inside a `client` block.
 [oauth2]: #oauth2-block
 [tls_config]: #tls_config-block
 [clustering]: #clustering-block
+[default_instance_label]: #default_instance_label-block
 
 ### client block
 
@@ -159,6 +161,19 @@ arguments.
 
 [using clustering]: {{< relref "../../concepts/clustering.md" >}}
 
+### default_instance_label block
+
+The `default_instance_label` block configures default "instance" label to be assigned with logs 
+scraped from inputted targets.
+If `default_instance_label` block is not provided, the target's address is used as the default `instance` label.
+
+The following arguments are supported:
+
+Name                     | Type                | Description                                                   | Default | Required
+------------------------ | ------------------- | ------------------------------------------------------------- | ------- | --------
+`enabled`                | `bool`              | Enable setting default "instance" label value.                | true    | no
+`value`                  | `string`            | Value for the label.                                          |         | no
+
 ## Exported fields
 
 `loki.source.kubernetes` does not export any fields.

internal/component/loki/source/kubernetes/kubernetes.go

@@ -45,11 +45,15 @@ type Arguments struct {
 	Client commonk8s.ClientArguments `river:"client,block,optional"`
 
 	Clustering cluster.ComponentBlock `river:"clustering,block,optional"`
+
+	// DefaultInstanceLabel configures "instance" label
+	DefaultInstanceLabel kubetail.DefaultInstanceLabel `river:"default_instance_label,block,optional"`
 }
 
 // DefaultArguments holds default settings for loki.source.kubernetes.
 var DefaultArguments = Arguments{
-	Client: commonk8s.DefaultClientArguments,
+	Client:               commonk8s.DefaultClientArguments,
+	DefaultInstanceLabel: kubetail.DefaultInstanceLabel{Enabled: true},
 }
 
 // SetToDefault implements river.Defaulter.
@@ -193,7 +197,7 @@ func (c *Component) resyncTargets(targets []discovery.Target) {
 	tailTargets := make([]*kubetail.Target, 0, len(targets))
 	for _, target := range targets {
 		lset := target.Labels()
-		processed, err := kubetail.PrepareLabels(lset, c.opts.ID)
+		processed, err := kubetail.PrepareLabels(lset, c.opts.ID, c.args.DefaultInstanceLabel)
 		if err != nil {
 			// TODO(rfratto): should this set the health of the component?
 			level.Error(c.log).Log("msg", "failed to process input target", "target", lset.String(), "err", err)

internal/component/loki/source/kubernetes/kubernetes_test.go

@@ -13,6 +13,10 @@ func TestRiverConfig(t *testing.T) {
 		{"__address__" = "localhost:9090", "foo" = "bar"},
 		{"__address__" = "localhost:8080", "foo" = "buzz"},
 	]
+	default_instance_label {
+		enabled = true
+		value = "test"
+	}
     forward_to = []
 	client {
 		api_server = "localhost:9091"

internal/component/loki/source/kubernetes/kubetail/target.go

@@ -41,6 +41,11 @@ type Target struct {
 	lastEntry time.Time
 }
 
+type DefaultInstanceLabel struct {
+	Enabled bool   `river:"enabled,attr,optional"`
+	Value   string `river:"value,attr,optional"`
+}
+
 // NewTarget creates a new Target which can be passed to a tailer.
 func NewTarget(origLabels labels.Labels, lset labels.Labels) *Target {
 	// Precompute some values based on labels so we don't have to continually
@@ -156,7 +161,7 @@ func (t *Target) LastEntry() time.Time {
 //
 // Validation of lset fails if there is no label indicating the pod namespace,
 // pod name, or container name.
-func PrepareLabels(lset labels.Labels, defaultJob string) (res labels.Labels, err error) {
+func PrepareLabels(lset labels.Labels, defaultJob string, defaultInstanceLabel DefaultInstanceLabel) (res labels.Labels, err error) {
 	tailLabels := []labels.Label{
 		{Name: model.JobLabel, Value: defaultJob},
 	}
@@ -220,8 +225,11 @@ func PrepareLabels(lset labels.Labels, defaultJob string) (res labels.Labels, er
 	}
 
 	// Default the instance label to the target address.
-	if !lset.Has(model.InstanceLabel) {
+	if !lset.Has(model.InstanceLabel) && defaultInstanceLabel.Enabled {
 		defaultInstance := fmt.Sprintf("%s/%s:%s", podNamespace, podName, podContainerName)
+		if defaultInstanceLabel.Value != "" {
+			defaultInstance = defaultInstanceLabel.Value
+		}
 		lb.Set(model.InstanceLabel, defaultInstance)
 	}
 

internal/component/loki/source/podlogs/reconciler.go

@@ -228,7 +228,8 @@ func (r *reconciler) reconcilePodLogs(ctx context.Context, cli client.Client, po
 			processedLabels, _ := relabel.Process(targetLabels.Copy(), relabelRules...)
 
 			defaultJob := fmt.Sprintf("%s/%s:%s", podLogs.Namespace, podLogs.Name, container.Name)
-			finalLabels, err := kubetail.PrepareLabels(processedLabels, defaultJob)
+			// TODO: make default "instance" label for pod logs configurable.
+			finalLabels, err := kubetail.PrepareLabels(processedLabels, defaultJob, kubetail.DefaultInstanceLabel{Enabled: true})
 
 			if err != nil {
 				discoveredPod.Containers = append(discoveredPod.Containers, DiscoveredContainer{