[summerospp]add fluentbit mqtt plugin (#911)

* [summerospp]add fluentbit mqtt plugin Signed-off-by: “sjliu1” <“sj_liu123@163.com”> * [summerospp]add fluentbit mqtt plugin Signed-off-by: “sjliu1” <“sj_liu123@163.com”> * [summerospp]add fluentbit mqtt plugin Signed-off-by: “sjliu1” <“sj_liu123@163.com”> --------- Signed-off-by: “sjliu1” <“sj_liu123@163.com”> Co-authored-by: “sjliu1” <“sj_liu123@163.com”>
fluent · Sep 9, 2023 · 8e3b536 · 8e3b536
1 parent a669907
commit 8e3b536
Show file tree

Hide file tree

Showing 10 changed files with 147 additions and 12 deletions.
diff --git a/apis/fluentbit/v1alpha2/clusterinput_types.go b/apis/fluentbit/v1alpha2/clusterinput_types.go
@@ -55,10 +55,12 @@ type InputSpec struct {
 	CustomPlugin *custom.CustomPlugin `json:"customPlugin,omitempty"`
 	// Forward defines forward  input plugin configuration
 	Forward *input.Forward `json:"forward,omitempty"`
-	// OpenTelemetry defines forward  input plugin configuration
+	// OpenTelemetry defines the OpenTelemetry input plugin configuration
 	OpenTelemetry *input.OpenTelemetry `json:"openTelemetry,omitempty"`
-	// HTTP defines forward  input plugin configuration
+	// HTTP defines the HTTP input plugin configuration
 	HTTP *input.HTTP `json:"http,omitempty"`
+	// MQTT defines the MQTT input plugin configuration
+	MQTT *input.MQTT `json:"mqtt,omitempty"`
 }
 
 // +kubebuilder:object:root=true

diff --git a/apis/fluentbit/v1alpha2/plugins/input/mqtt.go b/apis/fluentbit/v1alpha2/plugins/input/mqtt.go
@@ -0,0 +1,38 @@
+package input
+
+import (
+	"fmt"
+
+	"github.com/fluent/fluent-operator/v2/apis/fluentbit/v1alpha2/plugins"
+	"github.com/fluent/fluent-operator/v2/apis/fluentbit/v1alpha2/plugins/params"
+)
+
+// +kubebuilder:object:generate:=true
+
+// The MQTT input plugin, allows to retrieve messages/data from MQTT control packets over a TCP connection. <br />
+// The incoming data to receive must be a JSON map. <br />
+// **For full documentation, refer to https://docs.fluentbit.io/manual/pipeline/inputs/mqtt**
+type MQTT struct {
+	// Listener network interface, default: 0.0.0.0
+	Listen string `json:"listen,omitempty"`
+	// TCP port where listening for connections, default: 1883
+	// +kubebuilder:validation:Minimum:=1
+	// +kubebuilder:validation:Maximum:=65535
+	Port *int32 `json:"port,omitempty"`
+}
+
+func (_ *MQTT) Name() string {
+	return "mqtt"
+}
+
+// implement Section() method
+func (m *MQTT) Params(_ plugins.SecretLoader) (*params.KVs, error) {
+	kvs := params.NewKVs()
+	if m.Listen != "" {
+		kvs.Insert("Listen", m.Listen)
+	}
+	if m.Port != nil {
+		kvs.Insert("Port", fmt.Sprint(*m.Port))
+	}
+	return kvs, nil
+}
diff --git a/apis/fluentbit/v1alpha2/plugins/input/zz_generated.deepcopy.go b/apis/fluentbit/v1alpha2/plugins/input/zz_generated.deepcopy.go
diff --git a/apis/fluentbit/v1alpha2/zz_generated.deepcopy.go b/apis/fluentbit/v1alpha2/zz_generated.deepcopy.go
diff --git a/charts/fluent-operator/charts/fluent-bit-crds/crds/fluentbit.fluent.io_clusterinputs.yaml b/charts/fluent-operator/charts/fluent-bit-crds/crds/fluentbit.fluent.io_clusterinputs.yaml
@@ -124,7 +124,7 @@ spec:
                     type: string
                 type: object
               http:
-                description: HTTP defines forward  input plugin configuration
+                description: HTTP defines the HTTP input plugin configuration
                 properties:
                   bufferChunkSize:
                     description: This sets the chunk size for incoming incoming JSON
@@ -236,6 +236,20 @@ spec:
                 - debug
                 - trace
                 type: string
+              mqtt:
+                description: MQTT defines the MQTT input plugin configuration
+                properties:
+                  listen:
+                    description: 'Listener network interface, default: 0.0.0.0'
+                    type: string
+                  port:
+                    description: 'TCP port where listening for connections, default:
+                      1883'
+                    format: int32
+                    maximum: 65535
+                    minimum: 1
+                    type: integer
+                type: object
               nodeExporterMetrics:
                 description: NodeExporterMetrics defines Node Exporter Metrics Input
                   configuration.
@@ -261,7 +275,8 @@ spec:
                     type: string
                 type: object
               openTelemetry:
-                description: OpenTelemetry defines forward  input plugin configuration
+                description: OpenTelemetry defines the OpenTelemetry input plugin
+                  configuration
                 properties:
                   bufferChunkSize:
                     description: This sets the chunk size for incoming incoming JSON

diff --git a/config/crd/bases/fluentbit.fluent.io_clusterinputs.yaml b/config/crd/bases/fluentbit.fluent.io_clusterinputs.yaml
@@ -124,7 +124,7 @@ spec:
                     type: string
                 type: object
               http:
-                description: HTTP defines forward  input plugin configuration
+                description: HTTP defines the HTTP input plugin configuration
                 properties:
                   bufferChunkSize:
                     description: This sets the chunk size for incoming incoming JSON
@@ -236,6 +236,20 @@ spec:
                 - debug
                 - trace
                 type: string
+              mqtt:
+                description: MQTT defines the MQTT input plugin configuration
+                properties:
+                  listen:
+                    description: 'Listener network interface, default: 0.0.0.0'
+                    type: string
+                  port:
+                    description: 'TCP port where listening for connections, default:
+                      1883'
+                    format: int32
+                    maximum: 65535
+                    minimum: 1
+                    type: integer
+                type: object
               nodeExporterMetrics:
                 description: NodeExporterMetrics defines Node Exporter Metrics Input
                   configuration.
@@ -261,7 +275,8 @@ spec:
                     type: string
                 type: object
               openTelemetry:
-                description: OpenTelemetry defines forward  input plugin configuration
+                description: OpenTelemetry defines the OpenTelemetry input plugin
+                  configuration
                 properties:
                   bufferChunkSize:
                     description: This sets the chunk size for incoming incoming JSON

diff --git a/docs/fluentbit.md b/docs/fluentbit.md
@@ -419,8 +419,9 @@ InputSpec defines the desired state of ClusterInput
 | fluentBitMetrics | FluentBitMetrics defines Fluent Bit Metrics Input configuration. | *[input.FluentbitMetrics](plugins/input/fluentbitmetrics.md) |
 | customPlugin | CustomPlugin defines Custom Input configuration. | *custom.CustomPlugin |
 | forward | Forward defines forward  input plugin configuration | *[input.Forward](plugins/input/forward.md) |
-| openTelemetry | OpenTelemetry defines forward  input plugin configuration | *[input.OpenTelemetry](plugins/input/opentelemetry.md) |
-| http | HTTP defines forward  input plugin configuration | *[input.OpenTelemetry](plugins/input/opentelemetry.md) |
+| openTelemetry | OpenTelemetry defines the OpenTelemetry input plugin configuration | *[input.OpenTelemetry](plugins/input/opentelemetry.md) |
+| http | HTTP defines the HTTP input plugin configuration | *[input.HTTP](plugins/input/http.md) |
+| mqtt | MQTT defines the MQTT input plugin configuration | *[input.MQTT](plugins/input/mqtt.md) |
 
 [Back to TOC](#table-of-contents)
 # NamespacedFluentBitCfgSpec

diff --git a/docs/plugins/fluentbit/input/mqtt.md b/docs/plugins/fluentbit/input/mqtt.md
@@ -0,0 +1,9 @@
+# MQTT
+
+The MQTT input plugin, allows to retrieve messages/data from MQTT control packets over a TCP connection. <br /> The incoming data to receive must be a JSON map. <br /> **For full documentation, refer to https://docs.fluentbit.io/manual/pipeline/inputs/mqtt**
+
+
+| Field | Description | Scheme |
+| ----- | ----------- | ------ |
+| listen | Listener network interface, default: 0.0.0.0 | string |
+| port | TCP port where listening for connections, default: 1883 | *int32 |
diff --git a/manifests/setup/fluent-operator-crd.yaml b/manifests/setup/fluent-operator-crd.yaml
@@ -1873,7 +1873,7 @@ spec:
                     type: string
                 type: object
               http:
-                description: HTTP defines forward  input plugin configuration
+                description: HTTP defines the HTTP input plugin configuration
                 properties:
                   bufferChunkSize:
                     description: This sets the chunk size for incoming incoming JSON
@@ -1985,6 +1985,20 @@ spec:
                 - debug
                 - trace
                 type: string
+              mqtt:
+                description: MQTT defines the MQTT input plugin configuration
+                properties:
+                  listen:
+                    description: 'Listener network interface, default: 0.0.0.0'
+                    type: string
+                  port:
+                    description: 'TCP port where listening for connections, default:
+                      1883'
+                    format: int32
+                    maximum: 65535
+                    minimum: 1
+                    type: integer
+                type: object
               nodeExporterMetrics:
                 description: NodeExporterMetrics defines Node Exporter Metrics Input
                   configuration.
@@ -2010,7 +2024,8 @@ spec:
                     type: string
                 type: object
               openTelemetry:
-                description: OpenTelemetry defines forward  input plugin configuration
+                description: OpenTelemetry defines the OpenTelemetry input plugin
+                  configuration
                 properties:
                   bufferChunkSize:
                     description: This sets the chunk size for incoming incoming JSON

diff --git a/manifests/setup/setup.yaml b/manifests/setup/setup.yaml
@@ -1873,7 +1873,7 @@ spec:
                     type: string
                 type: object
               http:
-                description: HTTP defines forward  input plugin configuration
+                description: HTTP defines the HTTP input plugin configuration
                 properties:
                   bufferChunkSize:
                     description: This sets the chunk size for incoming incoming JSON
@@ -1985,6 +1985,20 @@ spec:
                 - debug
                 - trace
                 type: string
+              mqtt:
+                description: MQTT defines the MQTT input plugin configuration
+                properties:
+                  listen:
+                    description: 'Listener network interface, default: 0.0.0.0'
+                    type: string
+                  port:
+                    description: 'TCP port where listening for connections, default:
+                      1883'
+                    format: int32
+                    maximum: 65535
+                    minimum: 1
+                    type: integer
+                type: object
               nodeExporterMetrics:
                 description: NodeExporterMetrics defines Node Exporter Metrics Input
                   configuration.
@@ -2010,7 +2024,8 @@ spec:
                     type: string
                 type: object
               openTelemetry:
-                description: OpenTelemetry defines forward  input plugin configuration
+                description: OpenTelemetry defines the OpenTelemetry input plugin
+                  configuration
                 properties:
                   bufferChunkSize:
                     description: This sets the chunk size for incoming incoming JSON