diff --git a/CHANGELOG.md b/CHANGELOG.md index 16cbfa16..c286042c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -31,7 +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)) -- `PubSubPublisher` degrades to a noop when GCP credentials are absent or when `DISABLE_GCP_PUBSUB` is set, so Osprey runs without GCP config instead of failing ([#388](https://github.com/roostorg/osprey/pull/388) by [@julietshen](https://github.com/julietshen)) +- Pub/Sub publishers degrade to a noop when GCP credentials are absent or when `DISABLE_GCP_PUBSUB` is set, so Osprey runs without GCP config instead of failing: `PubSubPublisher` ([#388](https://github.com/roostorg/osprey/pull/388)) and `AsyncPubSubPublisher` ([#411](https://github.com/roostorg/osprey/pull/411)) (by [@julietshen](https://github.com/julietshen)) ### Fixed diff --git a/osprey_async_worker/src/osprey/async_worker/lib/publisher.py b/osprey_async_worker/src/osprey/async_worker/lib/publisher.py index 9b1e6410..46da53b8 100644 --- a/osprey_async_worker/src/osprey/async_worker/lib/publisher.py +++ b/osprey_async_worker/src/osprey/async_worker/lib/publisher.py @@ -9,6 +9,7 @@ from google.api_core.retry import Retry from google.cloud import pubsub_v1 +from osprey.worker.lib.gcp_credentials import gcp_credentials_available, gcp_pubsub_disabled from osprey.worker.lib.instruments import metrics from pydantic import BaseModel @@ -29,6 +30,12 @@ class AsyncPubSubPublisher: Messages are buffered in an asyncio.Queue and flushed either when the batch reaches max_messages or after max_latency_seconds. + + Degrades to noop mode when the DISABLE_GCP_PUBSUB env var is set, or 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 and the + publish paths return immediately. A one-time warning is logged at construction + so the inert state is visible. """ def __init__( @@ -39,15 +46,31 @@ 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 + if gcp_pubsub_disabled(): + self._enabled = False + logger.warning( + 'DISABLE_GCP_PUBSUB is set, AsyncPubSubPublisher disabled (project=%s, topic=%s)', + project_id, + topic_id, + ) + return + self._enabled = gcp_credentials_available() + if not self._enabled: + logger.warning( + 'GCP credentials not detected, AsyncPubSubPublisher running in noop mode (project=%s, topic=%s)', + project_id, + topic_id, + ) + return 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.""" @@ -129,6 +152,8 @@ def publish(self, data: BaseModel) -> None: def publish_bytes(self, data: bytes) -> None: """Queue raw bytes for async batched publishing.""" + if not self._enabled: + return self._ensure_started() try: self._queue.put_nowait(data) @@ -138,6 +163,8 @@ def publish_bytes(self, data: bytes) -> None: async def stop(self) -> None: """Flush remaining messages and stop.""" + if not self._enabled: + return if self._flush_task is not None: self._flush_task.cancel() try: diff --git a/osprey_async_worker/src/osprey/async_worker/tests/test_publisher.py b/osprey_async_worker/src/osprey/async_worker/tests/test_publisher.py index e589f4cd..8f26ef96 100644 --- a/osprey_async_worker/src/osprey/async_worker/tests/test_publisher.py +++ b/osprey_async_worker/src/osprey/async_worker/tests/test_publisher.py @@ -2,13 +2,18 @@ from unittest.mock import MagicMock, patch +import pytest from google.api_core.exceptions import NotFound +from osprey.async_worker.lib import publisher as publisher_module from osprey.async_worker.lib.publisher import _PUBLISH_RETRY, AsyncPubSubPublisher def _make_publisher(): - """Return an AsyncPubSubPublisher with a mocked PublisherClient.""" - with patch('osprey.async_worker.lib.publisher.pubsub_v1.PublisherClient'): + """Return an enabled AsyncPubSubPublisher with a mocked PublisherClient.""" + with ( + patch('osprey.async_worker.lib.publisher.gcp_credentials_available', return_value=True), + patch('osprey.async_worker.lib.publisher.pubsub_v1.PublisherClient'), + ): publisher = AsyncPubSubPublisher(project_id='proj', topic_id='topic') publisher._client = MagicMock() return publisher @@ -48,3 +53,50 @@ def test_permanent_failure_metric_fires(mock_metrics): assert len(failure_calls) == 1 assert failure_calls[0][0][0] == 'async_pubsub_publisher.publish.failure' assert f'error:{exc.__class__.__name__}' in failure_calls[0][1]['tags'] + + +def test_noops_when_creds_absent(caplog: pytest.LogCaptureFixture) -> None: + with ( + patch('osprey.async_worker.lib.publisher.gcp_credentials_available', return_value=False), + patch('osprey.async_worker.lib.publisher.pubsub_v1.PublisherClient') as client_cls, + ): + with caplog.at_level('WARNING', logger=publisher_module.logger.name): + publisher = AsyncPubSubPublisher(project_id='proj', topic_id='topic') + assert publisher._enabled is False + assert not client_cls.called + assert not hasattr(publisher, '_client') + assert 'noop mode' in caplog.text + assert 'project=proj' in caplog.text + assert 'topic=topic' in caplog.text + + +def test_disabled_via_env(monkeypatch, caplog: pytest.LogCaptureFixture) -> None: + monkeypatch.setenv('DISABLE_GCP_PUBSUB', 'true') + with ( + patch('osprey.async_worker.lib.publisher.gcp_credentials_available') as cred_check, + patch('osprey.async_worker.lib.publisher.pubsub_v1.PublisherClient') as client_cls, + ): + with caplog.at_level('WARNING', logger=publisher_module.logger.name): + publisher = AsyncPubSubPublisher(project_id='proj', topic_id='topic') + assert publisher._enabled is False + assert not client_cls.called + # The opt-out short-circuits before probing credentials. + assert not cred_check.called + assert 'DISABLE_GCP_PUBSUB' in caplog.text + + +@patch('osprey.async_worker.lib.publisher.metrics') +def test_publish_bytes_short_circuits_silently_when_disabled(mock_metrics) -> None: + with patch('osprey.async_worker.lib.publisher.gcp_credentials_available', return_value=False): + publisher = AsyncPubSubPublisher(project_id='proj', topic_id='topic') + publisher.publish_bytes(b'data') + + mock_metrics.increment.assert_not_called() + + +async def test_stop_short_circuits_when_disabled() -> None: + with patch('osprey.async_worker.lib.publisher.gcp_credentials_available', return_value=False): + publisher = AsyncPubSubPublisher(project_id='proj', topic_id='topic') + await publisher.stop() + + assert not hasattr(publisher, '_client') diff --git a/osprey_worker/src/osprey/worker/lib/data_exporters/test/conftest.py b/osprey_worker/src/osprey/worker/lib/data_exporters/test/conftest.py index ebe6099f..abf13658 100644 --- a/osprey_worker/src/osprey/worker/lib/data_exporters/test/conftest.py +++ b/osprey_worker/src/osprey/worker/lib/data_exporters/test/conftest.py @@ -28,5 +28,5 @@ def pubsub_client_mock(monkeypatch_session, monkeypatch) -> MagicMock: # monkeypatch so it reverts after each test instead of leaking into later # modules (e.g. test_publisher.py, whose disabled-publisher tests would then # build a real client and hang on stop()). - monkeypatch.setattr(publisher, '_check_gcp_credentials', lambda: True) + monkeypatch.setattr(publisher, 'gcp_credentials_available', lambda: True) return pubsub_client_mock diff --git a/osprey_worker/src/osprey/worker/lib/gcp_credentials.py b/osprey_worker/src/osprey/worker/lib/gcp_credentials.py new file mode 100644 index 00000000..540b44cd --- /dev/null +++ b/osprey_worker/src/osprey/worker/lib/gcp_credentials.py @@ -0,0 +1,36 @@ +"""Shared detection of whether GCP Pub/Sub publishing should be active. + +Used by the Pub/Sub publishers to decide whether to run for real or degrade to +a noop (e.g. local dev or adopter environments without GCP). Publishing is off +when DISABLE_GCP_PUBSUB is set, or when GCP credentials cannot be resolved. The +credential result is cached at module level so google.auth.default() is probed +at most once per process, and guarded by a lock so concurrent constructors +probe only once. +""" + +import os +import threading + +import google.auth +from google.auth.exceptions import DefaultCredentialsError + +_gcp_credentials_available: bool | None = None +_gcp_credentials_lock = threading.Lock() + + +def gcp_pubsub_disabled() -> bool: + return os.environ.get('DISABLE_GCP_PUBSUB', '').lower() == 'true' + + +def gcp_credentials_available() -> bool: + global _gcp_credentials_available + if _gcp_credentials_available is None: + with _gcp_credentials_lock: + # Re-check under the lock so concurrent callers probe google.auth.default() only once. + if _gcp_credentials_available is None: + try: + google.auth.default() + _gcp_credentials_available = True + except DefaultCredentialsError: + _gcp_credentials_available = False + return _gcp_credentials_available diff --git a/osprey_worker/src/osprey/worker/lib/publisher.py b/osprey_worker/src/osprey/worker/lib/publisher.py index 40846241..ebb90506 100644 --- a/osprey_worker/src/osprey/worker/lib/publisher.py +++ b/osprey_worker/src/osprey/worker/lib/publisher.py @@ -1,12 +1,9 @@ import abc import logging -import os -import threading from typing import TypeVar -import google.auth -from google.auth.exceptions import DefaultCredentialsError from google.cloud import pubsub_v1 +from osprey.worker.lib.gcp_credentials import gcp_credentials_available, gcp_pubsub_disabled from osprey.worker.lib.instruments import metrics from osprey.worker.lib.pubsub.publisher_client import BatchPubsubPublisherClient from pydantic import BaseModel @@ -72,7 +69,7 @@ def __init__( self._project = project_id self._raise_on_error = raise_on_error self._tags = [f'project:{self._project}', f'topic:{self._topic_name}'] - if _gcp_pubsub_disabled(): + if gcp_pubsub_disabled(): self._enabled = False logger.warning( 'DISABLE_GCP_PUBSUB is set, PubSubPublisher disabled (project=%s, topic=%s)', @@ -80,7 +77,7 @@ def __init__( topic_id, ) return - self._enabled = _check_gcp_credentials() + self._enabled = gcp_credentials_available() if not self._enabled: logger.warning( 'GCP credentials not detected, PubSubPublisher running in noop mode (project=%s, topic=%s)', @@ -139,25 +136,3 @@ def publish(self, data: str, attributes: dict[str, str] | None = None) -> None: attributes = {} super().publish(data, attributes) # type: ignore[type-var] - - -def _gcp_pubsub_disabled() -> bool: - return os.environ.get('DISABLE_GCP_PUBSUB', '').lower() == 'true' - - -_gcp_credentials_available: bool | None = None -_gcp_credentials_lock = threading.Lock() - - -def _check_gcp_credentials() -> bool: - global _gcp_credentials_available - if _gcp_credentials_available is None: - with _gcp_credentials_lock: - # Re-check under the lock so concurrent constructors probe google.auth.default() only once. - if _gcp_credentials_available is None: - try: - google.auth.default() - _gcp_credentials_available = True - except DefaultCredentialsError: - _gcp_credentials_available = False - return _gcp_credentials_available diff --git a/osprey_worker/src/osprey/worker/lib/tests/test_gcp_credentials.py b/osprey_worker/src/osprey/worker/lib/tests/test_gcp_credentials.py new file mode 100644 index 00000000..db49815b --- /dev/null +++ b/osprey_worker/src/osprey/worker/lib/tests/test_gcp_credentials.py @@ -0,0 +1,41 @@ +from typing import Iterator +from unittest.mock import MagicMock, patch + +import pytest +from google.auth.exceptions import DefaultCredentialsError +from osprey.worker.lib import gcp_credentials +from osprey.worker.lib.gcp_credentials import gcp_credentials_available, gcp_pubsub_disabled + + +@pytest.fixture(autouse=True) +def reset_cred_cache() -> Iterator[None]: + gcp_credentials._gcp_credentials_available = None + yield + gcp_credentials._gcp_credentials_available = None + + +def test_true_when_default_resolves() -> None: + with patch.object(gcp_credentials.google.auth, 'default', return_value=(MagicMock(), 'proj')): + assert gcp_credentials_available() is True + + +def test_false_when_default_raises() -> None: + with patch.object(gcp_credentials.google.auth, 'default', side_effect=DefaultCredentialsError()): + assert gcp_credentials_available() is False + + +def test_result_is_cached() -> None: + with patch.object(gcp_credentials.google.auth, 'default', return_value=(MagicMock(), 'proj')) as default_mock: + gcp_credentials_available() + gcp_credentials_available() + gcp_credentials_available() + assert default_mock.call_count == 1 + + +def test_pubsub_disabled_reads_env(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv('DISABLE_GCP_PUBSUB', raising=False) + assert gcp_pubsub_disabled() is False + monkeypatch.setenv('DISABLE_GCP_PUBSUB', 'true') + assert gcp_pubsub_disabled() is True + monkeypatch.setenv('DISABLE_GCP_PUBSUB', 'false') + assert gcp_pubsub_disabled() is False diff --git a/osprey_worker/src/osprey/worker/lib/tests/test_publisher.py b/osprey_worker/src/osprey/worker/lib/tests/test_publisher.py index 037cbd03..d4218385 100644 --- a/osprey_worker/src/osprey/worker/lib/tests/test_publisher.py +++ b/osprey_worker/src/osprey/worker/lib/tests/test_publisher.py @@ -1,40 +1,13 @@ -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 +from osprey.worker.lib.publisher import PubSubPublisher def test_pubsub_publisher_constructs_client_when_creds_present() -> None: with ( - patch.object(publisher.google.auth, 'default', return_value=(MagicMock(), 'proj')), + patch.object(publisher, 'gcp_credentials_available', return_value=True), patch.object(publisher, 'BatchPubsubPublisherClient') as client_cls, ): pub = PubSubPublisher('proj', 'topic') @@ -44,7 +17,7 @@ def test_pubsub_publisher_constructs_client_when_creds_present() -> None: 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, 'gcp_credentials_available', return_value=False), patch.object(publisher, 'BatchPubsubPublisherClient') as client_cls, ): with caplog.at_level('WARNING', logger=publisher.logger.name): @@ -61,7 +34,7 @@ def test_pubsub_publisher_disabled_via_env( ) -> None: monkeypatch.setenv('DISABLE_GCP_PUBSUB', 'true') with ( - patch.object(publisher, '_check_gcp_credentials') as cred_check, + patch.object(publisher, 'gcp_credentials_available') as cred_check, patch.object(publisher, 'BatchPubsubPublisherClient') as client_cls, ): with caplog.at_level('WARNING', logger=publisher.logger.name): @@ -75,7 +48,7 @@ def test_pubsub_publisher_disabled_via_env( def test_publish_short_circuits_silently_when_disabled() -> None: with ( - patch.object(publisher.google.auth, 'default', side_effect=DefaultCredentialsError()), + patch.object(publisher, 'gcp_credentials_available', return_value=False), patch.object(publisher, 'metrics') as metrics_mock, ): pub = PubSubPublisher('proj', 'topic') @@ -84,7 +57,7 @@ def test_publish_short_circuits_silently_when_disabled() -> None: def test_stop_short_circuits_when_disabled() -> None: - with patch.object(publisher.google.auth, 'default', side_effect=DefaultCredentialsError()): + with patch.object(publisher, 'gcp_credentials_available', return_value=False): pub = PubSubPublisher('proj', 'topic') pub.stop() assert not hasattr(pub, '_publisher')