diff --git a/pkg/objstore/s3store/BUILD.bazel b/pkg/objstore/s3store/BUILD.bazel index 3921b110e5ba0..3cc45d1a7f753 100644 --- a/pkg/objstore/s3store/BUILD.bazel +++ b/pkg/objstore/s3store/BUILD.bazel @@ -4,6 +4,8 @@ go_library( name = "s3store", srcs = [ "client.go", + "gcs_s3_fault.go", + "gcs_s3_signer.go", "interface.go", "ks3.go", "logger.go", @@ -28,6 +30,7 @@ go_library( "@com_github_aws_aws_sdk_go_v2//aws", "@com_github_aws_aws_sdk_go_v2//aws/ratelimit", "@com_github_aws_aws_sdk_go_v2//aws/retry", + "@com_github_aws_aws_sdk_go_v2//aws/signer/v4:signer", "@com_github_aws_aws_sdk_go_v2_config//:config", "@com_github_aws_aws_sdk_go_v2_credentials//:credentials", "@com_github_aws_aws_sdk_go_v2_credentials//stscreds", @@ -57,6 +60,7 @@ go_test( timeout = "short", srcs = [ "client_test.go", + "gcs_test.go", "main_test.go", "retry_test.go", "s3_flags_test.go", diff --git a/pkg/objstore/s3store/gcs_s3_fault.go b/pkg/objstore/s3store/gcs_s3_fault.go new file mode 100644 index 0000000000000..89cf3b4c886bd --- /dev/null +++ b/pkg/objstore/s3store/gcs_s3_fault.go @@ -0,0 +1,174 @@ +// Copyright 2026 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package s3store + +import ( + "errors" + "io" + "math/rand" + "net/http" + "os" + "strconv" + "strings" + "sync" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + "go.uber.org/zap" +) + +const ( + gcsS3FaultRateEnv = "TIDB_GCS_S3_COMPAT_FAULT_RATE" + gcsS3FaultModeEnv = "TIDB_GCS_S3_COMPAT_FAULT_MODE" + gcsS3FaultMaxEnv = "TIDB_GCS_S3_COMPAT_FAULT_MAX" + + // This branch enables low-rate GCS S3-compatible request faults by default + // for real-cluster E2E retry validation. Set gcsS3FaultRateEnv to 0 to disable. + defaultGCSS3FaultRate = 0.02 + + gcsS3FaultModeHTTP500 = "http500" + gcsS3FaultModeConnectionReset = "connection-reset" + gcsS3FaultModeMixed = "mixed" +) + +type gcsS3FaultInjectingHTTPClient struct { + base aws.HTTPClient + rate float64 + mode string + max int64 + count int64 + rand *rand.Rand + logger *zap.Logger + mu sync.Mutex +} + +func newGCSS3FaultInjectingHTTPClient(base aws.HTTPClient, logger *zap.Logger) aws.HTTPClient { + rate := defaultGCSS3FaultRate + if rateValue, ok := os.LookupEnv(gcsS3FaultRateEnv); ok { + rateValue = strings.TrimSpace(rateValue) + parsedRate, err := strconv.ParseFloat(rateValue, 64) + if err != nil || parsedRate < 0 || parsedRate > 1 { + logger.Warn( + "ignore invalid GCS S3-compatible fault injection rate", + zap.String("env", gcsS3FaultRateEnv), + zap.String("value", rateValue), + ) + return nil + } + rate = parsedRate + } + if rate == 0 { + return nil + } + + mode := strings.TrimSpace(os.Getenv(gcsS3FaultModeEnv)) + if mode == "" { + mode = gcsS3FaultModeMixed + } + switch mode { + case gcsS3FaultModeHTTP500, gcsS3FaultModeConnectionReset, gcsS3FaultModeMixed: + default: + logger.Warn( + "ignore invalid GCS S3-compatible fault injection mode", + zap.String("env", gcsS3FaultModeEnv), + zap.String("value", mode), + ) + return nil + } + + var maxFaults int64 + maxValue := strings.TrimSpace(os.Getenv(gcsS3FaultMaxEnv)) + if maxValue != "" { + parsedMaxFaults, err := strconv.ParseInt(maxValue, 10, 64) + if err != nil || parsedMaxFaults <= 0 { + logger.Warn( + "ignore invalid GCS S3-compatible fault injection max count", + zap.String("env", gcsS3FaultMaxEnv), + zap.String("value", maxValue), + ) + maxFaults = 0 + } else { + maxFaults = parsedMaxFaults + } + } + + logger.Warn( + "enable GCS S3-compatible fault injection", + zap.String("rateEnv", gcsS3FaultRateEnv), + zap.Float64("rate", rate), + zap.String("modeEnv", gcsS3FaultModeEnv), + zap.String("mode", mode), + zap.String("maxEnv", gcsS3FaultMaxEnv), + zap.Int64("max", maxFaults), + ) + return &gcsS3FaultInjectingHTTPClient{ + base: base, + rate: rate, + mode: mode, + max: maxFaults, + rand: rand.New(rand.NewSource(time.Now().UnixNano())), + logger: logger, + } +} + +func (c *gcsS3FaultInjectingHTTPClient) Do(req *http.Request) (*http.Response, error) { + if inject, mode := c.pickFaultMode(); inject { + c.logger.Warn( + "inject GCS S3-compatible request fault", + zap.String("mode", mode), + zap.String("method", req.Method), + zap.String("url", req.URL.Redacted()), + ) + switch mode { + case gcsS3FaultModeConnectionReset: + return nil, errors.New("read tcp 127.0.0.1:12345->127.0.0.1:443: read: connection reset by peer") + default: + return &http.Response{ + StatusCode: http.StatusInternalServerError, + Status: "500 " + http.StatusText(http.StatusInternalServerError), + Header: http.Header{ + "Content-Type": []string{"application/xml"}, + }, + Body: io.NopCloser(strings.NewReader("InternalErrorinjected fault")), + Request: req, + }, nil + } + } + + if c.base != nil { + return c.base.Do(req) + } + return http.DefaultClient.Do(req) +} + +func (c *gcsS3FaultInjectingHTTPClient) pickFaultMode() (bool, string) { + c.mu.Lock() + defer c.mu.Unlock() + + if c.max > 0 && c.count >= c.max { + return false, "" + } + if c.rand.Float64() >= c.rate { + return false, "" + } + c.count++ + if c.mode != gcsS3FaultModeMixed { + return true, c.mode + } + if c.rand.Intn(2) == 0 { + return true, gcsS3FaultModeHTTP500 + } + return true, gcsS3FaultModeConnectionReset +} diff --git a/pkg/objstore/s3store/gcs_s3_signer.go b/pkg/objstore/s3store/gcs_s3_signer.go new file mode 100644 index 0000000000000..9db875ec64d0b --- /dev/null +++ b/pkg/objstore/s3store/gcs_s3_signer.go @@ -0,0 +1,58 @@ +// Copyright 2026 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package s3store + +import ( + "context" + "net/http" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/aws/signer/v4" +) + +type gcsS3CompatibleSigner struct { + signer *v4.Signer +} + +func newGCSS3CompatibleSigner() *gcsS3CompatibleSigner { + return &gcsS3CompatibleSigner{signer: v4.NewSigner()} +} + +func (s *gcsS3CompatibleSigner) SignHTTP( + ctx context.Context, + credentials aws.Credentials, + r *http.Request, + payloadHash string, + service string, + region string, + signingTime time.Time, + optFns ...func(*v4.SignerOptions), +) error { + // GCS S3 interoperability rejects signatures that include Accept-Encoding. + // Keep sending the header, but exclude it from the canonical request. + savedHeaders := http.Header{} + for _, key := range []string{"Accept-Encoding"} { + if values, ok := r.Header[key]; ok { + savedHeaders[key] = append([]string(nil), values...) + r.Header.Del(key) + } + } + err := s.signer.SignHTTP(ctx, credentials, r, payloadHash, service, region, signingTime, optFns...) + for key, values := range savedHeaders { + r.Header[key] = values + } + return err +} diff --git a/pkg/objstore/s3store/gcs_test.go b/pkg/objstore/s3store/gcs_test.go new file mode 100644 index 0000000000000..06a3c2158feb2 --- /dev/null +++ b/pkg/objstore/s3store/gcs_test.go @@ -0,0 +1,241 @@ +// Copyright 2026 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package s3store + +import ( + "context" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + + backuppb "github.com/pingcap/kvproto/pkg/brpb" + "github.com/pingcap/tidb/pkg/objstore/storeapi" + "github.com/stretchr/testify/require" +) + +func TestIsGCSS3Compatible(t *testing.T) { + require.True(t, isGCSS3Compatible(&backuppb.S3{ + Provider: "gcs", + Endpoint: "http://127.0.0.1:9000", + })) + require.True(t, isGCSS3Compatible(&backuppb.S3{ + Provider: "ceph", + Endpoint: "https://storage.googleapis.com", + })) + require.True(t, isGCSS3Compatible(&backuppb.S3{ + Endpoint: "https://storage.googleapis.com/", + })) + require.False(t, isGCSS3Compatible(&backuppb.S3{ + Provider: "ceph", + Endpoint: "https://s3.example.com", + })) + require.False(t, isGCSS3Compatible(&backuppb.S3{ + Endpoint: "://bad-endpoint", + })) +} + +func TestGCSS3CompatibleSignerSkipsAcceptEncodingWithRetry(t *testing.T) { + t.Setenv(gcsS3FaultRateEnv, "0") + + const listObjectsV2Response = ` + + bucket + + 0 + 1 + false +` + + type requestInfo struct { + method string + signedHeaders string + listType string + statusCode int + writeErr error + } + + var ( + mu sync.Mutex + getAttempts int + requests []requestInfo + ) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + info := requestInfo{ + method: r.Method, + signedHeaders: getSignedHeaders(r.Header.Get("Authorization")), + listType: r.URL.Query().Get("list-type"), + } + defer func() { + mu.Lock() + requests = append(requests, info) + mu.Unlock() + }() + + switch r.Method { + case http.MethodHead: + info.statusCode = http.StatusOK + w.WriteHeader(http.StatusOK) + case http.MethodGet: + mu.Lock() + getAttempts++ + getAttempt := getAttempts + mu.Unlock() + + if getAttempt == 1 { + info.statusCode = http.StatusInternalServerError + w.WriteHeader(http.StatusInternalServerError) + return + } + + info.statusCode = http.StatusOK + w.Header().Set("Content-Type", "application/xml") + _, info.writeErr = w.Write([]byte(listObjectsV2Response)) + default: + info.statusCode = http.StatusNotFound + w.WriteHeader(http.StatusNotFound) + } + })) + defer server.Close() + + storage, err := NewS3Storage(context.Background(), &backuppb.S3{ + Bucket: "bucket", + Endpoint: server.URL, + Provider: "gcs", + ForcePathStyle: true, + AccessKey: "access-key", + SecretAccessKey: "secret-access-key", + }, &storeapi.Options{ + CheckPermissions: []storeapi.Permission{storeapi.AccessBuckets, storeapi.ListObjects}, + }) + require.NoError(t, err) + require.NotNil(t, storage) + + mu.Lock() + observedRequests := append([]requestInfo(nil), requests...) + mu.Unlock() + require.Len(t, observedRequests, 3) + + var headSeen bool + var listStatuses []int + for _, req := range observedRequests { + require.NoError(t, req.writeErr) + require.NotEmpty(t, req.signedHeaders) + require.NotContains(t, req.signedHeaders, "accept-encoding") + require.Contains(t, req.signedHeaders, "amz-sdk-invocation-id") + require.Contains(t, req.signedHeaders, "amz-sdk-request") + require.Contains(t, req.signedHeaders, "host") + require.Contains(t, req.signedHeaders, "x-amz-content-sha256") + require.Contains(t, req.signedHeaders, "x-amz-date") + + switch req.method { + case http.MethodHead: + headSeen = true + require.Equal(t, http.StatusOK, req.statusCode) + case http.MethodGet: + require.Equal(t, "2", req.listType) + listStatuses = append(listStatuses, req.statusCode) + default: + require.Failf(t, "unexpected request method", "method: %s", req.method) + } + } + require.True(t, headSeen) + require.Equal(t, []int{http.StatusInternalServerError, http.StatusOK}, listStatuses) +} + +func TestGCSS3CompatibleFaultInjectionRetries(t *testing.T) { + const listObjectsV2Response = ` + + bucket + + 0 + 1 + false +` + + for _, mode := range []string{gcsS3FaultModeHTTP500, gcsS3FaultModeConnectionReset} { + t.Run(mode, func(t *testing.T) { + t.Setenv(gcsS3FaultRateEnv, "1") + t.Setenv(gcsS3FaultModeEnv, mode) + t.Setenv(gcsS3FaultMaxEnv, "1") + + type faultRequestInfo struct { + method string + listType string + writeErr error + } + + var ( + mu sync.Mutex + requests []faultRequestInfo + ) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + info := faultRequestInfo{ + method: r.Method, + listType: r.URL.Query().Get("list-type"), + } + defer func() { + mu.Lock() + requests = append(requests, info) + mu.Unlock() + }() + + switch r.Method { + case http.MethodHead: + w.WriteHeader(http.StatusOK) + case http.MethodGet: + w.Header().Set("Content-Type", "application/xml") + _, info.writeErr = w.Write([]byte(listObjectsV2Response)) + default: + w.WriteHeader(http.StatusNotFound) + } + })) + defer server.Close() + + storage, err := NewS3Storage(context.Background(), &backuppb.S3{ + Bucket: "bucket", + Endpoint: server.URL, + Provider: "gcs", + ForcePathStyle: true, + AccessKey: "access-key", + SecretAccessKey: "secret-access-key", + }, &storeapi.Options{ + CheckPermissions: []storeapi.Permission{storeapi.AccessBuckets, storeapi.ListObjects}, + }) + require.NoError(t, err) + require.NotNil(t, storage) + + mu.Lock() + observedRequests := append([]faultRequestInfo(nil), requests...) + mu.Unlock() + require.Len(t, observedRequests, 2) + require.Equal(t, http.MethodHead, observedRequests[0].method) + require.Equal(t, http.MethodGet, observedRequests[1].method) + require.Equal(t, "2", observedRequests[1].listType) + require.NoError(t, observedRequests[1].writeErr) + }) + } +} + +func getSignedHeaders(authorization string) string { + for _, part := range strings.Split(authorization, ",") { + part = strings.TrimSpace(part) + if strings.HasPrefix(part, "SignedHeaders=") { + return strings.TrimPrefix(part, "SignedHeaders=") + } + } + return "" +} diff --git a/pkg/objstore/s3store/store.go b/pkg/objstore/s3store/store.go index 9e66b12ea49e0..892cc9167eca7 100644 --- a/pkg/objstore/s3store/store.go +++ b/pkg/objstore/s3store/store.go @@ -17,6 +17,7 @@ package s3store import ( "context" "fmt" + "net/url" "strings" alicred "github.com/aliyun/alibaba-cloud-sdk-go/sdk/auth/credentials" @@ -43,6 +44,12 @@ import ( const ( defaultRegion = "us-east-1" + gcsProvider = "gcs" + // GCS S3 interoperability documents storage.googleapis.com as the XML API + // endpoint for S3-compatible tools. + // See https://cloud.google.com/storage/docs/interoperability and + // https://cloud.google.com/storage/docs/request-endpoints. + gcsEndpoint = "storage.googleapis.com" // to check the cloud type by endpoint tag. domainAliyun = "aliyuncs.com" ) @@ -50,6 +57,7 @@ const ( // NewS3Storage initialize a new s3 storage for metadata. func NewS3Storage(ctx context.Context, backend *backuppb.S3, opts *storeapi.Options) (obj *s3like.Storage, errRet error) { qs := *backend + gcsS3Compatible := isGCSS3Compatible(&qs) // Start with default configuration loading var configOpts []func(*config.LoadOptions) error @@ -108,6 +116,13 @@ func NewS3Storage(ctx context.Context, backend *backuppb.S3, opts *storeapi.Opti if opts.HTTPClient != nil { cfg.HTTPClient = opts.HTTPClient } + var gcsS3FaultHTTPClient aws.HTTPClient + if gcsS3Compatible { + gcsS3FaultHTTPClient = newGCSS3FaultInjectingHTTPClient(cfg.HTTPClient, logger) + if gcsS3FaultHTTPClient != nil { + cfg.HTTPClient = gcsS3FaultHTTPClient + } + } // Configure S3-specific options var s3Opts []func(*s3.Options) @@ -125,6 +140,11 @@ func NewS3Storage(ctx context.Context, backend *backuppb.S3, opts *storeapi.Opti // These logs will be printed when log level is `DEBUG`. o.ClientLogMode |= aws.LogRetries | aws.LogRequest | aws.LogResponse | aws.LogDeprecatedUsage }) + if gcsS3Compatible { + s3Opts = append(s3Opts, func(o *s3.Options) { + o.HTTPSignerV4 = newGCSS3CompatibleSigner() + }) + } // ⚠️ Do NOT set a global endpoint in the AWS config. // Setting a global endpoint will break AssumeRoleWithWebIdentity, @@ -140,6 +160,11 @@ func NewS3Storage(ctx context.Context, backend *backuppb.S3, opts *storeapi.Opti o.HTTPClient = opts.HTTPClient }) } + if gcsS3FaultHTTPClient != nil { + s3Opts = append(s3Opts, func(o *s3.Options) { + o.HTTPClient = gcsS3FaultHTTPClient + }) + } // When using a profile, let AWS SDK handle credentials through the profile // Don't call autoNewCred as it interferes with profile-based authentication if qs.Profile == "" { @@ -227,7 +252,7 @@ func NewS3Storage(ctx context.Context, backend *backuppb.S3, opts *storeapi.Opti // Perform region detection and validation var detectedRegion string - officialS3 := len(qs.Provider) == 0 || qs.Provider == "aws" + officialS3 := !gcsS3Compatible && (len(qs.Provider) == 0 || qs.Provider == "aws") if officialS3 { // For AWS provider, detect the actual bucket region // In AWS SDK v2, GetBucketRegion has a simpler signature @@ -300,6 +325,20 @@ func NewS3Storage(ctx context.Context, backend *backuppb.S3, opts *storeapi.Opti return s3Storage, nil } +func isGCSS3Compatible(qs *backuppb.S3) bool { + if strings.EqualFold(qs.Provider, gcsProvider) { + return true + } + if qs.Endpoint == "" { + return false + } + u, err := url.Parse(qs.Endpoint) + if err != nil { + return false + } + return strings.EqualFold(u.Hostname(), gcsEndpoint) +} + // IsObjectLockEnabled checks whether the S3 bucket has Object Lock enabled. func IsObjectLockEnabled(svc S3API, options *backuppb.S3) bool { input := &s3.GetObjectLockConfigurationInput{