feat: expose distinct warehouse event names per connection

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

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,50 @@ 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 event FROM events "
+    "WHERE environment_key = %(environment_key)s "
+    "GROUP BY event ORDER BY max(timestamp) DESC 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`, 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:
+        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 +1147,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

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)

api/tests/unit/experimentation/test_services.py

@@ -22,6 +22,7 @@
     MetricSpec,
     ResultsAggregates,
     RolloutSpec,
+    WarehouseEventNames,
     WarehouseEventStats,
 )
 from experimentation.models import (
@@ -108,29 +109,137 @@ 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 "
+        "SELECT event FROM events "
         "WHERE environment_key = %(environment_key)s "
-        "ORDER BY event",
-        {"environment_key": "env-key-123"},
+        "GROUP BY event ORDER BY max(timestamp) DESC 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 event FROM events "
+        "WHERE environment_key = %(environment_key)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()
+    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(
     mocker: MockerFixture,
@@ -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",
     [