From 2cb6a791a26fc8d2846525edb8743bf26d733c0c Mon Sep 17 00:00:00 2001
From: Rafal Skolasinski <r.j.skolasinski@gmail.com>
Date: Tue, 15 Aug 2023 14:31:15 +0100
Subject: [PATCH 01/19] add helper target for go modules

---
 Makefile | 11 +++++++++++
 1 file changed, 11 insertions(+)

diff --git a/Makefile b/Makefile
index 829d27aeb3..1772ed627a 100644
--- a/Makefile
+++ b/Makefile
@@ -32,6 +32,17 @@ undeploy-k8s:
 # Dev
 #
 
+
+init-go-modules:
+	go work init || echo "go modules already initialized"
+	go work use operator
+	go work use scheduler
+	go work use apis/go
+	go work use components/tls
+	go work use hodometer
+	go work use tests/integration
+
+
 # use -W option for warnings as errors
 docs_build_html:
 	cd docs && \

From 7eac5f5219a511d84a394af1fc2b2e98314ec7bc Mon Sep 17 00:00:00 2001
From: Rafal Skolasinski <r.j.skolasinski@gmail.com>
Date: Mon, 21 Aug 2023 15:45:29 +0100
Subject: [PATCH 02/19] initial OAUTH configuration code for Kafka

---
 .dockerignore                                 |   5 +
 components/tls/pkg/oauth/k8s_secret.go        | 171 ++++++++++++++++++
 components/tls/pkg/oauth/oauth.go             |  33 ++++
 components/tls/pkg/oauth/store.go             | 116 ++++++++++++
 components/tls/pkg/password/k8s_secret.go     |   2 +-
 components/tls/pkg/password/store.go          |  15 +-
 components/tls/pkg/tls/util.go                |   1 +
 .../templates/seldon-v2-components.yaml       |   6 +-
 .../seldon-core-v2-setup/values.yaml          |   2 +
 .../seldon-core-v2-setup/values.yaml.template |   2 +
 .../patch_modelgateway.yaml                   |   2 +
 .../patch_pipelinegateway.yaml                |   2 +
 k8s/yaml/components.yaml                      |   4 +
 scheduler/Dockerfile.modelgateway             |   5 +
 scheduler/Dockerfile.pipelinegateway          |   5 +
 scheduler/pkg/kafka/config/tls.go             |  75 +++++++-
 16 files changed, 430 insertions(+), 16 deletions(-)
 create mode 100644 components/tls/pkg/oauth/k8s_secret.go
 create mode 100644 components/tls/pkg/oauth/oauth.go
 create mode 100644 components/tls/pkg/oauth/store.go

diff --git a/.dockerignore b/.dockerignore
index 95b90d9494..7dbb9c4e55 100644
--- a/.dockerignore
+++ b/.dockerignore
@@ -25,3 +25,8 @@ scheduler/mlrepo
 scheduler/mnt
 scheduler/notebooks
 scheduler/venv
+
+# General
+go.work
+go.work.sum
+
diff --git a/components/tls/pkg/oauth/k8s_secret.go b/components/tls/pkg/oauth/k8s_secret.go
new file mode 100644
index 0000000000..6df5efce9b
--- /dev/null
+++ b/components/tls/pkg/oauth/k8s_secret.go
@@ -0,0 +1,171 @@
+/*
+Copyright 2023 Seldon Technologies Ltd.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package oauth
+
+import (
+	"context"
+	"fmt"
+	"sync"
+
+	log "github.com/sirupsen/logrus"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/informers"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/cache"
+
+	"github.com/seldonio/seldon-core/components/tls/v2/pkg/k8s"
+)
+
+type OAUTHSecretHandler struct {
+	clientset   kubernetes.Interface
+	secretName  string
+	namespace   string
+	stopper     chan struct{}
+	logger      log.FieldLogger
+	mu          sync.RWMutex
+	oauthConfig OAUTHConfig
+}
+
+func NewOAUTHSecretHandler(secretName string, clientset kubernetes.Interface, namespace string, prefix string, locationSuffix string, logger log.FieldLogger) (*OAUTHSecretHandler, error) {
+	if clientset == nil {
+		var err error
+		clientset, err = k8s.CreateClientset()
+		if err != nil {
+			logger.WithError(err).Error("Failed to create clientset for OAUTH secret handler")
+			return nil, err
+		}
+	}
+	return &OAUTHSecretHandler{
+		clientset:  clientset,
+		secretName: secretName,
+		namespace:  namespace,
+		stopper:    make(chan struct{}),
+		logger:     logger,
+	}, nil
+}
+
+func (s *OAUTHSecretHandler) GetOAUTHConfig() OAUTHConfig {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	return s.oauthConfig
+}
+
+func (s *OAUTHSecretHandler) Stop() {
+	close(s.stopper)
+}
+
+func (s *OAUTHSecretHandler) saveOAUTHFromSecret(secret *corev1.Secret) error {
+	// Read and Save oauthbearer method
+	method, ok := secret.Data[methodKey]
+	if !ok {
+		return fmt.Errorf("Failed to find %s in secret %s", methodKey, secret.Name)
+	}
+	s.oauthConfig.Method = string(method)
+
+	// Read and Save oauthbearer client id
+	clientID, ok := secret.Data[clientIDKey]
+	if !ok {
+		return fmt.Errorf("Failed to find %s in secret %s", clientIDKey, secret.Name)
+	}
+	s.oauthConfig.ClientID = string(clientID)
+
+	// Read and Save oauthbearer client secret
+	clientSecret, ok := secret.Data[clientSecretKey]
+	if !ok {
+		return fmt.Errorf("Failed to find %s in secret %s", clientSecretKey, secret.Name)
+	}
+	s.oauthConfig.ClientSecret = string(clientSecret)
+
+	// Read and Save oauthbearer token endpoint url
+	tokenEndpointURL, ok := secret.Data[tokenEndpointURLKey]
+	if !ok {
+		return fmt.Errorf("Failed to find %s in secret %s", tokenEndpointURLKey, secret.Name)
+	}
+	s.oauthConfig.TokenEndpointURL = string(tokenEndpointURL)
+
+	// Read and Save oauthbearer extensions
+	extensions, ok := secret.Data[extensionsKey]
+	if !ok {
+		return fmt.Errorf("Failed to find %s in secret %s", extensionsKey, secret.Name)
+	}
+	s.oauthConfig.Extensions = string(extensions)
+
+	return nil
+}
+
+func (s *OAUTHSecretHandler) onAdd(obj interface{}) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	logger := s.logger.WithField("func", "onAdd")
+	secret := obj.(*corev1.Secret)
+	if secret.Name == s.secretName {
+		logger.Infof("OAUTH Secret %s added", s.secretName)
+		err := s.saveOAUTHFromSecret(secret)
+		if err != nil {
+			logger.WithError(err).Errorf("Failed to extract OAUTH from secret %s", secret.Name)
+		}
+	}
+}
+
+func (s *OAUTHSecretHandler) onUpdate(oldObj, newObj interface{}) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	logger := s.logger.WithField("func", "onUpdate")
+	secret := newObj.(*corev1.Secret)
+	if secret.Name == s.secretName {
+		logger.Infof("OAUTH Secret %s updated", s.secretName)
+		err := s.saveOAUTHFromSecret(secret)
+		if err != nil {
+			logger.WithError(err).Errorf("Failed to extract OAUTH from secret %s", secret.Name)
+		}
+	}
+}
+
+func (s *OAUTHSecretHandler) onDelete(obj interface{}) {
+	logger := s.logger.WithField("func", "onDelete")
+	secret := obj.(*corev1.Secret)
+	if secret.Name == s.secretName {
+		logger.Warnf("Secret %s deleted", secret.Name)
+	}
+}
+
+func (s *OAUTHSecretHandler) loadOAUTH(secretName string) error {
+	secret, err := s.clientset.CoreV1().Secrets(s.namespace).Get(context.Background(), secretName, metav1.GetOptions{})
+	if err != nil {
+		return err
+	}
+	return s.saveOAUTHFromSecret(secret)
+}
+
+func (s *OAUTHSecretHandler) GetOAUTHAndWatch() error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	err := s.loadOAUTH(s.secretName)
+	if err != nil {
+		return err
+	}
+	coreInformers := informers.NewSharedInformerFactoryWithOptions(s.clientset, 0, informers.WithNamespace(s.namespace))
+	coreInformers.Core().V1().Secrets().Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc:    s.onAdd,
+		UpdateFunc: s.onUpdate,
+		DeleteFunc: s.onDelete,
+	})
+	coreInformers.WaitForCacheSync(s.stopper)
+	coreInformers.Start(s.stopper)
+	return nil
+}
diff --git a/components/tls/pkg/oauth/oauth.go b/components/tls/pkg/oauth/oauth.go
new file mode 100644
index 0000000000..060e414fbb
--- /dev/null
+++ b/components/tls/pkg/oauth/oauth.go
@@ -0,0 +1,33 @@
+/*
+Copyright 2022 Seldon Technologies Ltd.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package oauth
+
+const (
+	methodKey           = "method"
+	clientIDKey         = "client_id"
+	clientSecretKey     = "client_secret"
+	tokenEndpointURLKey = "token_endpoint_url"
+	extensionsKey       = "extensions"
+)
+
+type OAUTHConfig struct {
+	Method           string
+	ClientID         string
+	ClientSecret     string
+	TokenEndpointURL string
+	Extensions       string
+}
diff --git a/components/tls/pkg/oauth/store.go b/components/tls/pkg/oauth/store.go
new file mode 100644
index 0000000000..53b4e96ec5
--- /dev/null
+++ b/components/tls/pkg/oauth/store.go
@@ -0,0 +1,116 @@
+/*
+Copyright 2023 Seldon Technologies Ltd.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package oauth
+
+import (
+	"fmt"
+	"os"
+
+	"github.com/sirupsen/logrus"
+	"k8s.io/client-go/kubernetes"
+
+	"github.com/seldonio/seldon-core/components/tls/v2/pkg/util"
+)
+
+const (
+	envSecretSuffix = "_OAUTH_SECRET_NAME"
+	envNamespace    = "POD_NAMESPACE"
+)
+
+type funcOAUTHServerOption struct {
+	f func(options *OAUTHStoreOptions)
+}
+
+func (fdo *funcOAUTHServerOption) apply(do *OAUTHStoreOptions) {
+	fdo.f(do)
+}
+
+func newFuncServerOption(f func(options *OAUTHStoreOptions)) *funcOAUTHServerOption {
+	return &funcOAUTHServerOption{
+		f: f,
+	}
+}
+
+type OAUTHStore interface {
+	GetOAUTHConfig() OAUTHConfig
+	Stop()
+}
+
+type OAUTHStoreOption interface {
+	apply(options *OAUTHStoreOptions)
+}
+
+type OAUTHStoreOptions struct {
+	prefix         string
+	locationSuffix string
+	clientset      kubernetes.Interface
+}
+
+func (c OAUTHStoreOptions) String() string {
+	return fmt.Sprintf("prefix=%s locationSuffix=%s clientset=%v",
+		c.prefix, c.locationSuffix, c.clientset)
+}
+
+func getDefaultOAUTHStoreOptions() OAUTHStoreOptions {
+	return OAUTHStoreOptions{}
+}
+
+func Prefix(prefix string) OAUTHStoreOption {
+	return newFuncServerOption(func(o *OAUTHStoreOptions) {
+		o.prefix = prefix
+	})
+}
+
+func LocationSuffix(suffix string) OAUTHStoreOption {
+	return newFuncServerOption(func(o *OAUTHStoreOptions) {
+		o.locationSuffix = suffix
+	})
+}
+func ClientSet(clientSet kubernetes.Interface) OAUTHStoreOption {
+	return newFuncServerOption(func(o *OAUTHStoreOptions) {
+		o.clientset = clientSet
+	})
+}
+
+func NewOAUTHStore(opt ...OAUTHStoreOption) (OAUTHStore, error) {
+	opts := getDefaultOAUTHStoreOptions()
+	for _, o := range opt {
+		o.apply(&opts)
+	}
+	logger := logrus.New().WithField("source", "OAUTHStore")
+	logger.Infof("Options:%s", opts.String())
+	if secretName, ok := util.GetEnv(opts.prefix, envSecretSuffix); ok {
+		logger.Infof("Starting new OAUTH k8s secret store for %s from secret %s", opts.prefix, secretName)
+		namespace, ok := os.LookupEnv(envNamespace)
+		logger.Infof("Namespace %s", namespace)
+		if !ok {
+			return nil, fmt.Errorf("Namespace env var %s not found and needed for OAUTH secret", envNamespace)
+		}
+		ps, err := NewOAUTHSecretHandler(secretName, opts.clientset, namespace, opts.prefix, opts.locationSuffix, logger)
+		if err != nil {
+			return nil, err
+		}
+		err = ps.GetOAUTHAndWatch()
+		if err != nil {
+			return nil, err
+		}
+		return ps, nil
+	} else {
+		// NOT IMPLEMENTED ERROR
+		return nil, fmt.Errorf("OAUTH mechanism is currently only supported on K8s")
+	}
+}
diff --git a/components/tls/pkg/password/k8s_secret.go b/components/tls/pkg/password/k8s_secret.go
index 2559d01e87..09e455e82f 100644
--- a/components/tls/pkg/password/k8s_secret.go
+++ b/components/tls/pkg/password/k8s_secret.go
@@ -106,7 +106,7 @@ func (s *PasswordSecretHandler) onUpdate(oldObj, newObj interface{}) {
 	logger := s.logger.WithField("func", "onUpdate")
 	secret := newObj.(*corev1.Secret)
 	if secret.Name == s.secretName {
-		logger.Infof("TLS Secret %s updated", s.secretName)
+		logger.Infof("Password Secret %s updated", s.secretName)
 		err := s.savePasswordFromSecret(secret)
 		if err != nil {
 			logger.WithError(err).Errorf("Failed to extract password from secret %s", secret.Name)
diff --git a/components/tls/pkg/password/store.go b/components/tls/pkg/password/store.go
index fe1a2965f5..19a8ff4596 100644
--- a/components/tls/pkg/password/store.go
+++ b/components/tls/pkg/password/store.go
@@ -31,16 +31,16 @@ const (
 	envNamespace    = "POD_NAMESPACE"
 )
 
-type funcTLSServerOption struct {
+type funcPasswordServerOption struct {
 	f func(options *PasswordStoreOptions)
 }
 
-func (fdo *funcTLSServerOption) apply(do *PasswordStoreOptions) {
+func (fdo *funcPasswordServerOption) apply(do *PasswordStoreOptions) {
 	fdo.f(do)
 }
 
-func newFuncServerOption(f func(options *PasswordStoreOptions)) *funcTLSServerOption {
-	return &funcTLSServerOption{
+func newFuncServerOption(f func(options *PasswordStoreOptions)) *funcPasswordServerOption {
+	return &funcPasswordServerOption{
 		f: f,
 	}
 }
@@ -61,8 +61,8 @@ type PasswordStoreOptions struct {
 }
 
 func (c PasswordStoreOptions) String() string {
-	return fmt.Sprintf("prefix=%s clientset=%v",
-		c.prefix, c.clientset)
+	return fmt.Sprintf("prefix=%s locationSuffix=%s clientset=%v",
+		c.prefix, c.locationSuffix, c.clientset)
 }
 
 func getDefaultPasswordStoreOptions() PasswordStoreOptions {
@@ -96,8 +96,9 @@ func NewPasswordStore(opt ...PasswordStoreOption) (PasswordStore, error) {
 	if secretName, ok := util.GetEnv(opts.prefix, envSecretSuffix); ok {
 		logger.Infof("Starting new password k8s secret store for %s from secret %s", opts.prefix, secretName)
 		namespace, ok := os.LookupEnv(envNamespace)
+		logger.Infof("Namespace %s", namespace)
 		if !ok {
-			return nil, fmt.Errorf("Namespace env var %s not found and needed for secret TLS", envNamespace)
+			return nil, fmt.Errorf("Namespace env var %s not found and needed for password secret", envNamespace)
 		}
 		ps, err := NewPasswordSecretHandler(secretName, opts.clientset, namespace, opts.prefix, opts.locationSuffix, logger)
 		if err != nil {
diff --git a/components/tls/pkg/tls/util.go b/components/tls/pkg/tls/util.go
index 7b9ae50aab..97a72ac37c 100644
--- a/components/tls/pkg/tls/util.go
+++ b/components/tls/pkg/tls/util.go
@@ -30,6 +30,7 @@ const (
 	EnvSASLMechanismSuffix   = "_SASL_MECHANISM"
 	SASLMechanismSCRAMSHA512 = "SCRAM-SHA-512"
 	SASLMechanismSCRAMSHA256 = "SCRAM-SHA-256"
+	SASLMechanismOAUTHBEARER = "OAUTHBEARER"
 	SASLMechanismPlain       = "PLAIN"
 
 	EnvEndpointIdentificationMechanismSuffix = "_TLS_ENDPOINT_IDENTIFICATION_ALGORITHM"
diff --git a/k8s/helm-charts/seldon-core-v2-setup/templates/seldon-v2-components.yaml b/k8s/helm-charts/seldon-core-v2-setup/templates/seldon-v2-components.yaml
index 65ac3e61ff..b81c6db23c 100644
--- a/k8s/helm-charts/seldon-core-v2-setup/templates/seldon-v2-components.yaml
+++ b/k8s/helm-charts/seldon-core-v2-setup/templates/seldon-v2-components.yaml
@@ -1030,6 +1030,8 @@ spec:
           value: '{{ .Values.security.kafka.protocol }}'
         - name: KAFKA_SASL_MECHANISM
           value: '{{ .Values.security.kafka.sasl.mechanism }}'
+        - name: KAFKA_CLIENT_OAUTH_SECRET_NAME
+          value: '{{ .Values.security.kafka.sasl.oauth.secret }}'
         - name: KAFKA_CLIENT_TLS_ENDPOINT_IDENTIFICATION_ALGORITHM
           value: '{{ .Values.security.kafka.ssl.client.endpointIdentificationAlgorithm
             }}'
@@ -1184,6 +1186,8 @@ spec:
           value: '{{ .Values.security.kafka.protocol }}'
         - name: KAFKA_SASL_MECHANISM
           value: '{{ .Values.security.kafka.sasl.mechanism }}'
+        - name: KAFKA_CLIENT_OAUTH_SECRET_NAME
+          value: '{{ .Values.security.kafka.sasl.oauth.secret }}'
         - name: KAFKA_CLIENT_TLS_ENDPOINT_IDENTIFICATION_ALGORITHM
           value: '{{ .Values.security.kafka.ssl.client.endpointIdentificationAlgorithm
             }}'
@@ -1666,8 +1670,6 @@ spec:
       - containerPort: 9000
         name: server-http
         protocol: TCP
-      - containerPort: 8082
-        name: server-metrics
       readinessProbe:
         httpGet:
           path: /v2/health/live
diff --git a/k8s/helm-charts/seldon-core-v2-setup/values.yaml b/k8s/helm-charts/seldon-core-v2-setup/values.yaml
index 306f0a1947..6601cab500 100644
--- a/k8s/helm-charts/seldon-core-v2-setup/values.yaml
+++ b/k8s/helm-charts/seldon-core-v2-setup/values.yaml
@@ -25,6 +25,8 @@ security:
         username: seldon
         secret:
         passwordPath: /tmp/sasl/kafka/client/password
+      oauth:
+        secret:
     ssl:
       client:
         secret:
diff --git a/k8s/helm-charts/seldon-core-v2-setup/values.yaml.template b/k8s/helm-charts/seldon-core-v2-setup/values.yaml.template
index 36ea8f14a7..a81f3d8ea0 100644
--- a/k8s/helm-charts/seldon-core-v2-setup/values.yaml.template
+++ b/k8s/helm-charts/seldon-core-v2-setup/values.yaml.template
@@ -25,6 +25,8 @@ security:
         username: seldon
         secret:
         passwordPath: /tmp/sasl/kafka/client/password
+      oauth:
+        secret:
     ssl:
       client:
         secret:
diff --git a/k8s/kustomize/helm-components-sc/patch_modelgateway.yaml b/k8s/kustomize/helm-components-sc/patch_modelgateway.yaml
index fc39ed2674..2a9d688851 100644
--- a/k8s/kustomize/helm-components-sc/patch_modelgateway.yaml
+++ b/k8s/kustomize/helm-components-sc/patch_modelgateway.yaml
@@ -34,6 +34,8 @@ spec:
             value: '{{ .Values.security.kafka.protocol }}'
           - name: KAFKA_SASL_MECHANISM
             value: '{{ .Values.security.kafka.sasl.mechanism }}'
+          - name: KAFKA_CLIENT_OAUTH_SECRET_NAME
+            value: '{{ .Values.security.kafka.sasl.oauth.secret }}'
           - name: KAFKA_CLIENT_TLS_ENDPOINT_IDENTIFICATION_ALGORITHM
             value: '{{ .Values.security.kafka.ssl.client.endpointIdentificationAlgorithm }}'
           - name: KAFKA_CLIENT_TLS_SECRET_NAME
diff --git a/k8s/kustomize/helm-components-sc/patch_pipelinegateway.yaml b/k8s/kustomize/helm-components-sc/patch_pipelinegateway.yaml
index 17a8c162f4..ec854c090f 100644
--- a/k8s/kustomize/helm-components-sc/patch_pipelinegateway.yaml
+++ b/k8s/kustomize/helm-components-sc/patch_pipelinegateway.yaml
@@ -23,6 +23,8 @@ spec:
             value: '{{ .Values.security.kafka.protocol }}'
           - name: KAFKA_SASL_MECHANISM
             value: '{{ .Values.security.kafka.sasl.mechanism }}'
+          - name: KAFKA_CLIENT_OAUTH_SECRET_NAME
+            value: '{{ .Values.security.kafka.sasl.oauth.secret }}'
           - name: KAFKA_CLIENT_TLS_ENDPOINT_IDENTIFICATION_ALGORITHM
             value: '{{ .Values.security.kafka.ssl.client.endpointIdentificationAlgorithm }}'
           - name: KAFKA_CLIENT_TLS_SECRET_NAME
diff --git a/k8s/yaml/components.yaml b/k8s/yaml/components.yaml
index b28cf4585d..8bad4e62ef 100644
--- a/k8s/yaml/components.yaml
+++ b/k8s/yaml/components.yaml
@@ -669,6 +669,8 @@ spec:
           value: 'PLAINTEXT'
         - name: KAFKA_SASL_MECHANISM
           value: 'SCRAM-SHA-512'
+        - name: KAFKA_CLIENT_OAUTH_SECRET_NAME
+          value: ''
         - name: KAFKA_CLIENT_TLS_ENDPOINT_IDENTIFICATION_ALGORITHM
           value: ''
         - name: KAFKA_CLIENT_TLS_SECRET_NAME
@@ -816,6 +818,8 @@ spec:
           value: 'PLAINTEXT'
         - name: KAFKA_SASL_MECHANISM
           value: 'SCRAM-SHA-512'
+        - name: KAFKA_CLIENT_OAUTH_SECRET_NAME
+          value: ''
         - name: KAFKA_CLIENT_TLS_ENDPOINT_IDENTIFICATION_ALGORITHM
           value: ''
         - name: KAFKA_CLIENT_TLS_SECRET_NAME
diff --git a/scheduler/Dockerfile.modelgateway b/scheduler/Dockerfile.modelgateway
index 5912e62cf6..c325a57dac 100644
--- a/scheduler/Dockerfile.modelgateway
+++ b/scheduler/Dockerfile.modelgateway
@@ -12,6 +12,11 @@ FROM registry.access.redhat.com/ubi9/ubi-minimal as certs
 
 # Kafka dependencies necessitate leaving CGo enabled and using a base image with C dependencies
 FROM registry.access.redhat.com/ubi9/ubi-micro:9.2-9
+
+#Kafka OIDC token retriva certs: https://github.com/confluentinc/librdkafka/issues/3751
+COPY --from=certs /etc/ssl/certs/ca-bundle.crt /etc/ssl/certs/ca-certificates.crt
+
+# Broker Certificates
 COPY --from=certs /etc/ssl/certs/ca-bundle.crt /tmp/certs/kafka/broker/ca.crt
 RUN chmod -R 777 /tmp/certs/
 COPY --from=builder /build/scheduler/bin/modelgateway /bin/modelgateway
diff --git a/scheduler/Dockerfile.pipelinegateway b/scheduler/Dockerfile.pipelinegateway
index 8b85abb056..e61962281a 100644
--- a/scheduler/Dockerfile.pipelinegateway
+++ b/scheduler/Dockerfile.pipelinegateway
@@ -12,6 +12,11 @@ FROM registry.access.redhat.com/ubi9/ubi-minimal as certs
 
 # Kafka dependencies necessitate leaving CGo enabled and using a base image with C dependencies
 FROM registry.access.redhat.com/ubi9/ubi-micro:9.2-9
+
+#Kafka OIDC token retriva certs: https://github.com/confluentinc/librdkafka/issues/3751
+COPY --from=certs /etc/ssl/certs/ca-bundle.crt /etc/ssl/certs/ca-certificates.crt
+
+# Broker Certificates
 COPY --from=certs /etc/ssl/certs/ca-bundle.crt /tmp/certs/kafka/broker/ca.crt
 RUN chmod -R 777 /tmp/certs/
 COPY --from=builder /build/scheduler/bin/pipelinegateway /bin/pipelinegateway
diff --git a/scheduler/pkg/kafka/config/tls.go b/scheduler/pkg/kafka/config/tls.go
index 61841c8d7b..a11c93ec5f 100644
--- a/scheduler/pkg/kafka/config/tls.go
+++ b/scheduler/pkg/kafka/config/tls.go
@@ -21,6 +21,7 @@ import (
 
 	"github.com/confluentinc/confluent-kafka-go/kafka"
 
+	"github.com/seldonio/seldon-core/components/tls/v2/pkg/oauth"
 	"github.com/seldonio/seldon-core/components/tls/v2/pkg/password"
 	"github.com/seldonio/seldon-core/components/tls/v2/pkg/tls"
 	"github.com/seldonio/seldon-core/components/tls/v2/pkg/util"
@@ -39,7 +40,7 @@ func AddKafkaSSLOptions(config kafka.ConfigMap) error {
 	switch protocol {
 	case tls.SecurityProtocolSSL:
 		return setupTLSAuthentication(config)
-	case tls.SecurityProtocolSASLSSL: // Note: we don't support SASL_PLAINTXT
+	case tls.SecurityProtocolSASLSSL: // Note: we don't support SASL_PLAINTEXT
 		return setupSASLSSLAuthentication(config)
 	case tls.SecurityProtocolPlaintxt:
 		return nil
@@ -48,17 +49,37 @@ func AddKafkaSSLOptions(config kafka.ConfigMap) error {
 }
 
 func setupSASLSSLAuthentication(config kafka.ConfigMap) error {
-	// Set the SASL mechanism
+
 	mechanism := tls.GetSASLMechanismFromEnv(tls.EnvSecurityPrefixKafka)
-	if (mechanism != tls.SASLMechanismPlain) && (mechanism != tls.SASLMechanismSCRAMSHA256) && (mechanism != tls.SASLMechanismSCRAMSHA512) {
-		return fmt.Errorf("Provided SASL mechanism %s is not supported", mechanism)
+
+	// TODO: Remove before merge (overwrite for testing)
+	mechanism = tls.SASLMechanismOAUTHBEARER
+
+	var err error
+	switch mechanism {
+	case tls.SASLMechanismPlain:
+		err = configureSASLSSLSCRAM(mechanism, config)
+	case tls.SASLMechanismSCRAMSHA256, tls.SASLMechanismSCRAMSHA512:
+		err = configureSASLSSLSCRAM(mechanism, config)
+	case tls.SASLMechanismOAUTHBEARER:
+		err = configureSASLSSLOAUTHBEARER(mechanism, config)
+	default:
+		err = fmt.Errorf("Provided SASL mechanism %s is not supported", mechanism)
 	}
+
+	return err
+}
+
+func configureSASLSSLSCRAM(mechanism string, config kafka.ConfigMap) error {
+	// Set the SASL mechanism
 	config["security.protocol"] = "SASL_SSL"
 	config["sasl.mechanism"] = mechanism
 
 	// Set the SASL username and password
-	ps, err := password.NewPasswordStore(password.Prefix(EnvKafkaClientPrefix),
-		password.LocationSuffix(EnvPasswordLocationSuffix))
+	ps, err := password.NewPasswordStore(
+		password.Prefix(EnvKafkaClientPrefix),
+		password.LocationSuffix(EnvPasswordLocationSuffix),
+	)
 	if err != nil {
 		return err
 	}
@@ -88,6 +109,48 @@ func setupSASLSSLAuthentication(config kafka.ConfigMap) error {
 	return nil
 }
 
+func configureSASLSSLOAUTHBEARER(mechanism string, config kafka.ConfigMap) error {
+	// Set the SASL mechanism
+	config["security.protocol"] = "SASL_SSL"
+	config["sasl.mechanism"] = "OAUTHBEARER"
+
+	// Set OAUTH Configuration
+	oauthStore, err := oauth.NewOAUTHStore(
+		oauth.Prefix(EnvKafkaClientPrefix),
+		oauth.LocationSuffix(EnvPasswordLocationSuffix),
+	)
+	if err != nil {
+		return err
+	}
+
+	oauthConfig := oauthStore.GetOAUTHConfig()
+
+	config["sasl.oauthbearer.method"] = oauthConfig.Method
+	config["sasl.oauthbearer.client.id"] = oauthConfig.ClientID
+	config["sasl.oauthbearer.client.secret"] = oauthConfig.ClientSecret
+	config["sasl.oauthbearer.token.endpoint.url"] = oauthConfig.TokenEndpointURL
+	config["sasl.oauthbearer.extensions"] = oauthConfig.Extensions
+
+	// Set the TLS Certificate
+	cs, err := tls.NewCertificateStore(
+		tls.ValidationOnly(true),
+		tls.ValidationPrefix(EnvKafkaBrokerPrefix),
+	)
+	if err != nil {
+		return err
+	}
+	caCert := cs.GetValidationCertificate()
+	// issue is that ca.pem does not work with multiple certificates defined
+	// see https://github.com/confluentinc/confluent-kafka-go/issues/827 (Fixed needs updating and testing in our code)
+
+	config["ssl.ca.location"] = caCert.CaPath
+
+	endpointMechanism := tls.GetEndpointIdentificationMechanismFromEnv(tls.EnvSecurityPrefixKafkaClient)
+	config["ssl.endpoint.identification.algorithm"] = endpointMechanism
+
+	return nil
+}
+
 func setupTLSAuthentication(config kafka.ConfigMap) error {
 	var cs *tls.CertificateStore
 	cs, err := tls.NewCertificateStore(

From a147e06c4b898f678e7e8ed4ddf5d2fb00ee4e75 Mon Sep 17 00:00:00 2001
From: Rafal Skolasinski <r.j.skolasinski@gmail.com>
Date: Mon, 21 Aug 2023 15:50:30 +0100
Subject: [PATCH 03/19] fix missing port setting;

---
 .../templates/seldon-v2-components.yaml                |  2 ++
 k8s/yaml/components.yaml                               |  2 ++
 operator/config/serverconfigs/mlserver.yaml            | 10 ++++++----
 3 files changed, 10 insertions(+), 4 deletions(-)

diff --git a/k8s/helm-charts/seldon-core-v2-setup/templates/seldon-v2-components.yaml b/k8s/helm-charts/seldon-core-v2-setup/templates/seldon-v2-components.yaml
index b81c6db23c..41bc6b6b75 100644
--- a/k8s/helm-charts/seldon-core-v2-setup/templates/seldon-v2-components.yaml
+++ b/k8s/helm-charts/seldon-core-v2-setup/templates/seldon-v2-components.yaml
@@ -1670,6 +1670,8 @@ spec:
       - containerPort: 9000
         name: server-http
         protocol: TCP
+      - containerPort: 8082
+        name: server-metrics
       readinessProbe:
         httpGet:
           path: /v2/health/live
diff --git a/k8s/yaml/components.yaml b/k8s/yaml/components.yaml
index 8bad4e62ef..fb753254c4 100644
--- a/k8s/yaml/components.yaml
+++ b/k8s/yaml/components.yaml
@@ -1282,6 +1282,8 @@ spec:
       - containerPort: 9000
         name: server-http
         protocol: TCP
+      - containerPort: 8082
+        name: server-metrics
       readinessProbe:
         httpGet:
           path: /v2/health/live
diff --git a/operator/config/serverconfigs/mlserver.yaml b/operator/config/serverconfigs/mlserver.yaml
index 0d13bc5d26..c9499af91b 100644
--- a/operator/config/serverconfigs/mlserver.yaml
+++ b/operator/config/serverconfigs/mlserver.yaml
@@ -40,7 +40,7 @@ spec:
       command:
         - /bin/agent
       args:
-        - --tracing-config-path=/mnt/tracing/tracing.json 
+        - --tracing-config-path=/mnt/tracing/tracing.json
       name: agent
       env:
       - name: SELDON_SERVER_CAPABILITIES
@@ -56,11 +56,11 @@ spec:
       - name: SELDON_SERVER_HTTP_PORT
         value: "9000"
       - name: SELDON_SERVER_GRPC_PORT
-        value: "9500"          
+        value: "9500"
       - name: SELDON_REVERSE_PROXY_HTTP_PORT
         value: "9001"
       - name: SELDON_REVERSE_PROXY_GRPC_PORT
-        value: "9501"          
+        value: "9501"
       - name: SELDON_SCHEDULER_HOST
         value: "seldon-scheduler"
       - name: SELDON_SCHEDULER_PORT
@@ -153,7 +153,7 @@ spec:
           path: /v2/health/live
           port: server-http
         initialDelaySeconds: 5
-        periodSeconds: 5          
+        periodSeconds: 5
       startupProbe:
         httpGet:
           path: /v2/health/live
@@ -168,6 +168,8 @@ spec:
       - containerPort: 9000
         name: server-http
         protocol: TCP
+      - containerPort: 8082
+        name: server-metrics
       volumeMounts:
       - mountPath: /mnt/agent
         name: mlserver-models

From cbc762e3bdf540bdcab6c2790f1080b0a68e573a Mon Sep 17 00:00:00 2001
From: Rafal Skolasinski <r.j.skolasinski@gmail.com>
Date: Mon, 21 Aug 2023 16:58:38 +0100
Subject: [PATCH 04/19] fix prefix of some global variables

---
 components/tls/pkg/oauth/k8s_secret.go | 20 ++++++++++----------
 components/tls/pkg/oauth/oauth.go      | 10 +++++-----
 2 files changed, 15 insertions(+), 15 deletions(-)

diff --git a/components/tls/pkg/oauth/k8s_secret.go b/components/tls/pkg/oauth/k8s_secret.go
index 6df5efce9b..c6436710a0 100644
--- a/components/tls/pkg/oauth/k8s_secret.go
+++ b/components/tls/pkg/oauth/k8s_secret.go
@@ -71,37 +71,37 @@ func (s *OAUTHSecretHandler) Stop() {
 
 func (s *OAUTHSecretHandler) saveOAUTHFromSecret(secret *corev1.Secret) error {
 	// Read and Save oauthbearer method
-	method, ok := secret.Data[methodKey]
+	method, ok := secret.Data[SecretKeyMethod]
 	if !ok {
-		return fmt.Errorf("Failed to find %s in secret %s", methodKey, secret.Name)
+		return fmt.Errorf("Failed to find %s in secret %s", SecretKeyMethod, secret.Name)
 	}
 	s.oauthConfig.Method = string(method)
 
 	// Read and Save oauthbearer client id
-	clientID, ok := secret.Data[clientIDKey]
+	clientID, ok := secret.Data[SecretKeyClientID]
 	if !ok {
-		return fmt.Errorf("Failed to find %s in secret %s", clientIDKey, secret.Name)
+		return fmt.Errorf("Failed to find %s in secret %s", SecretKeyClientID, secret.Name)
 	}
 	s.oauthConfig.ClientID = string(clientID)
 
 	// Read and Save oauthbearer client secret
-	clientSecret, ok := secret.Data[clientSecretKey]
+	clientSecret, ok := secret.Data[SecretKeyClientSecret]
 	if !ok {
-		return fmt.Errorf("Failed to find %s in secret %s", clientSecretKey, secret.Name)
+		return fmt.Errorf("Failed to find %s in secret %s", SecretKeyClientSecret, secret.Name)
 	}
 	s.oauthConfig.ClientSecret = string(clientSecret)
 
 	// Read and Save oauthbearer token endpoint url
-	tokenEndpointURL, ok := secret.Data[tokenEndpointURLKey]
+	tokenEndpointURL, ok := secret.Data[SecretKeyTokenEndpointURL]
 	if !ok {
-		return fmt.Errorf("Failed to find %s in secret %s", tokenEndpointURLKey, secret.Name)
+		return fmt.Errorf("Failed to find %s in secret %s", SecretKeyTokenEndpointURL, secret.Name)
 	}
 	s.oauthConfig.TokenEndpointURL = string(tokenEndpointURL)
 
 	// Read and Save oauthbearer extensions
-	extensions, ok := secret.Data[extensionsKey]
+	extensions, ok := secret.Data[SecretKeyExtensions]
 	if !ok {
-		return fmt.Errorf("Failed to find %s in secret %s", extensionsKey, secret.Name)
+		return fmt.Errorf("Failed to find %s in secret %s", SecretKeyExtensions, secret.Name)
 	}
 	s.oauthConfig.Extensions = string(extensions)
 
diff --git a/components/tls/pkg/oauth/oauth.go b/components/tls/pkg/oauth/oauth.go
index 060e414fbb..eb25995b9a 100644
--- a/components/tls/pkg/oauth/oauth.go
+++ b/components/tls/pkg/oauth/oauth.go
@@ -17,11 +17,11 @@ limitations under the License.
 package oauth
 
 const (
-	methodKey           = "method"
-	clientIDKey         = "client_id"
-	clientSecretKey     = "client_secret"
-	tokenEndpointURLKey = "token_endpoint_url"
-	extensionsKey       = "extensions"
+	SecretKeyMethod           = "method"
+	SecretKeyClientID         = "client_id"
+	SecretKeyClientSecret     = "client_secret"
+	SecretKeyTokenEndpointURL = "token_endpoint_url"
+	SecretKeyExtensions       = "extensions"
 )
 
 type OAUTHConfig struct {

From 83b80c8273b4f5e8cf89d9b40acb98d6af09abee Mon Sep 17 00:00:00 2001
From: Rafal Skolasinski <r.j.skolasinski@gmail.com>
Date: Mon, 21 Aug 2023 16:58:50 +0100
Subject: [PATCH 05/19] temporarily fix lint

---
 scheduler/pkg/kafka/config/tls.go | 1 +
 1 file changed, 1 insertion(+)

diff --git a/scheduler/pkg/kafka/config/tls.go b/scheduler/pkg/kafka/config/tls.go
index a11c93ec5f..4a331168a8 100644
--- a/scheduler/pkg/kafka/config/tls.go
+++ b/scheduler/pkg/kafka/config/tls.go
@@ -53,6 +53,7 @@ func setupSASLSSLAuthentication(config kafka.ConfigMap) error {
 	mechanism := tls.GetSASLMechanismFromEnv(tls.EnvSecurityPrefixKafka)
 
 	// TODO: Remove before merge (overwrite for testing)
+	_ = mechanism
 	mechanism = tls.SASLMechanismOAUTHBEARER
 
 	var err error

From 79390153d4f1045b2430020bf98e7040f05302c2 Mon Sep 17 00:00:00 2001
From: Rafal Skolasinski <r.j.skolasinski@gmail.com>
Date: Mon, 21 Aug 2023 17:09:58 +0100
Subject: [PATCH 06/19] rename tls.go to auth.go

---
 scheduler/pkg/kafka/config/{tls.go => auth.go} | 0
 1 file changed, 0 insertions(+), 0 deletions(-)
 rename scheduler/pkg/kafka/config/{tls.go => auth.go} (100%)

diff --git a/scheduler/pkg/kafka/config/tls.go b/scheduler/pkg/kafka/config/auth.go
similarity index 100%
rename from scheduler/pkg/kafka/config/tls.go
rename to scheduler/pkg/kafka/config/auth.go

From 4e522271bd062bf79ae84fa39f923b169d8a154f Mon Sep 17 00:00:00 2001
From: Rafal Skolasinski <r.j.skolasinski@gmail.com>
Date: Wed, 23 Aug 2023 22:46:27 +0100
Subject: [PATCH 07/19] move oauth package under kafka config

---
 scheduler/pkg/kafka/config/auth.go                | 15 ++++++++-------
 .../pkg/kafka/config}/oauth/k8s_secret.go         |  0
 .../pkg/kafka/config}/oauth/oauth.go              |  0
 .../pkg/kafka/config}/oauth/store.go              |  0
 4 files changed, 8 insertions(+), 7 deletions(-)
 rename {components/tls/pkg => scheduler/pkg/kafka/config}/oauth/k8s_secret.go (100%)
 rename {components/tls/pkg => scheduler/pkg/kafka/config}/oauth/oauth.go (100%)
 rename {components/tls/pkg => scheduler/pkg/kafka/config}/oauth/store.go (100%)

diff --git a/scheduler/pkg/kafka/config/auth.go b/scheduler/pkg/kafka/config/auth.go
index 4a331168a8..8ceea4013e 100644
--- a/scheduler/pkg/kafka/config/auth.go
+++ b/scheduler/pkg/kafka/config/auth.go
@@ -21,18 +21,19 @@ import (
 
 	"github.com/confluentinc/confluent-kafka-go/kafka"
 
-	"github.com/seldonio/seldon-core/components/tls/v2/pkg/oauth"
 	"github.com/seldonio/seldon-core/components/tls/v2/pkg/password"
 	"github.com/seldonio/seldon-core/components/tls/v2/pkg/tls"
 	"github.com/seldonio/seldon-core/components/tls/v2/pkg/util"
+	"github.com/seldonio/seldon-core/scheduler/v2/pkg/kafka/config/oauth"
 )
 
 const (
-	EnvKafkaClientPrefix      = "KAFKA_CLIENT"
-	EnvKafkaBrokerPrefix      = "KAFKA_BROKER"
-	EnvSASLUsernameSuffix     = "_SASL_USERNAME"
-	EnvPasswordLocationSuffix = "_SASL_PASSWORD_LOCATION"
-	DefaultSASLUsername       = "seldon"
+	EnvKafkaClientPrefix         = "KAFKA_CLIENT"
+	EnvKafkaBrokerPrefix         = "KAFKA_BROKER"
+	EnvSASLUsernameSuffix        = "_SASL_USERNAME"
+	EnvPasswordLocationSuffix    = "_SASL_PASSWORD_LOCATION"
+	EnvOAUTHConfigLocationSuffix = "_OAUTH_CONFIG_LOCATION"
+	DefaultSASLUsername          = "seldon"
 )
 
 func AddKafkaSSLOptions(config kafka.ConfigMap) error {
@@ -118,7 +119,7 @@ func configureSASLSSLOAUTHBEARER(mechanism string, config kafka.ConfigMap) error
 	// Set OAUTH Configuration
 	oauthStore, err := oauth.NewOAUTHStore(
 		oauth.Prefix(EnvKafkaClientPrefix),
-		oauth.LocationSuffix(EnvPasswordLocationSuffix),
+		oauth.LocationSuffix(EnvOAUTHConfigLocationSuffix),
 	)
 	if err != nil {
 		return err
diff --git a/components/tls/pkg/oauth/k8s_secret.go b/scheduler/pkg/kafka/config/oauth/k8s_secret.go
similarity index 100%
rename from components/tls/pkg/oauth/k8s_secret.go
rename to scheduler/pkg/kafka/config/oauth/k8s_secret.go
diff --git a/components/tls/pkg/oauth/oauth.go b/scheduler/pkg/kafka/config/oauth/oauth.go
similarity index 100%
rename from components/tls/pkg/oauth/oauth.go
rename to scheduler/pkg/kafka/config/oauth/oauth.go
diff --git a/components/tls/pkg/oauth/store.go b/scheduler/pkg/kafka/config/oauth/store.go
similarity index 100%
rename from components/tls/pkg/oauth/store.go
rename to scheduler/pkg/kafka/config/oauth/store.go

From 1b3e5422d21581f12df857327b988ab8cfa39cde Mon Sep 17 00:00:00 2001
From: Rafal Skolasinski <r.j.skolasinski@gmail.com>
Date: Wed, 23 Aug 2023 23:32:37 +0100
Subject: [PATCH 08/19] add test

---
 .../pkg/kafka/config/oauth/k8s_secret.go      |  2 +-
 scheduler/pkg/kafka/config/oauth/store.go     |  2 +-
 .../pkg/kafka/config/oauth/store_test.go      | 94 +++++++++++++++++++
 .../config/oauth/testdata/k8s_secret.yaml     | 12 +++
 4 files changed, 108 insertions(+), 2 deletions(-)
 create mode 100644 scheduler/pkg/kafka/config/oauth/store_test.go
 create mode 100644 scheduler/pkg/kafka/config/oauth/testdata/k8s_secret.yaml

diff --git a/scheduler/pkg/kafka/config/oauth/k8s_secret.go b/scheduler/pkg/kafka/config/oauth/k8s_secret.go
index c6436710a0..7a8154210d 100644
--- a/scheduler/pkg/kafka/config/oauth/k8s_secret.go
+++ b/scheduler/pkg/kafka/config/oauth/k8s_secret.go
@@ -41,7 +41,7 @@ type OAUTHSecretHandler struct {
 	oauthConfig OAUTHConfig
 }
 
-func NewOAUTHSecretHandler(secretName string, clientset kubernetes.Interface, namespace string, prefix string, locationSuffix string, logger log.FieldLogger) (*OAUTHSecretHandler, error) {
+func NewOAUTHSecretHandler(secretName string, clientset kubernetes.Interface, namespace string, prefix string, logger log.FieldLogger) (*OAUTHSecretHandler, error) {
 	if clientset == nil {
 		var err error
 		clientset, err = k8s.CreateClientset()
diff --git a/scheduler/pkg/kafka/config/oauth/store.go b/scheduler/pkg/kafka/config/oauth/store.go
index 53b4e96ec5..5ef2464a2f 100644
--- a/scheduler/pkg/kafka/config/oauth/store.go
+++ b/scheduler/pkg/kafka/config/oauth/store.go
@@ -100,7 +100,7 @@ func NewOAUTHStore(opt ...OAUTHStoreOption) (OAUTHStore, error) {
 		if !ok {
 			return nil, fmt.Errorf("Namespace env var %s not found and needed for OAUTH secret", envNamespace)
 		}
-		ps, err := NewOAUTHSecretHandler(secretName, opts.clientset, namespace, opts.prefix, opts.locationSuffix, logger)
+		ps, err := NewOAUTHSecretHandler(secretName, opts.clientset, namespace, opts.prefix, logger)
 		if err != nil {
 			return nil, err
 		}
diff --git a/scheduler/pkg/kafka/config/oauth/store_test.go b/scheduler/pkg/kafka/config/oauth/store_test.go
new file mode 100644
index 0000000000..35be5858c9
--- /dev/null
+++ b/scheduler/pkg/kafka/config/oauth/store_test.go
@@ -0,0 +1,94 @@
+/*
+Copyright 2023 Seldon Technologies Ltd.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package oauth
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/ghodss/yaml"
+
+	. "github.com/onsi/gomega"
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes/fake"
+)
+
+func unMarshallYamlStrict(data []byte, msg interface{}) error {
+	jsonData, err := yaml.YAMLToJSON(data)
+	if err != nil {
+		return err
+	}
+	d := json.NewDecoder(bytes.NewReader(jsonData))
+	d.DisallowUnknownFields() // So we fail if not exactly as required in schema
+	err = d.Decode(msg)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func moveStringDataToData(secret *v1.Secret) {
+	secret.Data = make(map[string][]byte)
+	for key, val := range secret.StringData {
+		secret.Data[key] = []byte(val)
+	}
+}
+
+func TestNewOAUTHStoreWithSecret(t *testing.T) {
+	g := NewGomegaWithT(t)
+	secretData, err := os.ReadFile("testdata/k8s_secret.yaml")
+	g.Expect(err).To(BeNil())
+
+	secret := &v1.Secret{}
+	err = unMarshallYamlStrict(secretData, secret)
+	g.Expect(err).To(BeNil())
+
+	moveStringDataToData(secret)
+
+	prefix := "prefix"
+
+	t.Setenv(fmt.Sprintf("%s%s", prefix, envSecretSuffix), secret.Name)
+	t.Setenv(envNamespace, secret.Namespace)
+
+	clientset := fake.NewSimpleClientset(secret)
+	ps, err := NewOAUTHStore(Prefix(prefix), ClientSet(clientset))
+	g.Expect(err).To(BeNil())
+
+	oauthConfig := ps.GetOAUTHConfig()
+	g.Expect(oauthConfig.Method).To(Equal("OIDC"))
+	g.Expect(oauthConfig.ClientID).To(Equal("test-client-id"))
+	g.Expect(oauthConfig.ClientSecret).To(Equal("test-client-secret"))
+	g.Expect(oauthConfig.TokenEndpointURL).To(Equal("https://keycloak.example.com/auth/realms/example-realm/protocol/openid-connect/token"))
+	g.Expect(oauthConfig.Extensions).To(Equal("logicalCluster=logic-1234,identityPoolId=pool-1234"))
+
+	newClientID := "new-client-id"
+	secret.Data["client_id"] = []byte(newClientID)
+
+	_, err = clientset.CoreV1().Secrets(secret.Namespace).Update(context.Background(), secret, metav1.UpdateOptions{})
+	g.Expect(err).To(BeNil())
+	time.Sleep(time.Millisecond * 500)
+
+	oauthConfig = ps.GetOAUTHConfig()
+	g.Expect(oauthConfig.ClientID).To(Equal("new-client-id"))
+	ps.Stop()
+}
diff --git a/scheduler/pkg/kafka/config/oauth/testdata/k8s_secret.yaml b/scheduler/pkg/kafka/config/oauth/testdata/k8s_secret.yaml
new file mode 100644
index 0000000000..c17bef970c
--- /dev/null
+++ b/scheduler/pkg/kafka/config/oauth/testdata/k8s_secret.yaml
@@ -0,0 +1,12 @@
+apiVersion: v1
+kind: Secret
+metadata:
+  name: cc-oauth-test-secret
+  namespace: default
+type: Opaque
+stringData:
+  method: OIDC
+  client_id: test-client-id
+  client_secret: test-client-secret
+  token_endpoint_url: https://keycloak.example.com/auth/realms/example-realm/protocol/openid-connect/token
+  extensions: logicalCluster=logic-1234,identityPoolId=pool-1234

From 57c82321f186349701741f3bd7bbc02a4bb53820 Mon Sep 17 00:00:00 2001
From: Rafal Skolasinski <r.j.skolasinski@gmail.com>
Date: Thu, 24 Aug 2023 09:52:28 +0100
Subject: [PATCH 09/19] lint

---
 scheduler/pkg/kafka/config/auth.go             | 1 +
 scheduler/pkg/kafka/config/oauth/store_test.go | 1 -
 2 files changed, 1 insertion(+), 1 deletion(-)

diff --git a/scheduler/pkg/kafka/config/auth.go b/scheduler/pkg/kafka/config/auth.go
index 8ceea4013e..f49657da89 100644
--- a/scheduler/pkg/kafka/config/auth.go
+++ b/scheduler/pkg/kafka/config/auth.go
@@ -24,6 +24,7 @@ import (
 	"github.com/seldonio/seldon-core/components/tls/v2/pkg/password"
 	"github.com/seldonio/seldon-core/components/tls/v2/pkg/tls"
 	"github.com/seldonio/seldon-core/components/tls/v2/pkg/util"
+
 	"github.com/seldonio/seldon-core/scheduler/v2/pkg/kafka/config/oauth"
 )
 
diff --git a/scheduler/pkg/kafka/config/oauth/store_test.go b/scheduler/pkg/kafka/config/oauth/store_test.go
index 35be5858c9..411c438aad 100644
--- a/scheduler/pkg/kafka/config/oauth/store_test.go
+++ b/scheduler/pkg/kafka/config/oauth/store_test.go
@@ -26,7 +26,6 @@ import (
 	"time"
 
 	"github.com/ghodss/yaml"
-
 	. "github.com/onsi/gomega"
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

From 4168809e73827f912eb8b7e6552c028bf661a228 Mon Sep 17 00:00:00 2001
From: Rafal Skolasinski <r.j.skolasinski@gmail.com>
Date: Thu, 24 Aug 2023 10:11:12 +0100
Subject: [PATCH 10/19] go mod tidy

---
 scheduler/go.mod | 1 +
 scheduler/go.sum | 1 +
 2 files changed, 2 insertions(+)

diff --git a/scheduler/go.mod b/scheduler/go.mod
index 2f179378be..3d3dfd6bab 100644
--- a/scheduler/go.mod
+++ b/scheduler/go.mod
@@ -9,6 +9,7 @@ require (
 	github.com/dgraph-io/badger/v3 v3.2103.2
 	github.com/envoyproxy/go-control-plane v0.11.1
 	github.com/fsnotify/fsnotify v1.5.1
+	github.com/ghodss/yaml v1.0.0
 	github.com/go-playground/validator/v10 v10.9.0
 	github.com/golang/protobuf v1.5.3
 	github.com/google/go-cmp v0.5.9
diff --git a/scheduler/go.sum b/scheduler/go.sum
index 38b2f091ba..cbac1c2f3f 100644
--- a/scheduler/go.sum
+++ b/scheduler/go.sum
@@ -226,6 +226,7 @@ github.com/fsnotify/fsnotify v1.5.1 h1:mZcQUHVQUQWoPXXtuf9yuEXKudkV2sx1E06UadKWp
 github.com/fsnotify/fsnotify v1.5.1/go.mod h1:T3375wBYaZdLLcVNkcVbzGHY7f1l/uK5T5Ai1i3InKU=
 github.com/getkin/kin-openapi v0.76.0/go.mod h1:660oXbgy5JFMKreazJaQTw7o+X00qeSyhcnluiMv+Xg=
 github.com/getsentry/raven-go v0.2.0/go.mod h1:KungGk8q33+aIAZUIVWZDr2OfAEBsO49PX4NzFV5kcQ=
+github.com/ghodss/yaml v1.0.0 h1:wQHKEahhL6wmXdzwWG11gIVCkOv05bNOh+Rxn0yngAk=
 github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
 github.com/go-gl/glfw v0.0.0-20190409004039-e6da0acd62b1/go.mod h1:vR7hzQXu2zJy9AVAgeJqvqgH9Q5CA+iKCZ2gyEVpxRU=
 github.com/go-gl/glfw/v3.3/glfw v0.0.0-20191125211704-12ad95a8df72/go.mod h1:tQ2UAYgL5IevRw8kRxooKSPJfGvJ9fJQFa0TUsXzTg8=

From 5e6ea5aef31e45faa93958c26faaf77f1aa76ba0 Mon Sep 17 00:00:00 2001
From: Rafal Skolasinski <r.j.skolasinski@gmail.com>
Date: Thu, 24 Aug 2023 11:16:39 +0100
Subject: [PATCH 11/19] update Kafka from 1.9.1 to 1.9.2

---
 scheduler/go.mod | 2 +-
 scheduler/go.sum | 4 ++--
 2 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/scheduler/go.mod b/scheduler/go.mod
index 3d3dfd6bab..01fe9bae7b 100644
--- a/scheduler/go.mod
+++ b/scheduler/go.mod
@@ -5,7 +5,7 @@ go 1.19
 require (
 	github.com/OneOfOne/xxhash v1.2.8
 	github.com/cenkalti/backoff/v4 v4.1.2
-	github.com/confluentinc/confluent-kafka-go v1.9.1
+	github.com/confluentinc/confluent-kafka-go v1.9.2
 	github.com/dgraph-io/badger/v3 v3.2103.2
 	github.com/envoyproxy/go-control-plane v0.11.1
 	github.com/fsnotify/fsnotify v1.5.1
diff --git a/scheduler/go.sum b/scheduler/go.sum
index cbac1c2f3f..b1e6849aa1 100644
--- a/scheduler/go.sum
+++ b/scheduler/go.sum
@@ -144,8 +144,8 @@ github.com/cockroachdb/datadriven v0.0.0-20200714090401-bf6692d28da5/go.mod h1:h
 github.com/cockroachdb/errors v1.2.4/go.mod h1:rQD95gz6FARkaKkQXUksEje/d9a6wBJoCr5oaCLELYA=
 github.com/cockroachdb/logtags v0.0.0-20190617123548-eb05cc24525f/go.mod h1:i/u985jwjWRlyHXQbwatDASoW0RMlZ/3i9yJHE2xLkI=
 github.com/confluentinc/confluent-kafka-go v1.8.2/go.mod h1:u2zNLny2xq+5rWeTQjFHbDzzNuba4P1vo31r9r4uAdg=
-github.com/confluentinc/confluent-kafka-go v1.9.1 h1:L3aW6KvTyrq/+BOMnDm9xJylhAEoAgqhoaJbMPe3GQI=
-github.com/confluentinc/confluent-kafka-go v1.9.1/go.mod h1:ptXNqsuDfYbAE/LBW6pnwWZElUoWxHoV8E43DCrliyo=
+github.com/confluentinc/confluent-kafka-go v1.9.2 h1:gV/GxhMBUb03tFWkN+7kdhg+zf+QUM+wVkI9zwh770Q=
+github.com/confluentinc/confluent-kafka-go v1.9.2/go.mod h1:ptXNqsuDfYbAE/LBW6pnwWZElUoWxHoV8E43DCrliyo=
 github.com/coreos/bbolt v1.3.2/go.mod h1:iRUV2dpdMOn7Bo10OQBFzIJO9kkE559Wcmn+qkEiiKk=
 github.com/coreos/etcd v3.3.10+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
 github.com/coreos/etcd v3.3.13+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=

From 48a46b50510c97213d24b3f996a11dcd005c52bc Mon Sep 17 00:00:00 2001
From: Rafal Skolasinski <r.j.skolasinski@gmail.com>
Date: Thu, 24 Aug 2023 11:17:51 +0100
Subject: [PATCH 12/19] add scope support for kafka oauth protocol

---
 scheduler/pkg/kafka/config/auth.go                        | 1 +
 scheduler/pkg/kafka/config/oauth/k8s_secret.go            | 7 +++++++
 scheduler/pkg/kafka/config/oauth/oauth.go                 | 2 ++
 scheduler/pkg/kafka/config/oauth/store_test.go            | 1 +
 scheduler/pkg/kafka/config/oauth/testdata/k8s_secret.yaml | 1 +
 5 files changed, 12 insertions(+)

diff --git a/scheduler/pkg/kafka/config/auth.go b/scheduler/pkg/kafka/config/auth.go
index f49657da89..6ccb00c5df 100644
--- a/scheduler/pkg/kafka/config/auth.go
+++ b/scheduler/pkg/kafka/config/auth.go
@@ -131,6 +131,7 @@ func configureSASLSSLOAUTHBEARER(mechanism string, config kafka.ConfigMap) error
 	config["sasl.oauthbearer.method"] = oauthConfig.Method
 	config["sasl.oauthbearer.client.id"] = oauthConfig.ClientID
 	config["sasl.oauthbearer.client.secret"] = oauthConfig.ClientSecret
+	config["sasl.oauthbearer.scope"] = oauthConfig.Scope
 	config["sasl.oauthbearer.token.endpoint.url"] = oauthConfig.TokenEndpointURL
 	config["sasl.oauthbearer.extensions"] = oauthConfig.Extensions
 
diff --git a/scheduler/pkg/kafka/config/oauth/k8s_secret.go b/scheduler/pkg/kafka/config/oauth/k8s_secret.go
index 7a8154210d..513e28f183 100644
--- a/scheduler/pkg/kafka/config/oauth/k8s_secret.go
+++ b/scheduler/pkg/kafka/config/oauth/k8s_secret.go
@@ -91,6 +91,13 @@ func (s *OAUTHSecretHandler) saveOAUTHFromSecret(secret *corev1.Secret) error {
 	}
 	s.oauthConfig.ClientSecret = string(clientSecret)
 
+	// Read and Save oauthbearer scope
+	scope, ok := secret.Data[SecretKeyScope]
+	if !ok {
+		return fmt.Errorf("Failed to find %s in secret %s", SecretKeyScope, secret.Name)
+	}
+	s.oauthConfig.Scope = string(scope)
+
 	// Read and Save oauthbearer token endpoint url
 	tokenEndpointURL, ok := secret.Data[SecretKeyTokenEndpointURL]
 	if !ok {
diff --git a/scheduler/pkg/kafka/config/oauth/oauth.go b/scheduler/pkg/kafka/config/oauth/oauth.go
index eb25995b9a..ccab0af917 100644
--- a/scheduler/pkg/kafka/config/oauth/oauth.go
+++ b/scheduler/pkg/kafka/config/oauth/oauth.go
@@ -20,6 +20,7 @@ const (
 	SecretKeyMethod           = "method"
 	SecretKeyClientID         = "client_id"
 	SecretKeyClientSecret     = "client_secret"
+	SecretKeyScope            = "scope"
 	SecretKeyTokenEndpointURL = "token_endpoint_url"
 	SecretKeyExtensions       = "extensions"
 )
@@ -28,6 +29,7 @@ type OAUTHConfig struct {
 	Method           string
 	ClientID         string
 	ClientSecret     string
+	Scope            string
 	TokenEndpointURL string
 	Extensions       string
 }
diff --git a/scheduler/pkg/kafka/config/oauth/store_test.go b/scheduler/pkg/kafka/config/oauth/store_test.go
index 411c438aad..7f1503471c 100644
--- a/scheduler/pkg/kafka/config/oauth/store_test.go
+++ b/scheduler/pkg/kafka/config/oauth/store_test.go
@@ -77,6 +77,7 @@ func TestNewOAUTHStoreWithSecret(t *testing.T) {
 	g.Expect(oauthConfig.Method).To(Equal("OIDC"))
 	g.Expect(oauthConfig.ClientID).To(Equal("test-client-id"))
 	g.Expect(oauthConfig.ClientSecret).To(Equal("test-client-secret"))
+	g.Expect(oauthConfig.Scope).To(Equal("test scope"))
 	g.Expect(oauthConfig.TokenEndpointURL).To(Equal("https://keycloak.example.com/auth/realms/example-realm/protocol/openid-connect/token"))
 	g.Expect(oauthConfig.Extensions).To(Equal("logicalCluster=logic-1234,identityPoolId=pool-1234"))
 
diff --git a/scheduler/pkg/kafka/config/oauth/testdata/k8s_secret.yaml b/scheduler/pkg/kafka/config/oauth/testdata/k8s_secret.yaml
index c17bef970c..3a93fc810c 100644
--- a/scheduler/pkg/kafka/config/oauth/testdata/k8s_secret.yaml
+++ b/scheduler/pkg/kafka/config/oauth/testdata/k8s_secret.yaml
@@ -10,3 +10,4 @@ stringData:
   client_secret: test-client-secret
   token_endpoint_url: https://keycloak.example.com/auth/realms/example-realm/protocol/openid-connect/token
   extensions: logicalCluster=logic-1234,identityPoolId=pool-1234
+  scope: test scope

From 50af6c43c095f5712c17da4f57f73f2aad267a88 Mon Sep 17 00:00:00 2001
From: Rafal Skolasinski <r.j.skolasinski@gmail.com>
Date: Thu, 31 Aug 2023 16:54:53 +0100
Subject: [PATCH 13/19] reuse secret variable set by
 security.kafka.sasl.client.secret

---
 .../seldon-core-v2-setup/templates/seldon-v2-components.yaml  | 4 ----
 k8s/helm-charts/seldon-core-v2-setup/values.yaml              | 2 --
 k8s/helm-charts/seldon-core-v2-setup/values.yaml.template     | 2 --
 k8s/kustomize/helm-components-sc/patch_modelgateway.yaml      | 2 --
 k8s/kustomize/helm-components-sc/patch_pipelinegateway.yaml   | 2 --
 k8s/yaml/components.yaml                                      | 4 ----
 scheduler/pkg/kafka/config/oauth/store.go                     | 2 +-
 7 files changed, 1 insertion(+), 17 deletions(-)

diff --git a/k8s/helm-charts/seldon-core-v2-setup/templates/seldon-v2-components.yaml b/k8s/helm-charts/seldon-core-v2-setup/templates/seldon-v2-components.yaml
index 41bc6b6b75..65ac3e61ff 100644
--- a/k8s/helm-charts/seldon-core-v2-setup/templates/seldon-v2-components.yaml
+++ b/k8s/helm-charts/seldon-core-v2-setup/templates/seldon-v2-components.yaml
@@ -1030,8 +1030,6 @@ spec:
           value: '{{ .Values.security.kafka.protocol }}'
         - name: KAFKA_SASL_MECHANISM
           value: '{{ .Values.security.kafka.sasl.mechanism }}'
-        - name: KAFKA_CLIENT_OAUTH_SECRET_NAME
-          value: '{{ .Values.security.kafka.sasl.oauth.secret }}'
         - name: KAFKA_CLIENT_TLS_ENDPOINT_IDENTIFICATION_ALGORITHM
           value: '{{ .Values.security.kafka.ssl.client.endpointIdentificationAlgorithm
             }}'
@@ -1186,8 +1184,6 @@ spec:
           value: '{{ .Values.security.kafka.protocol }}'
         - name: KAFKA_SASL_MECHANISM
           value: '{{ .Values.security.kafka.sasl.mechanism }}'
-        - name: KAFKA_CLIENT_OAUTH_SECRET_NAME
-          value: '{{ .Values.security.kafka.sasl.oauth.secret }}'
         - name: KAFKA_CLIENT_TLS_ENDPOINT_IDENTIFICATION_ALGORITHM
           value: '{{ .Values.security.kafka.ssl.client.endpointIdentificationAlgorithm
             }}'
diff --git a/k8s/helm-charts/seldon-core-v2-setup/values.yaml b/k8s/helm-charts/seldon-core-v2-setup/values.yaml
index 6601cab500..306f0a1947 100644
--- a/k8s/helm-charts/seldon-core-v2-setup/values.yaml
+++ b/k8s/helm-charts/seldon-core-v2-setup/values.yaml
@@ -25,8 +25,6 @@ security:
         username: seldon
         secret:
         passwordPath: /tmp/sasl/kafka/client/password
-      oauth:
-        secret:
     ssl:
       client:
         secret:
diff --git a/k8s/helm-charts/seldon-core-v2-setup/values.yaml.template b/k8s/helm-charts/seldon-core-v2-setup/values.yaml.template
index a81f3d8ea0..36ea8f14a7 100644
--- a/k8s/helm-charts/seldon-core-v2-setup/values.yaml.template
+++ b/k8s/helm-charts/seldon-core-v2-setup/values.yaml.template
@@ -25,8 +25,6 @@ security:
         username: seldon
         secret:
         passwordPath: /tmp/sasl/kafka/client/password
-      oauth:
-        secret:
     ssl:
       client:
         secret:
diff --git a/k8s/kustomize/helm-components-sc/patch_modelgateway.yaml b/k8s/kustomize/helm-components-sc/patch_modelgateway.yaml
index 2a9d688851..fc39ed2674 100644
--- a/k8s/kustomize/helm-components-sc/patch_modelgateway.yaml
+++ b/k8s/kustomize/helm-components-sc/patch_modelgateway.yaml
@@ -34,8 +34,6 @@ spec:
             value: '{{ .Values.security.kafka.protocol }}'
           - name: KAFKA_SASL_MECHANISM
             value: '{{ .Values.security.kafka.sasl.mechanism }}'
-          - name: KAFKA_CLIENT_OAUTH_SECRET_NAME
-            value: '{{ .Values.security.kafka.sasl.oauth.secret }}'
           - name: KAFKA_CLIENT_TLS_ENDPOINT_IDENTIFICATION_ALGORITHM
             value: '{{ .Values.security.kafka.ssl.client.endpointIdentificationAlgorithm }}'
           - name: KAFKA_CLIENT_TLS_SECRET_NAME
diff --git a/k8s/kustomize/helm-components-sc/patch_pipelinegateway.yaml b/k8s/kustomize/helm-components-sc/patch_pipelinegateway.yaml
index ec854c090f..17a8c162f4 100644
--- a/k8s/kustomize/helm-components-sc/patch_pipelinegateway.yaml
+++ b/k8s/kustomize/helm-components-sc/patch_pipelinegateway.yaml
@@ -23,8 +23,6 @@ spec:
             value: '{{ .Values.security.kafka.protocol }}'
           - name: KAFKA_SASL_MECHANISM
             value: '{{ .Values.security.kafka.sasl.mechanism }}'
-          - name: KAFKA_CLIENT_OAUTH_SECRET_NAME
-            value: '{{ .Values.security.kafka.sasl.oauth.secret }}'
           - name: KAFKA_CLIENT_TLS_ENDPOINT_IDENTIFICATION_ALGORITHM
             value: '{{ .Values.security.kafka.ssl.client.endpointIdentificationAlgorithm }}'
           - name: KAFKA_CLIENT_TLS_SECRET_NAME
diff --git a/k8s/yaml/components.yaml b/k8s/yaml/components.yaml
index fb753254c4..79859d3545 100644
--- a/k8s/yaml/components.yaml
+++ b/k8s/yaml/components.yaml
@@ -669,8 +669,6 @@ spec:
           value: 'PLAINTEXT'
         - name: KAFKA_SASL_MECHANISM
           value: 'SCRAM-SHA-512'
-        - name: KAFKA_CLIENT_OAUTH_SECRET_NAME
-          value: ''
         - name: KAFKA_CLIENT_TLS_ENDPOINT_IDENTIFICATION_ALGORITHM
           value: ''
         - name: KAFKA_CLIENT_TLS_SECRET_NAME
@@ -818,8 +816,6 @@ spec:
           value: 'PLAINTEXT'
         - name: KAFKA_SASL_MECHANISM
           value: 'SCRAM-SHA-512'
-        - name: KAFKA_CLIENT_OAUTH_SECRET_NAME
-          value: ''
         - name: KAFKA_CLIENT_TLS_ENDPOINT_IDENTIFICATION_ALGORITHM
           value: ''
         - name: KAFKA_CLIENT_TLS_SECRET_NAME
diff --git a/scheduler/pkg/kafka/config/oauth/store.go b/scheduler/pkg/kafka/config/oauth/store.go
index 5ef2464a2f..22e8167f65 100644
--- a/scheduler/pkg/kafka/config/oauth/store.go
+++ b/scheduler/pkg/kafka/config/oauth/store.go
@@ -27,7 +27,7 @@ import (
 )
 
 const (
-	envSecretSuffix = "_OAUTH_SECRET_NAME"
+	envSecretSuffix = "_SASL_SECRET_NAME"
 	envNamespace    = "POD_NAMESPACE"
 )
 

From 9f6c354596aad63e0e98a794c4e055513d9057ea Mon Sep 17 00:00:00 2001
From: Rafal Skolasinski <r.j.skolasinski@gmail.com>
Date: Thu, 31 Aug 2023 16:55:05 +0100
Subject: [PATCH 14/19] remove debug overwrite

---
 scheduler/pkg/kafka/config/auth.go | 4 ----
 1 file changed, 4 deletions(-)

diff --git a/scheduler/pkg/kafka/config/auth.go b/scheduler/pkg/kafka/config/auth.go
index 6ccb00c5df..f9b69da2db 100644
--- a/scheduler/pkg/kafka/config/auth.go
+++ b/scheduler/pkg/kafka/config/auth.go
@@ -54,10 +54,6 @@ func setupSASLSSLAuthentication(config kafka.ConfigMap) error {
 
 	mechanism := tls.GetSASLMechanismFromEnv(tls.EnvSecurityPrefixKafka)
 
-	// TODO: Remove before merge (overwrite for testing)
-	_ = mechanism
-	mechanism = tls.SASLMechanismOAUTHBEARER
-
 	var err error
 	switch mechanism {
 	case tls.SASLMechanismPlain:

From 52aacf5012b63169b9a39cad66be5ebc362c7ef2 Mon Sep 17 00:00:00 2001
From: Rafal Skolasinski <r.j.skolasinski@gmail.com>
Date: Fri, 1 Sep 2023 23:09:34 +0100
Subject: [PATCH 15/19] add documentation

---
 .../security/aws-msk-sasl.md                  |  2 +
 .../security/azure-event-hub-sasl.md          |  2 +
 .../security/confluent-oauth.md               | 65 +++++++++++++++++++
 .../security/confluent-sasl.md                |  5 +-
 .../kubernetes-installation/security/index.md |  2 +
 .../security/strimzi-mtls.md                  |  4 +-
 .../security/strimzi-sasl.md                  |  2 +
 .../source/contents/kubernetes/kafka/index.md |  1 +
 .../values-confluent-kafka-oauth.yaml.tmpl    | 21 ++++++
 .../values-confluent-kafka-sasl.yaml.tmpl     |  2 +-
 10 files changed, 103 insertions(+), 3 deletions(-)
 create mode 100644 docs/source/contents/getting-started/kubernetes-installation/security/confluent-oauth.md
 create mode 100644 k8s/samples/values-confluent-kafka-oauth.yaml.tmpl

diff --git a/docs/source/contents/getting-started/kubernetes-installation/security/aws-msk-sasl.md b/docs/source/contents/getting-started/kubernetes-installation/security/aws-msk-sasl.md
index 5d1a43383b..435136e316 100644
--- a/docs/source/contents/getting-started/kubernetes-installation/security/aws-msk-sasl.md
+++ b/docs/source/contents/getting-started/kubernetes-installation/security/aws-msk-sasl.md
@@ -24,6 +24,8 @@ kubectl create secret generic aws-msk-kafka-secret -n seldon-mesh --from-literal
 
 ## Configure Seldon Core v2
 
+Configure Seldon Core v2 by setting following Helm values:
+
 ```{literalinclude} ../../../../../../k8s/samples/values-aws-msk-kafka-sasl-scram.yaml.tmpl
 :language: yaml
 ```
diff --git a/docs/source/contents/getting-started/kubernetes-installation/security/azure-event-hub-sasl.md b/docs/source/contents/getting-started/kubernetes-installation/security/azure-event-hub-sasl.md
index fce070718c..b15e1ba448 100644
--- a/docs/source/contents/getting-started/kubernetes-installation/security/azure-event-hub-sasl.md
+++ b/docs/source/contents/getting-started/kubernetes-installation/security/azure-event-hub-sasl.md
@@ -46,6 +46,8 @@ kubectl create secret generic azure-kafka-secret -n seldon-mesh --from-literal p
 
 ## Configure Seldon Core v2
 
+Configure Seldon Core v2 by setting following Helm values:
+
 ```{literalinclude} ../../../../../../k8s/samples/values-azure-event-hub-sasl.yaml.tmpl
 :language: yaml
 ```
diff --git a/docs/source/contents/getting-started/kubernetes-installation/security/confluent-oauth.md b/docs/source/contents/getting-started/kubernetes-installation/security/confluent-oauth.md
new file mode 100644
index 0000000000..932d64a0e6
--- /dev/null
+++ b/docs/source/contents/getting-started/kubernetes-installation/security/confluent-oauth.md
@@ -0,0 +1,65 @@
+# Confluent Cloud Oauth 2.0 Example
+
+> New in Seldon Core 2.7.0
+
+Seldon Core v2 can integrate with Confluent Cloud managed Kafka.
+In this example we use [Oauth 2.0 security mechanism](https://docs.confluent.io/cloud/current/access-management/authenticate/oauth/overview.html).
+
+
+## Configure Identity Provider in Confluent Cloud Console
+
+In your Confluent Cloud Console go to [Account & Access / Identity providers](https://confluent.cloud/settings/org/identity_providers) and register your Identity Provider.
+
+
+See Confluent Cloud [documentation](https://docs.confluent.io/cloud/current/access-management/authenticate/oauth/identity-providers.html) for further details.
+
+
+## Configure Identity Pool
+
+In your Confluent Cloud Console go to [Account & Access / Identity providers](https://confluent.cloud/settings/org/identity_providers) and add new identity pool to your newly registered Identity Provider.
+
+See Confluent Cloud [documentation](https://docs.confluent.io/cloud/current/access-management/authenticate/oauth/identity-pools.html) for further details.
+
+
+## Create Kubernetes Secret
+
+Seldon Core v2 expects oauth credentials to be in form of K8s secret
+```yaml
+apiVersion: v1
+kind: Secret
+metadata:
+  name: confluent-kafka-oauth
+  namespace: seldon-mesh
+type: Opaque
+stringData:
+  method: OIDC
+  client_id: <client id>
+  client_secret: <client secret>
+  token_endpoint_url: <token endpoint url>
+  extensions: logicalCluster=<cluster id>,identityPoolId=<identity pool id>
+  scope: test scope
+```
+
+You will need following information from Confluent Cloud:
+- Cluster ID: `Cluster Overview` → `Cluster Settings` → `General` → `Identification`
+- Identity Pool ID: `Accounts & access` → `Identity providers` → `<specific provider details>`
+
+Client ID, client secret and token endpoint url should come from identity provider, e.g. Keycloak or Azure AD.
+
+
+## Configure Seldon Core v2
+
+Configure Seldon Core v2 by setting following Helm values:
+
+```{literalinclude} ../../../../../../k8s/samples/values-confluent-kafka-oauth.yaml.tmpl
+:language: yaml
+```
+
+Note you may need to tweak `replicationFactor` and `numPartitions` to your cluster configuration.
+
+
+## Troubleshooting
+
+- First check Confluent Cloud [documentation](https://docs.confluent.io/cloud/current/overview.html).
+
+- Set the kafka config map debug setting to `all`. For Helm install you can set `kafka.debug=all`.
diff --git a/docs/source/contents/getting-started/kubernetes-installation/security/confluent-sasl.md b/docs/source/contents/getting-started/kubernetes-installation/security/confluent-sasl.md
index ce4190a5fb..b904a29b15 100644
--- a/docs/source/contents/getting-started/kubernetes-installation/security/confluent-sasl.md
+++ b/docs/source/contents/getting-started/kubernetes-installation/security/confluent-sasl.md
@@ -3,6 +3,7 @@
 > New in Seldon Core 2.5.0
 
 Seldon Core v2 can integrate with Confluent Cloud managed Kafka.
+In this example we use SASL security mechanism.
 
 
 ## Create API Keys
@@ -22,11 +23,13 @@ See Confluent Cloud [documentation](https://docs.confluent.io/cloud/current/clie
 
 Seldon Core v2 expects password to be in form of K8s secret
 ```bash
-kubectl create secret generic confluent-kafka-secret -n seldon-mesh --from-literal password="<Confluent Cloud API Secret>"
+kubectl create secret generic confluent-kafka-sasl -n seldon-mesh --from-literal password="<Confluent Cloud API Secret>"
 ```
 
 ## Configure Seldon Core v2
 
+Configure Seldon Core v2 by setting following Helm values:
+
 ```{literalinclude} ../../../../../../k8s/samples/values-confluent-kafka-sasl.yaml.tmpl
 :language: yaml
 ```
diff --git a/docs/source/contents/getting-started/kubernetes-installation/security/index.md b/docs/source/contents/getting-started/kubernetes-installation/security/index.md
index 1b28a5728e..7d13cc868b 100644
--- a/docs/source/contents/getting-started/kubernetes-installation/security/index.md
+++ b/docs/source/contents/getting-started/kubernetes-installation/security/index.md
@@ -37,6 +37,7 @@ Examples are shown below:
  * [SASL PLAIN with Azure Event Hub](azure-event-hub-sasl.md) example
  * [SASL SCRAM with Strimzi](strimzi-sasl.md) example
  * [SASL SCRAM with AWS MSK](aws-msk-sasl.md) example
+ * [SASL OAUTH with Confluent Cloud](confluent-oauth.md) example
 
 ## Data Plane
 
@@ -95,6 +96,7 @@ helm install seldon-v2-certs k8s/helm-charts/seldon-core-v2-certs/ -n seldon-mes
 strimzi-mtls.md
 strimzi-sasl.md
 confluent-sasl.md
+confluent-oauth.md
 azure-event-hub-sasl.md
 aws-msk-sasl.md
 aws-msk-mtls.md
diff --git a/docs/source/contents/getting-started/kubernetes-installation/security/strimzi-mtls.md b/docs/source/contents/getting-started/kubernetes-installation/security/strimzi-mtls.md
index c76b011dc7..5d438e92f8 100644
--- a/docs/source/contents/getting-started/kubernetes-installation/security/strimzi-mtls.md
+++ b/docs/source/contents/getting-started/kubernetes-installation/security/strimzi-mtls.md
@@ -32,6 +32,8 @@ kubectl create -f k8s/samples/strimzi-example-tls-user.yaml -n seldon-mesh
 
 Install seldon with the Strimzi certificate secrets using a custom values file. This sets the secret created by Strimzi for the user created above (`seldon`) and targets the server certificate authority secret from the name of the cluster created on install of the Kafka cluster (`seldon-cluster-ca-cert`).
 
+Configure Seldon Core v2 by setting following Helm values:
+
 ```{literalinclude} ../../../../../../k8s/samples/values-strimzi-kafka-mtls.yaml
 :language: yaml
 ```
@@ -45,4 +47,4 @@ You can now go ahead and install a SeldonRuntime in your desired install namespa
 
 ```
 helm install seldon-v2-runtime ../k8s/helm-charts/seldon-core-v2-runtime  -n seldon-mesh
-```
\ No newline at end of file
+```
diff --git a/docs/source/contents/getting-started/kubernetes-installation/security/strimzi-sasl.md b/docs/source/contents/getting-started/kubernetes-installation/security/strimzi-sasl.md
index abbef9af75..c7c7a1a071 100644
--- a/docs/source/contents/getting-started/kubernetes-installation/security/strimzi-sasl.md
+++ b/docs/source/contents/getting-started/kubernetes-installation/security/strimzi-sasl.md
@@ -18,6 +18,8 @@ This will call the Strimzi cluster Helm chart provided by the project with overr
 Install Core v2 with SASL settings using a custom values file.
 This sets the secret created by Strimzi for the user created above (`seldon`) and targets the server certificate authority secret from the name of the cluster created on install of the Kafka cluster (`seldon-cluster-ca-cert`).
 
+Configure Seldon Core v2 by setting following Helm values:
+
 ```{literalinclude} ../../../../../../k8s/samples/values-strimzi-kafka-sasl-scram.yaml
 :language: yaml
 ```
diff --git a/docs/source/contents/kubernetes/kafka/index.md b/docs/source/contents/kubernetes/kafka/index.md
index 09a360f3b7..7d8382a3de 100644
--- a/docs/source/contents/kubernetes/kafka/index.md
+++ b/docs/source/contents/kubernetes/kafka/index.md
@@ -18,6 +18,7 @@ This allow to take away all the complexity on running secure and scalable Kafka
 
 We currently have tested and documented integration with following managed solutions:
 - Confluent Cloud (security: SASL/PLAIN)
+- Confluent Cloud (security: OAuth 2.0)
 - Amazon MSK (security: mTLS)
 - Amazon MSK (security: SASL/SCRAM)
 - Azure Event Hub (security: SASL/PLAIN)
diff --git a/k8s/samples/values-confluent-kafka-oauth.yaml.tmpl b/k8s/samples/values-confluent-kafka-oauth.yaml.tmpl
new file mode 100644
index 0000000000..9ef23e1be8
--- /dev/null
+++ b/k8s/samples/values-confluent-kafka-oauth.yaml.tmpl
@@ -0,0 +1,21 @@
+kafka:
+  bootstrap: < Confluent Cloud Broker Endpoints >
+  topics:
+    replicationFactor: 3
+    numPartitions: 4
+  consumer:
+    messageMaxBytes: 8388608
+  producer:
+    messageMaxBytes: 8388608
+
+security:
+  kafka:
+    protocol: SASL_SSL
+    sasl:
+    mechanism: OAUTHBEARER
+    client:
+        secret: confluent-kafka-oauth
+    ssl:
+      client:
+        secret:
+        brokerValidationSecret:
diff --git a/k8s/samples/values-confluent-kafka-sasl.yaml.tmpl b/k8s/samples/values-confluent-kafka-sasl.yaml.tmpl
index 8374cffa79..7ba91559d8 100644
--- a/k8s/samples/values-confluent-kafka-sasl.yaml.tmpl
+++ b/k8s/samples/values-confluent-kafka-sasl.yaml.tmpl
@@ -15,7 +15,7 @@ security:
       mechanism: "PLAIN"
       client:
         username: < username >
-        secret: confluent-kafka-secret
+        secret: confluent-kafka-sasl
     ssl:
       client:
         secret:

From 20203d8f91c1311db12231087baf65f9bac443ef Mon Sep 17 00:00:00 2001
From: Rafal Skolasinski <r.j.skolasinski@gmail.com>
Date: Fri, 1 Sep 2023 23:34:33 +0100
Subject: [PATCH 16/19] minor fixes

---
 k8s/samples/values-confluent-kafka-oauth.yaml.tmpl | 6 +++---
 scheduler/Dockerfile.modelgateway                  | 2 +-
 scheduler/Dockerfile.pipelinegateway               | 2 +-
 3 files changed, 5 insertions(+), 5 deletions(-)

diff --git a/k8s/samples/values-confluent-kafka-oauth.yaml.tmpl b/k8s/samples/values-confluent-kafka-oauth.yaml.tmpl
index 9ef23e1be8..64ec232746 100644
--- a/k8s/samples/values-confluent-kafka-oauth.yaml.tmpl
+++ b/k8s/samples/values-confluent-kafka-oauth.yaml.tmpl
@@ -12,9 +12,9 @@ security:
   kafka:
     protocol: SASL_SSL
     sasl:
-    mechanism: OAUTHBEARER
-    client:
-        secret: confluent-kafka-oauth
+      mechanism: OAUTHBEARER
+      client:
+          secret: confluent-kafka-oauth
     ssl:
       client:
         secret:
diff --git a/scheduler/Dockerfile.modelgateway b/scheduler/Dockerfile.modelgateway
index c325a57dac..f7cb00b31d 100644
--- a/scheduler/Dockerfile.modelgateway
+++ b/scheduler/Dockerfile.modelgateway
@@ -13,7 +13,7 @@ FROM registry.access.redhat.com/ubi9/ubi-minimal as certs
 # Kafka dependencies necessitate leaving CGo enabled and using a base image with C dependencies
 FROM registry.access.redhat.com/ubi9/ubi-micro:9.2-9
 
-#Kafka OIDC token retriva certs: https://github.com/confluentinc/librdkafka/issues/3751
+# Kafka OIDC token retriva certs: https://github.com/confluentinc/librdkafka/issues/3751
 COPY --from=certs /etc/ssl/certs/ca-bundle.crt /etc/ssl/certs/ca-certificates.crt
 
 # Broker Certificates
diff --git a/scheduler/Dockerfile.pipelinegateway b/scheduler/Dockerfile.pipelinegateway
index e61962281a..faccaaee44 100644
--- a/scheduler/Dockerfile.pipelinegateway
+++ b/scheduler/Dockerfile.pipelinegateway
@@ -13,7 +13,7 @@ FROM registry.access.redhat.com/ubi9/ubi-minimal as certs
 # Kafka dependencies necessitate leaving CGo enabled and using a base image with C dependencies
 FROM registry.access.redhat.com/ubi9/ubi-micro:9.2-9
 
-#Kafka OIDC token retriva certs: https://github.com/confluentinc/librdkafka/issues/3751
+# Kafka OIDC token retriva certs: https://github.com/confluentinc/librdkafka/issues/3751
 COPY --from=certs /etc/ssl/certs/ca-bundle.crt /etc/ssl/certs/ca-certificates.crt
 
 # Broker Certificates

From b42b83cd2bfa9d766c355d8d29d2952636ac6145 Mon Sep 17 00:00:00 2001
From: Rafal Skolasinski <r.j.skolasinski@gmail.com>
Date: Mon, 4 Sep 2023 14:29:43 +0100
Subject: [PATCH 17/19] Update
 docs/source/contents/getting-started/kubernetes-installation/security/confluent-oauth.md

---
 .../kubernetes-installation/security/confluent-oauth.md         | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/docs/source/contents/getting-started/kubernetes-installation/security/confluent-oauth.md b/docs/source/contents/getting-started/kubernetes-installation/security/confluent-oauth.md
index 932d64a0e6..493ff1aaa4 100644
--- a/docs/source/contents/getting-started/kubernetes-installation/security/confluent-oauth.md
+++ b/docs/source/contents/getting-started/kubernetes-installation/security/confluent-oauth.md
@@ -37,7 +37,7 @@ stringData:
   client_secret: <client secret>
   token_endpoint_url: <token endpoint url>
   extensions: logicalCluster=<cluster id>,identityPoolId=<identity pool id>
-  scope: test scope
+  scope: api://<client id>/.default
 ```
 
 You will need following information from Confluent Cloud:

From f968852515492bf050cecc5a0a40dbbe9e9466e3 Mon Sep 17 00:00:00 2001
From: Rafal Skolasinski <r.j.skolasinski@gmail.com>
Date: Mon, 4 Sep 2023 14:43:20 +0100
Subject: [PATCH 18/19] make scope generic here

---
 .../kubernetes-installation/security/confluent-oauth.md         | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/docs/source/contents/getting-started/kubernetes-installation/security/confluent-oauth.md b/docs/source/contents/getting-started/kubernetes-installation/security/confluent-oauth.md
index 493ff1aaa4..933335e9b1 100644
--- a/docs/source/contents/getting-started/kubernetes-installation/security/confluent-oauth.md
+++ b/docs/source/contents/getting-started/kubernetes-installation/security/confluent-oauth.md
@@ -37,7 +37,7 @@ stringData:
   client_secret: <client secret>
   token_endpoint_url: <token endpoint url>
   extensions: logicalCluster=<cluster id>,identityPoolId=<identity pool id>
-  scope: api://<client id>/.default
+  scope: openid
 ```
 
 You will need following information from Confluent Cloud:

From 0a868d8b67c808485fc2b5c443610ad86684e38b Mon Sep 17 00:00:00 2001
From: Rafal Skolasinski <r.j.skolasinski@gmail.com>
Date: Mon, 4 Sep 2023 14:51:00 +0100
Subject: [PATCH 19/19] leave scope as empty to be set only when required

---
 .../kubernetes-installation/security/confluent-oauth.md         | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/docs/source/contents/getting-started/kubernetes-installation/security/confluent-oauth.md b/docs/source/contents/getting-started/kubernetes-installation/security/confluent-oauth.md
index 933335e9b1..fb9e5c09a5 100644
--- a/docs/source/contents/getting-started/kubernetes-installation/security/confluent-oauth.md
+++ b/docs/source/contents/getting-started/kubernetes-installation/security/confluent-oauth.md
@@ -37,7 +37,7 @@ stringData:
   client_secret: <client secret>
   token_endpoint_url: <token endpoint url>
   extensions: logicalCluster=<cluster id>,identityPoolId=<identity pool id>
-  scope: openid
+  scope: ""
 ```
 
 You will need following information from Confluent Cloud: