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

CHANGELOG.md

@@ -31,6 +31,7 @@ For more information about each release including git tags and artifacts, see [R
 - Event stream shows sensible defaults so first-load isn't empty ([#297](https://github.com/roostorg/osprey/pull/297) by [@haileyok](https://github.com/haileyok))
 - Replace `react-scripts` with `rsbuild`/`rspack` for UI builds ([#235](https://github.com/roostorg/osprey/pull/235) by [@chimosky](https://github.com/chimosky))
 - Migrate from npm to pnpm via Corepack ([#252](https://github.com/roostorg/osprey/pull/252) by [@haileyok](https://github.com/haileyok))
+- Osprey starts without GCP credentials instead of crashing: Pub/Sub publishing degrades to a noop (via a `make_publisher` factory that emits a `configuration.errors` metric) when credentials are absent, and can be disabled with `OSPREY_PUBSUB_ENABLED=false`. The GCP-backed telemetry itself (analytics, webhooks, rules-visualizer experiment metadata) stays inert until GCP or a replacement is configured ([#388](https://github.com/roostorg/osprey/pull/388) by [@julietshen](https://github.com/julietshen))
 
 ### Fixed
 

osprey_async_worker/src/osprey/async_worker/lib/publisher.py

@@ -29,6 +29,10 @@ class AsyncPubSubPublisher:
 
     Messages are buffered in an asyncio.Queue and flushed either when
     the batch reaches max_messages or after max_latency_seconds.
+
+    This is a plain transport that assumes GCP is configured. A callers-side
+    factory (as on the sync side) should decide whether to build this or a noop
+    publisher once the async worker wires it into a sink.
     """
 
     def __init__(
@@ -39,15 +43,15 @@ def __init__(
         max_latency_seconds: float = 1.0,
     ):
         self._topic_path = f'projects/{project_id}/topics/{topic_id}'
+        self._metric_tags = [f'project:{project_id}', f'topic:{topic_id}']
+        self._flush_task: asyncio.Task[None] | None = None
         self._client = pubsub_v1.PublisherClient(
             batch_settings=pubsub_v1.types.BatchSettings(max_messages=1),
         )
         self._queue: asyncio.Queue[bytes] = asyncio.Queue()
         self._max_messages = max_messages
         self._max_latency = max_latency_seconds
-        self._flush_task: asyncio.Task[None] | None = None
         self._started = False
-        self._metric_tags = [f'project:{project_id}', f'topic:{topic_id}']
 
     def _ensure_started(self) -> None:
         """Start the background flush task on first publish."""

osprey_worker/src/osprey/worker/cli/sinks.py

@@ -37,7 +37,7 @@
 from osprey.worker.lib.config import Config
 from osprey.worker.lib.osprey_engine import bootstrap_engine, bootstrap_engine_with_helpers, get_sources_provider
 from osprey.worker.lib.osprey_shared.logging import get_logger
-from osprey.worker.lib.publisher import PubSubPublisher
+from osprey.worker.lib.publisher import make_publisher
 from osprey.worker.lib.singletons import CONFIG, LABELS_PROVIDER
 from osprey.worker.lib.storage import postgres
 from osprey.worker.lib.storage.bigtable import osprey_bigtable
@@ -263,7 +263,7 @@ def run_bulk_label_sink(pooled: bool) -> None:
 
     analytics_pubsub_project_id = config.get_str('PUBSUB_DATA_PROJECT_ID', 'osprey-dev')
     analytics_pubsub_topic_id = config.get_str('PUBSUB_ANALYTICS_EVENT_TOPIC_ID', 'osprey-analytics')
-    analytics_publisher = PubSubPublisher(analytics_pubsub_project_id, analytics_pubsub_topic_id)
+    analytics_publisher = make_publisher(analytics_pubsub_project_id, analytics_pubsub_topic_id)
 
     def factory() -> BulkLabelSink:
         # NOTE: It's very important the input stream is created per-webhook sink
@@ -304,11 +304,11 @@ def rollback_bulk_label_effects(ctx: click.Context, task_id: int, include_ids_fr
 
     analytics_pubsub_project_id = config.get_str('PUBSUB_DATA_PROJECT_ID', 'osprey-dev')
     analytics_pubsub_topic_id = config.get_str('PUBSUB_ANALYTICS_EVENT_TOPIC_ID', 'osprey-analytics')
-    analytics_publisher = PubSubPublisher(analytics_pubsub_project_id, analytics_pubsub_topic_id)
+    analytics_publisher = make_publisher(analytics_pubsub_project_id, analytics_pubsub_topic_id)
 
     osprey_webhook_pubsub_project = config.get_str('PUBSUB_OSPREY_WEBHOOKS_PROJECT_ID', 'osprey-dev')
     osprey_webhook_pubsub_topic = config.get_str('PUBSUB_OSPREY_WEBHOOKS_TOPIC_ID', 'osprey-webhooks')
-    webhooks_publisher = PubSubPublisher(osprey_webhook_pubsub_project, osprey_webhook_pubsub_topic)
+    webhooks_publisher = make_publisher(osprey_webhook_pubsub_project, osprey_webhook_pubsub_topic)
 
     task = BulkLabelTask.get_one(task_id)
     if task is None:

osprey_worker/src/osprey/worker/lib/data_exporters/validation_result_exporter.py

@@ -5,7 +5,7 @@
     ValidateExperiments,
 )
 from osprey.worker.lib.data_exporters.models import ospreyExperimentMetadataUpdate
-from osprey.worker.lib.publisher import BasePublisher, PubSubPublisher
+from osprey.worker.lib.publisher import BasePublisher, make_publisher
 from osprey.worker.lib.singletons import CONFIG
 
 
@@ -56,4 +56,4 @@ def get_validation_result_exporter() -> BaseValidationResultExporter:
 
     pubsub_project_id = config.get_str('PUBSUB_DATA_PROJECT_ID', 'osprey-dev')
     pubsub_topic_id = config.get_str('PUBSUB_ANALYTICS_EVENT_TOPIC_ID', 'osprey-analytics')
-    return ExperimentValidationResultExporter(publisher=PubSubPublisher(pubsub_project_id, pubsub_topic_id))
+    return ExperimentValidationResultExporter(publisher=make_publisher(pubsub_project_id, pubsub_topic_id))

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

@@ -1,13 +1,17 @@
 import abc
+import logging
 from typing import TypeVar
 
+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 +44,13 @@ def stop(self) -> None:
 
 
 class PubSubPublisher(BasePublisher):
+    """Publishes Pydantic models to a Google Cloud Pub/Sub topic.
+
+    This is a plain transport: it assumes GCP is configured and builds its client
+    eagerly. Callers that may run without GCP should build publishers via
+    make_publisher(), which hands back a NullPublisher when publishing is off.
+    """
+
     def __init__(
         self,
         project_id: str,
@@ -103,3 +114,42 @@ def publish(self, data: str, attributes: dict[str, str] | None = None) -> None:
             attributes = {}
 
         super().publish(data, attributes)  # type: ignore[type-var]
+
+
+def make_publisher(
+    project_id: str,
+    topic_id: str,
+    raise_on_error: bool = False,
+    max_bytes: int = 2000000,
+    max_messages: int = 250,
+    max_latency: float = 1.0,
+) -> BasePublisher:
+    """Build a Pub/Sub publisher, or a NullPublisher when publishing is off.
+
+    Publishing is on by default and can be turned off by setting OSPREY_PUBSUB_ENABLED
+    to false, which returns a NullPublisher. When it is on but GCP credentials cannot
+    be resolved (e.g. local dev or adopter environments without GCP), PubSubPublisher
+    construction raises; that is a misconfiguration, so we log, emit a one-time
+    configuration.errors metric, and fall back to a NullPublisher rather than crash.
+    """
+    # Imported here to avoid pulling the config/singletons stack into this low-level module.
+    from osprey.worker.lib.singletons import CONFIG
+
+    if not CONFIG.instance().get_bool('OSPREY_PUBSUB_ENABLED', True):
+        return NullPublisher()
+    try:
+        return PubSubPublisher(
+            project_id,
+            topic_id,
+            raise_on_error=raise_on_error,
+            max_bytes=max_bytes,
+            max_messages=max_messages,
+            max_latency=max_latency,
+        )
+    except DefaultCredentialsError:
+        logger.warning('GCP credentials errored, publishing disabled (project=%s, topic=%s)', project_id, topic_id)
+        metrics.increment(
+            'configuration.errors',
+            tags=[f'project:{project_id}', f'topic:{topic_id}', 'reason:gcp_credentials_missing'],
+        )
+        return NullPublisher()

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

@@ -0,0 +1,53 @@
+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 NullPublisher, PubSubPublisher, make_publisher
+
+
+def _mock_config(pubsub_enabled: bool) -> MagicMock:
+    config = MagicMock()
+    config.instance.return_value.get_bool.return_value = pubsub_enabled
+    return config
+
+
+def test_make_publisher_returns_null_when_pubsub_not_enabled() -> None:
+    config = _mock_config(pubsub_enabled=False)
+    with (
+        patch('osprey.worker.lib.singletons.CONFIG', config),
+        patch.object(publisher, 'BatchPubsubPublisherClient') as client_cls,
+        patch.object(publisher, 'metrics') as metrics_mock,
+    ):
+        pub = make_publisher('proj', 'topic')
+    assert isinstance(pub, NullPublisher)
+    # On by default; explicitly disabling is not a misconfiguration, so no client and no metric.
+    config.instance.return_value.get_bool.assert_called_once_with('OSPREY_PUBSUB_ENABLED', True)
+    assert not client_cls.called
+    metrics_mock.increment.assert_not_called()
+
+
+def test_make_publisher_returns_pubsub_when_enabled_and_creds_present() -> None:
+    with (
+        patch('osprey.worker.lib.singletons.CONFIG', _mock_config(pubsub_enabled=True)),
+        patch.object(publisher, 'BatchPubsubPublisherClient') as client_cls,
+    ):
+        pub = make_publisher('proj', 'topic')
+    assert isinstance(pub, PubSubPublisher)
+    assert client_cls.called
+
+
+def test_make_publisher_falls_back_to_null_and_metric_when_creds_absent(caplog: pytest.LogCaptureFixture) -> None:
+    with (
+        patch('osprey.worker.lib.singletons.CONFIG', _mock_config(pubsub_enabled=True)),
+        patch.object(publisher, 'BatchPubsubPublisherClient', side_effect=DefaultCredentialsError()),
+        patch.object(publisher, 'metrics') as metrics_mock,
+    ):
+        with caplog.at_level('WARNING', logger=publisher.logger.name):
+            pub = make_publisher('proj', 'topic')
+    assert isinstance(pub, NullPublisher)
+    assert 'credentials errored' in caplog.text
+    metrics_mock.increment.assert_called_once_with(
+        'configuration.errors',
+        tags=['project:proj', 'topic:topic', 'reason:gcp_credentials_missing'],
+    )

osprey_worker/src/osprey/worker/ui_api/osprey/singletons.py

@@ -1,4 +1,4 @@
-from osprey.worker.lib.publisher import PubSubPublisher
+from osprey.worker.lib.publisher import BasePublisher, make_publisher
 from osprey.worker.lib.singleton import Singleton
 from osprey.worker.lib.singletons import CONFIG
 
@@ -7,11 +7,11 @@
 DRUID: Singleton[DruidClientHolder] = Singleton(DruidClientHolder)
 
 
-def _init_analytics_publisher() -> PubSubPublisher:
+def _init_analytics_publisher() -> BasePublisher:
     config = CONFIG.instance()
     project = config.get_str('PUBSUB_DATA_PROJECT_ID', 'osprey-dev')
     topic = config.get_str('PUBSUB_ANALYTICS_EVENT_TOPIC_ID', 'osprey-analytics')
-    return PubSubPublisher(project, topic)
+    return make_publisher(project, topic)
 
 
-ANALYTICS_PUBLISHER: Singleton[PubSubPublisher] = Singleton(_init_analytics_publisher)
+ANALYTICS_PUBLISHER: Singleton[BasePublisher] = Singleton(_init_analytics_publisher)