Repository: redpanda-data/connect (language: go)

Issue:
S3 output missing header since aws-sdk-go-v2 upgrade
Hi, I've recently attempted to upgrade from v4.24.0, but encountered an issue with how the S3 output behaves when writing to a bucket with object lock enabled.

Using Benthos with versions greater than 4.24.0, AWS rejects the PutObject request with error: "Content-MD5 OR x-amz-checksum- HTTP header is required for Put Object requests with Object Lock parameters".

I believe this is due to the switch to `aws-sdk-go-v2` [which occurred in v4.25.0](https://github.com/redpanda-data/connect/blob/main/CHANGELOG.md#4250---2024-03-01). I'm not very familiar with Go, but I've had a look through the code for `aws-sdk-go`, and it looks like the original version of the SDK would automatically set the `Content-MD5` header for each `PutObject` request. See [here](https://github.com/aws/aws-sdk-go/blob/main/service/s3/customizations.go#L56) and [here](https://github.com/aws/aws-sdk-go/blob/main/service/s3/body_hash.go#L80).

I don't expect the Benthos connector to derive the MD5 and apply it automatically like the old `aws-sdk-go` did, but would like a way to provide it to the output e.g.:

```
      aws_s3:
          bucket: ingest-prod-data
          path: ${! meta("write_path") }
          tags: {}
          region: eu-west-2
          content_type: application/octet-stream
          content_md5: $!{ meta("file_hash") }
``` 

I'd be grateful if someone could check my assumptions regarding the `aws-sdk-go` to make sure I'm reading it right, and for alternative suggestions for how to fix this, whether it needs to be fixed or we can work around it somehow.

Thanks



Discussion / hints:
Hey @dom-lee-naimuri 👋 As a workaround, would you mind trying to attach a `mapping` processor to that `aws_s3` output with `meta Content-MD5 = metadata("file_hash")`? Details here: https://docs.redpanda.com/redpanda-connect/components/outputs/aws_s3/#metadata-2. We can also add an extra field to the output if that's easier. PRs are welcome.
Hi @mihaitodor, thanks for this suggestion. I've done the following to test:

```
pipeline:
  processors:
    - mapping: |-
        meta "Content-MD5" = "value"
        meta mykey = "another value"


output:
  processors:
    - log:
          level: INFO
          message: '${! meta("Content-MD5") }'
  aws_s3:
    bucket: 'bucket-name'
    path: 'doms-test-file'
    tags: {}
    region: eu-west-2
    content_type: application/octet-stream

```

Unfortunately before being sent off in the request, metadata keys are prefixed with `x-amz-meta` and stored as user defined metadata in S3. I'll raise a PR to add `Content-MD5` to the S3 output when I get a chance

Fix patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 559ab9cf19..b5cafd4374 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -3,6 +3,12 @@ Changelog
 
 All notable changes to this project will be documented in this file.
 
+## 4.33.0 - TBD
+
+### Added
+
+- Field `content_md5` added to the `aws_s3` output. (@dom-lee-naimuri)
+
 ## 4.32.1 - 2024-07-24
 
 ### Changed
diff --git a/docs/modules/components/pages/outputs/aws_s3.adoc b/docs/modules/components/pages/outputs/aws_s3.adoc
index 23f03fbe7b..799777b6ef 100644
--- a/docs/modules/components/pages/outputs/aws_s3.adoc
+++ b/docs/modules/components/pages/outputs/aws_s3.adoc
@@ -71,6 +71,7 @@ output:
     cache_control: ""
     content_disposition: ""
     content_language: ""
+    content_md5: ""
     website_redirect_location: ""
     metadata:
       exclude_prefixes: []
@@ -258,6 +259,16 @@ The content language to set for each object.
 This field supports xref:configuration:interpolation.adoc#bloblang-queries[interpolation functions].
 
 
+*Type*: `string`
+
+*Default*: `""`
+
+=== `content_md5`
+
+The content MD5 to set for each object.
+This field supports xref:configuration:interpolation.adoc#bloblang-queries[interpolation functions].
+
+
 *Type*: `string`
 
 *Default*: `""`
diff --git a/internal/impl/aws/output_s3.go b/internal/impl/aws/output_s3.go
index a8542ef6cb..b375b382cd 100644
--- a/internal/impl/aws/output_s3.go
+++ b/internal/impl/aws/output_s3.go
@@ -45,6 +45,7 @@ const (
 	s3oFieldCacheControl            = "cache_control"
 	s3oFieldContentDisposition      = "content_disposition"
 	s3oFieldContentLanguage         = "content_language"
+	s3oFieldContentMD5              = "content_md5"
 	s3oFieldWebsiteRedirectLocation = "website_redirect_location"
 	s3oFieldMetadata                = "metadata"
 	s3oFieldStorageClass            = "storage_class"
@@ -69,6 +70,7 @@ type s3oConfig struct {
 	CacheControl            *service.InterpolatedString
 	ContentDisposition      *service.InterpolatedString
 	ContentLanguage         *service.InterpolatedString
+	ContentMD5              *service.InterpolatedString
 	WebsiteRedirectLocation *service.InterpolatedString
 	Metadata                *service.MetadataExcludeFilter
 	StorageClass            *service.InterpolatedString
@@ -121,6 +123,9 @@ func s3oConfigFromParsed(pConf *service.ParsedConfig) (conf s3oConfig, err error
 	if conf.ContentLanguage, err = pConf.FieldInterpolatedString(s3oFieldContentLanguage); err != nil {
 		return
 	}
+	if conf.ContentMD5, err = pConf.FieldInterpolatedString(s3oFieldContentMD5); err != nil {
+		return
+	}
 	if conf.WebsiteRedirectLocation, err = pConf.FieldInterpolatedString(s3oFieldWebsiteRedirectLocation); err != nil {
 		return
 	}
@@ -245,6 +250,10 @@ output:
 				Description("The content language to set for each object.").
 				Default("").
 				Advanced(),
+			service.NewInterpolatedStringField(s3oFieldContentMD5).
+				Description("The content MD5 to set for each object.").
+				Default("").
+				Advanced(),
 			service.NewInterpolatedStringField(s3oFieldWebsiteRedirectLocation).
 				Description("The website redirect location to set for each object.").
 				Default("").
@@ -371,6 +380,13 @@ func (a *amazonS3Writer) WriteBatch(wctx context.Context, msg service.MessageBat
 		if ce != "" {
 			contentLanguage = aws.String(ce)
 		}
+		var contentMD5 *string
+		if ce, err = msg.TryInterpolatedString(i, a.conf.ContentMD5); err != nil {
+			return fmt.Errorf("content MD5 interpolation: %w", err)
+		}
+		if ce != "" {
+			contentMD5 = aws.String(ce)
+		}
 		var websiteRedirectLocation *string
 		if ce, err = msg.TryInterpolatedString(i, a.conf.WebsiteRedirectLocation); err != nil {
 			return fmt.Errorf("website redirect location interpolation: %w", err)
@@ -408,6 +424,7 @@ func (a *amazonS3Writer) WriteBatch(wctx context.Context, msg service.MessageBat
 			CacheControl:            cacheControl,
 			ContentDisposition:      contentDisposition,
 			ContentLanguage:         contentLanguage,
+			ContentMD5:              contentMD5,
 			WebsiteRedirectLocation: websiteRedirectLocation,
 			StorageClass:            types.StorageClass(storageClass),
 			Metadata:                metadata,


Test patch: