You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by ke...@apache.org on 2026/10/16 14:33:51 UTC
(beam) branch master updated: Add stale GCS test bucket cleaner (#40386)
This is an automated email from the ASF dual-hosted git repository.
kennknowles pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new f3a5f4f1542 Add stale GCS test bucket cleaner (#40386)
f3a5f4f1542 is described below
commit f3a5f4f15425fb1979e8843ab089477f1274bf98
Author: Goutam Adwant <86...@users.noreply.github.com>
AuthorDate: Fri Oct 16 07:33:43 2026 -0700
Add stale GCS test bucket cleaner (#40386)
---
.test-infra/tools/README.md | 18 ++
.test-infra/tools/build.gradle | 10 +
.test-infra/tools/stale_gcs_buckets_cleaner.py | 112 ++++++++++++
.../tools/test_stale_gcs_buckets_cleaner.py | 202 +++++++++++++++++++++
4 files changed, 342 insertions(+)
diff --git a/.test-infra/tools/README.md b/.test-infra/tools/README.md
index 758c3885a81..f13b672ddde 100644
--- a/.test-infra/tools/README.md
+++ b/.test-infra/tools/README.md
@@ -22,6 +22,24 @@
## Tools
- [Python installer](#python-installer)
+- [Stale GCS test bucket cleaner](#stale-gcs-test-bucket-cleaner)
+
+### Stale GCS test bucket cleaner
+
+`stale_gcs_buckets_cleaner.py` deletes buckets in the `apache-beam-testing`
+project whose names match `apache-beam-temp-bucket-<UUID>` and are older than
+24 hours. These buckets are created by `GcsUtil` integration tests and can be
+left behind when a test is interrupted before teardown. The cleaner removes
+objects from eligible buckets before deleting them.
+
+The cleaner performs a dry run by default:
+
+```bash
+python3 stale_gcs_buckets_cleaner.py
+```
+
+The scheduled `Clean Up GCP Resources` workflow runs the cleaner through the
+`removeStaleGcsBuckets` Gradle task, which passes the explicit `--delete` flag.
### Python installer
diff --git a/.test-infra/tools/build.gradle b/.test-infra/tools/build.gradle
index 92b53f2fd56..8c61e357c04 100644
--- a/.test-infra/tools/build.gradle
+++ b/.test-infra/tools/build.gradle
@@ -40,6 +40,15 @@ task removeStaleSpannerResources(type: Exec) {
commandLine './stale_spanner_cleaner.sh'
}
+task testStaleGcsBucketsCleaner(type: Exec) {
+ commandLine 'bash', '-c', 'python3 -m pip install -r requirements.txt && python3 -m unittest test_stale_gcs_buckets_cleaner.py'
+}
+
+task removeStaleGcsBuckets(type: Exec) {
+ dependsOn tasks.testStaleGcsBucketsCleaner
+ commandLine 'python3', 'stale_gcs_buckets_cleaner.py', '--delete'
+}
+
task runStaleCleaner(type: Exec) {
commandLine 'bash', '-c', 'pip install -r requirements.txt && python3 stale_cleaner.py'
}
@@ -50,5 +59,6 @@ task cleanupOtherStaleResources {
finalizedBy tasks.removeStaleCbtInstances
finalizedBy tasks.removeStaleK8sWorkload
finalizedBy tasks.removeStaleSpannerResources
+ finalizedBy tasks.removeStaleGcsBuckets
finalizedBy tasks.runStaleCleaner
}
diff --git a/.test-infra/tools/stale_gcs_buckets_cleaner.py b/.test-infra/tools/stale_gcs_buckets_cleaner.py
new file mode 100644
index 00000000000..4d475018862
--- /dev/null
+++ b/.test-infra/tools/stale_gcs_buckets_cleaner.py
@@ -0,0 +1,112 @@
+#!/usr/bin/env python
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+"""Deletes stale GCS buckets left behind by GcsUtil integration tests."""
+
+import argparse
+import datetime
+import re
+
+from google.cloud import storage
+
+
+DEFAULT_PROJECT_ID = "apache-beam-testing"
+GCS_TEMP_BUCKET_PREFIX = "apache-beam-temp-bucket-"
+GCS_TEMP_BUCKET_PATTERN = re.compile(
+ rf"^{GCS_TEMP_BUCKET_PREFIX}"
+ r"[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-"
+ r"[0-9a-f]{12}$")
+DEFAULT_MAX_AGE = datetime.timedelta(hours=24)
+UTC = datetime.timezone.utc
+
+
+def _as_utc(value):
+ if value.tzinfo is None:
+ return value.replace(tzinfo=UTC)
+ return value.astimezone(UTC)
+
+
+def clean_stale_gcs_buckets(
+ client,
+ max_age=DEFAULT_MAX_AGE,
+ now=None,
+ dry_run=True,
+):
+ """Deletes test buckets whose creation time is older than ``max_age``.
+
+ Returns the names of buckets selected for deletion. If a deletion fails,
+ the cleaner attempts the remaining buckets before reporting the failures.
+ """
+ current_time = _as_utc(now or datetime.datetime.now(UTC))
+ selected = []
+ failures = []
+
+ buckets = client.list_buckets(
+ project=DEFAULT_PROJECT_ID, prefix=GCS_TEMP_BUCKET_PREFIX)
+ for bucket in buckets:
+ # Keep this check even though list_buckets filters server-side. It is
+ # the final guard before a destructive operation.
+ if not GCS_TEMP_BUCKET_PATTERN.fullmatch(bucket.name):
+ continue
+ if bucket.time_created is None:
+ print(f"Skipping {bucket.name}: creation time is unavailable")
+ continue
+ if current_time - _as_utc(bucket.time_created) <= max_age:
+ continue
+
+ selected.append(bucket.name)
+ if dry_run:
+ print(f"Dry run: would delete gs://{bucket.name}")
+ continue
+
+ print(f"Deleting stale test bucket gs://{bucket.name}")
+ try:
+ # Bucket.delete(force=True) refuses to empty more than 256 objects.
+ # Delete each listed generation, then delete the empty bucket.
+ for blob in bucket.list_blobs(versions=True):
+ blob.delete(if_generation_match=blob.generation)
+ bucket.delete()
+ except Exception as error:
+ failures.append(bucket.name)
+ print(f"Failed to delete gs://{bucket.name}: {error}")
+
+ if failures:
+ raise RuntimeError(f"Bucket failure count: {len(failures)}")
+
+ return selected
+
+
+def main():
+ parser = argparse.ArgumentParser(
+ description="Delete stale buckets created by GcsUtil integration tests.")
+ parser.add_argument(
+ "--delete",
+ action="store_true",
+ help="Delete eligible buckets instead of performing a dry run.",
+ )
+ args = parser.parse_args()
+
+ client = storage.Client(project=DEFAULT_PROJECT_ID)
+ selected = clean_stale_gcs_buckets(client, dry_run=not args.delete)
+ action = "Deleted" if args.delete else "Selected"
+ print(f"{action} {len(selected)} stale GCS test bucket(s)")
+
+
+if __name__ == "__main__":
+ main()
diff --git a/.test-infra/tools/test_stale_gcs_buckets_cleaner.py b/.test-infra/tools/test_stale_gcs_buckets_cleaner.py
new file mode 100644
index 00000000000..21cbaa1dd76
--- /dev/null
+++ b/.test-infra/tools/test_stale_gcs_buckets_cleaner.py
@@ -0,0 +1,202 @@
+#!/usr/bin/env python
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+import datetime
+import unittest
+from unittest import mock
+
+import stale_gcs_buckets_cleaner as cleaner
+
+
+UTC = datetime.timezone.utc
+NOW = datetime.datetime(2026, 9, 30, 12, tzinfo=UTC)
+STALE_BUCKET = "apache-beam-temp-bucket-123e4567-e89b-42d3-a456-426614174000"
+FRESH_BUCKET = "apache-beam-temp-bucket-123e4567-e89b-42d3-a456-426614174001"
+
+
+class FakeBlob:
+ def __init__(self, bucket, name, generation):
+ self.bucket = bucket
+ self.name = name
+ self.generation = generation
+
+ def delete(self, if_generation_match=None):
+ if if_generation_match != self.generation:
+ raise ValueError("blob generation precondition is required")
+ self.bucket.blobs.remove(self)
+
+
+class FakeBucket:
+ def __init__(self, name, created, blob_count=0):
+ self.name = name
+ self.time_created = created
+ self.blobs = [FakeBlob(self, f"blob-{i}", i + 1)
+ for i in range(blob_count)]
+ self.deleted = False
+ self.delete_error = None
+
+ def list_blobs(self, versions=False):
+ if not versions:
+ raise ValueError("all object generations must be listed")
+ return iter(list(self.blobs))
+
+ def delete(self):
+ if self.delete_error:
+ raise self.delete_error
+ if self.blobs:
+ raise ValueError("bucket is not empty")
+ self.deleted = True
+
+
+class FakeClient:
+ def __init__(self):
+ self.buckets = []
+
+ def list_buckets(self, project, prefix):
+ if project != "apache-beam-testing" or prefix != "apache-beam-temp-bucket-":
+ raise ValueError("unexpected bucket selection")
+ return iter(self.buckets)
+
+
+def bucket(name, created, blob_count=0):
+ return FakeBucket(name, created, blob_count)
+
+
+class StaleGcsBucketsCleanerTest(unittest.TestCase):
+ def setUp(self):
+ self.client = FakeClient()
+
+ def test_deletes_only_stale_test_buckets(self):
+ stale = bucket(
+ STALE_BUCKET, NOW - datetime.timedelta(hours=25))
+ fresh = bucket(
+ FRESH_BUCKET, NOW - datetime.timedelta(hours=23))
+ unrelated = bucket("other-bucket", NOW - datetime.timedelta(days=7))
+ missing_created = bucket(
+ "apache-beam-temp-bucket-123e4567-e89b-42d3-a456-426614174002",
+ None)
+ self.client.buckets = [
+ stale, fresh, unrelated, missing_created]
+
+ deleted = cleaner.clean_stale_gcs_buckets(
+ self.client, now=NOW, dry_run=False)
+
+ self.assertEqual(deleted, [stale.name])
+ self.assertTrue(stale.deleted)
+ self.assertFalse(fresh.deleted)
+ self.assertFalse(unrelated.deleted)
+ self.assertFalse(missing_created.deleted)
+
+ def test_preserves_bucket_at_age_threshold(self):
+ value = bucket(
+ STALE_BUCKET,
+ NOW - datetime.timedelta(hours=24))
+ self.client.buckets = [value]
+
+ deleted = cleaner.clean_stale_gcs_buckets(
+ self.client, now=NOW, dry_run=False)
+
+ self.assertEqual(deleted, [])
+ self.assertFalse(value.deleted)
+
+ def test_dry_run_does_not_delete_bucket(self):
+ value = bucket(
+ STALE_BUCKET, NOW - datetime.timedelta(days=2))
+ self.client.buckets = [value]
+
+ selected = cleaner.clean_stale_gcs_buckets(self.client, now=NOW)
+
+ self.assertEqual(selected, [value.name])
+ self.assertFalse(value.deleted)
+
+ def test_naive_creation_time_is_treated_as_utc(self):
+ value = bucket(
+ STALE_BUCKET,
+ datetime.datetime(2026, 9, 29, 11))
+ self.client.buckets = [value]
+
+ deleted = cleaner.clean_stale_gcs_buckets(
+ self.client, now=NOW, dry_run=False)
+
+ self.assertEqual(deleted, [value.name])
+ self.assertTrue(value.deleted)
+
+ def test_continues_after_delete_failure_and_reports_all_failures(self):
+ failed = bucket(
+ STALE_BUCKET, NOW - datetime.timedelta(days=2))
+ deleted = bucket(
+ FRESH_BUCKET, NOW - datetime.timedelta(days=2))
+ failed.delete_error = RuntimeError("permission denied")
+ self.client.buckets = [failed, deleted]
+
+ with self.assertRaisesRegex(
+ RuntimeError, "Bucket failure count: 1"):
+ cleaner.clean_stale_gcs_buckets(
+ self.client, now=NOW, dry_run=False)
+
+ self.assertFalse(failed.deleted)
+ self.assertTrue(deleted.deleted)
+
+ def test_deletes_bucket_with_more_than_256_objects(self):
+ value = bucket(STALE_BUCKET, NOW - datetime.timedelta(days=2), 300)
+ self.client.buckets = [value]
+
+ selected = cleaner.clean_stale_gcs_buckets(
+ self.client, now=NOW, dry_run=False)
+
+ self.assertEqual(selected, [value.name])
+ self.assertTrue(value.deleted)
+ self.assertEqual(value.blobs, [])
+
+ def test_preserves_prefixed_bucket_without_uuid_name(self):
+ value = bucket(
+ "apache-beam-temp-bucket-manual", NOW - datetime.timedelta(days=7))
+ self.client.buckets = [value]
+
+ deleted = cleaner.clean_stale_gcs_buckets(self.client, now=NOW)
+
+ self.assertEqual(deleted, [])
+ self.assertFalse(value.deleted)
+
+ @mock.patch("stale_gcs_buckets_cleaner.clean_stale_gcs_buckets")
+ @mock.patch("stale_gcs_buckets_cleaner.storage.Client")
+ def test_main_defaults_to_dry_run(self, client_class, clean):
+ clean.return_value = []
+
+ with mock.patch("sys.argv", ["stale_gcs_buckets_cleaner.py"]):
+ cleaner.main()
+
+ client_class.assert_called_once_with(project="apache-beam-testing")
+ clean.assert_called_once_with(client_class.return_value, dry_run=True)
+
+ @mock.patch("stale_gcs_buckets_cleaner.clean_stale_gcs_buckets")
+ @mock.patch("stale_gcs_buckets_cleaner.storage.Client")
+ def test_main_requires_delete_flag_for_deletion(self, client_class, clean):
+ clean.return_value = []
+
+ with mock.patch(
+ "sys.argv", ["stale_gcs_buckets_cleaner.py", "--delete"]):
+ cleaner.main()
+
+ clean.assert_called_once_with(client_class.return_value, dry_run=False)
+
+
+if __name__ == "__main__":
+ unittest.main()