Merge pull request #2018 from verdel/fix/rdkafka_options

fix: change rdkafka_options format from section to hash

Author: pepov

pkg/sdk/logging/model/output/kafka.go

@@ -15,6 +15,8 @@
 package output
 
 import (
+	"encoding/json"
+
 	"github.com/cisco-open/operator-tools/pkg/secret"
 
 	"github.com/kube-logging/logging-operator/pkg/sdk/logging/model/types"
@@ -355,6 +357,21 @@ func (e *KafkaOutputConfig) ToDirective(secretLoader secret.SecretLoader, id str
 	} else {
 		kafka.Params = params
 	}
+
+	if e.RdkafkaOptions != nil {
+		if rdkafkaOptions, err := types.NewStructToStringMapper(secretLoader).StringsMap(e.RdkafkaOptions); err != nil {
+			return nil, err
+		} else {
+			if len(rdkafkaOptions) > 0 {
+				marshaledRdkafkaOptions, err := json.Marshal(rdkafkaOptions)
+				if err != nil {
+					return nil, err
+				}
+				kafka.Params["rdkafka_options"] = string(marshaledRdkafkaOptions)
+			}
+		}
+	}
+
 	if e.Buffer == nil {
 		e.Buffer = &Buffer{}
 	}
@@ -363,13 +380,6 @@ func (e *KafkaOutputConfig) ToDirective(secretLoader secret.SecretLoader, id str
 	} else {
 		kafka.SubDirectives = append(kafka.SubDirectives, buffer)
 	}
-	if e.RdkafkaOptions != nil {
-		if rdkafkaOptions, err := e.RdkafkaOptions.ToDirective(secretLoader, id); err != nil {
-			return nil, err
-		} else {
-			kafka.SubDirectives = append(kafka.SubDirectives, rdkafkaOptions)
-		}
-	}
 
 	if e.Format != nil {
 		if format, err := e.Format.ToDirective(secretLoader, ""); err != nil {
@@ -383,9 +393,3 @@ func (e *KafkaOutputConfig) ToDirective(secretLoader secret.SecretLoader, id str
 	delete(kafka.Params, "use_rdkafka")
 	return kafka, nil
 }
-
-func (o *RdkafkaOptions) ToDirective(secretLoader secret.SecretLoader, id string) (types.Directive, error) {
-	return types.NewFlatDirective(types.PluginMeta{
-		Directive: "rdkafka_options",
-	}, o, secretLoader)
-}

pkg/sdk/logging/model/output/kafka_test.go

@@ -77,7 +77,7 @@ rdkafka_options:
   ssl.ca.location: /etc/ssl/certs/ca-certificates.crt
   ssl.certificate.location: /etc/ssl/certs/tls.crt
   ssl.key.location: /etc/ssl/certs/tls.key
-  ssl.key.password: password 
+  ssl.key.password: password
 format:
   type: json
 buffer:
@@ -91,6 +91,7 @@ buffer:
     @id test
     brokers kafka-headless.kafka.svc.cluster.local:29092
     default_topic topic
+    rdkafka_options {"sasl.mechanisms":"PLAIN","sasl.username":"user","security.protocol":"SASL_SSL","ssl.ca.location":"/etc/ssl/certs/ca-certificates.crt","ssl.certificate.location":"/etc/ssl/certs/tls.crt","ssl.key.location":"/etc/ssl/certs/tls.key","ssl.key.password":"password"}
     ssl_verify_hostname false
     <buffer tag,time>
       @type file
@@ -100,15 +101,6 @@ buffer:
       timekey_use_utc true
       timekey_wait 30s
     </buffer>
-    <rdkafka_options>
-      sasl.mechanisms PLAIN
-      sasl.username user
-      security.protocol SASL_SSL
-      ssl.ca.location /etc/ssl/certs/ca-certificates.crt
-      ssl.certificate.location /etc/ssl/certs/tls.crt
-      ssl.key.location /etc/ssl/certs/tls.key
-      ssl.key.password password
-    </rdkafka_options>
     <format>
       @type json
     </format>