From e20a96113c8c8a43dc279f6185023106521a77a2 Mon Sep 17 00:00:00 2001 From: aibek Date: Tue, 13 Oct 2026 11:58:46 -0400 Subject: [PATCH] feat(fluentbit): add workers option to http output Expose the generic Fluent Bit Workers output option on the http output, as already done for syslog, s3, firehose, loki and others. Regenerates deepcopy, CRDs, manifests and docs, and adds unit tests covering Params, Load and LoadAsYaml rendering. Signed-off-by: aibek --- .../v1alpha2/clusteroutput_types_test.go | 65 +++++++++++++++++++ .../v1alpha2/plugins/output/http_types.go | 3 + .../plugins/output/http_types_test.go | 2 + .../plugins/output/zz_generated.deepcopy.go | 5 ++ .../fluentbit.fluent.io_clusteroutputs.yaml | 6 ++ .../fluentbit.fluent.io_outputs.yaml | 6 ++ .../fluentbit.fluent.io_clusteroutputs.yaml | 6 ++ .../crds/fluentbit.fluent.io_outputs.yaml | 6 ++ .../fluentbit.fluent.io_clusteroutputs.yaml | 6 ++ .../bases/fluentbit.fluent.io_outputs.yaml | 6 ++ docs/plugins/fluentbit/output/http.md | 1 + manifests/setup/setup.yaml | 12 ++++ 12 files changed, 124 insertions(+) diff --git a/apis/fluentbit/v1alpha2/clusteroutput_types_test.go b/apis/fluentbit/v1alpha2/clusteroutput_types_test.go index cd61d836d..23b24ed93 100644 --- a/apis/fluentbit/v1alpha2/clusteroutput_types_test.go +++ b/apis/fluentbit/v1alpha2/clusteroutput_types_test.go @@ -571,3 +571,68 @@ func TestForwardOutput_RetainMetadataInForwardMode(t *testing.T) { g.Expect(err).NotTo(HaveOccurred()) g.Expect(resultTrue).To(ContainSubstring("Retain_Metadata_In_Forward_Mode true")) } + +func httpOutputWithWorkers() ClusterOutputList { + return ClusterOutputList{ + Items: []ClusterOutput{{ + TypeMeta: metav1.TypeMeta{ + APIVersion: "fluentbit.fluent.io/v1alpha2", + Kind: "ClusterOutput", + }, + ObjectMeta: metav1.ObjectMeta{ + Name: "http_output_with_workers", + }, + Spec: OutputSpec{ + MatchRegex: "app.*", + RetryLimit: "8", + HTTP: &output.HTTP{ + Host: "example.com", + Port: new(int32(9481)), + Uri: "/insert/jsonline", + Format: "json_lines", + Workers: new(int32(4)), + }, + }, + }}, + } +} + +func TestHTTPOutputWithWorkers_Load(t *testing.T) { + g := NewGomegaWithT(t) + sl := plugins.NewSecretLoader(nil, "testnamespace") + + expected := `[Output] + Name http + Match_Regex app.* + Retry_Limit 8 + host example.com + port 9481 + uri /insert/jsonline + format json_lines + workers 4 +` + + result, err := httpOutputWithWorkers().Load(sl) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(result).To(Equal(expected)) +} + +func TestHTTPOutputWithWorkers_LoadAsYaml(t *testing.T) { + g := NewGomegaWithT(t) + sl := plugins.NewSecretLoader(nil, "testnamespace") + + expected := `outputs: + - name: http + match_regex: app.* + retry_limit: 8 + host: example.com + port: 9481 + uri: /insert/jsonline + format: json_lines + workers: 4 +` + + result, err := httpOutputWithWorkers().LoadAsYaml(sl, 0) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(result).To(Equal(expected)) +} diff --git a/apis/fluentbit/v1alpha2/plugins/output/http_types.go b/apis/fluentbit/v1alpha2/plugins/output/http_types.go index 8dc3fa751..d7a3d33aa 100644 --- a/apis/fluentbit/v1alpha2/plugins/output/http_types.go +++ b/apis/fluentbit/v1alpha2/plugins/output/http_types.go @@ -67,6 +67,8 @@ type HTTP struct { *plugins.Networking `json:"networking,omitempty"` // Limit the maximum number of Chunks in the filesystem for the current output logical destination. TotalLimitSize string `json:"totalLimitSize,omitempty"` + // Enables dedicated thread(s) for this output. Default value is set since version 1.8.13. For previous versions is 0. + Workers *int32 `json:"workers,omitempty"` } // implement Name method @@ -125,6 +127,7 @@ func (h *HTTP) Params(sl plugins.SecretLoader) (*params.KVs, error) { } plugins.InsertKVString(kvs, "storage.total_limit_size", h.TotalLimitSize) + plugins.InsertKVField(kvs, "workers", h.Workers) return kvs, nil } diff --git a/apis/fluentbit/v1alpha2/plugins/output/http_types_test.go b/apis/fluentbit/v1alpha2/plugins/output/http_types_test.go index 54f581e61..c33f353a0 100644 --- a/apis/fluentbit/v1alpha2/plugins/output/http_types_test.go +++ b/apis/fluentbit/v1alpha2/plugins/output/http_types_test.go @@ -35,6 +35,7 @@ func TestHTTP_Params(t *testing.T) { TLS: &plugins.TLS{Verify: new(false)}, Networking: &plugins.Networking{SourceAddress: new("expected_source_address")}, TotalLimitSize: "512M", + Workers: new(int32(4)), } expected := params.NewKVs() @@ -50,6 +51,7 @@ func TestHTTP_Params(t *testing.T) { expected.Insert("tls.verify", "false") expected.Insert("net.source_address", "expected_source_address") expected.Insert("storage.total_limit_size", "512M") + expected.Insert("workers", "4") kvs, err := h.Params(sl) g.Expect(err).NotTo(HaveOccurred()) diff --git a/apis/fluentbit/v1alpha2/plugins/output/zz_generated.deepcopy.go b/apis/fluentbit/v1alpha2/plugins/output/zz_generated.deepcopy.go index 3d1ce9e10..b5b514a34 100644 --- a/apis/fluentbit/v1alpha2/plugins/output/zz_generated.deepcopy.go +++ b/apis/fluentbit/v1alpha2/plugins/output/zz_generated.deepcopy.go @@ -484,6 +484,11 @@ func (in *HTTP) DeepCopyInto(out *HTTP) { *out = new(plugins.Networking) (*in).DeepCopyInto(*out) } + if in.Workers != nil { + in, out := &in.Workers, &out.Workers + *out = new(int32) + **out = **in + } } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new HTTP. diff --git a/charts/fluent-operator-fluent-bit-crds/templates/fluentbit.fluent.io_clusteroutputs.yaml b/charts/fluent-operator-fluent-bit-crds/templates/fluentbit.fluent.io_clusteroutputs.yaml index c597e3475..41976750a 100644 --- a/charts/fluent-operator-fluent-bit-crds/templates/fluentbit.fluent.io_clusteroutputs.yaml +++ b/charts/fluent-operator-fluent-bit-crds/templates/fluentbit.fluent.io_clusteroutputs.yaml @@ -1939,6 +1939,12 @@ spec: description: 'Specify an optional HTTP URI for the target web server, e.g: /something' type: string + workers: + description: Enables dedicated thread(s) for this output. Default + value is set since version 1.8.13. For previous versions is + 0. + format: int32 + type: integer type: object influxDB: description: InfluxDB defines InfluxDB Output configuration. diff --git a/charts/fluent-operator-fluent-bit-crds/templates/fluentbit.fluent.io_outputs.yaml b/charts/fluent-operator-fluent-bit-crds/templates/fluentbit.fluent.io_outputs.yaml index 8af040af3..95448c939 100644 --- a/charts/fluent-operator-fluent-bit-crds/templates/fluentbit.fluent.io_outputs.yaml +++ b/charts/fluent-operator-fluent-bit-crds/templates/fluentbit.fluent.io_outputs.yaml @@ -1939,6 +1939,12 @@ spec: description: 'Specify an optional HTTP URI for the target web server, e.g: /something' type: string + workers: + description: Enables dedicated thread(s) for this output. Default + value is set since version 1.8.13. For previous versions is + 0. + format: int32 + type: integer type: object influxDB: description: InfluxDB defines InfluxDB Output configuration. diff --git a/charts/fluent-operator/crds/fluentbit.fluent.io_clusteroutputs.yaml b/charts/fluent-operator/crds/fluentbit.fluent.io_clusteroutputs.yaml index 2d18fe3c1..82b305512 100644 --- a/charts/fluent-operator/crds/fluentbit.fluent.io_clusteroutputs.yaml +++ b/charts/fluent-operator/crds/fluentbit.fluent.io_clusteroutputs.yaml @@ -1937,6 +1937,12 @@ spec: description: 'Specify an optional HTTP URI for the target web server, e.g: /something' type: string + workers: + description: Enables dedicated thread(s) for this output. Default + value is set since version 1.8.13. For previous versions is + 0. + format: int32 + type: integer type: object influxDB: description: InfluxDB defines InfluxDB Output configuration. diff --git a/charts/fluent-operator/crds/fluentbit.fluent.io_outputs.yaml b/charts/fluent-operator/crds/fluentbit.fluent.io_outputs.yaml index 111347fc3..afd512b22 100644 --- a/charts/fluent-operator/crds/fluentbit.fluent.io_outputs.yaml +++ b/charts/fluent-operator/crds/fluentbit.fluent.io_outputs.yaml @@ -1937,6 +1937,12 @@ spec: description: 'Specify an optional HTTP URI for the target web server, e.g: /something' type: string + workers: + description: Enables dedicated thread(s) for this output. Default + value is set since version 1.8.13. For previous versions is + 0. + format: int32 + type: integer type: object influxDB: description: InfluxDB defines InfluxDB Output configuration. diff --git a/config/crd/bases/fluentbit.fluent.io_clusteroutputs.yaml b/config/crd/bases/fluentbit.fluent.io_clusteroutputs.yaml index c6454f0c4..ed37759bb 100644 --- a/config/crd/bases/fluentbit.fluent.io_clusteroutputs.yaml +++ b/config/crd/bases/fluentbit.fluent.io_clusteroutputs.yaml @@ -1938,6 +1938,12 @@ spec: description: 'Specify an optional HTTP URI for the target web server, e.g: /something' type: string + workers: + description: Enables dedicated thread(s) for this output. Default + value is set since version 1.8.13. For previous versions is + 0. + format: int32 + type: integer type: object influxDB: description: InfluxDB defines InfluxDB Output configuration. diff --git a/config/crd/bases/fluentbit.fluent.io_outputs.yaml b/config/crd/bases/fluentbit.fluent.io_outputs.yaml index 9645142d4..939a3adac 100644 --- a/config/crd/bases/fluentbit.fluent.io_outputs.yaml +++ b/config/crd/bases/fluentbit.fluent.io_outputs.yaml @@ -1938,6 +1938,12 @@ spec: description: 'Specify an optional HTTP URI for the target web server, e.g: /something' type: string + workers: + description: Enables dedicated thread(s) for this output. Default + value is set since version 1.8.13. For previous versions is + 0. + format: int32 + type: integer type: object influxDB: description: InfluxDB defines InfluxDB Output configuration. diff --git a/docs/plugins/fluentbit/output/http.md b/docs/plugins/fluentbit/output/http.md index 945eba9b7..248ff1ae5 100644 --- a/docs/plugins/fluentbit/output/http.md +++ b/docs/plugins/fluentbit/output/http.md @@ -27,3 +27,4 @@ The http output plugin allows to flush your records into a HTTP endpoint.
| tls | HTTP output plugin supports TTL/SSL, for more details about the properties available and general configuration, please refer to the TLS/SSL section. | *[plugins.TLS](../tls.md) | | networking | Include fluentbit networking options for this output-plugin | *[plugins.Networking](../net.md) | | totalLimitSize | Limit the maximum number of Chunks in the filesystem for the current output logical destination. | string | +| workers | Enables dedicated thread(s) for this output. Default value is set since version 1.8.13. For previous versions is 0. | *int32 | diff --git a/manifests/setup/setup.yaml b/manifests/setup/setup.yaml index ae10584d1..b0ca45551 100644 --- a/manifests/setup/setup.yaml +++ b/manifests/setup/setup.yaml @@ -6140,6 +6140,12 @@ spec: description: 'Specify an optional HTTP URI for the target web server, e.g: /something' type: string + workers: + description: Enables dedicated thread(s) for this output. Default + value is set since version 1.8.13. For previous versions is + 0. + format: int32 + type: integer type: object influxDB: description: InfluxDB defines InfluxDB Output configuration. @@ -37627,6 +37633,12 @@ spec: description: 'Specify an optional HTTP URI for the target web server, e.g: /something' type: string + workers: + description: Enables dedicated thread(s) for this output. Default + value is set since version 1.8.13. For previous versions is + 0. + format: int32 + type: integer type: object influxDB: description: InfluxDB defines InfluxDB Output configuration.