Stale pubsub subscriptions cleaner (#35553)

* Implement PubSub Subscription stale cleaner as dry run
This cleaner will delete the stale PubSub subscriptions that do not have any associated topics.

* Add tests for the PubSub subscription deletion class

Author: ksobrenat32

.test-infra/tools/stale_cleaner.py

@@ -25,6 +25,7 @@
 
 # Resource types
 PUBSUB_TOPIC_RESOURCE = "pubsub_topic"
+PUBSUB_SUBSCRIPTION_RESOURCE = "pubsub_subscription"
 
 # Storage constants
 STORAGE_PREFIX = "stale_cleaner/"
@@ -34,6 +35,7 @@
 
 # Time constants (in seconds)
 DEFAULT_PUBSUB_TOPIC_THRESHOLD = 86400  # 1 day
+DEFAULT_PUBSUB_SUBSCRIPTION_THRESHOLD = 86400  # 1 day
 DEFAULT_TIME_THRESHOLD = 3600  # 1 hour
 
 # Default values for testing
@@ -285,6 +287,12 @@ def delete_stale(self, dry_run: bool = True) -> None:
 
 # PubSub topic cleaner
 class PubSubTopicCleaner(StaleCleaner):
+    """
+    This cleaner will delete PubSub topics that are stale based on the time threshold.
+    It uses the PubSub API to list and delete topics.
+    It also applies prefix filtering to only delete topics that match the specified prefixes.
+    """
+
     def __init__(self, project_id: str, bucket_name: str,
                     prefixes: list = None, time_threshold: int = DEFAULT_PUBSUB_TOPIC_THRESHOLD,
                     clock: Clock = None) -> None:
@@ -304,7 +312,50 @@ def _delete_resource(self, resource_name: str) -> None:
         print(f"{self.clock()} - Deleting PubSub topic {resource_name}")
         self.client.delete_topic(request={"topic": resource_name})
 
-if __name__ == "__main__":
+# PubSub Subscription cleaner
+class PubSubSubscriptionCleaner(StaleCleaner):
+    """
+    This cleaner will delete PubSub subscriptions that are stale based on the time threshold.
+    It uses the PubSub API to list and delete subscriptions.
+    It also applies prefix filtering to only delete subscriptions that match the specified prefixes.
+    It checks if the subscription is detached (whether it has a topic associated with it).
+    If it is detached, it will be considered stale and eligible for deletion.
+    """
+
+    def __init__(self, project_id: str, bucket_name: str,
+                    prefixes: list = None, time_threshold: int = DEFAULT_PUBSUB_SUBSCRIPTION_THRESHOLD,
+                    clock: Clock = None) -> None:
+        super().__init__(project_id, PUBSUB_SUBSCRIPTION_RESOURCE, bucket_name, prefixes, time_threshold, clock)
+        self.client = None  # Will be initialized in each method that needs it
+
+    def _active_resources(self) -> dict:
+        d = {}
+        self.client = pubsub_v1.SubscriberClient()
+
+        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):
+                    # Check if the subscription has a topic associated with it
+                    if subscription.detached:
+                        d[subscription_name] = GoogleCloudResource(resource_name=subscription_name, clock=self.clock)
+
+        return d
+
+    def _delete_resource(self, resource_name: str) -> None:
+        self.client = pubsub_v1.SubscriberClient()
+        print(f"{self.clock()} - Deleting PubSub subscription {resource_name}")
+        with self.client:
+            subscription_path = self.client.subscription_path(self.project_id, resource_name)
+            self.client.delete_subscription(request={"subscription": subscription_path})
+
+def clean_pubsub_topics():
+    """ Clean up stale PubSub topics in the specified GCP project.
+    This function initializes the PubSubTopicCleaner with the default project ID and bucket name,
+    and a predefined list of topic prefixes.
+    It then refreshes the resources and deletes any stale topics.
+    """
     project_id = DEFAULT_PROJECT_ID
     bucket_name = DEFAULT_BUCKET_NAME
 
@@ -355,3 +406,32 @@ def _delete_resource(self, resource_name: str) -> None:
 
     # Delete stale resources
     cleaner.delete_stale(dry_run=False)
+
+def clean_pubsub_subscriptions():
+    """ Clean up stale PubSub subscriptions in the specified GCP project.
+    This function initializes the PubSubSubscriptionCleaner with the default project ID and bucket name,
+    and a predefined list of subscription prefixes.
+    It then refreshes the resources and deletes any stale 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 = []
+
+    # Create a PubSubSubscriptionCleaner instance
+    cleaner = PubSubSubscriptionCleaner(project_id=project_id, bucket_name=bucket_name,
+                                        prefixes=prefixes, time_threshold=DEFAULT_PUBSUB_SUBSCRIPTION_THRESHOLD)
+
+    # Refresh resources
+    cleaner.refresh()
+
+    # Delete stale resources
+    cleaner.delete_stale(dry_run=True) # Keep dry_run=True to avoid accidental deletions during testing
+
+if __name__ == "__main__":
+    # Clean up stale PubSub topics
+    clean_pubsub_topics()
+
+    # Clean up stale PubSub subscriptions
+    clean_pubsub_subscriptions()

.test-infra/tools/test_stale_cleaner.py

@@ -25,6 +25,7 @@
     GoogleCloudResource,
     StaleCleaner,
     PubSubTopicCleaner,
+    PubSubSubscriptionCleaner,
     FakeClock,
     DEFAULT_TIME_THRESHOLD,
     PUBSUB_TOPIC_RESOURCE,
@@ -390,6 +391,86 @@ def test_delete_stale(self):
         self.assertNotIn("deleted-resource", self.cleaner.deleted_resources)
 
 
+class PubSubSubscriptionCleanerTest(unittest.TestCase):
+    """Tests for the PubSubSubscriptionCleaner class."""
+
+    def setUp(self):
+        """Set up test fixtures."""
+        self.project_id = "test-project"
+        self.bucket_name = "test-bucket"
+        self.prefixes = ["test-prefix"]
+        self.time_threshold = 86400  # 1 day
+        self.fake_clock = FakeClock("2025-05-28T10:00:00")
+
+        # Mock PubSub client
+        self.mock_client_patcher = mock.patch('google.cloud.pubsub_v1.SubscriberClient')
+        self.MockSubscriberClientClass = self.mock_client_patcher.start()
+        self.mock_subscriber_client = self.MockSubscriberClientClass.return_value
+
+        # Create a test cleaner
+        self.cleaner = PubSubSubscriptionCleaner(
+            project_id=self.project_id,
+            bucket_name=self.bucket_name,
+            prefixes=self.prefixes,
+            time_threshold=self.time_threshold,
+            clock=self.fake_clock
+        )
+
+        self.cleaner._write_resources = SilencedMock()
+        self.cleaner._stored_resources = SilencedMock(return_value={})
+
+    def tearDown(self):
+        """Tear down test fixtures."""
+        self.mock_client_patcher.stop()
+
+    def test_init(self):
+        """Test initialization."""
+        self.assertEqual(self.cleaner.project_id, self.project_id)
+        self.assertEqual(self.cleaner.bucket_name, self.bucket_name)
+        self.assertEqual(self.cleaner.prefixes, self.prefixes)
+        self.assertEqual(self.cleaner.time_threshold, self.time_threshold)
+        self.assertIsInstance(self.cleaner.clock, FakeClock)
+
+    def test_active_resources(self):
+        """Test _active_resources method."""
+        # Mock subscriptions
+        sub1 = mock.Mock()
+        sub1.name = "projects/test-project/subscriptions/test-prefix-sub1"
+        sub1.topic = "projects/test-project/topics/some-topic"
+
+        sub2 = mock.Mock()
+        sub2.name = "projects/test-project/subscriptions/test-prefix-sub2-detached"
+        sub2.topic = "_deleted-topic_"
+
+        sub3 = mock.Mock()
+        sub3.name = "projects/test-project/subscriptions/other-prefix-sub3"
+        sub3.topic = "projects/test-project/topics/another-topic"
+
+        self.mock_subscriber_client.list_subscriptions.return_value = [sub1, sub2, sub3]
+
+        with SilencePrint():
+            active = self.cleaner._active_resources()
+
+        self.assertIn("projects/test-project/subscriptions/test-prefix-sub1", active)
+        self.assertIn("projects/test-project/subscriptions/test-prefix-sub2-detached", active)
+        self.assertNotIn("projects/test-project/subscriptions/other-prefix-sub3", active)
+        self.assertEqual(len(active), 2)
+
+    def test_delete_resource(self):
+        """Test _delete_resource method."""
+        sub_name = "test-sub-to-delete"
+        subscription_path = f"projects/{self.project_id}/subscriptions/{sub_name}"
+        self.mock_subscriber_client.subscription_path.return_value = subscription_path
+
+        with SilencePrint():
+            self.cleaner._delete_resource(sub_name)
+
+        self.mock_subscriber_client.subscription_path.assert_called_once_with(self.project_id, sub_name)
+        self.mock_subscriber_client.delete_subscription.assert_called_once_with(
+            request={'subscription': subscription_path}
+        )
+
+
 class PubSubTopicCleanerTest(unittest.TestCase):
     """Tests for the PubSubTopicCleaner class."""
 
@@ -484,10 +565,10 @@ def test_delete_resource(self):
             self.cleaner._delete_resource(resource_name)
 
             # Check that delete_topic was called
-            self.cleaner.client.delete_topic.assert_called_once_with(name=resource_name)
+            self.cleaner.client.delete_topic.assert_called_once_with(request={'topic': resource_name})
 
             # Check that correct message was printed
-            mock_print.assert_called_once_with(f"{self.cleaner.clock()} - Deleting PubSub topic test-topic")
+            mock_print.assert_called_once_with(f"{self.cleaner.clock()} - Deleting PubSub topic {resource_name}")
 
 if __name__ == '__main__':
     unittest.main()