feat: fluentd output plugin kafka support rdkafka2 client

charts/logging-operator/crds/logging.banzaicloud.io_clusteroutputs.yaml

@@ -3424,6 +3424,8 @@ spec:
                     type: string
                   use_default_for_unknown_topic:
                     type: boolean
+                  use_rdkafka:
+                    type: boolean
                   username:
                     properties:
                       mountFrom:
@@ -10821,6 +10823,8 @@ spec:
                     type: string
                   use_default_for_unknown_topic:
                     type: boolean
+                  use_rdkafka:
+                    type: boolean
                   username:
                     properties:
                       mountFrom:

charts/logging-operator/crds/logging.banzaicloud.io_outputs.yaml

@@ -3420,6 +3420,8 @@ spec:
                     type: string
                   use_default_for_unknown_topic:
                     type: boolean
+                  use_rdkafka:
+                    type: boolean
                   username:
                     properties:
                       mountFrom:
@@ -10091,6 +10093,8 @@ spec:
                     type: string
                   use_default_for_unknown_topic:
                     type: boolean
+                  use_rdkafka:
+                    type: boolean
                   username:
                     properties:
                       mountFrom:

config/crd/bases/logging.banzaicloud.io_clusteroutputs.yaml

@@ -3424,6 +3424,8 @@ spec:
                     type: string
                   use_default_for_unknown_topic:
                     type: boolean
+                  use_rdkafka:
+                    type: boolean
                   username:
                     properties:
                       mountFrom:
@@ -10821,6 +10823,8 @@ spec:
                     type: string
                   use_default_for_unknown_topic:
                     type: boolean
+                  use_rdkafka:
+                    type: boolean
                   username:
                     properties:
                       mountFrom:

config/crd/bases/logging.banzaicloud.io_outputs.yaml

@@ -3420,6 +3420,8 @@ spec:
                     type: string
                   use_default_for_unknown_topic:
                     type: boolean
+                  use_rdkafka:
+                    type: boolean
                   username:
                     properties:
                       mountFrom:
@@ -10091,6 +10093,8 @@ spec:
                     type: string
                   use_default_for_unknown_topic:
                     type: boolean
+                  use_rdkafka:
+                    type: boolean
                   username:
                     properties:
                       mountFrom:

docs/configuration/plugins/outputs/gelf.md

@@ -13,7 +13,7 @@ generated_file: true
 
 ### buffer (*Buffer, optional) {#output config-buffer}
 
-Available since ghcr.io/kube-logging/fluentd:v1.16-full-build.139 [Buffer](../buffer/) 
+Available since ghcr.io/kube-logging/fluentd:v1.16-4.8-full [Buffer](../buffer/) 
 
 
 ### host (string, required) {#output config-host}

docs/configuration/plugins/outputs/kafka.md

@@ -33,7 +33,10 @@ spec:
 ## Configuration
 ## Kafka
 
-Send your logs to Kafka
+Send your logs to Kafka.
+Setting use_rdkafka to true opts for rdkafka2, which offers higher performance compared to ruby-kafka.
+(Note: requires fluentd image version v1.16-4.9-full or higher)
+-[more info](https://github.com/fluent/fluent-plugin-kafka#output-plugin)
 
 ### ack_timeout (int, optional) {#kafka-ack_timeout}
 
@@ -212,7 +215,7 @@ Client certificate key
 Verify certificate hostname 
 
 
-### sasl_over_ssl (bool, required) {#kafka-sasl_over_ssl}
+### sasl_over_ssl (*bool, optional) {#kafka-sasl_over_ssl}
 
 SASL over SSL
 
@@ -240,6 +243,11 @@ Use default for unknown topics
 
 Default: false
 
+### use_rdkafka (bool, optional) {#kafka-use_rdkafka}
+
+Use rdkafka2 instead of the legacy kafka2 output plugin. This plugin requires fluentd image version v1.16-4.9-full or higher. 
+
+
 ### username (*secret.Secret, optional) {#kafka-username}
 
 Username when using PLAIN/SCRAM SASL authentication 

e2e/common/helpers.go

@@ -125,7 +125,7 @@ func LoggingInfra(
 			ControlNamespace: nsInfra,
 			FluentdSpec: &v1beta1.FluentdSpec{
 				Image: v1beta1.ImageSpec{
-					Tag: "v1.16-4.8-base",
+					Tag: "v1.16-4.9-base",
 				},
 				DisablePvc: true,
 				Resources: v12.ResourceRequirements{

e2e/fluentd-aggregator-namespacelabel/fluentd_aggregator_test.go

@@ -103,7 +103,7 @@ func TestFluentdAggregator_NamespaceLabel(t *testing.T) {
 				},
 				FluentdSpec: &v1beta1.FluentdSpec{
 					Image: v1beta1.ImageSpec{
-						Tag: "v1.16-4.8-base",
+						Tag: "v1.16-4.9-base",
 					},
 					Resources: corev1.ResourceRequirements{
 						Limits: corev1.ResourceList{

e2e/fluentd-aggregator/fluentd_aggregator_test.go

@@ -91,7 +91,7 @@ func TestFluentdAggregator_MultiWorker(t *testing.T) {
 				},
 				FluentdSpec: &v1beta1.FluentdSpec{
 					Image: v1beta1.ImageSpec{
-						Tag: "v1.16-4.8-base",
+						Tag: "v1.16-4.9-base",
 					},
 					Resources: corev1.ResourceRequirements{
 						Limits: corev1.ResourceList{
@@ -247,7 +247,7 @@ func TestFluentdAggregator_ConfigChecks(t *testing.T) {
 				},
 				FluentdSpec: &v1beta1.FluentdSpec{
 					Image: v1beta1.ImageSpec{
-						Tag: "v1.16-4.8-base",
+						Tag: "v1.16-4.9-base",
 					},
 					Resources: corev1.ResourceRequirements{
 						Limits: corev1.ResourceList{

e2e/volumedrain/volumedrain_test.go

@@ -89,7 +89,7 @@ func TestVolumeDrain_Downscale(t *testing.T) {
 				},
 				FluentdSpec: &v1beta1.FluentdSpec{
 					Image: v1beta1.ImageSpec{
-						Tag: "v1.16-4.8-base",
+						Tag: "v1.16-4.9-base",
 					},
 					Resources: corev1.ResourceRequirements{
 						Limits: corev1.ResourceList{

pkg/sdk/logging/api/v1beta1/logging_types.go

@@ -172,7 +172,7 @@ const (
 	DefaultFluentbitConfigReloaderImageRepository = "ghcr.io/kube-logging/config-reloader"
 	DefaultFluentbitConfigReloaderImageTag        = "v0.0.5"
 	DefaultFluentdImageRepository                 = "ghcr.io/kube-logging/fluentd"
-	DefaultFluentdImageTag                        = "v1.16-4.8-full"
+	DefaultFluentdImageTag                        = "v1.16-4.9-full"
 	DefaultFluentdBufferStorageVolumeName         = "fluentd-buffer"
 	DefaultFluentdDrainWatchImageRepository       = "ghcr.io/kube-logging/fluentd-drain-watch"
 	DefaultFluentdDrainWatchImageTag              = "v0.2.1"

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

@@ -49,7 +49,7 @@ type GELFOutputConfig struct {
 	TLS *bool `json:"tls,omitempty"`
 	// TLS options (default: {}). For details, see [https://github.com/graylog-labs/gelf-rb/blob/72916932b789f7a6768c3cdd6ab69a3c942dbcef/lib/gelf/transport/tcp_tls.rb#L7-L12](https://github.com/graylog-labs/gelf-rb/blob/72916932b789f7a6768c3cdd6ab69a3c942dbcef/lib/gelf/transport/tcp_tls.rb#L7-L12).
 	TLSOptions map[string]string `json:"tls_options,omitempty"`
-	// Available since ghcr.io/kube-logging/fluentd:v1.16-full-build.139
+	// Available since ghcr.io/kube-logging/fluentd:v1.16-4.8-full
 	// +docLink:"Buffer,../buffer/"
 	Buffer *Buffer `json:"buffer,omitempty"`
 }

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

@@ -16,6 +16,7 @@ package output
 
 import (
 	"github.com/cisco-open/operator-tools/pkg/secret"
+
 	"github.com/kube-logging/logging-operator/pkg/sdk/logging/model/types"
 )
 
@@ -58,9 +59,13 @@ type _metaKafka interface{} //nolint:deadcode,unused
 
 // +kubebuilder:object:generate=true
 // +docName:"Kafka"
-// Send your logs to Kafka
+// Send your logs to Kafka.
+// Setting use_rdkafka to true opts for rdkafka2, which offers higher performance compared to ruby-kafka.
+// (Note: requires fluentd image version v1.16-4.9-full or higher)
+// -[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"`
 	// The list of all seed brokers, with their host and port information.
 	Brokers string `json:"brokers"`
 	// Topic Key (default: "topic")
@@ -95,7 +100,7 @@ type KafkaOutputConfig struct {
 	Idempotent bool `json:"idempotent,omitempty"`
 	// SASL over SSL (default: true)
 	// +kubebuilder:validation:Optional
-	SaslOverSSL bool           `json:"sasl_over_ssl"`
+	SaslOverSSL *bool          `json:"sasl_over_ssl,omitempty"`
 	Principal   string         `json:"principal,omitempty"`
 	Keytab      *secret.Secret `json:"keytab,omitempty"`
 	// Username when using PLAIN/SCRAM SASL authentication
@@ -141,7 +146,10 @@ type KafkaOutputConfig struct {
 }
 
 func (e *KafkaOutputConfig) ToDirective(secretLoader secret.SecretLoader, id string) (types.Directive, error) {
-	const pluginType = "kafka2"
+	pluginType := "kafka2"
+	if e.UseRdkafka {
+		pluginType = "rdkafka2"
+	}
 	kafka := &types.OutputPlugin{
 		PluginMeta: types.PluginMeta{
 			Type:      pluginType,
@@ -171,5 +179,8 @@ func (e *KafkaOutputConfig) ToDirective(secretLoader secret.SecretLoader, id str
 			kafka.SubDirectives = append(kafka.SubDirectives, format)
 		}
 	}
+
+	// remove use_rdkafka from params, it is not a valid parameter for plugin config
+	delete(kafka.Params, "use_rdkafka")
 	return kafka, nil
 }

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

@@ -62,3 +62,42 @@ buffer:
 	test := render.NewOutputPluginTest(t, kafka)
 	test.DiffResult(expected)
 }
+
+func TestRdkafka(t *testing.T) {
+	CONFIG := []byte(`
+brokers: kafka-headless.kafka.svc.cluster.local:29092
+default_topic: topic
+use_rdkafka: true
+ssl_verify_hostname: false
+format:
+  type: json
+buffer:
+  timekey: 1m
+  timekey_wait: 30s
+  timekey_use_utc: true
+`)
+	expected := `
+  <match **>
+    @type rdkafka2
+    @id test
+    brokers kafka-headless.kafka.svc.cluster.local:29092
+    default_topic topic
+    ssl_verify_hostname false
+    <buffer tag,time>
+      @type file
+      path /buffers/test.*.buffer
+      retry_forever true
+      timekey 1m
+      timekey_use_utc true
+      timekey_wait 30s
+    </buffer>
+    <format>
+      @type json
+    </format>
+  </match>
+`
+	kafka := &output.KafkaOutputConfig{}
+	require.NoError(t, yaml.Unmarshal(CONFIG, kafka))
+	test := render.NewOutputPluginTest(t, kafka)
+	test.DiffResult(expected)
+}