fix(cloudwatch): cap maxWrite at 1MB and validate configuration

vparfonov · claude · openshift-merge-bot[bot] · commit 218eb4af6707 · 2026-06-10T12:39:14.000Z
CloudWatch Logs PutLogEvents API has a 1MB (1,048,576 bytes) limit per
request. Previously, ClusterLogForwarder allowed setting tuning.maxWrite
to values exceeding this limit (e.g., 10M), causing batch upload failures
with "Upload too large" errors and event drops.

Implement two layers of protection:

1. Validation: Reject ClusterLogForwarder specs at admission time if
   CloudWatch output maxWrite exceeds 1MB, providing users with clear
   error messages. This follows the existing pattern used for Azure
   Logs Ingestion output which has the same limit.

2. Runtime capping: The generator always sets batch.max_bytes to 1MB
   in the Vector configuration, ensuring safe operation regardless of
   whether tuning is configured.

Fixes: LOG-9448

Co-Authored-By: Claude Haiku 4.5 &lt;noreply@anthropic.com&gt;
diff --git a/internal/generator/vector/output/aws/cloudwatch/cloudwatch.go b/internal/generator/vector/output/aws/cloudwatch/cloudwatch.go
@@ -22,6 +22,9 @@ import (
 const (
 	groupNameField                  = "cw_group_name"
 	templatedInternalGroupNameField = `{{ _internal.` + groupNameField + ` }}`
+
+	// CloudwatchDefaultMaxBytes CloudWatch Logs PutLogEvents API has a 1MB per request limit
+	CloudwatchDefaultMaxBytes = 1_048_576
 )
 
 func New(id string, o *adapters.Output, inputs []string, secrets observability.Secrets, op utils.Options) (_ string, sink types.Sink, tfs api.Transforms) {
@@ -42,7 +45,14 @@ func New(id string, o *adapters.Output, inputs []string, secrets observability.S
 		} else {
 			s.Compression = sinks.CompressionType(o.GetTuning().Compression)
 		}
-		s.Batch = common.NewApiBatch(o)
+		if batch := common.NewApiBatch(o); batch != nil {
+			if batch.MaxBytes > CloudwatchDefaultMaxBytes {
+				batch.MaxBytes = CloudwatchDefaultMaxBytes
+			}
+			s.Batch = batch
+		} else {
+			s.Batch = &sinks.Batch{MaxBytes: CloudwatchDefaultMaxBytes}
+		}
 		s.Buffer = common.NewApiBuffer(o)
 		s.Request = common.NewApiRequest(o)
 		s.TLS = tls.NewTls(o, secrets, op)
diff --git a/internal/generator/vector/output/aws/cloudwatch/files/cw_groupname_with_aws_credentials.toml b/internal/generator/vector/output/aws/cloudwatch/files/cw_groupname_with_aws_credentials.toml
@@ -43,5 +43,8 @@ except_fields = ["_internal"]
 credentials_file = "/var/run/ocp-collector/config/my-forwarder-aws-creds/credentials"
 profile = "output_cw"
 
+[sinks.cw.batch]
+max_bytes = 1048576
+
 [sinks.cw.healthcheck]
 enabled = false
diff --git a/internal/generator/vector/output/aws/cloudwatch/files/cw_key_auth_and_assume_role.toml b/internal/generator/vector/output/aws/cloudwatch/files/cw_key_auth_and_assume_role.toml
@@ -45,5 +45,8 @@ secret_access_key = "SECRET[kubernetes_secret.vector-cw-secret/aws_secret_access
 assume_role = "SECRET[kubernetes_secret.secretwithcredentials/my_role_arn]"
 external_id = "unique-external-id"
 
+[sinks.cw.batch]
+max_bytes = 1048576
+
 [sinks.cw.healthcheck]
 enabled = false
diff --git a/internal/generator/vector/output/aws/cloudwatch/files/cw_with_groupname.toml b/internal/generator/vector/output/aws/cloudwatch/files/cw_with_groupname.toml
@@ -42,5 +42,8 @@ except_fields = ["_internal"]
 access_key_id = "SECRET[kubernetes_secret.vector-cw-secret/aws_access_key_id]"
 secret_access_key = "SECRET[kubernetes_secret.vector-cw-secret/aws_secret_access_key]"
 
+[sinks.cw.batch]
+max_bytes = 1048576
+
 [sinks.cw.healthcheck]
 enabled = false
diff --git a/internal/generator/vector/output/aws/cloudwatch/files/cw_with_tls_and_default_mintls_ciphers.toml b/internal/generator/vector/output/aws/cloudwatch/files/cw_with_tls_and_default_mintls_ciphers.toml
@@ -43,6 +43,9 @@ except_fields = ["_internal"]
 min_tls_version = "VersionTLS12"
 ciphersuites = "TLS_AES_128_GCM_SHA256,TLS_AES_256_GCM_SHA384,TLS_CHACHA20_POLY1305_SHA256,ECDHE-ECDSA-AES128-GCM-SHA256,ECDHE-RSA-AES128-GCM-SHA256,ECDHE-ECDSA-AES256-GCM-SHA384,ECDHE-RSA-AES256-GCM-SHA384,ECDHE-ECDSA-CHACHA20-POLY1305,ECDHE-RSA-CHACHA20-POLY1305,DHE-RSA-AES128-GCM-SHA256,DHE-RSA-AES256-GCM-SHA384"
 
+[sinks.cw.batch]
+max_bytes = 1048576
+
 [sinks.cw.auth]
 access_key_id = "SECRET[kubernetes_secret.vector-cw-secret/aws_access_key_id]"
 secret_access_key = "SECRET[kubernetes_secret.vector-cw-secret/aws_secret_access_key]"
diff --git a/internal/generator/vector/output/aws/cloudwatch/files/cw_with_tls_spec.toml b/internal/generator/vector/output/aws/cloudwatch/files/cw_with_tls_spec.toml
@@ -44,6 +44,9 @@ key_file = "/var/run/ocp-collector/secrets/vector-cw-secret-tls/tls.key"
 crt_file = "/var/run/ocp-collector/secrets/vector-cw-secret-tls/tls.crt"
 ca_file = "/var/run/ocp-collector/secrets/vector-cw-secret-tls/ca-bundle.crt"
 
+[sinks.cw.batch]
+max_bytes = 1048576
+
 [sinks.cw.auth]
 access_key_id = "SECRET[kubernetes_secret.vector-cw-secret/aws_access_key_id]"
 secret_access_key = "SECRET[kubernetes_secret.vector-cw-secret/aws_secret_access_key]"
diff --git a/internal/generator/vector/output/aws/cloudwatch/files/cw_with_tls_spec_insecure_verify.toml b/internal/generator/vector/output/aws/cloudwatch/files/cw_with_tls_spec_insecure_verify.toml
@@ -46,6 +46,9 @@ key_file = "/var/run/ocp-collector/secrets/vector-cw-secret-tls/tls.key"
 crt_file = "/var/run/ocp-collector/secrets/vector-cw-secret-tls/tls.crt"
 ca_file = "/var/run/ocp-collector/secrets/vector-cw-secret-tls/ca-bundle.crt"
 
+[sinks.cw.batch]
+max_bytes = 1048576
+
 [sinks.cw.auth]
 access_key_id = "SECRET[kubernetes_secret.vector-cw-secret/aws_access_key_id]"
 secret_access_key = "SECRET[kubernetes_secret.vector-cw-secret/aws_secret_access_key]"
diff --git a/internal/generator/vector/output/aws/cloudwatch/files/cw_with_tuning.toml b/internal/generator/vector/output/aws/cloudwatch/files/cw_with_tuning.toml
@@ -40,7 +40,7 @@ codec = "json"
 except_fields = ["_internal"]
 
 [sinks.cw.batch]
-max_bytes = 10000000
+max_bytes = 1048576
 
 [sinks.cw.buffer]
 type = "disk"
diff --git a/internal/generator/vector/output/aws/cloudwatch/files/cw_with_url.toml b/internal/generator/vector/output/aws/cloudwatch/files/cw_with_url.toml
@@ -44,5 +44,8 @@ except_fields = ["_internal"]
 access_key_id = "SECRET[kubernetes_secret.vector-cw-secret/aws_access_key_id]"
 secret_access_key = "SECRET[kubernetes_secret.vector-cw-secret/aws_secret_access_key]"
 
+[sinks.cw.batch]
+max_bytes = 1048576
+
 [sinks.cw.healthcheck]
 enabled = false
diff --git a/internal/validations/observability/outputs/validate.go b/internal/validations/observability/outputs/validate.go
@@ -25,6 +25,9 @@ func Validate(context internalcontext.ForwarderContext) {
 		switch out.Type {
 		case obs.OutputTypeCloudwatch, obs.OutputTypeS3:
 			messages = append(messages, ValidateAwsAuth(out, context)...)
+			if out.Type == obs.OutputTypeCloudwatch {
+				messages = append(messages, validateCloudwatchMaxWrite(out)...)
+			}
 		case obs.OutputTypeGoogleCloudLogging:
 			messages = append(messages, ValidateGCLAuth(out, context)...)
 		case obs.OutputTypeHTTP:
diff --git a/internal/validations/observability/outputs/validate_cloudwatch.go b/internal/validations/observability/outputs/validate_cloudwatch.go
@@ -0,0 +1,22 @@
+package outputs
+
+import (
+	"fmt"
+
+	obs "github.com/openshift/cluster-logging-operator/api/observability/v1"
+	"github.com/openshift/cluster-logging-operator/internal/generator/vector/output/aws/cloudwatch"
+)
+
+func validateCloudwatchMaxWrite(output obs.OutputSpec) (results []string) {
+	if output.Type != obs.OutputTypeCloudwatch || output.Cloudwatch == nil {
+		return results
+	}
+	if output.Cloudwatch.Tuning == nil || output.Cloudwatch.Tuning.MaxWrite == nil {
+		return results
+	}
+	maxWrite := output.Cloudwatch.Tuning.MaxWrite.Value()
+	if maxWrite > cloudwatch.CloudwatchDefaultMaxBytes {
+		results = append(results, fmt.Sprintf("maxWrite must not exceed %d bytes for CloudWatch, got %d bytes", cloudwatch.CloudwatchDefaultMaxBytes, maxWrite))
+	}
+	return results
+}
diff --git a/internal/validations/observability/outputs/validate_cloudwatch_test.go b/internal/validations/observability/outputs/validate_cloudwatch_test.go
@@ -0,0 +1,70 @@
+package outputs
+
+import (
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	obs "github.com/openshift/cluster-logging-operator/api/observability/v1"
+	"github.com/openshift/cluster-logging-operator/internal/utils"
+	"k8s.io/apimachinery/pkg/api/resource"
+)
+
+var _ = Describe("[internal][validations] ClusterLogForwarder will validate maxWrite for Cloudwatch output", func() {
+	var (
+		spec obs.OutputSpec
+	)
+	BeforeEach(func() {
+		spec = obs.OutputSpec{
+			Name:       "cloudwatchOutput",
+			Type:       obs.OutputTypeCloudwatch,
+			Cloudwatch: &obs.Cloudwatch{},
+		}
+	})
+
+	Context("#validateCloudwatchMaxWrite", func() {
+
+		It("should pass validation when no tuning is set", func() {
+			Expect(validateCloudwatchMaxWrite(spec)).To(BeEmpty())
+		})
+
+		It("should pass validation when maxWrite is nil", func() {
+			spec.Cloudwatch.Tuning = &obs.CloudwatchTuningSpec{}
+			Expect(validateCloudwatchMaxWrite(spec)).To(BeEmpty())
+		})
+
+		It("should pass validation when maxWrite is under 1MB", func() {
+			spec.Cloudwatch.Tuning = &obs.CloudwatchTuningSpec{
+				BaseOutputTuningSpec: obs.BaseOutputTuningSpec{
+					MaxWrite: utils.GetPtr(resource.MustParse("500Ki")),
+				},
+			}
+			Expect(validateCloudwatchMaxWrite(spec)).To(BeEmpty())
+		})
+
+		It("should pass validation when maxWrite is exactly 1MB", func() {
+			spec.Cloudwatch.Tuning = &obs.CloudwatchTuningSpec{
+				BaseOutputTuningSpec: obs.BaseOutputTuningSpec{
+					MaxWrite: utils.GetPtr(resource.MustParse("1048576")),
+				},
+			}
+			Expect(validateCloudwatchMaxWrite(spec)).To(BeEmpty())
+		})
+
+		It("should fail validation when maxWrite exceeds 1MB", func() {
+			spec.Cloudwatch.Tuning = &obs.CloudwatchTuningSpec{
+				BaseOutputTuningSpec: obs.BaseOutputTuningSpec{
+					MaxWrite: utils.GetPtr(resource.MustParse("10M")),
+				},
+			}
+			Expect(validateCloudwatchMaxWrite(spec)).ToNot(BeEmpty())
+		})
+
+		It("should pass validation when output is not Cloudwatch", func() {
+			spec = obs.OutputSpec{
+				Name: "httpOutput",
+				Type: obs.OutputTypeHTTP,
+				HTTP: &obs.HTTP{},
+			}
+			Expect(validateCloudwatchMaxWrite(spec)).To(BeEmpty())
+		})
+	})
+})
