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.