From 361572db495212206f5c1d47ef6cd1e6b7bc4bd0 Mon Sep 17 00:00:00 2001 From: wadii Date: Mon, 27 Jul 2026 17:31:42 +0200 Subject: [PATCH 1/3] feat: verify events table exists, clickhouse event stats, default flagsmith_exp database --- api/experimentation/services.py | 95 ++++++++++++++----- api/experimentation/types.py | 2 +- .../unit/experimentation/test_services.py | 58 ++++++++++- api/tests/unit/experimentation/test_views.py | 4 +- .../observability/_events-catalogue.md | 6 +- 5 files changed, 134 insertions(+), 31 deletions(-) diff --git a/api/experimentation/services.py b/api/experimentation/services.py index 704aa6245cdf..5ee7b33871fd 100644 --- a/api/experimentation/services.py +++ b/api/experimentation/services.py @@ -799,6 +799,31 @@ class InternalAddressError(Exception): pass +class MissingEventsTableError(Exception): + pass + + +def _connect_customer_clickhouse(connection: WarehouseConnection) -> Client: + """Build a client for the customer's ClickHouse instance from stored + connection details, re-checking the host against internal address ranges + first: DNS may resolve differently than it did at validation time, and + rows may predate host validation.""" + config = typing.cast(ClickHouseConfig, connection.config or {}) + credentials = typing.cast(ClickHouseCredentials, connection.credentials or {}) + if is_internal_address(config["host"], include_shared=True): + raise InternalAddressError(config["host"]) + return Client( + config["host"], + port=config["port"], + user=config["username"], + password=credentials["password"], + database=config["database"], + secure=config["secure"], + connect_timeout=CLICKHOUSE_CONNECT_TIMEOUT_SECONDS, + send_receive_timeout=CLICKHOUSE_VERIFY_TIMEOUT_SECONDS, + ) + + def _describe_verification_error(error: Exception) -> str: if isinstance(error, clickhouse_errors.ServerException): # 516 = AUTHENTICATION_FAILED (not in clickhouse_driver.errors.ErrorCodes) @@ -813,6 +838,11 @@ def _describe_verification_error(error: Exception) -> str: return "Could not connect to the host." if isinstance(error, InternalAddressError): return "Host must not target internal or private network addresses." + if isinstance(error, MissingEventsTableError): + return ( + "Events table not found in the configured database. " + "Run the setup SQL to create it." + ) if isinstance(error, KeyError): return "Stored connection details are incomplete." return "Verification failed." @@ -822,30 +852,19 @@ def verify_clickhouse_connection( connection: WarehouseConnection, persist: bool = True, ) -> None: - """Run SELECT 1 against the customer's ClickHouse and set the status to - connected or errored; never raises. With persist=False, the status is only - set on the in-memory instance, allowing unsaved connections to be tested.""" + """Run SELECT 1 against the customer's ClickHouse, check the events table + exists, and set the status to connected or errored; never raises. With + persist=False, the status is only set on the in-memory instance, allowing + unsaved connections to be tested.""" log = logger.bind(environment__id=connection.environment_id) try: log = log.bind(organisation__id=connection.environment.project.organisation_id) - config = typing.cast(ClickHouseConfig, connection.config or {}) - credentials = typing.cast(ClickHouseCredentials, connection.credentials or {}) - # Re-check right before connecting: DNS may resolve differently than it - # did at validation time, and rows may predate host validation. - if is_internal_address(config["host"], include_shared=True): - raise InternalAddressError(config["host"]) - client = Client( - config["host"], - port=config["port"], - user=config["username"], - password=credentials["password"], - database=config["database"], - secure=config["secure"], - connect_timeout=CLICKHOUSE_CONNECT_TIMEOUT_SECONDS, - send_receive_timeout=CLICKHOUSE_VERIFY_TIMEOUT_SECONDS, - ) + client = _connect_customer_clickhouse(connection) try: client.execute("SELECT 1") + rows = client.execute("EXISTS TABLE events") + if not rows[0][0]: + raise MissingEventsTableError() finally: client.disconnect() except Exception as error: @@ -893,9 +912,19 @@ def annotate_warehouse_event_stats( connection: WarehouseConnection, environment_key: str, ) -> None: - """Attach live warehouse event stats to a flagsmith connection. No-op for - non-flagsmith connections or when no warehouse is configured; leaves stats - unset when the warehouse is unreachable. Read-only: never changes status.""" + """Attach live warehouse event stats to a connection — from the managed + warehouse for flagsmith connections, from the customer's instance for + clickhouse ones. No-op for other types or when no warehouse is configured; + leaves stats unset when the warehouse is unreachable. Read-only: never + changes status.""" + if connection.warehouse_type == WarehouseType.CLICKHOUSE: + try: + connection.event_stats = _get_customer_warehouse_event_stats( + connection, environment_key + ) + except Exception: + pass + return if ( connection.warehouse_type != WarehouseType.FLAGSMITH or not settings.EXPERIMENTATION_CLICKHOUSE_URL @@ -905,3 +934,25 @@ def annotate_warehouse_event_stats( connection.event_stats = get_warehouse_event_stats(environment_key) except Exception: return + + +def _get_customer_warehouse_event_stats( + connection: WarehouseConnection, + environment_key: str, +) -> WarehouseEventStats: + """Return event counts recorded for `environment_key` in the customer's + ClickHouse instance.""" + client = _connect_customer_clickhouse(connection) + try: + rows = client.execute( + "SELECT count() AS total, uniqExact(event) AS unique " + "FROM events WHERE environment_key = %(environment_key)s", + {"environment_key": environment_key}, + ) + finally: + client.disconnect() + total, unique = rows[0] if rows else (0, 0) + return WarehouseEventStats( + total_events_received=int(total), + unique_events_count=int(unique), + ) diff --git a/api/experimentation/types.py b/api/experimentation/types.py index 6122499cb6b8..60751ff3508f 100644 --- a/api/experimentation/types.py +++ b/api/experimentation/types.py @@ -57,7 +57,7 @@ class ClickHouseConfig(TypedDict): CLICKHOUSE_DEFAULTS: ClickHouseConfig = { "host": "", "port": 9440, - "database": "flagsmith", + "database": "flagsmith_exp", "username": "default", "secure": True, } diff --git a/api/tests/unit/experimentation/test_services.py b/api/tests/unit/experimentation/test_services.py index 9ee8437f86ca..9c970c939af4 100644 --- a/api/tests/unit/experimentation/test_services.py +++ b/api/tests/unit/experimentation/test_services.py @@ -40,7 +40,9 @@ from experimentation.results_query import _MetricSlot from experimentation.services import ( InternalAddressError, + MissingEventsTableError, _describe_verification_error, + annotate_warehouse_event_stats, verify_clickhouse_connection, ) from experimentation.stats import VariantStats @@ -2031,6 +2033,7 @@ def test_verify_clickhouse_connection__reachable__sets_connected( ) -> None: # Given mock_client = mocker.patch("experimentation.services.Client") + mock_client.return_value.execute.return_value = [(1,)] success_count_before = _verification_count("success") clickhouse_connection.status_detail = "stale detail" clickhouse_connection.save() @@ -2052,7 +2055,10 @@ def test_verify_clickhouse_connection__reachable__sets_connected( connect_timeout=5, send_receive_timeout=5, ) - mock_client.return_value.execute.assert_called_once_with("SELECT 1") + assert mock_client.return_value.execute.call_args_list == [ + mocker.call("SELECT 1"), + mocker.call("EXISTS TABLE events"), + ] mock_client.return_value.disconnect.assert_called_once_with() assert _verification_count("success") == success_count_before + 1 assert { @@ -2071,14 +2077,20 @@ def test_verify_clickhouse_connection__reachable__sets_connected( Exception("connection refused"), "Verification failed.", ), + ( + {"password": "hunter2"}, + [[(1,)], [(0,)]], + "Events table not found in the configured database. " + "Run the setup SQL to create it.", + ), (None, None, "Stored connection details are incomplete."), ], - ids=["driver_error", "missing_credentials"], + ids=["driver_error", "missing_events_table", "missing_credentials"], ) def test_verify_clickhouse_connection__failure__sets_errored_with_detail( clickhouse_connection: WarehouseConnection, credentials: dict[str, str] | None, - execute_side_effect: Exception | None, + execute_side_effect: Exception | list[list[tuple[int]]] | None, expected_detail: str, log: StructuredLogCapture, mocker: MockerFixture, @@ -2161,6 +2173,11 @@ def test_verify_clickhouse_connection__internal_host__sets_errored_without_conne InternalAddressError("10.0.0.5"), "Host must not target internal or private network addresses.", ), + ( + MissingEventsTableError(), + "Events table not found in the configured database. " + "Run the setup SQL to create it.", + ), ( KeyError("host"), "Stored connection details are incomplete.", @@ -2178,6 +2195,7 @@ def test_verify_clickhouse_connection__internal_host__sets_errored_without_conne "builtin_timeout_error", "network_error", "internal_address_error", + "missing_events_table_error", "key_error", "generic_exception", ], @@ -2191,3 +2209,37 @@ def test_describe_verification_error__known_error_types__returns_expected_detail # Then assert detail == expected_detail + + +@pytest.mark.parametrize( + "execute_side_effect, expected_stats", + [ + ( + [[(42, 7)]], + WarehouseEventStats(total_events_received=42, unique_events_count=7), + ), + (Exception("connection refused"), None), + ], + ids=["reachable", "unreachable"], +) +def test_annotate_warehouse_event_stats__clickhouse_connection__queries_customer_instance( + clickhouse_connection: WarehouseConnection, + execute_side_effect: Exception | list[list[tuple[int, int]]], + expected_stats: WarehouseEventStats | None, + mocker: MockerFixture, +) -> None: + # Given + mock_client = mocker.patch("experimentation.services.Client") + mock_client.return_value.execute.side_effect = execute_side_effect + + # When + annotate_warehouse_event_stats(clickhouse_connection, "test-env-key") + + # Then + assert getattr(clickhouse_connection, "event_stats", None) == expected_stats + mock_client.return_value.execute.assert_called_once_with( + "SELECT count() AS total, uniqExact(event) AS unique " + "FROM events WHERE environment_key = %(environment_key)s", + {"environment_key": "test-env-key"}, + ) + mock_client.return_value.disconnect.assert_called_once_with() diff --git a/api/tests/unit/experimentation/test_views.py b/api/tests/unit/experimentation/test_views.py index 45c4b6cd16ed..751452122d9b 100644 --- a/api/tests/unit/experimentation/test_views.py +++ b/api/tests/unit/experimentation/test_views.py @@ -1238,7 +1238,7 @@ def test_post__clickhouse_minimal_payload__applies_defaults_and_generates_name( assert response.json()["config"] == { "host": "ch.example.com", "port": 9440, - "database": "flagsmith", + "database": "flagsmith_exp", "username": "default", "secure": True, } @@ -1535,7 +1535,7 @@ def test_post__clickhouse_verification_outcome__returns_201_with_status( assert response.json()["status"] == expected_status assert response.json()["status_detail"] == expected_detail assert "credentials" not in response.json() - mock_client.return_value.execute.assert_called_once_with("SELECT 1") + assert mock_client.return_value.execute.call_args_list[0] == mocker.call("SELECT 1") def test_get__clickhouse__credentials_not_in_response( diff --git a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md index cd06797f5cd7..95a42c257520 100644 --- a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md @@ -619,7 +619,7 @@ Attributes: ### `warehouse.connection.connected` Logged at `info` from: - - `api/experimentation/services.py:884` + - `api/experimentation/services.py:903` Attributes: - `environment.id` @@ -637,7 +637,7 @@ Attributes: ### `warehouse.connection.verification_failed` Logged at `warning` from: - - `api/experimentation/services.py:859` + - `api/experimentation/services.py:878` Attributes: - `environment.id` @@ -647,7 +647,7 @@ Attributes: ### `warehouse.connection.verification_succeeded` Logged at `info` from: - - `api/experimentation/services.py:869` + - `api/experimentation/services.py:888` Attributes: - `environment.id` From abfb468b22cec0f630754245e5a0e2db5e9c5b48 Mon Sep 17 00:00:00 2001 From: wadii Date: Tue, 28 Jul 2026 10:13:23 +0200 Subject: [PATCH 2/3] fix: cache customer event stats, throttle warehouse connection reads --- api/app/settings/common.py | 1 + api/app/settings/test.py | 1 + api/experimentation/services.py | 67 ++++++++++++------- api/experimentation/views.py | 3 + .../unit/experimentation/test_services.py | 13 ++++ api/tests/unit/experimentation/test_views.py | 23 ++++--- 6 files changed, 75 insertions(+), 33 deletions(-) diff --git a/api/app/settings/common.py b/api/app/settings/common.py index 146a1535eefb..cb3bcde34998 100644 --- a/api/app/settings/common.py +++ b/api/app/settings/common.py @@ -367,6 +367,7 @@ "user": USER_THROTTLE_RATE, "influx_query": "5/min", "warehouse_connection_write": "10/min", + "warehouse_connection_read": "60/min", }, "DEFAULT_FILTER_BACKENDS": ["django_filters.rest_framework.DjangoFilterBackend"], "DEFAULT_RENDERER_CLASSES": [ diff --git a/api/app/settings/test.py b/api/app/settings/test.py index c6e4954a94be..66078bbd32de 100644 --- a/api/app/settings/test.py +++ b/api/app/settings/test.py @@ -28,6 +28,7 @@ "master_api_key": "100000/day", "influx_query": "50/min", "warehouse_connection_write": "1000/min", + "warehouse_connection_read": "1000/min", } AWS_SSE_LOGS_BUCKET_NAME = "test_bucket" diff --git a/api/experimentation/services.py b/api/experimentation/services.py index 5ee7b33871fd..b5b0d4ada6d6 100644 --- a/api/experimentation/services.py +++ b/api/experimentation/services.py @@ -9,6 +9,7 @@ from clickhouse_driver import errors as clickhouse_errors from clickhouse_driver.util.helpers import parse_url from django.conf import settings +from django.core.cache import cache from django.db import transaction from django.db.models import Q from django.utils import timezone @@ -95,6 +96,9 @@ CLICKHOUSE_CONNECT_TIMEOUT_SECONDS = 5 CLICKHOUSE_QUERY_TIMEOUT_SECONDS = 30 CLICKHOUSE_VERIFY_TIMEOUT_SECONDS = 5 +CUSTOMER_EVENT_STATS_CACHE_SECONDS = 60 + +_CUSTOMER_EVENT_STATS_UNAVAILABLE = "unavailable" def is_warehouse_feature_enabled(organisation: Organisation) -> bool: @@ -140,9 +144,8 @@ def get_unique_event_names(environment_key: str) -> list[str]: return [row[0] for row in rows] -def get_warehouse_event_stats(environment_key: str) -> WarehouseEventStats: - """Return event counts recorded for `environment_key` in the warehouse.""" - rows = _get_clickhouse_client().execute( +def _query_event_stats(client: Client, environment_key: str) -> WarehouseEventStats: + rows = client.execute( "SELECT count() AS total, uniqExact(event) AS unique " "FROM events WHERE environment_key = %(environment_key)s", {"environment_key": environment_key}, @@ -154,6 +157,11 @@ def get_warehouse_event_stats(environment_key: str) -> WarehouseEventStats: ) +def get_warehouse_event_stats(environment_key: str) -> WarehouseEventStats: + """Return event counts recorded for `environment_key` in the warehouse.""" + return _query_event_stats(_get_clickhouse_client(), environment_key) + + EXPOSURE_BUCKETS_QUERY = ( _EXPOSURES_CTE + """ @@ -918,12 +926,9 @@ def annotate_warehouse_event_stats( leaves stats unset when the warehouse is unreachable. Read-only: never changes status.""" if connection.warehouse_type == WarehouseType.CLICKHOUSE: - try: - connection.event_stats = _get_customer_warehouse_event_stats( - connection, environment_key - ) - except Exception: - pass + stats = _get_customer_warehouse_event_stats_cached(connection, environment_key) + if stats is not None: + connection.event_stats = stats return if ( connection.warehouse_type != WarehouseType.FLAGSMITH @@ -936,23 +941,37 @@ def annotate_warehouse_event_stats( return -def _get_customer_warehouse_event_stats( +def _get_customer_warehouse_event_stats_cached( connection: WarehouseConnection, environment_key: str, -) -> WarehouseEventStats: +) -> WarehouseEventStats | None: """Return event counts recorded for `environment_key` in the customer's - ClickHouse instance.""" - client = _connect_customer_clickhouse(connection) + 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}" + cached = cache.get(cache_key) + if isinstance(cached, WarehouseEventStats): + return cached + if cached == _CUSTOMER_EVENT_STATS_UNAVAILABLE: + return None try: - rows = client.execute( - "SELECT count() AS total, uniqExact(event) AS unique " - "FROM events WHERE environment_key = %(environment_key)s", - {"environment_key": environment_key}, + client = _connect_customer_clickhouse(connection) + try: + stats = _query_event_stats(client, environment_key) + finally: + client.disconnect() + except Exception: + cache.set( + cache_key, + _CUSTOMER_EVENT_STATS_UNAVAILABLE, + CUSTOMER_EVENT_STATS_CACHE_SECONDS, ) - finally: - client.disconnect() - total, unique = rows[0] if rows else (0, 0) - return WarehouseEventStats( - total_events_received=int(total), - unique_events_count=int(unique), - ) + logger.warning( + "connection.event_stats_failed", + environment__id=connection.environment_id, + exc_info=True, + ) + return None + cache.set(cache_key, stats, CUSTOMER_EVENT_STATS_CACHE_SECONDS) + return stats diff --git a/api/experimentation/views.py b/api/experimentation/views.py index a97545dd6cc8..785aa8ae367e 100644 --- a/api/experimentation/views.py +++ b/api/experimentation/views.py @@ -106,6 +106,9 @@ def get_throttles(self) -> list[BaseThrottle]: ): self.throttle_scope = "warehouse_connection_write" return [*super().get_throttles(), ScopedRateThrottle()] + if self.action in ("list", "retrieve"): + self.throttle_scope = "warehouse_connection_read" + return [*super().get_throttles(), ScopedRateThrottle()] return super().get_throttles() def perform_create(self, serializer: BaseSerializer[WarehouseConnection]) -> None: diff --git a/api/tests/unit/experimentation/test_services.py b/api/tests/unit/experimentation/test_services.py index 9c970c939af4..a1537f1aed75 100644 --- a/api/tests/unit/experimentation/test_services.py +++ b/api/tests/unit/experimentation/test_services.py @@ -2224,8 +2224,10 @@ def test_describe_verification_error__known_error_types__returns_expected_detail ) def test_annotate_warehouse_event_stats__clickhouse_connection__queries_customer_instance( clickhouse_connection: WarehouseConnection, + reset_cache: None, execute_side_effect: Exception | list[list[tuple[int, int]]], expected_stats: WarehouseEventStats | None, + log: StructuredLogCapture, mocker: MockerFixture, ) -> None: # Given @@ -2243,3 +2245,14 @@ def test_annotate_warehouse_event_stats__clickhouse_connection__queries_customer {"environment_key": "test-env-key"}, ) mock_client.return_value.disconnect.assert_called_once_with() + assert any( + event["event"] == "connection.event_stats_failed" for event in log.events + ) == (expected_stats is None) + + # When — the outcome is cached, so a second request doesn't reconnect + fresh_connection = WarehouseConnection.objects.get(id=clickhouse_connection.id) + annotate_warehouse_event_stats(fresh_connection, "test-env-key") + + # Then + mock_client.assert_called_once() + assert getattr(fresh_connection, "event_stats", None) == expected_stats diff --git a/api/tests/unit/experimentation/test_views.py b/api/tests/unit/experimentation/test_views.py index 751452122d9b..51bc1db7437e 100644 --- a/api/tests/unit/experimentation/test_views.py +++ b/api/tests/unit/experimentation/test_views.py @@ -1090,16 +1090,21 @@ def test_test_warehouse_connection__non_admin__returns_403( @pytest.mark.parametrize( - "action", + "action, expected_scope", [ - "create", - "update", - "partial_update", - "test_warehouse_connection", - "test_warehouse_connection_config", + ("create", "warehouse_connection_write"), + ("update", "warehouse_connection_write"), + ("partial_update", "warehouse_connection_write"), + ("test_warehouse_connection", "warehouse_connection_write"), + ("test_warehouse_connection_config", "warehouse_connection_write"), + ("list", "warehouse_connection_read"), + ("retrieve", "warehouse_connection_read"), ], ) -def test_get_throttles__write_actions__returns_scoped_throttle(action: str) -> None: +def test_get_throttles__throttled_actions__returns_scoped_throttle( + action: str, + expected_scope: str, +) -> None: # Given view = WarehouseConnectionViewSet() view.action = action @@ -1109,12 +1114,12 @@ def test_get_throttles__write_actions__returns_scoped_throttle(action: str) -> N # Then assert any(isinstance(t, ScopedRateThrottle) for t in throttles) - assert view.throttle_scope == "warehouse_connection_write" + assert view.throttle_scope == expected_scope @pytest.mark.parametrize( "action", - ["list", "retrieve", "destroy"], + ["destroy"], ) def test_get_throttles__other_actions__returns_view_default_throttles( action: str, From e96947d5f2e0001751b1c608106ce767a7cb1ac2 Mon Sep 17 00:00:00 2001 From: "flagsmith-engineering[bot]" Date: Tue, 28 Jul 2026 08:31:40 +0000 Subject: [PATCH 3/3] chore: Update documentation artefacts --- .../observability/_events-catalogue.md | 21 +++++++++++++------ 1 file changed, 15 insertions(+), 6 deletions(-) diff --git a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md index 95a42c257520..a0bf94e9a960 100644 --- a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md @@ -619,16 +619,25 @@ Attributes: ### `warehouse.connection.connected` Logged at `info` from: - - `api/experimentation/services.py:903` + - `api/experimentation/services.py:911` Attributes: - `environment.id` - `organisation.id` +### `warehouse.connection.event_stats_failed` + +Logged at `warning` from: + - `api/experimentation/services.py:970` + +Attributes: + - `environment.id` + - `exc_info` + ### `warehouse.connection.test_event_sent` Logged at `info` from: - - `api/experimentation/services.py:790` + - `api/experimentation/services.py:798` Attributes: - `environment.id` @@ -637,7 +646,7 @@ Attributes: ### `warehouse.connection.verification_failed` Logged at `warning` from: - - `api/experimentation/services.py:878` + - `api/experimentation/services.py:886` Attributes: - `environment.id` @@ -647,7 +656,7 @@ Attributes: ### `warehouse.connection.verification_succeeded` Logged at `info` from: - - `api/experimentation/services.py:888` + - `api/experimentation/services.py:896` Attributes: - `environment.id` @@ -656,7 +665,7 @@ Attributes: ### `warehouse.srm.overallocated` Logged at `error` from: - - `api/experimentation/services.py:412` + - `api/experimentation/services.py:420` Attributes: - `environment.id` @@ -666,7 +675,7 @@ Attributes: ### `warehouse.srm.unkeyed_variant` Logged at `error` from: - - `api/experimentation/services.py:398` + - `api/experimentation/services.py:406` Attributes: - `environment.id`