From 9dbb11acd823fb54cf76cf06628ec6e5290f27f8 Mon Sep 17 00:00:00 2001 From: wadii Date: Wed, 12 Aug 2026 16:34:04 +0200 Subject: [PATCH 1/4] fix(experimentation): retry experiment computes on transient ClickHouse errors --- api/experimentation/services.py | 38 +++++++--- api/experimentation/tasks.py | 14 +++- .../unit/experimentation/test_services.py | 56 ++++++++++++-- api/tests/unit/experimentation/test_tasks.py | 75 +++++++++++++++++++ api/tests/unit/experimentation/test_views.py | 2 +- .../observability/_events-catalogue.md | 32 ++++---- 6 files changed, 180 insertions(+), 37 deletions(-) diff --git a/api/experimentation/services.py b/api/experimentation/services.py index ee8c1c2bff75..d91e4ad02142 100644 --- a/api/experimentation/services.py +++ b/api/experimentation/services.py @@ -2,10 +2,10 @@ import hashlib import json +import threading import time import typing from dataclasses import replace -from functools import lru_cache import structlog from clickhouse_connect.driver.exceptions import ClickHouseError @@ -104,6 +104,7 @@ CLICKHOUSE_CONNECT_TIMEOUT_SECONDS = 5 CLICKHOUSE_QUERY_TIMEOUT_SECONDS = 30 +CLICKHOUSE_BACKGROUND_QUERY_TIMEOUT_SECONDS = 120 CLICKHOUSE_VERIFY_TIMEOUT_SECONDS = 5 CLICKHOUSE_EVENT_NAMES_TIMEOUT_SECONDS = 15 CUSTOMER_EVENT_STATS_CACHE_SECONDS = 60 @@ -148,19 +149,30 @@ def is_experiment_feature_enabled(organisation: Organisation) -> bool: ) -@lru_cache(maxsize=1) -def _get_clickhouse_client() -> Client: +_clickhouse_clients = threading.local() + + +def _get_clickhouse_client( + send_receive_timeout: int = CLICKHOUSE_QUERY_TIMEOUT_SECONDS, +) -> Client: """Build a clickhouse-driver client for the experimentation event store. The database is taken from the DSN path, so queries can reference the `events` table unqualified. Connect and query timeouts are bounded unless the DSN overrides them. + + clickhouse-driver clients are not thread-safe, so one client is cached per + thread and requested timeout. """ - host, kwargs = parse_url(settings.EXPERIMENTATION_CLICKHOUSE_URL) - kwargs.setdefault("connect_timeout", CLICKHOUSE_CONNECT_TIMEOUT_SECONDS) - kwargs.setdefault("send_receive_timeout", CLICKHOUSE_QUERY_TIMEOUT_SECONDS) - kwargs.setdefault("client_name", settings.CLICKHOUSE_CONNECTION_CLIENT_NAME) - return Client(host, **kwargs) + clients: dict[int, Client] = getattr(_clickhouse_clients, "clients", None) or {} + _clickhouse_clients.clients = clients + if (client := clients.get(send_receive_timeout)) is None: + host, kwargs = parse_url(settings.EXPERIMENTATION_CLICKHOUSE_URL) + kwargs.setdefault("connect_timeout", CLICKHOUSE_CONNECT_TIMEOUT_SECONDS) + kwargs.setdefault("send_receive_timeout", send_receive_timeout) + kwargs.setdefault("client_name", settings.CLICKHOUSE_CONNECTION_CLIENT_NAME) + client = clients[send_receive_timeout] = Client(host, **kwargs) + return client _CLICKHOUSE_EVENT_NAMES_QUERY = ( @@ -344,7 +356,9 @@ def get_exposure_buckets( window_end: datetime, granularity: ExposureGranularity, ) -> list[ExposureBucket]: - rows = _get_clickhouse_client().execute( + rows = _get_clickhouse_client( + send_receive_timeout=CLICKHOUSE_BACKGROUND_QUERY_TIMEOUT_SECONDS, + ).execute( EXPOSURE_BUCKETS_QUERY.format( bucket_function=_EXPOSURE_BUCKET_FUNCTIONS[granularity] ), @@ -387,9 +401,9 @@ def get_metric_variant_stats( } builder.add_metric_params(params) - rows, columns = _get_clickhouse_client().execute( - builder.build_query(), params, with_column_types=True - ) + rows, columns = _get_clickhouse_client( + send_receive_timeout=CLICKHOUSE_BACKGROUND_QUERY_TIMEOUT_SECONDS, + ).execute(builder.build_query(), params, with_column_types=True) exposure_counts, metric_stats = builder.decode_rows( rows, [name for name, _type in columns] ) diff --git a/api/experimentation/tasks.py b/api/experimentation/tasks.py index 78abee5723b3..9f1aab839882 100644 --- a/api/experimentation/tasks.py +++ b/api/experimentation/tasks.py @@ -6,6 +6,7 @@ register_recurring_task, register_task_handler, ) +from task_processor.exceptions import TaskBackoffError from environments.models import Environment, EnvironmentAPIKey from experimentation import ingestion_sync_service @@ -28,6 +29,11 @@ deliver_warehouse_events, ) +# Warehouse computes may legitimately run for up to +# `CLICKHOUSE_BACKGROUND_QUERY_TIMEOUT_SECONDS`, e.g. while an idled +# ClickHouse Cloud service wakes up. +COMPUTE_TASK_TIMEOUT = timedelta(minutes=3) + logger = structlog.get_logger("experimentation") @@ -162,7 +168,7 @@ def clean_up_old_warehouse_delivery_logs() -> None: ).delete() -@register_task_handler() +@register_task_handler(timeout=COMPUTE_TASK_TIMEOUT) def compute_experiment_exposures(experiment_id: int) -> None: experiment = ( Experiment.objects.select_related("environment__project", "feature") @@ -194,12 +200,14 @@ def compute_experiment_exposures(experiment_id: int) -> None: environment__id=experiment.environment_id, organisation__id=experiment.environment.project.organisation_id, ) + if isinstance(exc, OSError): + raise TaskBackoffError() from exc return exposures.record_refresh(summary, as_of) -@register_task_handler() +@register_task_handler(timeout=COMPUTE_TASK_TIMEOUT) def compute_experiment_results(experiment_id: int) -> None: experiment = ( Experiment.objects.select_related("environment__project", "feature") @@ -229,6 +237,8 @@ def compute_experiment_results(experiment_id: int) -> None: environment__id=experiment.environment_id, organisation__id=experiment.environment.project.organisation_id, ) + if isinstance(exc, OSError): + raise TaskBackoffError() from exc return results.record_refresh(summary, as_of) diff --git a/api/tests/unit/experimentation/test_services.py b/api/tests/unit/experimentation/test_services.py index 19fbb481b840..397dcdef8986 100644 --- a/api/tests/unit/experimentation/test_services.py +++ b/api/tests/unit/experimentation/test_services.py @@ -1,3 +1,4 @@ +import threading from dataclasses import asdict from datetime import datetime, timezone from unittest.mock import MagicMock @@ -62,7 +63,7 @@ def test_get_clickhouse_client__configured_url__builds_client_with_timeouts( "clickhouse://user:pass@ch.example.com:9440/flagsmith_exp?secure=True" ) mock_client_cls = mocker.patch("experimentation.services.Client") - services._get_clickhouse_client.cache_clear() + services._clickhouse_clients.__dict__.clear() # When client = services._get_clickhouse_client() @@ -80,7 +81,7 @@ def test_get_clickhouse_client__configured_url__builds_client_with_timeouts( client_name=settings.CLICKHOUSE_CONNECTION_CLIENT_NAME, ) assert client is mock_client_cls.return_value - services._get_clickhouse_client.cache_clear() + services._clickhouse_clients.__dict__.clear() def test_get_clickhouse_client__dsn_timeouts__are_preserved( @@ -92,7 +93,7 @@ def test_get_clickhouse_client__dsn_timeouts__are_preserved( "clickhouse://ch.example.com:9000/db?connect_timeout=1&send_receive_timeout=2" ) mock_client_cls = mocker.patch("experimentation.services.Client") - services._get_clickhouse_client.cache_clear() + services._clickhouse_clients.__dict__.clear() # When services._get_clickhouse_client() @@ -106,7 +107,44 @@ def test_get_clickhouse_client__dsn_timeouts__are_preserved( send_receive_timeout=2, client_name=settings.CLICKHOUSE_CONNECTION_CLIENT_NAME, ) - services._get_clickhouse_client.cache_clear() + services._clickhouse_clients.__dict__.clear() + + +def test_get_clickhouse_client__per_thread_and_timeout__caches_distinct_clients( + mocker: MockerFixture, + settings: SettingsWrapper, +) -> None: + # Given + settings.EXPERIMENTATION_CLICKHOUSE_URL = "clickhouse://ch.example.com/db" + mock_client_cls = mocker.patch( + "experimentation.services.Client", + side_effect=lambda *args, **kwargs: mocker.Mock(), + ) + services._clickhouse_clients.__dict__.clear() + + # When + client = services._get_clickhouse_client() + same_client = services._get_clickhouse_client() + background_client = services._get_clickhouse_client( + send_receive_timeout=services.CLICKHOUSE_BACKGROUND_QUERY_TIMEOUT_SECONDS, + ) + other_thread_clients = [] + thread = threading.Thread( + target=lambda: other_thread_clients.append(services._get_clickhouse_client()) + ) + thread.start() + thread.join() + + # Then + assert client is same_client + assert background_client is not client + assert other_thread_clients[0] is not client + assert mock_client_cls.call_count == 3 + assert ( + mock_client_cls.call_args_list[1].kwargs["send_receive_timeout"] + == services.CLICKHOUSE_BACKGROUND_QUERY_TIMEOUT_SECONDS + ) + services._clickhouse_clients.__dict__.clear() @pytest.mark.parametrize( @@ -327,7 +365,7 @@ def test_get_exposure_buckets__day_granularity__queries_and_maps_rows( ] mock_client = mocker.Mock() mock_client.execute.return_value = rows - mocker.patch( + mock_get_client = mocker.patch( "experimentation.services._get_clickhouse_client", return_value=mock_client, ) @@ -378,6 +416,9 @@ def test_get_exposure_buckets__day_granularity__queries_and_maps_rows( "window_start": window_start, "window_end": window_end, } + mock_get_client.assert_called_once_with( + send_receive_timeout=services.CLICKHOUSE_BACKGROUND_QUERY_TIMEOUT_SECONDS, + ) def test_get_exposure_buckets__hour_granularity__buckets_by_hour( @@ -795,7 +836,7 @@ def test_get_metric_variant_stats__metrics__queries_and_maps_rows( ] mock_client = mocker.Mock() mock_client.execute.return_value = (rows, _result_columns(4)) - mocker.patch( + mock_get_client = mocker.patch( "experimentation.services._get_clickhouse_client", return_value=mock_client, ) @@ -862,6 +903,9 @@ def test_get_metric_variant_stats__metrics__queries_and_maps_rows( assert params["metric_2_event"] == "page_view" assert params["metric_3_event"] == "session" assert params["window_end"] == window_end + mock_get_client.assert_called_once_with( + send_receive_timeout=services.CLICKHOUSE_BACKGROUND_QUERY_TIMEOUT_SECONDS, + ) def test_get_metric_variant_stats__three_variants__maps_all_variants( diff --git a/api/tests/unit/experimentation/test_tasks.py b/api/tests/unit/experimentation/test_tasks.py index aa378cd44d0a..8155cca3ac2e 100644 --- a/api/tests/unit/experimentation/test_tasks.py +++ b/api/tests/unit/experimentation/test_tasks.py @@ -14,6 +14,7 @@ from prometheus_client import REGISTRY from pytest_mock import MockerFixture from pytest_structlog import StructuredLogCapture +from task_processor.exceptions import TaskBackoffError from environments.models import Environment, EnvironmentAPIKey from experimentation import warehouse_delivery_service @@ -39,6 +40,7 @@ WarehouseDeliveryOutcome, WarehouseType, ) +from experimentation.services import CLICKHOUSE_BACKGROUND_QUERY_TIMEOUT_SECONDS from experimentation.stats import VariantStats from experimentation.tasks import ( clean_up_old_warehouse_delivery_logs, @@ -355,6 +357,30 @@ def test_compute_experiment_exposures__warehouse_error__records_failure( ) +def test_compute_experiment_exposures__transient_warehouse_error__records_failure_and_backs_off( + experiment: Experiment, + mocker: MockerFixture, + log: StructuredLogCapture, +) -> None: + # Given + experiment.status = ExperimentStatus.RUNNING + experiment.started_at = datetime(2026, 6, 10, tzinfo=dt_timezone.utc) + experiment.save() + mocker.patch( + "experimentation.tasks.compute_exposures_summary", + side_effect=TimeoutError("The read operation timed out"), + ) + + # When + with pytest.raises(TaskBackoffError): + compute_experiment_exposures(experiment_id=experiment.id) + + # Then + exposures = ExperimentExposures.objects.get(experiment=experiment) + assert exposures.last_error_at is not None + assert log.has("exposures.compute_failed", level="error") + + def test_compute_experiment_exposures__not_started_experiment__skips( experiment: Experiment, mocker: MockerFixture, @@ -531,6 +557,55 @@ def test_compute_experiment_results__warehouse_error__records_failure( ] +@pytest.mark.parametrize( + "exc", + [ + TimeoutError("The read operation timed out"), + ConnectionResetError("Connection reset by peer"), + ], + ids=["timeout", "reset"], +) +def test_compute_experiment_results__transient_warehouse_error__records_failure_and_backs_off( + experiment: Experiment, + mocker: MockerFixture, + log: StructuredLogCapture, + exc: Exception, +) -> None: + # Given + experiment.status = ExperimentStatus.RUNNING + experiment.started_at = datetime(2026, 6, 10, tzinfo=dt_timezone.utc) + experiment.save() + mocker.patch( + "experimentation.tasks.compute_results_summary", + side_effect=exc, + ) + + # When + with pytest.raises(TaskBackoffError): + compute_experiment_results(experiment_id=experiment.id) + + # Then + results = ExperimentResults.objects.get(experiment=experiment) + assert results.last_error_at is not None + assert log.has("results.compute_failed", level="error") + + +@pytest.mark.parametrize( + "task_handler", + [compute_experiment_exposures, compute_experiment_results], + ids=["exposures", "results"], +) +def test_compute_experiment_task_handlers__task_timeout__exceeds_background_query_timeout( + task_handler: Any, +) -> None: + # Given + task_timeout = task_handler.timeout + + # When / Then + assert task_timeout is not None + assert task_timeout.total_seconds() > CLICKHOUSE_BACKGROUND_QUERY_TIMEOUT_SECONDS + + def test_compute_experiment_results__not_started_experiment__skips( experiment: Experiment, mocker: MockerFixture, diff --git a/api/tests/unit/experimentation/test_views.py b/api/tests/unit/experimentation/test_views.py index 543c3b26969e..58d07903755b 100644 --- a/api/tests/unit/experimentation/test_views.py +++ b/api/tests/unit/experimentation/test_views.py @@ -34,7 +34,7 @@ def mock_clickhouse_stats( events re-patch experimentation.services.get_warehouse_event_stats; tests for the unconfigured/erroring paths override the setting / raise.""" settings.EXPERIMENTATION_CLICKHOUSE_URL = "clickhouse://localhost:9000/test" - services._get_clickhouse_client.cache_clear() + services._clickhouse_clients.__dict__.clear() mock_client = mocker.Mock() mock_client.execute.return_value = [(0, 0)] return mocker.patch( diff --git a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md index 8bafb8eaf0c9..8b835107176f 100644 --- a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md @@ -150,7 +150,7 @@ Attributes: ### `experimentation.exposures.compute_failed` Logged at `error` from: - - `api/experimentation/tasks.py:189` + - `api/experimentation/tasks.py:195` Attributes: - `environment.id` @@ -218,7 +218,7 @@ Attributes: ### `experimentation.results.compute_failed` Logged at `error` from: - - `api/experimentation/tasks.py:225` + - `api/experimentation/tasks.py:233` Attributes: - `environment.id` @@ -690,7 +690,7 @@ Attributes: ### `warehouse.connection.connected` Logged at `info` from: - - `api/experimentation/services.py:1118` + - `api/experimentation/services.py:1132` Attributes: - `environment.id` @@ -699,8 +699,8 @@ Attributes: ### `warehouse.connection.event_names_failed` Logged at `warning` from: - - `api/experimentation/services.py:223` - - `api/experimentation/services.py:1218` + - `api/experimentation/services.py:235` + - `api/experimentation/services.py:1232` Attributes: - `environment.id` @@ -710,7 +710,7 @@ Attributes: ### `warehouse.connection.event_stats_failed` Logged at `warning` from: - - `api/experimentation/services.py:1181` + - `api/experimentation/services.py:1195` Attributes: - `environment.id` @@ -719,7 +719,7 @@ Attributes: ### `warehouse.connection.test_event_sent` Logged at `info` from: - - `api/experimentation/services.py:892` + - `api/experimentation/services.py:906` Attributes: - `environment.id` @@ -728,7 +728,7 @@ Attributes: ### `warehouse.connection.verification_failed` Logged at `warning` from: - - `api/experimentation/services.py:1093` + - `api/experimentation/services.py:1107` Attributes: - `environment.id` @@ -738,7 +738,7 @@ Attributes: ### `warehouse.connection.verification_succeeded` Logged at `info` from: - - `api/experimentation/services.py:1103` + - `api/experimentation/services.py:1117` Attributes: - `environment.id` @@ -747,7 +747,7 @@ Attributes: ### `warehouse.delivery.all_objects_rejected` Logged at `error` from: - - `api/experimentation/services.py:1048` + - `api/experimentation/services.py:1062` Attributes: - `connection.id` @@ -758,7 +758,7 @@ Attributes: ### `warehouse.delivery.budget_exhausted` Logged at `info` from: - - `api/experimentation/services.py:937` + - `api/experimentation/services.py:951` Attributes: - `connection.id` @@ -769,7 +769,7 @@ Attributes: ### `warehouse.delivery.completed` Logged at `info` from: - - `api/experimentation/services.py:1058` + - `api/experimentation/services.py:1072` Attributes: - `connection.id` @@ -782,7 +782,7 @@ Attributes: ### `warehouse.delivery.failed` Logged at `error` from: - - `api/experimentation/services.py:1031` + - `api/experimentation/services.py:1045` Attributes: - `connection.id` @@ -793,7 +793,7 @@ Attributes: ### `warehouse.delivery.object_rejected` Logged at `error` from: - - `api/experimentation/services.py:966` + - `api/experimentation/services.py:980` Attributes: - `connection.id` @@ -805,7 +805,7 @@ Attributes: ### `warehouse.srm.overallocated` Logged at `error` from: - - `api/experimentation/services.py:514` + - `api/experimentation/services.py:528` Attributes: - `environment.id` @@ -815,7 +815,7 @@ Attributes: ### `warehouse.srm.unkeyed_variant` Logged at `error` from: - - `api/experimentation/services.py:500` + - `api/experimentation/services.py:514` Attributes: - `environment.id` From ee6860f6f0bc38b1a9207f342cd1abb8a2e120e8 Mon Sep 17 00:00:00 2001 From: wadii Date: Wed, 12 Aug 2026 16:44:42 +0200 Subject: [PATCH 2/4] feat: remove slop --- api/experimentation/tasks.py | 3 --- 1 file changed, 3 deletions(-) diff --git a/api/experimentation/tasks.py b/api/experimentation/tasks.py index 9f1aab839882..2fcf4dec9ea8 100644 --- a/api/experimentation/tasks.py +++ b/api/experimentation/tasks.py @@ -29,9 +29,6 @@ deliver_warehouse_events, ) -# Warehouse computes may legitimately run for up to -# `CLICKHOUSE_BACKGROUND_QUERY_TIMEOUT_SECONDS`, e.g. while an idled -# ClickHouse Cloud service wakes up. COMPUTE_TASK_TIMEOUT = timedelta(minutes=3) logger = structlog.get_logger("experimentation") From 5e0bc974e2d7cdd0a31274c6d52b0b22b28e0e6c Mon Sep 17 00:00:00 2001 From: "flagsmith-engineering[bot]" Date: Wed, 12 Aug 2026 15:23:30 +0000 Subject: [PATCH 3/4] chore: Update documentation artefacts --- .../observability/_events-catalogue.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md index 8b835107176f..8ce840ec46cd 100644 --- a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md @@ -150,7 +150,7 @@ Attributes: ### `experimentation.exposures.compute_failed` Logged at `error` from: - - `api/experimentation/tasks.py:195` + - `api/experimentation/tasks.py:192` Attributes: - `environment.id` @@ -218,7 +218,7 @@ Attributes: ### `experimentation.results.compute_failed` Logged at `error` from: - - `api/experimentation/tasks.py:233` + - `api/experimentation/tasks.py:230` Attributes: - `environment.id` From 99485529a72040df61095b3ea8d11865234a1f64 Mon Sep 17 00:00:00 2001 From: wadii Date: Wed, 12 Aug 2026 18:25:28 +0200 Subject: [PATCH 4/4] fix: revert thread-local ClickHouse client caching, keep per-timeout cache --- api/experimentation/services.py | 25 ++++++----------- .../unit/experimentation/test_services.py | 24 ++++++---------- api/tests/unit/experimentation/test_views.py | 2 +- .../observability/_events-catalogue.md | 28 +++++++++---------- 4 files changed, 31 insertions(+), 48 deletions(-) diff --git a/api/experimentation/services.py b/api/experimentation/services.py index d91e4ad02142..dbc938f9c65c 100644 --- a/api/experimentation/services.py +++ b/api/experimentation/services.py @@ -2,10 +2,10 @@ import hashlib import json -import threading import time import typing from dataclasses import replace +from functools import lru_cache import structlog from clickhouse_connect.driver.exceptions import ClickHouseError @@ -149,9 +149,7 @@ def is_experiment_feature_enabled(organisation: Organisation) -> bool: ) -_clickhouse_clients = threading.local() - - +@lru_cache(maxsize=2) def _get_clickhouse_client( send_receive_timeout: int = CLICKHOUSE_QUERY_TIMEOUT_SECONDS, ) -> Client: @@ -159,20 +157,13 @@ def _get_clickhouse_client( The database is taken from the DSN path, so queries can reference the `events` table unqualified. Connect and query timeouts are bounded unless the - DSN overrides them. - - clickhouse-driver clients are not thread-safe, so one client is cached per - thread and requested timeout. + DSN overrides them. One client is cached per requested timeout. """ - clients: dict[int, Client] = getattr(_clickhouse_clients, "clients", None) or {} - _clickhouse_clients.clients = clients - if (client := clients.get(send_receive_timeout)) is None: - host, kwargs = parse_url(settings.EXPERIMENTATION_CLICKHOUSE_URL) - kwargs.setdefault("connect_timeout", CLICKHOUSE_CONNECT_TIMEOUT_SECONDS) - kwargs.setdefault("send_receive_timeout", send_receive_timeout) - kwargs.setdefault("client_name", settings.CLICKHOUSE_CONNECTION_CLIENT_NAME) - client = clients[send_receive_timeout] = Client(host, **kwargs) - return client + host, kwargs = parse_url(settings.EXPERIMENTATION_CLICKHOUSE_URL) + kwargs.setdefault("connect_timeout", CLICKHOUSE_CONNECT_TIMEOUT_SECONDS) + kwargs.setdefault("send_receive_timeout", send_receive_timeout) + kwargs.setdefault("client_name", settings.CLICKHOUSE_CONNECTION_CLIENT_NAME) + return Client(host, **kwargs) _CLICKHOUSE_EVENT_NAMES_QUERY = ( diff --git a/api/tests/unit/experimentation/test_services.py b/api/tests/unit/experimentation/test_services.py index 397dcdef8986..b363be8f794e 100644 --- a/api/tests/unit/experimentation/test_services.py +++ b/api/tests/unit/experimentation/test_services.py @@ -1,4 +1,3 @@ -import threading from dataclasses import asdict from datetime import datetime, timezone from unittest.mock import MagicMock @@ -63,7 +62,7 @@ def test_get_clickhouse_client__configured_url__builds_client_with_timeouts( "clickhouse://user:pass@ch.example.com:9440/flagsmith_exp?secure=True" ) mock_client_cls = mocker.patch("experimentation.services.Client") - services._clickhouse_clients.__dict__.clear() + services._get_clickhouse_client.cache_clear() # When client = services._get_clickhouse_client() @@ -81,7 +80,7 @@ def test_get_clickhouse_client__configured_url__builds_client_with_timeouts( client_name=settings.CLICKHOUSE_CONNECTION_CLIENT_NAME, ) assert client is mock_client_cls.return_value - services._clickhouse_clients.__dict__.clear() + services._get_clickhouse_client.cache_clear() def test_get_clickhouse_client__dsn_timeouts__are_preserved( @@ -93,7 +92,7 @@ def test_get_clickhouse_client__dsn_timeouts__are_preserved( "clickhouse://ch.example.com:9000/db?connect_timeout=1&send_receive_timeout=2" ) mock_client_cls = mocker.patch("experimentation.services.Client") - services._clickhouse_clients.__dict__.clear() + services._get_clickhouse_client.cache_clear() # When services._get_clickhouse_client() @@ -107,10 +106,10 @@ def test_get_clickhouse_client__dsn_timeouts__are_preserved( send_receive_timeout=2, client_name=settings.CLICKHOUSE_CONNECTION_CLIENT_NAME, ) - services._clickhouse_clients.__dict__.clear() + services._get_clickhouse_client.cache_clear() -def test_get_clickhouse_client__per_thread_and_timeout__caches_distinct_clients( +def test_get_clickhouse_client__per_timeout__caches_distinct_clients( mocker: MockerFixture, settings: SettingsWrapper, ) -> None: @@ -120,7 +119,7 @@ def test_get_clickhouse_client__per_thread_and_timeout__caches_distinct_clients( "experimentation.services.Client", side_effect=lambda *args, **kwargs: mocker.Mock(), ) - services._clickhouse_clients.__dict__.clear() + services._get_clickhouse_client.cache_clear() # When client = services._get_clickhouse_client() @@ -128,23 +127,16 @@ def test_get_clickhouse_client__per_thread_and_timeout__caches_distinct_clients( background_client = services._get_clickhouse_client( send_receive_timeout=services.CLICKHOUSE_BACKGROUND_QUERY_TIMEOUT_SECONDS, ) - other_thread_clients = [] - thread = threading.Thread( - target=lambda: other_thread_clients.append(services._get_clickhouse_client()) - ) - thread.start() - thread.join() # Then assert client is same_client assert background_client is not client - assert other_thread_clients[0] is not client - assert mock_client_cls.call_count == 3 + assert mock_client_cls.call_count == 2 assert ( mock_client_cls.call_args_list[1].kwargs["send_receive_timeout"] == services.CLICKHOUSE_BACKGROUND_QUERY_TIMEOUT_SECONDS ) - services._clickhouse_clients.__dict__.clear() + services._get_clickhouse_client.cache_clear() @pytest.mark.parametrize( diff --git a/api/tests/unit/experimentation/test_views.py b/api/tests/unit/experimentation/test_views.py index 58d07903755b..543c3b26969e 100644 --- a/api/tests/unit/experimentation/test_views.py +++ b/api/tests/unit/experimentation/test_views.py @@ -34,7 +34,7 @@ def mock_clickhouse_stats( events re-patch experimentation.services.get_warehouse_event_stats; tests for the unconfigured/erroring paths override the setting / raise.""" settings.EXPERIMENTATION_CLICKHOUSE_URL = "clickhouse://localhost:9000/test" - services._clickhouse_clients.__dict__.clear() + services._get_clickhouse_client.cache_clear() mock_client = mocker.Mock() mock_client.execute.return_value = [(0, 0)] return mocker.patch( diff --git a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md index 8ce840ec46cd..004a6b977fb6 100644 --- a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md @@ -690,7 +690,7 @@ Attributes: ### `warehouse.connection.connected` Logged at `info` from: - - `api/experimentation/services.py:1132` + - `api/experimentation/services.py:1123` Attributes: - `environment.id` @@ -699,8 +699,8 @@ Attributes: ### `warehouse.connection.event_names_failed` Logged at `warning` from: - - `api/experimentation/services.py:235` - - `api/experimentation/services.py:1232` + - `api/experimentation/services.py:226` + - `api/experimentation/services.py:1223` Attributes: - `environment.id` @@ -710,7 +710,7 @@ Attributes: ### `warehouse.connection.event_stats_failed` Logged at `warning` from: - - `api/experimentation/services.py:1195` + - `api/experimentation/services.py:1186` Attributes: - `environment.id` @@ -719,7 +719,7 @@ Attributes: ### `warehouse.connection.test_event_sent` Logged at `info` from: - - `api/experimentation/services.py:906` + - `api/experimentation/services.py:897` Attributes: - `environment.id` @@ -728,7 +728,7 @@ Attributes: ### `warehouse.connection.verification_failed` Logged at `warning` from: - - `api/experimentation/services.py:1107` + - `api/experimentation/services.py:1098` Attributes: - `environment.id` @@ -738,7 +738,7 @@ Attributes: ### `warehouse.connection.verification_succeeded` Logged at `info` from: - - `api/experimentation/services.py:1117` + - `api/experimentation/services.py:1108` Attributes: - `environment.id` @@ -747,7 +747,7 @@ Attributes: ### `warehouse.delivery.all_objects_rejected` Logged at `error` from: - - `api/experimentation/services.py:1062` + - `api/experimentation/services.py:1053` Attributes: - `connection.id` @@ -758,7 +758,7 @@ Attributes: ### `warehouse.delivery.budget_exhausted` Logged at `info` from: - - `api/experimentation/services.py:951` + - `api/experimentation/services.py:942` Attributes: - `connection.id` @@ -769,7 +769,7 @@ Attributes: ### `warehouse.delivery.completed` Logged at `info` from: - - `api/experimentation/services.py:1072` + - `api/experimentation/services.py:1063` Attributes: - `connection.id` @@ -782,7 +782,7 @@ Attributes: ### `warehouse.delivery.failed` Logged at `error` from: - - `api/experimentation/services.py:1045` + - `api/experimentation/services.py:1036` Attributes: - `connection.id` @@ -793,7 +793,7 @@ Attributes: ### `warehouse.delivery.object_rejected` Logged at `error` from: - - `api/experimentation/services.py:980` + - `api/experimentation/services.py:971` Attributes: - `connection.id` @@ -805,7 +805,7 @@ Attributes: ### `warehouse.srm.overallocated` Logged at `error` from: - - `api/experimentation/services.py:528` + - `api/experimentation/services.py:519` Attributes: - `environment.id` @@ -815,7 +815,7 @@ Attributes: ### `warehouse.srm.unkeyed_variant` Logged at `error` from: - - `api/experimentation/services.py:514` + - `api/experimentation/services.py:505` Attributes: - `environment.id`