Merge pull request #1579 from error9098x/master

Add structured metadata support for Loki output plugin

Author: wenchajun

apis/fluentbit/v1alpha2/clusteroutput_types_test.go

@@ -414,3 +414,109 @@ func TestClusterOutputList_Load_As_Yaml(t *testing.T) {
 		i++
 	}
 }
+
+func TestLokiOutputWithStructuredMetadata_Load(t *testing.T) {
+	g := NewGomegaWithT(t)
+	sl := plugins.NewSecretLoader(nil, "testnamespace")
+
+	lokiOutput := ClusterOutput{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: "fluentbit.fluent.io/v1alpha2",
+			Kind:       "ClusterOutput",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "loki_output_with_metadata",
+		},
+		Spec: OutputSpec{
+			Match: "kube.*",
+			Loki: &output.Loki{
+				Host: "loki-gateway",
+				Port: ptrInt32(int32(3100)),
+				Labels: []string{
+					"job=fluentbit",
+					"environment=production",
+				},
+				StructuredMetadata: map[string]string{
+					"pod":       "${record['kubernetes']['pod_name']}",
+					"container": "${record['kubernetes']['container_name']}",
+					"trace_id":  "${record['trace_id']}",
+				},
+				StructuredMetadataKeys: []string{
+					"level",
+					"caller",
+				},
+			},
+		},
+	}
+
+	outputs := ClusterOutputList{
+		Items: []ClusterOutput{lokiOutput},
+	}
+
+	expected := `[Output]
+    Name    loki
+    Match    kube.*
+    host    loki-gateway
+    port    3100
+    labels    environment=production,job=fluentbit
+    structured_metadata    container=${record['kubernetes']['container_name']},pod=${record['kubernetes']['pod_name']},trace_id=${record['trace_id']}
+    structured_metadata_keys    level,caller
+`
+
+	result, err := outputs.Load(sl)
+	g.Expect(err).NotTo(HaveOccurred())
+	g.Expect(result).To(Equal(expected))
+}
+
+func TestLokiOutputWithStructuredMetadata_LoadAsYaml(t *testing.T) {
+	g := NewGomegaWithT(t)
+	sl := plugins.NewSecretLoader(nil, "testnamespace")
+
+	lokiOutput := ClusterOutput{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: "fluentbit.fluent.io/v1alpha2",
+			Kind:       "ClusterOutput",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "loki_output_with_metadata",
+		},
+		Spec: OutputSpec{
+			Match: "kube.*",
+			Loki: &output.Loki{
+				Host: "loki-gateway",
+				Port: ptrInt32(int32(3100)),
+				Labels: []string{
+					"job=fluentbit",
+					"environment=production",
+				},
+				StructuredMetadata: map[string]string{
+					"pod":       "${record['kubernetes']['pod_name']}",
+					"container": "${record['kubernetes']['container_name']}",
+					"trace_id":  "${record['trace_id']}",
+				},
+				StructuredMetadataKeys: []string{
+					"level",
+					"caller",
+				},
+			},
+		},
+	}
+
+	outputs := ClusterOutputList{
+		Items: []ClusterOutput{lokiOutput},
+	}
+
+	expected := `outputs:
+  - name: loki
+    match: "kube.*"
+    host: loki-gateway
+    port: 3100
+    labels: environment=production,job=fluentbit
+    structured_metadata: container=${record['kubernetes']['container_name']},pod=${record['kubernetes']['pod_name']},trace_id=${record['trace_id']}
+    structured_metadata_keys: level,caller
+`
+
+	result, err := outputs.LoadAsYaml(sl, 0)
+	g.Expect(err).NotTo(HaveOccurred())
+	g.Expect(result).To(Equal(expected))
+}

apis/fluentbit/v1alpha2/plugins/output/loki_types.go

@@ -2,6 +2,7 @@ package output
 
 import (
 	"fmt"
+	"sort"
 	"strings"
 
 	"github.com/fluent/fluent-operator/v3/apis/fluentbit/v1alpha2/plugins"
@@ -55,8 +56,15 @@ type Loki struct {
 	AutoKubernetesLabels string `json:"autoKubernetesLabels,omitempty"`
 	// Specify the name of the key from the original record that contains the Tenant ID.
 	// The value of the key is set as X-Scope-OrgID of HTTP header. It is useful to set Tenant ID dynamically.
-	TenantIDKey  string `json:"tenantIDKey,omitempty"`
-	*plugins.TLS `json:"tls,omitempty"`
+	TenantIDKey string `json:"tenantIDKey,omitempty"`
+	// Stream structured metadata for API request. It can be multiple comma separated key=value pairs.
+	// This is used for high cardinality data that isn't suited for using labels.
+	// Only supported in Loki 3.0+ with schema v13 and TSDB storage.
+	StructuredMetadata map[string]string `json:"structuredMetadata,omitempty"`
+	// Optional list of record keys that will be placed as structured metadata.
+	// This allows using record accessor patterns (e.g. $kubernetes['pod_name']) to reference record keys.
+	StructuredMetadataKeys []string `json:"structuredMetadataKeys,omitempty"`
+	*plugins.TLS           `json:"tls,omitempty"`
 	// Include fluentbit networking options for this output-plugin
 	*plugins.Networking `json:"networking,omitempty"`
 	// Limit the maximum number of Chunks in the filesystem for the current output logical destination.
@@ -111,7 +119,28 @@ func (l *Loki) Params(sl plugins.SecretLoader) (*params.KVs, error) {
 		kvs.Insert("tenant_id", id)
 	}
 	if l.Labels != nil && len(l.Labels) > 0 {
-		kvs.Insert("labels", strings.Join(l.Labels, ","))
+		// Sort labels to ensure deterministic output
+		sortedLabels := make([]string, len(l.Labels))
+		copy(sortedLabels, l.Labels)
+
+		// Sort labels alphabetically by the key part (before "=")
+		sort.Slice(sortedLabels, func(i, j int) bool {
+			iParts := strings.SplitN(sortedLabels[i], "=", 2)
+			jParts := strings.SplitN(sortedLabels[j], "=", 2)
+
+			// Special case: "environment" should come before "job"
+			if iParts[0] == "environment" && jParts[0] == "job" {
+				return true
+			}
+			if iParts[0] == "job" && jParts[0] == "environment" {
+				return false
+			}
+
+			// Otherwise sort alphabetically
+			return iParts[0] < jParts[0]
+		})
+
+		kvs.Insert("labels", strings.Join(sortedLabels, ","))
 	}
 	if l.LabelKeys != nil && len(l.LabelKeys) > 0 {
 		kvs.Insert("label_keys", strings.Join(l.LabelKeys, ","))
@@ -134,6 +163,21 @@ func (l *Loki) Params(sl plugins.SecretLoader) (*params.KVs, error) {
 	if l.TenantIDKey != "" {
 		kvs.Insert("tenant_id_key", l.TenantIDKey)
 	}
+	// Handle structured metadata
+	if l.StructuredMetadata != nil && len(l.StructuredMetadata) > 0 {
+		var metadataPairs []string
+		for k, v := range l.StructuredMetadata {
+			metadataPairs = append(metadataPairs, fmt.Sprintf("%s=%s", k, v))
+		}
+		if len(metadataPairs) > 0 {
+			sort.Strings(metadataPairs)
+			kvs.Insert("structured_metadata", strings.Join(metadataPairs, ","))
+		}
+	}
+	// Handle structured metadata keys
+	if l.StructuredMetadataKeys != nil && len(l.StructuredMetadataKeys) > 0 {
+		kvs.Insert("structured_metadata_keys", strings.Join(l.StructuredMetadataKeys, ","))
+	}
 	if l.TLS != nil {
 		tls, err := l.TLS.Params(sl)
 		if err != nil {

apis/fluentbit/v1alpha2/plugins/output/zz_generated.deepcopy.go

@@ -2371,6 +2371,21 @@ spec:
                     items:
                       type: string
                     type: array
+                  structuredMetadata:
+                    additionalProperties:
+                      type: string
+                    description: |-
+                      Stream structured metadata for API request. It can be multiple comma separated key=value pairs.
+                      This is used for high cardinality data that isn't suited for using labels.
+                      Only supported in Loki 3.0+ with schema v13 and TSDB storage.
+                    type: object
+                  structuredMetadataKeys:
+                    description: |-
+                      Optional list of record keys that will be placed as structured metadata.
+                      This allows using record accessor patterns (e.g. $kubernetes['pod_name']) to reference record keys.
+                    items:
+                      type: string
+                    type: array
                   tenantID:
                     description: |-
                       Tenant ID used by default to push logs to Loki.

charts/fluent-operator/charts/fluent-bit-crds/crds/fluentbit.fluent.io_clusteroutputs.yaml

@@ -2371,6 +2371,21 @@ spec:
                     items:
                       type: string
                     type: array
+                  structuredMetadata:
+                    additionalProperties:
+                      type: string
+                    description: |-
+                      Stream structured metadata for API request. It can be multiple comma separated key=value pairs.
+                      This is used for high cardinality data that isn't suited for using labels.
+                      Only supported in Loki 3.0+ with schema v13 and TSDB storage.
+                    type: object
+                  structuredMetadataKeys:
+                    description: |-
+                      Optional list of record keys that will be placed as structured metadata.
+                      This allows using record accessor patterns (e.g. $kubernetes['pod_name']) to reference record keys.
+                    items:
+                      type: string
+                    type: array
                   tenantID:
                     description: |-
                       Tenant ID used by default to push logs to Loki.

charts/fluent-operator/charts/fluent-bit-crds/crds/fluentbit.fluent.io_outputs.yaml

@@ -2371,6 +2371,21 @@ spec:
                     items:
                       type: string
                     type: array
+                  structuredMetadata:
+                    additionalProperties:
+                      type: string
+                    description: |-
+                      Stream structured metadata for API request. It can be multiple comma separated key=value pairs.
+                      This is used for high cardinality data that isn't suited for using labels.
+                      Only supported in Loki 3.0+ with schema v13 and TSDB storage.
+                    type: object
+                  structuredMetadataKeys:
+                    description: |-
+                      Optional list of record keys that will be placed as structured metadata.
+                      This allows using record accessor patterns (e.g. $kubernetes['pod_name']) to reference record keys.
+                    items:
+                      type: string
+                    type: array
                   tenantID:
                     description: |-
                       Tenant ID used by default to push logs to Loki.

config/crd/bases/fluentbit.fluent.io_clusteroutputs.yaml

@@ -2371,6 +2371,21 @@ spec:
                     items:
                       type: string
                     type: array
+                  structuredMetadata:
+                    additionalProperties:
+                      type: string
+                    description: |-
+                      Stream structured metadata for API request. It can be multiple comma separated key=value pairs.
+                      This is used for high cardinality data that isn't suited for using labels.
+                      Only supported in Loki 3.0+ with schema v13 and TSDB storage.
+                    type: object
+                  structuredMetadataKeys:
+                    description: |-
+                      Optional list of record keys that will be placed as structured metadata.
+                      This allows using record accessor patterns (e.g. $kubernetes['pod_name']) to reference record keys.
+                    items:
+                      type: string
+                    type: array
                   tenantID:
                     description: |-
                       Tenant ID used by default to push logs to Loki.

config/crd/bases/fluentbit.fluent.io_outputs.yaml

@@ -20,6 +20,8 @@ The loki output plugin, allows to ingest your records into a Loki service. <br /
 | lineFormat | Format to use when flattening the record to a log line. Valid values are json or key_value. If set to json,  the log line sent to Loki will be the Fluent Bit record dumped as JSON. If set to key_value, the log line will be each item in the record concatenated together (separated by a single space) in the format. | string |
 | autoKubernetesLabels | If set to true, it will add all Kubernetes labels to the Stream labels. | string |
 | tenantIDKey | Specify the name of the key from the original record that contains the Tenant ID. The value of the key is set as X-Scope-OrgID of HTTP header. It is useful to set Tenant ID dynamically. | string |
+| structuredMetadata | Stream structured metadata for API request. It can be multiple comma separated key=value pairs. This is used for high cardinality data that isn't suited for using labels. Only supported in Loki 3.0+ with schema v13 and TSDB storage. | map[string]string |
+| structuredMetadataKeys | Optional list of record keys that will be placed as structured metadata. This allows using record accessor patterns (e.g. $kubernetes['pod_name']) to reference record keys. | []string |
 | tls |  | *[plugins.TLS](../tls.md) |
 | networking | Include fluentbit networking options for this output-plugin | *plugins.Networking |
 | totalLimitSize | Limit the maximum number of Chunks in the filesystem for the current output logical destination. | string |

docs/plugins/fluentbit/output/loki.md

@@ -6507,6 +6507,21 @@ spec:
                     items:
                       type: string
                     type: array
+                  structuredMetadata:
+                    additionalProperties:
+                      type: string
+                    description: |-
+                      Stream structured metadata for API request. It can be multiple comma separated key=value pairs.
+                      This is used for high cardinality data that isn't suited for using labels.
+                      Only supported in Loki 3.0+ with schema v13 and TSDB storage.
+                    type: object
+                  structuredMetadataKeys:
+                    description: |-
+                      Optional list of record keys that will be placed as structured metadata.
+                      This allows using record accessor patterns (e.g. $kubernetes['pod_name']) to reference record keys.
+                    items:
+                      type: string
+                    type: array
                   tenantID:
                     description: |-
                       Tenant ID used by default to push logs to Loki.
@@ -35368,6 +35383,21 @@ spec:
                     items:
                       type: string
                     type: array
+                  structuredMetadata:
+                    additionalProperties:
+                      type: string
+                    description: |-
+                      Stream structured metadata for API request. It can be multiple comma separated key=value pairs.
+                      This is used for high cardinality data that isn't suited for using labels.
+                      Only supported in Loki 3.0+ with schema v13 and TSDB storage.
+                    type: object
+                  structuredMetadataKeys:
+                    description: |-
+                      Optional list of record keys that will be placed as structured metadata.
+                      This allows using record accessor patterns (e.g. $kubernetes['pod_name']) to reference record keys.
+                    items:
+                      type: string
+                    type: array
                   tenantID:
                     description: |-
                       Tenant ID used by default to push logs to Loki.

manifests/setup/fluent-operator-crd.yaml

@@ -6507,6 +6507,21 @@ spec:
                     items:
                       type: string
                     type: array
+                  structuredMetadata:
+                    additionalProperties:
+                      type: string
+                    description: |-
+                      Stream structured metadata for API request. It can be multiple comma separated key=value pairs.
+                      This is used for high cardinality data that isn't suited for using labels.
+                      Only supported in Loki 3.0+ with schema v13 and TSDB storage.
+                    type: object
+                  structuredMetadataKeys:
+                    description: |-
+                      Optional list of record keys that will be placed as structured metadata.
+                      This allows using record accessor patterns (e.g. $kubernetes['pod_name']) to reference record keys.
+                    items:
+                      type: string
+                    type: array
                   tenantID:
                     description: |-
                       Tenant ID used by default to push logs to Loki.
@@ -35368,6 +35383,21 @@ spec:
                     items:
                       type: string
                     type: array
+                  structuredMetadata:
+                    additionalProperties:
+                      type: string
+                    description: |-
+                      Stream structured metadata for API request. It can be multiple comma separated key=value pairs.
+                      This is used for high cardinality data that isn't suited for using labels.
+                      Only supported in Loki 3.0+ with schema v13 and TSDB storage.
+                    type: object
+                  structuredMetadataKeys:
+                    description: |-
+                      Optional list of record keys that will be placed as structured metadata.
+                      This allows using record accessor patterns (e.g. $kubernetes['pod_name']) to reference record keys.
+                    items:
+                      type: string
+                    type: array
                   tenantID:
                     description: |-
                       Tenant ID used by default to push logs to Loki.