adapt to new aws code architecture

flow/connectors/clickhouse/clickhouse.go

@@ -113,7 +113,7 @@ func NewClickhouseConnector(
 			AccessKeyID:     config.AccessKeyId,
 			SecretAccessKey: config.SecretAccessKey,
 		},
-		EndpointUrl: nil,
+		EndpointUrl: config.Endpoint,
 		Region:      config.Region,
 	})
 	if err != nil {

flow/connectors/clickhouse/qrep_avro_sync.go

@@ -42,7 +42,9 @@ func (s *ClickhouseAvroSyncMethod) CopyStageToDestination(ctx context.Context, a
 
 	avroFileUrl := fmt.Sprintf("https://%s.s3.%s.amazonaws.com/%s", s3o.Bucket,
 		s.connector.credsProvider.Provider.GetRegion(), avroFile.FilePath)
-	if strings.Contains(s.connector.creds.Endpoint, "storage.googleapis.com") {
+
+	endpoint := s.connector.credsProvider.Provider.GetEndpointURL()
+	if strings.Contains(endpoint, "storage.googleapis.com") {
 		avroFileUrl = fmt.Sprintf("https://storage.googleapis.com/%s/%s", s3o.Bucket, avroFile.FilePath)
 	}
 
@@ -126,7 +128,8 @@ func (s *ClickhouseAvroSyncMethod) SyncQRepRecords(
 
 	avroFileUrl := fmt.Sprintf("https://%s.s3.%s.amazonaws.com/%s", s3o.Bucket,
 		s.connector.credsProvider.Provider.GetRegion(), avroFile.FilePath)
-	if strings.Contains(s.connector.creds.Endpoint, "storage.googleapis.com") {
+	endpoint := s.connector.credsProvider.Provider.GetEndpointURL()
+	if strings.Contains(endpoint, "storage.googleapis.com") {
 		avroFileUrl = fmt.Sprintf("https://storage.googleapis.com/%s/%s", s3o.Bucket, avroFile.FilePath)
 	}
 

flow/connectors/utils/aws.go

@@ -57,6 +57,7 @@ type AWSCredentialsProvider interface {
 	Retrieve(ctx context.Context) (AWSCredentials, error)
 	GetUnderlyingProvider() aws.CredentialsProvider
 	GetRegion() string
+	GetEndpointURL() string
 }
 
 type ConfigBasedAWSCredentialsProvider struct {
@@ -71,6 +72,15 @@ func (r *ConfigBasedAWSCredentialsProvider) GetRegion() string {
 	return r.config.Region
 }
 
+func (r *ConfigBasedAWSCredentialsProvider) GetEndpointURL() string {
+	endpoint := ""
+	if r.config.BaseEndpoint != nil {
+		endpoint = *r.config.BaseEndpoint
+	}
+
+	return endpoint
+}
+
 // Retrieve should be called as late as possible in order to have credentials with latest expiry
 func (r *ConfigBasedAWSCredentialsProvider) Retrieve(ctx context.Context) (AWSCredentials, error) {
 	retrieved, err := r.config.Credentials.Retrieve(ctx)
@@ -105,6 +115,15 @@ func (s *StaticAWSCredentialsProvider) Retrieve(ctx context.Context) (AWSCredent
 	return s.credentials, nil
 }
 
+func (s *StaticAWSCredentialsProvider) GetEndpointURL() string {
+	endpoint := ""
+	if s.credentials.EndpointUrl != nil {
+		endpoint = *s.credentials.EndpointUrl
+	}
+
+	return endpoint
+}
+
 func NewStaticAWSCredentialsProvider(credentials AWSCredentials, region string) AWSCredentialsProvider {
 	return &StaticAWSCredentialsProvider{
 		credentials: credentials,
@@ -206,17 +225,14 @@ func CreateS3Client(ctx context.Context, credsProvider AWSCredentialsProvider) (
 		options.Region = credsProvider.GetRegion()
 		options.Credentials = credsProvider.GetUnderlyingProvider()
 		if awsCredentials.EndpointUrl != nil {
-			options.BaseEndpoint = awsCredentials.EndpointUrl
-			if strings.Contains(*awsCredentials.EndpointUrl, "storage.googleapis.com") {
-				// Assign custom client with our own transport
-				options.HTTPClient = &http.Client{
-					Transport: &RecalculateV4Signature{
-						next:        http.DefaultTransport,
-						signer:      v4.NewSigner(),
-						credentials: credsProvider.GetUnderlyingProvider(),
-						region:      credsProvider.GetRegion(),
-					},
-				}
+			// Assign custom client with our own transport
+			options.HTTPClient = &http.Client{
+				Transport: &RecalculateV4Signature{
+					next:        http.DefaultTransport,
+					signer:      v4.NewSigner(),
+					credentials: credsProvider.GetUnderlyingProvider(),
+					region:      credsProvider.GetRegion(),
+				},
 			}
 		}
 	})