feat: Add structured metadata support for Loki output plugin

Signed-off-by: ProCodec <[email protected]>

Author: error9098x

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

@@ -3,6 +3,7 @@ package output
 import (
 	"fmt"
 	"strings"
+	"sort"
 
 	"github.com/fluent/fluent-operator/v3/apis/fluentbit/v1alpha2/plugins"
 	"github.com/fluent/fluent-operator/v3/apis/fluentbit/v1alpha2/plugins/params"
@@ -56,6 +57,13 @@ type Loki struct {
 	// 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"`
+	// 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"`
@@ -134,6 +142,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 {