LOG-5991: Setting authentication to token for ES output causes invalid configuration error

Author: Clee2691

internal/generator/vector/output/elasticsearch/auth.go

@@ -0,0 +1,52 @@
+package elasticsearch
+
+import (
+	obs "github.com/openshift/cluster-logging-operator/api/observability/v1"
+	"github.com/openshift/cluster-logging-operator/internal/api/observability"
+	"github.com/openshift/cluster-logging-operator/internal/constants"
+	"github.com/openshift/cluster-logging-operator/internal/generator/framework"
+	"github.com/openshift/cluster-logging-operator/internal/generator/vector/helpers"
+	"github.com/openshift/cluster-logging-operator/internal/utils"
+)
+
+type BearerToken struct {
+	ID    string
+	Token string
+}
+
+func (bt BearerToken) Name() string {
+	return "esBearerTokenTemplate"
+}
+
+func (bt BearerToken) Template() string {
+	return `{{define "` + bt.Name() + `" -}}
+[sinks.{{.ID}}.request.headers]
+Authorization = "Bearer {{.Token}}"
+{{end}}
+`
+}
+
+func NewBearerToken(id string, spec *obs.HTTPAuthentication, secrets observability.Secrets, op framework.Options) BearerToken {
+	bt := BearerToken{}
+	if spec != nil {
+		key := spec.Token
+		bt.ID = id
+		switch key.From {
+		case obs.BearerTokenFromSecret:
+			if key.Secret != nil {
+				bt.Token = helpers.SecretFrom(&obs.SecretReference{
+					SecretName: key.Secret.Name,
+					Key:        key.Secret.Key,
+				})
+			}
+		case obs.BearerTokenFromServiceAccount:
+			if name, found := utils.GetOption(op, framework.OptionServiceAccountTokenSecretName, ""); found {
+				bt.Token = helpers.SecretFrom(&obs.SecretReference{
+					Key:        constants.TokenKey,
+					SecretName: name,
+				})
+			}
+		}
+	}
+	return bt
+}

internal/generator/vector/output/elasticsearch/elasticsearch.go

@@ -83,9 +83,14 @@ if exists(.kubernetes.event.metadata.uid) {
 		common.NewBuffer(id, strategy),
 		common.NewRequest(id, strategy),
 		tls.New(id, o.TLS, secrets, op, Option{Name: URL, Value: o.Elasticsearch.URL}),
-		auth.HTTPAuth(id, o.Elasticsearch.Authentication, secrets, op),
 	)
 
+	if o.Elasticsearch.Authentication != nil && o.Elasticsearch.Authentication.Token != nil {
+		outputs = append(outputs, NewBearerToken(id, o.Elasticsearch.Authentication, secrets, op))
+	} else if o.Elasticsearch.Authentication != nil && o.Elasticsearch.Authentication.Username != nil && o.Elasticsearch.Authentication.Password != nil {
+		outputs = append(outputs, auth.NewBasic(id, o.Elasticsearch.Authentication, secrets))
+	}
+
 	return outputs
 }
 

internal/generator/vector/output/elasticsearch/elasticsearch_test.go

@@ -22,6 +22,7 @@ var _ = Describe("Generate Vector config", func() {
 		secretName = "es-1"
 		aUserName  = "testuser"
 		aPassword  = "testpass"
+		aToken     = "my-token"
 	)
 	var (
 		adapter fake.Output
@@ -71,6 +72,7 @@ var _ = Describe("Generate Vector config", func() {
 				Data: map[string][]byte{
 					constants.ClientUsername: []byte(aUserName),
 					constants.ClientPassword: []byte(aPassword),
+					constants.TokenKey:       []byte(aToken),
 				},
 			},
 		}
@@ -98,6 +100,26 @@ var _ = Describe("Generate Vector config", func() {
 		Expect(string(exp)).To(EqualConfigFrom(conf))
 	},
 		Entry("with username,password", nil, false, framework.NoOptions, "es_with_auth_username_password.toml"),
+		Entry("with custom bearer token", func(spec *obs.OutputSpec) {
+			spec.Elasticsearch.Authentication = &obs.HTTPAuthentication{
+				Token: &obs.BearerToken{
+					From: obs.BearerTokenFromSecret,
+					Secret: &obs.BearerTokenSecretKey{
+						Name: secretName,
+						Key:  constants.TokenKey,
+					},
+				},
+			}
+		}, false, framework.NoOptions, "es_with_auth_custom_bearer_token.toml"),
+		Entry("with serviceaccount token", func(spec *obs.OutputSpec) {
+			spec.Elasticsearch.Authentication = &obs.HTTPAuthentication{
+				Token: &obs.BearerToken{
+					From: obs.BearerTokenFromServiceAccount,
+				},
+			}
+		}, false, framework.Options{
+			framework.OptionServiceAccountTokenSecretName: "my-service-account-token",
+		}, "es_with_auth_serviceaccount_token.toml"),
 		Entry("with tls key,cert,ca-bundle", func(spec *obs.OutputSpec) {
 			spec.Elasticsearch.Authentication = nil
 			spec.TLS = tlsSpec

internal/generator/vector/output/elasticsearch/es_with_auth_custom_bearer_token.toml

@@ -0,0 +1,21 @@
+# 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]
+Authorization = "Bearer SECRET[kubernetes_secret.es-1/token]"

internal/generator/vector/output/elasticsearch/es_with_auth_serviceaccount_token.toml

@@ -0,0 +1,21 @@
+# 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]
+Authorization = "Bearer SECRET[kubernetes_secret.my-service-account-token/token]"

test/framework/functional/elasticsearch.go

@@ -3,13 +3,14 @@ package functional
 import (
 	"context"
 	"fmt"
-	"github.com/openshift/cluster-logging-operator/internal/api/initialize"
 	"net"
 	"os"
 	"strconv"
 	"strings"
 	"time"
 
+	"github.com/openshift/cluster-logging-operator/internal/api/initialize"
+
 	obs "github.com/openshift/cluster-logging-operator/api/observability/v1"
 
 	internalobs "github.com/openshift/cluster-logging-operator/internal/api/observability"
@@ -156,6 +157,13 @@ func (f *CollectorFunctionalFramework) RunCommand(container string, cmd ...strin
 	return out, err
 }
 
+func (f *CollectorFunctionalFramework) RunCommandInPod(pod *corev1.Pod, container string, cmd ...string) (string, error) {
+	log.V(2).Info("Running", "container", container, "cmd", cmd)
+	out, err := testruntime.ExecOc(pod, strings.ToLower(container), cmd[0], cmd[1:]...)
+	log.V(2).Info("Exec'd", "out", out, "err", err)
+	return out, err
+}
+
 func (f *CollectorFunctionalFramework) AddOutputContainersVisitors() []runtime.PodBuilderVisitor {
 	visitors := []runtime.PodBuilderVisitor{
 		func(b *runtime.PodBuilder) error {
@@ -274,8 +282,10 @@ func (f *CollectorFunctionalFramework) DeployWithVisitors(visitors []runtime.Pod
 			WithImagePullPolicy(corev1.PullAlways).ResourceRequirements(resources), FunctionalNodeName).
 		End()
 	for _, visit := range visitors {
-		if err = visit(b); err != nil {
-			return err
+		if visit != nil {
+			if err = visit(b); err != nil {
+				return err
+			}
 		}
 	}
 
@@ -417,7 +427,7 @@ func (f *CollectorFunctionalFramework) addOutputContainers(b *runtime.PodBuilder
 				return err
 			}
 		case obs.OutputTypeElasticsearch:
-			if err := f.AddES7Output(b, output); err != nil {
+			if err := f.AddESOutput(ElasticsearchVersion(output.Elasticsearch.Version), b, output, nil); err != nil {
 				return err
 			}
 		case obs.OutputTypeHTTP: