NO-JIRA: support custom headers for Elasticsearch output

api/observability/v1/output_types.go

@@ -504,6 +504,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.

config/crd/bases/observability.openshift.io_clusterlogforwarders.yaml

@@ -2125,6 +2125,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.

docs/reference/operator/api_logging_v1alpha1.adoc

@@ -81,6 +81,10 @@ Type:: array
 |name|string|  Name must match the name of one entry in pod.spec.resourceClaims of
 the Pod where this field is used. It makes that resource available
 inside a container.
+|request|string|  *(optional)* Request is the name chosen for a request in the referenced claim.
+If empty, everything from the claim is made available, otherwise
+only the result of this request.
+
 |======================
 
 === .spec.resources.limits

docs/reference/operator/api_observability_v1.adoc

@@ -1276,7 +1276,7 @@ NOTE2: If this filter is used in a pipeline with GoogleCloudLogging, `.hostname`
 
 === .spec.filters[].prune.in[]
 
-FieldPath represents a path to find a value for a given field.  The format must a value that can be converted to a
+FieldPath represents a path to find a value for a given field.  The format must be a value that can be converted to a
 valid collector configuration. It is a dot delimited path to a field in the log record. It must start with a `.`.
 The path can contain alphanumeric characters and underscores (a-zA-Z0-9_).
 If segments contain characters outside of this range, the segment must be quoted.
@@ -1286,7 +1286,7 @@ Type:: array
 
 === .spec.filters[].prune.notIn[]
 
-FieldPath represents a path to find a value for a given field.  The format must a value that can be converted to a
+FieldPath represents a path to find a value for a given field.  The format must be a value that can be converted to a
 valid collector configuration. It is a dot delimited path to a field in the log record. It must start with a `.`.
 The path can contain alphanumeric characters and underscores (a-zA-Z0-9_).
 If segments contain characters outside of this range, the segment must be quoted.
@@ -2020,6 +2020,8 @@ The 'username@password' 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.
@@ -2129,6 +2131,10 @@ Type:: object
 
 |======================
 
+=== .spec.outputs[].elasticsearch.headers
+
+Type:: object
+
 === .spec.outputs[].elasticsearch.tuning
 
 Type:: object
@@ -3147,6 +3153,39 @@ Example:
 
 3. foo.{.bar.baz||.qux.quux.corge||.grault||"nil"}-waldo.fred{.plugh||"none"}
 
+|indexedFields|array|  IndexedFields are the list of fields to be indexed by Splunk, increase storage usage, they should be used sparingly and only for high-value fields that provide significant search benefits.
+Nested fields are flattened into top-level fields.
+Field paths are joined using dot notation, and unsupported characters are replaced with underscores (_).
+Non-string values are automatically converted to strings (e.g., 3 → "3", true → "true").
+Object values serialized as JSON strings (e.g., { status: 200 } → "{\"status\":200}").
+
+Examples: [`.kubernetes`, `.log_type`, '.kubernetes.labels.foobar', `.kubernetes.labels."foo-bar/baz"`]
+|payloadKey|string|  PayloadKey specifies record field to use as payload.
+The PayloadKey must be a single field path.
+
+Field paths must only contain alphanumeric and underscores. Any field with other characters must be quoted.
+
+By default, payloadKey is not set, which means the complete log record is forwarded as the payload.
+Use payloadKey carefully. Selecting a single field as the payload may cause other important information in the
+log to be dropped, potentially leading to inconsistent or incomplete log events.
+
+Examples: `.kubernetes`, `.log_type`, '.kubernetes.labels.foobar', `.kubernetes.labels."foo-bar/baz"`
+
+|source|string|  Source identifies the origin of a log event.
+The Source can be a combination of static and dynamic values consisting of field paths followed by `||` followed by another field path or a static value.
+A dynamic value is encased in single curly brackets `{}` and MUST end with a static fallback value separated with `||`.
+Static values can only contain alphanumeric characters along with dashes, underscores, dots and forward slashes.
+If not specified will be detected according to .log_source and .log_type value.
+Details see in: docs/features/logforwarding/outputs/splunk-forwarding.adoc
+
+Example:
+
+1. foo-{.bar||"none"}
+
+2. {.foo||.bar||"missing"}
+
+3. foo.{.bar.baz||.qux.quux.corge||.grault||"nil"}-waldo.fred{.plugh||"none"}
+
 |tuning|object|  Tuning specs tuning for the output
 
 |======================
@@ -3181,6 +3220,16 @@ Type:: object
 
 |======================
 
+=== .spec.outputs[].splunk.indexedFields[]
+
+FieldPath represents a path to find a value for a given field.  The format must be a value that can be converted to a
+valid collector configuration. It is a dot delimited path to a field in the log record. It must start with a `.`.
+The path can contain alphanumeric characters and underscores (a-zA-Z0-9_).
+If segments contain characters outside of this range, the segment must be quoted.
+Examples: `.kubernetes.namespace_name`, `.log_type`, '.kubernetes.labels.foobar', `.kubernetes.labels."foo-bar/baz"`
+
+Type:: array
+
 === .spec.outputs[].splunk.tuning
 
 Type:: object
@@ -3255,6 +3304,15 @@ uucp cron authpriv ftp ntp security console solaris-cron
 
 local0 local1 local2 local3 local4 local5 local6 local7
 
+The Facility can be a combination of static and dynamic values consisting of field paths followed by `||` followed by another field path or a static value.
+A dynamic value is encased in single curly brackets `{}` and MUST end with a static fallback value separated with `||`.
+
+Static values can only contain alphanumeric characters along with dashes, underscores, dots and forward slashes.
+
+Example:
+
+1. {.foo||"user"}
+
 |msgId|string|  MsgId is MSGID part of the syslog-msg header. This supports template syntax to allow dynamic per-event values.
 
 The MsgId can be a combination of static and dynamic values consisting of field paths followed by `||` followed by another field path or a static value.
@@ -3321,6 +3379,15 @@ The value can be a decimal integer or one of these case-insensitive keywords:
 
 Emergency Alert Critical Error Warning Notice Informational Debug
 
+The Severity can be a combination of static and dynamic values consisting of field paths followed by `||` followed by another field path or a static value.
+A dynamic value is encased in single curly brackets `{}` and MUST end with a static fallback value separated with `||`.
+
+Static values can only contain alphanumeric characters along with dashes, underscores, dots and forward slashes.
+
+Example:
+
+1. {.foo||"Error"}
+
 |tuning|object|  Tuning specs tuning for the output
 
 |url|string|  An absolute URL, with a scheme. Valid schemes are: `tcp`, `tls`, `udp`

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
+}

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"),
 	)
 })

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]"