diff --git a/.github/workflows/entrypoint.yaml b/.github/workflows/entrypoint.yaml index 64deb6f9d0876..b8a7e63cca164 100644 --- a/.github/workflows/entrypoint.yaml +++ b/.github/workflows/entrypoint.yaml @@ -2,13 +2,16 @@ name: MatrixOne ALL CI on: pull_request_target: - types: [ opened, synchronize, reopened ] - branches: [ main,'[0-9]+.[0-9]+*' ] + types: [ opened, synchronize, reopened, ready_for_review ] + branches: [ main, 4.2-dev, '[0-9]+.[0-9]+*' ] concurrency: group: ${{ github.event.pull_request.head.repo.full_name}}/${{ github.event.pull_request.head.ref }}/${{ github.workflow }} cancel-in-progress: true +permissions: + contents: read + jobs: check-pr-valid: runs-on: ubuntu-latest @@ -23,6 +26,35 @@ jobs: title_for_find_issue: "Which issue(s) this PR fixes:" title_for_find_content: "What this PR does / why we need it:" + bvt-group-plan: + name: Plan complementary BVT groups + needs: check-pr-valid + if: ${{ needs.check-pr-valid.outputs.pr_valid == 'true' && github.base_ref != '3.0-dev' }} + runs-on: ubuntu-latest + outputs: + compose_group: ${{ steps.plan.outputs.compose_group }} + launch_group: ${{ steps.plan.outputs.launch_group }} + generation: ${{ steps.plan.outputs.generation }} + steps: + - id: plan + name: Assign both BVT groups + run: | + set -euo pipefail + compose_group=$(( (${{ github.run_id }} + ${{ github.run_attempt }}) % 2 )) + launch_group=$((1 - compose_group)) + generation='${{ github.run_id }}-${{ github.run_attempt }}' + { + echo "compose_group=${compose_group}" + echo "launch_group=${launch_group}" + echo "generation=${generation}" + } >> "$GITHUB_OUTPUT" + { + echo '### BVT group plan' + echo "- Compose + Proxy: group ${compose_group}" + echo "- Launch + Pessimistic: group ${launch_group}" + echo "- Generation: ${generation}" + } >> "$GITHUB_STEP_SUMMARY" + # ============ 非 3.0-dev 分支使用 @main ============ matrixone-ci: name: Matrixone CI @@ -31,6 +63,15 @@ jobs: uses: matrixorigin/CI/.github/workflows/ci.yaml@main secrets: inherit + matrixone-ut-coverage: + name: Matrixone UT Coverage + needs: [check-pr-valid, bvt-group-plan] + if: ${{ needs.check-pr-valid.outputs.pr_valid == 'true' && github.base_ref != '3.0-dev' }} + uses: matrixorigin/CI/.github/workflows/coverage-ut.yaml@main + with: + bvt_generation: ${{ needs.bvt-group-plan.outputs.generation }} + secrets: inherit + matrixone-upgrade-ci: name: Matrixone Upgrade CI needs: check-pr-valid @@ -40,23 +81,39 @@ jobs: matrixone-compose-ci: name: Matrixone Compose CI - needs: check-pr-valid + needs: [check-pr-valid, bvt-group-plan] if: ${{ needs.check-pr-valid.outputs.pr_valid == 'true' && github.base_ref != '3.0-dev' }} - uses: matrixorigin/CI/.github/workflows/e2e-compose.yaml@main + uses: matrixorigin/CI/.github/workflows/e2e-compose-parallel.yaml@main + with: + bvt_group: ${{ needs.bvt-group-plan.outputs.compose_group }} + bvt_generation: ${{ needs.bvt-group-plan.outputs.generation }} secrets: inherit matrixone-standalone-ci: name: Matrixone Standlone CI - needs: check-pr-valid + needs: [check-pr-valid, bvt-group-plan] if: ${{ needs.check-pr-valid.outputs.pr_valid == 'true' && github.base_ref != '3.0-dev' }} - uses: matrixorigin/CI/.github/workflows/e2e-standalone.yaml@main + uses: matrixorigin/CI/.github/workflows/e2e-standalone-parallel.yaml@main + with: + bvt_group: ${{ needs.bvt-group-plan.outputs.launch_group }} + bvt_generation: ${{ needs.bvt-group-plan.outputs.generation }} secrets: inherit matrixone-utils-ci: name: Matrixone Utils CI needs: check-pr-valid if: ${{ needs.check-pr-valid.outputs.pr_valid == 'true' && github.base_ref != '3.0-dev' }} - uses: matrixorigin/CI/.github/workflows/utils.yaml@main + uses: matrixorigin/CI/.github/workflows/utils-parallel.yaml@main + secrets: inherit + + matrixone-coverage-merge: + name: Matrixone Utils CI + needs: [check-pr-valid, bvt-group-plan, matrixone-ut-coverage, matrixone-compose-ci, matrixone-standalone-ci] + if: ${{ always() && needs.check-pr-valid.outputs.pr_valid == 'true' && github.base_ref != '3.0-dev' }} + uses: matrixorigin/CI/.github/workflows/coverage-merge.yaml@main + with: + prerequisites_ready: ${{ needs.bvt-group-plan.result == 'success' && needs.matrixone-ut-coverage.result == 'success' && needs.matrixone-ut-coverage.outputs.coverage_ready == 'true' && needs.matrixone-compose-ci.result == 'success' && needs.matrixone-standalone-ci.result == 'success' }} + expected_bvt_generation: ${{ needs.bvt-group-plan.outputs.generation }} secrets: inherit # ============ 3.0-dev 分支使用 @release/3.0-dev ============ diff --git a/.github/workflows/iceberg-connector.yml b/.github/workflows/iceberg-connector.yml index 9e16921c76b3f..03c49da8ae4b9 100644 --- a/.github/workflows/iceberg-connector.yml +++ b/.github/workflows/iceberg-connector.yml @@ -2,7 +2,9 @@ name: Iceberg Connector E2E on: pull_request: + types: [opened, synchronize, reopened, ready_for_review] paths: + - '.github/workflows/entrypoint.yaml' - '.github/workflows/iceberg-connector.yml' - 'Makefile' - 'optools/iceberg_ci.bash' diff --git a/etc/launch-tae-compose/README.md b/etc/launch-tae-compose/README.md index 0283622461dbc..033281e22c2df 100644 --- a/etc/launch-tae-compose/README.md +++ b/etc/launch-tae-compose/README.md @@ -33,6 +33,9 @@ docker compose -f etc/launch-tae-compose/compose.yaml --profile launch-multi-cn # cn-1 docker compose -f etc/launch-tae-compose/compose.yaml --profile launch-multi-cn logs cn-1 +# proxy +docker compose -f etc/launch-tae-compose/compose.yaml --profile launch-multi-cn logs proxy + # tn docker compose -f etc/launch-tae-compose/compose.yaml --profile launch-multi-cn logs tn @@ -47,17 +50,15 @@ docker compose -f etc/launch-tae-compose/compose.yaml --profile launch-multi-cn username: minio password: minio123 -## connect service +## connect service through proxy -cn-0 ```shell mysql -h 127.0.0.1 -P 6001 -udump -p111 ``` -cn-1: -```shell -mysql -h 127.0.0.1 -P 7001 -udump -p111 -``` +Port `6001` is the Proxy entrypoint. CN SQL ports are intentionally not +published to the host; clients should enter through Proxy so that both CNs are +eligible backends. ## down diff --git a/etc/launch-tae-compose/compose.yaml b/etc/launch-tae-compose/compose.yaml index 68ad29147bd1b..fdd6188201bbd 100644 --- a/etc/launch-tae-compose/compose.yaml +++ b/etc/launch-tae-compose/compose.yaml @@ -10,11 +10,13 @@ x-mo-common: &mo-common # Typecheck: enable type checking for ToSliceNoTypeCheck and ToSliceNoTypeCheck2 # Default: 0 (disabled), set to 1 to enable TYPECHECK: ${TYPECHECK:-0} + GOBUILD_OPT: ${GOBUILD_OPT:-} image: matrixorigin/matrixone:latest volumes: - ../../etc/launch-tae-compose/config:/config - ../../test:/test - ../../docker-compose-log:/log + - ../../coverage:/coverage restart: on-failure tty: true @@ -23,20 +25,22 @@ services: cn-0: container_name: cn-0 <<: *mo-common - entrypoint: ["/bin/bash","-c","set -euo pipefail; /mo-service -debug-http :12345 -cfg /config/cn-0.toml | tee -a /log/cn-0.log"] + # exec makes mo-service PID 1, so docker compose down delivers SIGTERM to + # the instrumented process and lets it flush GOCOVERDIR before shutdown. + entrypoint: ["/bin/bash","-c","set -euo pipefail; exec /mo-service -debug-http :12345 -cfg /config/cn-0.toml > >(tee -a /log/cn-0.log) 2>&1"] profiles: - launch - launch-multi-cn depends_on: - tn ports: - - "6001:6001" - "6002" - "12345:12345" environment: - AWS_ACCESS_KEY_ID=minio - AWS_SECRET_ACCESS_KEY=minio123 - mo_reuse_enable_checker=true + - GOCOVERDIR=/coverage networks: monet: aliases: @@ -45,11 +49,10 @@ services: cn-1: container_name: cn-1 <<: *mo-common - entrypoint: ["/bin/bash","-c","set -euo pipefail; /mo-service -debug-http :12345 -cfg /config/cn-1.toml | tee -a /log/cn-1.log"] + entrypoint: ["/bin/bash","-c","set -euo pipefail; exec /mo-service -debug-http :12345 -cfg /config/cn-1.toml > >(tee -a /log/cn-1.log) 2>&1"] depends_on: - tn ports: - - "7001:6001" - "6002" - "22345:12345" profiles: @@ -59,15 +62,40 @@ services: - AWS_ACCESS_KEY_ID=minio - AWS_SECRET_ACCESS_KEY=minio123 - mo_reuse_enable_checker=true + - GOCOVERDIR=/coverage networks: monet: aliases: - "cn1" + proxy: + container_name: proxy + <<: *mo-common + entrypoint: ["/bin/bash","-c","set -euo pipefail; exec /mo-service -debug-http :12345 -cfg /config/proxy.toml > >(tee -a /log/proxy.log) 2>&1"] + profiles: + - launch + - launch-multi-cn + depends_on: + - logservice + - cn-0 + - cn-1 + ports: + - "6001:6009" + - "32346:12345" + environment: + - AWS_ACCESS_KEY_ID=minio + - AWS_SECRET_ACCESS_KEY=minio123 + - mo_reuse_enable_checker=true + - GOCOVERDIR=/coverage + networks: + monet: + aliases: + - "proxy" + tn: container_name: tn <<: *mo-common - entrypoint: ["/bin/bash","-c","set -euo pipefail; /mo-service -debug-http :12345 -cfg /config/tn.toml | tee -a /log/tn.log"] + entrypoint: ["/bin/bash","-c","set -euo pipefail; exec /mo-service -debug-http :12345 -cfg /config/tn.toml > >(tee -a /log/tn.log) 2>&1"] profiles: - launch - launch-multi-cn @@ -80,6 +108,7 @@ services: - AWS_ACCESS_KEY_ID=minio - AWS_SECRET_ACCESS_KEY=minio123 - mo_reuse_enable_checker=true + - GOCOVERDIR=/coverage networks: monet: aliases: @@ -91,7 +120,7 @@ services: profiles: - launch - launch-multi-cn - entrypoint: ["/bin/bash","-c","set -euo pipefail; /mo-service -debug-http :12345 -cfg /config/log.toml | tee -a /log/logservice.log"] + entrypoint: ["/bin/bash","-c","set -euo pipefail; exec /mo-service -debug-http :12345 -cfg /config/log.toml > >(tee -a /log/logservice.log) 2>&1"] depends_on: - createbuckets ports: @@ -103,6 +132,7 @@ services: - AWS_ACCESS_KEY_ID=minio - AWS_SECRET_ACCESS_KEY=minio123 - mo_reuse_enable_checker=true + - GOCOVERDIR=/coverage networks: monet: aliases: diff --git a/etc/launch-tae-compose/config/cn-0.toml b/etc/launch-tae-compose/config/cn-0.toml index 582e86d0bf512..14990afe9902d 100644 --- a/etc/launch-tae-compose/config/cn-0.toml +++ b/etc/launch-tae-compose/config/cn-0.toml @@ -54,6 +54,7 @@ type = "distributed-tae" [cn.frontend] port = 6001 +proxy-enabled = true [cn.frontend.iceberg] enable = true diff --git a/etc/launch-tae-compose/config/cn-1.toml b/etc/launch-tae-compose/config/cn-1.toml index 134f1decdea0c..aa7823f65fe75 100644 --- a/etc/launch-tae-compose/config/cn-1.toml +++ b/etc/launch-tae-compose/config/cn-1.toml @@ -54,6 +54,7 @@ type = "distributed-tae" [cn.frontend] port = 6001 +proxy-enabled = true [cn.frontend.iceberg] enable = true diff --git a/etc/launch-tae-compose/config/proxy.toml b/etc/launch-tae-compose/config/proxy.toml new file mode 100644 index 0000000000000..92aa38a590527 --- /dev/null +++ b/etc/launch-tae-compose/config/proxy.toml @@ -0,0 +1,43 @@ +service-type = "PROXY" +data-dir = "/var/lib/matrixone/data" + +[log] +level = "info" +format = "console" +max-size = 512 + +[hakeeper-client] +service-addresses = ["logservice:32001"] + +[[fileservice]] +backend = "DISK-V2" +data-dir = "/var/lib/matrixone/data" +name = "LOCAL" + +[[fileservice]] +backend = "MINIO" +name = "SHARED" + +[fileservice.s3] +bucket = "mo-test" +endpoint = "http://minio:9000" +key-prefix = "server/data" +cert-files = ["/etc/ssl/cert.pem"] + +[fileservice.cache] +memory-capacity = "512MB" +disk-capacity = "1GB" +disk-path = "mo-data/file-service-cache" + +[[fileservice]] +backend = "MINIO" +name = "ETL" + +[fileservice.s3] +bucket = "mo-test" +endpoint = "http://minio:9000" +key-prefix = "server/etl" + +[proxy] +uuid = "rd1dccb4-4d3c-41f8-b482-5251dc7a41bf" +listen-address = "0.0.0.0:6009" diff --git a/optools/images/Dockerfile b/optools/images/Dockerfile index 87e179de75701..f787d4bcc7b98 100644 --- a/optools/images/Dockerfile +++ b/optools/images/Dockerfile @@ -8,6 +8,7 @@ RUN go env -w GOPROXY=${GOPROXY} # Typecheck: enable type checking for ToSliceNoTypeCheck and ToSliceNoTypeCheck2 # Default: 0 (disabled), set to 1 to enable ARG TYPECHECK=0 +ARG GOBUILD_OPT="" RUN mkdir -p /go/src/github.com/matrixorigin/matrixone @@ -28,9 +29,9 @@ RUN module_files="$(sha256sum go.mod go.sum)" && \ COPY . . RUN if [ "$TYPECHECK" = "1" ]; then \ - make clean && make build TYPECHECK=1; \ + make clean && make build TYPECHECK=1 GOBUILD_OPT="${GOBUILD_OPT}"; \ else \ - make clean && make build; \ + make clean && make build GOBUILD_OPT="${GOBUILD_OPT}"; \ fi FROM matrixorigin/ubuntu:22.04 @@ -42,8 +43,9 @@ COPY --from=builder /go/src/github.com/matrixorigin/matrixone/cgo/*.so /usr/loca COPY --from=builder /go/src/github.com/matrixorigin/matrixone/pkg/monlp/tokenizer/dict /usr/local/share/jieba ENV MO_JIEBA_DICT_DIR=/usr/local/share/jieba -# ldconfig and run mo-service to check if the shared library is found -RUN ldconfig && /mo-service -h +# ldconfig and run mo-service to check if the shared library is found. A +# coverage-instrumented binary requires a writable GOCOVERDIR even for -h. +RUN mkdir -p /tmp/coverage && ldconfig && GOCOVERDIR=/tmp/coverage /mo-service -h WORKDIR / diff --git a/optools/run_bvt_group.sh b/optools/run_bvt_group.sh new file mode 100644 index 0000000000000..1984aac8bfb6b --- /dev/null +++ b/optools/run_bvt_group.sh @@ -0,0 +1,72 @@ +#!/usr/bin/env bash + +# Run one of the two time-balanced BVT groups. The latest BVT timing assigns +# about 1501s to group 0 and 1479s to group 1. Keep the statement-info +# producers and verifiers in one group: mo-tester runs directories in lexical order, so log, +# result_count, sql_source_type, and statement_query_type run before +# zz_statement_query_type. +# +# New top-level directories are assigned deterministically so every case is +# exercised immediately; rebalance the explicit list from collected timing. +set -euo pipefail + +if [[ $# -lt 3 || $# -gt 4 ]]; then + echo "Usage: $0 [resource-dir]" >&2 + exit 2 +fi + +tester_dir=$(cd "$1" && pwd) +case_root=$(cd "$2" && pwd) +group_id=$3 +resource_dir=${4:-} + +if ! [[ "${group_id}" =~ ^[01]$ ]]; then + echo "BVT group must be 0 or 1; got '${group_id}'" >&2 + exit 2 +fi + +declare -a included_scripts=() +declare -a resource_args=() +if [[ -n "${resource_dir}" ]]; then + resource_dir=$(cd "${resource_dir}" && pwd) + resource_args=(-s "${resource_dir}") +fi + +while IFS= read -r script_path; do + relative_path=${script_path#"${case_root}/"} + [[ "${relative_path}" == *optimistic* ]] && continue + top_level=${relative_path%%/*} + + case "${top_level}" in + array|auto_increment|benchmark|dataXtest|ddl|disttae|fake_pk|function|git4data|hint|join|keyword|load_data|mo_cloud|optimizer|pg_cast|plugin|prepare|procedure|query_result|sample|save_query_result|sequence|snapshot|sql_inject|stage|system|system_variable|temporary|tenant|tenxcloud_xx|time_window|union|util|vector|view) + assigned_group=0 + ;; + analyze|charset_collation|comment|cte|database|distinct|dml|dtype|expression|feature_limit|foreign_key|fulltext|geo|iceberg|log|metadata|operator|pessimistic_transaction|pitr|plan_cache|publication_subscription|qexec|recursive_cte|replace_statement|result_count|security|set|sql_source_type|statement_query_type|subquery|table|task|udf|window|zz_accesscontrol|zz_statement_query_type) + assigned_group=1 + ;; + *) + # Do not defer newly added suites: deterministically place them in + # one group and rebalance the explicit list from collected timing. + assigned_group=$(( $(printf '%s' "${top_level}" | cksum | awk '{print $1}') % 2 )) + ;; + esac + + if (( assigned_group == group_id )); then + included_scripts+=("${script_path}") + fi +done < <(find "${case_root}" -type f \( -name '*.sql' -o -name '*.test' \) -print | LC_ALL=C sort) + +if (( ${#included_scripts[@]} == 0 )); then + echo "BVT group ${group_id} has no test scripts" >&2 + exit 1 +fi + +include_list=$(IFS=,; echo "${included_scripts[*]}") +echo "Run BVT group ${group_id}: ${#included_scripts[@]} scripts" +if [[ -n "${resource_dir}" ]]; then + echo "BVT resource directory: ${resource_dir}" +fi +printf '%s\n' "${included_scripts[@]}" + +cd "${tester_dir}" +./run.sh -n -g -o -p "${case_root}" -i "${include_list}" "${resource_args[@]}" diff --git a/pkg/fileservice/aliyun_sdk.go b/pkg/fileservice/aliyun_sdk.go index 4765e0afe79bf..d0d4d4b954523 100644 --- a/pkg/fileservice/aliyun_sdk.go +++ b/pkg/fileservice/aliyun_sdk.go @@ -277,7 +277,7 @@ func (a *AliyunSDK) Write( if err != nil { return err } - _, err = DoWithRetry("write", func() (int, error) { + _, err = DoWithRetryContext(ctx, "write", func() (int, error) { return 0, a.putObject( ctx, key, @@ -413,7 +413,8 @@ func (a *AliyunSDK) listObjects(ctx context.Context, prefix string, cont string) if a.listMaxKeys > 0 { opts = append(opts, oss.MaxKeys(a.listMaxKeys)) } - return DoWithRetry( + return DoWithRetryContext( + ctx, "s3 list objects", func() (oss.ListObjectsResultV2, error) { perfcounter.Update(ctx, func(counter *perfcounter.CounterSet) { @@ -510,7 +511,8 @@ func (a *AliyunSDK) getObject(ctx context.Context, key string, min *int64, max * } opts = append(opts, oss.NormalizedRange(rang)) opts = append(opts, oss.RangeBehavior("standard")) - r, err := DoWithRetry( + r, err := DoWithRetryContext( + ctx, "s3 get object", func() (io.ReadCloser, error) { perfcounter.Update(ctx, func(counter *perfcounter.CounterSet) { @@ -548,7 +550,8 @@ func (a *AliyunSDK) getObject(ctx context.Context, key string, min *int64, max * func (a *AliyunSDK) deleteObject(ctx context.Context, key string) (bool, error) { ctx, task := gotrace.NewTask(ctx, "AliyunSDK.deleteObject") defer task.End() - return DoWithRetry( + return DoWithRetryContext( + ctx, "s3 delete object", func() (bool, error) { perfcounter.Update(ctx, func(counter *perfcounter.CounterSet) { @@ -570,7 +573,8 @@ func (a *AliyunSDK) deleteObject(ctx context.Context, key string) (bool, error) func (a *AliyunSDK) deleteObjects(ctx context.Context, keys ...string) (bool, error) { ctx, task := gotrace.NewTask(ctx, "AliyunSDK.deleteObjects") defer task.End() - return DoWithRetry( + return DoWithRetryContext( + ctx, "s3 delete objects", func() (bool, error) { perfcounter.Update(ctx, func(counter *perfcounter.CounterSet) { diff --git a/pkg/fileservice/aliyun_sdk_test.go b/pkg/fileservice/aliyun_sdk_test.go index 43ea229e08284..aad59baf2a40c 100644 --- a/pkg/fileservice/aliyun_sdk_test.go +++ b/pkg/fileservice/aliyun_sdk_test.go @@ -88,6 +88,40 @@ func TestAliyunSDKCopyObjectPropagatesCancellation(t *testing.T) { } } +func TestAliyunSDKListStopsRetryingAtContextDeadline(t *testing.T) { + client, err := oss.New( + "http://oss.test", + "id", + "secret", + oss.ForcePathStyle(true), + oss.HTTPClient(&http.Client{Transport: aliyunRetryTimeoutTransport{}}), + ) + require.NoError(t, err) + bucket, err := client.Bucket("bucket") + require.NoError(t, err) + sdk := &AliyunSDK{bucket: bucket} + + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + for _, err := range sdk.List(ctx, "") { + require.ErrorIs(t, err, context.DeadlineExceeded) + return + } + t.Fatal("expected list to return a deadline error") +} + +type aliyunRetryTimeoutTransport struct{} + +func (aliyunRetryTimeoutTransport) RoundTrip(*http.Request) (*http.Response, error) { + return nil, aliyunRetryTimeoutError{} +} + +type aliyunRetryTimeoutError struct{} + +func (aliyunRetryTimeoutError) Error() string { return "timeout" } +func (aliyunRetryTimeoutError) Timeout() bool { return true } +func (aliyunRetryTimeoutError) Temporary() bool { return true } + func TestAliyunPutObjectPhysicalAccounting(t *testing.T) { var fail bool server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { diff --git a/pkg/fileservice/aws_sdk_v2.go b/pkg/fileservice/aws_sdk_v2.go index 0195522cbab77..9695924067a09 100644 --- a/pkg/fileservice/aws_sdk_v2.go +++ b/pkg/fileservice/aws_sdk_v2.go @@ -365,7 +365,7 @@ func (a *AwsSDKv2) Write( if sizeHint == nil { // multipart - output, err := DoWithRetry("create multipart upload", func() (*s3.CreateMultipartUploadOutput, error) { + output, err := DoWithRetryContext(ctx, "create multipart upload", func() (*s3.CreateMultipartUploadOutput, error) { return a.client.CreateMultipartUpload(ctx, &s3.CreateMultipartUploadInput{ Bucket: ptrTo(a.bucket), Key: ptrTo(key), @@ -400,7 +400,7 @@ func (a *AwsSDKv2) Write( if len(content) == 0 { break } - uploadOutput, err := DoWithRetry("upload part", func() (*s3.UploadPartOutput, error) { + uploadOutput, err := DoWithRetryContext(ctx, "upload part", func() (*s3.UploadPartOutput, error) { recordS3PutRequest(ctx, a.perfCounterSets...) return a.client.UploadPart(ctx, &s3.UploadPartInput{ Bucket: ptrTo(a.bucket), @@ -426,7 +426,7 @@ func (a *AwsSDKv2) Write( } // complete - _, err = DoWithRetry("complete multipart upload", func() (*s3.CompleteMultipartUploadOutput, error) { + _, err = DoWithRetryContext(ctx, "complete multipart upload", func() (*s3.CompleteMultipartUploadOutput, error) { return a.client.CompleteMultipartUpload(ctx, &s3.CompleteMultipartUploadInput{ Bucket: ptrTo(a.bucket), Key: ptrTo(key), @@ -443,7 +443,7 @@ func (a *AwsSDKv2) Write( if err != nil { return err } - _, err = DoWithRetry("write", func() (*s3.PutObjectOutput, error) { + _, err = DoWithRetryContext(ctx, "write", func() (*s3.PutObjectOutput, error) { return a.putObject( ctx, &s3.PutObjectInput{ @@ -578,7 +578,7 @@ func (a *AwsSDKv2) WriteMultipartParallel( return a.Write(ctx, key, bytes.NewReader(data), &size, options.Expire) } - output, createErr := DoWithRetry("create multipart upload", func() (*s3.CreateMultipartUploadOutput, error) { + output, createErr := DoWithRetryContext(ctx, "create multipart upload", func() (*s3.CreateMultipartUploadOutput, error) { return a.client.CreateMultipartUpload(ctx, &s3.CreateMultipartUploadInput{ Bucket: ptrTo(a.bucket), Key: ptrTo(key), @@ -653,7 +653,7 @@ func (a *AwsSDKv2) WriteMultipartParallel( releasePartBuffer(job.part) return } - uploadOutput, uploadErr := DoWithRetry("upload part", func() (*s3.UploadPartOutput, error) { + uploadOutput, uploadErr := DoWithRetryContext(ctx, "upload part", func() (*s3.UploadPartOutput, error) { recordS3PutRequest(ctx, a.perfCounterSets...) return a.client.UploadPart(ctx, &s3.UploadPartInput{ Bucket: ptrTo(a.bucket), @@ -735,7 +735,7 @@ func (a *AwsSDKv2) WriteMultipartParallel( return cmp.Compare(*a.PartNumber, *b.PartNumber) }) - _, err = DoWithRetry("complete multipart upload", func() (*s3.CompleteMultipartUploadOutput, error) { + _, err = DoWithRetryContext(ctx, "complete multipart upload", func() (*s3.CompleteMultipartUploadOutput, error) { return a.client.CompleteMultipartUpload(ctx, &s3.CompleteMultipartUploadInput{ Bucket: ptrTo(a.bucket), Key: ptrTo(key), @@ -919,7 +919,8 @@ func (a *AwsSDKv2) deleteMultiObjOneByOne(ctx context.Context, objs []types.Obje func (a *AwsSDKv2) listObjects(ctx context.Context, params *s3.ListObjectsInput, optFns ...func(*s3.Options)) (*s3.ListObjectsOutput, error) { ctx, task := gotrace.NewTask(ctx, "AwsSDKv2.listObjects") defer task.End() - return DoWithRetry( + return DoWithRetryContext( + ctx, "s3 list objects", func() (*s3.ListObjectsOutput, error) { perfcounter.Update(ctx, func(counter *perfcounter.CounterSet) { @@ -978,7 +979,8 @@ func (a *AwsSDKv2) getObject(ctx context.Context, min *int64, max *int64, params rang = fmt.Sprintf("bytes=%d-", offset) } params.Range = &rang - output, err := DoWithRetry( + output, err := DoWithRetryContext( + ctx, "s3 get object", func() (*s3.GetObjectOutput, error) { LogEvent(ctx, str_awssdkv2_get_object_begin) @@ -1015,7 +1017,8 @@ func (a *AwsSDKv2) getObject(ctx context.Context, min *int64, max *int64, params func (a *AwsSDKv2) deleteObject(ctx context.Context, params *s3.DeleteObjectInput, optFns ...func(*s3.Options)) (*s3.DeleteObjectOutput, error) { ctx, task := gotrace.NewTask(ctx, "AwsSDKv2.deleteObject") defer task.End() - return DoWithRetry( + return DoWithRetryContext( + ctx, "s3 delete object", func() (*s3.DeleteObjectOutput, error) { perfcounter.Update(ctx, func(counter *perfcounter.CounterSet) { @@ -1031,7 +1034,8 @@ func (a *AwsSDKv2) deleteObject(ctx context.Context, params *s3.DeleteObjectInpu func (a *AwsSDKv2) deleteObjects(ctx context.Context, params *s3.DeleteObjectsInput, optFns ...func(*s3.Options)) (*s3.DeleteObjectsOutput, error) { ctx, task := gotrace.NewTask(ctx, "AwsSDKv2.deleteObjects") defer task.End() - return DoWithRetry( + return DoWithRetryContext( + ctx, "s3 delete objects", func() (*s3.DeleteObjectsOutput, error) { perfcounter.Update(ctx, func(counter *perfcounter.CounterSet) { diff --git a/pkg/fileservice/file_service_test.go b/pkg/fileservice/file_service_test.go index 38f8f2d6306b7..ae2f816cc108b 100644 --- a/pkg/fileservice/file_service_test.go +++ b/pkg/fileservice/file_service_test.go @@ -45,12 +45,26 @@ func testFileService( policy Policy, newFS func(name string) FileService, ) { + testFileServiceWithContext(t, policy, func(_ context.Context, name string) FileService { + return newFS(name) + }) +} +func testFileServiceWithContext( + t *testing.T, + policy Policy, + newFS func(context.Context, string) FileService, +) { fsName := time.Now().Format("fs-2006-01-02-15-04-05") + // Real object-storage specs are part of this shared test suite. Bound the + // constructor and the whole suite so a remote outage cannot consume the + // package-level Go test timeout. + testCtx, cancel := context.WithTimeout(t.Context(), 3*time.Minute) + defer cancel() t.Run("basic", func(t *testing.T) { - ctx := context.Background() - fs := newFS(fsName) + ctx := testCtx + fs := newFS(ctx, fsName) defer fs.Close(ctx) assert.True(t, strings.Contains(fs.Name(), fsName)) @@ -181,8 +195,8 @@ func testFileService( }) t.Run("WriterForRead", func(t *testing.T) { - fs := newFS(fsName) - ctx := context.Background() + ctx := testCtx + fs := newFS(ctx, fsName) defer fs.Close(ctx) err := fs.Write(ctx, IOVector{ @@ -233,8 +247,8 @@ func testFileService( }) t.Run("ReadCloserForRead", func(t *testing.T) { - ctx := context.Background() - fs := newFS(fsName) + ctx := testCtx + fs := newFS(ctx, fsName) defer fs.Close(ctx) err := fs.Write(ctx, IOVector{ @@ -312,8 +326,8 @@ func testFileService( }) t.Run("random", func(t *testing.T) { - fs := newFS(fsName) - ctx := context.Background() + ctx := testCtx + fs := newFS(ctx, fsName) defer fs.Close(ctx) for i := 0; i < 8; i++ { @@ -442,8 +456,8 @@ func testFileService( }) t.Run("tree", func(t *testing.T) { - fs := newFS(fsName) - ctx := context.Background() + ctx := testCtx + fs := newFS(ctx, fsName) defer fs.Close(ctx) for _, dir := range []string{ @@ -573,8 +587,8 @@ func testFileService( }) t.Run("errors", func(t *testing.T) { - fs := newFS(fsName) - ctx := context.Background() + ctx := testCtx + fs := newFS(ctx, fsName) defer fs.Close(ctx) err := fs.Read(ctx, &IOVector{ @@ -678,8 +692,8 @@ func testFileService( }) t.Run("cache data", func(t *testing.T) { - ctx := context.Background() - fs := newFS(fsName) + ctx := testCtx + fs := newFS(ctx, fsName) defer fs.Close(ctx) var counterSet perfcounter.CounterSet ctx = perfcounter.WithCounterSet(ctx, &counterSet) @@ -755,8 +769,8 @@ func testFileService( }) t.Run("ignore", func(t *testing.T) { - ctx := context.Background() - fs := newFS(fsName) + ctx := testCtx + fs := newFS(ctx, fsName) defer fs.Close(ctx) data := []byte("foo") @@ -793,8 +807,8 @@ func testFileService( }) t.Run("named path", func(t *testing.T) { - ctx := context.Background() - fs := newFS(fsName) + ctx := testCtx + fs := newFS(ctx, fsName) defer fs.Close(ctx) // write @@ -863,8 +877,8 @@ func testFileService( }) t.Run("issue6110", func(t *testing.T) { - ctx := context.Background() - fs := newFS(fsName) + ctx := testCtx + fs := newFS(ctx, fsName) defer fs.Close(ctx) err := fs.Write(ctx, IOVector{ @@ -886,8 +900,8 @@ func testFileService( }) t.Run("streaming write", func(t *testing.T) { - ctx := context.Background() - fs := newFS(fsName) + ctx := testCtx + fs := newFS(ctx, fsName) defer fs.Close(ctx) reader, writer := io.Pipe() @@ -981,11 +995,11 @@ func testFileService( }, Policy: policy, } - ctx, cancel := context.WithCancel(context.Background()) + cancelCtx, cancel := context.WithCancel(context.Background()) cancel() errCh := make(chan error) go func() { - err := fs.Write(ctx, vec) + err := fs.Write(cancelCtx, vec) errCh <- err }() select { @@ -1000,7 +1014,7 @@ func testFileService( t.Run("context cancel", func(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) cancel() - fs := newFS(fsName) + fs := newFS(testCtx, fsName) defer fs.Close(ctx) err := fs.Write(ctx, IOVector{ @@ -1023,7 +1037,7 @@ func testFileService( t.Run("NewReader and NewWriter", func(t *testing.T) { ctx, cancel := context.WithCancel(t.Context()) defer cancel() - fs := newFS(fsName) + fs := newFS(testCtx, fsName) defer fs.Close(ctx) rwFS, ok := fs.(ReaderWriterFileService) @@ -1047,7 +1061,7 @@ func testFileService( }) t.Run("NewReader and NewWriter error", func(t *testing.T) { - fs := newFS(fsName) + fs := newFS(testCtx, fsName) defer fs.Close(t.Context()) rwFS, ok := fs.(ReaderWriterFileService) diff --git a/pkg/fileservice/hdfs.go b/pkg/fileservice/hdfs.go index 42f943989c3d2..4a99376b902e6 100644 --- a/pkg/fileservice/hdfs.go +++ b/pkg/fileservice/hdfs.go @@ -133,7 +133,7 @@ func (h *HDFS) Delete(ctx context.Context, keys ...string) (err error) { }, h.perfCounterSets...) for _, key := range keys { - _, err := DoWithRetry("HDFS: delete", func() (bool, error) { + _, err := DoWithRetryContext(ctx, "HDFS: delete", func() (bool, error) { return true, h.client.Remove(h.keyToPath(key)) }, maxRetryAttemps, IsRetryableError) if err != nil { @@ -153,21 +153,21 @@ func (h *HDFS) keyToPath(key string) string { return ret } -func (h *HDFS) createTemp() (*hdfs.FileWriter, string, error) { +func (h *HDFS) createTemp(ctx context.Context) (*hdfs.FileWriter, string, error) { for { randomBytes := make([]byte, 26) rand.Read(randomBytes) name := h.keyToPath( base32.HexEncoding.WithPadding(base32.NoPadding).EncodeToString(randomBytes), ) - _, err := DoWithRetry("HDFS: stat", func() (os.FileInfo, error) { + _, err := DoWithRetryContext(ctx, "HDFS: stat", func() (os.FileInfo, error) { return h.client.Stat(name) }, maxRetryAttemps, IsRetryableError) if err == nil { // existed continue } - file, err := DoWithRetry("HDFS: create", func() (*hdfs.FileWriter, error) { + file, err := DoWithRetryContext(ctx, "HDFS: create", func() (*hdfs.FileWriter, error) { return h.client.Create(name) }, maxRetryAttemps, IsRetryableError) if err != nil { @@ -186,7 +186,7 @@ func (h *HDFS) Exists(ctx context.Context, key string) (bool, error) { counter.FileService.S3.Head.Add(1) }, h.perfCounterSets...) - _, err := DoWithRetry("HDFS: stat", func() (os.FileInfo, error) { + _, err := DoWithRetryContext(ctx, "HDFS: stat", func() (os.FileInfo, error) { return h.client.Stat(h.keyToPath(key)) }, maxRetryAttemps, IsRetryableError) if err != nil { @@ -212,7 +212,7 @@ func (h *HDFS) List(ctx context.Context, prefix string) iter.Seq2[*DirEntry, err dir, prefix := path.Split(prefix) - f, err := DoWithRetry("HDFS: open", func() (*hdfs.FileReader, error) { + f, err := DoWithRetryContext(ctx, "HDFS: open", func() (*hdfs.FileReader, error) { return h.client.Open(h.keyToPath(dir)) }, maxRetryAttemps, IsRetryableError) if err != nil { @@ -271,7 +271,7 @@ func (h *HDFS) Read(ctx context.Context, key string, min *int64, max *int64) (r counter.FileService.S3.Get.Add(1) }, h.perfCounterSets...) - f, err := DoWithRetry("HDFS: open", func() (*hdfs.FileReader, error) { + f, err := DoWithRetryContext(ctx, "HDFS: open", func() (*hdfs.FileReader, error) { return h.client.Open(h.keyToPath(key)) }, maxRetryAttemps, IsRetryableError) if err != nil { @@ -321,7 +321,7 @@ func (h *HDFS) Stat(ctx context.Context, key string) (size int64, err error) { counter.FileService.S3.Head.Add(1) }, h.perfCounterSets...) - stat, err := DoWithRetry("HDFS: stat", func() (os.FileInfo, error) { + stat, err := DoWithRetryContext(ctx, "HDFS: stat", func() (os.FileInfo, error) { return h.client.Stat(h.keyToPath(key)) }, maxRetryAttemps, IsRetryableError) if err != nil { @@ -345,7 +345,7 @@ func (h *HDFS) Write(ctx context.Context, key string, r io.Reader, sizeHint *int counter.FileService.S3.Put.Add(1) }, h.perfCounterSets...) - tempFile, tempFilePath, err := h.createTemp() + tempFile, tempFilePath, err := h.createTemp(ctx) if err != nil { return err } @@ -368,7 +368,7 @@ func (h *HDFS) Write(ctx context.Context, key string, r io.Reader, sizeHint *int return moerr.NewSizeNotMatchNoCtx(key) } - _, err = DoWithRetry("HDFS: FileWriter.Close", func() (bool, error) { + _, err = DoWithRetryContext(ctx, "HDFS: FileWriter.Close", func() (bool, error) { // may retrun 'replication in progress' error, retry until OK return true, tempFile.Close() }, maxRetryAttemps, IsRetryableError) @@ -377,7 +377,7 @@ func (h *HDFS) Write(ctx context.Context, key string, r io.Reader, sizeHint *int } filePath := h.keyToPath(key) - _, err = DoWithRetry("HDFS: MkdirAll", func() (bool, error) { + _, err = DoWithRetryContext(ctx, "HDFS: MkdirAll", func() (bool, error) { return true, h.client.MkdirAll(path.Dir(filePath), 0755) }, maxRetryAttemps, IsRetryableError) if err != nil { @@ -386,7 +386,7 @@ func (h *HDFS) Write(ctx context.Context, key string, r io.Reader, sizeHint *int } } - _, err = DoWithRetry("HDFS: Rename", func() (bool, error) { + _, err = DoWithRetryContext(ctx, "HDFS: Rename", func() (bool, error) { return true, h.client.Rename(tempFilePath, filePath) }, maxRetryAttemps, IsRetryableError) if err != nil { diff --git a/pkg/fileservice/minio_sdk.go b/pkg/fileservice/minio_sdk.go index ebd4e7e58ab12..653cf5bfc411d 100644 --- a/pkg/fileservice/minio_sdk.go +++ b/pkg/fileservice/minio_sdk.go @@ -369,7 +369,7 @@ func (a *MinioSDK) Write( if err != nil { return err } - _, err = DoWithRetry("write", func() (minio.UploadInfo, error) { + _, err = DoWithRetryContext(ctx, "write", func() (minio.UploadInfo, error) { return a.putObject( ctx, key, @@ -500,7 +500,8 @@ func (a *MinioSDK) deleteSingle(ctx context.Context, key string) error { func (a *MinioSDK) listObjects(ctx context.Context, prefix string, marker string) (minio.ListBucketResult, error) { ctx, task := gotrace.NewTask(ctx, "MinioSDK.listObjects") defer task.End() - return DoWithRetry( + return DoWithRetryContext( + ctx, "s3 list objects", func() (minio.ListBucketResult, error) { perfcounter.Update(ctx, func(counter *perfcounter.CounterSet) { @@ -579,7 +580,8 @@ func (a *MinioSDK) getObject(ctx context.Context, key string, min *int64, max *i } r, err := newRetryableReader( func(offset int64) (io.ReadCloser, error) { - obj, err := DoWithRetry( + obj, err := DoWithRetryContext( + ctx, "s3 get object", func() (obj *minio.Object, err error) { perfcounter.Update(ctx, func(counter *perfcounter.CounterSet) { @@ -619,7 +621,8 @@ func (a *MinioSDK) getObject(ctx context.Context, key string, min *int64, max *i func (a *MinioSDK) deleteObject(ctx context.Context, key string) (any, error) { ctx, task := gotrace.NewTask(ctx, "MinioSDK.deleteObject") defer task.End() - return DoWithRetry( + return DoWithRetryContext( + ctx, "s3 delete object", func() (any, error) { perfcounter.Update(ctx, func(counter *perfcounter.CounterSet) { @@ -638,7 +641,8 @@ func (a *MinioSDK) deleteObject(ctx context.Context, key string) (any, error) { func (a *MinioSDK) deleteObjects(ctx context.Context, keys ...string) (any, error) { ctx, task := gotrace.NewTask(ctx, "MinioSDK.deleteObjects") defer task.End() - return DoWithRetry( + return DoWithRetryContext( + ctx, "s3 delete objects", func() (any, error) { perfcounter.Update(ctx, func(counter *perfcounter.CounterSet) { diff --git a/pkg/fileservice/object_storage_test.go b/pkg/fileservice/object_storage_test.go index 01a9eac3140b8..f0df570162558 100644 --- a/pkg/fileservice/object_storage_test.go +++ b/pkg/fileservice/object_storage_test.go @@ -32,12 +32,32 @@ func testObjectStorage[T ObjectStorage]( sdkName string, newStorage func(t *testing.T) T, ) { + testObjectStorageWithContext(t, sdkName, func(t *testing.T, _ context.Context) T { + return newStorage(t) + }) +} + +func testObjectStorageWithContext[T ObjectStorage]( + t *testing.T, + sdkName string, + newStorage func(*testing.T, context.Context) T, +) { + testObjectStorageWithParentContext(t, sdkName, t.Context(), newStorage) +} +func testObjectStorageWithParentContext[T ObjectStorage]( + t *testing.T, + sdkName string, + parentCtx context.Context, + newStorage func(*testing.T, context.Context) T, +) { t.Run(sdkName, func(t *testing.T) { + testCtx, cancel := context.WithTimeout(parentCtx, 3*time.Minute) + defer cancel() t.Run("basic", func(t *testing.T) { - storage := newStorage(t) - ctx := context.Background() + storage := newStorage(t, testCtx) + ctx := testCtx prefix := time.Now().Format("2006-01-02-15-04-05.000000") name := path.Join(prefix, "foo") @@ -159,8 +179,8 @@ func testObjectStorage[T ObjectStorage]( }) t.Run("invalid write length", func(t *testing.T) { - storage := newStorage(t) - ctx := context.Background() + storage := newStorage(t, testCtx) + ctx := testCtx prefix := time.Now().Format("2006-01-02-15-04-05.000000") name := path.Join(prefix, "foo") @@ -183,8 +203,8 @@ func testObjectStorage[T ObjectStorage]( }) t.Run("write empty", func(t *testing.T) { - storage := newStorage(t) - ctx := context.Background() + storage := newStorage(t, testCtx) + ctx := testCtx prefix := time.Now().Format("2006-01-02-15-04-05.000000") name := path.Join(prefix, "foo") @@ -207,13 +227,15 @@ func TestObjectStorages(t *testing.T) { for _, args := range objectStorageArgumentsForTest("test", t) { t.Run(args.Name, func(t *testing.T) { + specCtx, cancel := context.WithTimeout(t.Context(), 3*time.Minute) + defer cancel() switch { case strings.HasPrefix(strings.ToLower(args.Endpoint), "hdfs"): // HDFS - testObjectStorage(t, "hdfs", func(t *testing.T) *HDFS { - storage, err := NewHDFS(context.Background(), args, nil) + testObjectStorageWithParentContext(t, "hdfs", specCtx, func(t *testing.T, ctx context.Context) *HDFS { + storage, err := NewHDFS(ctx, args, nil) if err != nil { t.Fatal(err) } @@ -222,8 +244,8 @@ func TestObjectStorages(t *testing.T) { case args.Endpoint == "disk": // disk - testObjectStorage(t, "disk", func(t *testing.T) *diskObjectStorage { - storage, err := newDiskObjectStorage(context.Background(), args, nil) + testObjectStorageWithParentContext(t, "disk", specCtx, func(t *testing.T, ctx context.Context) *diskObjectStorage { + storage, err := newDiskObjectStorage(ctx, args, nil) if err != nil { t.Fatal(err) } @@ -232,23 +254,23 @@ func TestObjectStorages(t *testing.T) { case strings.Contains(args.Endpoint, "aliyun"): // aliyun - testObjectStorage(t, "aliyun", func(t *testing.T) *AliyunSDK { - storage, err := NewAliyunSDK(context.Background(), args, nil) + testObjectStorageWithParentContext(t, "aliyun", specCtx, func(t *testing.T, ctx context.Context) *AliyunSDK { + storage, err := NewAliyunSDK(ctx, args, nil) if err != nil { t.Fatal(err) } return storage }) if args.RoleARN == "" { - testObjectStorage(t, "minio", func(t *testing.T) *MinioSDK { - storage, err := NewMinioSDK(context.Background(), args, nil) + testObjectStorageWithParentContext(t, "minio", specCtx, func(t *testing.T, ctx context.Context) *MinioSDK { + storage, err := NewMinioSDK(ctx, args, nil) if err != nil { t.Fatal(err) } return storage }) - testObjectStorage(t, "aws", func(t *testing.T) *AwsSDKv2 { - storage, err := NewAwsSDKv2(context.Background(), args, nil) + testObjectStorageWithParentContext(t, "aws", specCtx, func(t *testing.T, ctx context.Context) *AwsSDKv2 { + storage, err := NewAwsSDKv2(ctx, args, nil) if err != nil { t.Fatal(err) } @@ -258,22 +280,22 @@ func TestObjectStorages(t *testing.T) { case strings.Contains(args.Endpoint, "qcloud"): // qcloud - testObjectStorage(t, "aws", func(t *testing.T) *AwsSDKv2 { - storage, err := NewAwsSDKv2(context.Background(), args, nil) + testObjectStorageWithParentContext(t, "aws", specCtx, func(t *testing.T, ctx context.Context) *AwsSDKv2 { + storage, err := NewAwsSDKv2(ctx, args, nil) if err != nil { t.Fatal(err) } return storage }) - testObjectStorage(t, "minio", func(t *testing.T) *MinioSDK { - storage, err := NewMinioSDK(context.Background(), args, nil) + testObjectStorageWithParentContext(t, "minio", specCtx, func(t *testing.T, ctx context.Context) *MinioSDK { + storage, err := NewMinioSDK(ctx, args, nil) if err != nil { t.Fatal(err) } return storage }) - testObjectStorage(t, "qcloud", func(t *testing.T) *QCloudSDK { - storage, err := NewQCloudSDK(context.Background(), args, nil) + testObjectStorageWithParentContext(t, "qcloud", specCtx, func(t *testing.T, ctx context.Context) *QCloudSDK { + storage, err := NewQCloudSDK(ctx, args, nil) if err != nil { t.Fatal(err) } @@ -282,15 +304,15 @@ func TestObjectStorages(t *testing.T) { case strings.Contains(args.Endpoint, "aws"): // AWS - testObjectStorage(t, "aws", func(t *testing.T) *AwsSDKv2 { - storage, err := NewAwsSDKv2(context.Background(), args, nil) + testObjectStorageWithParentContext(t, "aws", specCtx, func(t *testing.T, ctx context.Context) *AwsSDKv2 { + storage, err := NewAwsSDKv2(ctx, args, nil) if err != nil { t.Fatal(err) } return storage }) - testObjectStorage(t, "minio", func(t *testing.T) *MinioSDK { - storage, err := NewMinioSDK(context.Background(), args, nil) + testObjectStorageWithParentContext(t, "minio", specCtx, func(t *testing.T, ctx context.Context) *MinioSDK { + storage, err := NewMinioSDK(ctx, args, nil) if err != nil { t.Fatal(err) } @@ -299,15 +321,15 @@ func TestObjectStorages(t *testing.T) { case strings.Contains(args.Endpoint, "qiniu"): // qiniu - testObjectStorage(t, "minio", func(t *testing.T) *MinioSDK { - storage, err := NewMinioSDK(context.Background(), args, nil) + testObjectStorageWithParentContext(t, "minio", specCtx, func(t *testing.T, ctx context.Context) *MinioSDK { + storage, err := NewMinioSDK(ctx, args, nil) if err != nil { t.Fatal(err) } return storage }) - testObjectStorage(t, "aws", func(t *testing.T) *AwsSDKv2 { - storage, err := NewAwsSDKv2(context.Background(), args, nil) + testObjectStorageWithParentContext(t, "aws", specCtx, func(t *testing.T, ctx context.Context) *AwsSDKv2 { + storage, err := NewAwsSDKv2(ctx, args, nil) if err != nil { t.Fatal(err) } diff --git a/pkg/fileservice/qcloud_sdk.go b/pkg/fileservice/qcloud_sdk.go index e44664b980e07..226a564f25aba 100644 --- a/pkg/fileservice/qcloud_sdk.go +++ b/pkg/fileservice/qcloud_sdk.go @@ -49,6 +49,8 @@ type QCloudSDK struct { listMaxKeys int } +const qcloudMultipartAbortTimeout = 30 * time.Second + func NewQCloudSDK( ctx context.Context, args ObjectStorageArguments, @@ -122,7 +124,7 @@ func NewQCloudSDK( if !args.NoBucketValidation { // validate bucket - _, err := DoWithRetry("cos bucket head", func() (*cos.Response, error) { + _, err := DoWithRetryContext(ctx, "cos bucket head", func() (*cos.Response, error) { return client.Bucket.Head(ctx, &cos.BucketHeadOptions{}) }, maxRetryAttemps, IsRetryableError) if err != nil { @@ -285,7 +287,7 @@ func (a *QCloudSDK) Write( if err != nil { return err } - _, err = DoWithRetry("write", func() (int, error) { + _, err = DoWithRetryContext(ctx, "write", func() (int, error) { return 0, a.putObject( ctx, key, @@ -315,7 +317,7 @@ func (a *QCloudSDK) Write( if err != nil { return err } - _, err = DoWithRetry("write", func() (int, error) { + _, err = DoWithRetryContext(ctx, "write", func() (int, error) { if _, err := seeker.Seek(offset, io.SeekStart); err != nil { return 0, err } @@ -428,7 +430,7 @@ func (a *QCloudSDK) WriteMultipartParallel( Expires: expiresHeader, }, } - output, createErr := DoWithRetry("cos initiate multipart upload", func() (*cos.InitiateMultipartUploadResult, error) { + output, createErr := DoWithRetryContext(ctx, "cos initiate multipart upload", func() (*cos.InitiateMultipartUploadResult, error) { res, _, e := a.client.Object.InitiateMultipartUpload(ctx, key, initOpt) return res, e }, maxRetryAttemps, IsRetryableError) @@ -439,10 +441,20 @@ func (a *QCloudSDK) WriteMultipartParallel( defer func() { if err != nil { - _, _ = DoWithRetry("cos abort multipart upload", func() (*cos.Response, error) { - return a.client.Object.AbortMultipartUpload( - context.WithoutCancel(parentCtx), key, output.UploadID) - }, maxRetryAttemps, IsRetryableError) + // The upload context is normally canceled on the first part + // failure, but abort still needs a live context to remove the + // server-side multipart upload. Bound that detached cleanup so a + // broken COS endpoint cannot delay the original Write forever. + abortCtx, abortCancel := context.WithTimeoutCause( + context.WithoutCancel(parentCtx), + qcloudMultipartAbortTimeout, + context.DeadlineExceeded, + ) + defer abortCancel() + if abortErr := a.abortMultipartUpload(abortCtx, key, output.UploadID); abortErr != nil { + logutil.Warn("failed to abort cos multipart upload", + zap.Error(abortErr)) + } } }() @@ -501,7 +513,7 @@ func (a *QCloudSDK) WriteMultipartParallel( uploadOpt := &cos.ObjectUploadPartOptions{ ContentLength: int64(job.part.n), } - resp, uploadErr := DoWithRetry("cos upload part", func() (*cos.Response, error) { + resp, uploadErr := DoWithRetryContext(ctx, "cos upload part", func() (*cos.Response, error) { recordS3PutRequest(ctx, a.perfCounterSets...) return a.client.Object.UploadPart(ctx, key, output.UploadID, int(job.num), bytes.NewReader(job.part.buf[:job.part.n]), uploadOpt) }, maxRetryAttemps, IsRetryableError) @@ -584,7 +596,7 @@ func (a *QCloudSDK) WriteMultipartParallel( completeOpt := &cos.CompleteMultipartUploadOptions{ Parts: parts, } - _, err = DoWithRetry("cos complete multipart upload", func() (*cos.CompleteMultipartUploadResult, error) { + _, err = DoWithRetryContext(ctx, "cos complete multipart upload", func() (*cos.CompleteMultipartUploadResult, error) { res, _, e := a.client.Object.CompleteMultipartUpload(ctx, key, output.UploadID, completeOpt) return res, e }, maxRetryAttemps, IsRetryableError) @@ -595,6 +607,17 @@ func (a *QCloudSDK) WriteMultipartParallel( return nil } +func (a *QCloudSDK) abortMultipartUpload( + ctx context.Context, + key string, + uploadID string, +) error { + _, err := DoWithRetryContext(ctx, "cos abort multipart upload", func() (*cos.Response, error) { + return a.client.Object.AbortMultipartUpload(ctx, key, uploadID) + }, maxRetryAttemps, IsRetryableError) + return err +} + func (a *QCloudSDK) Read( ctx context.Context, key string, @@ -703,7 +726,8 @@ func (a *QCloudSDK) listObjects(ctx context.Context, prefix string, marker strin opts.MaxKeys = a.listMaxKeys } - return DoWithRetry( + return DoWithRetryContext( + ctx, "s3 list objects", func() (*cos.BucketGetResult, error) { perfcounter.Update(ctx, func(counter *perfcounter.CounterSet) { @@ -791,7 +815,8 @@ func (a *QCloudSDK) getObject(ctx context.Context, key string, min *int64, max * Range: rang, } - return DoWithRetry( + return DoWithRetryContext( + ctx, "s3 get object", func() (io.ReadCloser, error) { perfcounter.Update(ctx, func(counter *perfcounter.CounterSet) { @@ -823,7 +848,8 @@ func (a *QCloudSDK) getObject(ctx context.Context, key string, min *int64, max * func (a *QCloudSDK) deleteObject(ctx context.Context, key string) (bool, error) { ctx, task := gotrace.NewTask(ctx, "QCloudSDK.deleteObject") defer task.End() - return DoWithRetry( + return DoWithRetryContext( + ctx, "s3 delete object", func() (bool, error) { perfcounter.Update(ctx, func(counter *perfcounter.CounterSet) { @@ -842,7 +868,8 @@ func (a *QCloudSDK) deleteObject(ctx context.Context, key string) (bool, error) func (a *QCloudSDK) deleteObjects(ctx context.Context, keys ...string) (bool, error) { ctx, task := gotrace.NewTask(ctx, "QCloudSDK.deleteObjects") defer task.End() - return DoWithRetry( + return DoWithRetryContext( + ctx, "s3 delete objects", func() (bool, error) { objects := make([]cos.Object, 0, len(keys)) diff --git a/pkg/fileservice/qcloud_sdk_test.go b/pkg/fileservice/qcloud_sdk_test.go index 6a5d32a298eba..618b8c3c1879b 100644 --- a/pkg/fileservice/qcloud_sdk_test.go +++ b/pkg/fileservice/qcloud_sdk_test.go @@ -25,7 +25,9 @@ import ( "net/http/httptest" "net/url" "strings" + "sync/atomic" "testing" + "time" "github.com/matrixorigin/matrixone/pkg/common/moerr" "github.com/stretchr/testify/require" @@ -42,10 +44,10 @@ func TestQCloudSDK(t *testing.T) { t.Run(fmt.Sprintf("%s %s", args.Name, args.Bucket), func(t *testing.T) { - testObjectStorage(t, "qcloud", func(t *testing.T) *QCloudSDK { + testObjectStorageWithContext(t, "qcloud", func(t *testing.T, ctx context.Context) *QCloudSDK { args.KeyPrefix = fmt.Sprintf("%v", rand.Int64()) ret, err := NewQCloudSDK( - context.Background(), + ctx, args, nil, ) @@ -68,11 +70,11 @@ func TestQCloudSDK(t *testing.T) { t.Run(fmt.Sprintf("%s %s", args.Name, args.Bucket), func(t *testing.T) { t.Run("file service", func(t *testing.T) { - testFileService(t, 0, func(name string) FileService { + testFileServiceWithContext(t, 0, func(ctx context.Context, name string) FileService { args.Name = name args.KeyPrefix = fmt.Sprintf("%v", rand.Int64()) ret, err := NewS3FS( - context.Background(), + ctx, args, DisabledCacheConfig, nil, @@ -398,6 +400,44 @@ func TestQCloudSDKBasicObjectOperations(t *testing.T) { } } +func TestQCloudSDKListStopsRetryingAtContextDeadline(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusBadGateway) + })) + defer server.Close() + + sdk := newTestCOSClient(t, server) + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + + for _, err := range sdk.List(ctx, "") { + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("expected context deadline exceeded, got %v", err) + } + return + } + t.Fatal("expected list to return a deadline error") +} + +func TestQCloudSDKAbortMultipartStopsAtContextDeadline(t *testing.T) { + var requests atomic.Int64 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + requests.Add(1) + w.WriteHeader(http.StatusBadGateway) + })) + defer server.Close() + + sdk := newTestCOSClient(t, server) + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + + start := time.Now() + err := sdk.abortMultipartUpload(ctx, "object", "upload-id") + require.ErrorIs(t, err, context.DeadlineExceeded) + require.Less(t, time.Since(start), time.Second) + require.Positive(t, requests.Load()) +} + type timeoutError struct{} func (timeoutError) Error() string { diff --git a/pkg/fileservice/s3_fs_test.go b/pkg/fileservice/s3_fs_test.go index d0325519ea730..36d45422369d4 100644 --- a/pkg/fileservice/s3_fs_test.go +++ b/pkg/fileservice/s3_fs_test.go @@ -1753,12 +1753,12 @@ func TestS3FSFromSpecs(t *testing.T) { t.Run(args.Name, func(t *testing.T) { - testFileService(t, 0, func(name string) FileService { + testFileServiceWithContext(t, 0, func(ctx context.Context, name string) FileService { args.Name = name args.KeyPrefix = time.Now().Format("2006-01-02.15:04:05.000000") fs, err := NewS3FS( - context.Background(), + ctx, args, DisabledCacheConfig, nil,