From db97b6695e9a48eac51dc82880dce2bfd2ab3403 Mon Sep 17 00:00:00 2001 From: wadii Date: Thu, 6 Aug 2026 10:43:24 +0200 Subject: [PATCH 1/9] feat: expose distinct warehouse event names per connection --- api/experimentation/dataclasses.py | 6 + api/experimentation/services.py | 93 ++++++++++-- api/experimentation/views.py | 38 ++++- .../unit/experimentation/test_services.py | 139 +++++++++++++++--- api/tests/unit/experimentation/test_views.py | 78 +++++++++- .../observability/_events-catalogue.md | 33 +++-- 6 files changed, 340 insertions(+), 47 deletions(-) diff --git a/api/experimentation/dataclasses.py b/api/experimentation/dataclasses.py index 1766ea5aeb44..0c780a1529f8 100644 --- a/api/experimentation/dataclasses.py +++ b/api/experimentation/dataclasses.py @@ -24,6 +24,12 @@ class WarehouseEventStats: unique_events_count: int +@dataclass(frozen=True) +class WarehouseEventNames: + events: list[str] + is_truncated: bool + + @dataclass(frozen=True) class ExposureBucket: variant: str diff --git a/api/experimentation/services.py b/api/experimentation/services.py index e94da042c0f9..510f905ff199 100644 --- a/api/experimentation/services.py +++ b/api/experimentation/services.py @@ -42,6 +42,7 @@ ResultsAggregates, ResultsSummary, RolloutSpec, + WarehouseEventNames, WarehouseEventStats, ) from experimentation.metrics import ( @@ -103,8 +104,11 @@ CLICKHOUSE_QUERY_TIMEOUT_SECONDS = 30 CLICKHOUSE_VERIFY_TIMEOUT_SECONDS = 5 CUSTOMER_EVENT_STATS_CACHE_SECONDS = 60 +CUSTOMER_EVENT_NAMES_CACHE_SECONDS = 300 +WAREHOUSE_EVENT_NAMES_LIMIT = 500 _CUSTOMER_EVENT_STATS_UNAVAILABLE = "unavailable" +_CUSTOMER_EVENT_NAMES_UNAVAILABLE = "unavailable" # A delivery run stops taking on new objects after this long, leaving room for # the slowest possible in-flight insert to still land inside the task timeout. @@ -142,16 +146,49 @@ def _get_clickhouse_client() -> Client: return Client(host, **kwargs) -def get_unique_event_names(environment_key: str) -> list[str]: - """Return the distinct event names recorded for `environment_key`, - ordered alphabetically.""" - rows = _get_clickhouse_client().execute( - "SELECT DISTINCT event FROM events " - "WHERE environment_key = %(environment_key)s " - "ORDER BY event", - {"environment_key": environment_key}, +_EVENT_NAMES_QUERY = ( + "SELECT DISTINCT event FROM events " + "WHERE environment_key = %(environment_key)s " + "ORDER BY event LIMIT %(limit)s" +) + + +def _event_names_query_params(environment_key: str) -> dict[str, str | int]: + # Fetch one row past the limit so truncation is detectable. + return { + "environment_key": environment_key, + "limit": WAREHOUSE_EVENT_NAMES_LIMIT + 1, + } + + +def _build_event_names( + rows: "Sequence[Sequence[typing.Any]]", +) -> WarehouseEventNames: + names = [row[0] for row in rows] + return WarehouseEventNames( + events=names[:WAREHOUSE_EVENT_NAMES_LIMIT], + is_truncated=len(names) > WAREHOUSE_EVENT_NAMES_LIMIT, ) - return [row[0] for row in rows] + + +def get_warehouse_event_names( + connection: "WarehouseConnection", + environment_key: str, +) -> WarehouseEventNames | None: + """Return the distinct event names recorded for `environment_key`, capped + at WAREHOUSE_EVENT_NAMES_LIMIT; None when the warehouse is unavailable.""" + if connection.warehouse_type == WarehouseType.CLICKHOUSE: + return _get_customer_warehouse_event_names_cached(connection, environment_key) + if not settings.EXPERIMENTATION_CLICKHOUSE_URL: + return None + try: + rows = _get_clickhouse_client().execute( + _EVENT_NAMES_QUERY, + _event_names_query_params(environment_key), + ) + except Exception: + return None + return _build_event_names(rows) _EVENT_STATS_QUERY = ( @@ -1109,3 +1146,41 @@ def _get_customer_warehouse_event_stats_cached( return None cache.set(cache_key, stats, CUSTOMER_EVENT_STATS_CACHE_SECONDS) return stats + + +def _get_customer_warehouse_event_names_cached( + connection: "WarehouseConnection", + environment_key: str, +) -> WarehouseEventNames | None: + """Query the customer's ClickHouse instance, caching results — including + failures — to spare their host repeated connections.""" + cache_key = f"experimentation:customer_event_names:{connection.id}" + cached = cache.get(cache_key) + if isinstance(cached, WarehouseEventNames): + return cached + if cached == _CUSTOMER_EVENT_NAMES_UNAVAILABLE: + return None + try: + with warehouse_delivery_service.delivery_client( + connection, + send_receive_timeout=CLICKHOUSE_VERIFY_TIMEOUT_SECONDS, + ) as client: + rows = client.query( + _EVENT_NAMES_QUERY, + parameters=_event_names_query_params(environment_key), + ).result_rows + except Exception: + cache.set( + cache_key, + _CUSTOMER_EVENT_NAMES_UNAVAILABLE, + CUSTOMER_EVENT_NAMES_CACHE_SECONDS, + ) + logger.warning( + "connection.event_names_failed", + environment__id=connection.environment_id, + exc_info=True, + ) + return None + event_names = _build_event_names(rows) + cache.set(cache_key, event_names, CUSTOMER_EVENT_NAMES_CACHE_SECONDS) + return event_names diff --git a/api/experimentation/views.py b/api/experimentation/views.py index 785aa8ae367e..7a91b362c177 100644 --- a/api/experimentation/views.py +++ b/api/experimentation/views.py @@ -1,4 +1,5 @@ import logging +from dataclasses import asdict from datetime import timedelta from typing import Any @@ -65,6 +66,7 @@ create_metric_audit_log, create_warehouse_audit_log, enable_experiment_rollout, + get_warehouse_event_names, mark_warehouse_pending_connection, refresh_warehouse_connection_status, transition_experiment_status, @@ -106,7 +108,7 @@ def get_throttles(self) -> list[BaseThrottle]: ): self.throttle_scope = "warehouse_connection_write" return [*super().get_throttles(), ScopedRateThrottle()] - if self.action in ("list", "retrieve"): + if self.action in ("list", "retrieve", "events"): self.throttle_scope = "warehouse_connection_read" return [*super().get_throttles(), ScopedRateThrottle()] return super().get_throttles() @@ -222,6 +224,40 @@ def test_warehouse_connection_config( {"status": connection.status, "status_detail": connection.status_detail} ) + @extend_schema( + operation_id="api_v1_environments_warehouse_connections_events_list", + responses={ + 200: inline_serializer( + name="WarehouseEventNamesResult", + fields={ + "events": serializers.ListField(child=serializers.CharField()), + "is_truncated": serializers.BooleanField(), + }, + ) + }, + ) + @action(detail=True, methods=["get"], url_path="events") + def events(self, request: Request, **kwargs: object) -> Response: + """List the distinct event names in the connection's warehouse.""" + connection: WarehouseConnection = self.get_object() + if connection.warehouse_type not in ( + WarehouseType.FLAGSMITH, + WarehouseType.CLICKHOUSE, + ): + return Response( + {"detail": "Event listing is not supported for this warehouse type."}, + status=status.HTTP_400_BAD_REQUEST, + ) + event_names = get_warehouse_event_names( + connection, self.kwargs["environment_api_key"] + ) + if event_names is None: + return Response( + {"detail": "The warehouse is currently unreachable."}, + status=status.HTTP_503_SERVICE_UNAVAILABLE, + ) + return Response(asdict(event_names)) + def create(self, request: Request, *args: object, **kwargs: object) -> Response: environment = self._get_environment() serializer = self.get_serializer(data=request.data) diff --git a/api/tests/unit/experimentation/test_services.py b/api/tests/unit/experimentation/test_services.py index 3601638a7ce0..3441d0d4aabe 100644 --- a/api/tests/unit/experimentation/test_services.py +++ b/api/tests/unit/experimentation/test_services.py @@ -22,6 +22,7 @@ MetricSpec, ResultsAggregates, RolloutSpec, + WarehouseEventNames, WarehouseEventStats, ) from experimentation.models import ( @@ -108,28 +109,136 @@ def test_get_clickhouse_client__dsn_timeouts__are_preserved( services._get_clickhouse_client.cache_clear() -def test_get_unique_event_names__events_present__returns_ordered_names( +@pytest.mark.parametrize( + "rows, expected", + [ + ( + [("conversion",), ("page_view",)], + WarehouseEventNames(events=["conversion", "page_view"], is_truncated=False), + ), + ([], WarehouseEventNames(events=[], is_truncated=False)), + ( + [(f"event_{i:03d}",) for i in range(501)], + WarehouseEventNames( + events=[f"event_{i:03d}" for i in range(500)], is_truncated=True + ), + ), + ], + ids=["few", "none", "truncated"], +) +def test_get_warehouse_event_names__flagsmith_connection__returns_capped_names( + warehouse_connection: WarehouseConnection, + settings: SettingsWrapper, + rows: list[tuple[str]], + expected: WarehouseEventNames, mocker: MockerFixture, ) -> None: # Given + settings.EXPERIMENTATION_CLICKHOUSE_URL = "clickhouse://ch.example.com/db" mock_client = mocker.Mock() - mock_client.execute.return_value = [("conversion",), ("page_view",)] + mock_client.execute.return_value = rows mocker.patch( "experimentation.services._get_clickhouse_client", return_value=mock_client, ) # When - result = services.get_unique_event_names("env-key-123") + result = services.get_warehouse_event_names(warehouse_connection, "env-key-123") # Then - assert result == ["conversion", "page_view"] + assert result == expected mock_client.execute.assert_called_once_with( "SELECT DISTINCT event FROM events " "WHERE environment_key = %(environment_key)s " - "ORDER BY event", - {"environment_key": "env-key-123"}, + "ORDER BY event LIMIT %(limit)s", + {"environment_key": "env-key-123", "limit": 501}, + ) + + +@pytest.mark.parametrize( + "clickhouse_url, execute_side_effect", + [ + ("", None), + ("clickhouse://ch.example.com/db", Exception("connection refused")), + ], + ids=["unconfigured", "unreachable"], +) +def test_get_warehouse_event_names__flagsmith_warehouse_unavailable__returns_none( + warehouse_connection: WarehouseConnection, + settings: SettingsWrapper, + clickhouse_url: str, + execute_side_effect: Exception | None, + mocker: MockerFixture, +) -> None: + # Given + settings.EXPERIMENTATION_CLICKHOUSE_URL = clickhouse_url + mock_client = mocker.Mock() + mock_client.execute.side_effect = execute_side_effect + mocker.patch( + "experimentation.services._get_clickhouse_client", + return_value=mock_client, + ) + + # When + result = services.get_warehouse_event_names(warehouse_connection, "env-key-123") + + # Then + assert result is None + + +@pytest.mark.parametrize( + "query_result, expected", + [ + ( + [("conversion",), ("page_view",)], + WarehouseEventNames(events=["conversion", "page_view"], is_truncated=False), + ), + (Exception("connection refused"), None), + ], + ids=["reachable", "unreachable"], +) +def test_get_warehouse_event_names__clickhouse_connection__queries_customer_instance( + clickhouse_connection: WarehouseConnection, + reset_cache: None, + query_result: Exception | list[tuple[str]], + expected: WarehouseEventNames | None, + log: StructuredLogCapture, + mocker: MockerFixture, +) -> None: + # Given + get_client = mocker.patch( + "experimentation.warehouse_delivery_service.clickhouse_connect.get_client", + ) + if isinstance(query_result, Exception): + get_client.return_value.query.side_effect = query_result + else: + get_client.return_value.query.return_value = mocker.Mock( + result_rows=query_result + ) + + # When + result = services.get_warehouse_event_names(clickhouse_connection, "test-env-key") + + # Then + assert result == expected + get_client.return_value.query.assert_called_once_with( + "SELECT DISTINCT event FROM events " + "WHERE environment_key = %(environment_key)s " + "ORDER BY event LIMIT %(limit)s", + parameters={"environment_key": "test-env-key", "limit": 501}, ) + get_client.return_value.close.assert_called_once_with() + assert any( + event["event"] == "connection.event_names_failed" for event in log.events + ) == (expected is None) + + # When — the outcome is cached, so a second request doesn't reconnect + fresh_connection = WarehouseConnection.objects.get(id=clickhouse_connection.id) + second_result = services.get_warehouse_event_names(fresh_connection, "test-env-key") + + # Then + get_client.assert_called_once() + assert second_result == expected def test_get_exposure_buckets__day_granularity__queries_and_maps_rows( @@ -395,24 +504,6 @@ def test_build_exposures_summary__no_buckets__empty_summary() -> None: ) -def test_get_unique_event_names__no_events__returns_empty_list( - mocker: MockerFixture, -) -> None: - # Given - mock_client = mocker.Mock() - mock_client.execute.return_value = [] - mocker.patch( - "experimentation.services._get_clickhouse_client", - return_value=mock_client, - ) - - # When - result = services.get_unique_event_names("env-key-123") - - # Then - assert result == [] - - @pytest.mark.parametrize( "rows, expected_total, expected_unique", [ diff --git a/api/tests/unit/experimentation/test_views.py b/api/tests/unit/experimentation/test_views.py index fe01bad5b813..543c3b26969e 100644 --- a/api/tests/unit/experimentation/test_views.py +++ b/api/tests/unit/experimentation/test_views.py @@ -13,7 +13,7 @@ from audit.related_object_type import RelatedObjectType from environments.models import Environment from experimentation import services -from experimentation.dataclasses import WarehouseEventStats +from experimentation.dataclasses import WarehouseEventNames, WarehouseEventStats from experimentation.models import ( WarehouseConnection, WarehouseConnectionStatus, @@ -1846,3 +1846,79 @@ def test_patch__clickhouse_null_credentials__returns_400( assert "password" in response.json()["credentials"] clickhouse_connection.refresh_from_db() assert clickhouse_connection.credentials == {"password": "hunter2"} + + +@pytest.mark.parametrize( + "service_result, expected_status, expected_body", + [ + ( + WarehouseEventNames(events=["conversion"], is_truncated=False), + status.HTTP_200_OK, + {"events": ["conversion"], "is_truncated": False}, + ), + ( + None, + status.HTTP_503_SERVICE_UNAVAILABLE, + {"detail": "The warehouse is currently unreachable."}, + ), + ], + ids=["reachable", "unreachable"], +) +def test_get_events__warehouse_availability__maps_to_response( + admin_client: APIClient, + environment: Environment, + warehouse_connection: WarehouseConnection, + enable_features: EnableFeaturesFixture, + service_result: WarehouseEventNames | None, + expected_status: int, + expected_body: dict[str, object], + mocker: MockerFixture, +) -> None: + # Given + enable_features("experimentation_warehouse_connection") + mocker.patch( + "experimentation.views.get_warehouse_event_names", + return_value=service_result, + ) + url = reverse( + "api-v1:environments:experimentation:warehouse-connections-events", + args=[environment.api_key, warehouse_connection.id], + ) + + # When + response = admin_client.get(url) + + # Then + assert response.status_code == expected_status + assert response.json() == expected_body + + +def test_get_events__unsupported_type__returns_400( + admin_client: APIClient, + environment: Environment, + enable_features: EnableFeaturesFixture, + mocker: MockerFixture, +) -> None: + # Given + enable_features("experimentation_warehouse_connection") + get_event_names = mocker.patch("experimentation.views.get_warehouse_event_names") + connection = WarehouseConnection.objects.create( + environment=environment, + warehouse_type=WarehouseType.SNOWFLAKE, + name="Snowflake", + config={"account_identifier": "acme"}, + ) + url = reverse( + "api-v1:environments:experimentation:warehouse-connections-events", + args=[environment.api_key, connection.id], + ) + + # When + response = admin_client.get(url) + + # Then + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert response.json() == { + "detail": "Event listing is not supported for this warehouse type." + } + get_event_names.assert_not_called() diff --git a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md index 0c8a6b9066b4..e08e594eb16e 100644 --- a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md @@ -660,16 +660,25 @@ Attributes: ### `warehouse.connection.connected` Logged at `info` from: - - `api/experimentation/services.py:1041` + - `api/experimentation/services.py:1078` Attributes: - `environment.id` - `organisation.id` +### `warehouse.connection.event_names_failed` + +Logged at `warning` from: + - `api/experimentation/services.py:1178` + +Attributes: + - `environment.id` + - `exc_info` + ### `warehouse.connection.event_stats_failed` Logged at `warning` from: - - `api/experimentation/services.py:1104` + - `api/experimentation/services.py:1141` Attributes: - `environment.id` @@ -678,7 +687,7 @@ Attributes: ### `warehouse.connection.test_event_sent` Logged at `info` from: - - `api/experimentation/services.py:815` + - `api/experimentation/services.py:852` Attributes: - `environment.id` @@ -687,7 +696,7 @@ Attributes: ### `warehouse.connection.verification_failed` Logged at `warning` from: - - `api/experimentation/services.py:1016` + - `api/experimentation/services.py:1053` Attributes: - `environment.id` @@ -697,7 +706,7 @@ Attributes: ### `warehouse.connection.verification_succeeded` Logged at `info` from: - - `api/experimentation/services.py:1026` + - `api/experimentation/services.py:1063` Attributes: - `environment.id` @@ -706,7 +715,7 @@ Attributes: ### `warehouse.delivery.all_objects_rejected` Logged at `error` from: - - `api/experimentation/services.py:971` + - `api/experimentation/services.py:1008` Attributes: - `connection.id` @@ -717,7 +726,7 @@ Attributes: ### `warehouse.delivery.budget_exhausted` Logged at `info` from: - - `api/experimentation/services.py:860` + - `api/experimentation/services.py:897` Attributes: - `connection.id` @@ -728,7 +737,7 @@ Attributes: ### `warehouse.delivery.completed` Logged at `info` from: - - `api/experimentation/services.py:981` + - `api/experimentation/services.py:1018` Attributes: - `connection.id` @@ -741,7 +750,7 @@ Attributes: ### `warehouse.delivery.failed` Logged at `error` from: - - `api/experimentation/services.py:954` + - `api/experimentation/services.py:991` Attributes: - `connection.id` @@ -752,7 +761,7 @@ Attributes: ### `warehouse.delivery.object_rejected` Logged at `error` from: - - `api/experimentation/services.py:889` + - `api/experimentation/services.py:926` Attributes: - `connection.id` @@ -764,7 +773,7 @@ Attributes: ### `warehouse.srm.overallocated` Logged at `error` from: - - `api/experimentation/services.py:437` + - `api/experimentation/services.py:474` Attributes: - `environment.id` @@ -774,7 +783,7 @@ Attributes: ### `warehouse.srm.unkeyed_variant` Logged at `error` from: - - `api/experimentation/services.py:423` + - `api/experimentation/services.py:460` Attributes: - `environment.id` From f440aced7c0464be99c1e93c9905521f785fc32e Mon Sep 17 00:00:00 2001 From: wadii Date: Thu, 6 Aug 2026 10:53:59 +0200 Subject: [PATCH 2/9] feat: order warehouse event names by most recently seen --- api/experimentation/services.py | 9 +++++---- api/tests/unit/experimentation/test_services.py | 8 ++++---- 2 files changed, 9 insertions(+), 8 deletions(-) diff --git a/api/experimentation/services.py b/api/experimentation/services.py index 510f905ff199..624d690ea2f6 100644 --- a/api/experimentation/services.py +++ b/api/experimentation/services.py @@ -147,9 +147,9 @@ def _get_clickhouse_client() -> Client: _EVENT_NAMES_QUERY = ( - "SELECT DISTINCT event FROM events " + "SELECT event FROM events " "WHERE environment_key = %(environment_key)s " - "ORDER BY event LIMIT %(limit)s" + "GROUP BY event ORDER BY max(timestamp) DESC LIMIT %(limit)s" ) @@ -175,8 +175,9 @@ def get_warehouse_event_names( connection: "WarehouseConnection", environment_key: str, ) -> WarehouseEventNames | None: - """Return the distinct event names recorded for `environment_key`, capped - at WAREHOUSE_EVENT_NAMES_LIMIT; None when the warehouse is unavailable.""" + """Return the distinct event names recorded for `environment_key`, most + recently seen first, capped at WAREHOUSE_EVENT_NAMES_LIMIT; None when the + warehouse is unavailable.""" if connection.warehouse_type == WarehouseType.CLICKHOUSE: return _get_customer_warehouse_event_names_cached(connection, environment_key) if not settings.EXPERIMENTATION_CLICKHOUSE_URL: diff --git a/api/tests/unit/experimentation/test_services.py b/api/tests/unit/experimentation/test_services.py index 3441d0d4aabe..9d070c7143b7 100644 --- a/api/tests/unit/experimentation/test_services.py +++ b/api/tests/unit/experimentation/test_services.py @@ -148,9 +148,9 @@ def test_get_warehouse_event_names__flagsmith_connection__returns_capped_names( # Then assert result == expected mock_client.execute.assert_called_once_with( - "SELECT DISTINCT event FROM events " + "SELECT event FROM events " "WHERE environment_key = %(environment_key)s " - "ORDER BY event LIMIT %(limit)s", + "GROUP BY event ORDER BY max(timestamp) DESC LIMIT %(limit)s", {"environment_key": "env-key-123", "limit": 501}, ) @@ -222,9 +222,9 @@ def test_get_warehouse_event_names__clickhouse_connection__queries_customer_inst # Then assert result == expected get_client.return_value.query.assert_called_once_with( - "SELECT DISTINCT event FROM events " + "SELECT event FROM events " "WHERE environment_key = %(environment_key)s " - "ORDER BY event LIMIT %(limit)s", + "GROUP BY event ORDER BY max(timestamp) DESC LIMIT %(limit)s", parameters={"environment_key": "test-env-key", "limit": 501}, ) get_client.return_value.close.assert_called_once_with() From 76d3aa26e46b9cfc0cd3dcf875193acef119cf5b Mon Sep 17 00:00:00 2001 From: "flagsmith-engineering[bot]" Date: Thu, 6 Aug 2026 08:58:45 +0000 Subject: [PATCH 3/9] chore: Update documentation artefacts --- .../observability/_events-catalogue.md | 26 ++++++------ openapi.yaml | 40 +++++++++++++++++++ 2 files changed, 53 insertions(+), 13 deletions(-) diff --git a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md index e08e594eb16e..f62d4c1f6fba 100644 --- a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md @@ -660,7 +660,7 @@ Attributes: ### `warehouse.connection.connected` Logged at `info` from: - - `api/experimentation/services.py:1078` + - `api/experimentation/services.py:1079` Attributes: - `environment.id` @@ -669,7 +669,7 @@ Attributes: ### `warehouse.connection.event_names_failed` Logged at `warning` from: - - `api/experimentation/services.py:1178` + - `api/experimentation/services.py:1179` Attributes: - `environment.id` @@ -678,7 +678,7 @@ Attributes: ### `warehouse.connection.event_stats_failed` Logged at `warning` from: - - `api/experimentation/services.py:1141` + - `api/experimentation/services.py:1142` Attributes: - `environment.id` @@ -687,7 +687,7 @@ Attributes: ### `warehouse.connection.test_event_sent` Logged at `info` from: - - `api/experimentation/services.py:852` + - `api/experimentation/services.py:853` Attributes: - `environment.id` @@ -696,7 +696,7 @@ Attributes: ### `warehouse.connection.verification_failed` Logged at `warning` from: - - `api/experimentation/services.py:1053` + - `api/experimentation/services.py:1054` Attributes: - `environment.id` @@ -706,7 +706,7 @@ Attributes: ### `warehouse.connection.verification_succeeded` Logged at `info` from: - - `api/experimentation/services.py:1063` + - `api/experimentation/services.py:1064` Attributes: - `environment.id` @@ -715,7 +715,7 @@ Attributes: ### `warehouse.delivery.all_objects_rejected` Logged at `error` from: - - `api/experimentation/services.py:1008` + - `api/experimentation/services.py:1009` Attributes: - `connection.id` @@ -726,7 +726,7 @@ Attributes: ### `warehouse.delivery.budget_exhausted` Logged at `info` from: - - `api/experimentation/services.py:897` + - `api/experimentation/services.py:898` Attributes: - `connection.id` @@ -737,7 +737,7 @@ Attributes: ### `warehouse.delivery.completed` Logged at `info` from: - - `api/experimentation/services.py:1018` + - `api/experimentation/services.py:1019` Attributes: - `connection.id` @@ -750,7 +750,7 @@ Attributes: ### `warehouse.delivery.failed` Logged at `error` from: - - `api/experimentation/services.py:991` + - `api/experimentation/services.py:992` Attributes: - `connection.id` @@ -761,7 +761,7 @@ Attributes: ### `warehouse.delivery.object_rejected` Logged at `error` from: - - `api/experimentation/services.py:926` + - `api/experimentation/services.py:927` Attributes: - `connection.id` @@ -773,7 +773,7 @@ Attributes: ### `warehouse.srm.overallocated` Logged at `error` from: - - `api/experimentation/services.py:474` + - `api/experimentation/services.py:475` Attributes: - `environment.id` @@ -783,7 +783,7 @@ Attributes: ### `warehouse.srm.unkeyed_variant` Logged at `error` from: - - `api/experimentation/services.py:460` + - `api/experimentation/services.py:461` Attributes: - `environment.id` diff --git a/openapi.yaml b/openapi.yaml index 4086dbf7c689..86dbc206bb4a 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -7069,6 +7069,34 @@ paths: - Master API Key: [] tags: - Environments + '/api/v1/environments/{environment_api_key}/warehouse-connections/{connection_id}/events/': + get: + operationId: api_v1_environments_warehouse_connections_events_list + description: List the distinct event names in the connection's warehouse. + parameters: + - name: connection_id + in: path + description: A unique integer value identifying this warehouse connection. + required: true + schema: + type: integer + - name: environment_api_key + in: path + required: true + schema: + type: string + responses: + '200': + description: '' + content: + application/json: + schema: + $ref: '#/components/schemas/WarehouseEventNamesResult' + security: + - tokenAuth: [] + - Master API Key: [] + tags: + - Environments '/api/v1/environments/{environment_api_key}/warehouse-connections/{connection_id}/test-warehouse-connection/': post: operationId: api_v1_environments_warehouse_connections_test_warehouse_connection_create @@ -27950,6 +27978,18 @@ components: required: - status - status_detail + WarehouseEventNamesResult: + type: object + properties: + events: + type: array + items: + type: string + is_truncated: + type: boolean + required: + - events + - is_truncated WarehouseTypeEnum: description: |- * `flagsmith` - Flagsmith From 3fe89eab3da182dc20a17a0d898855bbeb24f9b5 Mon Sep 17 00:00:00 2001 From: wadii Date: Thu, 6 Aug 2026 11:07:55 +0200 Subject: [PATCH 4/9] refactor: tidy event names internals and shorten failure cache --- api/experimentation/services.py | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/api/experimentation/services.py b/api/experimentation/services.py index 624d690ea2f6..36d6d7ea1da4 100644 --- a/api/experimentation/services.py +++ b/api/experimentation/services.py @@ -105,10 +105,10 @@ CLICKHOUSE_VERIFY_TIMEOUT_SECONDS = 5 CUSTOMER_EVENT_STATS_CACHE_SECONDS = 60 CUSTOMER_EVENT_NAMES_CACHE_SECONDS = 300 +CUSTOMER_EVENT_NAMES_FAILURE_CACHE_SECONDS = 60 WAREHOUSE_EVENT_NAMES_LIMIT = 500 -_CUSTOMER_EVENT_STATS_UNAVAILABLE = "unavailable" -_CUSTOMER_EVENT_NAMES_UNAVAILABLE = "unavailable" +_CUSTOMER_EVENT_UNAVAILABLE = "unavailable" # A delivery run stops taking on new objects after this long, leaving room for # the slowest possible in-flight insert to still land inside the task timeout. @@ -146,7 +146,7 @@ def _get_clickhouse_client() -> Client: return Client(host, **kwargs) -_EVENT_NAMES_QUERY = ( +_CLICKHOUSE_EVENT_NAMES_QUERY = ( "SELECT event FROM events " "WHERE environment_key = %(environment_key)s " "GROUP BY event ORDER BY max(timestamp) DESC LIMIT %(limit)s" @@ -164,7 +164,7 @@ def _event_names_query_params(environment_key: str) -> dict[str, str | int]: def _build_event_names( rows: "Sequence[Sequence[typing.Any]]", ) -> WarehouseEventNames: - names = [row[0] for row in rows] + names = [event for (event,) in rows] return WarehouseEventNames( events=names[:WAREHOUSE_EVENT_NAMES_LIMIT], is_truncated=len(names) > WAREHOUSE_EVENT_NAMES_LIMIT, @@ -184,7 +184,7 @@ def get_warehouse_event_names( return None try: rows = _get_clickhouse_client().execute( - _EVENT_NAMES_QUERY, + _CLICKHOUSE_EVENT_NAMES_QUERY, _event_names_query_params(environment_key), ) except Exception: @@ -1121,7 +1121,7 @@ def _get_customer_warehouse_event_stats_cached( cached = cache.get(cache_key) if isinstance(cached, WarehouseEventStats): return cached - if cached == _CUSTOMER_EVENT_STATS_UNAVAILABLE: + if cached == _CUSTOMER_EVENT_UNAVAILABLE: return None try: with warehouse_delivery_service.delivery_client( @@ -1136,7 +1136,7 @@ def _get_customer_warehouse_event_stats_cached( except Exception: cache.set( cache_key, - _CUSTOMER_EVENT_STATS_UNAVAILABLE, + _CUSTOMER_EVENT_UNAVAILABLE, CUSTOMER_EVENT_STATS_CACHE_SECONDS, ) logger.warning( @@ -1159,7 +1159,7 @@ def _get_customer_warehouse_event_names_cached( cached = cache.get(cache_key) if isinstance(cached, WarehouseEventNames): return cached - if cached == _CUSTOMER_EVENT_NAMES_UNAVAILABLE: + if cached == _CUSTOMER_EVENT_UNAVAILABLE: return None try: with warehouse_delivery_service.delivery_client( @@ -1167,14 +1167,14 @@ def _get_customer_warehouse_event_names_cached( send_receive_timeout=CLICKHOUSE_VERIFY_TIMEOUT_SECONDS, ) as client: rows = client.query( - _EVENT_NAMES_QUERY, + _CLICKHOUSE_EVENT_NAMES_QUERY, parameters=_event_names_query_params(environment_key), ).result_rows except Exception: cache.set( cache_key, - _CUSTOMER_EVENT_NAMES_UNAVAILABLE, - CUSTOMER_EVENT_NAMES_CACHE_SECONDS, + _CUSTOMER_EVENT_UNAVAILABLE, + CUSTOMER_EVENT_NAMES_FAILURE_CACHE_SECONDS, ) logger.warning( "connection.event_names_failed", From 57fa33d2ca8460b522c110c05e556cde09a52b64 Mon Sep 17 00:00:00 2001 From: wadii Date: Thu, 6 Aug 2026 11:21:24 +0200 Subject: [PATCH 5/9] fix: clear customer warehouse caches when connection details change --- api/experimentation/services.py | 17 ++++++++-- api/experimentation/views.py | 8 +++-- api/tests/unit/experimentation/test_views.py | 34 ++++++++++++++++++++ 3 files changed, 55 insertions(+), 4 deletions(-) diff --git a/api/experimentation/services.py b/api/experimentation/services.py index 36d6d7ea1da4..274ede54b39b 100644 --- a/api/experimentation/services.py +++ b/api/experimentation/services.py @@ -110,6 +110,9 @@ _CUSTOMER_EVENT_UNAVAILABLE = "unavailable" +_CUSTOMER_EVENT_STATS_CACHE_KEY = "experimentation:customer_event_stats:{}" +_CUSTOMER_EVENT_NAMES_CACHE_KEY = "experimentation:customer_event_names:{}" + # A delivery run stops taking on new objects after this long, leaving room for # the slowest possible in-flight insert to still land inside the task timeout. DELIVERY_TIME_BUDGET_SECONDS = 210 @@ -1117,7 +1120,7 @@ def _get_customer_warehouse_event_stats_cached( ClickHouse instance, or None when it's unreachable. Results — including failures — are cached briefly so read endpoints don't open a connection to the customer's host on every request.""" - cache_key = f"experimentation:customer_event_stats:{connection.id}" + cache_key = _CUSTOMER_EVENT_STATS_CACHE_KEY.format(connection.id) cached = cache.get(cache_key) if isinstance(cached, WarehouseEventStats): return cached @@ -1155,7 +1158,7 @@ def _get_customer_warehouse_event_names_cached( ) -> WarehouseEventNames | None: """Query the customer's ClickHouse instance, caching results — including failures — to spare their host repeated connections.""" - cache_key = f"experimentation:customer_event_names:{connection.id}" + cache_key = _CUSTOMER_EVENT_NAMES_CACHE_KEY.format(connection.id) cached = cache.get(cache_key) if isinstance(cached, WarehouseEventNames): return cached @@ -1185,3 +1188,13 @@ def _get_customer_warehouse_event_names_cached( event_names = _build_event_names(rows) cache.set(cache_key, event_names, CUSTOMER_EVENT_NAMES_CACHE_SECONDS) return event_names + + +def clear_customer_warehouse_caches(connection: "WarehouseConnection") -> None: + """Drop cached warehouse reads so they don't outlive a connection change.""" + cache.delete_many( + [ + _CUSTOMER_EVENT_STATS_CACHE_KEY.format(connection.id), + _CUSTOMER_EVENT_NAMES_CACHE_KEY.format(connection.id), + ] + ) diff --git a/api/experimentation/views.py b/api/experimentation/views.py index 7a91b362c177..0d864675a770 100644 --- a/api/experimentation/views.py +++ b/api/experimentation/views.py @@ -62,6 +62,7 @@ from experimentation.services import ( annotate_warehouse_event_stats, apply_experiment_rollout, + clear_customer_warehouse_caches, create_experiment_audit_log, create_metric_audit_log, create_warehouse_audit_log, @@ -128,10 +129,13 @@ def perform_update(self, serializer: BaseSerializer[WarehouseConnection]) -> Non create_warehouse_audit_log( connection, self._get_user(self.request), action="updated" ) - if connection.warehouse_type == WarehouseType.CLICKHOUSE and ( + details_changed = ( "config" in serializer.validated_data or "credentials" in serializer.validated_data - ): + ) + if details_changed: + clear_customer_warehouse_caches(connection) + if connection.warehouse_type == WarehouseType.CLICKHOUSE and details_changed: verify_clickhouse_connection(connection) def perform_destroy(self, instance: WarehouseConnection) -> None: diff --git a/api/tests/unit/experimentation/test_views.py b/api/tests/unit/experimentation/test_views.py index 543c3b26969e..b8306a52e176 100644 --- a/api/tests/unit/experimentation/test_views.py +++ b/api/tests/unit/experimentation/test_views.py @@ -2,6 +2,7 @@ import pytest from clickhouse_connect.driver.exceptions import OperationalError +from django.core.cache import cache from django.urls import reverse from pytest_django.fixtures import SettingsWrapper from pytest_mock import MockerFixture @@ -1922,3 +1923,36 @@ def test_get_events__unsupported_type__returns_400( "detail": "Event listing is not supported for this warehouse type." } get_event_names.assert_not_called() + + +def test_patch__connection_details_changed__clears_customer_warehouse_caches( + admin_client: APIClient, + environment: Environment, + clickhouse_connection: WarehouseConnection, + enable_features: EnableFeaturesFixture, + reset_cache: None, + mocker: MockerFixture, +) -> None: + # Given + enable_features("experimentation_warehouse_connection") + mocker.patch("experimentation.views.verify_clickhouse_connection") + stats_key = f"experimentation:customer_event_stats:{clickhouse_connection.id}" + names_key = f"experimentation:customer_event_names:{clickhouse_connection.id}" + cache.set(stats_key, "stale", 300) + cache.set(names_key, "stale", 300) + url = reverse( + "api-v1:environments:experimentation:warehouse-connections-detail", + args=[environment.api_key, clickhouse_connection.id], + ) + + # When + response = admin_client.patch( + url, + data={"credentials": {"password": "new-password"}}, + format="json", + ) + + # Then + assert response.status_code == status.HTTP_200_OK + assert cache.get(stats_key) is None + assert cache.get(names_key) is None From 2b3617cafbb7c4a2572fd0c83cd380c18dc80a67 Mon Sep 17 00:00:00 2001 From: "flagsmith-engineering[bot]" Date: Thu, 6 Aug 2026 09:23:13 +0000 Subject: [PATCH 6/9] chore: Update documentation artefacts --- .../observability/_events-catalogue.md | 26 +++++++++---------- 1 file changed, 13 insertions(+), 13 deletions(-) diff --git a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md index f62d4c1f6fba..9870b6f0a0af 100644 --- a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md @@ -660,7 +660,7 @@ Attributes: ### `warehouse.connection.connected` Logged at `info` from: - - `api/experimentation/services.py:1079` + - `api/experimentation/services.py:1082` Attributes: - `environment.id` @@ -669,7 +669,7 @@ Attributes: ### `warehouse.connection.event_names_failed` Logged at `warning` from: - - `api/experimentation/services.py:1179` + - `api/experimentation/services.py:1182` Attributes: - `environment.id` @@ -678,7 +678,7 @@ Attributes: ### `warehouse.connection.event_stats_failed` Logged at `warning` from: - - `api/experimentation/services.py:1142` + - `api/experimentation/services.py:1145` Attributes: - `environment.id` @@ -687,7 +687,7 @@ Attributes: ### `warehouse.connection.test_event_sent` Logged at `info` from: - - `api/experimentation/services.py:853` + - `api/experimentation/services.py:856` Attributes: - `environment.id` @@ -696,7 +696,7 @@ Attributes: ### `warehouse.connection.verification_failed` Logged at `warning` from: - - `api/experimentation/services.py:1054` + - `api/experimentation/services.py:1057` Attributes: - `environment.id` @@ -706,7 +706,7 @@ Attributes: ### `warehouse.connection.verification_succeeded` Logged at `info` from: - - `api/experimentation/services.py:1064` + - `api/experimentation/services.py:1067` Attributes: - `environment.id` @@ -715,7 +715,7 @@ Attributes: ### `warehouse.delivery.all_objects_rejected` Logged at `error` from: - - `api/experimentation/services.py:1009` + - `api/experimentation/services.py:1012` Attributes: - `connection.id` @@ -726,7 +726,7 @@ Attributes: ### `warehouse.delivery.budget_exhausted` Logged at `info` from: - - `api/experimentation/services.py:898` + - `api/experimentation/services.py:901` Attributes: - `connection.id` @@ -737,7 +737,7 @@ Attributes: ### `warehouse.delivery.completed` Logged at `info` from: - - `api/experimentation/services.py:1019` + - `api/experimentation/services.py:1022` Attributes: - `connection.id` @@ -750,7 +750,7 @@ Attributes: ### `warehouse.delivery.failed` Logged at `error` from: - - `api/experimentation/services.py:992` + - `api/experimentation/services.py:995` Attributes: - `connection.id` @@ -761,7 +761,7 @@ Attributes: ### `warehouse.delivery.object_rejected` Logged at `error` from: - - `api/experimentation/services.py:927` + - `api/experimentation/services.py:930` Attributes: - `connection.id` @@ -773,7 +773,7 @@ Attributes: ### `warehouse.srm.overallocated` Logged at `error` from: - - `api/experimentation/services.py:475` + - `api/experimentation/services.py:478` Attributes: - `environment.id` @@ -783,7 +783,7 @@ Attributes: ### `warehouse.srm.unkeyed_variant` Logged at `error` from: - - `api/experimentation/services.py:461` + - `api/experimentation/services.py:464` Attributes: - `environment.id` From 56ca57451bf6a3741af98583a02d6ef27aa68b38 Mon Sep 17 00:00:00 2001 From: wadii Date: Thu, 6 Aug 2026 12:08:36 +0200 Subject: [PATCH 7/9] fix: race-safe warehouse cache keys and documented error responses --- api/experimentation/services.py | 29 ++++++------- api/experimentation/views.py | 18 ++++---- .../unit/experimentation/test_services.py | 41 +++++++++++++++++++ api/tests/unit/experimentation/test_views.py | 34 --------------- .../observability/_events-catalogue.md | 26 ++++++------ 5 files changed, 80 insertions(+), 68 deletions(-) diff --git a/api/experimentation/services.py b/api/experimentation/services.py index 274ede54b39b..8dcbde58a4d6 100644 --- a/api/experimentation/services.py +++ b/api/experimentation/services.py @@ -1,5 +1,7 @@ from __future__ import annotations +import hashlib +import json import time import typing from dataclasses import replace @@ -110,8 +112,17 @@ _CUSTOMER_EVENT_UNAVAILABLE = "unavailable" -_CUSTOMER_EVENT_STATS_CACHE_KEY = "experimentation:customer_event_stats:{}" -_CUSTOMER_EVENT_NAMES_CACHE_KEY = "experimentation:customer_event_names:{}" + +def _customer_cache_key(kind: str, connection: "WarehouseConnection") -> str: + """Key cached warehouse reads by the connection's details, so a config, + credential, or type change can neither serve nor store stale reads.""" + details = json.dumps( + [connection.warehouse_type, connection.config, connection.credentials], + sort_keys=True, + ) + digest = hashlib.sha256(details.encode()).hexdigest()[:12] + return f"experimentation:customer_{kind}:{connection.id}:{digest}" + # A delivery run stops taking on new objects after this long, leaving room for # the slowest possible in-flight insert to still land inside the task timeout. @@ -1120,7 +1131,7 @@ def _get_customer_warehouse_event_stats_cached( ClickHouse instance, or None when it's unreachable. Results — including failures — are cached briefly so read endpoints don't open a connection to the customer's host on every request.""" - cache_key = _CUSTOMER_EVENT_STATS_CACHE_KEY.format(connection.id) + cache_key = _customer_cache_key("event_stats", connection) cached = cache.get(cache_key) if isinstance(cached, WarehouseEventStats): return cached @@ -1158,7 +1169,7 @@ def _get_customer_warehouse_event_names_cached( ) -> WarehouseEventNames | None: """Query the customer's ClickHouse instance, caching results — including failures — to spare their host repeated connections.""" - cache_key = _CUSTOMER_EVENT_NAMES_CACHE_KEY.format(connection.id) + cache_key = _customer_cache_key("event_names", connection) cached = cache.get(cache_key) if isinstance(cached, WarehouseEventNames): return cached @@ -1188,13 +1199,3 @@ def _get_customer_warehouse_event_names_cached( event_names = _build_event_names(rows) cache.set(cache_key, event_names, CUSTOMER_EVENT_NAMES_CACHE_SECONDS) return event_names - - -def clear_customer_warehouse_caches(connection: "WarehouseConnection") -> None: - """Drop cached warehouse reads so they don't outlive a connection change.""" - cache.delete_many( - [ - _CUSTOMER_EVENT_STATS_CACHE_KEY.format(connection.id), - _CUSTOMER_EVENT_NAMES_CACHE_KEY.format(connection.id), - ] - ) diff --git a/api/experimentation/views.py b/api/experimentation/views.py index 0d864675a770..456a6e51c36e 100644 --- a/api/experimentation/views.py +++ b/api/experimentation/views.py @@ -62,7 +62,6 @@ from experimentation.services import ( annotate_warehouse_event_stats, apply_experiment_rollout, - clear_customer_warehouse_caches, create_experiment_audit_log, create_metric_audit_log, create_warehouse_audit_log, @@ -129,13 +128,10 @@ def perform_update(self, serializer: BaseSerializer[WarehouseConnection]) -> Non create_warehouse_audit_log( connection, self._get_user(self.request), action="updated" ) - details_changed = ( + if connection.warehouse_type == WarehouseType.CLICKHOUSE and ( "config" in serializer.validated_data or "credentials" in serializer.validated_data - ) - if details_changed: - clear_customer_warehouse_caches(connection) - if connection.warehouse_type == WarehouseType.CLICKHOUSE and details_changed: + ): verify_clickhouse_connection(connection) def perform_destroy(self, instance: WarehouseConnection) -> None: @@ -237,7 +233,15 @@ def test_warehouse_connection_config( "events": serializers.ListField(child=serializers.CharField()), "is_truncated": serializers.BooleanField(), }, - ) + ), + 400: inline_serializer( + name="WarehouseEventNamesUnsupported", + fields={"detail": serializers.CharField()}, + ), + 503: inline_serializer( + name="WarehouseEventNamesUnavailable", + fields={"detail": serializers.CharField()}, + ), }, ) @action(detail=True, methods=["get"], url_path="events") diff --git a/api/tests/unit/experimentation/test_services.py b/api/tests/unit/experimentation/test_services.py index 9d070c7143b7..1d2af6f8836e 100644 --- a/api/tests/unit/experimentation/test_services.py +++ b/api/tests/unit/experimentation/test_services.py @@ -241,6 +241,47 @@ def test_get_warehouse_event_names__clickhouse_connection__queries_customer_inst assert second_result == expected +@pytest.mark.parametrize( + "changed_field, new_value", + [ + ("config", {"host": "new.acme-corp.example"}), + ("credentials", {"password": "rotated"}), + ], + ids=["config", "credentials"], +) +def test_get_warehouse_event_names__connection_details_changed__bypasses_cache( + clickhouse_connection: WarehouseConnection, + reset_cache: None, + changed_field: str, + new_value: dict[str, str], + mocker: MockerFixture, +) -> None: + # Given — a cached result for the connection's current details + get_client = mocker.patch( + "experimentation.warehouse_delivery_service.clickhouse_connect.get_client", + ) + get_client.return_value.query.return_value = mocker.Mock( + result_rows=[("old_event",)] + ) + services.get_warehouse_event_names(clickhouse_connection, "test-env-key") + + # When — the connection details change + setattr( + clickhouse_connection, + changed_field, + {**getattr(clickhouse_connection, changed_field), **new_value}, + ) + clickhouse_connection.save() + get_client.return_value.query.return_value = mocker.Mock( + result_rows=[("new_event",)] + ) + result = services.get_warehouse_event_names(clickhouse_connection, "test-env-key") + + # Then — the stale cache entry is not served + assert result == WarehouseEventNames(events=["new_event"], is_truncated=False) + assert get_client.return_value.query.call_count == 2 + + def test_get_exposure_buckets__day_granularity__queries_and_maps_rows( mocker: MockerFixture, ) -> None: diff --git a/api/tests/unit/experimentation/test_views.py b/api/tests/unit/experimentation/test_views.py index b8306a52e176..543c3b26969e 100644 --- a/api/tests/unit/experimentation/test_views.py +++ b/api/tests/unit/experimentation/test_views.py @@ -2,7 +2,6 @@ import pytest from clickhouse_connect.driver.exceptions import OperationalError -from django.core.cache import cache from django.urls import reverse from pytest_django.fixtures import SettingsWrapper from pytest_mock import MockerFixture @@ -1923,36 +1922,3 @@ def test_get_events__unsupported_type__returns_400( "detail": "Event listing is not supported for this warehouse type." } get_event_names.assert_not_called() - - -def test_patch__connection_details_changed__clears_customer_warehouse_caches( - admin_client: APIClient, - environment: Environment, - clickhouse_connection: WarehouseConnection, - enable_features: EnableFeaturesFixture, - reset_cache: None, - mocker: MockerFixture, -) -> None: - # Given - enable_features("experimentation_warehouse_connection") - mocker.patch("experimentation.views.verify_clickhouse_connection") - stats_key = f"experimentation:customer_event_stats:{clickhouse_connection.id}" - names_key = f"experimentation:customer_event_names:{clickhouse_connection.id}" - cache.set(stats_key, "stale", 300) - cache.set(names_key, "stale", 300) - url = reverse( - "api-v1:environments:experimentation:warehouse-connections-detail", - args=[environment.api_key, clickhouse_connection.id], - ) - - # When - response = admin_client.patch( - url, - data={"credentials": {"password": "new-password"}}, - format="json", - ) - - # Then - assert response.status_code == status.HTTP_200_OK - assert cache.get(stats_key) is None - assert cache.get(names_key) is None diff --git a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md index 9870b6f0a0af..63978bd1ea68 100644 --- a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md @@ -660,7 +660,7 @@ Attributes: ### `warehouse.connection.connected` Logged at `info` from: - - `api/experimentation/services.py:1082` + - `api/experimentation/services.py:1093` Attributes: - `environment.id` @@ -669,7 +669,7 @@ Attributes: ### `warehouse.connection.event_names_failed` Logged at `warning` from: - - `api/experimentation/services.py:1182` + - `api/experimentation/services.py:1193` Attributes: - `environment.id` @@ -678,7 +678,7 @@ Attributes: ### `warehouse.connection.event_stats_failed` Logged at `warning` from: - - `api/experimentation/services.py:1145` + - `api/experimentation/services.py:1156` Attributes: - `environment.id` @@ -687,7 +687,7 @@ Attributes: ### `warehouse.connection.test_event_sent` Logged at `info` from: - - `api/experimentation/services.py:856` + - `api/experimentation/services.py:867` Attributes: - `environment.id` @@ -696,7 +696,7 @@ Attributes: ### `warehouse.connection.verification_failed` Logged at `warning` from: - - `api/experimentation/services.py:1057` + - `api/experimentation/services.py:1068` Attributes: - `environment.id` @@ -706,7 +706,7 @@ Attributes: ### `warehouse.connection.verification_succeeded` Logged at `info` from: - - `api/experimentation/services.py:1067` + - `api/experimentation/services.py:1078` Attributes: - `environment.id` @@ -715,7 +715,7 @@ Attributes: ### `warehouse.delivery.all_objects_rejected` Logged at `error` from: - - `api/experimentation/services.py:1012` + - `api/experimentation/services.py:1023` Attributes: - `connection.id` @@ -726,7 +726,7 @@ Attributes: ### `warehouse.delivery.budget_exhausted` Logged at `info` from: - - `api/experimentation/services.py:901` + - `api/experimentation/services.py:912` Attributes: - `connection.id` @@ -737,7 +737,7 @@ Attributes: ### `warehouse.delivery.completed` Logged at `info` from: - - `api/experimentation/services.py:1022` + - `api/experimentation/services.py:1033` Attributes: - `connection.id` @@ -750,7 +750,7 @@ Attributes: ### `warehouse.delivery.failed` Logged at `error` from: - - `api/experimentation/services.py:995` + - `api/experimentation/services.py:1006` Attributes: - `connection.id` @@ -761,7 +761,7 @@ Attributes: ### `warehouse.delivery.object_rejected` Logged at `error` from: - - `api/experimentation/services.py:930` + - `api/experimentation/services.py:941` Attributes: - `connection.id` @@ -773,7 +773,7 @@ Attributes: ### `warehouse.srm.overallocated` Logged at `error` from: - - `api/experimentation/services.py:478` + - `api/experimentation/services.py:489` Attributes: - `environment.id` @@ -783,7 +783,7 @@ Attributes: ### `warehouse.srm.unkeyed_variant` Logged at `error` from: - - `api/experimentation/services.py:464` + - `api/experimentation/services.py:475` Attributes: - `environment.id` From d59503b6c3ae9280886b8dd2ee253feb881fc597 Mon Sep 17 00:00:00 2001 From: "flagsmith-engineering[bot]" Date: Thu, 6 Aug 2026 10:11:42 +0000 Subject: [PATCH 8/9] chore: Update documentation artefacts --- openapi.yaml | 26 ++++++++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/openapi.yaml b/openapi.yaml index 86dbc206bb4a..ccfdf0f08579 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -7092,6 +7092,18 @@ paths: application/json: schema: $ref: '#/components/schemas/WarehouseEventNamesResult' + '400': + description: '' + content: + application/json: + schema: + $ref: '#/components/schemas/WarehouseEventNamesUnsupported' + '503': + description: '' + content: + application/json: + schema: + $ref: '#/components/schemas/WarehouseEventNamesUnavailable' security: - tokenAuth: [] - Master API Key: [] @@ -27990,6 +28002,20 @@ components: required: - events - is_truncated + WarehouseEventNamesUnavailable: + type: object + properties: + detail: + type: string + required: + - detail + WarehouseEventNamesUnsupported: + type: object + properties: + detail: + type: string + required: + - detail WarehouseTypeEnum: description: |- * `flagsmith` - Flagsmith From bffe6537692f084238b27b8a5c66a454e74116c4 Mon Sep 17 00:00:00 2001 From: wadii Date: Thu, 6 Aug 2026 14:15:02 +0200 Subject: [PATCH 9/9] refactor: address review on event names caching, timeouts and dispatch --- api/experimentation/services.py | 49 ++++++++++++++----- api/experimentation/views.py | 6 +-- .../unit/experimentation/test_services.py | 49 ++++++++++++++++--- .../observability/_events-catalogue.md | 28 ++++++----- 4 files changed, 95 insertions(+), 37 deletions(-) diff --git a/api/experimentation/services.py b/api/experimentation/services.py index 8dcbde58a4d6..ee8c1c2bff75 100644 --- a/api/experimentation/services.py +++ b/api/experimentation/services.py @@ -105,8 +105,9 @@ CLICKHOUSE_CONNECT_TIMEOUT_SECONDS = 5 CLICKHOUSE_QUERY_TIMEOUT_SECONDS = 30 CLICKHOUSE_VERIFY_TIMEOUT_SECONDS = 5 +CLICKHOUSE_EVENT_NAMES_TIMEOUT_SECONDS = 15 CUSTOMER_EVENT_STATS_CACHE_SECONDS = 60 -CUSTOMER_EVENT_NAMES_CACHE_SECONDS = 300 +EVENT_NAMES_CACHE_SECONDS = 300 CUSTOMER_EVENT_NAMES_FAILURE_CACHE_SECONDS = 60 WAREHOUSE_EVENT_NAMES_LIMIT = 500 @@ -114,10 +115,12 @@ def _customer_cache_key(kind: str, connection: "WarehouseConnection") -> str: - """Key cached warehouse reads by the connection's details, so a config, - credential, or type change can neither serve nor store stale reads.""" + """Key cached warehouse reads by the connection's non-secret details, so a + config or type change can neither serve nor store stale reads. Credentials + stay out of the key material: they don't determine what the warehouse + holds, so rotating them keeps the cache valid.""" details = json.dumps( - [connection.warehouse_type, connection.config, connection.credentials], + [connection.warehouse_type, connection.config], sort_keys=True, ) digest = hashlib.sha256(details.encode()).hexdigest()[:12] @@ -185,25 +188,47 @@ def _build_event_names( ) +EVENT_NAMES_SUPPORTED_WAREHOUSE_TYPES = ( + WarehouseType.FLAGSMITH, + WarehouseType.CLICKHOUSE, +) + + def get_warehouse_event_names( connection: "WarehouseConnection", environment_key: str, ) -> WarehouseEventNames | None: - """Return the distinct event names recorded for `environment_key`, most - recently seen first, capped at WAREHOUSE_EVENT_NAMES_LIMIT; None when the - warehouse is unavailable.""" if connection.warehouse_type == WarehouseType.CLICKHOUSE: - return _get_customer_warehouse_event_names_cached(connection, environment_key) + return _get_customer_clickhouse_event_names(connection, environment_key) + if connection.warehouse_type == WarehouseType.FLAGSMITH: + return _get_flagsmith_clickhouse_event_names(environment_key) + raise ValueError(f"Unsupported warehouse type: {connection.warehouse_type}") + + +def _get_flagsmith_clickhouse_event_names( + environment_key: str, +) -> WarehouseEventNames | None: if not settings.EXPERIMENTATION_CLICKHOUSE_URL: return None + cache_key = f"experimentation:event_names:{environment_key}" + cached = cache.get(cache_key) + if isinstance(cached, WarehouseEventNames): + return cached try: rows = _get_clickhouse_client().execute( _CLICKHOUSE_EVENT_NAMES_QUERY, _event_names_query_params(environment_key), ) except Exception: + logger.warning( + "connection.event_names_failed", + environment__key=environment_key, + exc_info=True, + ) return None - return _build_event_names(rows) + event_names = _build_event_names(rows) + cache.set(cache_key, event_names, EVENT_NAMES_CACHE_SECONDS) + return event_names _EVENT_STATS_QUERY = ( @@ -1163,7 +1188,7 @@ def _get_customer_warehouse_event_stats_cached( return stats -def _get_customer_warehouse_event_names_cached( +def _get_customer_clickhouse_event_names( connection: "WarehouseConnection", environment_key: str, ) -> WarehouseEventNames | None: @@ -1178,7 +1203,7 @@ def _get_customer_warehouse_event_names_cached( try: with warehouse_delivery_service.delivery_client( connection, - send_receive_timeout=CLICKHOUSE_VERIFY_TIMEOUT_SECONDS, + send_receive_timeout=CLICKHOUSE_EVENT_NAMES_TIMEOUT_SECONDS, ) as client: rows = client.query( _CLICKHOUSE_EVENT_NAMES_QUERY, @@ -1197,5 +1222,5 @@ def _get_customer_warehouse_event_names_cached( ) return None event_names = _build_event_names(rows) - cache.set(cache_key, event_names, CUSTOMER_EVENT_NAMES_CACHE_SECONDS) + cache.set(cache_key, event_names, EVENT_NAMES_CACHE_SECONDS) return event_names diff --git a/api/experimentation/views.py b/api/experimentation/views.py index 456a6e51c36e..5ab97d319c3c 100644 --- a/api/experimentation/views.py +++ b/api/experimentation/views.py @@ -60,6 +60,7 @@ WarehouseConnectionSerializer, ) from experimentation.services import ( + EVENT_NAMES_SUPPORTED_WAREHOUSE_TYPES, annotate_warehouse_event_stats, apply_experiment_rollout, create_experiment_audit_log, @@ -248,10 +249,7 @@ def test_warehouse_connection_config( def events(self, request: Request, **kwargs: object) -> Response: """List the distinct event names in the connection's warehouse.""" connection: WarehouseConnection = self.get_object() - if connection.warehouse_type not in ( - WarehouseType.FLAGSMITH, - WarehouseType.CLICKHOUSE, - ): + if connection.warehouse_type not in EVENT_NAMES_SUPPORTED_WAREHOUSE_TYPES: return Response( {"detail": "Event listing is not supported for this warehouse type."}, status=status.HTTP_400_BAD_REQUEST, diff --git a/api/tests/unit/experimentation/test_services.py b/api/tests/unit/experimentation/test_services.py index 1d2af6f8836e..19fbb481b840 100644 --- a/api/tests/unit/experimentation/test_services.py +++ b/api/tests/unit/experimentation/test_services.py @@ -129,6 +129,7 @@ def test_get_clickhouse_client__dsn_timeouts__are_preserved( def test_get_warehouse_event_names__flagsmith_connection__returns_capped_names( warehouse_connection: WarehouseConnection, settings: SettingsWrapper, + reset_cache: None, rows: list[tuple[str]], expected: WarehouseEventNames, mocker: MockerFixture, @@ -154,6 +155,15 @@ def test_get_warehouse_event_names__flagsmith_connection__returns_capped_names( {"environment_key": "env-key-123", "limit": 501}, ) + # When — the result is cached, so a second request doesn't hit the warehouse + second_result = services.get_warehouse_event_names( + warehouse_connection, "env-key-123" + ) + + # Then + mock_client.execute.assert_called_once() + assert second_result == expected + @pytest.mark.parametrize( "clickhouse_url, execute_side_effect", @@ -166,8 +176,10 @@ def test_get_warehouse_event_names__flagsmith_connection__returns_capped_names( def test_get_warehouse_event_names__flagsmith_warehouse_unavailable__returns_none( warehouse_connection: WarehouseConnection, settings: SettingsWrapper, + reset_cache: None, clickhouse_url: str, execute_side_effect: Exception | None, + log: StructuredLogCapture, mocker: MockerFixture, ) -> None: # Given @@ -184,6 +196,9 @@ def test_get_warehouse_event_names__flagsmith_warehouse_unavailable__returns_non # Then assert result is None + assert any( + event["event"] == "connection.event_names_failed" for event in log.events + ) == (execute_side_effect is not None) @pytest.mark.parametrize( @@ -242,18 +257,20 @@ def test_get_warehouse_event_names__clickhouse_connection__queries_customer_inst @pytest.mark.parametrize( - "changed_field, new_value", + "changed_field, new_value, expected_events, expected_query_count", [ - ("config", {"host": "new.acme-corp.example"}), - ("credentials", {"password": "rotated"}), + ("config", {"host": "new.acme-corp.example"}, ["new_event"], 2), + ("credentials", {"password": "rotated"}, ["old_event"], 1), ], - ids=["config", "credentials"], + ids=["config-bypasses-cache", "credentials-keep-cache"], ) -def test_get_warehouse_event_names__connection_details_changed__bypasses_cache( +def test_get_warehouse_event_names__connection_details_changed__cache_keyed_by_config( clickhouse_connection: WarehouseConnection, reset_cache: None, changed_field: str, new_value: dict[str, str], + expected_events: list[str], + expected_query_count: int, mocker: MockerFixture, ) -> None: # Given — a cached result for the connection's current details @@ -277,9 +294,25 @@ def test_get_warehouse_event_names__connection_details_changed__bypasses_cache( ) result = services.get_warehouse_event_names(clickhouse_connection, "test-env-key") - # Then — the stale cache entry is not served - assert result == WarehouseEventNames(events=["new_event"], is_truncated=False) - assert get_client.return_value.query.call_count == 2 + # Then — a config change re-queries; a credential rotation keeps the cache + assert result == WarehouseEventNames(events=expected_events, is_truncated=False) + assert get_client.return_value.query.call_count == expected_query_count + + +def test_get_warehouse_event_names__unsupported_type__raises( + environment: Environment, +) -> None: + # Given + connection = WarehouseConnection( + environment=environment, + warehouse_type=WarehouseType.SNOWFLAKE, + name="Snowflake", + config={"account_identifier": "acme"}, + ) + + # When / Then + with pytest.raises(ValueError, match="Unsupported warehouse type"): + services.get_warehouse_event_names(connection, "test-env-key") def test_get_exposure_buckets__day_granularity__queries_and_maps_rows( diff --git a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md index 63978bd1ea68..591712fced24 100644 --- a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md @@ -660,7 +660,7 @@ Attributes: ### `warehouse.connection.connected` Logged at `info` from: - - `api/experimentation/services.py:1093` + - `api/experimentation/services.py:1118` Attributes: - `environment.id` @@ -669,16 +669,18 @@ Attributes: ### `warehouse.connection.event_names_failed` Logged at `warning` from: - - `api/experimentation/services.py:1193` + - `api/experimentation/services.py:223` + - `api/experimentation/services.py:1218` Attributes: - `environment.id` + - `environment.key` - `exc_info` ### `warehouse.connection.event_stats_failed` Logged at `warning` from: - - `api/experimentation/services.py:1156` + - `api/experimentation/services.py:1181` Attributes: - `environment.id` @@ -687,7 +689,7 @@ Attributes: ### `warehouse.connection.test_event_sent` Logged at `info` from: - - `api/experimentation/services.py:867` + - `api/experimentation/services.py:892` Attributes: - `environment.id` @@ -696,7 +698,7 @@ Attributes: ### `warehouse.connection.verification_failed` Logged at `warning` from: - - `api/experimentation/services.py:1068` + - `api/experimentation/services.py:1093` Attributes: - `environment.id` @@ -706,7 +708,7 @@ Attributes: ### `warehouse.connection.verification_succeeded` Logged at `info` from: - - `api/experimentation/services.py:1078` + - `api/experimentation/services.py:1103` Attributes: - `environment.id` @@ -715,7 +717,7 @@ Attributes: ### `warehouse.delivery.all_objects_rejected` Logged at `error` from: - - `api/experimentation/services.py:1023` + - `api/experimentation/services.py:1048` Attributes: - `connection.id` @@ -726,7 +728,7 @@ Attributes: ### `warehouse.delivery.budget_exhausted` Logged at `info` from: - - `api/experimentation/services.py:912` + - `api/experimentation/services.py:937` Attributes: - `connection.id` @@ -737,7 +739,7 @@ Attributes: ### `warehouse.delivery.completed` Logged at `info` from: - - `api/experimentation/services.py:1033` + - `api/experimentation/services.py:1058` Attributes: - `connection.id` @@ -750,7 +752,7 @@ Attributes: ### `warehouse.delivery.failed` Logged at `error` from: - - `api/experimentation/services.py:1006` + - `api/experimentation/services.py:1031` Attributes: - `connection.id` @@ -761,7 +763,7 @@ Attributes: ### `warehouse.delivery.object_rejected` Logged at `error` from: - - `api/experimentation/services.py:941` + - `api/experimentation/services.py:966` Attributes: - `connection.id` @@ -773,7 +775,7 @@ Attributes: ### `warehouse.srm.overallocated` Logged at `error` from: - - `api/experimentation/services.py:489` + - `api/experimentation/services.py:514` Attributes: - `environment.id` @@ -783,7 +785,7 @@ Attributes: ### `warehouse.srm.unkeyed_variant` Logged at `error` from: - - `api/experimentation/services.py:475` + - `api/experimentation/services.py:500` Attributes: - `environment.id`