feat(publisher): degrade PubSubPublisher to noop when GCP credentials are absent

osprey_worker/src/osprey/worker/lib/publisher.py

@@ -1,13 +1,18 @@
 import abc
+import logging
 from typing import Dict, Optional, TypeVar
 
+import google.auth
+from google.auth.exceptions import DefaultCredentialsError
 from google.cloud import pubsub_v1
 from osprey.worker.lib.instruments import metrics
 from osprey.worker.lib.pubsub.publisher_client import BatchPubsubPublisherClient
 from pydantic import BaseModel
 
 _PydanticModelT = TypeVar('_PydanticModelT', bound=BaseModel)
 
+logger = logging.getLogger(__name__)
+
 
 class BasePublisher(abc.ABC):
     """
@@ -40,6 +45,15 @@ def stop(self) -> None:
 
 
 class PubSubPublisher(BasePublisher):
+    """Publishes Pydantic models to a Google Cloud Pub/Sub topic.
+
+    Degrades to noop mode when GCP credentials cannot be resolved at construction
+    time (e.g. local dev or adopter environments without GCP). In noop mode no
+    underlying client is built, publish() and stop() return immediately, and a
+    `PubSubPublisher.publisher.noop` metric is incremented per call so the inert
+    state is visible in dashboards. A one-time warning is logged at construction.
+    """
+
     def __init__(
         self,
         project_id: str,
@@ -54,15 +68,23 @@ def __init__(
             topic_id=topic_id,
         )
         self._project = project_id
+        self._raise_on_error = raise_on_error
+        self._tags = [f'project:{self._project}', f'topic:{self._topic_name}']
+        self._enabled = _check_gcp_credentials()
+        if not self._enabled:
+            logger.warning(
+                'GCP credentials not detected — PubSubPublisher running in noop mode (project=%s, topic=%s)',
+                project_id,
+                topic_id,
+            )
+            return
         batch_settings = pubsub_v1.types.BatchSettings(
             max_bytes=max_bytes,  # default 1MB
             max_messages=max_messages,  # default 100 messages
             max_latency=max_latency,
         )
         # self._publisher = pubsub_v1.PublisherClient(batch_settings=batch_settings)
         self._publisher = BatchPubsubPublisherClient(batch_settings=batch_settings)
-        self._raise_on_error = raise_on_error
-        self._tags = [f'project:{self._project}', f'topic:{self._topic_name}']
 
     def prepare_data(self, data: _PydanticModelT) -> bytes:
         """
@@ -72,6 +94,9 @@ def prepare_data(self, data: _PydanticModelT) -> bytes:
         return data.json(exclude_none=True).encode()
 
     def publish(self, data: _PydanticModelT, attributes: Optional[Dict[str, str]] = None) -> None:
+        if not self._enabled:
+            metrics.increment(f'{self.__class__.__name__}.publisher.noop', tags=self._tags)
+            return
         if attributes is None:
             attributes = {}
 
@@ -89,6 +114,8 @@ def publish(self, data: _PydanticModelT, attributes: Optional[Dict[str, str]] =
         metrics.increment(f'{self.__class__.__name__}.publisher.success', tags=self._tags)
 
     def stop(self) -> None:
+        if not self._enabled:
+            return
         self._publisher.stop()
 
 
@@ -103,3 +130,17 @@ def publish(self, data: str, attributes: Optional[Dict[str, str]] = None) -> Non
             attributes = {}
 
         super().publish(data, attributes)  # type: ignore[type-var]
+
+
+_gcp_credentials_available: Optional[bool] = None
+
+
+def _check_gcp_credentials() -> bool:
+    global _gcp_credentials_available
+    if _gcp_credentials_available is None:
+        try:
+            google.auth.default()
+            _gcp_credentials_available = True
+        except DefaultCredentialsError:
+            _gcp_credentials_available = False
+    return _gcp_credentials_available

osprey_worker/src/osprey/worker/lib/tests/test_publisher.py

@@ -0,0 +1,74 @@
+from typing import Iterator
+from unittest.mock import MagicMock, patch
+
+import pytest
+from google.auth.exceptions import DefaultCredentialsError
+from osprey.worker.lib import publisher
+from osprey.worker.lib.publisher import PubSubPublisher, _check_gcp_credentials
+
+
+@pytest.fixture(autouse=True)
+def reset_cred_cache() -> Iterator[None]:
+    publisher._gcp_credentials_available = None
+    yield
+    publisher._gcp_credentials_available = None
+
+
+def test_check_gcp_credentials_true_when_default_resolves() -> None:
+    with patch.object(publisher.google.auth, 'default', return_value=(MagicMock(), 'proj')):
+        assert _check_gcp_credentials() is True
+
+
+def test_check_gcp_credentials_false_when_default_raises() -> None:
+    with patch.object(publisher.google.auth, 'default', side_effect=DefaultCredentialsError()):
+        assert _check_gcp_credentials() is False
+
+
+def test_check_gcp_credentials_is_cached() -> None:
+    with patch.object(publisher.google.auth, 'default', return_value=(MagicMock(), 'proj')) as default_mock:
+        _check_gcp_credentials()
+        _check_gcp_credentials()
+        _check_gcp_credentials()
+    assert default_mock.call_count == 1
+
+
+def test_pubsub_publisher_constructs_client_when_creds_present() -> None:
+    with (
+        patch.object(publisher.google.auth, 'default', return_value=(MagicMock(), 'proj')),
+        patch.object(publisher, 'BatchPubsubPublisherClient') as client_cls,
+    ):
+        pub = PubSubPublisher('proj', 'topic')
+    assert pub._enabled is True
+    assert client_cls.called
+
+
+def test_pubsub_publisher_noops_when_creds_absent(caplog: pytest.LogCaptureFixture) -> None:
+    with (
+        patch.object(publisher.google.auth, 'default', side_effect=DefaultCredentialsError()),
+        patch.object(publisher, 'BatchPubsubPublisherClient') as client_cls,
+    ):
+        with caplog.at_level('WARNING', logger=publisher.logger.name):
+            pub = PubSubPublisher('proj', 'topic')
+    assert pub._enabled is False
+    assert not client_cls.called
+    assert 'noop mode' in caplog.text
+    assert 'project=proj' in caplog.text
+    assert 'topic=topic' in caplog.text
+
+
+def test_publish_short_circuits_and_emits_noop_metric_when_disabled() -> None:
+    with (
+        patch.object(publisher.google.auth, 'default', side_effect=DefaultCredentialsError()),
+        patch.object(publisher, 'metrics') as metrics_mock,
+    ):
+        pub = PubSubPublisher('proj', 'topic')
+        pub.publish(MagicMock())
+    metric_names = [call.args[0] for call in metrics_mock.increment.call_args_list]
+    assert metric_names == ['PubSubPublisher.publisher.noop']
+
+
+def test_stop_short_circuits_when_disabled() -> None:
+    with patch.object(publisher.google.auth, 'default', side_effect=DefaultCredentialsError()):
+        pub = PubSubPublisher('proj', 'topic')
+        pub.stop()
+    assert not hasattr(pub, '_publisher')