chore: fix rdkafka options tests

Signed-off-by: Peter Wilcsinszky <peter.wilcsinszky@axoflow.com>

Author: pepov

docs/configuration/plugins/outputs/kafka.md

@@ -184,7 +184,9 @@ Password when using PLAIN/SCRAM SASL authentication
 ### principal (string, optional) {#kafka-principal}
 
 
-### rdkafka_options (RdkafkaOptions, optional) {#kafka-rdkafka_options}
+### rdkafka_options (*RdkafkaOptions, optional) {#kafka-rdkafka_options}
+
+RdkafkaOptions represents the global configuration properties for librdkafka. 
 
 
 ### required_acks (int, optional) {#kafka-required_acks}

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

@@ -65,8 +65,9 @@ type _metaKafka interface{} //nolint:deadcode,unused
 // -[more info](https://github.com/fluent/fluent-plugin-kafka#output-plugin)
 type KafkaOutputConfig struct {
 	// Use rdkafka2 instead of the legacy kafka2 output plugin. This plugin requires fluentd image version v1.16-4.9-full or higher.
-	UseRdkafka     bool           `json:"use_rdkafka,omitempty"`
-	RdkafkaOptions RdkafkaOptions `json:"rdkafka_options,omitempty"`
+	UseRdkafka bool `json:"use_rdkafka,omitempty"`
+	// RdkafkaOptions represents the global configuration properties for librdkafka.
+	RdkafkaOptions *RdkafkaOptions `json:"rdkafka_options,omitempty"`
 	// The list of all seed brokers, with their host and port information.
 	Brokers string `json:"brokers"`
 	// Topic Key (default: "topic")
@@ -362,6 +363,13 @@ 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 {
@@ -375,3 +383,9 @@ 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

@@ -17,10 +17,11 @@ package output_test
 import (
 	"testing"
 
-	"github.com/kube-logging/logging-operator/pkg/sdk/logging/model/output"
-	"github.com/kube-logging/logging-operator/pkg/sdk/logging/model/render"
 	"github.com/stretchr/testify/require"
 	"sigs.k8s.io/yaml"
+
+	"github.com/kube-logging/logging-operator/pkg/sdk/logging/model/output"
+	"github.com/kube-logging/logging-operator/pkg/sdk/logging/model/render"
 )
 
 func TestKafka(t *testing.T) {
@@ -99,6 +100,15 @@ 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>