feat(plugin): Support customplugin (fluent#301)

Signed-off-by: Gentleelephant <[email protected]>

README.md

@@ -52,6 +52,7 @@ Once installed, the Fluent Operator provides the following features:
 		- [Monitoring](#monitoring)
 		- [Custom Parser](#custom-parser)
 		- [Misc](#misc)
+        - [Custom Plugin](#custom-plugin)
 	- [Roadmap](#roadmap)
 	- [Development](#development)
 		- [Requirements](#requirements)
@@ -264,6 +265,9 @@ For more info on various use cases of Fluent Operator Fluentd CRDs, you can refe
 
 If you want to learn more about Fluent-Operator, please refer to the [misc](docs/best-practice/misc.md).
 
+### Custom Plugin
+[Here](docs/best-practice/custom-plugin.md) you can learn how to use custom plugin in Fluent Bit.
+
 ## Roadmap
 
 - [x] Support containerd log format

apis/fluentbit/v1alpha2/clusterfilter_types.go

@@ -25,6 +25,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/fluent/fluent-operator/apis/fluentbit/v1alpha2/plugins"
+	"github.com/fluent/fluent-operator/apis/fluentbit/v1alpha2/plugins/custom"
 	"github.com/fluent/fluent-operator/apis/fluentbit/v1alpha2/plugins/filter"
 )
 
@@ -66,6 +67,8 @@ type FilterItem struct {
 	AWS *filter.AWS `json:"aws,omitempty"`
 	//Multiline defines a Multiline configuration.
 	Multiline *filter.Multiline `json:"multiline,omitempty"`
+	//CustomPlugin defines a Custom plugin configuration.
+	CustomPlugin *custom.CustomPlugin `json:"customPlugin,omitempty"`
 }
 
 // +kubebuilder:object:root=true
@@ -111,7 +114,9 @@ func (list ClusterFilterList) Load(sl plugins.SecretLoader) (string, error) {
 			}
 
 			buf.WriteString("[Filter]\n")
-			buf.WriteString(fmt.Sprintf("    Name    %s\n", p.Name()))
+			if p.Name() != "" {
+				buf.WriteString(fmt.Sprintf("    Name    %s\n", p.Name()))
+			}
 			if item.Spec.Match != "" {
 				buf.WriteString(fmt.Sprintf("    Match    %s\n", item.Spec.Match))
 			}

apis/fluentbit/v1alpha2/clusterfluentbitconfig_types_test.go

@@ -3,12 +3,14 @@ package v1alpha2
 import (
 	"testing"
 
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
 	"github.com/fluent/fluent-operator/apis/fluentbit/v1alpha2/plugins"
+	"github.com/fluent/fluent-operator/apis/fluentbit/v1alpha2/plugins/custom"
 	"github.com/fluent/fluent-operator/apis/fluentbit/v1alpha2/plugins/filter"
 	"github.com/fluent/fluent-operator/apis/fluentbit/v1alpha2/plugins/input"
 	"github.com/fluent/fluent-operator/apis/fluentbit/v1alpha2/plugins/output"
 	. "github.com/onsi/gomega"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
 var expected = `[Service]
@@ -65,6 +67,16 @@ var expected = `[Service]
     json_date_format    iso8601
     tls    On
     tls.verify    true
+[Output]
+    Name    kafka
+    Topics    fluentbit
+    Match    *
+    Brokers    192.168.100.32:9092
+    rdkafka.debug    All
+    rdkafka.request.required.acks    1
+    rdkafka.log.connection.close    false
+    rdkafka.log_level    7
+    rdkafka.metadata.broker.list    192.168.100.32:9092
 [Output]
     Name    opensearch
     Match    *
@@ -271,8 +283,24 @@ func Test_FluentBitConfig_RenderMainConfig(t *testing.T) {
 		},
 	}
 
+	kafkaOutput := ClusterOutput{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: "fluentbit.fluent.io/v1alpha2",
+			Kind:       "ClusterOutput",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:   "kafka_output",
+			Labels: labels,
+		},
+		Spec: OutputSpec{
+			CustomPlugin: &custom.CustomPlugin{
+				Config: "    Name    kafka\n    Topics    fluentbit\n    Match    *\n    Brokers    192.168.100.32:9092\n    rdkafka.debug    All\n    rdkafka.request.required.acks    1\n    rdkafka.log.connection.close    false\n    rdkafka.log_level    7\n    rdkafka.metadata.broker.list    192.168.100.32:9092",
+			},
+		},
+	}
+
 	outputs := ClusterOutputList{
-		Items: []ClusterOutput{syslogOut, httpOutput, openSearchOutput},
+		Items: []ClusterOutput{syslogOut, httpOutput, openSearchOutput, kafkaOutput},
 	}
 
 	cfg := ClusterFluentBitConfig{

apis/fluentbit/v1alpha2/clusterinput_types.go

@@ -22,9 +22,11 @@ import (
 	"reflect"
 	"sort"
 
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
 	"github.com/fluent/fluent-operator/apis/fluentbit/v1alpha2/plugins"
+	"github.com/fluent/fluent-operator/apis/fluentbit/v1alpha2/plugins/custom"
 	"github.com/fluent/fluent-operator/apis/fluentbit/v1alpha2/plugins/input"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
 // EDIT THIS FILE!  THIS IS SCAFFOLDING FOR YOU TO OWN!
@@ -47,6 +49,8 @@ type InputSpec struct {
 	PrometheusScrapeMetrics *input.PrometheusScrapeMetrics `json:"prometheusScrapeMetrics,omitempty"`
 	// FluentBitMetrics defines Fluent Bit Metrics Input configuration.
 	FluentBitMetrics *input.FluentbitMetrics `json:"fluentBitMetrics,omitempty"`
+	// CustomPlugin defines Custom Input configuration.
+	CustomPlugin *custom.CustomPlugin `json:"customPlugin,omitempty"`
 }
 
 // +kubebuilder:object:root=true
@@ -91,7 +95,9 @@ func (list ClusterInputList) Load(sl plugins.SecretLoader) (string, error) {
 			}
 
 			buf.WriteString("[Input]\n")
-			buf.WriteString(fmt.Sprintf("    Name    %s\n", p.Name()))
+			if p.Name() != "" {
+				buf.WriteString(fmt.Sprintf("    Name    %s\n", p.Name()))
+			}
 			if item.Spec.Alias != "" {
 				buf.WriteString(fmt.Sprintf("    Alias    %s\n", item.Spec.Alias))
 			}

apis/fluentbit/v1alpha2/clusteroutput_types.go

@@ -25,6 +25,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/fluent/fluent-operator/apis/fluentbit/v1alpha2/plugins"
+	"github.com/fluent/fluent-operator/apis/fluentbit/v1alpha2/plugins/custom"
 	"github.com/fluent/fluent-operator/apis/fluentbit/v1alpha2/plugins/output"
 )
 
@@ -75,6 +76,8 @@ type OutputSpec struct {
 	OpenTelemetry *output.OpenTelemetry `json:"opentelemetry,omitempty"`
 	// PrometheusRemoteWrite_types defines Prometheus Remote Write configuration.
 	PrometheusRemoteWrite *output.PrometheusRemoteWrite `json:"prometheusRemoteWrite,omitempty"`
+	// CustomPlugin defines Custom Output configuration.
+	CustomPlugin *custom.CustomPlugin `json:"customPlugin,omitempty"`
 }
 
 // +kubebuilder:object:root=true
@@ -119,7 +122,9 @@ func (list ClusterOutputList) Load(sl plugins.SecretLoader) (string, error) {
 			}
 
 			buf.WriteString("[Output]\n")
-			buf.WriteString(fmt.Sprintf("    Name    %s\n", p.Name()))
+			if p.Name() != "" {
+				buf.WriteString(fmt.Sprintf("    Name    %s\n", p.Name()))
+			}
 			if item.Spec.Match != "" {
 				buf.WriteString(fmt.Sprintf("    Match    %s\n", item.Spec.Match))
 			}

apis/fluentbit/v1alpha2/plugins/custom/custom_types.go

@@ -0,0 +1,37 @@
+package custom
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/fluent/fluent-operator/apis/fluentbit/v1alpha2/plugins"
+	"github.com/fluent/fluent-operator/apis/fluentbit/v1alpha2/plugins/params"
+)
+
+// +kubebuilder:object:generate:=true
+
+// CustomPlugin is used to support filter plugins that are not implemented yet
+type CustomPlugin struct {
+	Config string `json:"config,omitempty"`
+}
+
+func (c *CustomPlugin) Name() string {
+	return ""
+}
+
+func (a *CustomPlugin) Params(_ plugins.SecretLoader) (*params.KVs, error) {
+	kvs := params.NewKVs()
+	splits := strings.Split(a.Config, "\n")
+	for _, i := range splits {
+		if len(i) == 0 {
+			continue
+		}
+		fields := strings.Fields(i)
+		if len(fields) < 2 {
+			return nil, errors.New(fmt.Sprintf("invalid plugin config: %s", i))
+		}
+		kvs.Insert(fields[0], strings.Join(fields[1:], " "))
+	}
+	return kvs, nil
+}

charts/fluent-operator/crds/fluentbit.fluent.io_clusterfilters.yaml

@@ -84,6 +84,12 @@ spec:
                             is false.
                           type: boolean
                       type: object
+                    customPlugin:
+                      description: CustomPlugin defines a Custom plugin configuration.
+                      properties:
+                        config:
+                          type: string
+                      type: object
                     grep:
                       description: Grep defines Grep Filter configuration.
                       properties:

charts/fluent-operator/crds/fluentbit.fluent.io_clusterinputs.yaml

@@ -42,6 +42,12 @@ spec:
                 description: A user friendly alias name for this input plugin. Used
                   in metrics for distinction of each configured input.
                 type: string
+              customPlugin:
+                description: CustomPlugin defines Custom Input configuration.
+                properties:
+                  config:
+                    type: string
+                type: object
               dummy:
                 description: Dummy defines Dummy Input configuration.
                 properties:

charts/fluent-operator/crds/fluentbit.fluent.io_clusteroutputs.yaml

@@ -42,6 +42,12 @@ spec:
                 description: A user friendly alias name for this output plugin. Used
                   in metrics for distinction of each configured output.
                 type: string
+              customPlugin:
+                description: CustomPlugin defines Custom Output configuration.
+                properties:
+                  config:
+                    type: string
+                type: object
               datadog:
                 description: DataDog defines DataDog Output configuration.
                 properties:

config/crd/bases/fluentbit.fluent.io_clusterfilters.yaml

@@ -84,6 +84,12 @@ spec:
                             is false.
                           type: boolean
                       type: object
+                    customPlugin:
+                      description: CustomPlugin defines a Custom plugin configuration.
+                      properties:
+                        config:
+                          type: string
+                      type: object
                     grep:
                       description: Grep defines Grep Filter configuration.
                       properties:

config/crd/bases/fluentbit.fluent.io_clusterinputs.yaml

@@ -42,6 +42,12 @@ spec:
                 description: A user friendly alias name for this input plugin. Used
                   in metrics for distinction of each configured input.
                 type: string
+              customPlugin:
+                description: CustomPlugin defines Custom Input configuration.
+                properties:
+                  config:
+                    type: string
+                type: object
               dummy:
                 description: Dummy defines Dummy Input configuration.
                 properties:

config/crd/bases/fluentbit.fluent.io_clusteroutputs.yaml

@@ -42,6 +42,12 @@ spec:
                 description: A user friendly alias name for this output plugin. Used
                   in metrics for distinction of each configured output.
                 type: string
+              customPlugin:
+                description: CustomPlugin defines Custom Output configuration.
+                properties:
+                  config:
+                    type: string
+                type: object
               datadog:
                 description: DataDog defines DataDog Output configuration.
                 properties: