feat(fluentbit): Support more settings for firehose output

apis/fluentbit/v1alpha2/plugins/output/firehose_types.go

@@ -36,6 +36,17 @@ type Firehose struct {
 	STSEndpoint *string `json:"stsEndpoint,omitempty"`
 	// Immediately retry failed requests to AWS services once. This option does not affect the normal Fluent Bit retry mechanism with backoff. Instead, it enables an immediate retry with no delay for networking errors, which may help improve throughput when there are transient/random networking issues.
 	AutoRetryRequests *bool `json:"autoRetryRequests,omitempty"`
+	// Compression type to use when compressing the data. Valid values are: gzip, snappy, lz4, zstd. If you do not specify a compression type, the data will be sent uncompressed.
+	// +kubebuilder:validation:Enum=gzip;snappy;lz4;zstd
+	Compression *string `json:"compression,omitempty"`
+	// Specify an external ID for the STS API, can be used with the role_arn parameter if your role requires an external ID.
+	ExternalID *string `json:"externalID,omitempty"`
+	// Option to specify an AWS Profile for credentials.
+	Profile *string `json:"profile,omitempty"`
+	// Option to enable simple aggregation for the Firehose output plugin.
+	SimpleAggregation *bool `json:"simpleAggregation,omitempty"`
+	// Specify number of worker threads to use to output to Firehose
+	Workers *int32 `json:"workers,omitempty"`
 }
 
 // implement Section() method
@@ -76,5 +87,21 @@ func (l *Firehose) Params(sl plugins.SecretLoader) (*params.KVs, error) {
 		kvs.Insert("auto_retry_requests", strconv.FormatBool(*l.AutoRetryRequests))
 	}
 
+	if l.Compression != nil && *l.Compression != "" {
+		kvs.Insert("compression", *l.Compression)
+	}
+	if l.ExternalID != nil && *l.ExternalID != "" {
+		kvs.Insert("external_id", *l.ExternalID)
+	}
+	if l.SimpleAggregation != nil {
+		kvs.Insert("simple_aggregation", strconv.FormatBool(*l.SimpleAggregation))
+	}
+	if l.Profile != nil && *l.Profile != "" {
+		kvs.Insert("profile", *l.Profile)
+	}
+	if l.Workers != nil {
+		kvs.Insert("workers", strconv.FormatInt(int64(*l.Workers), 10))
+	}
+
 	return kvs, nil
 }

apis/fluentbit/v1alpha2/plugins/output/firehose_types_test.go

@@ -0,0 +1,55 @@
+package output
+
+import (
+	"testing"
+
+	"github.com/fluent/fluent-operator/v3/apis/fluentbit/v1alpha2/plugins"
+	"github.com/fluent/fluent-operator/v3/apis/fluentbit/v1alpha2/plugins/params"
+	"github.com/fluent/fluent-operator/v3/pkg/utils"
+	"github.com/onsi/gomega"
+)
+
+func TestOutput_Firehose_Params(t *testing.T) {
+	g := gomega.NewWithT(t)
+
+	sl := plugins.NewSecretLoader(nil, "test namespace")
+
+	fh := Firehose{
+		Region:            "us-east-1",
+		DeliveryStream:    "test_stream",
+		TimeKey:           utils.ToPtr("test_time_key"),
+		TimeKeyFormat:     utils.ToPtr("%Y-%m-%dT%H:%M:%S.%3N"),
+		DataKeys:          utils.ToPtr("test_data_keys"),
+		LogKey:            utils.ToPtr("test_time_key"),
+		RoleARN:           utils.ToPtr("arn:aws:iam:test"),
+		Endpoint:          utils.ToPtr("test_endpoint"),
+		STSEndpoint:       utils.ToPtr("test_sts_endpoint"),
+		AutoRetryRequests: utils.ToPtr(true),
+		ExternalID:        utils.ToPtr("test_external_id"),
+		Compression:       utils.ToPtr("gzip"),
+		SimpleAggregation: utils.ToPtr(true),
+		Profile:           utils.ToPtr("my-profile"),
+		Workers:           utils.ToPtr[int32](1),
+	}
+
+	expected := params.NewKVs()
+	expected.Insert("region", "us-east-1")
+	expected.Insert("delivery_stream", "test_stream")
+	expected.Insert("data_keys", "test_data_keys")
+	expected.Insert("log_key", "test_time_key")
+	expected.Insert("role_arn", "arn:aws:iam:test")
+	expected.Insert("endpoint", "test_endpoint")
+	expected.Insert("sts_endpoint", "test_sts_endpoint")
+	expected.Insert("time_key", "test_time_key")
+	expected.Insert("time_key_format", "%Y-%m-%dT%H:%M:%S.%3N")
+	expected.Insert("auto_retry_requests", "true")
+	expected.Insert("compression", "gzip")
+	expected.Insert("external_id", "test_external_id")
+	expected.Insert("simple_aggregation", "true")
+	expected.Insert("profile", "my-profile")
+	expected.Insert("workers", "1")
+
+	kvs, err := fh.Params(sl)
+	g.Expect(err).NotTo(gomega.HaveOccurred())
+	g.Expect(kvs).To(gomega.Equal(expected))
+}

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

@@ -993,6 +993,16 @@ spec:
                       with no delay for networking errors, which may help improve
                       throughput when there are transient/random networking issues.
                     type: boolean
+                  compression:
+                    description: 'Compression type to use when compressing the data.
+                      Valid values are: gzip, snappy, lz4, zstd. If you do not specify
+                      a compression type, the data will be sent uncompressed.'
+                    enum:
+                    - gzip
+                    - snappy
+                    - lz4
+                    - zstd
+                    type: string
                   dataKeys:
                     description: By default, the whole log record will be sent to
                       Kinesis. If you specify a key name(s) with this option, then
@@ -1009,19 +1019,31 @@ spec:
                     description: Specify a custom endpoint for the Kinesis Firehose
                       API.
                     type: string
+                  externalID:
+                    description: Specify an external ID for the STS API, can be used
+                      with the role_arn parameter if your role requires an external
+                      ID.
+                    type: string
                   logKey:
                     description: By default, the whole log record will be sent to
                       Firehose. If you specify a key name with this option, then only
                       the value of that key will be sent to Firehose. For example,
                       if you are using the Fluentd Docker log driver, you can specify
                       log_key log and only the log message will be sent to Firehose.
                     type: string
+                  profile:
+                    description: Option to specify an AWS Profile for credentials.
+                    type: string
                   region:
                     description: The AWS region.
                     type: string
                   roleARN:
                     description: ARN of an IAM role to assume (for cross account access).
                     type: string
+                  simpleAggregation:
+                    description: Option to enable simple aggregation for the Firehose
+                      output plugin.
+                    type: boolean
                   stsEndpoint:
                     description: Specify a custom endpoint for the STS API; used to
                       assume your custom role provided with role_arn.
@@ -1039,6 +1061,11 @@ spec:
                       ECS Container Agent v1.42.0 or later, otherwise the timestamps
                       associated with your container logs will only have second precision.
                     type: string
+                  workers:
+                    description: Specify number of worker threads to use to output
+                      to Firehose
+                    format: int32
+                    type: integer
                 required:
                 - deliveryStream
                 - region

charts/fluent-operator-fluent-bit-crds/templates/fluentbit.fluent.io_outputs.yaml

@@ -993,6 +993,16 @@ spec:
                       with no delay for networking errors, which may help improve
                       throughput when there are transient/random networking issues.
                     type: boolean
+                  compression:
+                    description: 'Compression type to use when compressing the data.
+                      Valid values are: gzip, snappy, lz4, zstd. If you do not specify
+                      a compression type, the data will be sent uncompressed.'
+                    enum:
+                    - gzip
+                    - snappy
+                    - lz4
+                    - zstd
+                    type: string
                   dataKeys:
                     description: By default, the whole log record will be sent to
                       Kinesis. If you specify a key name(s) with this option, then
@@ -1009,19 +1019,31 @@ spec:
                     description: Specify a custom endpoint for the Kinesis Firehose
                       API.
                     type: string
+                  externalID:
+                    description: Specify an external ID for the STS API, can be used
+                      with the role_arn parameter if your role requires an external
+                      ID.
+                    type: string
                   logKey:
                     description: By default, the whole log record will be sent to
                       Firehose. If you specify a key name with this option, then only
                       the value of that key will be sent to Firehose. For example,
                       if you are using the Fluentd Docker log driver, you can specify
                       log_key log and only the log message will be sent to Firehose.
                     type: string
+                  profile:
+                    description: Option to specify an AWS Profile for credentials.
+                    type: string
                   region:
                     description: The AWS region.
                     type: string
                   roleARN:
                     description: ARN of an IAM role to assume (for cross account access).
                     type: string
+                  simpleAggregation:
+                    description: Option to enable simple aggregation for the Firehose
+                      output plugin.
+                    type: boolean
                   stsEndpoint:
                     description: Specify a custom endpoint for the STS API; used to
                       assume your custom role provided with role_arn.
@@ -1039,6 +1061,11 @@ spec:
                       ECS Container Agent v1.42.0 or later, otherwise the timestamps
                       associated with your container logs will only have second precision.
                     type: string
+                  workers:
+                    description: Specify number of worker threads to use to output
+                      to Firehose
+                    format: int32
+                    type: integer
                 required:
                 - deliveryStream
                 - region

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

@@ -991,6 +991,16 @@ spec:
                       with no delay for networking errors, which may help improve
                       throughput when there are transient/random networking issues.
                     type: boolean
+                  compression:
+                    description: 'Compression type to use when compressing the data.
+                      Valid values are: gzip, snappy, lz4, zstd. If you do not specify
+                      a compression type, the data will be sent uncompressed.'
+                    enum:
+                    - gzip
+                    - snappy
+                    - lz4
+                    - zstd
+                    type: string
                   dataKeys:
                     description: By default, the whole log record will be sent to
                       Kinesis. If you specify a key name(s) with this option, then
@@ -1007,19 +1017,31 @@ spec:
                     description: Specify a custom endpoint for the Kinesis Firehose
                       API.
                     type: string
+                  externalID:
+                    description: Specify an external ID for the STS API, can be used
+                      with the role_arn parameter if your role requires an external
+                      ID.
+                    type: string
                   logKey:
                     description: By default, the whole log record will be sent to
                       Firehose. If you specify a key name with this option, then only
                       the value of that key will be sent to Firehose. For example,
                       if you are using the Fluentd Docker log driver, you can specify
                       log_key log and only the log message will be sent to Firehose.
                     type: string
+                  profile:
+                    description: Option to specify an AWS Profile for credentials.
+                    type: string
                   region:
                     description: The AWS region.
                     type: string
                   roleARN:
                     description: ARN of an IAM role to assume (for cross account access).
                     type: string
+                  simpleAggregation:
+                    description: Option to enable simple aggregation for the Firehose
+                      output plugin.
+                    type: boolean
                   stsEndpoint:
                     description: Specify a custom endpoint for the STS API; used to
                       assume your custom role provided with role_arn.
@@ -1037,6 +1059,11 @@ spec:
                       ECS Container Agent v1.42.0 or later, otherwise the timestamps
                       associated with your container logs will only have second precision.
                     type: string
+                  workers:
+                    description: Specify number of worker threads to use to output
+                      to Firehose
+                    format: int32
+                    type: integer
                 required:
                 - deliveryStream
                 - region

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

@@ -991,6 +991,16 @@ spec:
                       with no delay for networking errors, which may help improve
                       throughput when there are transient/random networking issues.
                     type: boolean
+                  compression:
+                    description: 'Compression type to use when compressing the data.
+                      Valid values are: gzip, snappy, lz4, zstd. If you do not specify
+                      a compression type, the data will be sent uncompressed.'
+                    enum:
+                    - gzip
+                    - snappy
+                    - lz4
+                    - zstd
+                    type: string
                   dataKeys:
                     description: By default, the whole log record will be sent to
                       Kinesis. If you specify a key name(s) with this option, then
@@ -1007,19 +1017,31 @@ spec:
                     description: Specify a custom endpoint for the Kinesis Firehose
                       API.
                     type: string
+                  externalID:
+                    description: Specify an external ID for the STS API, can be used
+                      with the role_arn parameter if your role requires an external
+                      ID.
+                    type: string
                   logKey:
                     description: By default, the whole log record will be sent to
                       Firehose. If you specify a key name with this option, then only
                       the value of that key will be sent to Firehose. For example,
                       if you are using the Fluentd Docker log driver, you can specify
                       log_key log and only the log message will be sent to Firehose.
                     type: string
+                  profile:
+                    description: Option to specify an AWS Profile for credentials.
+                    type: string
                   region:
                     description: The AWS region.
                     type: string
                   roleARN:
                     description: ARN of an IAM role to assume (for cross account access).
                     type: string
+                  simpleAggregation:
+                    description: Option to enable simple aggregation for the Firehose
+                      output plugin.
+                    type: boolean
                   stsEndpoint:
                     description: Specify a custom endpoint for the STS API; used to
                       assume your custom role provided with role_arn.
@@ -1037,6 +1059,11 @@ spec:
                       ECS Container Agent v1.42.0 or later, otherwise the timestamps
                       associated with your container logs will only have second precision.
                     type: string
+                  workers:
+                    description: Specify number of worker threads to use to output
+                      to Firehose
+                    format: int32
+                    type: integer
                 required:
                 - deliveryStream
                 - region