diff --git a/pyproject.toml b/pyproject.toml index cd772ac5..d14031e8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -16,7 +16,7 @@ [tool.poetry] name = "solnlib" -version = "8.1.1" +version = "8.2.0-beta.1" description = "The Splunk Software Development Kit for Splunk Solutions" authors = ["Splunk "] license = "Apache-2.0" diff --git a/solnlib/__init__.py b/solnlib/__init__.py index 0fe61d4e..e18103d3 100644 --- a/solnlib/__init__.py +++ b/solnlib/__init__.py @@ -55,4 +55,4 @@ "utils", ] -__version__ = "8.1.1" +__version__ = "8.2.0-beta.1" diff --git a/solnlib/_observability_flush.py b/solnlib/_observability_flush.py new file mode 100644 index 00000000..92d3dec1 --- /dev/null +++ b/solnlib/_observability_flush.py @@ -0,0 +1,161 @@ +# +# Copyright 2026 Splunk Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +"""Private background worker for OpenTelemetry metric flush and shutdown.""" + +import atexit +import os +import threading +import time +from typing import Callable, Optional + +_EXIT_GRACE_SECONDS = 5.0 + + +def _remaining_ms(deadline: float) -> int: + return max(0, int((deadline - time.monotonic()) * 1000)) + + +class _AsyncFlushController: + """Serialize one provider's explicit flushes and final shutdown.""" + + def __init__( + self, provider, logger, format_error: Callable[[BaseException], str] + ) -> None: + self._provider = provider + self._logger = logger + self._format_error = format_error + self._condition = threading.Condition() + self._pending_timeout = None + self._shutdown_deadline = None + self._worker = threading.Thread( + target=self._run, name="SolnlibMetricFlush", daemon=True + ) + self._worker.start() + + def request_flush(self, timeout_millis: float) -> None: + with self._condition: + if self._shutdown_deadline is None: + self._pending_timeout = int(timeout_millis) + self._condition.notify() + + def request_shutdown(self, deadline: float) -> None: + with self._condition: + if self._shutdown_deadline is None: + self._shutdown_deadline = deadline + self._condition.notify() + + def join_until(self, deadline: float) -> None: + self._worker.join(timeout=max(0.0, deadline - time.monotonic())) + + def _run(self) -> None: + while True: + flush_timeout, shutdown_deadline = self._wait_for_request() + if shutdown_deadline is None: + if flush_timeout is not None: + self._flush(flush_timeout) + continue + self._shut_down(flush_timeout, shutdown_deadline) + return + + def _has_request(self) -> bool: + return self._pending_timeout is not None or self._shutdown_deadline is not None + + def _wait_for_request(self) -> tuple[Optional[int], Optional[float]]: + with self._condition: + self._condition.wait_for(self._has_request) + flush_timeout, self._pending_timeout = self._pending_timeout, None + return flush_timeout, self._shutdown_deadline + + def _shut_down(self, flush_timeout: Optional[int], deadline: float) -> None: + # Readers with an infinite export interval do not collect on shutdown. + remaining = _remaining_ms(deadline) + if flush_timeout is not None and remaining: + self._flush(min(flush_timeout, remaining)) + try: + self._provider.shutdown(timeout_millis=_remaining_ms(deadline)) + except Exception as error: + self._logger.info( + "Failed to shut down metrics: %s", self._format_error(error) + ) + + def _flush(self, timeout_ms: int) -> None: + try: + self._provider.force_flush(timeout_millis=timeout_ms) + except Exception as error: + self._logger.info("Failed to flush metrics: %s", self._format_error(error)) + + def restart_after_fork(self) -> None: + """Replace parent-thread state after fork in the child process.""" + self._condition = threading.Condition() + self._pending_timeout = None + self._shutdown_deadline = None + self._worker = threading.Thread( + target=self._run, name="SolnlibMetricFlush", daemon=True + ) + self._worker.start() + + +class _FlushRegistry: + """Coordinate all metric providers against one process-exit deadline.""" + + def __init__(self) -> None: + self._lock = threading.Lock() + self._controllers = [] + self._registered = False + self._closed = False + + def register(self, provider, logger, format_error) -> _AsyncFlushController: + with self._lock: + if self._closed: + raise RuntimeError("Metrics shutdown has started") + if not self._registered: + atexit.register(_shutdown_all) + if hasattr(os, "register_at_fork"): + os.register_at_fork(after_in_child=self._after_fork_child) + self._registered = True + controller = _AsyncFlushController(provider, logger, format_error) + self._controllers.append(controller) + return controller + + def shutdown_all(self) -> None: + deadline = time.monotonic() + _EXIT_GRACE_SECONDS + with self._lock: + if self._closed: + return + self._closed = True + controllers = tuple(self._controllers) + for controller in controllers: + controller.request_shutdown(deadline) + for controller in controllers: + controller.join_until(deadline) + + def _after_fork_child(self) -> None: + self._lock = threading.Lock() + if self._closed: + return + for controller in self._controllers: + controller.restart_after_fork() + + +_registry = _FlushRegistry() + + +def _register_controller(provider, logger, format_error) -> _AsyncFlushController: + return _registry.register(provider, logger, format_error) + + +def _shutdown_all() -> None: + _registry.shutdown_all() diff --git a/solnlib/observability.py b/solnlib/observability.py index 5c3eae7f..a8249e1f 100644 --- a/solnlib/observability.py +++ b/solnlib/observability.py @@ -31,8 +31,8 @@ - :class:`StanzaObservabilityRecorder` — a stanza-scoped recorder that wraps ``ObservabilityService`` with a per-process singleton cache. Bind it to a single stanza name and call :meth:`~StanzaObservabilityRecorder.record` after - each batch of ingested events. Use as a context manager for automatic flush - on exit. + each batch of ingested events. Use as a context manager to schedule a + background flush on exit. Typical usage:: @@ -52,6 +52,7 @@ import threading import urllib.request from typing import Callable, ClassVar, Optional, Union +from ._observability_flush import _register_controller from .splunkenv import get_conf_stanzas from opentelemetry.metrics import Instrument, Meter from opentelemetry.sdk.metrics import MeterProvider, Counter, Histogram @@ -410,10 +411,12 @@ def __init__( self.event_bytes_counter: Optional[Counter] = None self._meter: Optional[Meter] = None self._provider: Optional[MeterProvider] = None + self._flush_controller = None for logger_name in _METRICS_SDK_LOGGERS: logging.getLogger(logger_name).addFilter(_downgrade_to_info_filter) + provider = None try: if type(modinput_type) is not str or not _is_safe_identifier_str( modinput_type @@ -451,34 +454,54 @@ def __init__( for exporter in extra_exporters or []: metric_readers.append(PeriodicExportingMetricReader(exporter)) - self._provider = MeterProvider( - resource=resource, metric_readers=metric_readers + provider = MeterProvider( + resource=resource, + metric_readers=metric_readers, + shutdown_on_exit=False, ) - self._meter = self._provider.get_meter(ta_name, ta_version) + meter = provider.get_meter(ta_name, ta_version) - self.event_count_counter = self._meter.create_counter( + event_count_counter = meter.create_counter( name="splunk.addon.events", description="Number of events ingested by the add-on modular input", unit="1", ) - self.event_bytes_counter = self._meter.create_counter( + event_bytes_counter = meter.create_counter( name="splunk.addon.events.bytes", description="Volume of data ingested by the add-on modular input", unit="By", ) - self._logger.info( - "ObservabilityService initialised: ta_name=%s ta_version=%s " - "modinput_type=%s", - ta_name, - ta_version, - modinput_type, + controller = _register_controller( + provider, self._logger, _safe_exception_str ) except Exception as error: + if provider is not None: + try: + provider.shutdown() + except Exception as cleanup_error: + self._logger.info( + "Failed to clean up metrics after init failure: %s", + _safe_exception_str(cleanup_error), + ) self._logger.info( "Failed to initialise ObservabilityService: %s", _safe_exception_str(error), ) + return + + self._provider = provider + self._meter = meter + self.event_count_counter = event_count_counter + self.event_bytes_counter = event_bytes_counter + self._flush_controller = controller + self._logger.info( + "ObservabilityService initialised: ta_name=%s ta_version=%s " + "modinput_type=%s", + ta_name, + ta_version, + modinput_type, + ) def _read_ta_info(self) -> tuple[Optional[str], Optional[str]]: """Read the add-on name and version from ``app.conf``. @@ -657,6 +680,7 @@ def _create_otlp_exporter(self) -> Optional[MetricExporter]: exporter = OTLPMetricExporter( endpoint=endpoint, credentials=credentials, + timeout=2.0, preferred_temporality={ Counter: AggregationTemporality.DELTA, Histogram: AggregationTemporality.DELTA, @@ -711,26 +735,34 @@ def register_instrument( return callback(self._meter) def flush(self, timeout_millis: float = 30_000) -> None: - """Force-flush all metric readers. + """Schedule a background flush of all metric readers. - Blocks until all buffered data points have been handed off to their - exporters or *timeout_millis* elapses. Call this before the modular - input process exits to avoid dropping the last batch of metrics. + Returns promptly without waiting for an export attempt to finish. + Repeated calls while an export is active are coalesced. At normal + process exit, solnlib gives its metric workers a single five-second + grace period for final export; the last metrics can be lost if it + expires. Prefer using :class:`StanzaObservabilityRecorder` as a context manager rather than calling this method directly — it calls :meth:`StanzaObservabilityRecorder.flush` on exit automatically. Args: - timeout_millis: Maximum time to wait for exporters to drain, in - milliseconds. Defaults to 30 seconds. + timeout_millis: Budget forwarded to the background + ``MeterProvider.force_flush()`` call, in milliseconds. + The budget starts when that call begins; waiting for the + worker is not included. Readers and exporters may exceed + this budget. It is not a hard limit on export duration. + Defaults to 30 seconds. """ - if self._provider is None: + if self._provider is None or self._flush_controller is None: return try: - self._provider.force_flush(timeout_millis=int(timeout_millis)) + self._flush_controller.request_flush(timeout_millis) except Exception as error: - self._logger.info("Failed to flush metrics: %s", _safe_exception_str(error)) + self._logger.info( + "Failed to schedule metrics flush: %s", _safe_exception_str(error) + ) _INT64_MIN = -(2**63) @@ -806,7 +838,8 @@ class StanzaObservabilityRecorder: **Best practices:** - Use as a context manager (``with`` statement) so that :meth:`flush` is - always called when the stanza collection loop exits, even on exceptions. + always scheduled when the stanza collection loop exits, even on + exceptions. This does not wait for export completion. - Pass the same *modinput_type* string for all stanzas of the same input type. The string should be lowercase, hyphenated, and stable across restarts (e.g. ``"event-hub"``). @@ -1013,15 +1046,15 @@ def record( self._service.event_bytes_counter.add(byte_count, attributes=attrs) def flush(self) -> None: - """Force-flush all metric readers. + """Schedule a background flush of all metric readers. - Delegates to :meth:`ObservabilityService.flush` (which calls - ``MeterProvider.force_flush()`` internally). Called automatically by - ``__exit__`` when the recorder is used as a context manager, so you - rarely need to call this directly. + Delegates to :meth:`ObservabilityService.flush`, which calls + ``MeterProvider.force_flush()`` on a daemon worker. Called + automatically by ``__exit__`` when used as a context manager. A + returned call means the attempt was scheduled, not completed. Call it explicitly only when you are not using the context manager and - need to guarantee delivery before the process exits:: + want to request an export before the process exits:: obs = StanzaObservabilityRecorder("my-input", logger, stanza_name) try: @@ -1036,10 +1069,10 @@ def __enter__(self) -> "StanzaObservabilityRecorder": return self def __exit__(self, *_) -> bool: - """Flush all metric readers and allow exceptions to propagate. + """Schedule a flush and allow exceptions to propagate. Returns ``False`` so any exception raised inside the ``with`` block - is re-raised after flushing. + is re-raised after the flush has been scheduled. """ self.flush() return False diff --git a/tests/unit/helpers/observability_flush_process.py b/tests/unit/helpers/observability_flush_process.py new file mode 100644 index 00000000..d92af18f --- /dev/null +++ b/tests/unit/helpers/observability_flush_process.py @@ -0,0 +1,86 @@ +# +# Copyright 2026 Splunk Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +"""Executable scenarios for process-level observability regression tests.""" + +import logging +import os +import sys +import threading + +from solnlib import _observability_flush as flushmod + + +def blocked_otel_exit(): + from opentelemetry.sdk.metrics.export import MetricExporter, MetricExportResult + + from solnlib.observability import ObservabilityService + + class SlowExporter(MetricExporter): + def export(self, metrics_data, timeout_millis=10_000, **kwargs): + print("OTEL_EXPORT_ENTERED", flush=True) + threading.Event().wait(10) + return MetricExportResult.SUCCESS + + def shutdown(self, timeout_millis=30_000, **kwargs): + pass + + def force_flush(self, timeout_millis=10_000): + return True + + flushmod._EXIT_GRACE_SECONDS = 0.5 + ObservabilityService._create_otlp_exporter = lambda self: None + service = ObservabilityService( + "test-input", + logging.getLogger("test"), + ta_name="test-ta", + ta_version="1.0.0", + extra_exporters=[SlowExporter()], + ) + service.event_count_counter.add(1) + print("MAIN_DONE", flush=True) + return 0 + + +def fork_worker(): + class Provider: + def force_flush(self, **kwargs): + pass + + def shutdown(self, **kwargs): + pass + + controller = flushmod._register_controller( + Provider(), logging.getLogger("test"), str + ) + parent_worker = controller._worker + pid = os.fork() + if pid == 0: + fresh = ( + controller._worker is not parent_worker and controller._worker.is_alive() + ) + print("CHILD_FRESH_WORKER=" + str(fresh), flush=True) + return 0 if fresh else 2 + _, status = os.waitpid(pid, 0) + print("PARENT_WORKER_ALIVE=" + str(parent_worker.is_alive()), flush=True) + return os.waitstatus_to_exitcode(status) + + +if __name__ == "__main__": + scenarios = { + "blocked-otel-exit": blocked_otel_exit, + "fork-worker": fork_worker, + } + sys.exit(scenarios[sys.argv[1]]()) diff --git a/tests/unit/test_observability.py b/tests/unit/test_observability.py index 84c907f9..fd53a367 100644 --- a/tests/unit/test_observability.py +++ b/tests/unit/test_observability.py @@ -633,6 +633,84 @@ def test_meter_is_available(self, logger, monkeypatch): # Assert assert svc._meter is not None + @pytest.mark.parametrize("failure_stage", ["counter", "registration"]) + @pytest.mark.parametrize("cleanup_fails", [False, True]) + def test_failed_initialisation_leaves_service_disabled( + self, logger, monkeypatch, failure_stage, cleanup_fails + ): + # Arrange + provider = MagicMock() + callback = MagicMock() + monkeypatch.setattr( + "solnlib.observability.MeterProvider", MagicMock(return_value=provider) + ) + register = MagicMock() + monkeypatch.setattr("solnlib.observability._register_controller", register) + error = RuntimeError("initialisation failure") + if failure_stage == "counter": + provider.get_meter.return_value.create_counter.side_effect = [ + MagicMock(), + error, + ] + else: + register.side_effect = error + if cleanup_fails: + provider.shutdown.side_effect = RuntimeError("cleanup failure") + + # Act + svc = _make_service(logger, monkeypatch) + instrument = svc.register_instrument(callback) + flush_result = svc.flush() + + # Assert + for attribute in ( + "_provider", + "_meter", + "_flush_controller", + "event_count_counter", + "event_bytes_counter", + ): + assert getattr(svc, attribute) is None + provider.shutdown.assert_called_once_with() + assert instrument is None + assert flush_result is None + callback.assert_not_called() + register.return_value.request_flush.assert_not_called() + assert any( + "initialisation failure" in call.args for call in logger.info.call_args_list + ) + + def test_closed_registry_does_not_raise_from_public_service( + self, logger, monkeypatch + ): + # Arrange + from solnlib import _observability_flush as flushmod + + registry = flushmod._FlushRegistry() + registry.shutdown_all() + monkeypatch.setattr(flushmod, "_registry", registry) + callback = MagicMock() + + # Act + svc = _make_service(logger, monkeypatch) + instrument = svc.register_instrument(callback) + flush_result = svc.flush() + + # Assert + assert svc._provider is None + assert svc._meter is None + assert svc._flush_controller is None + assert svc.event_count_counter is None + assert svc.event_bytes_counter is None + assert instrument is None + assert flush_result is None + callback.assert_not_called() + assert registry._controllers == [] + assert any( + "Metrics shutdown has started" in call.args + for call in logger.info.call_args_list + ) + def test_extra_exporter_is_added(self, logger, monkeypatch): # Arrange extra = MagicMock() @@ -905,7 +983,7 @@ def test_create_otlp_exporter_wrapper_forwards_shutdown( result.shutdown(timeout_millis=1234) mock_exporter.shutdown.assert_called_once_with(timeout_millis=1234) - def test_create_otlp_exporter_uses_delta_temporality( + def test_create_otlp_exporter_uses_delta_temporality_and_bounded_timeout( self, logger, monkeypatch, tmp_path ): # Arrange @@ -940,19 +1018,37 @@ def test_create_otlp_exporter_uses_delta_temporality( ObservabilityService._create_otlp_exporter(svc) # Assert — preferred_temporality must set DELTA for Counter and Histogram _, kwargs = mock_exporter_cls.call_args + assert kwargs["timeout"] == 2.0 temporality = kwargs["preferred_temporality"] assert temporality[Counter] == AggregationTemporality.DELTA assert temporality[Histogram] == AggregationTemporality.DELTA - def test_flush_calls_provider_force_flush(self, logger, monkeypatch): + def test_flush_schedules_without_calling_provider_on_caller( + self, logger, monkeypatch + ): # Arrange - svc = _make_service(logger, monkeypatch) - mock_provider = MagicMock() - svc._provider = mock_provider + controller = MagicMock() + provider_cls = MagicMock() + monkeypatch.setattr("solnlib.observability.MeterProvider", provider_cls) + monkeypatch.setattr( + "solnlib.observability.ObservabilityService._create_otlp_exporter", + lambda self: None, + ) + monkeypatch.setattr( + "solnlib.observability._register_controller", + lambda provider, logger, formatter: controller, + ) + svc = ObservabilityService( + "test-input", logger, ta_name="my_ta", ta_version="1.0.0" + ) + # Act svc.flush(timeout_millis=5_000) + # Assert - mock_provider.force_flush.assert_called_once_with(timeout_millis=5000) + controller.request_flush.assert_called_once_with(5_000) + provider_cls.return_value.force_flush.assert_not_called() + assert provider_cls.call_args.kwargs["shutdown_on_exit"] is False def test_flush_is_noop_when_provider_missing(self, logger, monkeypatch): # Arrange @@ -961,16 +1057,17 @@ def test_flush_is_noop_when_provider_missing(self, logger, monkeypatch): # Act / Assert — must not raise svc.flush() - def test_flush_logs_info_on_exception(self, logger, monkeypatch): + def test_flush_logs_info_when_scheduling_fails(self, logger, monkeypatch): # Arrange svc = _make_service(logger, monkeypatch) - mock_provider = MagicMock() - mock_provider.force_flush.side_effect = RuntimeError("boom") - svc._provider = mock_provider + svc._flush_controller = MagicMock() + svc._flush_controller.request_flush.side_effect = RuntimeError("boom") + # Act svc.flush() + # Assert - logger.info.assert_called() + assert any("boom" in call.args for call in logger.info.call_args_list) logger.warning.assert_not_called() def test_init_attaches_filter_to_metrics_sdk_loggers(self, logger, monkeypatch): @@ -1554,16 +1651,20 @@ def test_create_otlp_exporter_exception_uses_safe_str_no_exc_info( assert "exc_info" not in kwargs logger.warning.assert_not_called() - def test_flush_exception_uses_safe_str(self, logger, monkeypatch): + def test_flush_scheduling_error_uses_safe_str(self, logger, monkeypatch): + # Arrange monkeypatch.setattr( "solnlib.observability._safe_exception_str", lambda error: "SAFE" ) svc = _make_service(logger, monkeypatch) - mock_provider = MagicMock() - mock_provider.force_flush.side_effect = RuntimeError("boom") - svc._provider = mock_provider + svc._flush_controller = MagicMock() + svc._flush_controller.request_flush.side_effect = RuntimeError("boom") logger.reset_mock() + + # Act svc.flush() + + # Assert assert any("SAFE" in call.args for call in logger.info.call_args_list) logger.warning.assert_not_called() diff --git a/tests/unit/test_observability_flush.py b/tests/unit/test_observability_flush.py new file mode 100644 index 00000000..5b490395 --- /dev/null +++ b/tests/unit/test_observability_flush.py @@ -0,0 +1,344 @@ +# +# Copyright 2026 Splunk Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +import logging +import threading +import time +from unittest.mock import MagicMock + +import pytest +from opentelemetry.sdk.metrics.export import MetricExporter, MetricExportResult + +from solnlib import _observability_flush as flushmod +from solnlib._observability_flush import _AsyncFlushController +from solnlib.observability import ObservabilityService + + +@pytest.fixture +def controller_factory(): + controllers = [] + + def create(provider, logger=None): + controller = _AsyncFlushController( + provider, logger or MagicMock(spec=logging.Logger), str + ) + controllers.append(controller) + return controller + + yield create + + deadline = time.monotonic() + 1 + for controller in controllers: + controller.request_shutdown(deadline) + for controller in controllers: + controller.join_until(deadline) + + +@pytest.fixture +def registry_and_callbacks(monkeypatch): + registry = flushmod._FlushRegistry() + callbacks = [] + + def register_callback(callback): + callbacks.append(callback) + return callback + + monkeypatch.setattr(flushmod, "_registry", registry) + monkeypatch.setattr(flushmod.atexit, "register", register_callback) + if hasattr(flushmod.os, "register_at_fork"): + monkeypatch.setattr(flushmod.os, "register_at_fork", MagicMock()) + + yield registry, callbacks + + registry.shutdown_all() + + +def test_request_flush_returns_while_export_is_blocked(controller_factory): + # Arrange + entered = threading.Event() + release = threading.Event() + provider = MagicMock() + + def blocked_flush(**kwargs): + entered.set() + release.wait() + + provider.force_flush.side_effect = blocked_flush + controller = controller_factory(provider) + try: + # Act + start = time.monotonic() + controller.request_flush(30_000) + elapsed = time.monotonic() - start + export_started = entered.wait(1) + + # Assert + assert elapsed < 0.2 + assert export_started + provider.force_flush.assert_called_once_with(timeout_millis=30_000) + finally: + release.set() + + +def test_repeated_requests_during_export_coalesce_to_one_followup(controller_factory): + # Arrange + entered = threading.Event() + release = threading.Event() + second_done = threading.Event() + provider = MagicMock() + + def flush(**kwargs): + if provider.force_flush.call_count == 1: + entered.set() + release.wait() + else: + second_done.set() + + provider.force_flush.side_effect = flush + controller = controller_factory(provider) + try: + # Act + controller.request_flush(30_000) + first_started = entered.wait(1) + controller.request_flush(10_000) + controller.request_flush(5_000) + release.set() + followup_finished = second_done.wait(1) + + # Assert + assert first_started + assert followup_finished + assert provider.force_flush.call_count == 2 + assert provider.force_flush.call_args_list[-1].kwargs == { + "timeout_millis": 5_000 + } + finally: + release.set() + + +def test_worker_logs_export_error_without_raising_to_caller(controller_factory): + # Arrange + logged = threading.Event() + provider = MagicMock() + provider.force_flush.side_effect = RuntimeError("boom") + logger = MagicMock(spec=logging.Logger) + logger.info.side_effect = lambda *args: logged.set() + controller = controller_factory(provider, logger) + + # Act + controller.request_flush(30_000) + error_logged = logged.wait(1) + + # Assert + assert error_logged + assert "boom" in logger.info.call_args.args[-1] + + +def test_shutdown_is_once_and_later_flush_is_ignored(controller_factory): + # Arrange + provider = MagicMock() + controller = controller_factory(provider) + deadline = time.monotonic() + 1 + + # Act + controller.request_shutdown(deadline) + controller.request_shutdown(deadline) + controller.join_until(deadline) + controller.request_flush(30_000) + + # Assert + provider.shutdown.assert_called_once() + provider.force_flush.assert_not_called() + assert not controller._worker.is_alive() + + +def test_exit_budget_is_shared_by_two_controllers(monkeypatch, registry_and_callbacks): + # Arrange + monkeypatch.setattr(flushmod, "_EXIT_GRACE_SECONDS", 0.05) + registry, _ = registry_and_callbacks + release = threading.Event() + entered = [threading.Event(), threading.Event()] + controllers = [] + for gate in entered: + provider = MagicMock() + provider.shutdown.side_effect = lambda gate=gate, **kwargs: ( + gate.set(), + release.wait(), + ) + controllers.append(registry.register(provider, MagicMock(), str)) + try: + # Act + start = time.monotonic() + registry.shutdown_all() + elapsed = time.monotonic() - start + + # Assert + assert elapsed < 0.2 + assert all(gate.wait(1) for gate in entered) + finally: + release.set() + for controller in controllers: + controller.join_until(time.monotonic() + 1) + + +def test_one_atexit_registration_for_multiple_controllers(registry_and_callbacks): + # Arrange + registry, callbacks = registry_and_callbacks + + # Act + registry.register(MagicMock(), MagicMock(), str) + registry.register(MagicMock(), MagicMock(), str) + + # Assert + assert callbacks == [flushmod._shutdown_all] + + +def test_closed_registry_does_not_start_a_worker(monkeypatch, registry_and_callbacks): + # Arrange + registry, _ = registry_and_callbacks + registry.shutdown_all() + controller_cls = MagicMock() + monkeypatch.setattr(flushmod, "_AsyncFlushController", controller_cls) + + # Act + with pytest.raises(RuntimeError, match="shutdown has started"): + flushmod._register_controller(MagicMock(), MagicMock(), str) + + # Assert + controller_cls.assert_not_called() + assert registry._controllers == [] + + +def test_closed_registry_does_not_restart_workers_after_fork(registry_and_callbacks): + # Arrange + registry, _ = registry_and_callbacks + controller = registry.register(MagicMock(), MagicMock(), str) + registry.shutdown_all() + original_worker = controller._worker + try: + # Act + registry._after_fork_child() + + # Assert + assert controller._worker is original_worker + assert not controller._worker.is_alive() + finally: + deadline = time.monotonic() + 1 + controller.request_shutdown(deadline) + controller.join_until(deadline) + + +def test_exit_handler_cannot_register_a_worker_after_registry_shutdown( + registry_and_callbacks, +): + # Arrange + registry, callbacks = registry_and_callbacks + registration_errors = [] + + def late_handler(): + try: + flushmod._register_controller(MagicMock(), MagicMock(), str) + except RuntimeError as error: + registration_errors.append(error) + + callbacks.append(late_handler) + controller = flushmod._register_controller(MagicMock(), MagicMock(), str) + + # Act + for callback in reversed(callbacks): + callback() + + # Assert + assert len(registration_errors) == 1 + assert registry._controllers == [controller] + assert not controller._worker.is_alive() + + +def test_registered_exit_callback_attempts_shutdown_without_explicit_flush( + registry_and_callbacks, +): + # Arrange + _, callbacks = registry_and_callbacks + provider = MagicMock() + controller = flushmod._register_controller(provider, MagicMock(), str) + + # Act + callbacks[0]() + + # Assert + provider.shutdown.assert_called_once() + assert not controller._worker.is_alive() + + +def test_shutdown_exports_pending_flush_without_periodic_reader( + monkeypatch, registry_and_callbacks +): + # Arrange + monkeypatch.setenv("OTEL_METRIC_EXPORT_INTERVAL", "inf") + monkeypatch.setattr( + ObservabilityService, "_create_otlp_exporter", lambda self: None + ) + entered = threading.Event() + release = threading.Event() + exported_values = [] + + class RecordingExporter(MetricExporter): + def export(self, metrics_data, timeout_millis=10_000, **kwargs): + value = sum( + point.value + for resource in metrics_data.resource_metrics + for scope in resource.scope_metrics + for metric in scope.metrics + if metric.name == "splunk.addon.events" + for point in metric.data.data_points + ) + exported_values.append(value) + if len(exported_values) == 1: + entered.set() + release.wait() + return MetricExportResult.SUCCESS + + def shutdown(self, timeout_millis=30_000, **kwargs): + pass + + def force_flush(self, timeout_millis=10_000): + return True + + service = ObservabilityService( + "test-input", + MagicMock(spec=logging.Logger), + ta_name="test-ta", + ta_version="1.0.0", + extra_exporters=[RecordingExporter()], + ) + try: + # Act + service.event_count_counter.add(1) + service.flush() + first_export_started = entered.wait(1) + service.event_count_counter.add(2) + service.flush() + deadline = time.monotonic() + 1 + service._flush_controller.request_shutdown(deadline) + release.set() + service._flush_controller.join_until(deadline) + + # Assert + assert first_export_started + assert exported_values == [1, 3] + assert not service._flush_controller._worker.is_alive() + finally: + release.set() diff --git a/tests/unit/test_observability_lifecycle.py b/tests/unit/test_observability_lifecycle.py new file mode 100644 index 00000000..02b08220 --- /dev/null +++ b/tests/unit/test_observability_lifecycle.py @@ -0,0 +1,72 @@ +# +# Copyright 2026 Splunk Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +"""Process-level regressions that require interpreter shutdown or a real fork.""" + +import os +import subprocess +import sys +import time +from pathlib import Path + +import pytest + + +_REPO_ROOT = Path(__file__).resolve().parents[2] +_SCENARIO_SCRIPT = ( + Path(__file__).with_name("helpers") / "observability_flush_process.py" +) + + +def test_interpreter_exit_is_bounded_with_a_blocked_otel_export(): + # Arrange + env = dict(os.environ, PYTHONPATH=str(_REPO_ROOT)) + start = time.monotonic() + + # Act + result = subprocess.run( + [sys.executable, str(_SCENARIO_SCRIPT), "blocked-otel-exit"], + capture_output=True, + text=True, + env=env, + timeout=5, + ) + elapsed = time.monotonic() - start + + # Assert + assert result.returncode == 0, result.stderr + assert "MAIN_DONE" in result.stdout + assert "OTEL_EXPORT_ENTERED" in result.stdout + assert elapsed < 3 + + +@pytest.mark.skipif(not hasattr(os, "fork"), reason="fork is unavailable") +def test_child_gets_fresh_worker_after_fork(): + # Arrange + env = dict(os.environ, PYTHONPATH=str(_REPO_ROOT)) + + # Act + result = subprocess.run( + [sys.executable, str(_SCENARIO_SCRIPT), "fork-worker"], + capture_output=True, + text=True, + env=env, + timeout=5, + ) + + # Assert + assert result.returncode == 0, result.stderr + assert "CHILD_FRESH_WORKER=True" in result.stdout + assert "PARENT_WORKER_ALIVE=True" in result.stdout