From ba8061d25931d16b648170cd6183d67765d336a8 Mon Sep 17 00:00:00 2001 From: HansMarcus01 Date: Tue, 28 Jul 2026 19:15:50 -0600 Subject: [PATCH 1/3] Feat: A new prefix was added to the cleaner to clean up newly discovered orphaned subscriptions of taxirides topic --- .test-infra/tools/stale_cleaner.py | 13 +++++++++---- .test-infra/tools/test_stale_cleaner.py | 20 ++++++++++++++++---- 2 files changed, 25 insertions(+), 8 deletions(-) diff --git a/.test-infra/tools/stale_cleaner.py b/.test-infra/tools/stale_cleaner.py index 2c702cfa0fd8..ba0a94313cdb 100644 --- a/.test-infra/tools/stale_cleaner.py +++ b/.test-infra/tools/stale_cleaner.py @@ -337,8 +337,11 @@ def _active_resources(self) -> dict: subscription_name = subscription.name # Apply prefix filtering if prefixes are defined if not self.prefixes or any(subscription_name.startswith(f"{self.project_path}/subscriptions/{prefix}") for prefix in self.prefixes): - # Check if the subscription has a topic associated with it - if subscription.detached: + # Safe orphan detection: + # - Standard rule: detached subscriptions are eligible. + # - Taxi exception: subscriptions containing the NYC taxi prefix are eligible even if attached. + es_taxi = "taxirides-realtime_beam_" in subscription_name + if subscription.detached or es_taxi: d[subscription_name] = GoogleCloudResource(resource_name=subscription_name, clock=self.clock) return d @@ -416,8 +419,10 @@ def clean_pubsub_subscriptions(): project_id = DEFAULT_PROJECT_ID bucket_name = DEFAULT_BUCKET_NAME - # No prefixes are defined for subscriptions so we will delete all stale subscriptions - prefixes = [] + # Restrict subscription cleanup to the NYC taxi prefix only. + prefixes = [ + "taxirides-realtime_beam_", + ] # Create a PubSubSubscriptionCleaner instance cleaner = PubSubSubscriptionCleaner(project_id=project_id, bucket_name=bucket_name, diff --git a/.test-infra/tools/test_stale_cleaner.py b/.test-infra/tools/test_stale_cleaner.py index c53fbc1a44db..29691d8448b8 100644 --- a/.test-infra/tools/test_stale_cleaner.py +++ b/.test-infra/tools/test_stale_cleaner.py @@ -433,26 +433,38 @@ def test_init(self): def test_active_resources(self): """Test _active_resources method.""" + # Keep the existing standard prefix coverage and add the taxi-only cleanup prefix. + self.cleaner.prefixes = ["test-prefix", "taxirides-realtime_beam_"] + # Mock subscriptions sub1 = mock.Mock() sub1.name = "projects/test-project/subscriptions/test-prefix-sub1" sub1.topic = "projects/test-project/topics/some-topic" + sub1.detached = False sub2 = mock.Mock() sub2.name = "projects/test-project/subscriptions/test-prefix-sub2-detached" sub2.topic = "_deleted-topic_" + sub2.detached = True sub3 = mock.Mock() - sub3.name = "projects/test-project/subscriptions/other-prefix-sub3" - sub3.topic = "projects/test-project/topics/another-topic" + sub3.name = "projects/test-project/subscriptions/taxirides-realtime_beam_-102875" + sub3.topic = "projects/pubsub-public-data/topics/taxirides-realtime" + sub3.detached = False + + sub4 = mock.Mock() + sub4.name = "projects/test-project/subscriptions/other-prefix-sub3" + sub4.topic = "projects/test-project/topics/another-topic" + sub4.detached = False - self.mock_subscriber_client.list_subscriptions.return_value = [sub1, sub2, sub3] + self.mock_subscriber_client.list_subscriptions.return_value = [sub1, sub2, sub3, sub4] with SilencePrint(): active = self.cleaner._active_resources() - self.assertIn("projects/test-project/subscriptions/test-prefix-sub1", active) + self.assertNotIn("projects/test-project/subscriptions/test-prefix-sub1", active) self.assertIn("projects/test-project/subscriptions/test-prefix-sub2-detached", active) + self.assertIn("projects/test-project/subscriptions/taxirides-realtime_beam_-102875", active) self.assertNotIn("projects/test-project/subscriptions/other-prefix-sub3", active) self.assertEqual(len(active), 2) From e917eb41863e048ae3049de7a63e6857b7a15178 Mon Sep 17 00:00:00 2001 From: HansMarcus01 Date: Wed, 29 Jul 2026 22:07:27 -0600 Subject: [PATCH 2/3] Fix: Modifications were made to manage the cleanup of disconnected and active subscriptions under prefixes. --- .test-infra/tools/stale_cleaner.py | 13 ++-- .test-infra/tools/test_stale_cleaner.py | 80 +++++++++++++++---------- 2 files changed, 55 insertions(+), 38 deletions(-) diff --git a/.test-infra/tools/stale_cleaner.py b/.test-infra/tools/stale_cleaner.py index ba0a94313cdb..140e93a1917b 100644 --- a/.test-infra/tools/stale_cleaner.py +++ b/.test-infra/tools/stale_cleaner.py @@ -331,18 +331,19 @@ def __init__(self, project_id: str, bucket_name: str, def _active_resources(self) -> dict: d = {} self.client = pubsub_v1.SubscriberClient() + taxi_prefix = f"{self.project_path}/subscriptions/taxirides-realtime_beam_" with self.client: for subscription in self.client.list_subscriptions(request={"project": self.project_path}): subscription_name = subscription.name # Apply prefix filtering if prefixes are defined - if not self.prefixes or any(subscription_name.startswith(f"{self.project_path}/subscriptions/{prefix}") for prefix in self.prefixes): - # Safe orphan detection: - # - Standard rule: detached subscriptions are eligible. - # - Taxi exception: subscriptions containing the NYC taxi prefix are eligible even if attached. - es_taxi = "taxirides-realtime_beam_" in subscription_name - if subscription.detached or es_taxi: + if subscription.detached: d[subscription_name] = GoogleCloudResource(resource_name=subscription_name, clock=self.clock) + #Only attached subscriptions with the NYC taxi prefix are eligible. + elif any( + subscription_name.startswith(f"{self.project_path}/subscriptions/{prefix}") for prefix in self.prefixes + ) and subscription_name.startswith(taxi_prefix): + d[subscription_name] = GoogleCloudResource(resource_name=subscription_name, clock=self.clock) return d diff --git a/.test-infra/tools/test_stale_cleaner.py b/.test-infra/tools/test_stale_cleaner.py index 29691d8448b8..80b257b003a0 100644 --- a/.test-infra/tools/test_stale_cleaner.py +++ b/.test-infra/tools/test_stale_cleaner.py @@ -431,42 +431,58 @@ def test_init(self): self.assertEqual(self.cleaner.time_threshold, self.time_threshold) self.assertIsInstance(self.cleaner.clock, FakeClock) - def test_active_resources(self): - """Test _active_resources method.""" - # Keep the existing standard prefix coverage and add the taxi-only cleanup prefix. - self.cleaner.prefixes = ["test-prefix", "taxirides-realtime_beam_"] - - # Mock subscriptions - sub1 = mock.Mock() - sub1.name = "projects/test-project/subscriptions/test-prefix-sub1" - sub1.topic = "projects/test-project/topics/some-topic" - sub1.detached = False - - sub2 = mock.Mock() - sub2.name = "projects/test-project/subscriptions/test-prefix-sub2-detached" - sub2.topic = "_deleted-topic_" - sub2.detached = True - - sub3 = mock.Mock() - sub3.name = "projects/test-project/subscriptions/taxirides-realtime_beam_-102875" - sub3.topic = "projects/pubsub-public-data/topics/taxirides-realtime" - sub3.detached = False - - sub4 = mock.Mock() - sub4.name = "projects/test-project/subscriptions/other-prefix-sub3" - sub4.topic = "projects/test-project/topics/another-topic" - sub4.detached = False - - self.mock_subscriber_client.list_subscriptions.return_value = [sub1, sub2, sub3, sub4] + def test_active_resources_active_subscriptions(self): + """Valida que las suscripciones activas con el prefijo de taxirides sean identificadas.""" + self.cleaner.prefixes = ["taxirides-realtime_beam_"] + + # Suscripción activa con el prefijo correcto de taxis + sub_taxi_active = mock.Mock() + sub_taxi_active.name = f"projects/{self.project_id}/subscriptions/taxirides-realtime_beam_-12345" + sub_taxi_active.topic = "projects/pubsub-public-data/topics/taxirides-realtime" + sub_taxi_active.detached = False + + # Suscripción activa con un prefijo distinto + sub_other_active = mock.Mock() + sub_other_active.name = f"projects/{self.project_id}/subscriptions/other-prefix-sub" + sub_other_active.topic = f"projects/{self.project_id}/topics/another-topic" + sub_other_active.detached = False + + self.mock_subscriber_client.list_subscriptions.return_value = [sub_taxi_active, sub_other_active] with SilencePrint(): active = self.cleaner._active_resources() - self.assertNotIn("projects/test-project/subscriptions/test-prefix-sub1", active) - self.assertIn("projects/test-project/subscriptions/test-prefix-sub2-detached", active) - self.assertIn("projects/test-project/subscriptions/taxirides-realtime_beam_-102875", active) - self.assertNotIn("projects/test-project/subscriptions/other-prefix-sub3", active) - self.assertEqual(len(active), 2) + # Verificamos que solo capture la suscripción de taxi, descartando la otra + self.assertIn(sub_taxi_active.name, active) + self.assertNotIn(sub_other_active.name, active) + self.assertEqual(len(active), 1) + + def test_active_resources_detached_subscriptions(self): + """Valida que las suscripciones desconectadas se mantengan en la lista de activos a limpiar, + independientemente de los prefijos específicos de taxis.""" + self.cleaner.prefixes = ["test-prefix"] + + # Suscripción desconectada (debería incluirse en la recolección) + sub_detached = mock.Mock() + sub_detached.name = f"projects/{self.project_id}/subscriptions/test-prefix-detached" + sub_detached.topic = "_deleted-topic_" + sub_detached.detached = True + + # Suscripción conectada normal (debería ignorarse) + sub_attached = mock.Mock() + sub_attached.name = f"projects/{self.project_id}/subscriptions/test-prefix-attached" + sub_attached.topic = f"projects/{self.project_id}/topics/some-topic" + sub_attached.detached = False + + self.mock_subscriber_client.list_subscriptions.return_value = [sub_detached, sub_attached] + + with SilencePrint(): + active = self.cleaner._active_resources() + + # Verificamos que solo se registre la suscripción huérfana + self.assertIn(sub_detached.name, active) + self.assertNotIn(sub_attached.name, active) + self.assertEqual(len(active), 1) def test_delete_resource(self): """Test _delete_resource method.""" From 1927d003b062bf1c1296466fd94aac88c71218a7 Mon Sep 17 00:00:00 2001 From: HansMarcus01 Date: Thu, 30 Jul 2026 18:41:37 -0600 Subject: [PATCH 3/3] Fix: Deleting the hardcode about taxi prefix --- .test-infra/tools/stale_cleaner.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/.test-infra/tools/stale_cleaner.py b/.test-infra/tools/stale_cleaner.py index 140e93a1917b..a00bb3ad7ab5 100644 --- a/.test-infra/tools/stale_cleaner.py +++ b/.test-infra/tools/stale_cleaner.py @@ -331,7 +331,6 @@ def __init__(self, project_id: str, bucket_name: str, def _active_resources(self) -> dict: d = {} self.client = pubsub_v1.SubscriberClient() - taxi_prefix = f"{self.project_path}/subscriptions/taxirides-realtime_beam_" with self.client: for subscription in self.client.list_subscriptions(request={"project": self.project_path}): @@ -342,7 +341,7 @@ def _active_resources(self) -> dict: #Only attached subscriptions with the NYC taxi prefix are eligible. elif any( subscription_name.startswith(f"{self.project_path}/subscriptions/{prefix}") for prefix in self.prefixes - ) and subscription_name.startswith(taxi_prefix): + ): d[subscription_name] = GoogleCloudResource(resource_name=subscription_name, clock=self.clock) return d