feat: adds fluentd monitor_agent input plugin

apis/fluentd/v1alpha1/plugins/input/monitor_agent.go

@@ -0,0 +1,17 @@
+package input
+
+// The in_monitor_agent Input plugin exports Fluentd's internal metrics via REST API.
+type MonitorAgent struct {
+	// The port to listen to.
+	Port *int64 `json:"port,omitempty"`
+	// The bind address to listen to.
+	Bind *string `json:"bind,omitempty"`
+	// If you set this parameter, this plugin emits metrics as records.
+	Tag *string `json:"tag,omitempty"`
+	// The interval time between event emits. This will be used when "tag" is configured.
+	EmitInterval *int64 `json:"emitInterval,omitempty"`
+	// You can set this option to false to remove the config field from the response.
+	IncludeConfig *bool `json:"includeConfig,omitempty"`
+	// You can set this option to false to remove the retry field from the response.
+	IncludeRetry *bool `json:"includeRetry,omitempty"`
+}

apis/fluentd/v1alpha1/plugins/input/types.go

@@ -34,6 +34,8 @@ type Input struct {
 	Sample *Sample `json:"sample,omitempty"`
 	// Custom plugin type
 	CustomPlugin *custom.CustomPlugin `json:"customPlugin,omitempty"`
+	// monitor_agent plugin
+	MonitorAgent *MonitorAgent `json:"monitorAgent,omitempty"`
 }
 
 // DeepCopyInto implements the DeepCopyInto interface.
@@ -94,6 +96,11 @@ func (i *Input) Params(loader plugins.SecretLoader) (*params.PluginStore, error)
 		return ps, nil
 	}
 
+	if i.MonitorAgent != nil {
+		ps.InsertType(string(params.MonitorAgentType))
+		return i.monitorAgentPlugin(ps, loader), nil
+	}
+
 	return nil, errors.New("you must define an input plugin")
 }
 
@@ -365,4 +372,27 @@ func (i *Input) samplePlugin(parent *params.PluginStore, loader plugins.SecretLo
 	return parent
 }
 
+func (i *Input) monitorAgentPlugin(parent *params.PluginStore, loader plugins.SecretLoader) *params.PluginStore {
+	monitorAgentModel := i.MonitorAgent
+	if monitorAgentModel.Port != nil {
+		parent.InsertPairs("port", fmt.Sprint(*monitorAgentModel.Port))
+	}
+	if monitorAgentModel.Bind != nil {
+		parent.InsertPairs("bind", fmt.Sprint(*monitorAgentModel.Bind))
+	}
+	if monitorAgentModel.Tag != nil {
+		parent.InsertPairs("tag", fmt.Sprint(*monitorAgentModel.Tag))
+	}
+	if monitorAgentModel.EmitInterval != nil {
+		parent.InsertPairs("emit_interval", fmt.Sprint(*monitorAgentModel.EmitInterval))
+	}
+	if monitorAgentModel.IncludeConfig != nil {
+		parent.InsertPairs("include_config", fmt.Sprint(*monitorAgentModel.IncludeConfig))
+	}
+	if monitorAgentModel.IncludeRetry != nil {
+		parent.InsertPairs("include_retry", fmt.Sprint(*monitorAgentModel.IncludeRetry))
+	}
+	return parent
+}
+
 var _ plugins.Plugin = &Input{}

apis/fluentd/v1alpha1/plugins/params/const.go

@@ -51,6 +51,7 @@ const (
 	ForwardInputType InputType = "forward"
 	TailInputType    InputType = "tail"
 	SampleInputType  InputType = "sample"
+	MonitorAgentType InputType = "monitor_agent"
 
 	// Enums the supported filter types
 	RecordTransformerFilterType FilterType = "record_transformer"

apis/fluentd/v1alpha1/tests/expected/fluentd-global-cfg-input-monitorAgent.cfg

@@ -0,0 +1,25 @@
+<source>
+  @type  monitor_agent
+  bind  0.0.0.0
+  emit_interval  5
+  include_config  true
+  include_retry  true
+  port  24220
+  tag  example
+</source>
+<match **>
+  @id  main
+  @type  label_router
+  <route>
+    @label  @2d9e59757d3bfc66d93c3bc44b408922
+    <match>
+      namespaces  fluent
+    </match>
+  </route>
+</match>
+<label @2d9e59757d3bfc66d93c3bc44b408922>
+  <match foo.*>
+    @id  FluentdConfig-fluent-fluentd-config::cluster::clusteroutput::fluentd-output-stdout-0
+    @type  stdout
+  </match>
+</label>

apis/fluentd/v1alpha1/tests/helper_test.go

@@ -1,6 +1,7 @@
 package cfgrender
 
 import (
+	"fmt"
 	"strings"
 	"testing"
 
@@ -88,6 +89,27 @@ func Test_ClusterCfgInputSample(t *testing.T) {
 	}
 }
 
+func Test_ClusterCfgInputMonitorAgent(t *testing.T) {
+	g := NewGomegaWithT(t)
+	sl := plugins.NewSecretLoader(nil, Fluentd.Namespace, logr.Logger{})
+
+	psr := fluentdv1alpha1.NewGlobalPluginResources("main")
+	psr.CombineGlobalInputsPlugins(sl, FluentdInputMonitorAgent.Spec.GlobalInputs)
+
+	clustercfgRouter, err := psr.BuildCfgRouter(&FluentdConfig1)
+	g.Expect(err).NotTo(HaveOccurred())
+	clusterOutputs := []fluentdv1alpha1.ClusterOutput{FluentdClusterOutputTag}
+	clustercfgResources, _ := psr.PatchAndFilterClusterLevelResources(sl, FluentdConfig1.GetCfgId(), []fluentdv1alpha1.ClusterInput{}, []fluentdv1alpha1.ClusterFilter{}, clusterOutputs)
+	err = psr.WithCfgResources(*clustercfgRouter.Label, clustercfgResources)
+	g.Expect(err).NotTo(HaveOccurred())
+
+	for i := 0; i < maxRuntimes; i++ {
+		config, errs := psr.RenderMainConfig(false)
+		g.Expect(errs).NotTo(HaveOccurred())
+		g.Expect(string(getExpectedCfg("./expected/fluentd-global-cfg-input-monitorAgent.cfg"))).To(Equal(config))
+	}
+}
+
 func Test_ClusterCfgOutput2ES(t *testing.T) {
 	g := NewGomegaWithT(t)
 	sl := plugins.NewSecretLoader(nil, Fluentd.Namespace, logr.Logger{})
@@ -107,7 +129,7 @@ func Test_ClusterCfgOutput2ES(t *testing.T) {
 	i := 0
 	for i < maxRuntimes {
 		config, errs := psr.RenderMainConfig(false)
-		// fmt.Println(config)
+		fmt.Println(config)
 		g.Expect(errs).NotTo(HaveOccurred())
 		g.Expect(string(getExpectedCfg("./expected/fluentd-cluster-cfg-output-es.cfg"))).To(Equal(config))
 

apis/fluentd/v1alpha1/tests/tools.go

@@ -58,6 +58,32 @@ spec:
     matchLabels:
       config.fluentd.fluent.io/enabled: "true"
 `
+
+	FluentdInputMonitorAgent    fluentdv1alpha1.Fluentd
+	FluentdInputMonitorAgentRaw = `
+apiVersion: fluentd.fluent.io/v1alpha1
+kind: Fluentd
+metadata:
+name: fluentd
+namespace: fluent
+labels:
+  app.kubernetes.io/name: fluentd
+spec:
+  globalInputs:
+  - monitorAgent:
+      bind: 0.0.0.0
+      port: 24220
+      tag: example
+      emitInterval: 5
+      includeConfig: true
+      includeRetry: true
+  replicas: 1
+  image: kubesphere/fluentd:v1.15.3
+  fluentdCfgSelector:
+    matchLabels:
+      config.fluentd.fluent.io/enabled: "true"
+`
+
 	FluentdInputTail    fluentdv1alpha1.Fluentd
 	FluentdInputTailRaw = `
 apiVersion: fluentd.fluent.io/v1alpha1
@@ -575,6 +601,7 @@ func init() {
 			ParseIntoObject(FluentdRaw, &Fluentd)
 			ParseIntoObject(FluentdInputTailRaw, &FluentdInputTail)
 			ParseIntoObject(FluentdInputSampleRaw, &FluentdInputSample)
+			ParseIntoObject(FluentdInputMonitorAgentRaw, &FluentdInputMonitorAgent)
 			ParseIntoObject(FluentdClusterOutputTagRaw, &FluentdClusterOutputTag)
 			ParseIntoObject(FluentdClusterFluentdConfig1Raw, &FluentdClusterFluentdConfig1)
 			ParseIntoObject(FluentdClusterFluentdConfig2Raw, &FluentdClusterFluentdConfig2)

charts/fluent-operator/charts/fluentd-crds/crds/fluentd.fluent.io_fluentds.yaml

@@ -2212,6 +2212,34 @@ spec:
                       description: The @log_level parameter specifies the plugin-specific
                         logging level
                       type: string
+                    monitorAgent:
+                      description: monitor_agent plugin
+                      properties:
+                        bind:
+                          description: The bind address to listen to.
+                          type: string
+                        emitInterval:
+                          description: The interval time between event emits. This
+                            will be used when "tag" is configured.
+                          format: int64
+                          type: integer
+                        includeConfig:
+                          description: You can set this option to false to remove
+                            the config field from the response.
+                          type: boolean
+                        includeRetry:
+                          description: You can set this option to false to remove
+                            the retry field from the response.
+                          type: boolean
+                        port:
+                          description: The port to listen to.
+                          format: int64
+                          type: integer
+                        tag:
+                          description: If you set this parameter, this plugin emits
+                            metrics as records.
+                          type: string
+                      type: object
                     sample:
                       description: in_sample plugin
                       properties:

config/crd/bases/fluentd.fluent.io_fluentds.yaml

@@ -2212,6 +2212,34 @@ spec:
                       description: The @log_level parameter specifies the plugin-specific
                         logging level
                       type: string
+                    monitorAgent:
+                      description: monitor_agent plugin
+                      properties:
+                        bind:
+                          description: The bind address to listen to.
+                          type: string
+                        emitInterval:
+                          description: The interval time between event emits. This
+                            will be used when "tag" is configured.
+                          format: int64
+                          type: integer
+                        includeConfig:
+                          description: You can set this option to false to remove
+                            the config field from the response.
+                          type: boolean
+                        includeRetry:
+                          description: You can set this option to false to remove
+                            the retry field from the response.
+                          type: boolean
+                        port:
+                          description: The port to listen to.
+                          format: int64
+                          type: integer
+                        tag:
+                          description: If you set this parameter, this plugin emits
+                            metrics as records.
+                          type: string
+                      type: object
                     sample:
                       description: in_sample plugin
                       properties:

docs/plugins/fluentd/input/monitor_agent.md

@@ -0,0 +1,13 @@
+# MonitorAgent
+
+The in_monitor_agent Input plugin exports Fluentd's internal metrics via REST API.
+
+
+| Field | Description | Scheme |
+| ----- | ----------- | ------ |
+| port | The port to listen to. | *int64 |
+| bind | The bind address to listen to. | *string |
+| tag | If you set this parameter, this plugin emits metrics as records. | *string |
+| emitInterval | The interval time between event emits. This will be used when \"tag\" is configured. | *int64 |
+| includeConfig | You can set this option to false to remove the config field from the response. | *bool |
+| includeRetry | You can set this option to false to remove the retry field from the response. | *bool |

docs/plugins/fluentd/input/types.md

@@ -20,3 +20,4 @@ Input defines all available input plugins and their parameters
 | tail | in_tail plugin | *Tail |
 | sample | in_sample plugin | *Sample |
 | customPlugin | Custom plugin type | *custom.CustomPlugin |
+| monitor_agent | monitor_agent plugin | *MonitorAgent |

go.mod

@@ -58,12 +58,14 @@ require (
 	go.uber.org/atomic v1.9.0 // indirect
 	go.uber.org/multierr v1.7.0 // indirect
 	go.uber.org/zap v1.24.0 // indirect
+	golang.org/x/mod v0.12.0 // indirect
 	golang.org/x/net v0.14.0 // indirect
 	golang.org/x/oauth2 v0.0.0-20220223155221-ee480838109b // indirect
 	golang.org/x/sys v0.11.0 // indirect
 	golang.org/x/term v0.11.0 // indirect
 	golang.org/x/text v0.12.0 // indirect
 	golang.org/x/time v0.3.0 // indirect
+	golang.org/x/tools v0.12.0 // indirect
 	gomodules.xyz/jsonpatch/v2 v2.2.0 // indirect
 	google.golang.org/appengine v1.6.7 // indirect
 	google.golang.org/protobuf v1.28.1 // indirect
@@ -72,7 +74,9 @@ require (
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 	k8s.io/apiextensions-apiserver v0.26.1 // indirect
+	k8s.io/code-generator v0.26.1 // indirect
 	k8s.io/component-base v0.26.1 // indirect
+	k8s.io/gengo v0.0.0-20220902162205-c0856e24416d // indirect
 	k8s.io/kube-openapi v0.0.0-20230501164219-8b0f38b5fd1f // indirect
 	k8s.io/utils v0.0.0-20230209194617-a36077c30491 // indirect
 	sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd // indirect

go.sum

@@ -89,6 +89,7 @@ github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V
 github.com/go-logfmt/logfmt v0.5.0/go.mod h1:wCYkCAKZfumFQihp8CzCvQ3paCTfi41vtzG1KdI/P7A=
 github.com/go-logfmt/logfmt v0.5.1 h1:otpy5pqBCBZ1ng9RQ0dPu4PN7ba75Y/aA+UpowDyNVA=
 github.com/go-logfmt/logfmt v0.5.1/go.mod h1:WYhtIu8zTZfxdn5+rREduYbwxfcBr/Vr6KEVveWlfTs=
+github.com/go-logr/logr v0.2.0/go.mod h1:z6/tIYblkpsD+a4lm/fGIIU9mZ+XfAiaFtq7xTgseGU=
 github.com/go-logr/logr v1.2.0/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
 github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
 github.com/go-logr/logr v1.2.4 h1:g01GSCwiDw2xSZfjJ2/T9M+S6pFdcNtFYsp+Y43HYDQ=
@@ -349,6 +350,8 @@ golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzB
 golang.org/x/mod v0.1.1-0.20191107180719-034126e5016b/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg=
 golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
+golang.org/x/mod v0.12.0 h1:rmsUpXtvNzj340zd98LZ4KntptpfRHwpFOHG188oHXc=
+golang.org/x/mod v0.12.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
 golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
@@ -503,6 +506,7 @@ golang.org/x/tools v0.0.0-20200304193943-95d2e580d8eb/go.mod h1:o4KQGtdN14AW+yjs
 golang.org/x/tools v0.0.0-20200312045724-11d5b4c81c7d/go.mod h1:o4KQGtdN14AW+yjsvvwRTJJuXz8XRtIHtEnmAXLyFUw=
 golang.org/x/tools v0.0.0-20200331025713-a30bf2db82d4/go.mod h1:Sl4aGygMT6LrqrWclx+PTx3U+LnKx/seiNR+3G19Ar8=
 golang.org/x/tools v0.0.0-20200501065659-ab2804fb9c9d/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
+golang.org/x/tools v0.0.0-20200505023115-26f46d2f7ef8/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
 golang.org/x/tools v0.0.0-20200512131952-2bc93b1c0c88/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
 golang.org/x/tools v0.0.0-20200515010526-7d3b6ebf133d/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
 golang.org/x/tools v0.0.0-20200618134242-20370b0cb4b2/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
@@ -513,6 +517,7 @@ golang.org/x/tools v0.0.0-20200825202427-b303f430e36d/go.mod h1:njjCfa9FT2d7l9Bc
 golang.org/x/tools v0.0.0-20201224043029-2b0845dc783e/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.12.0 h1:YW6HUoUmYBpwSgyaGaZq1fHjrBjX1rlpZ54T6mu2kss=
+golang.org/x/tools v0.12.0/go.mod h1:Sc0INKfu04TlqNoRA1hgpFZbhYXHPr4V5DzpSBTPqQM=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
@@ -640,8 +645,13 @@ k8s.io/apimachinery v0.27.4 h1:CdxflD4AF61yewuid0fLl6bM4a3q04jWel0IlP+aYjs=
 k8s.io/apimachinery v0.27.4/go.mod h1:XNfZ6xklnMCOGGFNqXG7bUrQCoR04dh/E7FprV6pb+E=
 k8s.io/client-go v0.26.3 h1:k1UY+KXfkxV2ScEL3gilKcF7761xkYsSD6BC9szIu8s=
 k8s.io/client-go v0.26.3/go.mod h1:ZPNu9lm8/dbRIPAgteN30RSXea6vrCpFvq+MateTUuQ=
+k8s.io/code-generator v0.26.1 h1:dusFDsnNSKlMFYhzIM0jAO1OlnTN5WYwQQ+Ai12IIlo=
+k8s.io/code-generator v0.26.1/go.mod h1:OMoJ5Dqx1wgaQzKgc+ZWaZPfGjdRq/Y3WubFrZmeI3I=
 k8s.io/component-base v0.26.1 h1:4ahudpeQXHZL5kko+iDHqLj/FSGAEUnSVO0EBbgDd+4=
 k8s.io/component-base v0.26.1/go.mod h1:VHrLR0b58oC035w6YQiBSbtsf0ThuSwXP+p5dD/kAWU=
+k8s.io/gengo v0.0.0-20220902162205-c0856e24416d h1:U9tB195lKdzwqicbJvyJeOXV7Klv+wNAWENRnXEGi08=
+k8s.io/gengo v0.0.0-20220902162205-c0856e24416d/go.mod h1:FiNAH4ZV3gBg2Kwh89tzAEV2be7d5xI0vBa/VySYy3E=
+k8s.io/klog/v2 v2.2.0/go.mod h1:Od+F08eJP+W3HUb4pSrPpgp9DGU4GzlpG/TmITuYh/Y=
 k8s.io/klog/v2 v2.100.1 h1:7WCHKK6K8fNhTqfBhISHQ97KrnJNFZMcQvKp7gP/tmg=
 k8s.io/klog/v2 v2.100.1/go.mod h1:y1WjHnz7Dj687irZUWR/WLkLc5N1YHtjLdmgWjndZn0=
 k8s.io/kube-openapi v0.0.0-20230501164219-8b0f38b5fd1f h1:2kWPakN3i/k81b0gvD5C5FJ2kxm1WrQFanWchyKuqGg=
@@ -657,5 +667,6 @@ sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd h1:EDPBXCAspyGV4jQlpZSudPeMm
 sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd/go.mod h1:B8JuhiUyNFVKdsE8h686QcCxMaH6HrOAZj4vswFpcB0=
 sigs.k8s.io/structured-merge-diff/v4 v4.2.3 h1:PRbqxJClWWYMNV1dhaG4NsibJbArud9kFxnAMREiWFE=
 sigs.k8s.io/structured-merge-diff/v4 v4.2.3/go.mod h1:qjx8mGObPmV2aSZepjQjbmb2ihdVs8cGKBraizNC69E=
+sigs.k8s.io/yaml v1.2.0/go.mod h1:yfXDCHCao9+ENCvLSE62v9VSji2MKu5jeNfTrofGhJc=
 sigs.k8s.io/yaml v1.3.0 h1:a2VclLzOGrwOHDiV8EfBGhvjHvP46CtW5j6POvhYGGo=
 sigs.k8s.io/yaml v1.3.0/go.mod h1:GeOyir5tyXNByN85N/dRIT9es5UQNerPYEKK56eTBm8=

manifests/setup/fluent-operator-crd.yaml

@@ -21184,6 +21184,34 @@ spec:
                       description: The @log_level parameter specifies the plugin-specific
                         logging level
                       type: string
+                    monitorAgent:
+                      description: monitor_agent plugin
+                      properties:
+                        bind:
+                          description: The bind address to listen to.
+                          type: string
+                        emitInterval:
+                          description: The interval time between event emits. This
+                            will be used when "tag" is configured.
+                          format: int64
+                          type: integer
+                        includeConfig:
+                          description: You can set this option to false to remove
+                            the config field from the response.
+                          type: boolean
+                        includeRetry:
+                          description: You can set this option to false to remove
+                            the retry field from the response.
+                          type: boolean
+                        port:
+                          description: The port to listen to.
+                          format: int64
+                          type: integer
+                        tag:
+                          description: If you set this parameter, this plugin emits
+                            metrics as records.
+                          type: string
+                      type: object
                     sample:
                       description: in_sample plugin
                       properties: