LOG-7804: support custom headers for Elasticsearch output

AndrewChubatiuk · AndrewChubatiuk · commit 05eea16482c2 · 2025-10-16T21:13:57.000+03:00
diff --git a/api/observability/v1/output_types.go b/api/observability/v1/output_types.go
@@ -537,6 +537,12 @@ type Elasticsearch struct {
 	// +kubebuilder:validation:Required
 	// +operator-sdk:csv:customresourcedefinitions:type=spec,displayName="ElasticSearch Version",xDescriptors={"urn:alm:descriptor:com.tectonic.ui:number"}
 	Version int `json:"version"`
+
+	// Headers specify optional headers to be sent with the request
+	//
+	// +kubebuilder:validation:Optional
+	// +operator-sdk:csv:customresourcedefinitions:type=spec,displayName="Headers"
+	Headers map[string]string `json:"headers,omitempty"`
 }
 
 // GoogleCloudLoggingAuthentication contains configuration for authenticating requests to a GoogleCloudLogging output.
diff --git a/api/observability/v1/zz_generated.deepcopy.go b/api/observability/v1/zz_generated.deepcopy.go
diff --git a/config/crd/bases/observability.openshift.io_clusterlogforwarders.yaml b/config/crd/bases/observability.openshift.io_clusterlogforwarders.yaml
@@ -2193,6 +2193,12 @@ spec:
                               - secretName
                               type: object
                           type: object
+                        headers:
+                          additionalProperties:
+                            type: string
+                          description: Headers specify optional headers to be sent
+                            with the request
+                          type: object
                         index:
                           description: |-
                             Index is the index for the logs. This supports template syntax to allow dynamic per-event values.
diff --git a/docs/reference/operator/api_observability_v1.adoc b/docs/reference/operator/api_observability_v1.adoc
@@ -2051,6 +2051,8 @@ The &#39;username@password&#39; part of `url` is ignored.
 
 |authentication|object|  Authentication sets credentials for authenticating the requests.
 
+|headers|object|  Headers specify optional headers to be sent with the request
+
 |index|string|  Index is the index for the logs. This supports template syntax to allow dynamic per-event values.
 
 The Index can be a combination of static and dynamic values consisting of field paths followed by `||` followed by another field path or a static value.
@@ -2160,6 +2162,10 @@ Type:: object
 
 |======================
 
+=== .spec.outputs[].elasticsearch.headers
+
+Type:: object
+
 === .spec.outputs[].elasticsearch.tuning
 
 Type:: object
diff --git a/internal/generator/vector/output/elasticsearch/elasticsearch.go b/internal/generator/vector/output/elasticsearch/elasticsearch.go
@@ -81,7 +81,7 @@ if exists(.kubernetes.event.metadata.uid) {
 		common.NewAcknowledgments(id, strategy),
 		common.NewBatch(id, strategy),
 		common.NewBuffer(id, strategy),
-		common.NewRequest(id, strategy),
+		Request(id, o.Elasticsearch, strategy),
 		tls.New(id, o.TLS, secrets, op, Option{Name: URL, Value: o.Elasticsearch.URL}),
 	)
 
@@ -110,3 +110,11 @@ func Output(id string, o obs.OutputSpec, inputs []string, index string, secrets
 	}
 	return &es
 }
+
+func Request(id string, o *obs.Elasticsearch, strategy common.ConfigStrategy) *common.Request {
+	req := common.NewRequest(id, strategy)
+	if len(o.Headers) != 0 {
+		req.SetHeaders(o.Headers)
+	}
+	return req
+}
diff --git a/internal/generator/vector/output/elasticsearch/elasticsearch_test.go b/internal/generator/vector/output/elasticsearch/elasticsearch_test.go
@@ -145,5 +145,10 @@ var _ = Describe("Generate Vector config", func() {
 				BaseOutputTuningSpec: *baseTune,
 			}
 		}, true, framework.NoOptions, "es_with_tune.toml"),
+		Entry("with headers", func(spec *obs.OutputSpec) {
+			spec.Elasticsearch.Headers = map[string]string{
+				"Key": "Value",
+			}
+		}, true, framework.NoOptions, "es_with_headers.toml"),
 	)
 })
diff --git a/internal/generator/vector/output/elasticsearch/es_with_headers.toml b/internal/generator/vector/output/elasticsearch/es_with_headers.toml
@@ -0,0 +1,26 @@
+# Elasticsearch Index
+[transforms.es_1_index]
+type = "remap"
+inputs = ["application"]
+source = '''
+._internal.es_1_index = to_string!(._internal.log_type||"none")
+'''
+
+[sinks.es_1]
+type = "elasticsearch"
+inputs = ["es_1_index"]
+endpoints = ["https://es.svc.infra.cluster:9200"]
+bulk.index = "{{ _internal.es_1_index }}"
+bulk.action = "create"
+api_version = "v8"
+
+[sinks.es_1.encoding]
+except_fields = ["_internal"]
+
+[sinks.es_1.request]
+headers = {"Key"="Value"}
+
+[sinks.es_1.auth]
+strategy = "basic"
+user = "SECRET[kubernetes_secret.es-1/username]"
+password = "SECRET[kubernetes_secret.es-1/password]"
diff --git a/internal/validations/observability/outputs/validate.go b/internal/validations/observability/outputs/validate.go
@@ -29,6 +29,8 @@ func Validate(context internalcontext.ForwarderContext) {
 			messages = append(messages, validateHttpContentTypeHeaders(out)...)
 		case obs.OutputTypeLokiStack, obs.OutputTypeOTLP:
 			messages = append(messages, ValidateTechPreviewAnnotation(out, context)...)
+		case obs.OutputTypeElasticsearch:
+			messages = append(messages, validateElasticsearchHeaders(out)...)
 		}
 		// Set condition
 		if len(messages) > 0 {
diff --git a/internal/validations/observability/outputs/validate_elasticsearch_headers.go b/internal/validations/observability/outputs/validate_elasticsearch_headers.go
@@ -0,0 +1,26 @@
+package outputs
+
+import (
+	"fmt"
+	log "github.com/ViaQ/logerr/v2/log/static"
+	obs "github.com/openshift/cluster-logging-operator/api/observability/v1"
+	"strings"
+)
+
+// validateElasticsearchHeaders will validate Elasticsearch custom headers
+// it's not allowed to pass "Authorization" and "Content-Type" headers
+func validateElasticsearchHeaders(output obs.OutputSpec) (results []string) {
+	if output.Type == obs.OutputTypeElasticsearch && output.Elasticsearch != nil && len(output.Elasticsearch.Headers) > 0 {
+		var invalidHeaders []string
+		for headerName := range output.Elasticsearch.Headers {
+			if strings.ToLower(headerName) == "authorization" || strings.ToLower(headerName) == "content-type" {
+				invalidHeaders = append(invalidHeaders, headerName)
+			}
+		}
+		if len(invalidHeaders) > 0 {
+			log.V(3).Info("validateElasticsearchHeaders failed", "reason", "invalid headers found: ", strings.Join(invalidHeaders, ","))
+			results = append(results, fmt.Sprintf("invalid headers found: %s", strings.Join(invalidHeaders, ",")))
+		}
+	}
+	return results
+}
diff --git a/internal/validations/observability/outputs/validate_elasticsearch_headers_test.go b/internal/validations/observability/outputs/validate_elasticsearch_headers_test.go
@@ -0,0 +1,55 @@
+package outputs
+
+import (
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	"github.com/openshift/cluster-logging-operator/api/observability/v1"
+)
+
+var _ = Describe("[internal][validations] ClusterLogForwarder will validate headers in Elasticsearch Output", func() {
+	var (
+		es   *v1.Elasticsearch
+		spec v1.OutputSpec
+	)
+	BeforeEach(func() {
+		es = &v1.Elasticsearch{}
+		spec = v1.OutputSpec{
+			Name:          "esOutput",
+			Type:          v1.OutputTypeElasticsearch,
+			Elasticsearch: es,
+		}
+	})
+
+	Context("#validateElasticsearchHeaders", func() {
+
+		It("should pass validation with empty headers", func() {
+			Expect(validateElasticsearchHeaders(spec)).To(BeEmpty())
+		})
+		It("should pass validation when no invalid headers set", func() {
+			spec.Elasticsearch.Headers = map[string]string{
+				"Accept": "application/json",
+			}
+			Expect(validateElasticsearchHeaders(spec)).To(BeEmpty())
+		})
+		It("should fail validation when the Content-Type header is set", func() {
+			spec.Elasticsearch.Headers = map[string]string{
+				"Content-Type": "application/json",
+			}
+			Expect(validateElasticsearchHeaders(spec)).ToNot(BeEmpty())
+		})
+		It("should fail validation when the Authorization header is set", func() {
+			spec.Elasticsearch.Headers = map[string]string{
+				"Authorization": "test",
+			}
+			Expect(validateElasticsearchHeaders(spec)).ToNot(BeEmpty())
+		})
+		It("should pass validation when no Elasticsearch Output", func() {
+			spec = v1.OutputSpec{
+				Name:          "esOutput",
+				Type:          v1.OutputTypeElasticsearch,
+				Elasticsearch: &v1.Elasticsearch{},
+			}
+			Expect(validateElasticsearchHeaders(spec)).To(BeEmpty())
+		})
+	})
+})
diff --git a/test/framework/functional/framework.go b/test/framework/functional/framework.go
@@ -427,8 +427,14 @@ func (f *CollectorFunctionalFramework) addOutputContainers(b *runtime.PodBuilder
 				return err
 			}
 		case obs.OutputTypeElasticsearch:
-			if err := f.AddESOutput(ElasticsearchVersion(output.Elasticsearch.Version), b, output, nil); err != nil {
-				return err
+			if len(output.Elasticsearch.Headers) == 0 {
+				if err := f.AddESOutput(ElasticsearchVersion(output.Elasticsearch.Version), b, output, nil); err != nil {
+					return err
+				}
+			} else {
+				if err := f.AddVLOutput(b, output, nil); err != nil {
+					return err
+				}
 			}
 		case obs.OutputTypeHTTP:
 			if err := f.AddVectorHttpOutput(b, output); err != nil {
diff --git a/test/framework/functional/output_victorialogs.go b/test/framework/functional/output_victorialogs.go
@@ -0,0 +1,81 @@
+package functional
+
+import (
+	"bufio"
+	"context"
+	"fmt"
+	"strings"
+
+	obs "github.com/openshift/cluster-logging-operator/api/observability/v1"
+	"k8s.io/apimachinery/pkg/util/wait"
+
+	"github.com/openshift/cluster-logging-operator/internal/generator/url"
+	"github.com/openshift/cluster-logging-operator/internal/runtime"
+
+	log "github.com/ViaQ/logerr/v2/log/static"
+)
+
+func (f *CollectorFunctionalFramework) AddVLOutput(b *runtime.PodBuilder, output obs.OutputSpec, args []string) error {
+	log.V(2).Info("Adding output for victorialogs", "name", output.Name)
+	name := strings.ToLower(output.Name)
+
+	esURL, err := url.Parse(output.Elasticsearch.URL)
+	if err != nil {
+		return err
+	}
+
+	log.V(2).Info("Adding container", "name", name)
+	log.V(2).Info("Adding VictoriaLogs output container", "name", obs.OutputTypeElasticsearch)
+
+	cmdArgs := []string{
+		"-httpListenAddr=:" + esURL.Port(),
+		"-storageDataPath=/tmp/logs",
+	}
+	if len(args) > 0 {
+		cmdArgs = append(cmdArgs, args...)
+	}
+
+	b.AddContainer(name, "quay.io/victoriametrics/victoria-logs:v1.34.0").
+		AddRunAsUser(2000).
+		WithCmdArgs(cmdArgs).
+		End().
+		AddContainer(fmt.Sprintf("%s-sidecar", name), "quay.io/curl/curl:8.16.0").
+		WithCmd([]string{"/bin/sleep", "infinity"}).
+		AddRunAsUser(2000).
+		End()
+	return nil
+}
+
+func (f *CollectorFunctionalFramework) GetLogsFromVL(outputName string, headers map[string]string, options ...Option) (results []string, err error) {
+	port := "9428"
+	if found, o := OptionsInclude("port", options); found {
+		port = o.Value
+	}
+
+	pod := runtime.NewPod(f.Test.NS.Name, outputName)
+	if err = f.Test.Get(pod); err != nil {
+		pod = f.Pod
+	}
+
+	err = wait.PollUntilContextTimeout(context.TODO(), defaultRetryInterval, maxDuration, true, func(cxt context.Context) (done bool, err error) {
+		curlArgs := `-X GET -H 'Content-Type: application/json'`
+		for k, v := range headers {
+			curlArgs += fmt.Sprintf(" -H '%s: %s'", k, v)
+		}
+		cmd := fmt.Sprintf(`curl localhost:%s/select/logsql/query?query='*' %s`, port, curlArgs)
+		var result string
+		result, err = f.RunCommandInPod(pod, fmt.Sprintf("%s-sidecar", outputName), "sh", "-c", cmd)
+		if result != "" && err == nil {
+			scanner := bufio.NewScanner(strings.NewReader(result))
+			log.V(2).Info("results", "response", results)
+			for scanner.Scan() {
+				results = append(results, scanner.Text())
+			}
+			return true, nil
+		}
+		log.V(4).Info("Polling from VictoriaLogs", "err", err, "result", result)
+		return false, nil
+	})
+	log.V(4).Info("Returning", "logs", results)
+	return results, err
+}
diff --git a/test/framework/functional/read.go b/test/framework/functional/read.go
@@ -120,14 +120,26 @@ func (f *CollectorFunctionalFramework) ReadLogsFrom(outputName, sourceType strin
 			return f.ReadApplicationLogsFromKafka(sourceType, "localhost:9092", container)
 		}
 	case obs.OutputTypeElasticsearch:
-		readLogs = func() ([]string, error) {
-			option := Option{"port", "9200"}
-			if outputSpec.Elasticsearch != nil {
-				if esurl, err := url.Parse(outputSpec.Elasticsearch.URL); err == nil {
-					option.Value = esurl.Port()
+		if len(outputSpec.Elasticsearch.Headers) > 0 {
+			readLogs = func() ([]string, error) {
+				option := Option{"port", "9428"}
+				if outputSpec.Elasticsearch != nil {
+					if esurl, err := url.Parse(outputSpec.Elasticsearch.URL); err == nil {
+						option.Value = esurl.Port()
+					}
+				}
+				return f.GetLogsFromVL(outputName, outputSpec.Elasticsearch.Headers, option)
+			}
+		} else {
+			readLogs = func() ([]string, error) {
+				option := Option{"port", "9200"}
+				if outputSpec.Elasticsearch != nil {
+					if esurl, err := url.Parse(outputSpec.Elasticsearch.URL); err == nil {
+						option.Value = esurl.Port()
+					}
 				}
+				return f.GetLogsFromElasticSearch(outputName, sourceType, option)
 			}
-			return f.GetLogsFromElasticSearch(outputName, sourceType, option)
 		}
 	default:
 		readLogs = func() ([]string, error) {
diff --git a/test/functional/outputs/elasticsearch/forward_to_victorialogs_test.go b/test/functional/outputs/elasticsearch/forward_to_victorialogs_test.go

Original file line number	Diff line number	Diff line change
`@@ -81,7 +81,7 @@ if exists(.kubernetes.event.metadata.uid) {`
`81`	`81`	`common.NewAcknowledgments(id, strategy),`
`82`	`82`	`common.NewBatch(id, strategy),`
`83`	`83`	`common.NewBuffer(id, strategy),`
`84`		`- common.NewRequest(id, strategy),`
	`84`	`+ Request(id, o.Elasticsearch, strategy),`
`85`	`85`	`tls.New(id, o.TLS, secrets, op, Option{Name: URL, Value: o.Elasticsearch.URL}),`
`86`	`86`	`)`
`87`	`87`
`@@ -110,3 +110,11 @@ func Output(id string, o obs.OutputSpec, inputs []string, index string, secrets`
`110`	`110`	`}`
`111`	`111`	`return &es`
`112`	`112`	`}`
	`113`	`+`
	`114`	`+func Request(id string, o obs.Elasticsearch, strategy common.ConfigStrategy) common.Request {`
	`115`	`+ req := common.NewRequest(id, strategy)`
	`116`	`+ if len(o.Headers) != 0 {`
	`117`	`+ req.SetHeaders(o.Headers)`
	`118`	`+ }`
	`119`	`+ return req`
	`120`	`+}`
Original file line number	Diff line number	Diff line change
`@@ -145,5 +145,10 @@ var _ = Describe("Generate Vector config", func() {`
`145`	`145`	`BaseOutputTuningSpec: *baseTune,`
`146`	`146`	`}`
`147`	`147`	`}, true, framework.NoOptions, "es_with_tune.toml"),`
	`148`	`+ Entry("with headers", func(spec *obs.OutputSpec) {`
	`149`	`+ spec.Elasticsearch.Headers = map[string]string{`
	`150`	`+ "Key": "Value",`
	`151`	`+ }`
	`152`	`+ }, true, framework.NoOptions, "es_with_headers.toml"),`
`148`	`153`	`)`
`149`	`154`	`})`
Original file line number	Diff line number	Diff line change
`@@ -29,6 +29,8 @@ func Validate(context internalcontext.ForwarderContext) {`
`29`	`29`	`messages = append(messages, validateHttpContentTypeHeaders(out)...)`
`30`	`30`	`case obs.OutputTypeLokiStack, obs.OutputTypeOTLP:`
`31`	`31`	`messages = append(messages, ValidateTechPreviewAnnotation(out, context)...)`
	`32`	`+ case obs.OutputTypeElasticsearch:`
	`33`	`+ messages = append(messages, validateElasticsearchHeaders(out)...)`
`32`	`34`	`}`
`33`	`35`	`// Set condition`
`34`	`36`	`if len(messages) > 0 {`