diff --git a/ci/resources/stemcell-version-bump/go.mod b/ci/resources/stemcell-version-bump/go.mod index 38bbd087d..367c59be3 100644 --- a/ci/resources/stemcell-version-bump/go.mod +++ b/ci/resources/stemcell-version-bump/go.mod @@ -5,18 +5,18 @@ go 1.23.0 toolchain go1.24.1 require ( - cloud.google.com/go/storage v1.52.0 + cloud.google.com/go/storage v1.54.0 github.com/stretchr/testify v1.10.0 google.golang.org/api v0.232.0 ) require ( cel.dev/expr v0.20.0 // indirect - cloud.google.com/go v0.120.0 // indirect + cloud.google.com/go v0.121.0 // indirect cloud.google.com/go/auth v0.16.1 // indirect cloud.google.com/go/auth/oauth2adapt v0.2.8 // indirect cloud.google.com/go/compute/metadata v0.6.0 // indirect - cloud.google.com/go/iam v1.5.0 // indirect + cloud.google.com/go/iam v1.5.2 // indirect cloud.google.com/go/monitoring v1.24.0 // indirect github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.27.0 // indirect github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/metric v0.51.0 // indirect @@ -55,8 +55,8 @@ require ( golang.org/x/text v0.24.0 // indirect golang.org/x/time v0.11.0 // indirect google.golang.org/genproto v0.0.0-20250303144028-a0af3efb3deb // indirect - google.golang.org/genproto/googleapis/api v0.0.0-20250414145226-207652e42e2e // indirect - google.golang.org/genproto/googleapis/rpc v0.0.0-20250428153025-10db94c68c34 // indirect + google.golang.org/genproto/googleapis/api v0.0.0-20250505200425-f936aa4a68b2 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20250505200425-f936aa4a68b2 // indirect google.golang.org/grpc v1.72.0 // indirect google.golang.org/protobuf v1.36.6 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect diff --git a/ci/resources/stemcell-version-bump/go.sum b/ci/resources/stemcell-version-bump/go.sum index 9c30a900a..c549874da 100644 --- a/ci/resources/stemcell-version-bump/go.sum +++ b/ci/resources/stemcell-version-bump/go.sum @@ -1,23 +1,23 @@ cel.dev/expr v0.20.0 h1:OunBvVCfvpWlt4dN7zg3FM6TDkzOePe1+foGJ9AXeeI= cel.dev/expr v0.20.0/go.mod h1:MrpN08Q+lEBs+bGYdLxxHkZoUSsCp0nSKTs0nTymJgw= -cloud.google.com/go v0.120.0 h1:wc6bgG9DHyKqF5/vQvX1CiZrtHnxJjBlKUyF9nP6meA= -cloud.google.com/go v0.120.0/go.mod h1:/beW32s8/pGRuj4IILWQNd4uuebeT4dkOhKmkfit64Q= +cloud.google.com/go v0.121.0 h1:pgfwva8nGw7vivjZiRfrmglGWiCJBP+0OmDpenG/Fwg= +cloud.google.com/go v0.121.0/go.mod h1:rS7Kytwheu/y9buoDmu5EIpMMCI4Mb8ND4aeN4Vwj7Q= cloud.google.com/go/auth v0.16.1 h1:XrXauHMd30LhQYVRHLGvJiYeczweKQXZxsTbV9TiguU= cloud.google.com/go/auth v0.16.1/go.mod h1:1howDHJ5IETh/LwYs3ZxvlkXF48aSqqJUM+5o02dNOI= cloud.google.com/go/auth/oauth2adapt v0.2.8 h1:keo8NaayQZ6wimpNSmW5OPc283g65QNIiLpZnkHRbnc= cloud.google.com/go/auth/oauth2adapt v0.2.8/go.mod h1:XQ9y31RkqZCcwJWNSx2Xvric3RrU88hAYYbjDWYDL+c= cloud.google.com/go/compute/metadata v0.6.0 h1:A6hENjEsCDtC1k8byVsgwvVcioamEHvZ4j01OwKxG9I= cloud.google.com/go/compute/metadata v0.6.0/go.mod h1:FjyFAW1MW0C203CEOMDTu3Dk1FlqW3Rga40jzHL4hfg= -cloud.google.com/go/iam v1.5.0 h1:QlLcVMhbLGOjRcGe6VTGGTyQib8dRLK2B/kYNV0+2xs= -cloud.google.com/go/iam v1.5.0/go.mod h1:U+DOtKQltF/LxPEtcDLoobcsZMilSRwR7mgNL7knOpo= +cloud.google.com/go/iam v1.5.2 h1:qgFRAGEmd8z6dJ/qyEchAuL9jpswyODjA2lS+w234g8= +cloud.google.com/go/iam v1.5.2/go.mod h1:SE1vg0N81zQqLzQEwxL2WI6yhetBdbNQuTvIKCSkUHE= cloud.google.com/go/logging v1.13.0 h1:7j0HgAp0B94o1YRDqiqm26w4q1rDMH7XNRU34lJXHYc= cloud.google.com/go/logging v1.13.0/go.mod h1:36CoKh6KA/M0PbhPKMq6/qety2DCAErbhXT62TuXALA= -cloud.google.com/go/longrunning v0.6.6 h1:XJNDo5MUfMM05xK3ewpbSdmt7R2Zw+aQEMbdQR65Rbw= -cloud.google.com/go/longrunning v0.6.6/go.mod h1:hyeGJUrPHcx0u2Uu1UFSoYZLn4lkMrccJig0t4FI7yw= +cloud.google.com/go/longrunning v0.6.7 h1:IGtfDWHhQCgCjwQjV9iiLnUta9LBCo8R9QmAFsS/PrE= +cloud.google.com/go/longrunning v0.6.7/go.mod h1:EAFV3IZAKmM56TyiE6VAP3VoTzhZzySwI/YI1s/nRsY= cloud.google.com/go/monitoring v1.24.0 h1:csSKiCJ+WVRgNkRzzz3BPoGjFhjPY23ZTcaenToJxMM= cloud.google.com/go/monitoring v1.24.0/go.mod h1:Bd1PRK5bmQBQNnuGwHBfUamAV1ys9049oEPHnn4pcsc= -cloud.google.com/go/storage v1.52.0 h1:ROpzMW/IwipKtatA69ikxibdzQSiXJrY9f6IgBa9AlA= -cloud.google.com/go/storage v1.52.0/go.mod h1:4wrBAbAYUvYkbrf19ahGm4I5kDQhESSqN3CGEkMGvOY= +cloud.google.com/go/storage v1.54.0 h1:Du3XEyliAiftfyW0bwfdppm2MMLdpVAfiIg4T2nAI+0= +cloud.google.com/go/storage v1.54.0/go.mod h1:hIi9Boe8cHxTyaeqh7KMMwKg088VblFK46C2x/BWaZE= cloud.google.com/go/trace v1.11.3 h1:c+I4YFjxRQjvAhRmSsmjpASUKq88chOX854ied0K/pE= cloud.google.com/go/trace v1.11.3/go.mod h1:pt7zCYiDSQjC9Y2oqCsh9jF4GStB/hmjrYLsxRR27q8= github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.27.0 h1:ErKg/3iS1AKcTkf3yixlZ54f9U1rljCkQyEXWUnIUxc= @@ -119,10 +119,10 @@ google.golang.org/api v0.232.0 h1:qGnmaIMf7KcuwHOlF3mERVzChloDYwRfOJOrHt8YC3I= google.golang.org/api v0.232.0/go.mod h1:p9QCfBWZk1IJETUdbTKloR5ToFdKbYh2fkjsUL6vNoY= google.golang.org/genproto v0.0.0-20250303144028-a0af3efb3deb h1:ITgPrl429bc6+2ZraNSzMDk3I95nmQln2fuPstKwFDE= google.golang.org/genproto v0.0.0-20250303144028-a0af3efb3deb/go.mod h1:sAo5UzpjUwgFBCzupwhcLcxHVDK7vG5IqI30YnwX2eE= -google.golang.org/genproto/googleapis/api v0.0.0-20250414145226-207652e42e2e h1:UdXH7Kzbj+Vzastr5nVfccbmFsmYNygVLSPk1pEfDoY= -google.golang.org/genproto/googleapis/api v0.0.0-20250414145226-207652e42e2e/go.mod h1:085qFyf2+XaZlRdCgKNCIZ3afY2p4HHZdoIRpId8F4A= -google.golang.org/genproto/googleapis/rpc v0.0.0-20250428153025-10db94c68c34 h1:h6p3mQqrmT1XkHVTfzLdNz1u7IhINeZkz67/xTbOuWs= -google.golang.org/genproto/googleapis/rpc v0.0.0-20250428153025-10db94c68c34/go.mod h1:qQ0YXyHHx3XkvlzUtpXDkS29lDSafHMZBAZDc03LQ3A= +google.golang.org/genproto/googleapis/api v0.0.0-20250505200425-f936aa4a68b2 h1:vPV0tzlsK6EzEDHNNH5sa7Hs9bd7iXR7B1tSiPepkV0= +google.golang.org/genproto/googleapis/api v0.0.0-20250505200425-f936aa4a68b2/go.mod h1:pKLAc5OolXC3ViWGI62vvC0n10CpwAtRcTNCFwTKBEw= +google.golang.org/genproto/googleapis/rpc v0.0.0-20250505200425-f936aa4a68b2 h1:IqsN8hx+lWLqlN+Sc3DoMy/watjofWiU8sRFgQ8fhKM= +google.golang.org/genproto/googleapis/rpc v0.0.0-20250505200425-f936aa4a68b2/go.mod h1:qQ0YXyHHx3XkvlzUtpXDkS29lDSafHMZBAZDc03LQ3A= google.golang.org/grpc v1.72.0 h1:S7UkcVa60b5AAQTaO6ZKamFp1zMZSU0fGDK2WZLbBnM= google.golang.org/grpc v1.72.0/go.mod h1:wH5Aktxcg25y1I3w7H69nHfXdOG3UiadoBtjh3izSDM= google.golang.org/protobuf v1.36.6 h1:z1NpPI8ku2WgiWnf+t9wTPsn6eP1L7ksHUlkfLvd9xY= diff --git a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/iam/CHANGES.md b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/iam/CHANGES.md index 1c6498d56..7839f3b89 100644 --- a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/iam/CHANGES.md +++ b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/iam/CHANGES.md @@ -1,6 +1,22 @@ # Changes +## [1.5.2](https://github.com/googleapis/google-cloud-go/compare/iam/v1.5.1...iam/v1.5.2) (2025-04-15) + + +### Bug Fixes + +* **iam:** Update google.golang.org/api to 0.229.0 ([3319672](https://github.com/googleapis/google-cloud-go/commit/3319672f3dba84a7150772ccb5433e02dab7e201)) + +## [1.5.1](https://github.com/googleapis/google-cloud-go/compare/iam/v1.5.0...iam/v1.5.1) (2025-04-15) + + +### Documentation + +* **iam:** Formatting update for ListPolicyBindingsRequest ([dfdf404](https://github.com/googleapis/google-cloud-go/commit/dfdf404138728724aa6305c5c465ecc6fe5b1264)) +* **iam:** Minor doc update for ListPrincipalAccessBoundaryPoliciesResponse ([20f762c](https://github.com/googleapis/google-cloud-go/commit/20f762c528726a3f038d3e1f37e8a4952118badf)) +* **iam:** Minor doc update for ListPrincipalAccessBoundaryPoliciesResponse ([20f762c](https://github.com/googleapis/google-cloud-go/commit/20f762c528726a3f038d3e1f37e8a4952118badf)) + ## [1.5.0](https://github.com/googleapis/google-cloud-go/compare/iam/v1.4.2...iam/v1.5.0) (2025-03-31) diff --git a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/internal/.repo-metadata-full.json b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/internal/.repo-metadata-full.json index d72e82329..9dfae7977 100644 --- a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/internal/.repo-metadata-full.json +++ b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/internal/.repo-metadata-full.json @@ -1069,16 +1069,6 @@ "release_level": "preview", "library_type": "GAPIC_AUTO" }, - "cloud.google.com/go/debugger/apiv2": { - "api_shortname": "clouddebugger", - "distribution_name": "cloud.google.com/go/debugger/apiv2", - "description": "Stackdriver Debugger API", - "language": "go", - "client_library_type": "generated", - "client_documentation": "https://cloud.google.com/go/docs/reference/cloud.google.com/go/latest/debugger/apiv2", - "release_level": "preview", - "library_type": "GAPIC_AUTO" - }, "cloud.google.com/go/deploy/apiv1": { "api_shortname": "clouddeploy", "distribution_name": "cloud.google.com/go/deploy/apiv1", @@ -1099,6 +1089,16 @@ "release_level": "preview", "library_type": "GAPIC_AUTO" }, + "cloud.google.com/go/devicestreaming/apiv1": { + "api_shortname": "devicestreaming", + "distribution_name": "cloud.google.com/go/devicestreaming/apiv1", + "description": "Device Streaming API", + "language": "go", + "client_library_type": "generated", + "client_documentation": "https://cloud.google.com/go/docs/reference/cloud.google.com/go/devicestreaming/latest/apiv1", + "release_level": "preview", + "library_type": "GAPIC_AUTO" + }, "cloud.google.com/go/dialogflow/apiv2": { "api_shortname": "dialogflow", "distribution_name": "cloud.google.com/go/dialogflow/apiv2", @@ -1459,6 +1459,26 @@ "release_level": "stable", "library_type": "GAPIC_AUTO" }, + "cloud.google.com/go/iam/apiv3": { + "api_shortname": "iam", + "distribution_name": "cloud.google.com/go/iam/apiv3", + "description": "Identity and Access Management (IAM) API", + "language": "go", + "client_library_type": "generated", + "client_documentation": "https://cloud.google.com/go/docs/reference/cloud.google.com/go/iam/latest/apiv3", + "release_level": "preview", + "library_type": "GAPIC_AUTO" + }, + "cloud.google.com/go/iam/apiv3beta": { + "api_shortname": "iam", + "distribution_name": "cloud.google.com/go/iam/apiv3beta", + "description": "Identity and Access Management (IAM) API", + "language": "go", + "client_library_type": "generated", + "client_documentation": "https://cloud.google.com/go/docs/reference/cloud.google.com/go/iam/latest/apiv3beta", + "release_level": "preview", + "library_type": "GAPIC_AUTO" + }, "cloud.google.com/go/iam/credentials/apiv1": { "api_shortname": "iamcredentials", "distribution_name": "cloud.google.com/go/iam/credentials/apiv1", @@ -1599,6 +1619,16 @@ "release_level": "stable", "library_type": "GAPIC_AUTO" }, + "cloud.google.com/go/lustre/apiv1": { + "api_shortname": "lustre", + "distribution_name": "cloud.google.com/go/lustre/apiv1", + "description": "Google Cloud Managed Lustre API", + "language": "go", + "client_library_type": "generated", + "client_documentation": "https://cloud.google.com/go/docs/reference/cloud.google.com/go/lustre/latest/apiv1", + "release_level": "preview", + "library_type": "GAPIC_AUTO" + }, "cloud.google.com/go/managedidentities/apiv1": { "api_shortname": "managedidentities", "distribution_name": "cloud.google.com/go/managedidentities/apiv1", @@ -1632,7 +1662,7 @@ "cloud.google.com/go/maps/areainsights/apiv1": { "api_shortname": "areainsights", "distribution_name": "cloud.google.com/go/maps/areainsights/apiv1", - "description": "Places Insights API", + "description": "Places Aggregate API", "language": "go", "client_library_type": "generated", "client_documentation": "https://cloud.google.com/go/docs/reference/cloud.google.com/go/maps/latest/areainsights/apiv1", @@ -2139,13 +2169,13 @@ "release_level": "stable", "library_type": "GAPIC_MANUAL" }, - "cloud.google.com/go/pubsub/apiv1": { + "cloud.google.com/go/pubsub/v2/apiv1": { "api_shortname": "pubsub", - "distribution_name": "cloud.google.com/go/pubsub/apiv1", + "distribution_name": "cloud.google.com/go/pubsub/v2/apiv1", "description": "Cloud Pub/Sub API", "language": "go", "client_library_type": "generated", - "client_documentation": "https://cloud.google.com/go/docs/reference/cloud.google.com/go/pubsub/latest/apiv1", + "client_documentation": "https://cloud.google.com/go/docs/reference/cloud.google.com/go/pubsub/latest/v2/apiv1", "release_level": "stable", "library_type": "GAPIC_AUTO" }, @@ -2282,7 +2312,7 @@ "cloud.google.com/go/retail/apiv2": { "api_shortname": "retail", "distribution_name": "cloud.google.com/go/retail/apiv2", - "description": "Vertex AI Search for Retail API", + "description": "Vertex AI Search for commerce API", "language": "go", "client_library_type": "generated", "client_documentation": "https://cloud.google.com/go/docs/reference/cloud.google.com/go/retail/latest/apiv2", @@ -2292,7 +2322,7 @@ "cloud.google.com/go/retail/apiv2alpha": { "api_shortname": "retail", "distribution_name": "cloud.google.com/go/retail/apiv2alpha", - "description": "Vertex AI Search for Retail API", + "description": "Vertex AI Search for commerce API", "language": "go", "client_library_type": "generated", "client_documentation": "https://cloud.google.com/go/docs/reference/cloud.google.com/go/retail/latest/apiv2alpha", @@ -2302,7 +2332,7 @@ "cloud.google.com/go/retail/apiv2beta": { "api_shortname": "retail", "distribution_name": "cloud.google.com/go/retail/apiv2beta", - "description": "Vertex AI Search for Retail API", + "description": "Vertex AI Search for commerce API", "language": "go", "client_library_type": "generated", "client_documentation": "https://cloud.google.com/go/docs/reference/cloud.google.com/go/retail/latest/apiv2beta", @@ -2599,6 +2629,16 @@ "release_level": "preview", "library_type": "GAPIC_AUTO" }, + "cloud.google.com/go/shopping/merchant/issueresolution/apiv1beta": { + "api_shortname": "merchantapi", + "distribution_name": "cloud.google.com/go/shopping/merchant/issueresolution/apiv1beta", + "description": "Merchant API", + "language": "go", + "client_library_type": "generated", + "client_documentation": "https://cloud.google.com/go/docs/reference/cloud.google.com/go/shopping/latest/merchant/issueresolution/apiv1beta", + "release_level": "preview", + "library_type": "GAPIC_AUTO" + }, "cloud.google.com/go/shopping/merchant/lfp/apiv1beta": { "api_shortname": "merchantapi", "distribution_name": "cloud.google.com/go/shopping/merchant/lfp/apiv1beta", @@ -2619,6 +2659,16 @@ "release_level": "preview", "library_type": "GAPIC_AUTO" }, + "cloud.google.com/go/shopping/merchant/ordertracking/apiv1beta": { + "api_shortname": "merchantapi", + "distribution_name": "cloud.google.com/go/shopping/merchant/ordertracking/apiv1beta", + "description": "Merchant API", + "language": "go", + "client_library_type": "generated", + "client_documentation": "https://cloud.google.com/go/docs/reference/cloud.google.com/go/shopping/latest/merchant/ordertracking/apiv1beta", + "release_level": "preview", + "library_type": "GAPIC_AUTO" + }, "cloud.google.com/go/shopping/merchant/products/apiv1beta": { "api_shortname": "merchantapi", "distribution_name": "cloud.google.com/go/shopping/merchant/products/apiv1beta", @@ -2679,6 +2729,16 @@ "release_level": "stable", "library_type": "GAPIC_MANUAL" }, + "cloud.google.com/go/spanner/adapter/apiv1": { + "api_shortname": "spanner", + "distribution_name": "cloud.google.com/go/spanner/adapter/apiv1", + "description": "Cloud Spanner API", + "language": "go", + "client_library_type": "generated", + "client_documentation": "https://cloud.google.com/go/docs/reference/cloud.google.com/go/spanner/latest/adapter/apiv1", + "release_level": "preview", + "library_type": "GAPIC_AUTO" + }, "cloud.google.com/go/spanner/admin/database/apiv1": { "api_shortname": "spanner", "distribution_name": "cloud.google.com/go/spanner/admin/database/apiv1", diff --git a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/CHANGES.md b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/CHANGES.md index ca5cbb7c2..a571246f0 100644 --- a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/CHANGES.md +++ b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/CHANGES.md @@ -1,6 +1,40 @@ # Changes +## [1.54.0](https://github.com/googleapis/google-cloud-go/compare/storage/v1.53.0...storage/v1.54.0) (2025-05-12) + + +### Features + +* **storage:** Add experimental ZB API option ([#12214](https://github.com/googleapis/google-cloud-go/issues/12214)) ([f669982](https://github.com/googleapis/google-cloud-go/commit/f669982de2abf64759eccf5c38bd669488b9cf6a)) + + +### Bug Fixes + +* **storage:** Fix append writer hang ([#12201](https://github.com/googleapis/google-cloud-go/issues/12201)) ([7ce2a2a](https://github.com/googleapis/google-cloud-go/commit/7ce2a2ad3ae9deff28c73c1bcc2e7001770464eb)) +* **storage:** Retry unwrapped EOFs ([#12202](https://github.com/googleapis/google-cloud-go/issues/12202)) ([b2d42bd](https://github.com/googleapis/google-cloud-go/commit/b2d42bda6a398f3aa00030b6e99bbcb40f132ff7)) + +## [1.53.0](https://github.com/googleapis/google-cloud-go/compare/storage/v1.52.0...storage/v1.53.0) (2025-05-02) + + +### Features + +* **storage/control:** Add Anywhere cache control APIs ([83ae06c](https://github.com/googleapis/google-cloud-go/commit/83ae06c3ec7d190e38856ba4cfd8a13f08356b4d)) + + +### Bug Fixes + +* **storage:** Fix append edge cases ([#12074](https://github.com/googleapis/google-cloud-go/issues/12074)) ([0eee1f9](https://github.com/googleapis/google-cloud-go/commit/0eee1f99a7dc0d1bfc36fa43d78933cae47962ee)) +* **storage:** Fix retries for redirection errors. ([#12093](https://github.com/googleapis/google-cloud-go/issues/12093)) ([3e177e7](https://github.com/googleapis/google-cloud-go/commit/3e177e755f5bf6aa96e8712cc4adcba7eb6f04f6)) +* **storage:** Handle gRPC deadlines in tests. ([#12092](https://github.com/googleapis/google-cloud-go/issues/12092)) ([30b7cd2](https://github.com/googleapis/google-cloud-go/commit/30b7cd27771ccbd49b70ee106da36362ba8f1e87)) +* **storage:** Update offset on resumable upload retry ([#12086](https://github.com/googleapis/google-cloud-go/issues/12086)) ([6ce8fe5](https://github.com/googleapis/google-cloud-go/commit/6ce8fe5aec0ec7916eda4d1405cab5f5f65a5de8)) +* **storage:** Validate Bidi option for MRD ([#12033](https://github.com/googleapis/google-cloud-go/issues/12033)) ([d9018cf](https://github.com/googleapis/google-cloud-go/commit/d9018cf640a9ac25e2b23b75b3bcfa734379ab09)) + + +### Documentation + +* **storage/control:** Added comments for Anywhere cache messages ([83ae06c](https://github.com/googleapis/google-cloud-go/commit/83ae06c3ec7d190e38856ba4cfd8a13f08356b4d)) + ## [1.52.0](https://github.com/googleapis/google-cloud-go/compare/storage/v1.51.0...storage/v1.52.0) (2025-04-22) diff --git a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/doc.go b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/doc.go index 4fcfb7326..ec801fb54 100644 --- a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/doc.go +++ b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/doc.go @@ -333,12 +333,13 @@ to add a [custom audit logging] header: # gRPC API -This package includes support for the Cloud Storage gRPC API. The -implementation uses gRPC rather than the Default -JSON & XML APIs to make requests to Cloud Storage. -The Go Storage gRPC client is generally available. -The Notifications, Serivce Account HMAC -and GetServiceAccount RPCs are not supported through the gRPC client. +This package includes support for the [Cloud Storage gRPC API]. This +implementation uses gRPC rather than the default JSON & XML APIs +to make requests to Cloud Storage. All methods on the [Client] support +the gRPC API, with the exception of [GetServiceAccount], [Notification], +and [HMACKey] methods. + +The Cloud Storage gRPC API is generally available. To create a client which will use gRPC, use the alternate constructor: @@ -349,43 +350,31 @@ To create a client which will use gRPC, use the alternate constructor: } // Use client as usual. -Using the gRPC API inside GCP with a bucket in the same region can allow for -[Direct Connectivity] (enabling requests to skip some proxy steps and reducing -response latency). A warning is emmitted if gRPC is not used within GCP to -warn that Direct Connectivity could not be initialized. Direct Connectivity -is not required to access the gRPC API. +One major advantage of the gRPC API is that it can use [Direct Connectivity], +enabling requests to skip some proxy steps and reducing responce latency. +Requirements to use Direct Connectivity include: + + - Your application must be running inside Google Cloud. + - Your Cloud Storage [bucket location] must overlap with your VM or compute + environment zone. For example, if your VM is in us-east1a, your bucket + must be located in either us-east1 (single region), nam4 (dual region), + or us (multi-region). + - Your client must use service account authentication. + +Additional requirements for Direct Connectivity are documented in the +[Cloud Storage gRPC docs]. Dependencies for the gRPC API may slightly increase the size of binaries for applications depending on this package. If you are not using gRPC, you can use the build tag `disable_grpc_modules` to opt out of these dependencies and reduce the binary size. -The gRPC client emits metrics by default and will export the -gRPC telemetry discussed in [gRFC/66] and [gRFC/78] to -[Google Cloud Monitoring]. The metrics are accessible through Cloud Monitoring -API and you incur no additional cost for publishing the metrics. Google Cloud -Support can use this information to more quickly diagnose problems related to -GCS and gRPC. -Sending this data does not incur any billing charges, and requires minimal -CPU (a single RPC every minute) or memory (a few KiB to batch the -telemetry). - -To access the metrics you can view them through Cloud Monitoring -[metric explorer] with the prefix `storage.googleapis.com/client`. Metrics are emitted -every minute. - -You can disable metrics using the following example when creating a new gRPC -client using [WithDisabledClientMetrics]. - -The metrics exporter uses Cloud Monitoring API which determines -project ID and credentials doing the following: - -* Project ID is determined using OTel Resource Detector for the environment -otherwise it falls back to the project provided by [google.FindCredentials]. - -* Credentials are determined using [Application Default Credentials]. The -principal must have `roles/monitoring.metricWriter` role granted. If not a -logged warning will be emitted. Subsequent are silenced to prevent noisy logs. +The gRPC client is instrumented with Open Telemetry metrics which export to +Cloud Monitoring by default. More information is available in the +[gRPC client-side metrics] documentation, including information about +roles which must be enabled in order to do the export successfully. To +disable this export, you can use the [WithDisabledClientMetrics] client +option. # Storage Control API @@ -394,11 +383,6 @@ and Managed Folder operations) are supported via the autogenerated Storage Contr client, which is available as a subpackage in this module. See package docs at [cloud.google.com/go/storage/control/apiv2] or reference the [Storage Control API] docs. -[Application Default Credentials]: https://cloud.google.com/docs/authentication/application-default-credentials -[google.FindCredentials]: https://pkg.go.dev/golang.org/x/oauth2/google#FindDefaultCredentials -[gRFC/66]: https://github.com/grpc/proposal/blob/master/A66-otel-stats.md -[gRFC/78]: https://github.com/grpc/proposal/blob/master/A78-grpc-metrics-wrr-pf-xds.md -[Google Cloud Monitoring]: https://cloud.google.com/monitoring/docs [Cloud Storage IAM docs]: https://cloud.google.com/storage/docs/access-control/iam [XML POST Object docs]: https://cloud.google.com/storage/docs/xml-api/post-object [Cloud Storage retry docs]: https://cloud.google.com/storage/docs/retry-strategy @@ -408,7 +392,10 @@ client, which is available as a subpackage in this module. See package docs at [IAM Service Account Credentials API]: https://console.developers.google.com/apis/api/iamcredentials.googleapis.com/overview [custom audit logging]: https://cloud.google.com/storage/docs/audit-logging#add-custom-metadata [Storage Control API]: https://cloud.google.com/storage/docs/reference/rpc/google.storage.control.v2 -[metric explorer]: https://console.cloud.google.com/projectselector/monitoring/metrics-explorer +[Cloud Storage gRPC API]: https://cloud.google.com/storage/docs/enable-grpc-api [Direct Connectivity]: https://cloud.google.com/vpc-service-controls/docs/set-up-private-connectivity#direct-connectivity +[bucket location]: https://cloud.google.com/storage/docs/locations +[Cloud Storage gRPC docs]: https://cloud.google.com/storage/docs/enable-grpc-api#limitations +[gRPC client-side metrics]: https://cloud.google.com/storage/docs/client-side-metrics */ package storage // import "cloud.google.com/go/storage" diff --git a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/emulator_test.sh b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/emulator_test.sh index cf26432e0..4d8da2eeb 100644 --- a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/emulator_test.sh +++ b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/emulator_test.sh @@ -89,4 +89,5 @@ then fi # Run tests -go test -v -timeout 15m ./ ./dataflux -run="^Test(RetryConformance|.*Emulated)$" -short -race 2>&1 | tee -a sponge_log.log + +go test -v -timeout 17m ./ ./dataflux -run="^Test(RetryConformance|.*Emulated)$" -short -race 2>&1 | tee -a sponge_log.log diff --git a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/experimental/experimental.go b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/experimental/experimental.go index 5bcc59ad2..a178f9ebb 100644 --- a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/experimental/experimental.go +++ b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/experimental/experimental.go @@ -77,7 +77,8 @@ type ReadStallTimeoutConfig struct { // WithGRPCBidiReads provides an [option.ClientOption] that may be passed to // [cloud.google.com/go/storage.NewGRPCClient]. // It enables the client to use bi-directional gRPC APIs for downloads rather than the -// server streaming API. In particular, it allows users to use the [storage.MultiRangeDownloader] +// server streaming API. In particular, it allows users to use the +// [cloud.google.com/go/storage.MultiRangeDownloader] // surface, which requires bi-directional streaming. // // The bi-directional API is in private preview; please contact your account manager if @@ -85,3 +86,16 @@ type ReadStallTimeoutConfig struct { func WithGRPCBidiReads() option.ClientOption { return internal.WithGRPCBidiReads.(func() option.ClientOption)() } + +// WithZonalBucketAPIs provides an [option.ClientOption] that may be passed to +// [cloud.google.com/go/storage.NewGRPCClient]. +// It enables the client to use bi-directional gRPC APIs for downloads rather than the +// server streaming API (same as [WithGRPCBidiReads]) as well as appendable +// object semantics for uploads. By setting this option, both upload and download +// paths will use zonal bucket compatible APIs by default. +// +// Zonal buckets and rapid storage is in private preview; please contact your +// account manager if interested. +func WithZonalBucketAPIs() option.ClientOption { + return internal.WithZonalBucketAPIs.(func() option.ClientOption)() +} diff --git a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/grpc_client.go b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/grpc_client.go index 1dd564cda..84b9d4e9a 100644 --- a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/grpc_client.go +++ b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/grpc_client.go @@ -22,7 +22,6 @@ import ( "hash/crc32" "io" "log" - "net/url" "os" "sync" @@ -31,7 +30,6 @@ import ( gapic "cloud.google.com/go/storage/internal/apiv2" "cloud.google.com/go/storage/internal/apiv2/storagepb" "github.com/googleapis/gax-go/v2" - "google.golang.org/api/googleapi" "google.golang.org/api/iterator" "google.golang.org/api/option" "google.golang.org/api/option/internaloption" @@ -57,12 +55,6 @@ const ( // This is only used for the gRPC client. defaultConnPoolSize = 1 - // maxPerMessageWriteSize is the maximum amount of content that can be sent - // per WriteObjectRequest message. A buffer reaching this amount will - // precipitate a flush of the buffer. It is only used by the gRPC Writer - // implementation. - maxPerMessageWriteSize int = int(storagepb.ServiceConstants_MAX_WRITE_CHUNK_BYTES) - // globalProjectAlias is the project ID alias used for global buckets. // // This is only used for the gRPC API. @@ -140,7 +132,7 @@ func enableClientMetrics(ctx context.Context, s *settings, config storageConfig) // newGRPCStorageClient initializes a new storageClient that uses the gRPC // Storage API. -func newGRPCStorageClient(ctx context.Context, opts ...storageOption) (storageClient, error) { +func newGRPCStorageClient(ctx context.Context, opts ...storageOption) (*grpcStorageClient, error) { s := initSettings(opts...) s.clientOption = append(defaultGRPCOptions(), s.clientOption...) // Disable all gax-level retries in favor of retry logic in the veneer client. @@ -460,6 +452,9 @@ func (c *grpcStorageClient) ListObjects(ctx context.Context, bucket string, q *Q ctx = setUserProjectMetadata(ctx, s.userProject) } fetch := func(pageSize int, pageToken string) (token string, err error) { + // Add trace span around List API call within the fetch. + ctx, _ = startSpan(ctx, "grpcStorageClient.ObjectsListCall") + defer func() { endSpan(ctx, err) }() var objects []*storagepb.Object var gitr *gapic.ObjectIterator err = run(it.ctx, func(ctx context.Context) error { @@ -1063,6 +1058,10 @@ func contextMetadataFromBidiReadObject(req *storagepb.BidiReadObjectRequest) []s } func (c *grpcStorageClient) NewMultiRangeDownloader(ctx context.Context, params *newMultiRangeDownloaderParams, opts ...storageOption) (mr *MultiRangeDownloader, err error) { + if !c.config.grpcBidiReads { + return nil, errors.New("storage: MultiRangeDownloader requires the experimental.WithGRPCBidiReads option") + } + ctx = trace.StartSpan(ctx, "cloud.google.com/go/storage.grpcStorageClient.NewMultiRangeDownloader") defer func() { trace.EndSpan(ctx, err) }() s := callSettings(c.settings, opts...) @@ -1423,11 +1422,7 @@ func (mrd *gRPCBidiReader) activeRange() []mrdRange { // retryStream cancel's stream and reopen the stream again. func (mrd *gRPCBidiReader) retryStream(err error) error { - var shouldRetry = ShouldRetry - if mrd.settings.retry != nil && mrd.settings.retry.shouldRetry != nil { - shouldRetry = mrd.settings.retry.shouldRetry - } - if shouldRetry(err) { + if mrd.settings.retry.runShouldRetry(err) { // This will "close" the existing stream and immediately attempt to // reopen the stream, but will backoff if further attempts are necessary. // When Reopening the stream only failed readID will be added to stream. @@ -1736,116 +1731,6 @@ func (c *grpcStorageClient) NewRangeReader(ctx context.Context, params *newRange return r, nil } -func (c *grpcStorageClient) OpenWriter(params *openWriterParams, opts ...storageOption) (*io.PipeWriter, error) { - var offset int64 - errorf := params.setError - setObj := params.setObj - setFlush := params.setFlush - pr, pw := io.Pipe() - - s := callSettings(c.settings, opts...) - - retryDeadline := defaultWriteChunkRetryDeadline - if params.chunkRetryDeadline != 0 { - retryDeadline = params.chunkRetryDeadline - } - if s.retry == nil { - s.retry = defaultRetry.clone() - } - s.retry.maxRetryDuration = retryDeadline - - // Set Flush func for use by exported Writer.Flush. - var gw *gRPCWriter - setFlush(func() (int64, error) { - return gw.flush() - }) - gw, err := newGRPCWriter(c, s, params, pr, pw, params.setPipeWriter) - if err != nil { - errorf(err) - pr.CloseWithError(err) - close(params.donec) - return nil, err - } - - var o *storagepb.Object - - // If we are taking over an appendable object, send the first message here - // to get the append offset. - if params.appendGen > 0 { - // Create the buffer sender. This opens a stream and blocks until we - // get a response that tells us what offset to write from. - wbs, err := gw.newGRPCAppendTakeoverWriteBufferSender(params.ctx) - if err != nil { - return nil, fmt.Errorf("storage: creating buffer sender: %w", err) - } - // Propagate append offset to caller and buffer sending logic below. - params.setTakeoverOffset(wbs.takeoverOffset) - offset = wbs.takeoverOffset - gw.streamSender = wbs - o = wbs.objResource - setObj(newObjectFromProto(o)) - } - - // This function reads the data sent to the pipe and sends sets of messages - // on the gRPC client-stream as the buffer is filled. - go func() { - err := func() error { - // Unless the user told us the content type, we have to determine it from - // the first read. - if params.attrs.ContentType == "" && !params.forceEmptyContentType { - gw.reader, gw.spec.Resource.ContentType = gax.DetermineContentType(gw.reader) - } - - // Loop until there is an error or the Object has been finalized. - for { - // Note: This blocks until either the buffer is full or EOF is read. - recvd, doneReading, err := gw.read() - if err != nil { - return err - } - - uploadBuff := func(ctx context.Context) error { - obj, err := gw.uploadBuffer(ctx, recvd, offset, doneReading) - if obj != nil { - o = obj - setObj(newObjectFromProto(o)) - } - return err - } - - // Add routing headers to the context metadata for single-shot and resumable - // writes. Append writes need to set this at a lower level to pass the routing - // token. - bctx := gw.ctx - if !gw.append { - bctx = bucketContext(bctx, gw.bucket) - } - err = run(bctx, uploadBuff, gw.settings.retry, s.idempotent) - if err != nil { - return err - } - offset += int64(recvd) - - // When we are done reading data without errors, set the object and - // finish. - if doneReading { - // Build Object from server's response. - setObj(newObjectFromProto(o)) - return nil - } - } - }() - - // These calls are still valid if err is nil - err = checkCanceled(err) - errorf(err) - pr.CloseWithError(err) - close(params.donec) - }() - - return pw, nil -} - // IAM methods. func (c *grpcStorageClient) GetIamPolicy(ctx context.Context, resource string, version int32, opts ...storageOption) (*iampb.Policy, error) { @@ -2146,11 +2031,7 @@ func (r *gRPCReader) Close() error { func (r *gRPCReader) recv() error { databufs := mem.BufferSlice{} err := r.stream.RecvMsg(&databufs) - var shouldRetry = ShouldRetry - if r.settings.retry != nil && r.settings.retry.shouldRetry != nil { - shouldRetry = r.settings.retry.shouldRetry - } - if err != nil && shouldRetry(err) { + if err != nil && r.settings.retry.runShouldRetry(err) { // This will "close" the existing stream and immediately attempt to // reopen the stream, but will backoff if further attempts are necessary. // Reopening the stream Recvs the first message, so if retrying is @@ -2613,467 +2494,3 @@ func (r *gRPCReader) reopenStream() error { r.cancel = cancel return nil } - -func newGRPCWriter(c *grpcStorageClient, s *settings, params *openWriterParams, r io.Reader, pw *io.PipeWriter, setPipeWriter func(*io.PipeWriter)) (*gRPCWriter, error) { - if params.attrs.Retention != nil { - // TO-DO: remove once ObjectRetention is available - see b/308194853 - return nil, status.Errorf(codes.Unimplemented, "storage: object retention is not supported in gRPC") - } - - size := googleapi.MinUploadChunkSize - // A completely bufferless upload (params.chunkSize <= 0) is not possible in - // gRPC because the buffer must be provided to the message. Use the minimum - // size possible. - if params.chunkSize > 0 { - size = params.chunkSize - } - - // Round up chunksize to nearest 256KiB - if size%googleapi.MinUploadChunkSize != 0 { - size += googleapi.MinUploadChunkSize - (size % googleapi.MinUploadChunkSize) - } - - if s.userProject != "" { - params.ctx = setUserProjectMetadata(params.ctx, s.userProject) - } - - spec := &storagepb.WriteObjectSpec{ - Resource: params.attrs.toProtoObject(params.bucket), - Appendable: proto.Bool(params.append), - } - var appendSpec *storagepb.AppendObjectSpec - if params.appendGen > 0 { - appendSpec = &storagepb.AppendObjectSpec{ - Bucket: bucketResourceName(globalProjectAlias, params.bucket), - Object: params.attrs.Name, - Generation: params.appendGen, - } - } - // WriteObject doesn't support the generation condition, so use default. - if err := applyCondsProto("WriteObject", defaultGen, params.conds, spec); err != nil { - return nil, err - } - - return &gRPCWriter{ - buf: make([]byte, size), - c: c, - ctx: params.ctx, - reader: r, - pw: pw, - bucket: params.bucket, - attrs: params.attrs, - conds: params.conds, - spec: spec, - appendSpec: appendSpec, - encryptionKey: params.encryptionKey, - settings: s, - progress: params.progress, - setSize: params.setSize, - sendCRC32C: params.sendCRC32C, - forceOneShot: params.chunkSize <= 0, - forceEmptyContentType: params.forceEmptyContentType, - append: params.append, - finalizeOnClose: params.finalizeOnClose, - setPipeWriter: setPipeWriter, - flushComplete: make(chan int64), - }, nil -} - -// gRPCWriter is a wrapper around the the gRPC client-stream API that manages -// sending chunks of data provided by the user over the stream. -type gRPCWriter struct { - c *grpcStorageClient - buf []byte - reader io.Reader - pw *io.PipeWriter - setPipeWriter func(*io.PipeWriter) // used to set in parent storage.Writer - - ctx context.Context - - bucket string - attrs *ObjectAttrs - conds *Conditions - spec *storagepb.WriteObjectSpec - appendSpec *storagepb.AppendObjectSpec - encryptionKey []byte - settings *settings - progress func(int64) - setSize func(int64) - - sendCRC32C bool - forceOneShot bool - forceEmptyContentType bool - append bool - finalizeOnClose bool - - streamSender gRPCBidiWriteBufferSender - flushInProgress bool // true when the pipe is being recreated for a flush. - flushComplete chan int64 // use to signal back to flush call that flush to server was completed. -} - -func bucketContext(ctx context.Context, bucket string) context.Context { - hds := []string{"x-goog-request-params", fmt.Sprintf("bucket=projects/_/buckets/%s", url.QueryEscape(bucket))} - return gax.InsertMetadataIntoOutgoingContext(ctx, hds...) -} - -// drainInboundStream calls stream.Recv() repeatedly until an error is returned. -// It returns the last Resource received on the stream, or nil if no Resource -// was returned. drainInboundStream always returns a non-nil error. io.EOF -// indicates all messages were successfully read. -func drainInboundStream(stream storagepb.Storage_BidiWriteObjectClient) (object *storagepb.Object, err error) { - for err == nil { - var resp *storagepb.BidiWriteObjectResponse - resp, err = stream.Recv() - // GetResource() returns nil on a nil response - if resp.GetResource() != nil { - object = resp.GetResource() - } - } - return object, err -} - -func bidiWriteObjectRequest(buf []byte, offset int64, flush, finishWrite bool) *storagepb.BidiWriteObjectRequest { - var data *storagepb.BidiWriteObjectRequest_ChecksummedData - if buf != nil { - data = &storagepb.BidiWriteObjectRequest_ChecksummedData{ - ChecksummedData: &storagepb.ChecksummedData{ - Content: buf, - }, - } - } - req := &storagepb.BidiWriteObjectRequest{ - Data: data, - WriteOffset: offset, - FinishWrite: finishWrite, - Flush: flush, - StateLookup: flush, - } - return req -} - -type gRPCBidiWriteBufferSender interface { - // sendBuffer implementations should upload buf, respecting flush and - // finishWrite. Callers must guarantee that buf is not too long to fit in a - // gRPC message. - // - // If flush is true, implementations must not return until the data in buf is - // stable. If finishWrite is true, implementations must return the object on - // success. - sendBuffer(ctx context.Context, buf []byte, offset int64, flush, finishWrite bool) (*storagepb.Object, error) -} - -type gRPCOneshotBidiWriteBufferSender struct { - firstMessage *storagepb.BidiWriteObjectRequest - raw *gapic.Client - stream storagepb.Storage_BidiWriteObjectClient - settings *settings -} - -func (w *gRPCWriter) newGRPCOneshotBidiWriteBufferSender() (*gRPCOneshotBidiWriteBufferSender, error) { - firstMessage := &storagepb.BidiWriteObjectRequest{ - FirstMessage: &storagepb.BidiWriteObjectRequest_WriteObjectSpec{ - WriteObjectSpec: w.spec, - }, - CommonObjectRequestParams: toProtoCommonObjectRequestParams(w.encryptionKey), - // For a non-resumable upload, checksums must be sent in this message. - // TODO: Currently the checksums are only sent on the first message - // of the stream, but in the future, we must also support sending it - // on the *last* message of the stream (instead of the first). - ObjectChecksums: toProtoChecksums(w.sendCRC32C, w.attrs), - } - - return &gRPCOneshotBidiWriteBufferSender{ - firstMessage: firstMessage, - raw: w.c.raw, - settings: w.settings, - }, nil -} - -func (s *gRPCOneshotBidiWriteBufferSender) sendBuffer(ctx context.Context, buf []byte, offset int64, flush, finishWrite bool) (obj *storagepb.Object, err error) { - var firstMessage *storagepb.BidiWriteObjectRequest - if s.stream == nil { - s.stream, err = s.raw.BidiWriteObject(ctx, s.settings.gax...) - if err != nil { - return - } - firstMessage = s.firstMessage - } - req := bidiWriteObjectRequest(buf, offset, flush, finishWrite) - if firstMessage != nil { - proto.Merge(req, firstMessage) - } - - sendErr := s.stream.Send(req) - if sendErr != nil { - obj, err = drainInboundStream(s.stream) - s.stream = nil - if sendErr != io.EOF { - err = sendErr - } - return - } - // Oneshot uploads assume all flushes succeed - - if finishWrite { - s.stream.CloseSend() - // Oneshot uploads only read from the response stream on completion or - // failure - obj, err = drainInboundStream(s.stream) - s.stream = nil - if err == io.EOF { - err = nil - } - } - return -} - -type gRPCResumableBidiWriteBufferSender struct { - queryRetry *retryConfig - upid string - progress func(int64) - raw *gapic.Client - forceFirstMessage bool - stream storagepb.Storage_BidiWriteObjectClient - flushOffset int64 - settings *settings -} - -func (w *gRPCWriter) newGRPCResumableBidiWriteBufferSender(ctx context.Context) (*gRPCResumableBidiWriteBufferSender, error) { - req := &storagepb.StartResumableWriteRequest{ - WriteObjectSpec: w.spec, - CommonObjectRequestParams: toProtoCommonObjectRequestParams(w.encryptionKey), - // TODO: Currently the checksums are only sent on the request to initialize - // the upload, but in the future, we must also support sending it - // on the *last* message of the stream. - ObjectChecksums: toProtoChecksums(w.sendCRC32C, w.attrs), - } - - var upid string - err := run(ctx, func(ctx context.Context) error { - upres, err := w.c.raw.StartResumableWrite(ctx, req, w.settings.gax...) - upid = upres.GetUploadId() - return err - }, w.settings.retry, w.settings.idempotent) - if err != nil { - return nil, err - } - - // Set up an initial connection for the 0 offset, so we don't query state - // unnecessarily for the first buffer. If we fail, we'll just retry in the - // normal connect path. - stream, err := w.c.raw.BidiWriteObject(ctx, w.settings.gax...) - if err != nil { - stream = nil - } - - return &gRPCResumableBidiWriteBufferSender{ - queryRetry: w.settings.retry, - upid: upid, - progress: w.progress, - raw: w.c.raw, - forceFirstMessage: true, - stream: stream, - settings: w.settings, - }, nil -} - -// queryProgress is a helper that queries the status of the resumable upload -// associated with the given upload ID. -func (s *gRPCResumableBidiWriteBufferSender) queryProgress(ctx context.Context) (int64, error) { - var persistedSize int64 - err := run(ctx, func(ctx context.Context) error { - q, err := s.raw.QueryWriteStatus(ctx, &storagepb.QueryWriteStatusRequest{ - UploadId: s.upid, - }, s.settings.gax...) - // q.GetPersistedSize() will return 0 if q is nil. - persistedSize = q.GetPersistedSize() - return err - }, s.queryRetry, true) - - return persistedSize, err -} - -func (s *gRPCResumableBidiWriteBufferSender) sendBuffer(ctx context.Context, buf []byte, offset int64, flush, finishWrite bool) (obj *storagepb.Object, err error) { - if s.stream == nil { - // Determine offset and reconnect - s.flushOffset, err = s.queryProgress(ctx) - if err != nil { - return - } - s.stream, err = s.raw.BidiWriteObject(ctx, s.settings.gax...) - if err != nil { - return - } - s.forceFirstMessage = true - } - - // clean up buf. We'll still write the message if a flush/finishWrite was - // requested. - if offset < s.flushOffset { - trim := s.flushOffset - offset - if int64(len(buf)) <= trim { - trim = int64(len(buf)) - } - buf = buf[trim:] - } - if len(buf) == 0 && !flush && !finishWrite { - // no need to send anything - return nil, nil - } - - req := bidiWriteObjectRequest(buf, offset, flush, finishWrite) - if s.forceFirstMessage { - req.FirstMessage = &storagepb.BidiWriteObjectRequest_UploadId{UploadId: s.upid} - s.forceFirstMessage = false - } - - sendErr := s.stream.Send(req) - if sendErr != nil { - obj, err = drainInboundStream(s.stream) - s.stream = nil - if err == io.EOF { - // This is unexpected - we got an error on Send(), but not on Recv(). - // Bubble up the sendErr. - err = sendErr - } - return - } - - if finishWrite { - s.stream.CloseSend() - obj, err = drainInboundStream(s.stream) - s.stream = nil - if err == io.EOF { - err = nil - if obj.GetSize() > s.flushOffset { - s.progress(obj.GetSize()) - } - } - return - } - - if flush { - resp, err := s.stream.Recv() - if err != nil { - return nil, err - } - persistedOffset := resp.GetPersistedSize() - if persistedOffset > s.flushOffset { - s.flushOffset = persistedOffset - s.progress(s.flushOffset) - } - } - return -} - -// uploadBuffer uploads the buffer at the given offset using a bi-directional -// Write stream. It will open a new stream if necessary (on the first call or -// after resuming from failure) and chunk the buffer per maxPerMessageWriteSize. -// The final Object is returned on success if doneReading is true. -// -// Returns object and any error that is not retriable. -func (w *gRPCWriter) uploadBuffer(ctx context.Context, recvd int, start int64, doneReading bool) (obj *storagepb.Object, err error) { - if w.streamSender == nil { - if w.append { - // Appendable object semantics - w.streamSender, err = w.newGRPCAppendableObjectBufferSender() - } else if doneReading || w.forceOneShot { - // One shot semantics - w.streamSender, err = w.newGRPCOneshotBidiWriteBufferSender() - } else { - // Resumable write semantics - w.streamSender, err = w.newGRPCResumableBidiWriteBufferSender(ctx) - } - if err != nil { - return - } - } - - data := w.buf[:recvd] - offset := start - // We want to go through this loop at least once, in case we have to - // finishWrite with an empty buffer. - for { - // Send as much as we can fit into a single gRPC message. Only flush once, - // when sending the very last message. - l := maxPerMessageWriteSize - flush := false - if len(data) <= l { - l = len(data) - flush = true - } - obj, err = w.streamSender.sendBuffer(ctx, data[:l], offset, flush, flush && doneReading) - if err != nil { - return nil, err - } - data = data[l:] - offset += int64(l) - if len(data) == 0 { - // Update object size to match persisted offset. - if obj != nil { - obj.Size = offset - } - break - } - } - if w.flushInProgress { - w.setSize(offset) - w.flushInProgress = false - w.flushComplete <- offset - } - return -} - -// read copies the data in the reader to the given buffer and reports how much -// data was read into the buffer and if there is no more data to read (EOF). -// read returns when either 1. the buffer is full, 2. Writer.Flush was called, -// or 3. Writer.Close was called. -func (w *gRPCWriter) read() (int, bool, error) { - // Set n to -1 to start the Read loop. - var n, recvd int = -1, 0 - var err error - for err == nil && n != 0 { - // The routine blocks here until data is received. - n, err = w.reader.Read(w.buf[recvd:]) - recvd += n - } - var done bool - if err == io.EOF { - err = nil - // EOF can come from Writer.Flush or Writer.Close. - if w.flushInProgress { - // Reset pipe for additional writes after the flush. - pr, pw := io.Pipe() - w.reader = pr - w.pw = pw - w.setPipeWriter(pw) - } else { - done = true - } - } - return recvd, done, err -} - -// flush flushes the current buffer regardless of whether it is full or not. -// It's the implementation for Writer.Flush. -func (w *gRPCWriter) flush() (int64, error) { - if !w.append { - return 0, errors.New("Flush is supported only if Writer.Append is set to true") - } - - // Close PipeWriter to trigger EOF on read side of the stream. - w.flushInProgress = true - w.pw.Close() - - // Wait for flush to complete - offset := <-w.flushComplete - return offset, nil -} - -func checkCanceled(err error) error { - if status.Code(err) == codes.Canceled { - return context.Canceled - } - - return err -} diff --git a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/grpc_reader.go b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/grpc_reader.go index e1aecad65..099e6a615 100644 --- a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/grpc_reader.go +++ b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/grpc_reader.go @@ -424,11 +424,7 @@ func (r *gRPCReadObjectReader) recv() error { databufs := mem.BufferSlice{} err := r.stream.RecvMsg(&databufs) - var shouldRetry = ShouldRetry - if r.settings.retry != nil && r.settings.retry.shouldRetry != nil { - shouldRetry = r.settings.retry.shouldRetry - } - if err != nil && shouldRetry(err) { + if err != nil && r.settings.retry.runShouldRetry(err) { // This will "close" the existing stream and immediately attempt to // reopen the stream, but will backoff if further attempts are necessary. // Reopening the stream Recvs the first message, so if retrying is diff --git a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/grpc_writer.go b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/grpc_writer.go index 83d2c9e3e..37aceb05e 100644 --- a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/grpc_writer.go +++ b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/grpc_writer.go @@ -19,17 +19,649 @@ import ( "errors" "fmt" "io" + "net/url" "time" gapic "cloud.google.com/go/storage/internal/apiv2" "cloud.google.com/go/storage/internal/apiv2/storagepb" gax "github.com/googleapis/gax-go/v2" + "google.golang.org/api/googleapi" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" "google.golang.org/protobuf/proto" ) -const defaultWriteChunkRetryDeadline = 32 * time.Second +const ( + // defaultWriteChunkRetryDeadline is the default deadline for the upload + // of a single chunk. It can be overwritten by Writer.ChunkRetryDeadline. + defaultWriteChunkRetryDeadline = 32 * time.Second + // maxPerMessageWriteSize is the maximum amount of content that can be sent + // per WriteObjectRequest message. A buffer reaching this amount will + // precipitate a flush of the buffer. It is only used by the gRPC Writer + // implementation. + maxPerMessageWriteSize int = int(storagepb.ServiceConstants_MAX_WRITE_CHUNK_BYTES) +) + +func withBidiWriteObjectRedirectionErrorRetries(s *settings) (newr *retryConfig) { + oldr := s.retry + newr = oldr.clone() + if newr == nil { + newr = &retryConfig{} + } + if (oldr.policy == RetryIdempotent && !s.idempotent) || oldr.policy == RetryNever { + // We still retry redirection errors even when settings indicate not to + // retry. + // + // The protocol requires us to respect redirection errors, so RetryNever has + // to ignore them. + // + // Idempotency is always protected by redirection errors: they either + // contain a handle which can be used as idempotency information, or they do + // not contain a handle and are "affirmative failures" which indicate that + // no server-side action occurred. + newr.policy = RetryAlways + newr.shouldRetry = func(err error) bool { + return errors.Is(err, bidiWriteObjectRedirectionError{}) + } + return newr + } + // If retry settings allow retries normally, fall back to that behavior. + newr.shouldRetry = func(err error) bool { + if errors.Is(err, bidiWriteObjectRedirectionError{}) { + return true + } + v := oldr.runShouldRetry(err) + return v + } + return newr +} + +func (c *grpcStorageClient) OpenWriter(params *openWriterParams, opts ...storageOption) (*io.PipeWriter, error) { + var offset int64 + errorf := params.setError + setObj := params.setObj + setFlush := params.setFlush + pr, pw := io.Pipe() + + s := callSettings(c.settings, opts...) + + retryDeadline := defaultWriteChunkRetryDeadline + if params.chunkRetryDeadline != 0 { + retryDeadline = params.chunkRetryDeadline + } + if s.retry == nil { + s.retry = defaultRetry.clone() + } + if params.append { + s.retry = withBidiWriteObjectRedirectionErrorRetries(s) + } + s.retry.maxRetryDuration = retryDeadline + + // Set Flush func for use by exported Writer.Flush. + var gw *gRPCWriter + setFlush(func() (int64, error) { + return gw.flush() + }) + gw, err := newGRPCWriter(c, s, params, pr, pr, pw, params.setPipeWriter) + if err != nil { + errorf(err) + pr.CloseWithError(err) + close(params.donec) + return nil, err + } + + var o *storagepb.Object + + // If we are taking over an appendable object, send the first message here + // to get the append offset. + if params.appendGen > 0 { + // Create the buffer sender. This opens a stream and blocks until we + // get a response that tells us what offset to write from. + wbs, err := gw.newGRPCAppendTakeoverWriteBufferSender(params.ctx) + if err != nil { + return nil, fmt.Errorf("storage: creating buffer sender: %w", err) + } + // Propagate append offset to caller and buffer sending logic below. + params.setTakeoverOffset(wbs.takeoverOffset) + offset = wbs.takeoverOffset + gw.streamSender = wbs + o = wbs.objResource + setObj(newObjectFromProto(o)) + } + + // This function reads the data sent to the pipe and sends sets of messages + // on the gRPC client-stream as the buffer is filled. + go func() { + err := func() error { + // Unless the user told us the content type, we have to determine it from + // the first read. + if params.attrs.ContentType == "" && !params.forceEmptyContentType { + gw.reader, gw.spec.Resource.ContentType = gax.DetermineContentType(gw.reader) + } + + // Loop until there is an error or the Object has been finalized. + for { + // Note: This blocks until either the buffer is full or EOF is read. + recvd, doneReading, err := gw.read() + if err != nil { + return err + } + + uploadBuff := func(ctx context.Context) error { + obj, err := gw.uploadBuffer(ctx, recvd, offset, doneReading) + if obj != nil { + o = obj + setObj(newObjectFromProto(o)) + } + return err + } + + // Add routing headers to the context metadata for single-shot and resumable + // writes. Append writes need to set this at a lower level to pass the routing + // token. + bctx := gw.ctx + if !gw.append { + bctx = bucketContext(bctx, gw.bucket) + } + err = run(bctx, uploadBuff, gw.settings.retry, s.idempotent) + offset += int64(recvd) + // If this buffer upload was triggered by a flush, reset and + // communicate back the result. + if gw.flushInProgress { + gw.setSize(offset) + gw.flushInProgress = false + gw.flushComplete <- flushResult{offset: offset, err: err} + } + if err != nil { + return err + } + // When we are done reading data without errors, set the object and + // finish. + if doneReading { + // Build Object from server's response. + setObj(newObjectFromProto(o)) + return nil + } + } + }() + + // These calls are still valid if err is nil + err = checkCanceled(err) + errorf(err) + gw.pr.CloseWithError(err) + close(params.donec) + }() + + return pw, nil +} + +func newGRPCWriter(c *grpcStorageClient, s *settings, params *openWriterParams, r io.Reader, pr *io.PipeReader, pw *io.PipeWriter, setPipeWriter func(*io.PipeWriter)) (*gRPCWriter, error) { + if params.attrs.Retention != nil { + // TO-DO: remove once ObjectRetention is available - see b/308194853 + return nil, status.Errorf(codes.Unimplemented, "storage: object retention is not supported in gRPC") + } + + size := googleapi.MinUploadChunkSize + // A completely bufferless upload (params.chunkSize <= 0) is not possible in + // gRPC because the buffer must be provided to the message. Use the minimum + // size possible. + if params.chunkSize > 0 { + size = params.chunkSize + } + + // Round up chunksize to nearest 256KiB + if size%googleapi.MinUploadChunkSize != 0 { + size += googleapi.MinUploadChunkSize - (size % googleapi.MinUploadChunkSize) + } + + if s.userProject != "" { + params.ctx = setUserProjectMetadata(params.ctx, s.userProject) + } + + spec := &storagepb.WriteObjectSpec{ + Resource: params.attrs.toProtoObject(params.bucket), + Appendable: proto.Bool(params.append), + } + var appendSpec *storagepb.AppendObjectSpec + if params.appendGen > 0 { + appendSpec = &storagepb.AppendObjectSpec{ + Bucket: bucketResourceName(globalProjectAlias, params.bucket), + Object: params.attrs.Name, + Generation: params.appendGen, + } + } + // WriteObject doesn't support the generation condition, so use default. + if err := applyCondsProto("WriteObject", defaultGen, params.conds, spec); err != nil { + return nil, err + } + + return &gRPCWriter{ + buf: make([]byte, size), + c: c, + ctx: params.ctx, + reader: r, + pw: pw, + pr: pr, + bucket: params.bucket, + attrs: params.attrs, + conds: params.conds, + spec: spec, + appendSpec: appendSpec, + encryptionKey: params.encryptionKey, + settings: s, + progress: params.progress, + setSize: params.setSize, + sendCRC32C: params.sendCRC32C, + forceOneShot: params.chunkSize <= 0, + forceEmptyContentType: params.forceEmptyContentType, + append: params.append, + finalizeOnClose: params.finalizeOnClose, + setPipeWriter: setPipeWriter, + flushComplete: make(chan flushResult), + }, nil +} + +// gRPCWriter is a wrapper around the the gRPC client-stream API that manages +// sending chunks of data provided by the user over the stream. +type gRPCWriter struct { + c *grpcStorageClient + buf []byte + reader io.Reader + pr *io.PipeReader // Keep track of pr and pw to update post-flush + pw *io.PipeWriter + setPipeWriter func(*io.PipeWriter) // used to set in parent storage.Writer + + ctx context.Context + + bucket string + attrs *ObjectAttrs + conds *Conditions + spec *storagepb.WriteObjectSpec + appendSpec *storagepb.AppendObjectSpec + encryptionKey []byte + settings *settings + progress func(int64) + setSize func(int64) + + sendCRC32C bool + forceOneShot bool + forceEmptyContentType bool + append bool + finalizeOnClose bool + + streamSender gRPCBidiWriteBufferSender + flushInProgress bool // true when the pipe is being recreated for a flush. + flushComplete chan flushResult // use to signal back to flush call that flush to server was completed. +} + +type flushResult struct { + err error + offset int64 +} + +func bucketContext(ctx context.Context, bucket string) context.Context { + hds := []string{"x-goog-request-params", fmt.Sprintf("bucket=projects/_/buckets/%s", url.QueryEscape(bucket))} + return gax.InsertMetadataIntoOutgoingContext(ctx, hds...) +} + +// drainInboundStream calls stream.Recv() repeatedly until an error is returned. +// It returns the last Resource received on the stream, or nil if no Resource +// was returned. drainInboundStream always returns a non-nil error. io.EOF +// indicates all messages were successfully read. +func drainInboundStream(stream storagepb.Storage_BidiWriteObjectClient) (object *storagepb.Object, err error) { + for err == nil { + var resp *storagepb.BidiWriteObjectResponse + resp, err = stream.Recv() + // GetResource() returns nil on a nil response + if resp.GetResource() != nil { + object = resp.GetResource() + } + } + return object, err +} + +func bidiWriteObjectRequest(buf []byte, offset int64, flush, finishWrite bool) *storagepb.BidiWriteObjectRequest { + var data *storagepb.BidiWriteObjectRequest_ChecksummedData + if buf != nil { + data = &storagepb.BidiWriteObjectRequest_ChecksummedData{ + ChecksummedData: &storagepb.ChecksummedData{ + Content: buf, + }, + } + } + req := &storagepb.BidiWriteObjectRequest{ + Data: data, + WriteOffset: offset, + FinishWrite: finishWrite, + Flush: flush, + StateLookup: flush, + } + return req +} + +type gRPCBidiWriteBufferSender interface { + // sendBuffer implementations should upload buf, respecting flush and + // finishWrite. Callers must guarantee that buf is not too long to fit in a + // gRPC message. + // + // If flush is true, implementations must not return until the data in buf is + // stable. If finishWrite is true, implementations must return the object on + // success. + sendBuffer(ctx context.Context, buf []byte, offset int64, flush, finishWrite bool) (*storagepb.Object, error) +} + +type gRPCOneshotBidiWriteBufferSender struct { + firstMessage *storagepb.BidiWriteObjectRequest + raw *gapic.Client + stream storagepb.Storage_BidiWriteObjectClient + settings *settings +} + +func (w *gRPCWriter) newGRPCOneshotBidiWriteBufferSender() (*gRPCOneshotBidiWriteBufferSender, error) { + firstMessage := &storagepb.BidiWriteObjectRequest{ + FirstMessage: &storagepb.BidiWriteObjectRequest_WriteObjectSpec{ + WriteObjectSpec: w.spec, + }, + CommonObjectRequestParams: toProtoCommonObjectRequestParams(w.encryptionKey), + // For a non-resumable upload, checksums must be sent in this message. + // TODO: Currently the checksums are only sent on the first message + // of the stream, but in the future, we must also support sending it + // on the *last* message of the stream (instead of the first). + ObjectChecksums: toProtoChecksums(w.sendCRC32C, w.attrs), + } + + return &gRPCOneshotBidiWriteBufferSender{ + firstMessage: firstMessage, + raw: w.c.raw, + settings: w.settings, + }, nil +} + +func (s *gRPCOneshotBidiWriteBufferSender) sendBuffer(ctx context.Context, buf []byte, offset int64, flush, finishWrite bool) (obj *storagepb.Object, err error) { + var firstMessage *storagepb.BidiWriteObjectRequest + if s.stream == nil { + s.stream, err = s.raw.BidiWriteObject(ctx, s.settings.gax...) + if err != nil { + return + } + firstMessage = s.firstMessage + } + req := bidiWriteObjectRequest(buf, offset, flush, finishWrite) + if firstMessage != nil { + proto.Merge(req, firstMessage) + } + + sendErr := s.stream.Send(req) + if sendErr != nil { + obj, err = drainInboundStream(s.stream) + s.stream = nil + if sendErr != io.EOF { + err = sendErr + } + return + } + // Oneshot uploads assume all flushes succeed + + if finishWrite { + s.stream.CloseSend() + // Oneshot uploads only read from the response stream on completion or + // failure + obj, err = drainInboundStream(s.stream) + s.stream = nil + if err == io.EOF { + err = nil + } + } + return +} + +type gRPCResumableBidiWriteBufferSender struct { + queryRetry *retryConfig + upid string + progress func(int64) + raw *gapic.Client + forceFirstMessage bool + stream storagepb.Storage_BidiWriteObjectClient + flushOffset int64 + settings *settings +} + +func (w *gRPCWriter) newGRPCResumableBidiWriteBufferSender(ctx context.Context) (*gRPCResumableBidiWriteBufferSender, error) { + req := &storagepb.StartResumableWriteRequest{ + WriteObjectSpec: w.spec, + CommonObjectRequestParams: toProtoCommonObjectRequestParams(w.encryptionKey), + // TODO: Currently the checksums are only sent on the request to initialize + // the upload, but in the future, we must also support sending it + // on the *last* message of the stream. + ObjectChecksums: toProtoChecksums(w.sendCRC32C, w.attrs), + } + + var upid string + err := run(ctx, func(ctx context.Context) error { + upres, err := w.c.raw.StartResumableWrite(ctx, req, w.settings.gax...) + upid = upres.GetUploadId() + return err + }, w.settings.retry, w.settings.idempotent) + if err != nil { + return nil, err + } + + // Set up an initial connection for the 0 offset, so we don't query state + // unnecessarily for the first buffer. If we fail, we'll just retry in the + // normal connect path. + stream, err := w.c.raw.BidiWriteObject(ctx, w.settings.gax...) + if err != nil { + stream = nil + } + + return &gRPCResumableBidiWriteBufferSender{ + queryRetry: w.settings.retry, + upid: upid, + progress: w.progress, + raw: w.c.raw, + forceFirstMessage: true, + stream: stream, + settings: w.settings, + }, nil +} + +// queryProgress is a helper that queries the status of the resumable upload +// associated with the given upload ID. +func (s *gRPCResumableBidiWriteBufferSender) queryProgress(ctx context.Context) (int64, error) { + var persistedSize int64 + err := run(ctx, func(ctx context.Context) error { + q, err := s.raw.QueryWriteStatus(ctx, &storagepb.QueryWriteStatusRequest{ + UploadId: s.upid, + }, s.settings.gax...) + // q.GetPersistedSize() will return 0 if q is nil. + persistedSize = q.GetPersistedSize() + return err + }, s.queryRetry, true) + + return persistedSize, err +} + +func (s *gRPCResumableBidiWriteBufferSender) sendBuffer(ctx context.Context, buf []byte, offset int64, flush, finishWrite bool) (obj *storagepb.Object, err error) { + if s.stream == nil { + // Determine offset and reconnect + s.flushOffset, err = s.queryProgress(ctx) + if err != nil { + return + } + s.stream, err = s.raw.BidiWriteObject(ctx, s.settings.gax...) + if err != nil { + return + } + s.forceFirstMessage = true + } + + // clean up buf. We'll still write the message if a flush/finishWrite was + // requested. + if offset < s.flushOffset { + trim := s.flushOffset - offset + if int64(len(buf)) <= trim { + trim = int64(len(buf)) + } + buf = buf[trim:] + offset += trim + } + if len(buf) == 0 && !flush && !finishWrite { + // no need to send anything + return nil, nil + } + + req := bidiWriteObjectRequest(buf, offset, flush, finishWrite) + if s.forceFirstMessage { + req.FirstMessage = &storagepb.BidiWriteObjectRequest_UploadId{UploadId: s.upid} + s.forceFirstMessage = false + } + + sendErr := s.stream.Send(req) + if sendErr != nil { + obj, err = drainInboundStream(s.stream) + s.stream = nil + if err == io.EOF { + // This is unexpected - we got an error on Send(), but not on Recv(). + // Bubble up the sendErr. + err = sendErr + } + return + } + + if finishWrite { + s.stream.CloseSend() + obj, err = drainInboundStream(s.stream) + s.stream = nil + if err == io.EOF { + err = nil + if obj.GetSize() > s.flushOffset { + s.progress(obj.GetSize()) + } + } + return + } + + if flush { + resp, err := s.stream.Recv() + if err != nil { + return nil, err + } + persistedOffset := resp.GetPersistedSize() + if persistedOffset > s.flushOffset { + s.flushOffset = persistedOffset + s.progress(s.flushOffset) + } + } + return +} + +// uploadBuffer uploads the buffer at the given offset using a bi-directional +// Write stream. It will open a new stream if necessary (on the first call or +// after resuming from failure) and chunk the buffer per maxPerMessageWriteSize. +// The final Object is returned on success if doneReading is true. +// +// Returns object and any error that is not retriable. +func (w *gRPCWriter) uploadBuffer(ctx context.Context, recvd int, start int64, doneReading bool) (obj *storagepb.Object, err error) { + if w.streamSender == nil { + if w.append { + // Appendable object semantics + w.streamSender, err = w.newGRPCAppendableObjectBufferSender() + } else if doneReading || w.forceOneShot { + // One shot semantics + w.streamSender, err = w.newGRPCOneshotBidiWriteBufferSender() + } else { + // Resumable write semantics + w.streamSender, err = w.newGRPCResumableBidiWriteBufferSender(ctx) + } + if err != nil { + return + } + } + + data := w.buf[:recvd] + offset := start + // We want to go through this loop at least once, in case we have to + // finishWrite with an empty buffer. + for { + // Send as much as we can fit into a single gRPC message. Only flush once, + // when sending the very last message. + l := maxPerMessageWriteSize + flush := false + if len(data) <= l { + l = len(data) + flush = true + } + obj, err = w.streamSender.sendBuffer(ctx, data[:l], offset, flush, flush && doneReading) + if err != nil { + return nil, err + } + data = data[l:] + offset += int64(l) + if len(data) == 0 { + // Update object size to match persisted offset. + if obj != nil { + obj.Size = offset + } + break + } + } + return +} + +// read copies the data in the reader to the given buffer and reports how much +// data was read into the buffer and if there is no more data to read (EOF). +// read returns when either 1. the buffer is full, 2. Writer.Flush was called, +// or 3. Writer.Close was called. +func (w *gRPCWriter) read() (int, bool, error) { + // Set n to -1 to start the Read loop. + var n, recvd int = -1, 0 + var err error + for err == nil && n != 0 { + // The routine blocks here until data is received. + n, err = w.reader.Read(w.buf[recvd:]) + recvd += n + } + var done bool + if err == io.EOF { + err = nil + // EOF can come from Writer.Flush or Writer.Close. + if w.flushInProgress { + // Reset pipe for additional writes after the flush. + pr, pw := io.Pipe() + w.reader = pr + w.pw = pw + w.pr = pr + w.setPipeWriter(pw) + } else { + done = true + } + } + return recvd, done, err +} + +// flush flushes the current buffer regardless of whether it is full or not. +// It's the implementation for Writer.Flush. +func (w *gRPCWriter) flush() (int64, error) { + if !w.append { + return 0, errors.New("Flush is supported only if Writer.Append is set to true") + } + + // Close PipeWriter to trigger EOF on read side of the stream. + w.flushInProgress = true + w.pw.Close() + + // Wait for flush to complete + result := <-w.flushComplete + return result.offset, result.err +} + +func checkCanceled(err error) error { + if status.Code(err) == codes.Canceled { + return context.Canceled + } + + return err +} type gRPCAppendBidiWriteBufferSender struct { bucket string @@ -60,7 +692,7 @@ func (w *gRPCWriter) newGRPCAppendableObjectBufferSender() (*gRPCAppendBidiWrite s := &gRPCAppendBidiWriteBufferSender{ bucket: w.spec.GetResource().GetBucket(), raw: w.c.raw, - settings: w.c.settings, + settings: w.settings, firstMessage: &storagepb.BidiWriteObjectRequest{ FirstMessage: &storagepb.BidiWriteObjectRequest_WriteObjectSpec{ WriteObjectSpec: w.spec, @@ -82,7 +714,7 @@ func (w *gRPCWriter) newGRPCAppendTakeoverWriteBufferSender(ctx context.Context) s := &gRPCAppendBidiWriteBufferSender{ bucket: w.spec.GetResource().GetBucket(), raw: w.c.raw, - settings: w.c.settings, + settings: w.settings, firstMessage: &storagepb.BidiWriteObjectRequest{ FirstMessage: &storagepb.BidiWriteObjectRequest_AppendObjectSpec{ AppendObjectSpec: w.appendSpec, @@ -101,6 +733,11 @@ func (w *gRPCWriter) newGRPCAppendTakeoverWriteBufferSender(ctx context.Context) return nil, err } firstResp := <-s.recvs + // Check recvErr after getting the response. + if s.recvErr != nil { + return nil, s.recvErr + } + // Object resource is returned in the first response on takeover, so capture // this now. s.objResource = firstResp.GetResource() @@ -126,6 +763,10 @@ func (s *gRPCAppendBidiWriteBufferSender) connect(ctx context.Context) (err erro if s.firstMessage.GetAppendObjectSpec().GetGeneration() != 0 { return nil } + // Also always ok to reconnect if we've seen a redirect token + if s.routingToken != nil { + return nil + } // We can also reconnect if the first message has an if_generation_match or // if_metageneration_match condition. Note that negative conditions like @@ -208,7 +849,7 @@ func (s *gRPCAppendBidiWriteBufferSender) maybeUpdateFirstMessage(resp *storagep type bidiWriteObjectRedirectionError struct{} func (e bidiWriteObjectRedirectionError) Error() string { - return "BidiWriteObjectRedirectedError" + return "" } func (s *gRPCAppendBidiWriteBufferSender) handleRedirectionError(e *storagepb.BidiWriteObjectRedirectedError) bool { @@ -253,10 +894,10 @@ func (s *gRPCAppendBidiWriteBufferSender) receiveMessages(resps chan<- *storagep if st, ok := status.FromError(err); ok && st.Code() == codes.Aborted { for _, d := range st.Details() { if e, ok := d.(*storagepb.BidiWriteObjectRedirectedError); ok { - // If we can handle this error, replace it with the sentinel. Otherwise, - // report it to the user. + // If we can handle this error, wrap it with the sentinel so it gets + // retried. if ok := s.handleRedirectionError(e); ok { - err = bidiWriteObjectRedirectionError{} + err = fmt.Errorf("%w%w", bidiWriteObjectRedirectionError{}, err) } } } @@ -374,12 +1015,6 @@ func (s *gRPCAppendBidiWriteBufferSender) sendBuffer(ctx context.Context, buf [] err = s.recvErr } s.stream = nil - - // Retry transparently on a redirection error - if _, ok := err.(bidiWriteObjectRedirectionError); ok { - s.forceFirstMessage = true - continue - } return } } diff --git a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/http_client.go b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/http_client.go index eafa1a3b1..656201c08 100644 --- a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/http_client.go +++ b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/http_client.go @@ -344,6 +344,10 @@ func (c *httpStorageClient) ListObjects(ctx context.Context, bucket string, q *Q it.query = *q } fetch := func(pageSize int, pageToken string) (string, error) { + var err error + // Add trace span around List API call within the fetch. + ctx, _ = startSpan(ctx, "httpStorageClient.ObjectsListCall") + defer func() { endSpan(ctx, err) }() req := c.raw.Objects.List(bucket) if it.query.SoftDeleted { req.SoftDeleted(it.query.SoftDeleted) @@ -372,7 +376,6 @@ func (c *httpStorageClient) ListObjects(ctx context.Context, bucket string, q *Q req.MaxResults(int64(pageSize)) } var resp *raw.Objects - var err error err = run(it.ctx, func(ctx context.Context) error { resp, err = req.Context(ctx).Do() return err diff --git a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/internal/experimental.go b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/internal/experimental.go index 2fd5111fb..4482eab8a 100644 --- a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/internal/experimental.go +++ b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/internal/experimental.go @@ -33,4 +33,9 @@ var ( // WithGRPCBidiReads is a function which is implemented by the storage package. // It sets the gRPC client to use the BidiReadObject API for downloads. WithGRPCBidiReads any // func() option.ClientOption + + // WithZonalBucketAPIs is a function which is implemented by the storage package. + // It sets the gRPC client to use the BidiReadObject API for downloads and + // appendable object semantics by default for uploads. + WithZonalBucketAPIs any // func() option.ClientOption ) diff --git a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/internal/version.go b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/internal/version.go index 7651272b2..63de91334 100644 --- a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/internal/version.go +++ b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/internal/version.go @@ -15,4 +15,4 @@ package internal // Version is the current tagged release of the library. -const Version = "1.52.0" +const Version = "1.54.0" diff --git a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/invoke.go b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/invoke.go index 34b676c5f..cf5b0e913 100644 --- a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/invoke.go +++ b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/invoke.go @@ -53,6 +53,13 @@ var ( }) ) +func (r *retryConfig) runShouldRetry(err error) bool { + if r == nil || r.shouldRetry == nil { + return ShouldRetry(err) + } + return r.shouldRetry(err) +} + // run determines whether a retry is necessary based on the config and // idempotency information. It then calls the function with or without retries // as appropriate, using the configured settings. @@ -73,10 +80,6 @@ func run(ctx context.Context, call func(ctx context.Context) error, retry *retry bo.Initial = retry.backoff.Initial bo.Max = retry.backoff.Max } - var errorFunc func(err error) bool = ShouldRetry - if retry.shouldRetry != nil { - errorFunc = retry.shouldRetry - } var quitAfterTimer *time.Timer if retry.maxRetryDuration != 0 { @@ -103,7 +106,7 @@ func run(ctx context.Context, call func(ctx context.Context) error, retry *retry return true, fmt.Errorf("storage: retry failed after %v attempts; last error: %w", *retry.maxAttempts, lastErr) } attempts++ - retryable := errorFunc(lastErr) + retryable := retry.runShouldRetry(lastErr) // Explicitly check context cancellation so that we can distinguish between a // DEADLINE_EXCEEDED error from the server and a user-set context deadline. // Unfortunately gRPC will codes.DeadlineExceeded (which may be retryable if it's @@ -167,6 +170,15 @@ func ShouldRetry(err error) bool { return true } } + // TODO: remove when https://github.com/golang/go/issues/53472 is resolved. + // We don't want to retry io.EOF errors, since these can indicate normal + // functioning terminations such as internally in the case of Reader and + // externally in the case of iterator methods. However, the linked bug + // requires us to retry EOFs that it causes. We can distinguish + // EOFs caused by the bug because they are not wrapped correctly. + if !errors.Is(err, io.EOF) && strings.Contains(err.Error(), "EOF") { + return true + } case *net.DNSError: if e.IsTemporary { return true diff --git a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/option.go b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/option.go index 16d57644a..6548cc18e 100644 --- a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/option.go +++ b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/option.go @@ -41,6 +41,7 @@ func init() { storageinternal.WithMetricInterval = withMetricInterval storageinternal.WithReadStallTimeout = withReadStallTimeout storageinternal.WithGRPCBidiReads = withGRPCBidiReads + storageinternal.WithZonalBucketAPIs = withZonalBucketAPIs } // getDynamicReadReqIncreaseRateFromEnv returns the value set in the env variable. @@ -83,6 +84,7 @@ type storageConfig struct { manualReader *metric.ManualReader readStallTimeoutConfig *experimental.ReadStallTimeoutConfig grpcBidiReads bool + grpcAppendableUploads bool } // newStorageConfig generates a new storageConfig with all the given @@ -254,3 +256,17 @@ type withGRPCBidiReadsConfig struct { func (w *withGRPCBidiReadsConfig) ApplyStorageOpt(config *storageConfig) { config.grpcBidiReads = true } + +func withZonalBucketAPIs() option.ClientOption { + return &withZonalBucketAPIsConfig{} +} + +type withZonalBucketAPIsConfig struct { + internaloption.EmbeddableAdapter +} + +func (w *withZonalBucketAPIsConfig) ApplyStorageOpt(config *storageConfig) { + // Use both appendable upload semantics and bidi reads. + config.grpcAppendableUploads = true + config.grpcBidiReads = true +} diff --git a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/reader.go b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/reader.go index c3109f8b1..b20601dda 100644 --- a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/reader.go +++ b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/reader.go @@ -161,7 +161,9 @@ func (o *ObjectHandle) NewRangeReader(ctx context.Context, offset, length int64) // Must be called on a gRPC client created using [NewGRPCClient]. // // This uses the gRPC-specific bi-directional read API, which is in private -// preview; please contact your account manager if interested. +// preview; please contact your account manager if interested. The option +// [experimental.WithGRPCBidiReads] or [experimental.WithZonalBucketAPIs] +// must be selected in order to use this API. func (o *ObjectHandle) NewMultiRangeDownloader(ctx context.Context) (mrd *MultiRangeDownloader, err error) { // This span covers the life of the reader. It is closed via the context // in Reader.Close. diff --git a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/storage.go b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/storage.go index 18f461802..aead2df8d 100644 --- a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/storage.go +++ b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/storage.go @@ -126,6 +126,9 @@ type Client struct { // tc is the transport-agnostic client implemented with either gRPC or HTTP. tc storageClient + + // Option to use gRRPC appendable upload API was set. + grpcAppendableUploads bool } // NewClient creates a new Google Cloud Storage client using the HTTP transport. @@ -238,8 +241,10 @@ func NewGRPCClient(ctx context.Context, opts ...option.ClientOption) (*Client, e if err != nil { return nil, err } - - return &Client{tc: tc}, nil + return &Client{ + tc: tc, + grpcAppendableUploads: tc.config.grpcAppendableUploads, + }, nil } // CheckDirectConnectivitySupported checks if gRPC direct connectivity @@ -1107,7 +1112,9 @@ type ObjectAttrsToUpdate struct { } // Delete deletes the single specified object. -func (o *ObjectHandle) Delete(ctx context.Context) error { +func (o *ObjectHandle) Delete(ctx context.Context) (err error) { + ctx, _ = startSpan(ctx, "Object.Delete") + defer func() { endSpan(ctx, err) }() if err := o.validate(); err != nil { return err } @@ -1238,6 +1245,7 @@ func (o *ObjectHandle) NewWriter(ctx context.Context) *Writer { donec: make(chan struct{}), ObjectAttrs: ObjectAttrs{Name: o.object}, ChunkSize: googleapi.DefaultUploadChunkSize, + Append: o.c.grpcAppendableUploads, } } @@ -1267,7 +1275,7 @@ func (o *ObjectHandle) NewWriter(ctx context.Context) *Writer { // This feature is in preview and is not yet available for general use. func (o *ObjectHandle) NewWriterFromAppendableObject(ctx context.Context, opts *AppendableWriterOpts) (*Writer, int64, error) { ctx = trace.StartSpan(ctx, "cloud.google.com/go/storage.Object.Writer") - if o.gen == 0 { + if o.gen < 0 { return nil, 0, errors.New("storage: ObjectHandle.Generation must be set to use NewWriterFromAppendableObject") } w := &Writer{ diff --git a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/trace.go b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/trace.go index 67858b086..0fb023259 100644 --- a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/trace.go +++ b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/trace.go @@ -76,18 +76,23 @@ func endSpan(ctx context.Context, err error) { } } -// getCommonTraceOptions includes the common attributes used for Cloud Trace adoption tracking. +// getCommonTraceOptions makes a SpanStartOption with common attributes. func getCommonTraceOptions() []trace.SpanStartOption { opts := []trace.SpanStartOption{ - trace.WithAttributes( - attribute.String("gcp.client.version", internal.Version), - attribute.String("gcp.client.repo", gcpClientRepo), - attribute.String("gcp.client.artifact", gcpClientArtifact), - ), + trace.WithAttributes(getCommonAttributes()...), } return opts } +// getCommonAttributes includes the common attributes used for Cloud Trace adoption tracking. +func getCommonAttributes() []attribute.KeyValue { + return []attribute.KeyValue{ + attribute.String("gcp.client.version", internal.Version), + attribute.String("gcp.client.repo", gcpClientRepo), + attribute.String("gcp.client.artifact", gcpClientArtifact), + } +} + func appendPackageName(spanName string) string { return fmt.Sprintf("%s.%s", gcpClientArtifact, spanName) } diff --git a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/writer.go b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/writer.go index d43f28d28..a1b0b73c0 100644 --- a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/writer.go +++ b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/writer.go @@ -111,6 +111,9 @@ type Writer struct { // when Writer.Close() is called; otherwise, the object is left unfinalized // and can be appended to later. // + // Defaults to false unless the experiemental WithZonalBucketAPIs option was + // set. + // // Append is only supported for gRPC. This feature is in preview and is not // yet available for general use. Append bool diff --git a/ci/resources/stemcell-version-bump/vendor/modules.txt b/ci/resources/stemcell-version-bump/vendor/modules.txt index da7797cf4..c8b5b9d37 100644 --- a/ci/resources/stemcell-version-bump/vendor/modules.txt +++ b/ci/resources/stemcell-version-bump/vendor/modules.txt @@ -1,7 +1,7 @@ # cel.dev/expr v0.20.0 ## explicit; go 1.21.1 cel.dev/expr -# cloud.google.com/go v0.120.0 +# cloud.google.com/go v0.121.0 ## explicit; go 1.23.0 cloud.google.com/go/internal cloud.google.com/go/internal/optional @@ -30,7 +30,7 @@ cloud.google.com/go/auth/oauth2adapt # cloud.google.com/go/compute/metadata v0.6.0 ## explicit; go 1.21 cloud.google.com/go/compute/metadata -# cloud.google.com/go/iam v1.5.0 +# cloud.google.com/go/iam v1.5.2 ## explicit; go 1.23.0 cloud.google.com/go/iam cloud.google.com/go/iam/apiv1/iampb @@ -39,7 +39,7 @@ cloud.google.com/go/iam/apiv1/iampb cloud.google.com/go/monitoring/apiv3/v2 cloud.google.com/go/monitoring/apiv3/v2/monitoringpb cloud.google.com/go/monitoring/internal -# cloud.google.com/go/storage v1.52.0 +# cloud.google.com/go/storage v1.54.0 ## explicit; go 1.23.0 cloud.google.com/go/storage cloud.google.com/go/storage/experimental @@ -329,7 +329,7 @@ google.golang.org/genproto/googleapis/type/calendarperiod google.golang.org/genproto/googleapis/type/date google.golang.org/genproto/googleapis/type/expr google.golang.org/genproto/googleapis/type/timeofday -# google.golang.org/genproto/googleapis/api v0.0.0-20250414145226-207652e42e2e +# google.golang.org/genproto/googleapis/api v0.0.0-20250505200425-f936aa4a68b2 ## explicit; go 1.23.0 google.golang.org/genproto/googleapis/api google.golang.org/genproto/googleapis/api/annotations @@ -338,7 +338,7 @@ google.golang.org/genproto/googleapis/api/expr/v1alpha1 google.golang.org/genproto/googleapis/api/label google.golang.org/genproto/googleapis/api/metric google.golang.org/genproto/googleapis/api/monitoredres -# google.golang.org/genproto/googleapis/rpc v0.0.0-20250428153025-10db94c68c34 +# google.golang.org/genproto/googleapis/rpc v0.0.0-20250505200425-f936aa4a68b2 ## explicit; go 1.23.0 google.golang.org/genproto/googleapis/rpc/code google.golang.org/genproto/googleapis/rpc/errdetails