[ENH] Use GCS with aws-sdk-go-v2 (#5878)

## Description of changes

_Summarize the changes made by this PR._

- Improvements & Bug fixes
  - N/A
- New functionality
- Hacks the S3 client initialization in SysDB so that it can support GCS
- Source:
https://stackoverflow.com/questions/73717477/gcp-cloud-storage-golang-aws-sdk2-upload-file-with-s3-interoperability-creds

## Test plan

_How are these changes tested?_

- [ ] Tests pass locally with `pytest` for python, `yarn test` for js,
`cargo test` for rust

## Migration plan

_Are there any migrations, or any forwards/backwards compatibility
changes needed in order to make sure this change deploys reliably?_

## Observability plan

_What is the plan to instrument and monitor this change?_

## Documentation Changes

_Are all docstrings for user-facing APIs updated if required? Do we need
to make documentation changes in the [docs
section](https://github.com/chroma-core/chroma/tree/main/docs/docs.trychroma.com)?_

Author: Sicheng-Pan

go/cmd/coordinator/cmd.go

@@ -80,6 +80,7 @@ func init() {
 	Cmd.Flags().StringVar(&conf.MetaStoreConfig.AccessKeyID, "s3-access-key-id", "", "S3 access key ID")
 	Cmd.Flags().StringVar(&conf.MetaStoreConfig.SecretAccessKey, "s3-secret-access-key", "", "S3 secret access key")
 	Cmd.Flags().BoolVar(&conf.MetaStoreConfig.ForcePathStyle, "s3-force-path-style", false, "S3 force path style")
+	Cmd.Flags().BoolVar(&conf.MetaStoreConfig.GCSInterop, "s3-gcs-interop", false, "Enable Google Cloud Storage support for S3 client")
 
 	// Version file
 	Cmd.Flags().BoolVar(&conf.VersionFileEnabled, "version-file-enabled", false, "Enable version file")

go/pkg/sysdb/metastore/s3/impl.go

@@ -6,9 +6,12 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"net/http"
 	"strings"
+	"time"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
+	v4 "github.com/aws/aws-sdk-go-v2/aws/signer/v4"
 	"github.com/aws/aws-sdk-go-v2/config"
 	"github.com/aws/aws-sdk-go-v2/credentials"
 	"github.com/aws/aws-sdk-go-v2/service/s3"
@@ -20,6 +23,52 @@ import (
 	"google.golang.org/protobuf/proto"
 )
 
+// NOTE(sicheng): As a temporary solution we use the AWS SDK with GCS, but this approach needs a few tweaks:
+// https://stackoverflow.com/questions/73717477/gcp-cloud-storage-golang-aws-sdk2-upload-file-with-s3-interoperability-creds
+//
+// In summary, the AWS SDK we are using (1.36.3) is not fully compatible with the GCS because it uses and additional header
+// for signing. We need to add a middleware to remove the offending header, as suggested by the thread above.
+//
+// If we are upgrading AWS SDK to version higher than 1.73.0, we need additional tweak:
+// cfg.RequestChecksumCalculation = aws.RequestChecksumCalculationWhenRequired
+type RecalculateV4Signature struct {
+	next   http.RoundTripper
+	signer *v4.Signer
+	cfg    aws.Config
+}
+
+// NOTE(sicheng): Code borrowed from https://stackoverflow.com/questions/73717477/gcp-cloud-storage-golang-aws-sdk2-upload-file-with-s3-interoperability-creds
+func (lt *RecalculateV4Signature) RoundTrip(req *http.Request) (*http.Response, error) {
+	// store for later use
+	val := req.Header.Get("Accept-Encoding")
+
+	// delete the header so the header doesn't account for in the signature
+	req.Header.Del("Accept-Encoding")
+
+	// sign with the same date
+	timeString := req.Header.Get("X-Amz-Date")
+	timeDate, err := time.Parse("20060102T150405Z", timeString)
+	if err != nil {
+		return nil, err
+	}
+
+	creds, err := lt.cfg.Credentials.Retrieve(req.Context())
+	if err != nil {
+		return nil, err
+	}
+
+	err = lt.signer.SignHTTP(req.Context(), creds, req, v4.GetPayloadHash(req.Context()), "s3", lt.cfg.Region, timeDate)
+	if err != nil {
+		return nil, err
+	}
+
+	// Reset Accept-Encoding if desired
+	req.Header.Set("Accept-Encoding", val)
+
+	// follows up the original round tripper
+	return lt.next.RoundTrip(req)
+}
+
 // Path to Version Files in S3.
 // Example:
 // s3://<bucket-name>/tenant/<tenant_id>/databases/<database_id>/collections/<collection_id>/versionfiles/file_name
@@ -37,6 +86,7 @@ type S3MetaStoreConfig struct {
 	AccessKeyID             string
 	SecretAccessKey         string
 	ForcePathStyle          bool
+	GCSInterop              bool
 }
 
 type S3MetaStoreInterface interface {
@@ -101,35 +151,35 @@ func NewS3MetaStore(ctx context.Context, cfg S3MetaStoreConfig) (*S3MetaStore, e
 	var awsConfig aws.Config
 	var err error
 
+	awsConfigParts := []func(*config.LoadOptions) error{config.WithRegion(region)}
+
 	if cfg.AccessKeyID != "" && cfg.SecretAccessKey != "" {
 		creds := credentials.NewStaticCredentialsProvider(cfg.AccessKeyID, cfg.SecretAccessKey, "")
-		if cfg.Endpoint != "" {
-			awsConfig, err = config.LoadDefaultConfig(ctx,
-				config.WithCredentialsProvider(creds),
-				config.WithRegion(region),
-			)
-		} else {
-			awsConfig, err = config.LoadDefaultConfig(ctx,
-				config.WithCredentialsProvider(creds),
-				config.WithRegion(region),
-			)
-		}
-	} else {
-		if cfg.Endpoint != "" {
-			awsConfig, err = config.LoadDefaultConfig(ctx,
-				config.WithRegion(region),
-			)
-		} else {
-			awsConfig, err = config.LoadDefaultConfig(ctx,
-				config.WithRegion(region),
-			)
-		}
+		awsConfigParts = append(awsConfigParts, config.WithCredentialsProvider(creds))
 	}
 
+	if cfg.GCSInterop && cfg.Endpoint != "" {
+		resolver := aws.EndpointResolverWithOptionsFunc(func(service, region string, options ...any) (aws.Endpoint, error) {
+			return aws.Endpoint{
+				URL:               cfg.Endpoint,
+				SigningRegion:     cfg.Region,
+				Source:            aws.EndpointSourceCustom,
+				HostnameImmutable: true,
+			}, nil
+		})
+		awsConfigParts = append(awsConfigParts, config.WithEndpointResolverWithOptions(resolver))
+	}
+
+	awsConfig, err = config.LoadDefaultConfig(ctx, awsConfigParts...)
 	if err != nil {
 		return nil, err
 	}
 
+	// Add middleware to remove offending header for signing for GCS Support
+	if cfg.GCSInterop {
+		awsConfig.HTTPClient = &http.Client{Transport: &RecalculateV4Signature{http.DefaultTransport, v4.NewSigner(), awsConfig}}
+	}
+
 	// Create S3 client with optional path-style addressing and custom endpoint
 	otelaws.AppendMiddlewares(&awsConfig.APIOptions)
 	s3Client := s3.NewFromConfig(awsConfig, func(o *s3.Options) {
@@ -159,7 +209,7 @@ func NewS3MetaStore(ctx context.Context, cfg S3MetaStoreConfig) (*S3MetaStore, e
 	}
 
 	// Verify we have access to the bucket
-	_, err = s3Client.HeadBucket(ctx, &s3.HeadBucketInput{
+	_, err = s3Client.ListObjects(ctx, &s3.ListObjectsInput{
 		Bucket: aws.String(bucketName),
 	})
 	if err != nil {