From 0c34edde594e79118a42d0a3c7c5ee9de00330a8 Mon Sep 17 00:00:00 2001 From: Daniel Visca Date: Mon, 20 Jul 2026 13:41:01 -0700 Subject: [PATCH 1/2] feat(metrics): make the posthoganalytics SDK metrics API bump-ready for web and celery Pre-lands the wiring so client.metrics works the moment posthoganalytics is bumped to >=7.23: apps.py sets the module-level metrics config (service name, version, environment) that setup() will consume, and the celery worker_process_shutdown handler flushes the SDK's final aggregation window so recycled prefork children don't drop their tail samples. Both changes are inert on the currently pinned SDK version, which predates the metrics API. Generated-By: PostHog Code Task-Id: 470062dd-05c0-40bf-bfa3-bc89b194121e --- posthog/apps.py | 9 +++++++++ posthog/celery.py | 14 +++++++++++++ posthog/test/test_celery.py | 40 ++++++++++++++++++++++++++++++++++++- 3 files changed, 62 insertions(+), 1 deletion(-) diff --git a/posthog/apps.py b/posthog/apps.py index a080decbd310..9bb164384762 100644 --- a/posthog/apps.py +++ b/posthog/apps.py @@ -73,6 +73,15 @@ def ready(self): "service": settings.OTEL_SERVICE_NAME, "environment": os.getenv("OTEL_SERVICE_ENVIRONMENT"), } + # Config for the SDK's `client.metrics` API. The pinned SDK version predates + # the metrics API and ignores this attr; once posthoganalytics is bumped to + # >=7.23 it's picked up by setup(), so metrics get a real service.name + # instead of 'unknown_service'. + posthoganalytics.metrics = { # type: ignore[attr-defined] + "service_name": settings.OTEL_SERVICE_NAME or "posthog", + "service_version": os.getenv("COMMIT_SHA"), + "environment": os.getenv("OTEL_SERVICE_ENVIRONMENT"), + } if str_to_bool(os.environ.get("TEMPORAL_DISABLE_EXCEPTION_VARIABLE_CAPTURE", "false")): posthoganalytics.capture_exception_code_variables = False diff --git a/posthog/celery.py b/posthog/celery.py index 8cdc55719f55..4fe880ee60fa 100644 --- a/posthog/celery.py +++ b/posthog/celery.py @@ -227,6 +227,20 @@ def on_worker_process_shutdown(**kwargs) -> None: multiprocess.mark_process_dead(os.getpid()) + # Flush the posthoganalytics SDK's final metrics window: `client.metrics` + # aggregates in memory and flushes on an interval, so a recycled child + # (--max-tasks-per-child) would otherwise drop up to one interval of samples. + # Inert on SDK versions without the metrics API and on untouched/disabled clients. + import posthoganalytics # noqa: PLC0415 — keep the SDK off the celery import path + + try: + client = posthoganalytics.default_client + metrics = getattr(client, "metrics", None) if client is not None else None + if metrics is not None: + metrics.flush() + except Exception: + logger.warning("posthoganalytics_metrics_flush_failed", exc_info=True) + # Set up clickhouse query instrumentation @task_prerun.connect diff --git a/posthog/test/test_celery.py b/posthog/test/test_celery.py index 5b315b3b7176..4701fb461fbb 100644 --- a/posthog/test/test_celery.py +++ b/posthog/test/test_celery.py @@ -1,9 +1,47 @@ from unittest import TestCase -from unittest.mock import patch +from unittest.mock import MagicMock, patch +import posthoganalytics +from parameterized import parameterized + +from posthog.celery import on_worker_process_shutdown from posthog.tasks.tasks import clickhouse_errors_count +class TestWorkerShutdownFlushesAnalyticsMetrics(TestCase): + def test_flushes_sdk_metrics_tail_window(self) -> None: + client = MagicMock() + with patch.object(posthoganalytics, "default_client", client): + on_worker_process_shutdown() + client.metrics.flush.assert_called_once() + + @parameterized.expand( + [ + ("no_default_client", lambda: None), + # The pinned SDK version has no `metrics` API — the hook must stay + # inert (a bare `client.metrics.flush()` would raise on every + # worker recycle until the dependency is bumped). + ( + "real_client_on_pinned_sdk_version", + lambda: posthoganalytics.Client("phc_test", sync_mode=True, disabled=True), + ), + ("flush_raises", lambda: MagicMock(**{"metrics.flush.side_effect": RuntimeError("network down")})), + ] + ) + def test_handler_never_breaks_worker_shutdown(self, _name: str, client_factory) -> None: + with patch.object(posthoganalytics, "default_client", client_factory()): + on_worker_process_shutdown() + + +class TestAnalyticsMetricsConfig(TestCase): + def test_apps_ready_configures_module_level_metrics(self) -> None: + # Deleting the "unused" attr assignment in apps.py before the SDK bump + # would make the bump silently ship service_name='unknown_service'. + config = getattr(posthoganalytics, "metrics", None) + assert isinstance(config, dict) + assert config["service_name"] + + class TestCeleryMetrics(TestCase): @patch("posthog.clickhouse.client.sync_execute") @patch("posthog.metrics.push_to_gateway") From 99c2508a21c7b03778a3f23e109c481f1f753217 Mon Sep 17 00:00:00 2001 From: Daniel Visca Date: Tue, 21 Jul 2026 14:05:08 -0700 Subject: [PATCH 2/2] fix(metrics): bound celery shutdown flush and align service-name fallback Generated-By: PostHog Code Task-Id: 934a73f2-8545-4575-b4b2-497a4c181f9c --- posthog/apps.py | 4 +++- posthog/celery.py | 19 ++++++++++++++++++- posthog/test/test_celery.py | 23 +++++++++++++++++++++++ 3 files changed, 44 insertions(+), 2 deletions(-) diff --git a/posthog/apps.py b/posthog/apps.py index 9bb164384762..1696bc61f83c 100644 --- a/posthog/apps.py +++ b/posthog/apps.py @@ -78,7 +78,9 @@ def ready(self): # >=7.23 it's picked up by setup(), so metrics get a real service.name # instead of 'unknown_service'. posthoganalytics.metrics = { # type: ignore[attr-defined] - "service_name": settings.OTEL_SERVICE_NAME or "posthog", + # Same fallback as the OTel trace resource (otel_instrumentation.py) — + # metrics and traces from one process must share a service identity. + "service_name": settings.OTEL_SERVICE_NAME or "posthog-django-default", "service_version": os.getenv("COMMIT_SHA"), "environment": os.getenv("OTEL_SERVICE_ENVIRONMENT"), } diff --git a/posthog/celery.py b/posthog/celery.py index 4fe880ee60fa..9579934b4520 100644 --- a/posthog/celery.py +++ b/posthog/celery.py @@ -1,6 +1,7 @@ import os import time import errno +import threading from django.dispatch import receiver @@ -219,6 +220,9 @@ def on_worker_start(**kwargs) -> None: _initialize_worker_metrics() +_ANALYTICS_METRICS_FLUSH_TIMEOUT_SECONDS = 5.0 + + @worker_process_shutdown.connect def on_worker_process_shutdown(**kwargs) -> None: """Remove metric files for this child so recycled workers don't leak stale data.""" @@ -237,7 +241,20 @@ def on_worker_process_shutdown(**kwargs) -> None: client = posthoganalytics.default_client metrics = getattr(client, "metrics", None) if client is not None else None if metrics is not None: - metrics.flush() + # Bound the wait: an unreachable metrics endpoint must not hold a + # recycling child hostage (same hazard otel_instrumentation.py caps + # with a 5s force_flush). The daemon thread is abandoned on timeout. + def _flush() -> None: + try: + metrics.flush() + except Exception: + logger.warning("posthoganalytics_metrics_flush_failed", exc_info=True) + + flush_thread = threading.Thread(target=_flush, name="posthoganalytics-metrics-flush", daemon=True) + flush_thread.start() + flush_thread.join(timeout=_ANALYTICS_METRICS_FLUSH_TIMEOUT_SECONDS) + if flush_thread.is_alive(): + logger.warning("posthoganalytics_metrics_flush_timed_out") except Exception: logger.warning("posthoganalytics_metrics_flush_failed", exc_info=True) diff --git a/posthog/test/test_celery.py b/posthog/test/test_celery.py index 4701fb461fbb..3d09fa1d4fb5 100644 --- a/posthog/test/test_celery.py +++ b/posthog/test/test_celery.py @@ -1,9 +1,12 @@ +import threading + from unittest import TestCase from unittest.mock import MagicMock, patch import posthoganalytics from parameterized import parameterized +import posthog.celery from posthog.celery import on_worker_process_shutdown from posthog.tasks.tasks import clickhouse_errors_count @@ -15,6 +18,26 @@ def test_flushes_sdk_metrics_tail_window(self) -> None: on_worker_process_shutdown() client.metrics.flush.assert_called_once() + def test_hung_flush_does_not_stall_worker_recycling(self) -> None: + release = threading.Event() + flush_completed = threading.Event() + + def hung_flush() -> None: + release.wait(timeout=10) + flush_completed.set() + + client = MagicMock(**{"metrics.flush.side_effect": hung_flush}) + try: + with ( + patch.object(posthog.celery, "_ANALYTICS_METRICS_FLUSH_TIMEOUT_SECONDS", 0.05), + patch.object(posthoganalytics, "default_client", client), + ): + on_worker_process_shutdown() + # The handler must abandon the hung flush, not wait it out. + assert not flush_completed.is_set() + finally: + release.set() + @parameterized.expand( [ ("no_default_client", lambda: None),