Seldon add ssl

executor/.gitignore

@@ -10,3 +10,4 @@ openapi/
 executor/api/rest/openapi/
 triton-inference-server/	
 _operator
+certs/

executor/Dockerfile.executor

@@ -17,6 +17,7 @@ COPY api/ api/
 COPY predictor/ predictor/
 COPY logger/ logger/
 COPY k8s/ k8s/
+COPY certs/ certs/
 
 # Build
 RUN go build -a -o executor cmd/executor/main.go
@@ -55,6 +56,7 @@ RUN chmod -R 660 /openapi/
 FROM gcr.io/distroless/base-debian10:latest
 WORKDIR /
 COPY --from=builder /workspace/executor .
+COPY --from=builder /workspace/certs/ /certs/
 COPY licenses/license.txt licenses/license.txt
 COPY --from=builder /workspace/*.tar.gz licenses/mpl_source/
 

executor/Makefile

@@ -8,7 +8,7 @@ IMG_REDHAT ?= seldonio/${IMG_VERSION_REDHAT}
 
 EXECUTOR_FOLDERS ?= ./api/... ./predictor/... ./k8s/... ./logger/...
 
-KIND_NAME ?= kind
+KIND_NAME ?= ansible
 
 # Get the currently used golang install path (in GOPATH/bin, unless GOBIN is set)
 ifeq (,$(shell go env GOBIN))

executor/api/kafka/server.go

@@ -2,6 +2,13 @@ package kafka
 
 import (
 	"fmt"
+	"net/url"
+	"os"
+	"os/signal"
+	"reflect"
+	"syscall"
+	"time"
+
 	"github.com/cloudevents/sdk-go/pkg/bindings/http"
 	"github.com/confluentinc/confluent-kafka-go/kafka"
 	"github.com/go-logr/logr"
@@ -13,14 +20,9 @@ import (
 	"github.com/seldonio/seldon-core/executor/api/grpc/tensorflow"
 	"github.com/seldonio/seldon-core/executor/api/payload"
 	"github.com/seldonio/seldon-core/executor/api/rest"
+	"github.com/seldonio/seldon-core/executor/api/util"
 	"github.com/seldonio/seldon-core/executor/predictor"
 	v1 "github.com/seldonio/seldon-core/operator/apis/machinelearning.seldon.io/v1"
-	"net/url"
-	"os"
-	"os/signal"
-	"reflect"
-	"syscall"
-	"time"
 )
 
 const (
@@ -29,13 +31,22 @@ const (
 )
 
 const (
-	ENV_KAFKA_BROKER       = "KAFKA_BROKER"
-	ENV_KAFKA_INPUT_TOPIC  = "KAFKA_INPUT_TOPIC"
-	ENV_KAFKA_OUTPUT_TOPIC = "KAFKA_OUTPUT_TOPIC"
-	ENV_KAFKA_FULL_GRAPH   = "KAFKA_FULL_GRAPH"
-	ENV_KAFKA_WORKERS      = "KAFKA_WORKERS"
+	ENV_KAFKA_BROKER            = "KAFKA_BROKER"
+	ENV_KAFKA_INPUT_TOPIC       = "KAFKA_INPUT_TOPIC"
+	ENV_KAFKA_OUTPUT_TOPIC      = "KAFKA_OUTPUT_TOPIC"
+	ENV_KAFKA_FULL_GRAPH        = "KAFKA_FULL_GRAPH"
+	ENV_KAFKA_WORKERS           = "KAFKA_WORKERS"
+	ENV_KAFKA_SECURITY_PROTOCOL = "KAFKA_SECURITY_PROTOCOL"
 )
 
+type SslKakfa struct {
+	kafkaSslClientCertFile string
+	kafkaSslClientKeyFile  string
+	kafkaSslCACertFile     string
+	kafkaSecurityProtocol  string
+	kafkaSslClientKeyPass  string
+}
+
 type SeldonKafkaServer struct {
 	Client         client.SeldonApiClient
 	Producer       *kafka.Producer
@@ -49,6 +60,18 @@ type SeldonKafkaServer struct {
 	ServerUrl      *url.URL
 	Workers        int
 	Log            logr.Logger
+	// KafkaSecurityProtocol string
+}
+
+func getSslElements() *SslKakfa {
+	sslElements := SslKakfa{
+		kafkaSslClientCertFile: util.GetEnv("KAFKA_SSL_CLIENT_CERT_FILE", ""),
+		kafkaSslClientKeyFile:  util.GetEnv("KAFKA_SSL_CLIENT_KEY_FILE", ""),
+		kafkaSslCACertFile:     util.GetEnv("KAFKA_SSL_CA_CERT_FILE", ""),
+		kafkaSecurityProtocol:  util.GetEnv("KAFKA_SECURITY_PROTOCOL", ""),
+		kafkaSslClientKeyPass:  util.GetEnv("KAFKA_SSL_CLIENT_KEY_PASS", ""),
+	}
+	return &sslElements
 }
 
 func NewKafkaServer(fullGraph bool, workers int, deploymentName, namespace, protocol, transport string, annotations map[string]string, serverUrl *url.URL, predictor *v1.PredictorSpec, broker, topicIn, topicOut string, log logr.Logger) (*SeldonKafkaServer, error) {
@@ -76,12 +99,24 @@ func NewKafkaServer(fullGraph bool, workers int, deploymentName, namespace, prot
 			return nil, fmt.Errorf("Unknown transport %s", transport)
 		}
 	}
+	sslKakfaServer := getSslElements()
+	var producerConfigMap = kafka.ConfigMap{"bootstrap.servers": broker,
+		"go.delivery.reports": false, // Need this othewise will get memory leak
+	}
+	if broker != "" {
+		if sslKakfaServer.kafkaSecurityProtocol != "" {
+			// producerConfigMap["debug"] = "security,broker,protocol,metadata,topic"
+			producerConfigMap["security.protocol"] = sslKakfaServer.kafkaSecurityProtocol
+			producerConfigMap["ssl.ca.location"] = sslKakfaServer.kafkaSslCACertFile
+			producerConfigMap["ssl.key.location"] = sslKakfaServer.kafkaSslClientKeyFile
+			producerConfigMap["ssl.certificate.location"] = sslKakfaServer.kafkaSslClientCertFile
+			producerConfigMap["ssl.key.password"] = sslKakfaServer.kafkaSslClientKeyPass // Key password, if any
 
+		}
+	}
 	// Create Producer
 	log.Info("Creating producer", "broker", broker)
-	p, err := kafka.NewProducer(&kafka.ConfigMap{"bootstrap.servers": broker,
-		"go.delivery.reports": false, // Need this othewise will get memory leak
-	})
+	p, err := kafka.NewProducer(&producerConfigMap)
 	if err != nil {
 		return nil, err
 	}

executor/logger/dispatcher.go

@@ -1,15 +1,29 @@
 package logger
 
 import (
-	"github.com/go-logr/logr"
 	"os"
+
+	"github.com/go-logr/logr"
+	"github.com/seldonio/seldon-core/executor/api/util"
 )
 
 const (
 	ENV_LOGGER_KAFKA_BROKER = "LOGGER_KAFKA_BROKER"
 	ENV_LOGGER_KAFKA_TOPIC  = "LOGGER_KAFKA_TOPIC"
 )
 
+var WorkerQueue chan chan LogRequest
+
+func getSslElements() *SslKakfa {
+	sslElements := SslKakfa{
+		kafkaSslClientCertFile: util.GetEnv("KAFKA_SSL_CLIENT_CERT_FILE", ""),
+		kafkaSslClientKeyFile:  util.GetEnv("KAFKA_SSL_CLIENT_KEY_FILE", ""),
+		kafkaSslCACertFile:     util.GetEnv("KAFKA_SSL_CA_CERT_FILE", ""),
+		kafkaSecurityProtocol:  util.GetEnv("KAFKA_SECURITY_PROTOCOL", ""),
+		kafkaSslClientKeyPass:  util.GetEnv("KAFKA_SSL_CLIENT_KEY_PASS", ""),
+	}
+	return &sslElements
+}
 func StartDispatcher(nworkers int, logBufferSize int, writeTimeoutMs int, log logr.Logger, sdepName string, namespace string, predictorName string, kafkaBroker string, kafkaTopic string) error {
 	if kafkaBroker == "" {
 		kafkaBroker = os.Getenv(ENV_LOGGER_KAFKA_BROKER)
@@ -22,14 +36,24 @@ func StartDispatcher(nworkers int, logBufferSize int, writeTimeoutMs int, log lo
 			kafkaTopic = "seldon"
 		}
 	}
+	sslKakfa := getSslElements()
+	log.Info("kafkaSslClientCertFile", "clientcertfile", sslKakfa.kafkaSslClientCertFile)
+	if sslKakfa.kafkaSslClientCertFile != "" && sslKakfa.kafkaSslClientKeyFile != "" && sslKakfa.kafkaSslCACertFile != "" {
+		if sslKakfa.kafkaSecurityProtocol == "" {
+			sslKakfa.kafkaSecurityProtocol = "ssl"
+		}
 
+		// if kafkaSecurityProtocol != "ssl" && kafkaSecurityProtocol != "sasl_ssl" {
+		// 	log.Error("invalid config: kafka security protocol is not ssl based but ssl config is provided")
+		// }
+	}
 	workQueue = make(chan LogRequest, logBufferSize)
 	writeTimeoutMilliseconds = writeTimeoutMs
 
 	// Now, create all of our workers.
 	for i := 0; i < nworkers; i++ {
 		log.Info("Starting", "worker", i+1)
-		worker, err := NewWorker(i+1, workQueue, log, sdepName, namespace, predictorName, kafkaBroker, kafkaTopic)
+		worker, err := NewWorker(i+1, workQueue, log, sdepName, namespace, predictorName, kafkaBroker, kafkaTopic, *sslKakfa)
 		if err != nil {
 			return err
 		}

executor/logger/types.go

@@ -23,3 +23,11 @@ type LogRequest struct {
 	ModelId         string
 	RequestId       string
 }
+
+type SslKakfa struct {
+	kafkaSslClientCertFile string
+	kafkaSslClientKeyFile  string
+	kafkaSslCACertFile     string
+	kafkaSecurityProtocol  string
+	kafkaSslClientKeyPass  string
+}

executor/logger/worker.go

@@ -31,16 +31,29 @@ const (
 // NewWorker creates, and returns a new Worker object. Its only argument
 // is a channel that the worker can add itself to whenever it is done its
 // work.
-func NewWorker(id int, workQueue chan LogRequest, log logr.Logger, sdepName string, namespace string, predictorName string, kafkaBroker string, kafkaTopic string) (*Worker, error) {
+func NewWorker(id int, workQueue chan LogRequest, log logr.Logger, sdepName string, namespace string, predictorName string, kafkaBroker string, kafkaTopic string, sslKakfa SslKakfa) (*Worker, error) {
 
 	var producer *kafka.Producer
 	var err error
 	if kafkaBroker != "" {
 		log.Info("Creating producer", "broker", kafkaBroker, "topic", kafkaTopic)
-		producer, err = kafka.NewProducer(&kafka.ConfigMap{
-			"bootstrap.servers":   kafkaBroker,
+		var producerConfigMap = kafka.ConfigMap{"bootstrap.servers": kafkaBroker,
 			"go.delivery.reports": false, // Need this othewise will get memory leak
-		})
+		}
+		log.Info("kafkaSecurityProtocol", "kafkaSecurityProtocol", sslKakfa.kafkaSecurityProtocol)
+		if kafkaBroker != "" {
+			if sslKakfa.kafkaSecurityProtocol != "" {
+
+				// producerConfigMap["debug"] = "security,broker,protocol,metadata,topic"
+				producerConfigMap["security.protocol"] = sslKakfa.kafkaSecurityProtocol
+				producerConfigMap["ssl.ca.location"] = sslKakfa.kafkaSslCACertFile
+				producerConfigMap["ssl.key.location"] = sslKakfa.kafkaSslClientKeyFile
+				producerConfigMap["ssl.certificate.location"] = sslKakfa.kafkaSslClientCertFile
+				producerConfigMap["ssl.key.password"] = sslKakfa.kafkaSslClientKeyPass // Key password, if any
+
+			}
+		}
+		producer, err = kafka.NewProducer(&producerConfigMap)
 		if err != nil {
 			return nil, err
 		}
@@ -94,7 +107,7 @@ func getCEType(logReq LogRequest) (string, error) {
 }
 
 func (w *Worker) sendKafkaEvent(logReq LogRequest) error {
-
+	w.Log.Info("SENDING KAFKA EVENT", "LogReq", logReq)
 	reqType, err := getCEType(logReq)
 	if err != nil {
 		return err
@@ -109,12 +122,15 @@ func (w *Worker) sendKafkaEvent(logReq LogRequest) error {
 		{Key: NamespaceAttr, Value: []byte(w.Namespace)},
 		{Key: EndpointAttr, Value: []byte(w.PredictorName)},
 	}
-
+	w.Log.Info("kafkaHeaders is", "kafkaHeaders", kafkaHeaders)
 	err = w.Producer.Produce(&kafka.Message{
 		TopicPartition: kafka.TopicPartition{Topic: &w.KafkaTopic, Partition: kafka.PartitionAny},
 		Value:          *logReq.Bytes,
 		Headers:        kafkaHeaders,
 	}, nil)
+	w.Log.Info("Topic Partition:", "topic", kafka.TopicPartition{Topic: &w.KafkaTopic, Partition: kafka.PartitionAny})
+	w.Log.Info("Value:", "logReqBytes", *logReq.Bytes)
+	w.Log.Info("Kafka Headers:", "headers", kafkaHeaders)
 	if err != nil {
 		w.Log.Error(err, "Failed to produce response")
 		return err
@@ -187,6 +203,7 @@ func (w *Worker) Start() {
 				// Receive a work request.
 
 				if w.KafkaTopic != "" {
+					w.Log.Info("KAFKA TOPIC IS NOT NULL", "Topic", w.KafkaTopic)
 					if err := w.sendKafkaEvent(work); err != nil {
 						w.Log.Error(err, "Failed to send kafka log", "Topic", w.KafkaTopic)
 					}