From 86c920fa796a6dee077090dcd641802fdf85d430 Mon Sep 17 00:00:00 2001 From: Oded Viner Date: Mon, 29 Dec 2025 17:21:51 +0200 Subject: [PATCH] object: add support for aws sdk v2 in s3agent createbucket initialize both aws sdk v1 and v2 clients in s3agent. update createbucket to use sdk v2 while keeping all other methods on sdk v1. Signed-off-by: Oded Viner --- go.mod | 7 +- pkg/operator/ceph/object/s3-handlers.go | 69 +++++++++++++++----- pkg/operator/ceph/object/s3-handlers_test.go | 31 ++++++++- tests/scripts/generate-tls-config.sh | 23 +------ 4 files changed, 89 insertions(+), 41 deletions(-) diff --git a/go.mod b/go.mod index 857571a80..8708fbc12 100644 --- a/go.mod +++ b/go.mod @@ -15,8 +15,10 @@ replace ( require ( github.com/IBM/keyprotect-go-client v0.15.1 github.com/aws/aws-sdk-go v1.55.8 + github.com/aws/aws-sdk-go-v2 v1.41.5 github.com/aws/aws-sdk-go-v2/config v1.32.13 github.com/aws/aws-sdk-go-v2/credentials v1.19.13 + github.com/aws/aws-sdk-go-v2/service/s3 v1.95.0 github.com/aws/aws-sdk-go-v2/service/sns v1.39.15 github.com/aws/smithy-go v1.24.2 github.com/banzaicloud/k8s-objectmatcher v1.8.0 @@ -70,13 +72,16 @@ require ( github.com/AzureAD/microsoft-authentication-library-for-go v1.3.2 // indirect github.com/ansel1/merry v1.8.1 // indirect github.com/ansel1/merry/v2 v2.2.2 // indirect - github.com/aws/aws-sdk-go-v2 v1.41.5 // indirect + github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.4 // indirect github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.21 // indirect github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.21 // indirect github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.21 // indirect github.com/aws/aws-sdk-go-v2/internal/ini v1.8.6 // indirect + github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.16 // indirect github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.7 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.7 // indirect github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.21 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.16 // indirect github.com/aws/aws-sdk-go-v2/service/signin v1.0.9 // indirect github.com/aws/aws-sdk-go-v2/service/sso v1.30.14 // indirect github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.18 // indirect diff --git a/pkg/operator/ceph/object/s3-handlers.go b/pkg/operator/ceph/object/s3-handlers.go index 37894ac9c..ea3f6762d 100644 --- a/pkg/operator/ceph/object/s3-handlers.go +++ b/pkg/operator/ceph/object/s3-handlers.go @@ -18,11 +18,17 @@ package object import ( "bytes" + "context" "crypto/tls" "crypto/x509" "net/http" + "net/url" "strings" + v2aws "github.com/aws/aws-sdk-go-v2/aws" + v2creds "github.com/aws/aws-sdk-go-v2/credentials" + s3v2 "github.com/aws/aws-sdk-go-v2/service/s3" + s3types "github.com/aws/aws-sdk-go-v2/service/s3/types" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/aws/awserr" "github.com/aws/aws-sdk-go/aws/credentials" @@ -36,7 +42,8 @@ const CephRegion = "us-east-1" // S3Agent wraps the s3.S3 structure to allow for wrapper methods type S3Agent struct { - Client *s3.S3 + Client *s3.S3 // v1 client + ClientV2 *s3v2.Client // v2 client } func NewS3Agent(accessKey, secretKey, endpoint string, debug bool, tlsCert []byte, insecure bool, httpClient *http.Client) (*S3Agent, error) { @@ -44,6 +51,7 @@ func NewS3Agent(accessKey, secretKey, endpoint string, debug bool, tlsCert []byt if debug { logLevel = aws.LogDebug } + tlsEnabled := false if len(tlsCert) > 0 || insecure { tlsEnabled = true @@ -57,7 +65,10 @@ func NewS3Agent(accessKey, secretKey, endpoint string, debug bool, tlsCert []byt } } - session, err := awssession.NewSession( + // ----------------------------- + // SDK v1 client initialization + // ----------------------------- + v1Session, err := awssession.NewSession( aws.NewConfig(). WithRegion(CephRegion). WithCredentials(credentials.NewStaticCredentials(accessKey, secretKey, "")). @@ -69,11 +80,39 @@ func NewS3Agent(accessKey, secretKey, endpoint string, debug bool, tlsCert []byt WithLogLevel(logLevel), ) if err != nil { - return nil, err + return nil, errors.Wrap(err, "failed to create v1 session") } - svc := s3.New(session) + v1Client := s3.New(v1Session) + + // ----------------------------- + // SDK v2 client initialization + // ----------------------------- + scheme := "http" + if tlsEnabled { + scheme = "https" + } + u, perr := url.Parse(endpoint) + if perr != nil || (u.Scheme != "http" && u.Scheme != "https") { + u, _ = url.Parse(scheme + "://" + endpoint) + } + // Use the officially-supported v2 endpoint customization mechanism. + // This keeps S3's internal endpoint customizations (including hostname handling) + // consistent with the SDK's expectations, which is especially important for TLS. + baseEndpoint := u.String() + v2Cfg := v2aws.Config{ + Region: CephRegion, + Credentials: v2aws.NewCredentialsCache(v2creds.NewStaticCredentialsProvider(accessKey, secretKey, "")), + HTTPClient: httpClient, + BaseEndpoint: &baseEndpoint, + RetryMaxAttempts: 5, + RetryMode: v2aws.RetryModeStandard, + } + v2Client := s3v2.NewFromConfig(v2Cfg, func(o *s3v2.Options) { + o.UsePathStyle = true + }) return &S3Agent{ - Client: svc, + Client: v1Client, + ClientV2: v2Client, }, nil } @@ -93,22 +132,18 @@ func (s *S3Agent) createBucket(name string, infoLogging bool) error { } else { logger.Debugf("creating bucket %q", name) } - bucketInput := &s3.CreateBucketInput{ + + input := &s3v2.CreateBucketInput{ Bucket: &name, } - _, err := s.Client.CreateBucket(bucketInput) + _, err := s.ClientV2.CreateBucket(context.TODO(), input) if err != nil { - if aerr, ok := err.(awserr.Error); ok { - logger.Debugf("DEBUG: after s3 call, ok=%v, aerr=%v", ok, aerr) - switch aerr.Code() { - case s3.ErrCodeBucketAlreadyExists: - logger.Debugf("bucket %q already exists", name) - return nil - case s3.ErrCodeBucketAlreadyOwnedByYou: - logger.Debugf("bucket %q already owned by you", name) - return nil - } + var alreadyExists *s3types.BucketAlreadyExists + var alreadyOwned *s3types.BucketAlreadyOwnedByYou + if errors.As(err, &alreadyExists) || errors.As(err, &alreadyOwned) { + logger.Debugf("bucket %q already exists or is owned by you", name) + return nil } return errors.Wrapf(err, "failed to create bucket %q", name) } diff --git a/pkg/operator/ceph/object/s3-handlers_test.go b/pkg/operator/ceph/object/s3-handlers_test.go index 628dea1e5..aaf47ff26 100644 --- a/pkg/operator/ceph/object/s3-handlers_test.go +++ b/pkg/operator/ceph/object/s3-handlers_test.go @@ -37,6 +37,8 @@ func TestNewS3Agent(t *testing.T) { assert.NotEqual(t, aws.LogDebug, s3Agent.Client.Config.LogLevel) assert.Equal(t, nil, s3Agent.Client.Config.HTTPClient.Transport) assert.True(t, *s3Agent.Client.Config.DisableSSL) + assert.NotNil(t, s3Agent.ClientV2) + assert.Equal(t, "http://endpoint", *s3Agent.ClientV2.Options().BaseEndpoint) }) t.Run("test with debug without tls", func(t *testing.T) { debug := true @@ -47,6 +49,8 @@ func TestNewS3Agent(t *testing.T) { assert.Equal(t, &logLevel, s3Agent.Client.Config.LogLevel) assert.Nil(t, s3Agent.Client.Config.HTTPClient.Transport) assert.True(t, *s3Agent.Client.Config.DisableSSL) + assert.NotNil(t, s3Agent.ClientV2) + assert.Equal(t, "http://endpoint", *s3Agent.ClientV2.Options().BaseEndpoint) }) t.Run("test without tls client cert but insecure tls", func(t *testing.T) { debug := true @@ -56,6 +60,8 @@ func TestNewS3Agent(t *testing.T) { assert.NotNil(t, s3Agent.Client.Config.HTTPClient.Transport.(*http.Transport).TLSClientConfig.RootCAs) // still includes sys certs assert.True(t, s3Agent.Client.Config.HTTPClient.Transport.(*http.Transport).TLSClientConfig.InsecureSkipVerify) assert.False(t, *s3Agent.Client.Config.DisableSSL) + assert.NotNil(t, s3Agent.ClientV2) + assert.Equal(t, "https://endpoint", *s3Agent.ClientV2.Options().BaseEndpoint) }) t.Run("test with secure tls client cert", func(t *testing.T) { debug := true @@ -65,6 +71,8 @@ func TestNewS3Agent(t *testing.T) { assert.NoError(t, err) assert.NotNil(t, s3Agent.Client.Config.HTTPClient.Transport.(*http.Transport).TLSClientConfig.RootCAs) assert.False(t, *s3Agent.Client.Config.DisableSSL) + assert.NotNil(t, s3Agent.ClientV2) + assert.Equal(t, "https://endpoint", *s3Agent.ClientV2.Options().BaseEndpoint) }) t.Run("test with insecure tls client cert", func(t *testing.T) { debug := true @@ -75,6 +83,8 @@ func TestNewS3Agent(t *testing.T) { assert.NotNil(t, s3Agent.Client.Config.HTTPClient.Transport) assert.True(t, s3Agent.Client.Config.HTTPClient.Transport.(*http.Transport).TLSClientConfig.InsecureSkipVerify) assert.False(t, *s3Agent.Client.Config.DisableSSL) + assert.NotNil(t, s3Agent.ClientV2) + assert.Equal(t, "https://endpoint", *s3Agent.ClientV2.Options().BaseEndpoint) }) t.Run("test with custom http.Client", func(t *testing.T) { debug := true @@ -82,7 +92,6 @@ func TestNewS3Agent(t *testing.T) { insecure := false httpClient := &http.Client{ Transport: &http.Transport{ - // set some values to check if they are passed through MaxIdleConns: 7, MaxIdleConnsPerHost: 13, MaxConnsPerHost: 17, @@ -97,5 +106,25 @@ func TestNewS3Agent(t *testing.T) { assert.Equal(t, 7, transport.(*http.Transport).MaxIdleConns) assert.Equal(t, 13, transport.(*http.Transport).MaxIdleConnsPerHost) assert.Equal(t, 17, transport.(*http.Transport).MaxConnsPerHost) + assert.NotNil(t, s3Agent.ClientV2) + assert.Equal(t, "http://endpoint", *s3Agent.ClientV2.Options().BaseEndpoint) + }) + t.Run("v2 endpoint with host:port for TLS", func(t *testing.T) { + ep := "rook-ceph-rgw-store.test-ns.svc:443" + s3Agent, err := NewS3Agent(accessKey, secretKey, ep, false, nil, true, nil) + assert.NoError(t, err) + assert.Equal(t, "https://rook-ceph-rgw-store.test-ns.svc:443", *s3Agent.ClientV2.Options().BaseEndpoint) + }) + t.Run("v2 endpoint with host:port without TLS", func(t *testing.T) { + ep := "rook-ceph-rgw-store.test-ns.svc:80" + s3Agent, err := NewS3Agent(accessKey, secretKey, ep, false, nil, false, nil) + assert.NoError(t, err) + assert.Equal(t, "http://rook-ceph-rgw-store.test-ns.svc:80", *s3Agent.ClientV2.Options().BaseEndpoint) + }) + t.Run("v2 endpoint with full URL", func(t *testing.T) { + ep := "https://rook-ceph-rgw-store.test-ns.svc:443" + s3Agent, err := NewS3Agent(accessKey, secretKey, ep, false, nil, true, nil) + assert.NoError(t, err) + assert.Equal(t, "https://rook-ceph-rgw-store.test-ns.svc:443", *s3Agent.ClientV2.Options().BaseEndpoint) }) } diff --git a/tests/scripts/generate-tls-config.sh b/tests/scripts/generate-tls-config.sh index 3493ac4b8..a3a2c4016 100755 --- a/tests/scripts/generate-tls-config.sh +++ b/tests/scripts/generate-tls-config.sh @@ -33,27 +33,7 @@ openssl req -new -key "${DIR}"/"${SERVICE}".key -subj "/CN=system:node:${SERVICE export CSR_NAME=${SERVICE}-csr -# Minimum 1.19.0 kubernetes version is required for certificates.k8s.io/v1 version -SERVER_VERSION=$(kubectl version --short | awk -F "." '/Server Version/ {print $2}') -MINIMUM_VERSION=19 -if [ "${SERVER_VERSION}" -lt "${MINIMUM_VERSION}" ] -then - cat <"${DIR}"/csr.yaml - apiVersion: certificates.k8s.io/v1beta1 - kind: CertificateSigningRequest - metadata: - name: ${CSR_NAME} - spec: - groups: - - system:authenticated - request: $(cat "${DIR}"/server.csr | base64 | tr -d '\n') - usages: - - digital signature - - key encipherment - - server auth -EOF -else - cat <"${DIR}"/csr.yaml +cat <"${DIR}"/csr.yaml apiVersion: certificates.k8s.io/v1 kind: CertificateSigningRequest metadata: @@ -68,7 +48,6 @@ else - key encipherment - server auth EOF -fi kubectl create -f "${DIR}/"csr.yaml