From 8c57a9c85d9a50b96060237b930ec39d149acc2c Mon Sep 17 00:00:00 2001 From: Calvin Lee Date: Tue, 1 Sep 2026 19:51:16 -0400 Subject: [PATCH] feat(elasticsearch): enhance Elasticsearch output to support multiple endpoints --- .gitignore | 2 + api/observability/v1/output_types.go | 32 ++++++- api/observability/v1/zz_generated.deepcopy.go | 6 +- ...cluster-logging.clusterserviceversion.yaml | 9 +- ...ity.openshift.io_clusterlogforwarders.yaml | 24 ++++- ...ity.openshift.io_clusterlogforwarders.yaml | 24 ++++- ...cluster-logging.clusterserviceversion.yaml | 7 ++ .../operator/api_observability_v1.adoc | 90 +++++++++++++------ .../vector/api/sinks/elasticsearch_sink.go | 4 +- .../output/elasticsearch/elasticsearch.go | 13 ++- .../elasticsearch/elasticsearch_test.go | 21 ++++- .../es_with_multi_endpoints.toml | 19 ++++ .../es_with_url_and_endpoints.toml | 19 ++++ internal/network/network_policy_test.go | 8 +- internal/network/ports.go | 7 +- internal/network/ports_test.go | 42 +++++++-- .../outputs/validate_url_to_output_tls.go | 43 +++++---- .../validate_url_to_output_tls_test.go | 32 ++++++- .../security/container_security_test.go | 4 +- .../functional/output_elasticsearch.go | 26 ++++++ .../forward_to_elasticsearch_test.go | 78 ++++++++++++++++ .../outputs/multiple/multiple_test.go | 4 +- .../observability/cluster_log_forwarder.go | 4 +- 23 files changed, 427 insertions(+), 91 deletions(-) create mode 100644 internal/generator/vector/output/elasticsearch/es_with_multi_endpoints.toml create mode 100644 internal/generator/vector/output/elasticsearch/es_with_url_and_endpoints.toml diff --git a/.gitignore b/.gitignore index 3f9bb027ac..b5d3c4af2c 100644 --- a/.gitignore +++ b/.gitignore @@ -90,3 +90,5 @@ tags # End of https://www.gitignore.io/api/go,vim,emacs,visualstudiocode .cache/ .target/ + +.claude/settings.local.json \ No newline at end of file diff --git a/api/observability/v1/output_types.go b/api/observability/v1/output_types.go index ef971ef478..c0b4588eea 100644 --- a/api/observability/v1/output_types.go +++ b/api/observability/v1/output_types.go @@ -508,7 +508,7 @@ type Cloudwatch struct { // The 'username@password' part of `url` is ignored. // // +kubebuilder:validation:Optional - // +kubebuilder:validation:XValidation:rule="self == '' || isURL(self)", message="invalid URL" + // +kubebuilder:validation:XValidation:rule="self == '' || isURL(self)", message="invalid URL" // +operator-sdk:csv:customresourcedefinitions:type=spec,displayName="Destination URL",xDescriptors={"urn:alm:descriptor:com.tectonic.ui:text"} URL string `json:"url,omitempty"` @@ -676,8 +676,32 @@ type ElasticsearchTuningSpec struct { Compression string `json:"compression,omitempty"` } +// EndpointURL is a URL to an Elasticsearch endpoint. +// +kubebuilder:validation:XValidation:rule="isURL(self)", message="invalid URL" +type EndpointURL string + +// Elasticsearch provides optional extra properties for `type: elasticsearch` +// +// +kubebuilder:validation:XValidation:rule="has(self.url) || self.endpoints.size() > 0", message="URL or endpoints required" type Elasticsearch struct { - URLSpec `json:",inline"` + + // URL to send log records to. + // Basic TLS is enabled if the URL scheme requires it (for example 'https' or 'tls'). + // The 'username@password' part of `url` is ignored. + // + // +kubebuilder:validation:Optional + // +kubebuilder:validation:XValidation:rule="isURL(self)", message="invalid URL" + // +operator-sdk:csv:customresourcedefinitions:type=spec,displayName="Destination URL",xDescriptors={"urn:alm:descriptor:com.tectonic.ui:text"} + URL string `json:"url,omitempty"` + + // Endpoints is a list of Elasticsearch endpoints to send log records to. + // Vector distributes events across endpoints using load balancing with + // automatic failover. When both URL and Endpoints are provided, URL is + // prepended to the Endpoints list. + // + // +kubebuilder:validation:Optional + // +operator-sdk:csv:customresourcedefinitions:type=spec,displayName="Elasticsearch Endpoints" + Endpoints []EndpointURL `json:"endpoints,omitempty"` // Authentication sets credentials for authenticating the requests. // @@ -887,7 +911,7 @@ type HTTP struct { // ProxyURL URL of a HTTP or HTTPS proxy to be used instead of direct connection. // // +kubebuilder:validation:Optional - // +kubebuilder:validation:XValidation:rule="self == '' || isURL(self)", message="invalid URL" + // +kubebuilder:validation:XValidation:rule="self == '' || isURL(self)", message="invalid URL" ProxyURL string `json:"proxyURL,omitempty"` // Format defines data format used to send data to remote destination. @@ -1257,7 +1281,7 @@ type Loki struct { // ProxyURL URL of a HTTP or HTTPS proxy to be used instead of direct connection. // // +kubebuilder:validation:Optional - // +kubebuilder:validation:XValidation:rule="self == '' || isURL(self)", message="invalid URL" + // +kubebuilder:validation:XValidation:rule="self == '' || isURL(self)", message="invalid URL" ProxyURL string `json:"proxyURL,omitempty"` } diff --git a/api/observability/v1/zz_generated.deepcopy.go b/api/observability/v1/zz_generated.deepcopy.go index 0321bc5746..b290098e2c 100644 --- a/api/observability/v1/zz_generated.deepcopy.go +++ b/api/observability/v1/zz_generated.deepcopy.go @@ -707,7 +707,11 @@ func (in *DropTest) DeepCopy() *DropTest { // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *Elasticsearch) DeepCopyInto(out *Elasticsearch) { *out = *in - out.URLSpec = in.URLSpec + if in.Endpoints != nil { + in, out := &in.Endpoints, &out.Endpoints + *out = make([]EndpointURL, len(*in)) + copy(*out, *in) + } if in.Authentication != nil { in, out := &in.Authentication, &out.Authentication *out = new(HTTPAuthentication) diff --git a/bundle/manifests/cluster-logging.clusterserviceversion.yaml b/bundle/manifests/cluster-logging.clusterserviceversion.yaml index 97bc0be0ba..faa9cac747 100644 --- a/bundle/manifests/cluster-logging.clusterserviceversion.yaml +++ b/bundle/manifests/cluster-logging.clusterserviceversion.yaml @@ -82,7 +82,7 @@ metadata: categories: OpenShift Optional, Logging & Tracing, Observability certified: "false" containerImage: quay.io/openshift-logging/cluster-logging-operator:latest - createdAt: "2026-07-15T11:39:59Z" + createdAt: "2026-09-01T18:57:04Z" description: The Red Hat OpenShift Logging Operator for OCP provides a means for configuring and managing log collection and forwarding. features.operators.openshift.io/cnf: "false" @@ -934,6 +934,13 @@ spec: path: outputs[0].elasticsearch.authentication.username.secretName x-descriptors: - urn:alm:descriptor:com.tectonic.ui:text + - description: |- + Endpoints is a list of Elasticsearch endpoints to send log records to. + Vector distributes events across endpoints using load balancing with + automatic failover. When both URL and Endpoints are provided, URL is + prepended to the Endpoints list. + displayName: Elasticsearch Endpoints + path: outputs[0].elasticsearch.endpoints - description: Headers specify optional headers to be sent with the request displayName: Headers path: outputs[0].elasticsearch.headers diff --git a/bundle/manifests/observability.openshift.io_clusterlogforwarders.yaml b/bundle/manifests/observability.openshift.io_clusterlogforwarders.yaml index 3f4f5c35f1..ffd5867448 100644 --- a/bundle/manifests/observability.openshift.io_clusterlogforwarders.yaml +++ b/bundle/manifests/observability.openshift.io_clusterlogforwarders.yaml @@ -2393,7 +2393,7 @@ spec: type: string x-kubernetes-validations: - message: invalid URL - rule: self == '' || isURL(self) + rule: self == '' || isURL(self) required: - authentication - groupName @@ -2476,6 +2476,20 @@ spec: - secretName type: object type: object + endpoints: + description: |- + Endpoints is a list of Elasticsearch endpoints to send log records to. + Vector distributes events across endpoints using load balancing with + automatic failover. When both URL and Endpoints are provided, URL is + prepended to the Endpoints list. + items: + description: EndpointURL is a URL to an Elasticsearch + endpoint. + type: string + x-kubernetes-validations: + - message: invalid URL + rule: isURL(self) + type: array headers: additionalProperties: type: string @@ -2567,9 +2581,11 @@ spec: type: integer required: - index - - url - version type: object + x-kubernetes-validations: + - message: URL or endpoints required + rule: has(self.url) || self.endpoints.size() > 0 googleCloudLogging: description: GoogleCloudLogging configures forwarding log events to GCP (formally Stackdriver) Operations @@ -2822,7 +2838,7 @@ spec: type: string x-kubernetes-validations: - message: invalid URL - rule: self == '' || isURL(self) + rule: self == '' || isURL(self) timeout: description: Timeout specifies the Http request timeout in seconds. If not set, 10secs is used. @@ -3134,7 +3150,7 @@ spec: type: string x-kubernetes-validations: - message: invalid URL - rule: self == '' || isURL(self) + rule: self == '' || isURL(self) tenantKey: description: |- TenantKey is the tenant for the logs. This supports vector's template syntax to allow dynamic per-event values. diff --git a/config/crd/bases/observability.openshift.io_clusterlogforwarders.yaml b/config/crd/bases/observability.openshift.io_clusterlogforwarders.yaml index 3fe192785d..b436e3df26 100644 --- a/config/crd/bases/observability.openshift.io_clusterlogforwarders.yaml +++ b/config/crd/bases/observability.openshift.io_clusterlogforwarders.yaml @@ -2393,7 +2393,7 @@ spec: type: string x-kubernetes-validations: - message: invalid URL - rule: self == '' || isURL(self) + rule: self == '' || isURL(self) required: - authentication - groupName @@ -2476,6 +2476,20 @@ spec: - secretName type: object type: object + endpoints: + description: |- + Endpoints is a list of Elasticsearch endpoints to send log records to. + Vector distributes events across endpoints using load balancing with + automatic failover. When both URL and Endpoints are provided, URL is + prepended to the Endpoints list. + items: + description: EndpointURL is a URL to an Elasticsearch + endpoint. + type: string + x-kubernetes-validations: + - message: invalid URL + rule: isURL(self) + type: array headers: additionalProperties: type: string @@ -2567,9 +2581,11 @@ spec: type: integer required: - index - - url - version type: object + x-kubernetes-validations: + - message: URL or endpoints required + rule: has(self.url) || self.endpoints.size() > 0 googleCloudLogging: description: GoogleCloudLogging configures forwarding log events to GCP (formally Stackdriver) Operations @@ -2822,7 +2838,7 @@ spec: type: string x-kubernetes-validations: - message: invalid URL - rule: self == '' || isURL(self) + rule: self == '' || isURL(self) timeout: description: Timeout specifies the Http request timeout in seconds. If not set, 10secs is used. @@ -3134,7 +3150,7 @@ spec: type: string x-kubernetes-validations: - message: invalid URL - rule: self == '' || isURL(self) + rule: self == '' || isURL(self) tenantKey: description: |- TenantKey is the tenant for the logs. This supports vector's template syntax to allow dynamic per-event values. diff --git a/config/manifests/bases/cluster-logging.clusterserviceversion.yaml b/config/manifests/bases/cluster-logging.clusterserviceversion.yaml index 7411b3a0f3..60b1308837 100644 --- a/config/manifests/bases/cluster-logging.clusterserviceversion.yaml +++ b/config/manifests/bases/cluster-logging.clusterserviceversion.yaml @@ -857,6 +857,13 @@ spec: path: outputs[0].elasticsearch.authentication.username.secretName x-descriptors: - urn:alm:descriptor:com.tectonic.ui:text + - description: |- + Endpoints is a list of Elasticsearch endpoints to send log records to. + Vector distributes events across endpoints using load balancing with + automatic failover. When both URL and Endpoints are provided, URL is + prepended to the Endpoints list. + displayName: Elasticsearch Endpoints + path: outputs[0].elasticsearch.endpoints - description: Headers specify optional headers to be sent with the request displayName: Headers path: outputs[0].elasticsearch.headers diff --git a/docs/reference/operator/api_observability_v1.adoc b/docs/reference/operator/api_observability_v1.adoc index ee48a68d48..88f2fc2e41 100644 --- a/docs/reference/operator/api_observability_v1.adoc +++ b/docs/reference/operator/api_observability_v1.adoc @@ -1823,13 +1823,15 @@ Type:: object === .spec.outputs[].elasticsearch +Elasticsearch provides optional extra properties for `type: elasticsearch` + Type:: object [options="header"] |====================== |Property|Type|Description -|url|string| URL to send log records to. Basic TLS is enabled if the URL scheme requires it (for example 'https' or 'tls'). The 'username@password' part of `url` is ignored. |authentication|object| Authentication sets credentials for authenticating the requests. +|endpoints|array| Endpoints is a list of Elasticsearch endpoints to send log records to. Vector distributes events across endpoints using load balancing with automatic failover. When both URL and Endpoints are provided, URL is prepended to the Endpoints list. |headers|object| Headers specify optional headers to be sent with the request |index|string a| Index is the index for the logs. This supports template syntax to allow dynamic per-event values. @@ -1846,6 +1848,7 @@ Example: . foo.pass:[{.bar.baz\|\|.qux.quux.corge\|\|.grault\|\|"nil"}]-waldo.fredpass:[{.plugh\|\|"none"}] |tuning|object| Tuning specs tuning for the output +|url|string| URL to send log records to. Basic TLS is enabled if the URL scheme requires it (for example 'https' or 'tls'). The 'username@password' part of `url` is ignored. |version|int| Version specifies the API version of Elasticsearch to be used. Must be one of: 6-8 The value of '8' should be used when forwarding to Elasticsearch version v8 or greater. |====================== @@ -1915,6 +1918,12 @@ Type:: object |secretName|string| SecretName contains the name of the Secret containing the referenced value. |====================== +=== .spec.outputs[].elasticsearch.endpoints[] + +EndpointURL is a URL to an Elasticsearch endpoint. + +Type:: array + === .spec.outputs[].elasticsearch.headers Type:: object @@ -3126,20 +3135,26 @@ Type:: object |Property|Type|Description |custom|object a| *(optional)* custom is a user-defined TLS security profile. Be extremely careful using a custom -profile as invalid configurations can be catastrophic. An example custom profile -looks like this: +profile as invalid configurations can be catastrophic. +The supported groups list for this profile is empty by default. +An example custom profile looks like this: +minTLSVersion: VersionTLS11 ciphers: - ECDHE-ECDSA-CHACHA20-POLY1305 - ECDHE-RSA-CHACHA20-POLY1305 - ECDHE-RSA-AES128-GCM-SHA256 - ECDHE-ECDSA-AES128-GCM-SHA256 -minTLSVersion: VersionTLS11 |intermediate|object -a| *(optional)* intermediate is a TLS security profile based on: -https://wiki.mozilla.org/Security/Server_Side_TLS#Intermediate_compatibility_.28recommended.29 -and looks like this (yaml): +a| *(optional)* intermediate is a TLS profile for use when you do not need compatibility with +legacy clients and want to remain highly secure while being compatible with +most clients currently in use. +The supported groups list includes by default the following groups +in suggested preference order (ordering may not be honored by all implementations): +X25519MLKEM768, X25519, secp256r1, secp384r1. +This profile is equivalent to a Custom profile specified as: +minTLSVersion: VersionTLS12 ciphers: - TLS_AES_128_GCM_SHA256 @@ -3151,25 +3166,29 @@ ciphers: - ECDHE-RSA-AES256-GCM-SHA384 - ECDHE-ECDSA-CHACHA20-POLY1305 - ECDHE-RSA-CHACHA20-POLY1305 -- DHE-RSA-AES128-GCM-SHA256 -- DHE-RSA-AES256-GCM-SHA384 -minTLSVersion: VersionTLS12 |modern|object -a| *(optional)* modern is a TLS security profile based on: -https://wiki.mozilla.org/Security/Server_Side_TLS#Modern_compatibility -and looks like this (yaml): +a| *(optional)* modern is a TLS security profile for use with clients that support TLS 1.3 and +do not need backward compatibility for older clients. +The supported groups list includes by default the following groups +in suggested preference order (ordering may not be honored by all implementations): +X25519MLKEM768, X25519, secp256r1, secp384r1. +This profile is equivalent to a Custom profile specified as: +minTLSVersion: VersionTLS13 ciphers: - TLS_AES_128_GCM_SHA256 - TLS_AES_256_GCM_SHA384 - TLS_CHACHA20_POLY1305_SHA256 -minTLSVersion: VersionTLS13 |old|object -a| *(optional)* old is a TLS security profile based on: -https://wiki.mozilla.org/Security/Server_Side_TLS#Old_backward_compatibility -and looks like this (yaml): +a| *(optional)* old is a TLS profile for use when services need to be accessed by very old +clients or libraries and should be used only as a last resort. +The supported groups list includes by default the following groups +in suggested preference order (ordering may not be honored by all implementations): +X25519MLKEM768, X25519, secp256r1, secp384r1. +This profile is equivalent to a Custom profile specified as: +minTLSVersion: VersionTLS10 ciphers: - TLS_AES_128_GCM_SHA256 @@ -3181,9 +3200,6 @@ ciphers: - ECDHE-RSA-AES256-GCM-SHA384 - ECDHE-ECDSA-CHACHA20-POLY1305 - ECDHE-RSA-CHACHA20-POLY1305 -- DHE-RSA-AES128-GCM-SHA256 -- DHE-RSA-AES256-GCM-SHA384 -- DHE-RSA-CHACHA20-POLY1305 - ECDHE-ECDSA-AES128-SHA256 - ECDHE-RSA-AES128-SHA256 - ECDHE-ECDSA-AES128-SHA @@ -3192,8 +3208,6 @@ ciphers: - ECDHE-RSA-AES256-SHA384 - ECDHE-ECDSA-AES256-SHA - ECDHE-RSA-AES256-SHA -- DHE-RSA-AES128-SHA256 -- DHE-RSA-AES256-SHA256 - AES128-GCM-SHA256 - AES256-GCM-SHA384 - AES128-SHA256 @@ -3201,9 +3215,8 @@ ciphers: - AES128-SHA - AES256-SHA - DES-CBC3-SHA -minTLSVersion: VersionTLS10 -|type|string| *(optional)* type is one of Old, Intermediate, Modern or Custom. Custom provides the ability to specify individual TLS security profile parameters. Old, Intermediate and Modern are TLS security profiles based on: https://wiki.mozilla.org/Security/Server_Side_TLS#Recommended_configurations The profiles are intent based, so they may change over time as new ciphers are developed and existing ciphers are found to be insecure. Depending on precisely which ciphers are available to a process, the list may be reduced. Note that the Modern profile is currently not supported because it is not yet well adopted by common software libraries. +|type|string| *(optional)* type is one of Old, Intermediate, Modern or Custom. Custom provides the ability to specify individual TLS security profile parameters. The cipher and groups lists in these profiles are based on version 5.8 of the Mozilla Server Side TLS configuration guidelines. See: https://ssl-config.mozilla.org/guidelines/5.8.json The groups are listed in suggested preference order, with the most preferred group first. Note that not all platform components honor the ordering: Go-based components use Go's internal preference order and treat this list as a filter of allowed groups rather than an ordered preference. Note that X25519MLKEM768 is a post-quantum hybrid group that is not FIPS-approved and should be ignored by components running in FIPS mode. The profiles are intent based, so they may change over time as new ciphers are developed and existing ciphers are found to be insecure. Depending on precisely which ciphers are available to a process, the list may be reduced. |====================== === .spec.outputs[].tls.securityProfile.custom @@ -3215,13 +3228,32 @@ Type:: object |Property|Type|Description |ciphers|array a| ciphers is used to specify the cipher algorithms that are negotiated -during the TLS handshake. Operators may remove entries their operands -do not support. For example, to use DES-CBC3-SHA (yaml): +during the TLS handshake. Operators may remove entries that their operands +do not support. For example, to use only ECDHE-RSA-AES128-GCM-SHA256 (yaml): ciphers: -- DES-CBC3-SHA - -|minTLSVersion|string| minTLSVersion is used to specify the minimal version of the TLS protocol that is negotiated during the TLS handshake. For example, to use TLS versions 1.1, 1.2 and 1.3 (yaml): minTLSVersion: VersionTLS11 NOTE: currently the highest minTLSVersion allowed is VersionTLS12 +- ECDHE-RSA-AES128-GCM-SHA256 +TLS 1.3 cipher suites (e.g. TLS_AES_128_GCM_SHA256) are not configurable +and are always enabled when TLS 1.3 is negotiated. + +|groups|array +a| *(optional)* groups is an optional, ordered field used to specify the supported groups (formerly known as +elliptic curves) that are used during the TLS handshake. The order of the groups represents +a suggested preference, with the most preferred group first. Note that not all platform +components honor the ordering: Go-based components use Go's internal preference order and +treat this list as a filter of allowed groups rather than an ordered preference. +Operators may remove entries their operands do not support. +When omitted, this means no opinion and the platform is left to choose reasonable defaults which are +subject to change over time and may be different per platform component depending on the underlying TLS +libraries they use. If specified, the list must contain at least one and at most 7 groups, +and each group must be unique. +For example, to use X25519 and secp256r1 (yaml): +groups: + +- X25519 +- secp256r1 + +|minTLSVersion|string| minTLSVersion is used to specify the minimal version of the TLS protocol that is negotiated during the TLS handshake. For example, to use TLS versions 1.1, 1.2 and 1.3 (yaml): minTLSVersion: VersionTLS11 |====================== === .spec.outputs[].tls.securityProfile.intermediate diff --git a/internal/generator/vector/api/sinks/elasticsearch_sink.go b/internal/generator/vector/api/sinks/elasticsearch_sink.go index 4d52cd98bc..4512bd0053 100644 --- a/internal/generator/vector/api/sinks/elasticsearch_sink.go +++ b/internal/generator/vector/api/sinks/elasticsearch_sink.go @@ -18,12 +18,12 @@ type Elasticsearch struct { Proxy *Proxy `json:"proxy,omitempty" yaml:"proxy,omitempty" toml:"proxy,omitempty"` } -func NewElasticsearch(url string, init func(s *Elasticsearch), inputs ...string) (s *Elasticsearch) { +func NewElasticsearch(endpoints []string, init func(s *Elasticsearch), inputs ...string) (s *Elasticsearch) { sort.Strings(inputs) s = &Elasticsearch{ Type: types.SinkTypeElasticsearch, Inputs: inputs, - Endpoints: []string{url}, + Endpoints: endpoints, } if init != nil { init(s) diff --git a/internal/generator/vector/output/elasticsearch/elasticsearch.go b/internal/generator/vector/output/elasticsearch/elasticsearch.go index b9489b9e63..3b342d0622 100644 --- a/internal/generator/vector/output/elasticsearch/elasticsearch.go +++ b/internal/generator/vector/output/elasticsearch/elasticsearch.go @@ -19,6 +19,17 @@ import ( "github.com/openshift/cluster-logging-operator/internal/utils" ) +func mergeEndpoints(es *obs.Elasticsearch) []string { + var endpoints []string + if es.URL != "" { + endpoints = append(endpoints, es.URL) + } + for _, e := range es.Endpoints { + endpoints = append(endpoints, string(e)) + } + return endpoints +} + func New(id string, o *adapters.Output, inputs []string, secrets observability.Secrets, op utils.Options) (_ string, sink types.Sink, tfs api.Transforms) { componentID := helpers.MakeID(id, "index") tfs = api.Transforms{} @@ -31,7 +42,7 @@ if exists(.kubernetes.event.metadata.uid) { inputs = []string{addID} } tfs[componentID] = commontemplate.NewTemplateRemap(inputs, o.Elasticsearch.Index, componentID) - sink = sinks.NewElasticsearch(o.Elasticsearch.URL, func(s *sinks.Elasticsearch) { + sink = sinks.NewElasticsearch(mergeEndpoints(o.Elasticsearch), func(s *sinks.Elasticsearch) { s.Bulk = &sinks.Bulk{ Action: sinks.BulkActionCreate, Index: fmt.Sprintf("{{ _internal.%s }}", componentID), diff --git a/internal/generator/vector/output/elasticsearch/elasticsearch_test.go b/internal/generator/vector/output/elasticsearch/elasticsearch_test.go index 5874a3097a..74266e8eb7 100644 --- a/internal/generator/vector/output/elasticsearch/elasticsearch_test.go +++ b/internal/generator/vector/output/elasticsearch/elasticsearch_test.go @@ -49,9 +49,7 @@ var _ = Describe("Generate Vector config", func() { Type: obs.OutputTypeElasticsearch, Name: "es_1", Elasticsearch: &obs.Elasticsearch{ - URLSpec: obs.URLSpec{ - URL: "https://es.svc.infra.cluster:9200", - }, + URL: "https://es.svc.infra.cluster:9200", Index: `{.log_type||"none"}`, Authentication: &obs.HTTPAuthentication{ Username: &obs.SecretReference{ @@ -159,5 +157,22 @@ var _ = Describe("Generate Vector config", func() { "Key": "Value", } }, true, framework.NoOptions, "es_with_headers.toml"), + Entry("with multiple endpoints only", func(spec *obs.OutputSpec) { + spec.Elasticsearch.Authentication = nil + spec.Elasticsearch.URL = "" + spec.Elasticsearch.Endpoints = []obs.EndpointURL{ + "https://es1.example.com:9200", + "https://es2.example.com:9200", + "https://es3.example.com:9200", + } + }, false, framework.NoOptions, "es_with_multi_endpoints.toml"), + Entry("with url and endpoints combined", func(spec *obs.OutputSpec) { + spec.Elasticsearch.Authentication = nil + spec.Elasticsearch.URL = "https://es-primary.example.com:9200" + spec.Elasticsearch.Endpoints = []obs.EndpointURL{ + "https://es1.example.com:9200", + "https://es2.example.com:9200", + } + }, false, framework.NoOptions, "es_with_url_and_endpoints.toml"), ) }) diff --git a/internal/generator/vector/output/elasticsearch/es_with_multi_endpoints.toml b/internal/generator/vector/output/elasticsearch/es_with_multi_endpoints.toml new file mode 100644 index 0000000000..51ee7979d7 --- /dev/null +++ b/internal/generator/vector/output/elasticsearch/es_with_multi_endpoints.toml @@ -0,0 +1,19 @@ +[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://es1.example.com:9200", "https://es2.example.com:9200", "https://es3.example.com:9200"] +api_version = "v8" + +[sinks.es_1.bulk] +index = "{{ _internal.es_1_index }}" +action = "create" + +[sinks.es_1.encoding] +except_fields = ["_internal"] diff --git a/internal/generator/vector/output/elasticsearch/es_with_url_and_endpoints.toml b/internal/generator/vector/output/elasticsearch/es_with_url_and_endpoints.toml new file mode 100644 index 0000000000..2f1f61345b --- /dev/null +++ b/internal/generator/vector/output/elasticsearch/es_with_url_and_endpoints.toml @@ -0,0 +1,19 @@ +[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-primary.example.com:9200", "https://es1.example.com:9200", "https://es2.example.com:9200"] +api_version = "v8" + +[sinks.es_1.bulk] +index = "{{ _internal.es_1_index }}" +action = "create" + +[sinks.es_1.encoding] +except_fields = ["_internal"] diff --git a/internal/network/network_policy_test.go b/internal/network/network_policy_test.go index 220fb337fc..8992ab1778 100644 --- a/internal/network/network_policy_test.go +++ b/internal/network/network_policy_test.go @@ -177,9 +177,7 @@ var _ = Describe("Reconcile NetworkPolicy", func() { Name: "elasticsearch-output", Type: obsv1.OutputTypeElasticsearch, Elasticsearch: &obsv1.Elasticsearch{ - URLSpec: obsv1.URLSpec{ - URL: "https://elasticsearch.example.com:9200", - }, + URL: "https://elasticsearch.example.com:9200", }, }, { @@ -277,9 +275,7 @@ var _ = Describe("Reconcile NetworkPolicy", func() { Name: "elasticsearch-output", Type: obsv1.OutputTypeElasticsearch, Elasticsearch: &obsv1.Elasticsearch{ - URLSpec: obsv1.URLSpec{ - URL: "https://elasticsearch.example.com:9200", - }, + URL: "https://elasticsearch.example.com:9200", }, }, { diff --git a/internal/network/ports.go b/internal/network/ports.go index 19267537c5..a75b9abc52 100644 --- a/internal/network/ports.go +++ b/internal/network/ports.go @@ -88,7 +88,12 @@ func getPortProtocolFromOutputURLs(output obs.OutputSpec) []factory.PortProtocol switch output.Type { case obs.OutputTypeElasticsearch: if output.Elasticsearch != nil { - urlSlice = append(urlSlice, output.Elasticsearch.URL) + if output.Elasticsearch.URL != "" { + urlSlice = append(urlSlice, output.Elasticsearch.URL) + } + for _, e := range output.Elasticsearch.Endpoints { + urlSlice = append(urlSlice, string(e)) + } } case obs.OutputTypeSplunk: if output.Splunk != nil { diff --git a/internal/network/ports_test.go b/internal/network/ports_test.go index 798a7ccc96..c6cccb5137 100644 --- a/internal/network/ports_test.go +++ b/internal/network/ports_test.go @@ -125,7 +125,7 @@ var _ = Describe("Network Ports", func() { func(urlStr string, expectedPort int32) { output := obs.OutputSpec{ Type: obs.OutputTypeElasticsearch, - Elasticsearch: &obs.Elasticsearch{URLSpec: obs.URLSpec{URL: urlStr}}, + Elasticsearch: &obs.Elasticsearch{URL: urlStr}, } Expect(getPortProtocolFromOutputURLs(output)).To(Equal(makeTCPPorts(expectedPort))) }, @@ -137,6 +137,36 @@ var _ = Describe("Network Ports", func() { "http://es.example.com", constants.DefaultHTTPPort), ) + DescribeTable("Elasticsearch with multiple endpoints", + func(output obs.OutputSpec, expectedPorts []factory.PortProtocol) { + Expect(getPortProtocolFromOutputURLs(output)).To(ConsistOf(expectedPorts)) + }, + Entry("should extract ports from multiple endpoints", + obs.OutputSpec{ + Type: obs.OutputTypeElasticsearch, + Elasticsearch: &obs.Elasticsearch{ + Endpoints: []obs.EndpointURL{ + "https://es1.example.com:9200", + "https://es2.example.com:9300", + }, + }, + }, + makeTCPPorts(9200, 9300), + ), + Entry("should extract ports from url and endpoints combined", + obs.OutputSpec{ + Type: obs.OutputTypeElasticsearch, + Elasticsearch: &obs.Elasticsearch{ + URL: "https://es-primary.example.com:9200", + Endpoints: []obs.EndpointURL{ + "https://es1.example.com:9300", + }, + }, + }, + makeTCPPorts(9200, 9300), + ), + ) + DescribeTable("Splunk", func(urlStr string, expectedPort int32) { output := obs.OutputSpec{ @@ -362,7 +392,7 @@ var _ = Describe("Network Ports", func() { { Type: obs.OutputTypeElasticsearch, Elasticsearch: &obs.Elasticsearch{ - URLSpec: obs.URLSpec{URL: "https://es.example.com:9200"}, + URL: "https://es.example.com:9200", }, }, { @@ -388,13 +418,13 @@ var _ = Describe("Network Ports", func() { { Type: obs.OutputTypeElasticsearch, Elasticsearch: &obs.Elasticsearch{ - URLSpec: obs.URLSpec{URL: "https://es1.example.com:9200"}, + URL: "https://es1.example.com:9200", }, }, { Type: obs.OutputTypeElasticsearch, Elasticsearch: &obs.Elasticsearch{ - URLSpec: obs.URLSpec{URL: "https://es2.example.com:9200"}, + URL: "https://es2.example.com:9200", }, }, { @@ -413,7 +443,7 @@ var _ = Describe("Network Ports", func() { { Type: obs.OutputTypeElasticsearch, Elasticsearch: &obs.Elasticsearch{ - URLSpec: obs.URLSpec{URL: "https://es.example.com"}, + URL: "https://es.example.com", }, }, { @@ -443,7 +473,7 @@ var _ = Describe("Network Ports", func() { { Type: obs.OutputTypeElasticsearch, Elasticsearch: &obs.Elasticsearch{ - URLSpec: obs.URLSpec{URL: "https://es.example.com:9200"}, + URL: "https://es.example.com:9200", }, }, } diff --git a/internal/validations/observability/outputs/validate_url_to_output_tls.go b/internal/validations/observability/outputs/validate_url_to_output_tls.go index 00c5b52c7a..36fbfe0c6e 100644 --- a/internal/validations/observability/outputs/validate_url_to_output_tls.go +++ b/internal/validations/observability/outputs/validate_url_to_output_tls.go @@ -11,34 +11,43 @@ import ( // validateURLAccordingToTLS validate that if Output has TLS configuration Output URL scheme must be secure e.g. https, tls etc func validateURLAccordingToTLS(output obs.OutputSpec) (results []string) { - specURL := "" + var specURLs []string switch output.Type { case obs.OutputTypeCloudwatch: - specURL = output.Cloudwatch.URL + specURLs = append(specURLs, output.Cloudwatch.URL) case obs.OutputTypeElasticsearch: - specURL = output.Elasticsearch.URL + if output.Elasticsearch.URL != "" { + specURLs = append(specURLs, output.Elasticsearch.URL) + } + for _, e := range output.Elasticsearch.Endpoints { + specURLs = append(specURLs, string(e)) + } case obs.OutputTypeHTTP: - specURL = output.HTTP.URL + specURLs = append(specURLs, output.HTTP.URL) case obs.OutputTypeKafka: - specURL = output.Kafka.URL + specURLs = append(specURLs, output.Kafka.URL) case obs.OutputTypeLoki: - specURL = output.Loki.URL + specURLs = append(specURLs, output.Loki.URL) case obs.OutputTypeSplunk: - specURL = output.Splunk.URL + specURLs = append(specURLs, output.Splunk.URL) case obs.OutputTypeSyslog: - specURL = output.Syslog.URL + specURLs = append(specURLs, output.Syslog.URL) case obs.OutputTypeOTLP: - specURL = output.OTLP.URL + specURLs = append(specURLs, output.OTLP.URL) } - // some outputs not require to have output URL (e.g. Amazon CloudWatch or Google Cloud Logging) - if specURL != "" && output.TLS != nil { - u, _ := url.Parse(specURL) - scheme := strings.ToLower(u.Scheme) - if !url.IsTLSScheme(scheme) && (output.TLS.InsecureSkipVerify || output.TLS.TLSSecurityProfile != nil) { - log.V(3).Info("validateURLAccordingToTLS failed", "reason", "URL not secure but output has TLS configuration parameters", - "output URL", specURL, "output Name", output.Name) - results = append(results, fmt.Sprintf("URL scheme not secure: %v, but output has TLS configuration parameters", scheme)) + if output.TLS != nil { + for _, specURL := range specURLs { + if specURL == "" { + continue + } + u, _ := url.Parse(specURL) + scheme := strings.ToLower(u.Scheme) + if !url.IsTLSScheme(scheme) && (output.TLS.InsecureSkipVerify || output.TLS.TLSSecurityProfile != nil) { + log.V(3).Info("validateURLAccordingToTLS failed", "reason", "URL not secure but output has TLS configuration parameters", + "output URL", specURL, "output Name", output.Name) + results = append(results, fmt.Sprintf("URL scheme not secure: %v, but output has TLS configuration parameters", scheme)) + } } } return results diff --git a/internal/validations/observability/outputs/validate_url_to_output_tls_test.go b/internal/validations/observability/outputs/validate_url_to_output_tls_test.go index d051f4e25c..041c72604f 100644 --- a/internal/validations/observability/outputs/validate_url_to_output_tls_test.go +++ b/internal/validations/observability/outputs/validate_url_to_output_tls_test.go @@ -88,9 +88,7 @@ var _ = Describe("[internal][validations][observability][outputs] ClusterLogForw It("should pass validation when secure URL and exist TLS config: tls.TLSSecurityProfile", func() { spec.Type = obs.OutputTypeElasticsearch spec.Elasticsearch = &obs.Elasticsearch{ - URLSpec: obs.URLSpec{ - URL: "https://local.svc:514", - }, + URL: "https://local.svc:514", } spec.TLS = &obs.OutputTLSSpec{ @@ -100,6 +98,34 @@ var _ = Describe("[internal][validations][observability][outputs] ClusterLogForw } Expect(validateURLAccordingToTLS(spec)).To(BeEmpty()) }) + It("should fail validation when any endpoint has insecure URL with TLS config", func() { + spec.Type = obs.OutputTypeElasticsearch + spec.Elasticsearch = &obs.Elasticsearch{ + URL: "https://secure.svc:9200", + Endpoints: []obs.EndpointURL{ + "http://insecure.svc:9200", + }, + } + spec.TLS = &obs.OutputTLSSpec{ + InsecureSkipVerify: true, + } + Expect(validateURLAccordingToTLS(spec)).ToNot(BeEmpty()) + }) + It("should pass validation when all endpoints are secure with TLS config", func() { + spec.Type = obs.OutputTypeElasticsearch + spec.Elasticsearch = &obs.Elasticsearch{ + Endpoints: []obs.EndpointURL{ + "https://es1.svc:9200", + "https://es2.svc:9200", + }, + } + spec.TLS = &obs.OutputTLSSpec{ + TLSSecurityProfile: &configv1.TLSSecurityProfile{ + Type: configv1.TLSProfileOldType, + }, + } + Expect(validateURLAccordingToTLS(spec)).To(BeEmpty()) + }) It("should pass validation when URL is optional and TLS is spec'd", func() { spec.Type = obs.OutputTypeCloudwatch spec.Cloudwatch = &obs.Cloudwatch{} diff --git a/test/e2e/collection/security/container_security_test.go b/test/e2e/collection/security/container_security_test.go index db16750e53..59162071b4 100644 --- a/test/e2e/collection/security/container_security_test.go +++ b/test/e2e/collection/security/container_security_test.go @@ -83,9 +83,7 @@ var _ = Describe("Tests of collector container security stance", func() { Type: obs.OutputTypeElasticsearch, Elasticsearch: &obs.Elasticsearch{ Version: 6, - URLSpec: obs.URLSpec{ - URL: "http://foo.bar.svc:24224", - }, + URL: "http://foo.bar.svc:24224", }, }, }, diff --git a/test/framework/functional/output_elasticsearch.go b/test/framework/functional/output_elasticsearch.go index 22f1023a31..1ebcb6d11e 100644 --- a/test/framework/functional/output_elasticsearch.go +++ b/test/framework/functional/output_elasticsearch.go @@ -109,6 +109,32 @@ func (f *CollectorFunctionalFramework) AddESOutputWithTokenSecurity(b *runtime.P return f.AddESOutput(ElasticsearchVersion8, b, output, envVars) } +// ESNodeConfig defines configuration for an Elasticsearch node +type ESNodeConfig struct { + Name string + HTTPPort string +} + +// AddMultiESContainers adds multiple Elasticsearch containers to the pod for testing +// multi-endpoint scenarios. Each node runs in single-node mode with security disabled. +// This is a convenience wrapper around AddESOutput for creating multiple nodes. +func (f *CollectorFunctionalFramework) AddMultiESContainers(version ElasticsearchVersion, b *runtime.PodBuilder, nodes []ESNodeConfig) error { + for _, node := range nodes { + // Create a minimal OutputSpec for this node + output := obs.OutputSpec{ + Name: node.Name, + Elasticsearch: &obs.Elasticsearch{ + URL: fmt.Sprintf("http://localhost:%s", node.HTTPPort), + }, + } + // Reuse AddESOutput with default settings + if err := f.AddESOutput(version, b, output, nil); err != nil { + return err + } + } + return nil +} + func (f *CollectorFunctionalFramework) GetLogsFromElasticSearch(outputName string, outputLogType string, options ...Option) (results []string, err error) { index, ok := logTypeIndexMap[outputLogType] if !ok { diff --git a/test/functional/outputs/elasticsearch/forward_to_elasticsearch_test.go b/test/functional/outputs/elasticsearch/forward_to_elasticsearch_test.go index d1cee2b07e..54e5622083 100644 --- a/test/functional/outputs/elasticsearch/forward_to_elasticsearch_test.go +++ b/test/functional/outputs/elasticsearch/forward_to_elasticsearch_test.go @@ -3,6 +3,7 @@ package elasticsearch import ( "sort" "strconv" + "sync" "time" . "github.com/onsi/ginkgo/v2" @@ -319,4 +320,81 @@ var _ = Describe("[Functional][Outputs][ElasticSearch] Logforwarding to ElasticS Expect(outputTestLog).To(matchers.FitLogFormatTemplate(outputLogTemplate)) }) }) + + Context("with multiple endpoints", func() { + + const ( + esIndex = "application-write" + esBaseAddress = "http://0.0.0.0:" + ) + + var ( + // ES node configurations: name and HTTP port + esNodes = []functional.ESNodeConfig{ + {Name: "es-node-1", HTTPPort: "9200"}, + {Name: "es-node-2", HTTPPort: "9800"}, + } + ) + + AfterEach(func() { + framework.Cleanup() + }) + + // Helper: add two ES v8 containers on different ports + addMultiESContainers := func(b *runtime.PodBuilder) error { + return framework.AddMultiESContainers(functional.ElasticsearchVersion8, b, esNodes) + } + + // getLogsFromAllEndpoints queries both ES nodes in parallel and returns + // the combined log count. This avoids a sequential 5-minute timeout on a + // node that received zero logs due to non-deterministic load balancing. + getLogsFromAllEndpoints := func() int { + var logs1, logs2 []string + var wg sync.WaitGroup + + queryNode := func(result *[]string, nodeName, index string, opts ...functional.Option) { + defer wg.Done() + defer GinkgoRecover() + *result, _ = framework.GetLogsFromElasticSearchIndex(nodeName, index, opts...) + } + + wg.Add(2) + go queryNode(&logs1, esNodes[0].Name, esIndex) + go queryNode(&logs2, esNodes[1].Name, esIndex, functional.Option{Name: "port", Value: esNodes[1].HTTPPort}) + wg.Wait() + + return len(logs1) + len(logs2) + } + + DescribeTable("should deliver logs to multiple endpoints", func(esSpec *obs.Elasticsearch) { + framework = functional.NewCollectorFunctionalFramework() + obstestruntime.NewClusterLogForwarderBuilder(framework.Forwarder). + FromInput(obs.InputTypeApplication). + ToOutputWithVisitor(func(output *obs.OutputSpec) { + output.Name = "es-multi" + output.Type = obs.OutputTypeElasticsearch + output.Elasticsearch = esSpec + output.Elasticsearch.Index = `{.log_type||"notfound"}-write` + output.Elasticsearch.Version = 8 + }, "es-multi") + + Expect(framework.DeployWithVisitor(addMultiESContainers)).To(BeNil()) + Expect(framework.WritesApplicationLogs(10)).To(BeNil()) + + Expect(getLogsFromAllEndpoints()).To(Equal(10), "Expected 10 total logs across both ES nodes") + }, + Entry("using endpoints only", &obs.Elasticsearch{ + Endpoints: []obs.EndpointURL{ + obs.EndpointURL(esBaseAddress + esNodes[0].HTTPPort), + obs.EndpointURL(esBaseAddress + esNodes[1].HTTPPort), + }, + }), + Entry("using url and endpoints combined", &obs.Elasticsearch{ + URL: esBaseAddress + esNodes[0].HTTPPort, + Endpoints: []obs.EndpointURL{ + obs.EndpointURL(esBaseAddress + esNodes[1].HTTPPort), + }, + }), + ) + }) }) diff --git a/test/functional/outputs/multiple/multiple_test.go b/test/functional/outputs/multiple/multiple_test.go index 09c5b3b569..db99fe3a22 100644 --- a/test/functional/outputs/multiple/multiple_test.go +++ b/test/functional/outputs/multiple/multiple_test.go @@ -82,9 +82,7 @@ var _ = Describe("[Functional][Outputs][Multiple] tests", func() { ToOutputWithVisitor(func(output *obs.OutputSpec) { output.Type = obs.OutputTypeElasticsearch output.Elasticsearch = &obs.Elasticsearch{ - URLSpec: obs.URLSpec{ - URL: "http://0.0.0.0:9800", - }, + URL: "http://0.0.0.0:9800", Index: "foo", Version: 8, } diff --git a/test/runtime/observability/cluster_log_forwarder.go b/test/runtime/observability/cluster_log_forwarder.go index 987da06d59..f4529c471f 100644 --- a/test/runtime/observability/cluster_log_forwarder.go +++ b/test/runtime/observability/cluster_log_forwarder.go @@ -150,9 +150,7 @@ func (p *PipelineBuilder) ToElasticSearchOutput(visitors ...func(output *obs.Out output.Name = string(obs.OutputTypeElasticsearch) output.Type = obs.OutputTypeElasticsearch output.Elasticsearch = &obs.Elasticsearch{ - URLSpec: obs.URLSpec{ - URL: "http://0.0.0.0:9200", - }, + URL: "http://0.0.0.0:9200", Index: `{.log_type||"notfound"}-write`, Version: 8, }