From 7f471ad52b500ce8425c15a94c819e69499ee63d Mon Sep 17 00:00:00 2001 From: Gagan Trivedi Date: Fri, 18 Sep 2026 13:57:03 +0530 Subject: [PATCH 1/8] feat(experimentation): publish warehouse connections to Redis and apply delivery status The warehouse-delivery service reads each environment's connection from experimentation:environment_warehouses: and leaves each connection's outcome in the experimentation:warehouse_delivery_status hash. sync_environment_ingestion now writes the connection, with the credentials as the same Fernet ciphertext the database holds, before the destination so the service never sees an event it cannot place, and removes it after. A one-minute task copies the outcomes onto WarehouseConnection.status and status_detail. The update hook also fires on config and credentials changes. --- api/core/fields.py | 9 +- api/experimentation/dataclasses.py | 11 ++ api/experimentation/ingestion_sync_service.py | 85 ++++++++- api/experimentation/models.py | 6 +- api/experimentation/tasks.py | 54 +++++- api/tests/unit/core/test_fields.py | 13 +- .../test_ingestion_sync_service.py | 162 ++++++++++++++++ api/tests/unit/experimentation/test_models.py | 9 +- api/tests/unit/experimentation/test_tasks.py | 175 +++++++++++++++++- .../observability/_events-catalogue.md | 33 +++- 10 files changed, 535 insertions(+), 22 deletions(-) diff --git a/api/core/fields.py b/api/core/fields.py index 90051714cfd3..c6e5dbf1b568 100644 --- a/api/core/fields.py +++ b/api/core/fields.py @@ -42,11 +42,18 @@ def _get_fernet() -> Fernet: return Fernet(base64.urlsafe_b64encode(digest)) +def encrypt_json(value: Any) -> str: + """Encrypts a JSON value exactly as ``EncryptedJSONField`` stores it, so the + same ciphertext can be handed to another service that holds + ``WAREHOUSE_CREDENTIALS_SECRET``.""" + return _get_fernet().encrypt(json.dumps(value).encode()).decode() + + class EncryptedJSONField(models.TextField[Any, Any]): def get_prep_value(self, value: Any) -> str | None: if value is None: return None - return _get_fernet().encrypt(json.dumps(value).encode()).decode() + return encrypt_json(value) def from_db_value( self, diff --git a/api/experimentation/dataclasses.py b/api/experimentation/dataclasses.py index 528d0b4e76f8..076ec3782bf7 100644 --- a/api/experimentation/dataclasses.py +++ b/api/experimentation/dataclasses.py @@ -125,3 +125,14 @@ class ResultsSummary: # totals divide: a bucket's own conversions over its own new identities # compares different people and can exceed 100%. exposures_timeseries: ExposuresTimeseries + + +@dataclass(frozen=True) +class WarehouseDeliveryStatus: + """One outcome the warehouse-delivery service left for a connection: + whether its last insert into the customer's warehouse worked and, if not, + the sentence the dashboard should show.""" + + connection_id: int + status: str + detail: str | None diff --git a/api/experimentation/ingestion_sync_service.py b/api/experimentation/ingestion_sync_service.py index 734e4fe8f0bb..69e6e002de67 100644 --- a/api/experimentation/ingestion_sync_service.py +++ b/api/experimentation/ingestion_sync_service.py @@ -1,20 +1,43 @@ from __future__ import annotations +import json from functools import lru_cache -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, cast +import structlog from django.conf import settings from redis.cluster import RedisCluster +from core.fields import encrypt_json +from experimentation.dataclasses import WarehouseDeliveryStatus + if TYPE_CHECKING: from datetime import datetime INGESTION_ENVIRONMENT_KEY_PREFIX = "experimentation:environment_keys:" INGESTION_ENVIRONMENT_DESTINATION_PREFIX = "experimentation:environment_destinations:" - +# Read by the warehouse-delivery service to find the warehouse an environment's +# events go to. The rest of the key is the environment's client API key, the +# same value the ingestion server puts on each Kafka message. +INGESTION_ENVIRONMENT_WAREHOUSE_PREFIX = "experimentation:environment_warehouses:" +# One hash the warehouse-delivery service writes each connection's latest +# outcome into, under the connection id. Emptied by +# apply_warehouse_delivery_statuses once a minute. +WAREHOUSE_DELIVERY_STATUS_KEY = "experimentation:warehouse_delivery_status" + +# Returns every field and value of the hash and deletes it in the same step, +# so an outcome the delivery service writes while we are reading is never +# deleted unread. +_POP_HASH_SCRIPT = """ +local entries = redis.call('HGETALL', KEYS[1]) +redis.call('DEL', KEYS[1]) +return entries +""" SOCKET_TIMEOUT = 1 +logger = structlog.get_logger("experimentation") + @lru_cache(maxsize=1) def _get_client() -> RedisCluster: @@ -52,3 +75,61 @@ def set_ingestion_destination(client_api_key: str, *, topic: str) -> None: def delete_ingestion_destination(client_api_key: str) -> None: redis_key = f"{INGESTION_ENVIRONMENT_DESTINATION_PREFIX}{client_api_key}" _get_client().delete(redis_key) + + +def set_ingestion_warehouse( + client_api_key: str, + *, + connection_id: int, + warehouse_type: str, + config: dict[str, object], + credentials: dict[str, object] | None, +) -> None: + """Publishes the warehouse the delivery service should insert the + environment's events into. Credentials travel as the same Fernet + ciphertext the database holds, so only a service that has + WAREHOUSE_CREDENTIALS_SECRET can read them out of Redis.""" + redis_key = f"{INGESTION_ENVIRONMENT_WAREHOUSE_PREFIX}{client_api_key}" + document = { + "connection_id": connection_id, + "warehouse_type": warehouse_type, + "config": config, + "credentials": encrypt_json(credentials) if credentials is not None else None, + } + _get_client().set(redis_key, json.dumps(document)) + + +def delete_ingestion_warehouse(client_api_key: str) -> None: + redis_key = f"{INGESTION_ENVIRONMENT_WAREHOUSE_PREFIX}{client_api_key}" + _get_client().delete(redis_key) + + +def pop_warehouse_delivery_statuses() -> list[WarehouseDeliveryStatus]: + """Takes every outcome the warehouse-delivery service has left in Redis, + emptying the hash as it goes. An entry that cannot be read is logged and + skipped rather than blocking the others.""" + # The stub types eval for the async client too; this client is synchronous + # and a Lua HGETALL comes back as a flat field, value, field, value list. + entries = cast( + list[bytes], + _get_client().eval(_POP_HASH_SCRIPT, 1, WAREHOUSE_DELIVERY_STATUS_KEY), + ) + statuses: list[WarehouseDeliveryStatus] = [] + for field, value in zip(entries[::2], entries[1::2], strict=True): + try: + outcome = json.loads(value) + detail = outcome.get("detail") + status = WarehouseDeliveryStatus( + connection_id=int(field), + status=str(outcome["status"]), + detail=str(detail) if detail is not None else None, + ) + except (ValueError, KeyError, TypeError, AttributeError): + logger.warning( + "delivery_status.unreadable", + field=field.decode(errors="replace"), + exc_info=True, + ) + continue + statuses.append(status) + return statuses diff --git a/api/experimentation/models.py b/api/experimentation/models.py index 8f352061bdcd..333cbb1f1d59 100644 --- a/api/experimentation/models.py +++ b/api/experimentation/models.py @@ -78,7 +78,11 @@ class Meta: ] @hook(AFTER_CREATE) # type: ignore[misc] - @hook(AFTER_UPDATE, when="warehouse_type", has_changed=True) # type: ignore[misc] + @hook( # type: ignore[misc] + AFTER_UPDATE, + when_any=["warehouse_type", "config", "credentials"], + has_changed=True, + ) @hook(AFTER_DELETE) # type: ignore[misc] def sync_to_ingestion(self) -> None: from experimentation.tasks import sync_environment_ingestion diff --git a/api/experimentation/tasks.py b/api/experimentation/tasks.py index 6875bc16e7aa..eb73764805df 100644 --- a/api/experimentation/tasks.py +++ b/api/experimentation/tasks.py @@ -1,8 +1,12 @@ from datetime import timedelta import structlog +from django.conf import settings from django.utils import timezone -from task_processor.decorators import register_task_handler +from task_processor.decorators import ( + register_recurring_task, + register_task_handler, +) from task_processor.exceptions import TaskBackoffError from environments.models import Environment, EnvironmentAPIKey @@ -12,6 +16,8 @@ Experiment, ExperimentExposures, ExperimentResults, + WarehouseConnection, + WarehouseConnectionStatus, WarehouseType, ) from experimentation.services import ( @@ -47,14 +53,24 @@ def sync_environment_ingestion(environment_id: int) -> None: for api_key in environment.api_keys.all(): ingestion_sync_service.delete_ingestion_key(api_key.key) ingestion_sync_service.delete_ingestion_destination(environment.api_key) + ingestion_sync_service.delete_ingestion_warehouse(environment.api_key) return - # Destination first, then keys. As soon as a key is in Redis the ingestion - # server accepts events for it, and if no destination is stored yet those - # events go to Flagsmith's own topic instead of the external warehouse one. + # Connection details, then destination, then keys. Each step makes the next + # one safe: the warehouse-delivery service drops events for an environment + # whose connection it cannot find in Redis, and the ingestion server sends + # events for an environment with no destination to Flagsmith's own topic. if connection.warehouse_type == WarehouseType.FLAGSMITH: ingestion_sync_service.delete_ingestion_destination(environment.api_key) + ingestion_sync_service.delete_ingestion_warehouse(environment.api_key) else: + ingestion_sync_service.set_ingestion_warehouse( + environment.api_key, + connection_id=connection.id, + warehouse_type=connection.warehouse_type, + config=connection.config or {}, + credentials=connection.credentials, + ) ingestion_sync_service.set_ingestion_destination( environment.api_key, topic=EXTERNAL_WAREHOUSE_EVENTS_TOPIC, @@ -97,6 +113,36 @@ def remove_environment_ingestion_key(key: str) -> None: ingestion_sync_service.delete_ingestion_key(key) +@register_recurring_task(run_every=timedelta(minutes=1), timeout=timedelta(minutes=1)) +def apply_warehouse_delivery_statuses() -> None: + """Copies the outcomes the warehouse-delivery service left in Redis onto + the connections, so the dashboard shows whether a customer's warehouse is + taking their events. That service never writes to Postgres; this task is + the only path from it to the connection row.""" + if not settings.INGESTION_REDIS_URL: + return + for outcome in ingestion_sync_service.pop_warehouse_delivery_statuses(): + if outcome.status not in WarehouseConnectionStatus.values: + logger.warning( + "delivery_status.unknown", + connection__id=outcome.connection_id, + status=outcome.status, + ) + continue + updated = WarehouseConnection.objects.filter(id=outcome.connection_id).update( + status=outcome.status, + status_detail=outcome.detail[:255] if outcome.detail else None, + ) + # A connected outcome arrives for every live connection every minute, + # so only the failures are worth an event. + if updated and outcome.status == WarehouseConnectionStatus.ERRORED: + logger.warning( + "warehouse_connection.delivery_errored", + connection__id=outcome.connection_id, + status__detail=outcome.detail, + ) + + @register_task_handler(timeout=COMPUTE_TASK_TIMEOUT) def compute_experiment_exposures(experiment_id: int) -> None: experiment = ( diff --git a/api/tests/unit/core/test_fields.py b/api/tests/unit/core/test_fields.py index e67cc9a81254..f5e385413796 100644 --- a/api/tests/unit/core/test_fields.py +++ b/api/tests/unit/core/test_fields.py @@ -3,7 +3,7 @@ from pytest_django.fixtures import SettingsWrapper from pytest_structlog import StructuredLogCapture -from core.fields import EncryptedJSONField, NoSSRFURLField +from core.fields import EncryptedJSONField, NoSSRFURLField, encrypt_json from integrations.gitlab.serializers import GitLabConfigurationSerializer @@ -118,3 +118,14 @@ def test_get_lookup__non_isnull__raises_not_implemented() -> None: with pytest.raises(NotImplementedError): field.get_lookup("exact") assert field.get_lookup("isnull") is not None + + +def test_encrypt_json__value__is_what_the_field_reads_back() -> None: + # Given + field = EncryptedJSONField() + + # When + token = encrypt_json({"password": "hunter2"}) + + # Then another holder of the secret, or the field itself, reads the value + assert field.from_db_value(token, None, None) == {"password": "hunter2"} diff --git a/api/tests/unit/experimentation/test_ingestion_sync_service.py b/api/tests/unit/experimentation/test_ingestion_sync_service.py index 7d4b6cd61de8..572cfb77f134 100644 --- a/api/tests/unit/experimentation/test_ingestion_sync_service.py +++ b/api/tests/unit/experimentation/test_ingestion_sync_service.py @@ -1,11 +1,15 @@ +import json from datetime import datetime from datetime import timezone as dt_timezone import pytest from pytest_mock import MockerFixture +from pytest_structlog import StructuredLogCapture from redis.exceptions import RedisError +from core.fields import EncryptedJSONField from experimentation import ingestion_sync_service +from experimentation.dataclasses import WarehouseDeliveryStatus def test_get_client__configured_url__builds_redis_cluster_with_socket_options( @@ -174,3 +178,161 @@ def test_delete_ingestion_key__redis_error__propagates( # When / Then with pytest.raises(RedisError, match="boom"): ingestion_sync_service.delete_ingestion_key("ser.test-key-001") + + +def test_set_ingestion_warehouse__connection_details__writes_document_with_encrypted_credentials( + mocker: MockerFixture, +) -> None: + # Given + mock_client = mocker.Mock() + mocker.patch( + "experimentation.ingestion_sync_service._get_client", + return_value=mock_client, + ) + config = {"host": "ch.acme-corp.example", "port": 8443, "secure": True} + + # When + ingestion_sync_service.set_ingestion_warehouse( + "client-env-key", + connection_id=42, + warehouse_type="clickhouse", + config=config, + credentials={"password": "hunter2"}, + ) + + # Then the document sits under the environment's client key, and the + # password only ever reaches Redis as the ciphertext the database holds + mock_client.set.assert_called_once() + redis_key, raw = mock_client.set.call_args.args + assert redis_key == "experimentation:environment_warehouses:client-env-key" + assert "hunter2" not in raw + document = json.loads(raw) + assert document["connection_id"] == 42 + assert document["warehouse_type"] == "clickhouse" + assert document["config"] == config + assert EncryptedJSONField().from_db_value(document["credentials"], None, None) == { + "password": "hunter2" + } + + +def test_set_ingestion_warehouse__no_credentials__writes_null( + mocker: MockerFixture, +) -> None: + # Given a warehouse type the API stores no credentials for + mock_client = mocker.Mock() + mocker.patch( + "experimentation.ingestion_sync_service._get_client", + return_value=mock_client, + ) + + # When + ingestion_sync_service.set_ingestion_warehouse( + "client-env-key", + connection_id=42, + warehouse_type="snowflake", + config={"account_identifier": "acme"}, + credentials=None, + ) + + # Then + _, raw = mock_client.set.call_args.args + assert json.loads(raw)["credentials"] is None + + +def test_delete_ingestion_warehouse__client_key__deletes_from_redis( + mocker: MockerFixture, +) -> None: + # Given + mock_client = mocker.Mock() + mocker.patch( + "experimentation.ingestion_sync_service._get_client", + return_value=mock_client, + ) + + # When + ingestion_sync_service.delete_ingestion_warehouse("client-env-key") + + # Then + mock_client.delete.assert_called_once_with( + "experimentation:environment_warehouses:client-env-key", + ) + + +def test_pop_warehouse_delivery_statuses__entries_in_hash__read_and_cleared_in_one_step( + mocker: MockerFixture, +) -> None: + # Given two outcomes the delivery service left, as Redis hands them back + mock_client = mocker.Mock() + mock_client.eval.return_value = [ + b"42", + b'{"status": "errored", "detail": "Authentication failed.", "at": 1758000000.0}', + b"7", + b'{"status": "connected", "detail": null, "at": 1758000001.0}', + ] + mocker.patch( + "experimentation.ingestion_sync_service._get_client", + return_value=mock_client, + ) + + # When + statuses = ingestion_sync_service.pop_warehouse_delivery_statuses() + + # Then both are returned, and the hash was read and deleted by one script + # so nothing written in between is lost + assert statuses == [ + WarehouseDeliveryStatus( + connection_id=42, status="errored", detail="Authentication failed." + ), + WarehouseDeliveryStatus(connection_id=7, status="connected", detail=None), + ] + mock_client.eval.assert_called_once_with( + ingestion_sync_service._POP_HASH_SCRIPT, + 1, + "experimentation:warehouse_delivery_status", + ) + + +def test_pop_warehouse_delivery_statuses__empty_hash__returns_nothing( + mocker: MockerFixture, +) -> None: + # Given + mock_client = mocker.Mock() + mock_client.eval.return_value = [] + mocker.patch( + "experimentation.ingestion_sync_service._get_client", + return_value=mock_client, + ) + + # When / Then + assert ingestion_sync_service.pop_warehouse_delivery_statuses() == [] + + +def test_pop_warehouse_delivery_statuses__unreadable_entries__skipped_and_logged( + mocker: MockerFixture, + log: StructuredLogCapture, +) -> None: + # Given a field that is not a connection id, a value that is not JSON, and + # one good entry + mock_client = mocker.Mock() + mock_client.eval.return_value = [ + b"not-an-id", + b'{"status": "errored"}', + b"42", + b"not json", + b"7", + b'{"status": "connected"}', + ] + mocker.patch( + "experimentation.ingestion_sync_service._get_client", + return_value=mock_client, + ) + + # When + statuses = ingestion_sync_service.pop_warehouse_delivery_statuses() + + # Then the good entry still gets through + assert statuses == [ + WarehouseDeliveryStatus(connection_id=7, status="connected", detail=None) + ] + assert log.has("delivery_status.unreadable", level="warning", field="not-an-id") + assert log.has("delivery_status.unreadable", level="warning", field="42") diff --git a/api/tests/unit/experimentation/test_models.py b/api/tests/unit/experimentation/test_models.py index 3ef9016f7b21..01c8895b2832 100644 --- a/api/tests/unit/experimentation/test_models.py +++ b/api/tests/unit/experimentation/test_models.py @@ -70,14 +70,16 @@ def test_warehouse_connection__after_delete__enqueues_ingestion_sync_task( "field, value, expected_enqueued", [ pytest.param("warehouse_type", WarehouseType.CLICKHOUSE, True, id="type"), + pytest.param("config", {"host": "ch.acme-corp.example"}, True, id="config"), + pytest.param("credentials", {"password": "rotated"}, True, id="credentials"), pytest.param("name", "renamed", False, id="name"), ], ) -def test_warehouse_connection__after_update__enqueues_ingestion_sync_task_on_type_change( +def test_warehouse_connection__after_update__enqueues_ingestion_sync_task_on_detail_change( warehouse_connection: WarehouseConnection, mocker: MockerFixture, field: str, - value: str, + value: object, expected_enqueued: bool, ) -> None: # Given @@ -89,7 +91,8 @@ def test_warehouse_connection__after_update__enqueues_ingestion_sync_task_on_typ setattr(warehouse_connection, field, value) warehouse_connection.save() - # Then switching warehouse type re-routes the environment; other edits don't + # Then anything the delivery service reads from Redis republishes the + # connection; a rename does not if expected_enqueued: mock_task.delay.assert_called_once_with( kwargs={"environment_id": warehouse_connection.environment_id}, diff --git a/api/tests/unit/experimentation/test_tasks.py b/api/tests/unit/experimentation/test_tasks.py index d02f5f4937f6..e6c6380e35b6 100644 --- a/api/tests/unit/experimentation/test_tasks.py +++ b/api/tests/unit/experimentation/test_tasks.py @@ -6,6 +6,7 @@ import pytest from django.utils import timezone from freezegun import freeze_time +from pytest_django.fixtures import SettingsWrapper from pytest_mock import MockerFixture from pytest_structlog import StructuredLogCapture from task_processor.exceptions import TaskBackoffError @@ -17,6 +18,7 @@ ExposuresTimeseriesPoint, MetricResult, ResultsSummary, + WarehouseDeliveryStatus, ) from experimentation.models import ( Experiment, @@ -24,10 +26,12 @@ ExperimentResults, ExperimentStatus, WarehouseConnection, + WarehouseConnectionStatus, ) from experimentation.services import CLICKHOUSE_BACKGROUND_QUERY_TIMEOUT_SECONDS from experimentation.stats import VariantStats from experimentation.tasks import ( + apply_warehouse_delivery_statuses, compute_experiment_exposures, compute_experiment_results, remove_environment_ingestion_key, @@ -61,10 +65,12 @@ def test_sync_environment_ingestion__flagsmith_connection__whitelists_valid_keys # When sync_environment_ingestion(environment_id=environment.id) - # Then the environment follows the default pipeline, and only the client - # key and the valid server-side key are whitelisted + # Then the environment follows the default pipeline with no warehouse + # published, and only the client key and the valid server-side key are + # whitelisted assert mock_service.mock_calls == [ mocker.call.delete_ingestion_destination(environment.api_key), + mocker.call.delete_ingestion_warehouse(environment.api_key), mocker.call.set_ingestion_key( environment.api_key, environment_key=environment.api_key, @@ -77,7 +83,7 @@ def test_sync_environment_ingestion__flagsmith_connection__whitelists_valid_keys ] -def test_sync_environment_ingestion__external_connection__routes_before_whitelisting( +def test_sync_environment_ingestion__external_connection__publishes_then_routes_then_whitelists( clickhouse_connection: WarehouseConnection, environment: Environment, mocker: MockerFixture, @@ -88,9 +94,16 @@ def test_sync_environment_ingestion__external_connection__routes_before_whitelis # When sync_environment_ingestion(environment_id=environment.id) - # Then the environment is routed to the topic before its key is whitelisted, - # so no event can reach the default topic in between + # Then the connection is in Redis before events are routed to the topic, and + # the key is whitelisted last, so no event arrives anywhere unplaced assert mock_service.mock_calls == [ + mocker.call.set_ingestion_warehouse( + environment.api_key, + connection_id=clickhouse_connection.id, + warehouse_type="clickhouse", + config=clickhouse_connection.config, + credentials={"password": "hunter2"}, + ), mocker.call.set_ingestion_destination( environment.api_key, topic="external_warehouse_events", @@ -128,6 +141,7 @@ def test_sync_environment_ingestion__connection_deleted__removes_keys_and_destin mocker.call.delete_ingestion_key(active_key.key), mocker.call.delete_ingestion_key(inactive_key.key), mocker.call.delete_ingestion_destination(environment.api_key), + mocker.call.delete_ingestion_warehouse(environment.api_key), ] @@ -143,11 +157,12 @@ def test_sync_environment_ingestion__environment_deleted__removes_keys_and_desti # When sync_environment_ingestion(environment_id=environment.id) - # Then the deleted environment is still found, so its key and destination - # are removed rather than left accepting events + # Then the deleted environment is still found, so its key, destination and + # connection are removed rather than left accepting events assert mock_service.mock_calls == [ mocker.call.delete_ingestion_key(environment.api_key), mocker.call.delete_ingestion_destination(environment.api_key), + mocker.call.delete_ingestion_warehouse(environment.api_key), ] @@ -674,3 +689,149 @@ def test_compute_experiment_results__experiment_deleted_after_enqueue__skips( # Then the task exits without raising into the task processor mock_compute.assert_not_called() + + +def test_apply_warehouse_delivery_statuses__errored_outcome__marks_connection_and_logs( + clickhouse_connection: WarehouseConnection, + mocker: MockerFixture, + settings: SettingsWrapper, + log: StructuredLogCapture, +) -> None: + # Given the delivery service found the customer's warehouse refusing our + # login, and also left an outcome for a connection that no longer exists + settings.INGESTION_REDIS_URL = "redis://ingestion:6379" + mocker.patch( + "experimentation.tasks.ingestion_sync_service.pop_warehouse_delivery_statuses", + return_value=[ + WarehouseDeliveryStatus( + connection_id=clickhouse_connection.id, + status="errored", + detail="Authentication failed.", + ), + WarehouseDeliveryStatus( + connection_id=404404, status="connected", detail=None + ), + ], + ) + + # When + apply_warehouse_delivery_statuses() + + # Then the dashboard shows the failure, and the missing connection is ignored + clickhouse_connection.refresh_from_db() + assert clickhouse_connection.status == WarehouseConnectionStatus.ERRORED + assert clickhouse_connection.status_detail == "Authentication failed." + assert log.events == [ + { + "event": "warehouse_connection.delivery_errored", + "level": "warning", + "connection__id": clickhouse_connection.id, + "status__detail": "Authentication failed.", + } + ] + + +def test_apply_warehouse_delivery_statuses__connected_outcome__clears_detail_quietly( + clickhouse_connection: WarehouseConnection, + mocker: MockerFixture, + settings: SettingsWrapper, + log: StructuredLogCapture, +) -> None: + # Given a connection the dashboard currently shows as errored, whose + # warehouse has started taking events again + settings.INGESTION_REDIS_URL = "redis://ingestion:6379" + clickhouse_connection.status = WarehouseConnectionStatus.ERRORED + clickhouse_connection.status_detail = "Could not connect to the host." + clickhouse_connection.save() + mocker.patch( + "experimentation.tasks.ingestion_sync_service.pop_warehouse_delivery_statuses", + return_value=[ + WarehouseDeliveryStatus( + connection_id=clickhouse_connection.id, status="connected", detail=None + ) + ], + ) + + # When + apply_warehouse_delivery_statuses() + + # Then the connection recovers, and a routine success is not logged + clickhouse_connection.refresh_from_db() + assert clickhouse_connection.status == WarehouseConnectionStatus.CONNECTED + assert clickhouse_connection.status_detail is None + assert log.events == [] + + +def test_apply_warehouse_delivery_statuses__unknown_status__skipped_and_logged( + clickhouse_connection: WarehouseConnection, + mocker: MockerFixture, + settings: SettingsWrapper, + log: StructuredLogCapture, +) -> None: + # Given a status value the connection model has no choice for + settings.INGESTION_REDIS_URL = "redis://ingestion:6379" + mocker.patch( + "experimentation.tasks.ingestion_sync_service.pop_warehouse_delivery_statuses", + return_value=[ + WarehouseDeliveryStatus( + connection_id=clickhouse_connection.id, status="retrying", detail=None + ) + ], + ) + + # When + apply_warehouse_delivery_statuses() + + # Then the connection is left as it was rather than failing the save + clickhouse_connection.refresh_from_db() + assert clickhouse_connection.status == WarehouseConnectionStatus.CREATED + assert log.has( + "delivery_status.unknown", + level="warning", + connection__id=clickhouse_connection.id, + status="retrying", + ) + + +def test_apply_warehouse_delivery_statuses__long_detail__cut_to_the_column_length( + clickhouse_connection: WarehouseConnection, + mocker: MockerFixture, + settings: SettingsWrapper, +) -> None: + # Given a detail longer than status_detail can hold + settings.INGESTION_REDIS_URL = "redis://ingestion:6379" + mocker.patch( + "experimentation.tasks.ingestion_sync_service.pop_warehouse_delivery_statuses", + return_value=[ + WarehouseDeliveryStatus( + connection_id=clickhouse_connection.id, + status="errored", + detail="x" * 300, + ) + ], + ) + + # When + apply_warehouse_delivery_statuses() + + # Then + clickhouse_connection.refresh_from_db() + assert clickhouse_connection.status_detail == "x" * 255 + + +def test_apply_warehouse_delivery_statuses__ingestion_redis_not_configured__does_nothing( + db: None, + mocker: MockerFixture, + settings: SettingsWrapper, +) -> None: + # Given a self-hosted installation with no ingestion Redis + settings.INGESTION_REDIS_URL = "" + mock_pop = mocker.patch( + "experimentation.tasks.ingestion_sync_service.pop_warehouse_delivery_statuses", + ) + + # When + apply_warehouse_delivery_statuses() + + # Then + mock_pop.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 428096849b15..0a3ca4fa4f90 100644 --- a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md @@ -178,7 +178,7 @@ Attributes: ### `core.encrypted_field.decrypt_failed` Logged at `warning` from: - - `api/core/fields.py:62` + - `api/core/fields.py:69` Attributes: - `exc_info` @@ -192,10 +192,28 @@ Attributes: - `environment_api_key` - `environment_id` +### `experimentation.delivery_status.unknown` + +Logged at `warning` from: + - `api/experimentation/tasks.py:126` + +Attributes: + - `connection.id` + - `status` + +### `experimentation.delivery_status.unreadable` + +Logged at `warning` from: + - `api/experimentation/ingestion_sync_service.py:128` + +Attributes: + - `exc_info` + - `field` + ### `experimentation.exposures.compute_failed` Logged at `error` from: - - `api/experimentation/tasks.py:124` + - `api/experimentation/tasks.py:170` Attributes: - `environment.id` @@ -207,7 +225,7 @@ Attributes: ### `experimentation.results.compute_failed` Logged at `error` from: - - `api/experimentation/tasks.py:162` + - `api/experimentation/tasks.py:208` Attributes: - `environment.id` @@ -231,6 +249,15 @@ Attributes: - `feature.id` - `rollout.percentage` +### `experimentation.warehouse_connection.delivery_errored` + +Logged at `warning` from: + - `api/experimentation/tasks.py:139` + +Attributes: + - `connection.id` + - `status.detail` + ### `feature_health.feature_health_event_dismissal_not_supported` Logged at `warning` from: From afe51e611f1516cd8188a2048559254e66599a78 Mon Sep 17 00:00:00 2001 From: Gagan Trivedi Date: Fri, 18 Sep 2026 14:22:51 +0530 Subject: [PATCH 2/8] refactor(core): move warehouse credential encryption out of the field module encrypt_warehouse_credentials and decrypt_warehouse_credentials live in core/warehouse_credentials.py, named for the secret they are bound to. EncryptedJSONField and the ingestion sync service both call them. --- api/core/fields.py | 28 ++++----------- api/core/warehouse_credentials.py | 27 +++++++++++++++ api/experimentation/ingestion_sync_service.py | 6 ++-- api/tests/unit/core/test_fields.py | 13 +------ .../unit/core/test_warehouse_credentials.py | 34 +++++++++++++++++++ .../test_ingestion_sync_service.py | 4 +-- .../observability/_events-catalogue.md | 4 +-- 7 files changed, 77 insertions(+), 39 deletions(-) create mode 100644 api/core/warehouse_credentials.py create mode 100644 api/tests/unit/core/test_warehouse_credentials.py diff --git a/api/core/fields.py b/api/core/fields.py index c6e5dbf1b568..b83125415619 100644 --- a/api/core/fields.py +++ b/api/core/fields.py @@ -1,14 +1,14 @@ -import base64 -import hashlib -import json from typing import Any, TypeVar import structlog -from cryptography.fernet import Fernet, InvalidToken -from django.conf import settings +from cryptography.fernet import InvalidToken from django.db import models from core.validators import validate_http_url_scheme, validate_no_internal_address +from core.warehouse_credentials import ( + decrypt_warehouse_credentials, + encrypt_warehouse_credentials, +) logger = structlog.get_logger("core") @@ -36,24 +36,11 @@ class NoSSRFURLField(models.URLField[_ST, _GT]): ] -def _get_fernet() -> Fernet: - secret: str = settings.WAREHOUSE_CREDENTIALS_SECRET - digest = hashlib.sha256(secret.encode()).digest() - return Fernet(base64.urlsafe_b64encode(digest)) - - -def encrypt_json(value: Any) -> str: - """Encrypts a JSON value exactly as ``EncryptedJSONField`` stores it, so the - same ciphertext can be handed to another service that holds - ``WAREHOUSE_CREDENTIALS_SECRET``.""" - return _get_fernet().encrypt(json.dumps(value).encode()).decode() - - class EncryptedJSONField(models.TextField[Any, Any]): def get_prep_value(self, value: Any) -> str | None: if value is None: return None - return encrypt_json(value) + return encrypt_warehouse_credentials(value) def from_db_value( self, @@ -64,11 +51,10 @@ def from_db_value( if value is None: return None try: - plaintext = _get_fernet().decrypt(value.encode()) + return decrypt_warehouse_credentials(value) except InvalidToken: logger.warning("encrypted_field.decrypt_failed", exc_info=True) return None - return json.loads(plaintext) def get_lookup(self, lookup_name: str) -> Any: if lookup_name != "isnull": diff --git a/api/core/warehouse_credentials.py b/api/core/warehouse_credentials.py new file mode 100644 index 000000000000..ca59d2840ac2 --- /dev/null +++ b/api/core/warehouse_credentials.py @@ -0,0 +1,27 @@ +import base64 +import hashlib +import json +from typing import Any + +from cryptography.fernet import Fernet +from django.conf import settings + + +def _warehouse_credentials_fernet() -> Fernet: + """The cipher for warehouse credentials, keyed on + WAREHOUSE_CREDENTIALS_SECRET. The key is the SHA-256 of the secret, so any + service holding the same secret builds the same cipher and can read what + another encrypted.""" + secret: str = settings.WAREHOUSE_CREDENTIALS_SECRET + digest = hashlib.sha256(secret.encode()).digest() + return Fernet(base64.urlsafe_b64encode(digest)) + + +def encrypt_warehouse_credentials(value: Any) -> str: + return _warehouse_credentials_fernet().encrypt(json.dumps(value).encode()).decode() + + +def decrypt_warehouse_credentials(token: str) -> Any: + """Raises ``cryptography.fernet.InvalidToken`` when the token was made + under a different secret.""" + return json.loads(_warehouse_credentials_fernet().decrypt(token.encode())) diff --git a/api/experimentation/ingestion_sync_service.py b/api/experimentation/ingestion_sync_service.py index 69e6e002de67..1ba2401eed9a 100644 --- a/api/experimentation/ingestion_sync_service.py +++ b/api/experimentation/ingestion_sync_service.py @@ -8,7 +8,7 @@ from django.conf import settings from redis.cluster import RedisCluster -from core.fields import encrypt_json +from core.warehouse_credentials import encrypt_warehouse_credentials from experimentation.dataclasses import WarehouseDeliveryStatus if TYPE_CHECKING: @@ -94,7 +94,9 @@ def set_ingestion_warehouse( "connection_id": connection_id, "warehouse_type": warehouse_type, "config": config, - "credentials": encrypt_json(credentials) if credentials is not None else None, + "credentials": encrypt_warehouse_credentials(credentials) + if credentials is not None + else None, } _get_client().set(redis_key, json.dumps(document)) diff --git a/api/tests/unit/core/test_fields.py b/api/tests/unit/core/test_fields.py index f5e385413796..e67cc9a81254 100644 --- a/api/tests/unit/core/test_fields.py +++ b/api/tests/unit/core/test_fields.py @@ -3,7 +3,7 @@ from pytest_django.fixtures import SettingsWrapper from pytest_structlog import StructuredLogCapture -from core.fields import EncryptedJSONField, NoSSRFURLField, encrypt_json +from core.fields import EncryptedJSONField, NoSSRFURLField from integrations.gitlab.serializers import GitLabConfigurationSerializer @@ -118,14 +118,3 @@ def test_get_lookup__non_isnull__raises_not_implemented() -> None: with pytest.raises(NotImplementedError): field.get_lookup("exact") assert field.get_lookup("isnull") is not None - - -def test_encrypt_json__value__is_what_the_field_reads_back() -> None: - # Given - field = EncryptedJSONField() - - # When - token = encrypt_json({"password": "hunter2"}) - - # Then another holder of the secret, or the field itself, reads the value - assert field.from_db_value(token, None, None) == {"password": "hunter2"} diff --git a/api/tests/unit/core/test_warehouse_credentials.py b/api/tests/unit/core/test_warehouse_credentials.py new file mode 100644 index 000000000000..a5190b253ea9 --- /dev/null +++ b/api/tests/unit/core/test_warehouse_credentials.py @@ -0,0 +1,34 @@ +import pytest +from cryptography.fernet import InvalidToken +from pytest_django.fixtures import SettingsWrapper + +from core.warehouse_credentials import ( + decrypt_warehouse_credentials, + encrypt_warehouse_credentials, +) + + +def test_encrypt_warehouse_credentials__value__comes_back_intact_under_the_same_secret() -> ( + None +): + # Given / When + token = encrypt_warehouse_credentials({"password": "hunter2"}) + + # Then the value is unreadable as stored and decrypts to what went in + assert "hunter2" not in token + assert decrypt_warehouse_credentials(token) == {"password": "hunter2"} + + +def test_decrypt_warehouse_credentials__token_from_another_secret__raises_invalid_token( + settings: SettingsWrapper, +) -> None: + # Given a token made under one secret + settings.WAREHOUSE_CREDENTIALS_SECRET = "old-secret" + token = encrypt_warehouse_credentials({"password": "hunter2"}) + + # When the secret has changed + settings.WAREHOUSE_CREDENTIALS_SECRET = "new-secret" + + # Then the caller learns the token is unreadable rather than getting garbage + with pytest.raises(InvalidToken): + decrypt_warehouse_credentials(token) diff --git a/api/tests/unit/experimentation/test_ingestion_sync_service.py b/api/tests/unit/experimentation/test_ingestion_sync_service.py index 572cfb77f134..154f51152df9 100644 --- a/api/tests/unit/experimentation/test_ingestion_sync_service.py +++ b/api/tests/unit/experimentation/test_ingestion_sync_service.py @@ -7,7 +7,7 @@ from pytest_structlog import StructuredLogCapture from redis.exceptions import RedisError -from core.fields import EncryptedJSONField +from core.warehouse_credentials import decrypt_warehouse_credentials from experimentation import ingestion_sync_service from experimentation.dataclasses import WarehouseDeliveryStatus @@ -210,7 +210,7 @@ def test_set_ingestion_warehouse__connection_details__writes_document_with_encry assert document["connection_id"] == 42 assert document["warehouse_type"] == "clickhouse" assert document["config"] == config - assert EncryptedJSONField().from_db_value(document["credentials"], None, None) == { + assert decrypt_warehouse_credentials(document["credentials"]) == { "password": "hunter2" } diff --git a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md index 0a3ca4fa4f90..4d9178e285e7 100644 --- a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md @@ -178,7 +178,7 @@ Attributes: ### `core.encrypted_field.decrypt_failed` Logged at `warning` from: - - `api/core/fields.py:69` + - `api/core/fields.py:56` Attributes: - `exc_info` @@ -204,7 +204,7 @@ Attributes: ### `experimentation.delivery_status.unreadable` Logged at `warning` from: - - `api/experimentation/ingestion_sync_service.py:128` + - `api/experimentation/ingestion_sync_service.py:130` Attributes: - `exc_info` From 82a2525c108c3e598c8125babdc66555cb24a424 Mon Sep 17 00:00:00 2001 From: Gagan Trivedi Date: Fri, 18 Sep 2026 14:30:41 +0530 Subject: [PATCH 3/8] refactor(experimentation): move EncryptedJSONField and credential encryption into the app Both are only about a warehouse connection's credentials. core/fields.py keeps NoSSRFURLField alone. Migration 0010 now imports the field from its new module, so the migration state matches the model and no new migration is needed. --- api/core/fields.py | 38 +----------- api/experimentation/fields.py | 45 ++++++++++++++ api/experimentation/ingestion_sync_service.py | 2 +- ...onnection_credentials_and_status_detail.py | 4 +- api/experimentation/models.py | 2 +- .../warehouse_credentials.py | 0 api/tests/unit/core/test_fields.py | 58 +----------------- api/tests/unit/experimentation/test_fields.py | 59 +++++++++++++++++++ .../test_ingestion_sync_service.py | 2 +- .../test_warehouse_credentials.py | 2 +- .../observability/_events-catalogue.md | 16 ++--- 11 files changed, 120 insertions(+), 108 deletions(-) create mode 100644 api/experimentation/fields.py rename api/{core => experimentation}/warehouse_credentials.py (100%) create mode 100644 api/tests/unit/experimentation/test_fields.py rename api/tests/unit/{core => experimentation}/test_warehouse_credentials.py (95%) diff --git a/api/core/fields.py b/api/core/fields.py index b83125415619..4206c6f3060a 100644 --- a/api/core/fields.py +++ b/api/core/fields.py @@ -1,16 +1,8 @@ -from typing import Any, TypeVar +from typing import TypeVar -import structlog -from cryptography.fernet import InvalidToken from django.db import models from core.validators import validate_http_url_scheme, validate_no_internal_address -from core.warehouse_credentials import ( - decrypt_warehouse_credentials, - encrypt_warehouse_credentials, -) - -logger = structlog.get_logger("core") _ST = TypeVar("_ST") _GT = TypeVar("_GT") @@ -34,31 +26,3 @@ class NoSSRFURLField(models.URLField[_ST, _GT]): validate_http_url_scheme, validate_no_internal_address, ] - - -class EncryptedJSONField(models.TextField[Any, Any]): - def get_prep_value(self, value: Any) -> str | None: - if value is None: - return None - return encrypt_warehouse_credentials(value) - - def from_db_value( - self, - value: str | None, - expression: object, - connection: object, - ) -> Any: - if value is None: - return None - try: - return decrypt_warehouse_credentials(value) - except InvalidToken: - logger.warning("encrypted_field.decrypt_failed", exc_info=True) - return None - - def get_lookup(self, lookup_name: str) -> Any: - if lookup_name != "isnull": - raise NotImplementedError( - "EncryptedJSONField only supports isnull lookups." - ) - return super().get_lookup(lookup_name) diff --git a/api/experimentation/fields.py b/api/experimentation/fields.py new file mode 100644 index 000000000000..6e77de5e16f6 --- /dev/null +++ b/api/experimentation/fields.py @@ -0,0 +1,45 @@ +from typing import Any + +import structlog +from cryptography.fernet import InvalidToken +from django.db import models + +from experimentation.warehouse_credentials import ( + decrypt_warehouse_credentials, + encrypt_warehouse_credentials, +) + +logger = structlog.get_logger("experimentation") + + +class EncryptedJSONField(models.TextField[Any, Any]): + """Stores a JSON value as Fernet ciphertext keyed on + WAREHOUSE_CREDENTIALS_SECRET. Used for a warehouse connection's + credentials, so the same ciphertext can be handed to the warehouse-delivery + service.""" + + def get_prep_value(self, value: Any) -> str | None: + if value is None: + return None + return encrypt_warehouse_credentials(value) + + def from_db_value( + self, + value: str | None, + expression: object, + connection: object, + ) -> Any: + if value is None: + return None + try: + return decrypt_warehouse_credentials(value) + except InvalidToken: + logger.warning("encrypted_field.decrypt_failed", exc_info=True) + return None + + def get_lookup(self, lookup_name: str) -> Any: + if lookup_name != "isnull": + raise NotImplementedError( + "EncryptedJSONField only supports isnull lookups." + ) + return super().get_lookup(lookup_name) diff --git a/api/experimentation/ingestion_sync_service.py b/api/experimentation/ingestion_sync_service.py index 1ba2401eed9a..e563cf56f0a5 100644 --- a/api/experimentation/ingestion_sync_service.py +++ b/api/experimentation/ingestion_sync_service.py @@ -8,8 +8,8 @@ from django.conf import settings from redis.cluster import RedisCluster -from core.warehouse_credentials import encrypt_warehouse_credentials from experimentation.dataclasses import WarehouseDeliveryStatus +from experimentation.warehouse_credentials import encrypt_warehouse_credentials if TYPE_CHECKING: from datetime import datetime diff --git a/api/experimentation/migrations/0010_warehouse_connection_credentials_and_status_detail.py b/api/experimentation/migrations/0010_warehouse_connection_credentials_and_status_detail.py index 0f117529259a..df68ebe98185 100644 --- a/api/experimentation/migrations/0010_warehouse_connection_credentials_and_status_detail.py +++ b/api/experimentation/migrations/0010_warehouse_connection_credentials_and_status_detail.py @@ -1,6 +1,6 @@ from django.db import migrations, models -import core.fields +import experimentation.fields class Migration(migrations.Migration): @@ -12,7 +12,7 @@ class Migration(migrations.Migration): migrations.AddField( model_name="warehouseconnection", name="credentials", - field=core.fields.EncryptedJSONField(blank=True, null=True), + field=experimentation.fields.EncryptedJSONField(blank=True, null=True), ), migrations.AddField( model_name="warehouseconnection", diff --git a/api/experimentation/models.py b/api/experimentation/models.py index 333cbb1f1d59..a9e393f13b08 100644 --- a/api/experimentation/models.py +++ b/api/experimentation/models.py @@ -13,7 +13,6 @@ hook, ) -from core.fields import EncryptedJSONField from core.models import SoftDeleteExportableModel from environments.models import Environment from experimentation.dataclasses import ( @@ -21,6 +20,7 @@ ResultsSummary, WarehouseEventStats, ) +from experimentation.fields import EncryptedJSONField from experimentation.types import MetricDefinition # A computation's payload is the serialised form of its summary dataclass; the diff --git a/api/core/warehouse_credentials.py b/api/experimentation/warehouse_credentials.py similarity index 100% rename from api/core/warehouse_credentials.py rename to api/experimentation/warehouse_credentials.py diff --git a/api/tests/unit/core/test_fields.py b/api/tests/unit/core/test_fields.py index e67cc9a81254..ce9f5ff46e27 100644 --- a/api/tests/unit/core/test_fields.py +++ b/api/tests/unit/core/test_fields.py @@ -1,9 +1,7 @@ import pytest from django.core.exceptions import ValidationError -from pytest_django.fixtures import SettingsWrapper -from pytest_structlog import StructuredLogCapture -from core.fields import EncryptedJSONField, NoSSRFURLField +from core.fields import NoSSRFURLField from integrations.gitlab.serializers import GitLabConfigurationSerializer @@ -64,57 +62,3 @@ def test_no_ssrf_url_field__model_serializer__rejects_non_http_scheme() -> None: # Then assert is_valid is False - - -def test_get_prep_value__json_value__returns_ciphertext_that_roundtrips() -> None: - # Given - field = EncryptedJSONField() - value = {"password": "hunter2"} - - # When - stored = field.get_prep_value(value) - - # Then - assert stored is not None - assert "hunter2" not in stored - assert field.from_db_value(stored, None, None) == value - - -def test_field_methods__none__returns_none() -> None: - # Given - field = EncryptedJSONField() - - # When & Then - assert field.get_prep_value(None) is None - assert field.from_db_value(None, None, None) is None - - -def test_from_db_value__secret_key_changed__returns_none_and_logs( - settings: SettingsWrapper, - log: StructuredLogCapture, -) -> None: - # Given - settings.WAREHOUSE_CREDENTIALS_SECRET = "old-secret" - field = EncryptedJSONField() - stored = field.get_prep_value({"password": "hunter2"}) - settings.WAREHOUSE_CREDENTIALS_SECRET = "new-secret" - - # When - value = field.from_db_value(stored, None, None) - - # Then - assert value is None - assert { - "level": "warning", - "event": "encrypted_field.decrypt_failed", - } in [{"level": e["level"], "event": e["event"]} for e in log.events] - - -def test_get_lookup__non_isnull__raises_not_implemented() -> None: - # Given - field = EncryptedJSONField() - - # When & Then - with pytest.raises(NotImplementedError): - field.get_lookup("exact") - assert field.get_lookup("isnull") is not None diff --git a/api/tests/unit/experimentation/test_fields.py b/api/tests/unit/experimentation/test_fields.py new file mode 100644 index 000000000000..2e0c0fda5106 --- /dev/null +++ b/api/tests/unit/experimentation/test_fields.py @@ -0,0 +1,59 @@ +import pytest +from pytest_django.fixtures import SettingsWrapper +from pytest_structlog import StructuredLogCapture + +from experimentation.fields import EncryptedJSONField + + +def test_get_prep_value__json_value__returns_ciphertext_that_roundtrips() -> None: + # Given + field = EncryptedJSONField() + value = {"password": "hunter2"} + + # When + stored = field.get_prep_value(value) + + # Then + assert stored is not None + assert "hunter2" not in stored + assert field.from_db_value(stored, None, None) == value + + +def test_field_methods__none__returns_none() -> None: + # Given + field = EncryptedJSONField() + + # When & Then + assert field.get_prep_value(None) is None + assert field.from_db_value(None, None, None) is None + + +def test_from_db_value__secret_key_changed__returns_none_and_logs( + settings: SettingsWrapper, + log: StructuredLogCapture, +) -> None: + # Given + settings.WAREHOUSE_CREDENTIALS_SECRET = "old-secret" + field = EncryptedJSONField() + stored = field.get_prep_value({"password": "hunter2"}) + settings.WAREHOUSE_CREDENTIALS_SECRET = "new-secret" + + # When + value = field.from_db_value(stored, None, None) + + # Then + assert value is None + assert { + "level": "warning", + "event": "encrypted_field.decrypt_failed", + } in [{"level": e["level"], "event": e["event"]} for e in log.events] + + +def test_get_lookup__non_isnull__raises_not_implemented() -> None: + # Given + field = EncryptedJSONField() + + # When & Then + with pytest.raises(NotImplementedError): + field.get_lookup("exact") + assert field.get_lookup("isnull") is not None diff --git a/api/tests/unit/experimentation/test_ingestion_sync_service.py b/api/tests/unit/experimentation/test_ingestion_sync_service.py index 154f51152df9..423b39a2ccd8 100644 --- a/api/tests/unit/experimentation/test_ingestion_sync_service.py +++ b/api/tests/unit/experimentation/test_ingestion_sync_service.py @@ -7,9 +7,9 @@ from pytest_structlog import StructuredLogCapture from redis.exceptions import RedisError -from core.warehouse_credentials import decrypt_warehouse_credentials from experimentation import ingestion_sync_service from experimentation.dataclasses import WarehouseDeliveryStatus +from experimentation.warehouse_credentials import decrypt_warehouse_credentials def test_get_client__configured_url__builds_redis_cluster_with_socket_options( diff --git a/api/tests/unit/core/test_warehouse_credentials.py b/api/tests/unit/experimentation/test_warehouse_credentials.py similarity index 95% rename from api/tests/unit/core/test_warehouse_credentials.py rename to api/tests/unit/experimentation/test_warehouse_credentials.py index a5190b253ea9..e7a17dd2ef2c 100644 --- a/api/tests/unit/core/test_warehouse_credentials.py +++ b/api/tests/unit/experimentation/test_warehouse_credentials.py @@ -2,7 +2,7 @@ from cryptography.fernet import InvalidToken from pytest_django.fixtures import SettingsWrapper -from core.warehouse_credentials import ( +from experimentation.warehouse_credentials import ( decrypt_warehouse_credentials, encrypt_warehouse_credentials, ) diff --git a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md index 4d9178e285e7..b21304065427 100644 --- a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md @@ -175,14 +175,6 @@ Attributes: - `error.message` - `source` -### `core.encrypted_field.decrypt_failed` - -Logged at `warning` from: - - `api/core/fields.py:56` - -Attributes: - - `exc_info` - ### `dynamodb.environment_document_compressed` Logged at `info` from: @@ -210,6 +202,14 @@ Attributes: - `exc_info` - `field` +### `experimentation.encrypted_field.decrypt_failed` + +Logged at `warning` from: + - `api/experimentation/fields.py:37` + +Attributes: + - `exc_info` + ### `experimentation.exposures.compute_failed` Logged at `error` from: From f4a49ccb9b739502bf10bc4eb6a9b22a3c23f093 Mon Sep 17 00:00:00 2001 From: Gagan Trivedi Date: Fri, 18 Sep 2026 15:31:31 +0530 Subject: [PATCH 4/8] docs(experimentation): drop the WarehouseDeliveryStatus docstring, the fields say it --- api/experimentation/dataclasses.py | 4 ---- 1 file changed, 4 deletions(-) diff --git a/api/experimentation/dataclasses.py b/api/experimentation/dataclasses.py index 076ec3782bf7..2d52c54f358f 100644 --- a/api/experimentation/dataclasses.py +++ b/api/experimentation/dataclasses.py @@ -129,10 +129,6 @@ class ResultsSummary: @dataclass(frozen=True) class WarehouseDeliveryStatus: - """One outcome the warehouse-delivery service left for a connection: - whether its last insert into the customer's warehouse worked and, if not, - the sentence the dashboard should show.""" - connection_id: int status: str detail: str | None From 86786d2930322a8830f00f970462c12f7dbe732f Mon Sep 17 00:00:00 2001 From: Gagan Trivedi Date: Fri, 18 Sep 2026 15:45:46 +0530 Subject: [PATCH 5/8] refactor(experimentation): split the warehouse-delivery Redis contract from the ingestion one warehouse_delivery_sync_service.py holds what the warehouse-delivery service reads and writes: publish_warehouse_connection, remove_warehouse_connection, pop_warehouse_delivery_statuses. ingestion_sync_service.py is back to keys and destinations for the ingestion server. Both use the client in ingestion_redis.py. The old warehouse_delivery_service.py, which verifies a connection when it is saved, is renamed warehouse_verification_service.py. --- api/experimentation/ingestion_redis.py | 17 ++ api/experimentation/ingestion_sync_service.py | 109 +--------- api/experimentation/services.py | 14 +- api/experimentation/tasks.py | 10 +- .../warehouse_delivery_sync_service.py | 90 ++++++++ ...e.py => warehouse_verification_service.py} | 0 api/tests/unit/experimentation/conftest.py | 6 +- .../experimentation/test_ingestion_redis.py | 28 +++ .../test_ingestion_sync_service.py | 198 +----------------- .../unit/experimentation/test_services.py | 12 +- api/tests/unit/experimentation/test_tasks.py | 85 +++++--- api/tests/unit/experimentation/test_views.py | 14 +- .../test_warehouse_delivery_sync_service.py | 166 +++++++++++++++ ...=> test_warehouse_verification_service.py} | 34 +-- .../observability/_events-catalogue.md | 2 +- 15 files changed, 420 insertions(+), 365 deletions(-) create mode 100644 api/experimentation/ingestion_redis.py create mode 100644 api/experimentation/warehouse_delivery_sync_service.py rename api/experimentation/{warehouse_delivery_service.py => warehouse_verification_service.py} (100%) create mode 100644 api/tests/unit/experimentation/test_ingestion_redis.py create mode 100644 api/tests/unit/experimentation/test_warehouse_delivery_sync_service.py rename api/tests/unit/experimentation/{test_warehouse_delivery_service.py => test_warehouse_verification_service.py} (83%) diff --git a/api/experimentation/ingestion_redis.py b/api/experimentation/ingestion_redis.py new file mode 100644 index 000000000000..21d885d9a099 --- /dev/null +++ b/api/experimentation/ingestion_redis.py @@ -0,0 +1,17 @@ +from functools import lru_cache + +from django.conf import settings +from redis.cluster import RedisCluster + +SOCKET_TIMEOUT = 1 + + +@lru_cache(maxsize=1) +def get_client() -> RedisCluster: + """The Redis the ingestion server and the warehouse-delivery service read + their configuration from and, for the latter, write outcomes to.""" + return RedisCluster.from_url( # type: ignore[no-any-return] + settings.INGESTION_REDIS_URL, + socket_timeout=SOCKET_TIMEOUT, + socket_keepalive=True, + ) diff --git a/api/experimentation/ingestion_sync_service.py b/api/experimentation/ingestion_sync_service.py index e563cf56f0a5..c8a9f85ec746 100644 --- a/api/experimentation/ingestion_sync_service.py +++ b/api/experimentation/ingestion_sync_service.py @@ -1,51 +1,14 @@ from __future__ import annotations -import json -from functools import lru_cache -from typing import TYPE_CHECKING, cast +from typing import TYPE_CHECKING -import structlog -from django.conf import settings -from redis.cluster import RedisCluster - -from experimentation.dataclasses import WarehouseDeliveryStatus -from experimentation.warehouse_credentials import encrypt_warehouse_credentials +from experimentation.ingestion_redis import get_client if TYPE_CHECKING: from datetime import datetime INGESTION_ENVIRONMENT_KEY_PREFIX = "experimentation:environment_keys:" INGESTION_ENVIRONMENT_DESTINATION_PREFIX = "experimentation:environment_destinations:" -# Read by the warehouse-delivery service to find the warehouse an environment's -# events go to. The rest of the key is the environment's client API key, the -# same value the ingestion server puts on each Kafka message. -INGESTION_ENVIRONMENT_WAREHOUSE_PREFIX = "experimentation:environment_warehouses:" -# One hash the warehouse-delivery service writes each connection's latest -# outcome into, under the connection id. Emptied by -# apply_warehouse_delivery_statuses once a minute. -WAREHOUSE_DELIVERY_STATUS_KEY = "experimentation:warehouse_delivery_status" - -# Returns every field and value of the hash and deletes it in the same step, -# so an outcome the delivery service writes while we are reading is never -# deleted unread. -_POP_HASH_SCRIPT = """ -local entries = redis.call('HGETALL', KEYS[1]) -redis.call('DEL', KEYS[1]) -return entries -""" - -SOCKET_TIMEOUT = 1 - -logger = structlog.get_logger("experimentation") - - -@lru_cache(maxsize=1) -def _get_client() -> RedisCluster: - return RedisCluster.from_url( # type: ignore[no-any-return] - settings.INGESTION_REDIS_URL, - socket_timeout=SOCKET_TIMEOUT, - socket_keepalive=True, - ) def set_ingestion_key( @@ -55,7 +18,7 @@ def set_ingestion_key( expires_at: datetime | None = None, ) -> None: redis_key = f"{INGESTION_ENVIRONMENT_KEY_PREFIX}{key}" - _get_client().set( + get_client().set( redis_key, environment_key, exat=int(expires_at.timestamp()) if expires_at is not None else None, @@ -64,74 +27,14 @@ def set_ingestion_key( def delete_ingestion_key(key: str) -> None: redis_key = f"{INGESTION_ENVIRONMENT_KEY_PREFIX}{key}" - _get_client().delete(redis_key) + get_client().delete(redis_key) def set_ingestion_destination(client_api_key: str, *, topic: str) -> None: redis_key = f"{INGESTION_ENVIRONMENT_DESTINATION_PREFIX}{client_api_key}" - _get_client().set(redis_key, topic) + get_client().set(redis_key, topic) def delete_ingestion_destination(client_api_key: str) -> None: redis_key = f"{INGESTION_ENVIRONMENT_DESTINATION_PREFIX}{client_api_key}" - _get_client().delete(redis_key) - - -def set_ingestion_warehouse( - client_api_key: str, - *, - connection_id: int, - warehouse_type: str, - config: dict[str, object], - credentials: dict[str, object] | None, -) -> None: - """Publishes the warehouse the delivery service should insert the - environment's events into. Credentials travel as the same Fernet - ciphertext the database holds, so only a service that has - WAREHOUSE_CREDENTIALS_SECRET can read them out of Redis.""" - redis_key = f"{INGESTION_ENVIRONMENT_WAREHOUSE_PREFIX}{client_api_key}" - document = { - "connection_id": connection_id, - "warehouse_type": warehouse_type, - "config": config, - "credentials": encrypt_warehouse_credentials(credentials) - if credentials is not None - else None, - } - _get_client().set(redis_key, json.dumps(document)) - - -def delete_ingestion_warehouse(client_api_key: str) -> None: - redis_key = f"{INGESTION_ENVIRONMENT_WAREHOUSE_PREFIX}{client_api_key}" - _get_client().delete(redis_key) - - -def pop_warehouse_delivery_statuses() -> list[WarehouseDeliveryStatus]: - """Takes every outcome the warehouse-delivery service has left in Redis, - emptying the hash as it goes. An entry that cannot be read is logged and - skipped rather than blocking the others.""" - # The stub types eval for the async client too; this client is synchronous - # and a Lua HGETALL comes back as a flat field, value, field, value list. - entries = cast( - list[bytes], - _get_client().eval(_POP_HASH_SCRIPT, 1, WAREHOUSE_DELIVERY_STATUS_KEY), - ) - statuses: list[WarehouseDeliveryStatus] = [] - for field, value in zip(entries[::2], entries[1::2], strict=True): - try: - outcome = json.loads(value) - detail = outcome.get("detail") - status = WarehouseDeliveryStatus( - connection_id=int(field), - status=str(outcome["status"]), - detail=str(detail) if detail is not None else None, - ) - except (ValueError, KeyError, TypeError, AttributeError): - logger.warning( - "delivery_status.unreadable", - field=field.decode(errors="replace"), - exc_info=True, - ) - continue - statuses.append(status) - return statuses + get_client().delete(redis_key) diff --git a/api/experimentation/services.py b/api/experimentation/services.py index 3476227826df..a7ef2d9b6502 100644 --- a/api/experimentation/services.py +++ b/api/experimentation/services.py @@ -23,7 +23,7 @@ from cohorts.models import Cohort from core.dataclasses import AuthorData from environments.tasks import rebuild_environment_document -from experimentation import warehouse_delivery_service +from experimentation import warehouse_verification_service from experimentation.constants import ( CONTROL_VARIANT_KEY, EXPERIMENT_FLAG, @@ -1402,15 +1402,15 @@ def verify_clickhouse_connection( log = logger.bind(environment__id=connection.environment_id) try: log = log.bind(organisation__id=connection.environment.project.organisation_id) - with warehouse_delivery_service.delivery_client( + with warehouse_verification_service.delivery_client( connection, send_receive_timeout=CLICKHOUSE_VERIFY_TIMEOUT_SECONDS, ) as client: - warehouse_delivery_service.check_events_table_exists(client) + warehouse_verification_service.check_events_table_exists(client) except Exception as error: connection.status = WarehouseConnectionStatus.ERRORED - connection.status_detail = warehouse_delivery_service.describe_warehouse_error( - error + connection.status_detail = ( + warehouse_verification_service.describe_warehouse_error(error) ) if persist: connection.save(update_fields=["status", "status_detail"]) @@ -1490,7 +1490,7 @@ def _get_customer_warehouse_event_stats_cached( if cached == _CUSTOMER_EVENT_UNAVAILABLE: return None try: - with warehouse_delivery_service.delivery_client( + with warehouse_verification_service.delivery_client( connection, send_receive_timeout=CLICKHOUSE_VERIFY_TIMEOUT_SECONDS, ) as client: @@ -1528,7 +1528,7 @@ def _get_customer_clickhouse_event_names( if cached == _CUSTOMER_EVENT_UNAVAILABLE: return None try: - with warehouse_delivery_service.delivery_client( + with warehouse_verification_service.delivery_client( connection, send_receive_timeout=CLICKHOUSE_EVENT_NAMES_TIMEOUT_SECONDS, ) as client: diff --git a/api/experimentation/tasks.py b/api/experimentation/tasks.py index eb73764805df..c8eefcecf0da 100644 --- a/api/experimentation/tasks.py +++ b/api/experimentation/tasks.py @@ -10,7 +10,7 @@ from task_processor.exceptions import TaskBackoffError from environments.models import Environment, EnvironmentAPIKey -from experimentation import ingestion_sync_service +from experimentation import ingestion_sync_service, warehouse_delivery_sync_service from experimentation.constants import EXTERNAL_WAREHOUSE_EVENTS_TOPIC from experimentation.models import ( Experiment, @@ -53,7 +53,7 @@ def sync_environment_ingestion(environment_id: int) -> None: for api_key in environment.api_keys.all(): ingestion_sync_service.delete_ingestion_key(api_key.key) ingestion_sync_service.delete_ingestion_destination(environment.api_key) - ingestion_sync_service.delete_ingestion_warehouse(environment.api_key) + warehouse_delivery_sync_service.remove_warehouse_connection(environment.api_key) return # Connection details, then destination, then keys. Each step makes the next @@ -62,9 +62,9 @@ def sync_environment_ingestion(environment_id: int) -> None: # events for an environment with no destination to Flagsmith's own topic. if connection.warehouse_type == WarehouseType.FLAGSMITH: ingestion_sync_service.delete_ingestion_destination(environment.api_key) - ingestion_sync_service.delete_ingestion_warehouse(environment.api_key) + warehouse_delivery_sync_service.remove_warehouse_connection(environment.api_key) else: - ingestion_sync_service.set_ingestion_warehouse( + warehouse_delivery_sync_service.publish_warehouse_connection( environment.api_key, connection_id=connection.id, warehouse_type=connection.warehouse_type, @@ -121,7 +121,7 @@ def apply_warehouse_delivery_statuses() -> None: the only path from it to the connection row.""" if not settings.INGESTION_REDIS_URL: return - for outcome in ingestion_sync_service.pop_warehouse_delivery_statuses(): + for outcome in warehouse_delivery_sync_service.pop_warehouse_delivery_statuses(): if outcome.status not in WarehouseConnectionStatus.values: logger.warning( "delivery_status.unknown", diff --git a/api/experimentation/warehouse_delivery_sync_service.py b/api/experimentation/warehouse_delivery_sync_service.py new file mode 100644 index 000000000000..ec42fb54858a --- /dev/null +++ b/api/experimentation/warehouse_delivery_sync_service.py @@ -0,0 +1,90 @@ +import json +from typing import cast + +import structlog + +from experimentation.dataclasses import WarehouseDeliveryStatus +from experimentation.ingestion_redis import get_client +from experimentation.warehouse_credentials import encrypt_warehouse_credentials + +# Read by the warehouse-delivery service to find the warehouse an environment's +# events go to. The rest of the key is the environment's client API key, the +# same value the ingestion server puts on each Kafka message. +WAREHOUSE_CONNECTION_KEY_PREFIX = "experimentation:environment_warehouses:" +# One hash the warehouse-delivery service writes each connection's latest +# outcome into, under the connection id. Emptied by +# apply_warehouse_delivery_statuses once a minute. +WAREHOUSE_DELIVERY_STATUS_KEY = "experimentation:warehouse_delivery_status" + +# Returns every field and value of the hash and deletes it in the same step, +# so an outcome the delivery service writes while we are reading is never +# deleted unread. +_POP_HASH_SCRIPT = """ +local entries = redis.call('HGETALL', KEYS[1]) +redis.call('DEL', KEYS[1]) +return entries +""" + +logger = structlog.get_logger("experimentation") + + +def publish_warehouse_connection( + client_api_key: str, + *, + connection_id: int, + warehouse_type: str, + config: dict[str, object], + credentials: dict[str, object] | None, +) -> None: + """Tells the warehouse-delivery service which warehouse the environment's + events go to. Credentials travel as the same Fernet ciphertext the database + holds, so only a service that has WAREHOUSE_CREDENTIALS_SECRET can read + them out of Redis.""" + redis_key = f"{WAREHOUSE_CONNECTION_KEY_PREFIX}{client_api_key}" + document = { + "connection_id": connection_id, + "warehouse_type": warehouse_type, + "config": config, + "credentials": ( + encrypt_warehouse_credentials(credentials) + if credentials is not None + else None + ), + } + get_client().set(redis_key, json.dumps(document)) + + +def remove_warehouse_connection(client_api_key: str) -> None: + redis_key = f"{WAREHOUSE_CONNECTION_KEY_PREFIX}{client_api_key}" + get_client().delete(redis_key) + + +def pop_warehouse_delivery_statuses() -> list[WarehouseDeliveryStatus]: + """Takes every outcome the warehouse-delivery service has left in Redis, + emptying the hash as it goes. An entry that cannot be read is logged and + skipped rather than blocking the others.""" + # The stub types eval for the async client too; this client is synchronous + # and a Lua HGETALL comes back as a flat field, value, field, value list. + entries = cast( + list[bytes], + get_client().eval(_POP_HASH_SCRIPT, 1, WAREHOUSE_DELIVERY_STATUS_KEY), + ) + statuses: list[WarehouseDeliveryStatus] = [] + for field, value in zip(entries[::2], entries[1::2], strict=True): + try: + outcome = json.loads(value) + detail = outcome.get("detail") + status = WarehouseDeliveryStatus( + connection_id=int(field), + status=str(outcome["status"]), + detail=str(detail) if detail is not None else None, + ) + except (ValueError, KeyError, TypeError, AttributeError): + logger.warning( + "delivery_status.unreadable", + field=field.decode(errors="replace"), + exc_info=True, + ) + continue + statuses.append(status) + return statuses diff --git a/api/experimentation/warehouse_delivery_service.py b/api/experimentation/warehouse_verification_service.py similarity index 100% rename from api/experimentation/warehouse_delivery_service.py rename to api/experimentation/warehouse_verification_service.py diff --git a/api/tests/unit/experimentation/conftest.py b/api/tests/unit/experimentation/conftest.py index 261b8aea15ee..12fdbdee5015 100644 --- a/api/tests/unit/experimentation/conftest.py +++ b/api/tests/unit/experimentation/conftest.py @@ -7,7 +7,7 @@ from core.dataclasses import AuthorData from environments.models import Environment -from experimentation import ingestion_sync_service +from experimentation import ingestion_redis from experimentation.dataclasses import AudienceSpec, RolloutSpec from experimentation.models import ( Experiment, @@ -38,8 +38,8 @@ def __call__( @pytest.fixture(autouse=True) def mock_ingestion_redis_client(mocker: MockerFixture) -> None: - ingestion_sync_service._get_client.cache_clear() - mocker.patch("experimentation.ingestion_sync_service.RedisCluster.from_url") + ingestion_redis.get_client.cache_clear() + mocker.patch("experimentation.ingestion_redis.RedisCluster.from_url") @pytest.fixture() diff --git a/api/tests/unit/experimentation/test_ingestion_redis.py b/api/tests/unit/experimentation/test_ingestion_redis.py new file mode 100644 index 000000000000..6dead50f2600 --- /dev/null +++ b/api/tests/unit/experimentation/test_ingestion_redis.py @@ -0,0 +1,28 @@ +from pytest_django.fixtures import SettingsWrapper +from pytest_mock import MockerFixture + +from experimentation import ingestion_redis + + +def test_get_client__configured_url__builds_redis_cluster_with_socket_options( + mocker: MockerFixture, + settings: SettingsWrapper, +) -> None: + # Given + settings.INGESTION_REDIS_URL = "redis://ingestion:6379" + mock_from_url = mocker.patch( + "experimentation.ingestion_redis.RedisCluster.from_url", + ) + ingestion_redis.get_client.cache_clear() + + # When + client = ingestion_redis.get_client() + + # Then + mock_from_url.assert_called_once_with( + "redis://ingestion:6379", + socket_timeout=ingestion_redis.SOCKET_TIMEOUT, + socket_keepalive=True, + ) + assert client is mock_from_url.return_value + ingestion_redis.get_client.cache_clear() diff --git a/api/tests/unit/experimentation/test_ingestion_sync_service.py b/api/tests/unit/experimentation/test_ingestion_sync_service.py index 423b39a2ccd8..a26cb4766949 100644 --- a/api/tests/unit/experimentation/test_ingestion_sync_service.py +++ b/api/tests/unit/experimentation/test_ingestion_sync_service.py @@ -1,37 +1,11 @@ -import json from datetime import datetime from datetime import timezone as dt_timezone import pytest from pytest_mock import MockerFixture -from pytest_structlog import StructuredLogCapture from redis.exceptions import RedisError from experimentation import ingestion_sync_service -from experimentation.dataclasses import WarehouseDeliveryStatus -from experimentation.warehouse_credentials import decrypt_warehouse_credentials - - -def test_get_client__configured_url__builds_redis_cluster_with_socket_options( - mocker: MockerFixture, - settings: object, -) -> None: - # Given - settings.INGESTION_REDIS_URL = "redis://ingestion:6379" # type: ignore[attr-defined] - mock_from_url = mocker.patch( - "experimentation.ingestion_sync_service.RedisCluster.from_url", - ) - - # When - client = ingestion_sync_service._get_client() - - # Then - mock_from_url.assert_called_once_with( - "redis://ingestion:6379", - socket_timeout=ingestion_sync_service.SOCKET_TIMEOUT, - socket_keepalive=True, - ) - assert client is mock_from_url.return_value def test_set_ingestion_key__no_expiry__writes_environment_key_without_ttl( @@ -40,7 +14,7 @@ def test_set_ingestion_key__no_expiry__writes_environment_key_without_ttl( # Given mock_client = mocker.Mock() mocker.patch( - "experimentation.ingestion_sync_service._get_client", + "experimentation.ingestion_sync_service.get_client", return_value=mock_client, ) @@ -64,7 +38,7 @@ def test_set_ingestion_key__expiry__writes_environment_key_with_ttl( # Given mock_client = mocker.Mock() mocker.patch( - "experimentation.ingestion_sync_service._get_client", + "experimentation.ingestion_sync_service.get_client", return_value=mock_client, ) expires_at = datetime(2026, 9, 1, tzinfo=dt_timezone.utc) @@ -90,7 +64,7 @@ def test_delete_ingestion_key__valid_key__deletes_from_redis( # Given mock_client = mocker.Mock() mocker.patch( - "experimentation.ingestion_sync_service._get_client", + "experimentation.ingestion_sync_service.get_client", return_value=mock_client, ) @@ -109,7 +83,7 @@ def test_set_ingestion_destination__valid_topic__writes_topic( # Given mock_client = mocker.Mock() mocker.patch( - "experimentation.ingestion_sync_service._get_client", + "experimentation.ingestion_sync_service.get_client", return_value=mock_client, ) @@ -132,7 +106,7 @@ def test_delete_ingestion_destination__valid_key__deletes_from_redis( # Given mock_client = mocker.Mock() mocker.patch( - "experimentation.ingestion_sync_service._get_client", + "experimentation.ingestion_sync_service.get_client", return_value=mock_client, ) @@ -152,7 +126,7 @@ def test_set_ingestion_key__redis_error__propagates( mock_client = mocker.Mock() mock_client.set.side_effect = RedisError("boom") mocker.patch( - "experimentation.ingestion_sync_service._get_client", + "experimentation.ingestion_sync_service.get_client", return_value=mock_client, ) @@ -171,168 +145,10 @@ def test_delete_ingestion_key__redis_error__propagates( mock_client = mocker.Mock() mock_client.delete.side_effect = RedisError("boom") mocker.patch( - "experimentation.ingestion_sync_service._get_client", + "experimentation.ingestion_sync_service.get_client", return_value=mock_client, ) # When / Then with pytest.raises(RedisError, match="boom"): ingestion_sync_service.delete_ingestion_key("ser.test-key-001") - - -def test_set_ingestion_warehouse__connection_details__writes_document_with_encrypted_credentials( - mocker: MockerFixture, -) -> None: - # Given - mock_client = mocker.Mock() - mocker.patch( - "experimentation.ingestion_sync_service._get_client", - return_value=mock_client, - ) - config = {"host": "ch.acme-corp.example", "port": 8443, "secure": True} - - # When - ingestion_sync_service.set_ingestion_warehouse( - "client-env-key", - connection_id=42, - warehouse_type="clickhouse", - config=config, - credentials={"password": "hunter2"}, - ) - - # Then the document sits under the environment's client key, and the - # password only ever reaches Redis as the ciphertext the database holds - mock_client.set.assert_called_once() - redis_key, raw = mock_client.set.call_args.args - assert redis_key == "experimentation:environment_warehouses:client-env-key" - assert "hunter2" not in raw - document = json.loads(raw) - assert document["connection_id"] == 42 - assert document["warehouse_type"] == "clickhouse" - assert document["config"] == config - assert decrypt_warehouse_credentials(document["credentials"]) == { - "password": "hunter2" - } - - -def test_set_ingestion_warehouse__no_credentials__writes_null( - mocker: MockerFixture, -) -> None: - # Given a warehouse type the API stores no credentials for - mock_client = mocker.Mock() - mocker.patch( - "experimentation.ingestion_sync_service._get_client", - return_value=mock_client, - ) - - # When - ingestion_sync_service.set_ingestion_warehouse( - "client-env-key", - connection_id=42, - warehouse_type="snowflake", - config={"account_identifier": "acme"}, - credentials=None, - ) - - # Then - _, raw = mock_client.set.call_args.args - assert json.loads(raw)["credentials"] is None - - -def test_delete_ingestion_warehouse__client_key__deletes_from_redis( - mocker: MockerFixture, -) -> None: - # Given - mock_client = mocker.Mock() - mocker.patch( - "experimentation.ingestion_sync_service._get_client", - return_value=mock_client, - ) - - # When - ingestion_sync_service.delete_ingestion_warehouse("client-env-key") - - # Then - mock_client.delete.assert_called_once_with( - "experimentation:environment_warehouses:client-env-key", - ) - - -def test_pop_warehouse_delivery_statuses__entries_in_hash__read_and_cleared_in_one_step( - mocker: MockerFixture, -) -> None: - # Given two outcomes the delivery service left, as Redis hands them back - mock_client = mocker.Mock() - mock_client.eval.return_value = [ - b"42", - b'{"status": "errored", "detail": "Authentication failed.", "at": 1758000000.0}', - b"7", - b'{"status": "connected", "detail": null, "at": 1758000001.0}', - ] - mocker.patch( - "experimentation.ingestion_sync_service._get_client", - return_value=mock_client, - ) - - # When - statuses = ingestion_sync_service.pop_warehouse_delivery_statuses() - - # Then both are returned, and the hash was read and deleted by one script - # so nothing written in between is lost - assert statuses == [ - WarehouseDeliveryStatus( - connection_id=42, status="errored", detail="Authentication failed." - ), - WarehouseDeliveryStatus(connection_id=7, status="connected", detail=None), - ] - mock_client.eval.assert_called_once_with( - ingestion_sync_service._POP_HASH_SCRIPT, - 1, - "experimentation:warehouse_delivery_status", - ) - - -def test_pop_warehouse_delivery_statuses__empty_hash__returns_nothing( - mocker: MockerFixture, -) -> None: - # Given - mock_client = mocker.Mock() - mock_client.eval.return_value = [] - mocker.patch( - "experimentation.ingestion_sync_service._get_client", - return_value=mock_client, - ) - - # When / Then - assert ingestion_sync_service.pop_warehouse_delivery_statuses() == [] - - -def test_pop_warehouse_delivery_statuses__unreadable_entries__skipped_and_logged( - mocker: MockerFixture, - log: StructuredLogCapture, -) -> None: - # Given a field that is not a connection id, a value that is not JSON, and - # one good entry - mock_client = mocker.Mock() - mock_client.eval.return_value = [ - b"not-an-id", - b'{"status": "errored"}', - b"42", - b"not json", - b"7", - b'{"status": "connected"}', - ] - mocker.patch( - "experimentation.ingestion_sync_service._get_client", - return_value=mock_client, - ) - - # When - statuses = ingestion_sync_service.pop_warehouse_delivery_statuses() - - # Then the good entry still gets through - assert statuses == [ - WarehouseDeliveryStatus(connection_id=7, status="connected", detail=None) - ] - assert log.has("delivery_status.unreadable", level="warning", field="not-an-id") - assert log.has("delivery_status.unreadable", level="warning", field="42") diff --git a/api/tests/unit/experimentation/test_services.py b/api/tests/unit/experimentation/test_services.py index 1932355a1381..d5ac97c51e70 100644 --- a/api/tests/unit/experimentation/test_services.py +++ b/api/tests/unit/experimentation/test_services.py @@ -270,7 +270,7 @@ def test_get_warehouse_event_names__clickhouse_connection__queries_customer_inst ) -> None: # Given get_client = mocker.patch( - "experimentation.warehouse_delivery_service.clickhouse_connect.get_client", + "experimentation.warehouse_verification_service.clickhouse_connect.get_client", ) if isinstance(query_result, Exception): get_client.return_value.query.side_effect = query_result @@ -323,7 +323,7 @@ def test_get_warehouse_event_names__connection_details_changed__cache_keyed_by_c ) -> None: # Given — a cached result for the connection's current details get_client = mocker.patch( - "experimentation.warehouse_delivery_service.clickhouse_connect.get_client", + "experimentation.warehouse_verification_service.clickhouse_connect.get_client", ) get_client.return_value.query.return_value = mocker.Mock( result_rows=[("old_event",)] @@ -2646,7 +2646,7 @@ def test_verify_clickhouse_connection__reachable__sets_connected( ) -> None: # Given get_client = mocker.patch( - "experimentation.warehouse_delivery_service.clickhouse_connect.get_client", + "experimentation.warehouse_verification_service.clickhouse_connect.get_client", ) success_count_before = _verification_count("success") clickhouse_connection.status_detail = "stale detail" @@ -2709,7 +2709,7 @@ def test_verify_clickhouse_connection__failure__sets_errored_with_detail( ) -> None: # Given get_client = mocker.patch( - "experimentation.warehouse_delivery_service.clickhouse_connect.get_client", + "experimentation.warehouse_verification_service.clickhouse_connect.get_client", ) if isinstance(query_results, list): get_client.return_value.query.side_effect = [ @@ -2740,7 +2740,7 @@ def test_verify_clickhouse_connection__internal_host__sets_errored_without_conne ) -> None: # Given get_client = mocker.patch( - "experimentation.warehouse_delivery_service.clickhouse_connect.get_client", + "experimentation.warehouse_verification_service.clickhouse_connect.get_client", ) clickhouse_connection.config = { **(clickhouse_connection.config or {}), @@ -2782,7 +2782,7 @@ def test_annotate_warehouse_event_stats__clickhouse_connection__queries_customer ) -> None: # Given get_client = mocker.patch( - "experimentation.warehouse_delivery_service.clickhouse_connect.get_client", + "experimentation.warehouse_verification_service.clickhouse_connect.get_client", ) if isinstance(query_result, Exception): get_client.return_value.query.side_effect = query_result diff --git a/api/tests/unit/experimentation/test_tasks.py b/api/tests/unit/experimentation/test_tasks.py index e6c6380e35b6..5814b625e29e 100644 --- a/api/tests/unit/experimentation/test_tasks.py +++ b/api/tests/unit/experimentation/test_tasks.py @@ -60,7 +60,14 @@ def test_sync_environment_ingestion__flagsmith_connection__whitelists_valid_keys name="expired", expires_at=timezone.now() - timedelta(days=1), ) - mock_service = mocker.patch("experimentation.tasks.ingestion_sync_service") + mock_service = mocker.Mock() + mock_service.attach_mock( + mocker.patch("experimentation.tasks.ingestion_sync_service"), "ingestion" + ) + mock_service.attach_mock( + mocker.patch("experimentation.tasks.warehouse_delivery_sync_service"), + "delivery", + ) # When sync_environment_ingestion(environment_id=environment.id) @@ -69,13 +76,13 @@ def test_sync_environment_ingestion__flagsmith_connection__whitelists_valid_keys # published, and only the client key and the valid server-side key are # whitelisted assert mock_service.mock_calls == [ - mocker.call.delete_ingestion_destination(environment.api_key), - mocker.call.delete_ingestion_warehouse(environment.api_key), - mocker.call.set_ingestion_key( + mocker.call.ingestion.delete_ingestion_destination(environment.api_key), + mocker.call.delivery.remove_warehouse_connection(environment.api_key), + mocker.call.ingestion.set_ingestion_key( environment.api_key, environment_key=environment.api_key, ), - mocker.call.set_ingestion_key( + mocker.call.ingestion.set_ingestion_key( valid_key.key, environment_key=environment.api_key, expires_at=valid_key.expires_at, @@ -89,7 +96,14 @@ def test_sync_environment_ingestion__external_connection__publishes_then_routes_ mocker: MockerFixture, ) -> None: # Given - mock_service = mocker.patch("experimentation.tasks.ingestion_sync_service") + mock_service = mocker.Mock() + mock_service.attach_mock( + mocker.patch("experimentation.tasks.ingestion_sync_service"), "ingestion" + ) + mock_service.attach_mock( + mocker.patch("experimentation.tasks.warehouse_delivery_sync_service"), + "delivery", + ) # When sync_environment_ingestion(environment_id=environment.id) @@ -97,18 +111,18 @@ def test_sync_environment_ingestion__external_connection__publishes_then_routes_ # Then the connection is in Redis before events are routed to the topic, and # the key is whitelisted last, so no event arrives anywhere unplaced assert mock_service.mock_calls == [ - mocker.call.set_ingestion_warehouse( + mocker.call.delivery.publish_warehouse_connection( environment.api_key, connection_id=clickhouse_connection.id, warehouse_type="clickhouse", config=clickhouse_connection.config, credentials={"password": "hunter2"}, ), - mocker.call.set_ingestion_destination( + mocker.call.ingestion.set_ingestion_destination( environment.api_key, topic="external_warehouse_events", ), - mocker.call.set_ingestion_key( + mocker.call.ingestion.set_ingestion_key( environment.api_key, environment_key=environment.api_key, ), @@ -129,7 +143,14 @@ def test_sync_environment_ingestion__connection_deleted__removes_keys_and_destin inactive_key = EnvironmentAPIKey.objects.create( environment=environment, name="inactive", active=False ) - mock_service = mocker.patch("experimentation.tasks.ingestion_sync_service") + mock_service = mocker.Mock() + mock_service.attach_mock( + mocker.patch("experimentation.tasks.ingestion_sync_service"), "ingestion" + ) + mock_service.attach_mock( + mocker.patch("experimentation.tasks.warehouse_delivery_sync_service"), + "delivery", + ) # When sync_environment_ingestion(environment_id=environment.id) @@ -137,11 +158,11 @@ def test_sync_environment_ingestion__connection_deleted__removes_keys_and_destin # Then the client key and every server-side key are removed regardless of # state, and the destination routing is cleared assert mock_service.mock_calls == [ - mocker.call.delete_ingestion_key(environment.api_key), - mocker.call.delete_ingestion_key(active_key.key), - mocker.call.delete_ingestion_key(inactive_key.key), - mocker.call.delete_ingestion_destination(environment.api_key), - mocker.call.delete_ingestion_warehouse(environment.api_key), + mocker.call.ingestion.delete_ingestion_key(environment.api_key), + mocker.call.ingestion.delete_ingestion_key(active_key.key), + mocker.call.ingestion.delete_ingestion_key(inactive_key.key), + mocker.call.ingestion.delete_ingestion_destination(environment.api_key), + mocker.call.delivery.remove_warehouse_connection(environment.api_key), ] @@ -152,7 +173,14 @@ def test_sync_environment_ingestion__environment_deleted__removes_keys_and_desti ) -> None: # Given the environment is soft-deleted, taking its connection with it environment.delete() - mock_service = mocker.patch("experimentation.tasks.ingestion_sync_service") + mock_service = mocker.Mock() + mock_service.attach_mock( + mocker.patch("experimentation.tasks.ingestion_sync_service"), "ingestion" + ) + mock_service.attach_mock( + mocker.patch("experimentation.tasks.warehouse_delivery_sync_service"), + "delivery", + ) # When sync_environment_ingestion(environment_id=environment.id) @@ -160,9 +188,9 @@ def test_sync_environment_ingestion__environment_deleted__removes_keys_and_desti # Then the deleted environment is still found, so its key, destination and # connection are removed rather than left accepting events assert mock_service.mock_calls == [ - mocker.call.delete_ingestion_key(environment.api_key), - mocker.call.delete_ingestion_destination(environment.api_key), - mocker.call.delete_ingestion_warehouse(environment.api_key), + mocker.call.ingestion.delete_ingestion_key(environment.api_key), + mocker.call.ingestion.delete_ingestion_destination(environment.api_key), + mocker.call.delivery.remove_warehouse_connection(environment.api_key), ] @@ -171,7 +199,14 @@ def test_sync_environment_ingestion__missing_environment__does_nothing( mocker: MockerFixture, ) -> None: # Given - mock_service = mocker.patch("experimentation.tasks.ingestion_sync_service") + mock_service = mocker.Mock() + mock_service.attach_mock( + mocker.patch("experimentation.tasks.ingestion_sync_service"), "ingestion" + ) + mock_service.attach_mock( + mocker.patch("experimentation.tasks.warehouse_delivery_sync_service"), + "delivery", + ) # When sync_environment_ingestion(environment_id=404404) @@ -701,7 +736,7 @@ def test_apply_warehouse_delivery_statuses__errored_outcome__marks_connection_an # login, and also left an outcome for a connection that no longer exists settings.INGESTION_REDIS_URL = "redis://ingestion:6379" mocker.patch( - "experimentation.tasks.ingestion_sync_service.pop_warehouse_delivery_statuses", + "experimentation.tasks.warehouse_delivery_sync_service.pop_warehouse_delivery_statuses", return_value=[ WarehouseDeliveryStatus( connection_id=clickhouse_connection.id, @@ -744,7 +779,7 @@ def test_apply_warehouse_delivery_statuses__connected_outcome__clears_detail_qui clickhouse_connection.status_detail = "Could not connect to the host." clickhouse_connection.save() mocker.patch( - "experimentation.tasks.ingestion_sync_service.pop_warehouse_delivery_statuses", + "experimentation.tasks.warehouse_delivery_sync_service.pop_warehouse_delivery_statuses", return_value=[ WarehouseDeliveryStatus( connection_id=clickhouse_connection.id, status="connected", detail=None @@ -771,7 +806,7 @@ def test_apply_warehouse_delivery_statuses__unknown_status__skipped_and_logged( # Given a status value the connection model has no choice for settings.INGESTION_REDIS_URL = "redis://ingestion:6379" mocker.patch( - "experimentation.tasks.ingestion_sync_service.pop_warehouse_delivery_statuses", + "experimentation.tasks.warehouse_delivery_sync_service.pop_warehouse_delivery_statuses", return_value=[ WarehouseDeliveryStatus( connection_id=clickhouse_connection.id, status="retrying", detail=None @@ -801,7 +836,7 @@ def test_apply_warehouse_delivery_statuses__long_detail__cut_to_the_column_lengt # Given a detail longer than status_detail can hold settings.INGESTION_REDIS_URL = "redis://ingestion:6379" mocker.patch( - "experimentation.tasks.ingestion_sync_service.pop_warehouse_delivery_statuses", + "experimentation.tasks.warehouse_delivery_sync_service.pop_warehouse_delivery_statuses", return_value=[ WarehouseDeliveryStatus( connection_id=clickhouse_connection.id, @@ -827,7 +862,7 @@ def test_apply_warehouse_delivery_statuses__ingestion_redis_not_configured__does # Given a self-hosted installation with no ingestion Redis settings.INGESTION_REDIS_URL = "" mock_pop = mocker.patch( - "experimentation.tasks.ingestion_sync_service.pop_warehouse_delivery_statuses", + "experimentation.tasks.warehouse_delivery_sync_service.pop_warehouse_delivery_statuses", ) # When diff --git a/api/tests/unit/experimentation/test_views.py b/api/tests/unit/experimentation/test_views.py index 543c3b26969e..6bfd9b35397f 100644 --- a/api/tests/unit/experimentation/test_views.py +++ b/api/tests/unit/experimentation/test_views.py @@ -1226,7 +1226,7 @@ def test_post__clickhouse_minimal_payload__applies_defaults_and_generates_name( # Given enable_features("experimentation_warehouse_connection") mocker.patch( - "experimentation.warehouse_delivery_service.clickhouse_connect.get_client", + "experimentation.warehouse_verification_service.clickhouse_connect.get_client", ) # When @@ -1523,7 +1523,7 @@ def test_post__clickhouse_verification_outcome__returns_201_with_status( # Given enable_features("experimentation_warehouse_connection") mock_client = mocker.patch( - "experimentation.warehouse_delivery_service.clickhouse_connect.get_client", + "experimentation.warehouse_verification_service.clickhouse_connect.get_client", ) mock_client.return_value.query.side_effect = query_side_effect @@ -1576,7 +1576,7 @@ def test_patch__clickhouse_config_without_credentials__keeps_stored_password( # Given enable_features("experimentation_warehouse_connection") mock_client = mocker.patch( - "experimentation.warehouse_delivery_service.clickhouse_connect.get_client", + "experimentation.warehouse_verification_service.clickhouse_connect.get_client", ) url = reverse( "api-v1:environments:experimentation:warehouse-connections-detail", @@ -1611,7 +1611,7 @@ def test_patch__clickhouse_name_only__does_not_reverify( # Given enable_features("experimentation_warehouse_connection") mock_client = mocker.patch( - "experimentation.warehouse_delivery_service.clickhouse_connect.get_client", + "experimentation.warehouse_verification_service.clickhouse_connect.get_client", ) url = reverse( "api-v1:environments:experimentation:warehouse-connections-detail", @@ -1636,7 +1636,7 @@ def test_put__clickhouse_name_only__preserves_config_and_credentials( # Given enable_features("experimentation_warehouse_connection") mocker.patch( - "experimentation.warehouse_delivery_service.clickhouse_connect.get_client", + "experimentation.warehouse_verification_service.clickhouse_connect.get_client", ) url = reverse( "api-v1:environments:experimentation:warehouse-connections-detail", @@ -1672,7 +1672,7 @@ def test_test_warehouse_connection__clickhouse__reverifies_and_returns_status( clickhouse_connection.status = WarehouseConnectionStatus.ERRORED clickhouse_connection.save() mocker.patch( - "experimentation.warehouse_delivery_service.clickhouse_connect.get_client", + "experimentation.warehouse_verification_service.clickhouse_connect.get_client", ) url = reverse( "api-v1:environments:experimentation:" @@ -1743,7 +1743,7 @@ def test_test_warehouse_connection_config__payload__returns_expected_response( # Given enable_features("experimentation_warehouse_connection") mocker.patch( - "experimentation.warehouse_delivery_service.clickhouse_connect.get_client", + "experimentation.warehouse_verification_service.clickhouse_connect.get_client", side_effect=client_side_effect, ) url = reverse( diff --git a/api/tests/unit/experimentation/test_warehouse_delivery_sync_service.py b/api/tests/unit/experimentation/test_warehouse_delivery_sync_service.py new file mode 100644 index 000000000000..f3b1409bd318 --- /dev/null +++ b/api/tests/unit/experimentation/test_warehouse_delivery_sync_service.py @@ -0,0 +1,166 @@ +import json + +from pytest_mock import MockerFixture +from pytest_structlog import StructuredLogCapture + +from experimentation import warehouse_delivery_sync_service +from experimentation.dataclasses import WarehouseDeliveryStatus +from experimentation.warehouse_credentials import decrypt_warehouse_credentials + + +def test_publish_warehouse_connection__connection_details__writes_document_with_encrypted_credentials( + mocker: MockerFixture, +) -> None: + # Given + mock_client = mocker.Mock() + mocker.patch( + "experimentation.warehouse_delivery_sync_service.get_client", + return_value=mock_client, + ) + config = {"host": "ch.acme-corp.example", "port": 8443, "secure": True} + + # When + warehouse_delivery_sync_service.publish_warehouse_connection( + "client-env-key", + connection_id=42, + warehouse_type="clickhouse", + config=config, + credentials={"password": "hunter2"}, + ) + + # Then the document sits under the environment's client key, and the + # password only ever reaches Redis as the ciphertext the database holds + mock_client.set.assert_called_once() + redis_key, raw = mock_client.set.call_args.args + assert redis_key == "experimentation:environment_warehouses:client-env-key" + assert "hunter2" not in raw + document = json.loads(raw) + assert document["connection_id"] == 42 + assert document["warehouse_type"] == "clickhouse" + assert document["config"] == config + assert decrypt_warehouse_credentials(document["credentials"]) == { + "password": "hunter2" + } + + +def test_publish_warehouse_connection__no_credentials__writes_null( + mocker: MockerFixture, +) -> None: + # Given a warehouse type the API stores no credentials for + mock_client = mocker.Mock() + mocker.patch( + "experimentation.warehouse_delivery_sync_service.get_client", + return_value=mock_client, + ) + + # When + warehouse_delivery_sync_service.publish_warehouse_connection( + "client-env-key", + connection_id=42, + warehouse_type="snowflake", + config={"account_identifier": "acme"}, + credentials=None, + ) + + # Then + _, raw = mock_client.set.call_args.args + assert json.loads(raw)["credentials"] is None + + +def test_remove_warehouse_connection__client_key__deletes_from_redis( + mocker: MockerFixture, +) -> None: + # Given + mock_client = mocker.Mock() + mocker.patch( + "experimentation.warehouse_delivery_sync_service.get_client", + return_value=mock_client, + ) + + # When + warehouse_delivery_sync_service.remove_warehouse_connection("client-env-key") + + # Then + mock_client.delete.assert_called_once_with( + "experimentation:environment_warehouses:client-env-key", + ) + + +def test_pop_warehouse_delivery_statuses__entries_in_hash__read_and_cleared_in_one_step( + mocker: MockerFixture, +) -> None: + # Given two outcomes the delivery service left, as Redis hands them back + mock_client = mocker.Mock() + mock_client.eval.return_value = [ + b"42", + b'{"status": "errored", "detail": "Authentication failed.", "at": 1758000000.0}', + b"7", + b'{"status": "connected", "detail": null, "at": 1758000001.0}', + ] + mocker.patch( + "experimentation.warehouse_delivery_sync_service.get_client", + return_value=mock_client, + ) + + # When + statuses = warehouse_delivery_sync_service.pop_warehouse_delivery_statuses() + + # Then both are returned, and the hash was read and deleted by one script + # so nothing written in between is lost + assert statuses == [ + WarehouseDeliveryStatus( + connection_id=42, status="errored", detail="Authentication failed." + ), + WarehouseDeliveryStatus(connection_id=7, status="connected", detail=None), + ] + mock_client.eval.assert_called_once_with( + warehouse_delivery_sync_service._POP_HASH_SCRIPT, + 1, + "experimentation:warehouse_delivery_status", + ) + + +def test_pop_warehouse_delivery_statuses__empty_hash__returns_nothing( + mocker: MockerFixture, +) -> None: + # Given + mock_client = mocker.Mock() + mock_client.eval.return_value = [] + mocker.patch( + "experimentation.warehouse_delivery_sync_service.get_client", + return_value=mock_client, + ) + + # When / Then + assert warehouse_delivery_sync_service.pop_warehouse_delivery_statuses() == [] + + +def test_pop_warehouse_delivery_statuses__unreadable_entries__skipped_and_logged( + mocker: MockerFixture, + log: StructuredLogCapture, +) -> None: + # Given a field that is not a connection id, a value that is not JSON, and + # one good entry + mock_client = mocker.Mock() + mock_client.eval.return_value = [ + b"not-an-id", + b'{"status": "errored"}', + b"42", + b"not json", + b"7", + b'{"status": "connected"}', + ] + mocker.patch( + "experimentation.warehouse_delivery_sync_service.get_client", + return_value=mock_client, + ) + + # When + statuses = warehouse_delivery_sync_service.pop_warehouse_delivery_statuses() + + # Then the good entry still gets through + assert statuses == [ + WarehouseDeliveryStatus(connection_id=7, status="connected", detail=None) + ] + assert log.has("delivery_status.unreadable", level="warning", field="not-an-id") + assert log.has("delivery_status.unreadable", level="warning", field="42") diff --git a/api/tests/unit/experimentation/test_warehouse_delivery_service.py b/api/tests/unit/experimentation/test_warehouse_verification_service.py similarity index 83% rename from api/tests/unit/experimentation/test_warehouse_delivery_service.py rename to api/tests/unit/experimentation/test_warehouse_verification_service.py index 98d3c0685f31..eaa8c9f76ff1 100644 --- a/api/tests/unit/experimentation/test_warehouse_delivery_service.py +++ b/api/tests/unit/experimentation/test_warehouse_verification_service.py @@ -3,7 +3,7 @@ from pytest_mock import MockerFixture from urllib3 import PoolManager -from experimentation import warehouse_delivery_service +from experimentation import warehouse_verification_service from experimentation.models import WarehouseConnection @@ -15,10 +15,10 @@ def test_delivery_client__incomplete_config__raises_config_error( # When / Then with pytest.raises( - warehouse_delivery_service.DeliveryConfigError, + warehouse_verification_service.DeliveryConfigError, match="incomplete", ): - with warehouse_delivery_service.delivery_client( + with warehouse_verification_service.delivery_client( clickhouse_connection, send_receive_timeout=5, ): @@ -33,10 +33,10 @@ def test_delivery_client__internal_host__raises_config_error( # When / Then with pytest.raises( - warehouse_delivery_service.DeliveryConfigError, + warehouse_verification_service.DeliveryConfigError, match="internal or private", ): - with warehouse_delivery_service.delivery_client( + with warehouse_verification_service.delivery_client( clickhouse_connection, send_receive_timeout=5, ): @@ -49,11 +49,11 @@ def test_delivery_client__valid_config__yields_http_client_and_closes( ) -> None: # Given get_client = mocker.patch( - "experimentation.warehouse_delivery_service.clickhouse_connect.get_client", + "experimentation.warehouse_verification_service.clickhouse_connect.get_client", ) # When - with warehouse_delivery_service.delivery_client( + with warehouse_verification_service.delivery_client( clickhouse_connection, send_receive_timeout=5, ) as client: @@ -75,7 +75,7 @@ def test_delivery_client__valid_config__yields_http_client_and_closes( pool_manager = get_client.call_args.kwargs["pool_mgr"] assert isinstance( pool_manager, - warehouse_delivery_service._NoRedirectPoolManager, + warehouse_verification_service._NoRedirectPoolManager, ) get_client.return_value.close.assert_not_called() @@ -88,12 +88,12 @@ def test_delivery_client__body_raises__still_closes_client( ) -> None: # Given get_client = mocker.patch( - "experimentation.warehouse_delivery_service.clickhouse_connect.get_client", + "experimentation.warehouse_verification_service.clickhouse_connect.get_client", ) # When a query inside the block fails with pytest.raises(RuntimeError, match="boom"): - with warehouse_delivery_service.delivery_client( + with warehouse_verification_service.delivery_client( clickhouse_connection, send_receive_timeout=5, ): @@ -121,10 +121,10 @@ def test_check_events_table_exists__exists_query_result__raises_only_when_missin # When / Then if expected_raise: - with pytest.raises(warehouse_delivery_service.MissingEventsTableError): - warehouse_delivery_service.check_events_table_exists(client) + with pytest.raises(warehouse_verification_service.MissingEventsTableError): + warehouse_verification_service.check_events_table_exists(client) else: - warehouse_delivery_service.check_events_table_exists(client) + warehouse_verification_service.check_events_table_exists(client) client.query.assert_called_once_with("EXISTS TABLE events") @@ -132,7 +132,7 @@ def test_check_events_table_exists__exists_query_result__raises_only_when_missin "error, expected_detail", [ pytest.param( - warehouse_delivery_service.DeliveryConfigError( + warehouse_verification_service.DeliveryConfigError( "Stored connection details are incomplete." ), "Stored connection details are incomplete.", @@ -165,7 +165,7 @@ def test_check_events_table_exists__exists_query_result__raises_only_when_missin id="other-server-error", ), pytest.param( - warehouse_delivery_service.MissingEventsTableError(), + warehouse_verification_service.MissingEventsTableError(), "Events table not found in the configured database. " "Run the setup SQL to create it.", id="missing-events-table", @@ -184,7 +184,7 @@ def test_describe_warehouse_error__known_failures__returns_user_facing_detail( # Given a parametrised verification failure # When - detail = warehouse_delivery_service.describe_warehouse_error(error) + detail = warehouse_verification_service.describe_warehouse_error(error) # Then assert detail == expected_detail @@ -196,7 +196,7 @@ def test_no_redirect_pool_manager__urlopen__refuses_to_follow_redirects( # Given a manager asked to follow redirects, as clickhouse-connect's own # request path does urlopen = mocker.patch.object(PoolManager, "urlopen") - manager = warehouse_delivery_service._NoRedirectPoolManager() + manager = warehouse_verification_service._NoRedirectPoolManager() # When manager.urlopen("POST", "https://ch.acme-corp.example/", redirect=True) diff --git a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md index b21304065427..09df28550a6b 100644 --- a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md @@ -196,7 +196,7 @@ Attributes: ### `experimentation.delivery_status.unreadable` Logged at `warning` from: - - `api/experimentation/ingestion_sync_service.py:130` + - `api/experimentation/warehouse_delivery_sync_service.py:83` Attributes: - `exc_info` From 11bcdc05742941f55badcdbd4b648ca5ee7b4aae Mon Sep 17 00:00:00 2001 From: Gagan Trivedi Date: Fri, 18 Sep 2026 16:03:29 +0530 Subject: [PATCH 6/8] feat(experimentation): show delivery failures on fetch instead of copying them into Postgres The connection list and detail views read the delivery service's latest outcome for each verified external connection from Redis with one HMGET and, when it failed, show errored with the reason. Nothing is saved; a connection that failed verification keeps that result. Redis being down or unconfigured falls back to the stored status. The one-minute apply_warehouse_delivery_statuses task and the Lua pop go away, and removing a connection also forgets its outcome so nothing stale shows. --- api/experimentation/services.py | 31 ++- api/experimentation/tasks.py | 51 ++--- api/experimentation/views.py | 3 + .../warehouse_delivery_sync_service.py | 76 ++++--- .../unit/experimentation/test_services.py | 90 +++++++++ api/tests/unit/experimentation/test_tasks.py | 162 +-------------- api/tests/unit/experimentation/test_views.py | 45 ++++- .../test_warehouse_delivery_sync_service.py | 190 ++++++++++-------- .../observability/_events-catalogue.md | 44 ++-- 9 files changed, 360 insertions(+), 332 deletions(-) diff --git a/api/experimentation/services.py b/api/experimentation/services.py index a7ef2d9b6502..26a1ee92c4de 100644 --- a/api/experimentation/services.py +++ b/api/experimentation/services.py @@ -23,7 +23,10 @@ from cohorts.models import Cohort from core.dataclasses import AuthorData from environments.tasks import rebuild_environment_document -from experimentation import warehouse_verification_service +from experimentation import ( + warehouse_delivery_sync_service, + warehouse_verification_service, +) from experimentation.constants import ( CONTROL_VARIANT_KEY, EXPERIMENT_FLAG, @@ -1450,6 +1453,32 @@ def refresh_warehouse_connection_status( return connection +def annotate_warehouse_delivery_statuses( + connections: Sequence[WarehouseConnection], +) -> None: + """For external connections that passed verification, show what the + warehouse-delivery service last saw: a warehouse that has started refusing + events reads as errored with the reason, instead of the connected status + stored when it was saved. A connection that failed verification keeps that + result. Read-only: nothing is saved.""" + verified = [ + connection + for connection in connections + if connection.warehouse_type != WarehouseType.FLAGSMITH + and connection.status == WarehouseConnectionStatus.CONNECTED + ] + if not verified: + return + statuses = warehouse_delivery_sync_service.get_warehouse_delivery_statuses( + [connection.id for connection in verified] + ) + for connection in verified: + outcome = statuses.get(connection.id) + if outcome is not None and outcome.status == WarehouseConnectionStatus.ERRORED: + connection.status = WarehouseConnectionStatus.ERRORED + connection.status_detail = outcome.detail + + def annotate_warehouse_event_stats( connection: WarehouseConnection, environment_key: str, diff --git a/api/experimentation/tasks.py b/api/experimentation/tasks.py index c8eefcecf0da..21c53bd06236 100644 --- a/api/experimentation/tasks.py +++ b/api/experimentation/tasks.py @@ -1,12 +1,8 @@ from datetime import timedelta import structlog -from django.conf import settings from django.utils import timezone -from task_processor.decorators import ( - register_recurring_task, - register_task_handler, -) +from task_processor.decorators import register_task_handler from task_processor.exceptions import TaskBackoffError from environments.models import Environment, EnvironmentAPIKey @@ -17,7 +13,6 @@ ExperimentExposures, ExperimentResults, WarehouseConnection, - WarehouseConnectionStatus, WarehouseType, ) from experimentation.services import ( @@ -53,7 +48,15 @@ def sync_environment_ingestion(environment_id: int) -> None: for api_key in environment.api_keys.all(): ingestion_sync_service.delete_ingestion_key(api_key.key) ingestion_sync_service.delete_ingestion_destination(environment.api_key) - warehouse_delivery_sync_service.remove_warehouse_connection(environment.api_key) + # The connection is gone, so include deleted ones to find its id. + connection_ids = ( + WarehouseConnection.objects.all_with_deleted() + .filter(environment_id=environment.id) + .values_list("id", flat=True) + ) + warehouse_delivery_sync_service.remove_warehouse_connection( + environment.api_key, connection_ids=list(connection_ids) + ) return # Connection details, then destination, then keys. Each step makes the next @@ -62,7 +65,9 @@ def sync_environment_ingestion(environment_id: int) -> None: # events for an environment with no destination to Flagsmith's own topic. if connection.warehouse_type == WarehouseType.FLAGSMITH: ingestion_sync_service.delete_ingestion_destination(environment.api_key) - warehouse_delivery_sync_service.remove_warehouse_connection(environment.api_key) + warehouse_delivery_sync_service.remove_warehouse_connection( + environment.api_key, connection_ids=[connection.id] + ) else: warehouse_delivery_sync_service.publish_warehouse_connection( environment.api_key, @@ -113,36 +118,6 @@ def remove_environment_ingestion_key(key: str) -> None: ingestion_sync_service.delete_ingestion_key(key) -@register_recurring_task(run_every=timedelta(minutes=1), timeout=timedelta(minutes=1)) -def apply_warehouse_delivery_statuses() -> None: - """Copies the outcomes the warehouse-delivery service left in Redis onto - the connections, so the dashboard shows whether a customer's warehouse is - taking their events. That service never writes to Postgres; this task is - the only path from it to the connection row.""" - if not settings.INGESTION_REDIS_URL: - return - for outcome in warehouse_delivery_sync_service.pop_warehouse_delivery_statuses(): - if outcome.status not in WarehouseConnectionStatus.values: - logger.warning( - "delivery_status.unknown", - connection__id=outcome.connection_id, - status=outcome.status, - ) - continue - updated = WarehouseConnection.objects.filter(id=outcome.connection_id).update( - status=outcome.status, - status_detail=outcome.detail[:255] if outcome.detail else None, - ) - # A connected outcome arrives for every live connection every minute, - # so only the failures are worth an event. - if updated and outcome.status == WarehouseConnectionStatus.ERRORED: - logger.warning( - "warehouse_connection.delivery_errored", - connection__id=outcome.connection_id, - status__detail=outcome.detail, - ) - - @register_task_handler(timeout=COMPUTE_TASK_TIMEOUT) def compute_experiment_exposures(experiment_id: int) -> None: experiment = ( diff --git a/api/experimentation/views.py b/api/experimentation/views.py index 822ee0f88ad5..bfb180ee0afa 100644 --- a/api/experimentation/views.py +++ b/api/experimentation/views.py @@ -61,6 +61,7 @@ ) from experimentation.services import ( EVENT_NAMES_SUPPORTED_WAREHOUSE_TYPES, + annotate_warehouse_delivery_statuses, annotate_warehouse_event_stats, apply_experiment_rollout, create_experiment_audit_log, @@ -150,12 +151,14 @@ def list(self, request: Request, *args: object, **kwargs: object) -> Response: if not exclude_event_stats: for connection in connections: annotate_warehouse_event_stats(connection, environment_api_key) + annotate_warehouse_delivery_statuses(connections) serializer = self.get_serializer(connections, many=True) return Response(serializer.data) def retrieve(self, request: Request, *args: object, **kwargs: object) -> Response: connection = self.get_object() annotate_warehouse_event_stats(connection, self.kwargs["environment_api_key"]) + annotate_warehouse_delivery_statuses([connection]) serializer = self.get_serializer(connection) return Response(serializer.data) diff --git a/api/experimentation/warehouse_delivery_sync_service.py b/api/experimentation/warehouse_delivery_sync_service.py index ec42fb54858a..b02d5aac1eb3 100644 --- a/api/experimentation/warehouse_delivery_sync_service.py +++ b/api/experimentation/warehouse_delivery_sync_service.py @@ -1,7 +1,10 @@ import json +from collections.abc import Iterable, Sequence from typing import cast import structlog +from django.conf import settings +from redis.exceptions import RedisError from experimentation.dataclasses import WarehouseDeliveryStatus from experimentation.ingestion_redis import get_client @@ -12,19 +15,9 @@ # same value the ingestion server puts on each Kafka message. WAREHOUSE_CONNECTION_KEY_PREFIX = "experimentation:environment_warehouses:" # One hash the warehouse-delivery service writes each connection's latest -# outcome into, under the connection id. Emptied by -# apply_warehouse_delivery_statuses once a minute. +# outcome into, under the connection id, overwriting the previous one. WAREHOUSE_DELIVERY_STATUS_KEY = "experimentation:warehouse_delivery_status" -# Returns every field and value of the hash and deletes it in the same step, -# so an outcome the delivery service writes while we are reading is never -# deleted unread. -_POP_HASH_SCRIPT = """ -local entries = redis.call('HGETALL', KEYS[1]) -redis.call('DEL', KEYS[1]) -return entries -""" - logger = structlog.get_logger("experimentation") @@ -54,37 +47,60 @@ def publish_warehouse_connection( get_client().set(redis_key, json.dumps(document)) -def remove_warehouse_connection(client_api_key: str) -> None: +def remove_warehouse_connection( + client_api_key: str, + *, + connection_ids: Iterable[int], +) -> None: + """Stops the warehouse-delivery service delivering for the environment and + forgets the outcomes it left for these connections, so a connection that + is deleted or switched back to Flagsmith's warehouse never shows a stale + failure.""" redis_key = f"{WAREHOUSE_CONNECTION_KEY_PREFIX}{client_api_key}" - get_client().delete(redis_key) + client = get_client() + client.delete(redis_key) + fields = [str(connection_id) for connection_id in connection_ids] + if fields: + client.hdel(WAREHOUSE_DELIVERY_STATUS_KEY, *fields) -def pop_warehouse_delivery_statuses() -> list[WarehouseDeliveryStatus]: - """Takes every outcome the warehouse-delivery service has left in Redis, - emptying the hash as it goes. An entry that cannot be read is logged and - skipped rather than blocking the others.""" - # The stub types eval for the async client too; this client is synchronous - # and a Lua HGETALL comes back as a flat field, value, field, value list. - entries = cast( - list[bytes], - get_client().eval(_POP_HASH_SCRIPT, 1, WAREHOUSE_DELIVERY_STATUS_KEY), - ) - statuses: list[WarehouseDeliveryStatus] = [] - for field, value in zip(entries[::2], entries[1::2], strict=True): +def get_warehouse_delivery_statuses( + connection_ids: Sequence[int], +) -> dict[int, WarehouseDeliveryStatus]: + """The latest outcome the warehouse-delivery service left for each of these + connections, by id. A connection it has never delivered for is absent. + + Returns nothing at all when the ingestion Redis is not configured or does + not answer, so the connections page never depends on it being up.""" + if not connection_ids or not settings.INGESTION_REDIS_URL: + return {} + fields = [str(connection_id) for connection_id in connection_ids] + try: + # The stub types hmget for the async client too; this client is + # synchronous. + values = cast( + list[bytes | None], + get_client().hmget(WAREHOUSE_DELIVERY_STATUS_KEY, fields), + ) + except RedisError: + logger.warning("delivery_status.unavailable", exc_info=True) + return {} + statuses: dict[int, WarehouseDeliveryStatus] = {} + for connection_id, value in zip(connection_ids, values, strict=True): + if value is None: + continue try: outcome = json.loads(value) detail = outcome.get("detail") - status = WarehouseDeliveryStatus( - connection_id=int(field), + statuses[connection_id] = WarehouseDeliveryStatus( + connection_id=connection_id, status=str(outcome["status"]), detail=str(detail) if detail is not None else None, ) except (ValueError, KeyError, TypeError, AttributeError): logger.warning( "delivery_status.unreadable", - field=field.decode(errors="replace"), + connection__id=connection_id, exc_info=True, ) - continue - statuses.append(status) return statuses diff --git a/api/tests/unit/experimentation/test_services.py b/api/tests/unit/experimentation/test_services.py index d5ac97c51e70..8571a361c6db 100644 --- a/api/tests/unit/experimentation/test_services.py +++ b/api/tests/unit/experimentation/test_services.py @@ -37,6 +37,7 @@ MetricSpec, ResultsAggregates, RolloutSpec, + WarehouseDeliveryStatus, WarehouseEventNames, WarehouseEventStats, ) @@ -3817,3 +3818,92 @@ def _rule_ids() -> set[int]: ) assert _audit_log_count() == 2 assert _rule_ids() == rule_ids + + +def test_annotate_warehouse_delivery_statuses__verified_connection_failing_delivery__shows_errored( + clickhouse_connection: WarehouseConnection, + mocker: MockerFixture, +) -> None: + # Given a connection that passed verification when it was saved, whose + # warehouse has since started refusing our login + clickhouse_connection.status = WarehouseConnectionStatus.CONNECTED + mock_get = mocker.patch( + "experimentation.services.warehouse_delivery_sync_service.get_warehouse_delivery_statuses", + return_value={ + clickhouse_connection.id: WarehouseDeliveryStatus( + connection_id=clickhouse_connection.id, + status="errored", + detail="Authentication failed.", + ) + }, + ) + + # When + services.annotate_warehouse_delivery_statuses([clickhouse_connection]) + + # Then the dashboard shows the failure and its reason, without a save + mock_get.assert_called_once_with([clickhouse_connection.id]) + assert clickhouse_connection.status == WarehouseConnectionStatus.ERRORED + assert clickhouse_connection.status_detail == "Authentication failed." + stored = WarehouseConnection.objects.get(id=clickhouse_connection.id) + assert stored.status == WarehouseConnectionStatus.CREATED + + +@pytest.mark.parametrize( + "outcome", + [ + pytest.param(None, id="never-delivered"), + pytest.param( + WarehouseDeliveryStatus(connection_id=0, status="connected", detail=None), + id="delivering", + ), + ], +) +def test_annotate_warehouse_delivery_statuses__verified_connection_delivering__unchanged( + clickhouse_connection: WarehouseConnection, + mocker: MockerFixture, + outcome: WarehouseDeliveryStatus | None, +) -> None: + # Given a verified connection the delivery service has no complaint about + clickhouse_connection.status = WarehouseConnectionStatus.CONNECTED + mocker.patch( + "experimentation.services.warehouse_delivery_sync_service.get_warehouse_delivery_statuses", + return_value={clickhouse_connection.id: outcome} if outcome else {}, + ) + + # When + services.annotate_warehouse_delivery_statuses([clickhouse_connection]) + + # Then + assert clickhouse_connection.status == WarehouseConnectionStatus.CONNECTED + assert clickhouse_connection.status_detail is None + + +def test_annotate_warehouse_delivery_statuses__unverified_or_flagsmith__redis_not_consulted( + clickhouse_connection: WarehouseConnection, + environment: Environment, + mocker: MockerFixture, +) -> None: + # Given a ClickHouse connection that failed verification, and a Flagsmith + # connection, which the delivery service never handles + clickhouse_connection.status = WarehouseConnectionStatus.ERRORED + clickhouse_connection.status_detail = "Could not connect to the host." + flagsmith_connection = WarehouseConnection( + environment=environment, + warehouse_type=WarehouseType.FLAGSMITH, + name="Flagsmith", + status=WarehouseConnectionStatus.CONNECTED, + ) + mock_get = mocker.patch( + "experimentation.services.warehouse_delivery_sync_service.get_warehouse_delivery_statuses", + ) + + # When + services.annotate_warehouse_delivery_statuses( + [clickhouse_connection, flagsmith_connection] + ) + + # Then the verification result stands, and Redis is not even asked + mock_get.assert_not_called() + assert clickhouse_connection.status == WarehouseConnectionStatus.ERRORED + assert clickhouse_connection.status_detail == "Could not connect to the host." diff --git a/api/tests/unit/experimentation/test_tasks.py b/api/tests/unit/experimentation/test_tasks.py index 5814b625e29e..83ed9a3ed516 100644 --- a/api/tests/unit/experimentation/test_tasks.py +++ b/api/tests/unit/experimentation/test_tasks.py @@ -6,7 +6,6 @@ import pytest from django.utils import timezone from freezegun import freeze_time -from pytest_django.fixtures import SettingsWrapper from pytest_mock import MockerFixture from pytest_structlog import StructuredLogCapture from task_processor.exceptions import TaskBackoffError @@ -18,7 +17,6 @@ ExposuresTimeseriesPoint, MetricResult, ResultsSummary, - WarehouseDeliveryStatus, ) from experimentation.models import ( Experiment, @@ -26,12 +24,10 @@ ExperimentResults, ExperimentStatus, WarehouseConnection, - WarehouseConnectionStatus, ) from experimentation.services import CLICKHOUSE_BACKGROUND_QUERY_TIMEOUT_SECONDS from experimentation.stats import VariantStats from experimentation.tasks import ( - apply_warehouse_delivery_statuses, compute_experiment_exposures, compute_experiment_results, remove_environment_ingestion_key, @@ -77,7 +73,9 @@ def test_sync_environment_ingestion__flagsmith_connection__whitelists_valid_keys # whitelisted assert mock_service.mock_calls == [ mocker.call.ingestion.delete_ingestion_destination(environment.api_key), - mocker.call.delivery.remove_warehouse_connection(environment.api_key), + mocker.call.delivery.remove_warehouse_connection( + environment.api_key, connection_ids=[warehouse_connection.id] + ), mocker.call.ingestion.set_ingestion_key( environment.api_key, environment_key=environment.api_key, @@ -162,7 +160,9 @@ def test_sync_environment_ingestion__connection_deleted__removes_keys_and_destin mocker.call.ingestion.delete_ingestion_key(active_key.key), mocker.call.ingestion.delete_ingestion_key(inactive_key.key), mocker.call.ingestion.delete_ingestion_destination(environment.api_key), - mocker.call.delivery.remove_warehouse_connection(environment.api_key), + mocker.call.delivery.remove_warehouse_connection( + environment.api_key, connection_ids=[clickhouse_connection.id] + ), ] @@ -190,7 +190,9 @@ def test_sync_environment_ingestion__environment_deleted__removes_keys_and_desti assert mock_service.mock_calls == [ mocker.call.ingestion.delete_ingestion_key(environment.api_key), mocker.call.ingestion.delete_ingestion_destination(environment.api_key), - mocker.call.delivery.remove_warehouse_connection(environment.api_key), + mocker.call.delivery.remove_warehouse_connection( + environment.api_key, connection_ids=[clickhouse_connection.id] + ), ] @@ -724,149 +726,3 @@ def test_compute_experiment_results__experiment_deleted_after_enqueue__skips( # Then the task exits without raising into the task processor mock_compute.assert_not_called() - - -def test_apply_warehouse_delivery_statuses__errored_outcome__marks_connection_and_logs( - clickhouse_connection: WarehouseConnection, - mocker: MockerFixture, - settings: SettingsWrapper, - log: StructuredLogCapture, -) -> None: - # Given the delivery service found the customer's warehouse refusing our - # login, and also left an outcome for a connection that no longer exists - settings.INGESTION_REDIS_URL = "redis://ingestion:6379" - mocker.patch( - "experimentation.tasks.warehouse_delivery_sync_service.pop_warehouse_delivery_statuses", - return_value=[ - WarehouseDeliveryStatus( - connection_id=clickhouse_connection.id, - status="errored", - detail="Authentication failed.", - ), - WarehouseDeliveryStatus( - connection_id=404404, status="connected", detail=None - ), - ], - ) - - # When - apply_warehouse_delivery_statuses() - - # Then the dashboard shows the failure, and the missing connection is ignored - clickhouse_connection.refresh_from_db() - assert clickhouse_connection.status == WarehouseConnectionStatus.ERRORED - assert clickhouse_connection.status_detail == "Authentication failed." - assert log.events == [ - { - "event": "warehouse_connection.delivery_errored", - "level": "warning", - "connection__id": clickhouse_connection.id, - "status__detail": "Authentication failed.", - } - ] - - -def test_apply_warehouse_delivery_statuses__connected_outcome__clears_detail_quietly( - clickhouse_connection: WarehouseConnection, - mocker: MockerFixture, - settings: SettingsWrapper, - log: StructuredLogCapture, -) -> None: - # Given a connection the dashboard currently shows as errored, whose - # warehouse has started taking events again - settings.INGESTION_REDIS_URL = "redis://ingestion:6379" - clickhouse_connection.status = WarehouseConnectionStatus.ERRORED - clickhouse_connection.status_detail = "Could not connect to the host." - clickhouse_connection.save() - mocker.patch( - "experimentation.tasks.warehouse_delivery_sync_service.pop_warehouse_delivery_statuses", - return_value=[ - WarehouseDeliveryStatus( - connection_id=clickhouse_connection.id, status="connected", detail=None - ) - ], - ) - - # When - apply_warehouse_delivery_statuses() - - # Then the connection recovers, and a routine success is not logged - clickhouse_connection.refresh_from_db() - assert clickhouse_connection.status == WarehouseConnectionStatus.CONNECTED - assert clickhouse_connection.status_detail is None - assert log.events == [] - - -def test_apply_warehouse_delivery_statuses__unknown_status__skipped_and_logged( - clickhouse_connection: WarehouseConnection, - mocker: MockerFixture, - settings: SettingsWrapper, - log: StructuredLogCapture, -) -> None: - # Given a status value the connection model has no choice for - settings.INGESTION_REDIS_URL = "redis://ingestion:6379" - mocker.patch( - "experimentation.tasks.warehouse_delivery_sync_service.pop_warehouse_delivery_statuses", - return_value=[ - WarehouseDeliveryStatus( - connection_id=clickhouse_connection.id, status="retrying", detail=None - ) - ], - ) - - # When - apply_warehouse_delivery_statuses() - - # Then the connection is left as it was rather than failing the save - clickhouse_connection.refresh_from_db() - assert clickhouse_connection.status == WarehouseConnectionStatus.CREATED - assert log.has( - "delivery_status.unknown", - level="warning", - connection__id=clickhouse_connection.id, - status="retrying", - ) - - -def test_apply_warehouse_delivery_statuses__long_detail__cut_to_the_column_length( - clickhouse_connection: WarehouseConnection, - mocker: MockerFixture, - settings: SettingsWrapper, -) -> None: - # Given a detail longer than status_detail can hold - settings.INGESTION_REDIS_URL = "redis://ingestion:6379" - mocker.patch( - "experimentation.tasks.warehouse_delivery_sync_service.pop_warehouse_delivery_statuses", - return_value=[ - WarehouseDeliveryStatus( - connection_id=clickhouse_connection.id, - status="errored", - detail="x" * 300, - ) - ], - ) - - # When - apply_warehouse_delivery_statuses() - - # Then - clickhouse_connection.refresh_from_db() - assert clickhouse_connection.status_detail == "x" * 255 - - -def test_apply_warehouse_delivery_statuses__ingestion_redis_not_configured__does_nothing( - db: None, - mocker: MockerFixture, - settings: SettingsWrapper, -) -> None: - # Given a self-hosted installation with no ingestion Redis - settings.INGESTION_REDIS_URL = "" - mock_pop = mocker.patch( - "experimentation.tasks.warehouse_delivery_sync_service.pop_warehouse_delivery_statuses", - ) - - # When - apply_warehouse_delivery_statuses() - - # Then - mock_pop.assert_not_called() diff --git a/api/tests/unit/experimentation/test_views.py b/api/tests/unit/experimentation/test_views.py index 6bfd9b35397f..046d9965f250 100644 --- a/api/tests/unit/experimentation/test_views.py +++ b/api/tests/unit/experimentation/test_views.py @@ -13,7 +13,11 @@ from audit.related_object_type import RelatedObjectType from environments.models import Environment from experimentation import services -from experimentation.dataclasses import WarehouseEventNames, WarehouseEventStats +from experimentation.dataclasses import ( + WarehouseDeliveryStatus, + WarehouseEventNames, + WarehouseEventStats, +) from experimentation.models import ( WarehouseConnection, WarehouseConnectionStatus, @@ -1922,3 +1926,42 @@ 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_list__verified_connection_failing_delivery__shows_errored_without_saving( + admin_client: APIClient, + environment: Environment, + enable_features: EnableFeaturesFixture, + clickhouse_connection: WarehouseConnection, + mocker: MockerFixture, +) -> None: + # Given a connection that passed verification when saved, whose warehouse + # has since started refusing the events the delivery service sends + enable_features("experimentation_warehouse_connection") + clickhouse_connection.status = WarehouseConnectionStatus.CONNECTED + clickhouse_connection.save() + mocker.patch( + "experimentation.services.warehouse_delivery_sync_service.get_warehouse_delivery_statuses", + return_value={ + clickhouse_connection.id: WarehouseDeliveryStatus( + connection_id=clickhouse_connection.id, + status="errored", + detail="Authentication failed.", + ) + }, + ) + url = reverse( + "api-v1:environments:experimentation:warehouse-connections-list", + args=[environment.api_key], + ) + + # When + response = admin_client.get(url, {"exclude_event_stats": "true"}) + + # Then the dashboard sees the failure, while the stored verification stands + assert response.status_code == status.HTTP_200_OK + (connection,) = response.json() + assert connection["status"] == "errored" + assert connection["status_detail"] == "Authentication failed." + clickhouse_connection.refresh_from_db() + assert clickhouse_connection.status == WarehouseConnectionStatus.CONNECTED diff --git a/api/tests/unit/experimentation/test_warehouse_delivery_sync_service.py b/api/tests/unit/experimentation/test_warehouse_delivery_sync_service.py index f3b1409bd318..db51ca4bc712 100644 --- a/api/tests/unit/experimentation/test_warehouse_delivery_sync_service.py +++ b/api/tests/unit/experimentation/test_warehouse_delivery_sync_service.py @@ -1,22 +1,34 @@ import json +from unittest.mock import Mock +import pytest +from pytest_django.fixtures import SettingsWrapper from pytest_mock import MockerFixture from pytest_structlog import StructuredLogCapture +from redis.exceptions import RedisError from experimentation import warehouse_delivery_sync_service from experimentation.dataclasses import WarehouseDeliveryStatus from experimentation.warehouse_credentials import decrypt_warehouse_credentials +STATUS_KEY = "experimentation:warehouse_delivery_status" -def test_publish_warehouse_connection__connection_details__writes_document_with_encrypted_credentials( - mocker: MockerFixture, -) -> None: - # Given - mock_client = mocker.Mock() + +@pytest.fixture() +def redis_client(mocker: MockerFixture, settings: SettingsWrapper) -> Mock: + settings.INGESTION_REDIS_URL = "redis://ingestion:6379" + client = Mock() mocker.patch( "experimentation.warehouse_delivery_sync_service.get_client", - return_value=mock_client, + return_value=client, ) + return client + + +def test_publish_warehouse_connection__connection_details__writes_document_with_encrypted_credentials( + redis_client: Mock, +) -> None: + # Given config = {"host": "ch.acme-corp.example", "port": 8443, "secure": True} # When @@ -30,8 +42,8 @@ def test_publish_warehouse_connection__connection_details__writes_document_with_ # Then the document sits under the environment's client key, and the # password only ever reaches Redis as the ciphertext the database holds - mock_client.set.assert_called_once() - redis_key, raw = mock_client.set.call_args.args + redis_client.set.assert_called_once() + redis_key, raw = redis_client.set.call_args.args assert redis_key == "experimentation:environment_warehouses:client-env-key" assert "hunter2" not in raw document = json.loads(raw) @@ -44,14 +56,9 @@ def test_publish_warehouse_connection__connection_details__writes_document_with_ def test_publish_warehouse_connection__no_credentials__writes_null( - mocker: MockerFixture, + redis_client: Mock, ) -> None: # Given a warehouse type the API stores no credentials for - mock_client = mocker.Mock() - mocker.patch( - "experimentation.warehouse_delivery_sync_service.get_client", - return_value=mock_client, - ) # When warehouse_delivery_sync_service.publish_warehouse_connection( @@ -63,104 +70,123 @@ def test_publish_warehouse_connection__no_credentials__writes_null( ) # Then - _, raw = mock_client.set.call_args.args + _, raw = redis_client.set.call_args.args assert json.loads(raw)["credentials"] is None -def test_remove_warehouse_connection__client_key__deletes_from_redis( - mocker: MockerFixture, +def test_remove_warehouse_connection__connection_ids__deletes_document_and_outcomes( + redis_client: Mock, ) -> None: # Given - mock_client = mocker.Mock() - mocker.patch( - "experimentation.warehouse_delivery_sync_service.get_client", - return_value=mock_client, - ) # When - warehouse_delivery_sync_service.remove_warehouse_connection("client-env-key") + warehouse_delivery_sync_service.remove_warehouse_connection( + "client-env-key", connection_ids=[42, 43] + ) - # Then - mock_client.delete.assert_called_once_with( + # Then the delivery service stops finding the environment, and any failure + # it recorded for these connections can no longer be shown + redis_client.delete.assert_called_once_with( "experimentation:environment_warehouses:client-env-key", ) + redis_client.hdel.assert_called_once_with(STATUS_KEY, "42", "43") -def test_pop_warehouse_delivery_statuses__entries_in_hash__read_and_cleared_in_one_step( - mocker: MockerFixture, +def test_remove_warehouse_connection__no_connection_ids__deletes_document_only( + redis_client: Mock, ) -> None: - # Given two outcomes the delivery service left, as Redis hands them back - mock_client = mocker.Mock() - mock_client.eval.return_value = [ - b"42", + # Given an environment that never had a connection to forget outcomes for + + # When + warehouse_delivery_sync_service.remove_warehouse_connection( + "client-env-key", connection_ids=[] + ) + + # Then + redis_client.delete.assert_called_once() + redis_client.hdel.assert_not_called() + + +def test_get_warehouse_delivery_statuses__outcomes_in_hash__returned_by_connection( + redis_client: Mock, +) -> None: + # Given outcomes for two of three connections, as Redis hands them back + redis_client.hmget.return_value = [ b'{"status": "errored", "detail": "Authentication failed.", "at": 1758000000.0}', - b"7", + None, b'{"status": "connected", "detail": null, "at": 1758000001.0}', ] - mocker.patch( - "experimentation.warehouse_delivery_sync_service.get_client", - return_value=mock_client, - ) # When - statuses = warehouse_delivery_sync_service.pop_warehouse_delivery_statuses() + statuses = warehouse_delivery_sync_service.get_warehouse_delivery_statuses( + [42, 43, 7] + ) - # Then both are returned, and the hash was read and deleted by one script - # so nothing written in between is lost - assert statuses == [ - WarehouseDeliveryStatus( + # Then one round trip fetches them all, and the connection with no + # outcome is simply absent + redis_client.hmget.assert_called_once_with(STATUS_KEY, ["42", "43", "7"]) + assert statuses == { + 42: WarehouseDeliveryStatus( connection_id=42, status="errored", detail="Authentication failed." ), - WarehouseDeliveryStatus(connection_id=7, status="connected", detail=None), - ] - mock_client.eval.assert_called_once_with( - warehouse_delivery_sync_service._POP_HASH_SCRIPT, - 1, - "experimentation:warehouse_delivery_status", - ) + 7: WarehouseDeliveryStatus(connection_id=7, status="connected", detail=None), + } -def test_pop_warehouse_delivery_statuses__empty_hash__returns_nothing( - mocker: MockerFixture, +def test_get_warehouse_delivery_statuses__unreadable_outcome__skipped_and_logged( + redis_client: Mock, + log: StructuredLogCapture, ) -> None: - # Given - mock_client = mocker.Mock() - mock_client.eval.return_value = [] - mocker.patch( - "experimentation.warehouse_delivery_sync_service.get_client", - return_value=mock_client, - ) + # Given one value that is not JSON next to a good one + redis_client.hmget.return_value = [b"not json", b'{"status": "connected"}'] - # When / Then - assert warehouse_delivery_sync_service.pop_warehouse_delivery_statuses() == [] + # When + statuses = warehouse_delivery_sync_service.get_warehouse_delivery_statuses([42, 7]) + # Then the good one still gets through + assert list(statuses) == [7] + assert log.has("delivery_status.unreadable", level="warning", connection__id=42) -def test_pop_warehouse_delivery_statuses__unreadable_entries__skipped_and_logged( - mocker: MockerFixture, + +def test_get_warehouse_delivery_statuses__redis_unavailable__returns_nothing_and_logs( + redis_client: Mock, log: StructuredLogCapture, ) -> None: - # Given a field that is not a connection id, a value that is not JSON, and - # one good entry - mock_client = mocker.Mock() - mock_client.eval.return_value = [ - b"not-an-id", - b'{"status": "errored"}', - b"42", - b"not json", - b"7", - b'{"status": "connected"}', - ] - mocker.patch( - "experimentation.warehouse_delivery_sync_service.get_client", - return_value=mock_client, + # Given the ingestion Redis does not answer + redis_client.hmget.side_effect = RedisError("timeout") + + # When + statuses = warehouse_delivery_sync_service.get_warehouse_delivery_statuses([42]) + + # Then the caller falls back to the stored status rather than failing + assert statuses == {} + assert log.has("delivery_status.unavailable", level="warning") + + +def test_get_warehouse_delivery_statuses__redis_not_configured__returns_nothing( + mocker: MockerFixture, + settings: SettingsWrapper, +) -> None: + # Given a self-hosted installation with no ingestion Redis + settings.INGESTION_REDIS_URL = "" + get_client = mocker.patch( + "experimentation.warehouse_delivery_sync_service.get_client" ) # When - statuses = warehouse_delivery_sync_service.pop_warehouse_delivery_statuses() + statuses = warehouse_delivery_sync_service.get_warehouse_delivery_statuses([42]) - # Then the good entry still gets through - assert statuses == [ - WarehouseDeliveryStatus(connection_id=7, status="connected", detail=None) - ] - assert log.has("delivery_status.unreadable", level="warning", field="not-an-id") - assert log.has("delivery_status.unreadable", level="warning", field="42") + # Then Redis is not even contacted + assert statuses == {} + get_client.assert_not_called() + + +def test_get_warehouse_delivery_statuses__no_connections__returns_nothing( + redis_client: Mock, +) -> None: + # Given / When + statuses = warehouse_delivery_sync_service.get_warehouse_delivery_statuses([]) + + # Then + assert statuses == {} + redis_client.hmget.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 09df28550a6b..bcbdfe4784f7 100644 --- a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md @@ -184,23 +184,22 @@ Attributes: - `environment_api_key` - `environment_id` -### `experimentation.delivery_status.unknown` +### `experimentation.delivery_status.unavailable` Logged at `warning` from: - - `api/experimentation/tasks.py:126` + - `api/experimentation/warehouse_delivery_sync_service.py:86` Attributes: - - `connection.id` - - `status` + - `exc_info` ### `experimentation.delivery_status.unreadable` Logged at `warning` from: - - `api/experimentation/warehouse_delivery_sync_service.py:83` + - `api/experimentation/warehouse_delivery_sync_service.py:101` Attributes: + - `connection.id` - `exc_info` - - `field` ### `experimentation.encrypted_field.decrypt_failed` @@ -213,7 +212,7 @@ Attributes: ### `experimentation.exposures.compute_failed` Logged at `error` from: - - `api/experimentation/tasks.py:170` + - `api/experimentation/tasks.py:145` Attributes: - `environment.id` @@ -225,7 +224,7 @@ Attributes: ### `experimentation.results.compute_failed` Logged at `error` from: - - `api/experimentation/tasks.py:208` + - `api/experimentation/tasks.py:183` Attributes: - `environment.id` @@ -236,7 +235,7 @@ Attributes: ### `experimentation.rollout.applied` Logged at `info` from: - - `api/experimentation/services.py:1270` + - `api/experimentation/services.py:1273` Attributes: - `audience.match` @@ -249,15 +248,6 @@ Attributes: - `feature.id` - `rollout.percentage` -### `experimentation.warehouse_connection.delivery_errored` - -Logged at `warning` from: - - `api/experimentation/tasks.py:139` - -Attributes: - - `connection.id` - - `status.detail` - ### `feature_health.feature_health_event_dismissal_not_supported` Logged at `warning` from: @@ -825,7 +815,7 @@ Attributes: ### `warehouse.connection.connected` Logged at `info` from: - - `api/experimentation/services.py:1445` + - `api/experimentation/services.py:1448` Attributes: - `environment.id` @@ -834,8 +824,8 @@ Attributes: ### `warehouse.connection.event_names_failed` Logged at `warning` from: - - `api/experimentation/services.py:268` - - `api/experimentation/services.py:1545` + - `api/experimentation/services.py:271` + - `api/experimentation/services.py:1574` Attributes: - `environment.id` @@ -845,7 +835,7 @@ Attributes: ### `warehouse.connection.event_stats_failed` Logged at `warning` from: - - `api/experimentation/services.py:1508` + - `api/experimentation/services.py:1537` Attributes: - `environment.id` @@ -854,7 +844,7 @@ Attributes: ### `warehouse.connection.test_event_sent` Logged at `info` from: - - `api/experimentation/services.py:1385` + - `api/experimentation/services.py:1388` Attributes: - `environment.id` @@ -863,7 +853,7 @@ Attributes: ### `warehouse.connection.verification_failed` Logged at `warning` from: - - `api/experimentation/services.py:1420` + - `api/experimentation/services.py:1423` Attributes: - `environment.id` @@ -873,7 +863,7 @@ Attributes: ### `warehouse.connection.verification_succeeded` Logged at `info` from: - - `api/experimentation/services.py:1430` + - `api/experimentation/services.py:1433` Attributes: - `environment.id` @@ -882,7 +872,7 @@ Attributes: ### `warehouse.srm.overallocated` Logged at `error` from: - - `api/experimentation/services.py:624` + - `api/experimentation/services.py:627` Attributes: - `environment.id` @@ -892,7 +882,7 @@ Attributes: ### `warehouse.srm.unkeyed_variant` Logged at `error` from: - - `api/experimentation/services.py:610` + - `api/experimentation/services.py:613` Attributes: - `environment.id` From b5949c6d4349913f58f066bacbe945c12ad5b025 Mon Sep 17 00:00:00 2001 From: Gagan Trivedi Date: Fri, 18 Sep 2026 16:07:36 +0530 Subject: [PATCH 7/8] docs(experimentation): shorten the delivery status overlay docstring --- api/experimentation/services.py | 5 +---- .../observability/_events-catalogue.md | 4 ++-- 2 files changed, 3 insertions(+), 6 deletions(-) diff --git a/api/experimentation/services.py b/api/experimentation/services.py index 26a1ee92c4de..d41c9773b64f 100644 --- a/api/experimentation/services.py +++ b/api/experimentation/services.py @@ -1457,10 +1457,7 @@ def annotate_warehouse_delivery_statuses( connections: Sequence[WarehouseConnection], ) -> None: """For external connections that passed verification, show what the - warehouse-delivery service last saw: a warehouse that has started refusing - events reads as errored with the reason, instead of the connected status - stored when it was saved. A connection that failed verification keeps that - result. Read-only: nothing is saved.""" + warehouse-delivery service last saw. Read-only: nothing is saved.""" verified = [ connection for connection in connections diff --git a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md index bcbdfe4784f7..27358427b732 100644 --- a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md @@ -825,7 +825,7 @@ Attributes: Logged at `warning` from: - `api/experimentation/services.py:271` - - `api/experimentation/services.py:1574` + - `api/experimentation/services.py:1571` Attributes: - `environment.id` @@ -835,7 +835,7 @@ Attributes: ### `warehouse.connection.event_stats_failed` Logged at `warning` from: - - `api/experimentation/services.py:1537` + - `api/experimentation/services.py:1534` Attributes: - `environment.id` From 38438dd7b46ec49db1a2d3db2ec3b9aa46dd6c05 Mon Sep 17 00:00:00 2001 From: Gagan Trivedi Date: Thu, 24 Sep 2026 08:55:29 +0530 Subject: [PATCH 8/8] fix(experimentation): forget stale delivery outcomes, fall back when the cluster is unreachable A connection whose details changed kept showing the outcome the delivery service recorded for its old details, such as a failure the customer had just fixed. sync_environment_ingestion now deletes a connection's outcome whenever it publishes or removes the connection, through its own delete_warehouse_delivery_statuses, so publish and remove each do one thing. Creating the cluster client connects immediately, and a cluster it cannot reach raises RedisClusterException, which is not a RedisError, so the connections page returned a 500 instead of falling back to stored statuses. --- api/experimentation/tasks.py | 14 ++++-- .../warehouse_delivery_sync_service.py | 23 +++------ api/tests/unit/experimentation/test_tasks.py | 23 ++++++--- .../test_warehouse_delivery_sync_service.py | 49 ++++++++++++++----- .../observability/_events-catalogue.md | 8 +-- 5 files changed, 74 insertions(+), 43 deletions(-) diff --git a/api/experimentation/tasks.py b/api/experimentation/tasks.py index 21c53bd06236..4e3e992f0350 100644 --- a/api/experimentation/tasks.py +++ b/api/experimentation/tasks.py @@ -54,8 +54,9 @@ def sync_environment_ingestion(environment_id: int) -> None: .filter(environment_id=environment.id) .values_list("id", flat=True) ) - warehouse_delivery_sync_service.remove_warehouse_connection( - environment.api_key, connection_ids=list(connection_ids) + warehouse_delivery_sync_service.remove_warehouse_connection(environment.api_key) + warehouse_delivery_sync_service.delete_warehouse_delivery_statuses( + list(connection_ids) ) return @@ -65,8 +66,9 @@ def sync_environment_ingestion(environment_id: int) -> None: # events for an environment with no destination to Flagsmith's own topic. if connection.warehouse_type == WarehouseType.FLAGSMITH: ingestion_sync_service.delete_ingestion_destination(environment.api_key) - warehouse_delivery_sync_service.remove_warehouse_connection( - environment.api_key, connection_ids=[connection.id] + warehouse_delivery_sync_service.remove_warehouse_connection(environment.api_key) + warehouse_delivery_sync_service.delete_warehouse_delivery_statuses( + [connection.id] ) else: warehouse_delivery_sync_service.publish_warehouse_connection( @@ -76,6 +78,10 @@ def sync_environment_ingestion(environment_id: int) -> None: config=connection.config or {}, credentials=connection.credentials, ) + # Its last outcome was about the details just replaced. + warehouse_delivery_sync_service.delete_warehouse_delivery_statuses( + [connection.id] + ) ingestion_sync_service.set_ingestion_destination( environment.api_key, topic=EXTERNAL_WAREHOUSE_EVENTS_TOPIC, diff --git a/api/experimentation/warehouse_delivery_sync_service.py b/api/experimentation/warehouse_delivery_sync_service.py index b02d5aac1eb3..5e8f9a1fef3a 100644 --- a/api/experimentation/warehouse_delivery_sync_service.py +++ b/api/experimentation/warehouse_delivery_sync_service.py @@ -4,7 +4,7 @@ import structlog from django.conf import settings -from redis.exceptions import RedisError +from redis.exceptions import RedisClusterException, RedisError from experimentation.dataclasses import WarehouseDeliveryStatus from experimentation.ingestion_redis import get_client @@ -47,21 +47,14 @@ def publish_warehouse_connection( get_client().set(redis_key, json.dumps(document)) -def remove_warehouse_connection( - client_api_key: str, - *, - connection_ids: Iterable[int], -) -> None: - """Stops the warehouse-delivery service delivering for the environment and - forgets the outcomes it left for these connections, so a connection that - is deleted or switched back to Flagsmith's warehouse never shows a stale - failure.""" - redis_key = f"{WAREHOUSE_CONNECTION_KEY_PREFIX}{client_api_key}" - client = get_client() - client.delete(redis_key) +def remove_warehouse_connection(client_api_key: str) -> None: + get_client().delete(f"{WAREHOUSE_CONNECTION_KEY_PREFIX}{client_api_key}") + + +def delete_warehouse_delivery_statuses(connection_ids: Iterable[int]) -> None: fields = [str(connection_id) for connection_id in connection_ids] if fields: - client.hdel(WAREHOUSE_DELIVERY_STATUS_KEY, *fields) + get_client().hdel(WAREHOUSE_DELIVERY_STATUS_KEY, *fields) def get_warehouse_delivery_statuses( @@ -82,7 +75,7 @@ def get_warehouse_delivery_statuses( list[bytes | None], get_client().hmget(WAREHOUSE_DELIVERY_STATUS_KEY, fields), ) - except RedisError: + except (RedisError, RedisClusterException): logger.warning("delivery_status.unavailable", exc_info=True) return {} statuses: dict[int, WarehouseDeliveryStatus] = {} diff --git a/api/tests/unit/experimentation/test_tasks.py b/api/tests/unit/experimentation/test_tasks.py index 83ed9a3ed516..3c685f5d50e8 100644 --- a/api/tests/unit/experimentation/test_tasks.py +++ b/api/tests/unit/experimentation/test_tasks.py @@ -73,8 +73,9 @@ def test_sync_environment_ingestion__flagsmith_connection__whitelists_valid_keys # whitelisted assert mock_service.mock_calls == [ mocker.call.ingestion.delete_ingestion_destination(environment.api_key), - mocker.call.delivery.remove_warehouse_connection( - environment.api_key, connection_ids=[warehouse_connection.id] + mocker.call.delivery.remove_warehouse_connection(environment.api_key), + mocker.call.delivery.delete_warehouse_delivery_statuses( + [warehouse_connection.id] ), mocker.call.ingestion.set_ingestion_key( environment.api_key, @@ -106,8 +107,9 @@ def test_sync_environment_ingestion__external_connection__publishes_then_routes_ # When sync_environment_ingestion(environment_id=environment.id) - # Then the connection is in Redis before events are routed to the topic, and - # the key is whitelisted last, so no event arrives anywhere unplaced + # Then the connection is in Redis before events are routed to the topic, its + # outcome from before these details is forgotten, and the key is + # whitelisted last, so no event arrives anywhere unplaced assert mock_service.mock_calls == [ mocker.call.delivery.publish_warehouse_connection( environment.api_key, @@ -116,6 +118,9 @@ def test_sync_environment_ingestion__external_connection__publishes_then_routes_ config=clickhouse_connection.config, credentials={"password": "hunter2"}, ), + mocker.call.delivery.delete_warehouse_delivery_statuses( + [clickhouse_connection.id] + ), mocker.call.ingestion.set_ingestion_destination( environment.api_key, topic="external_warehouse_events", @@ -160,8 +165,9 @@ def test_sync_environment_ingestion__connection_deleted__removes_keys_and_destin mocker.call.ingestion.delete_ingestion_key(active_key.key), mocker.call.ingestion.delete_ingestion_key(inactive_key.key), mocker.call.ingestion.delete_ingestion_destination(environment.api_key), - mocker.call.delivery.remove_warehouse_connection( - environment.api_key, connection_ids=[clickhouse_connection.id] + mocker.call.delivery.remove_warehouse_connection(environment.api_key), + mocker.call.delivery.delete_warehouse_delivery_statuses( + [clickhouse_connection.id] ), ] @@ -190,8 +196,9 @@ def test_sync_environment_ingestion__environment_deleted__removes_keys_and_desti assert mock_service.mock_calls == [ mocker.call.ingestion.delete_ingestion_key(environment.api_key), mocker.call.ingestion.delete_ingestion_destination(environment.api_key), - mocker.call.delivery.remove_warehouse_connection( - environment.api_key, connection_ids=[clickhouse_connection.id] + mocker.call.delivery.remove_warehouse_connection(environment.api_key), + mocker.call.delivery.delete_warehouse_delivery_statuses( + [clickhouse_connection.id] ), ] diff --git a/api/tests/unit/experimentation/test_warehouse_delivery_sync_service.py b/api/tests/unit/experimentation/test_warehouse_delivery_sync_service.py index db51ca4bc712..b2d54041a2af 100644 --- a/api/tests/unit/experimentation/test_warehouse_delivery_sync_service.py +++ b/api/tests/unit/experimentation/test_warehouse_delivery_sync_service.py @@ -5,7 +5,7 @@ from pytest_django.fixtures import SettingsWrapper from pytest_mock import MockerFixture from pytest_structlog import StructuredLogCapture -from redis.exceptions import RedisError +from redis.exceptions import RedisClusterException, RedisError from experimentation import warehouse_delivery_sync_service from experimentation.dataclasses import WarehouseDeliveryStatus @@ -74,36 +74,41 @@ def test_publish_warehouse_connection__no_credentials__writes_null( assert json.loads(raw)["credentials"] is None -def test_remove_warehouse_connection__connection_ids__deletes_document_and_outcomes( +def test_remove_warehouse_connection__client_api_key__deletes_document( redis_client: Mock, ) -> None: # Given # When - warehouse_delivery_sync_service.remove_warehouse_connection( - "client-env-key", connection_ids=[42, 43] - ) + warehouse_delivery_sync_service.remove_warehouse_connection("client-env-key") - # Then the delivery service stops finding the environment, and any failure - # it recorded for these connections can no longer be shown + # Then the delivery service stops finding the environment redis_client.delete.assert_called_once_with( "experimentation:environment_warehouses:client-env-key", ) + + +def test_delete_warehouse_delivery_statuses__connection_ids__deletes_their_outcomes( + redis_client: Mock, +) -> None: + # Given + + # When + warehouse_delivery_sync_service.delete_warehouse_delivery_statuses([42, 43]) + + # Then any outcome recorded for these connections can no longer be shown redis_client.hdel.assert_called_once_with(STATUS_KEY, "42", "43") -def test_remove_warehouse_connection__no_connection_ids__deletes_document_only( +def test_delete_warehouse_delivery_statuses__no_connection_ids__does_not_call_redis( redis_client: Mock, ) -> None: # Given an environment that never had a connection to forget outcomes for # When - warehouse_delivery_sync_service.remove_warehouse_connection( - "client-env-key", connection_ids=[] - ) + warehouse_delivery_sync_service.delete_warehouse_delivery_statuses([]) # Then - redis_client.delete.assert_called_once() redis_client.hdel.assert_not_called() @@ -163,6 +168,26 @@ def test_get_warehouse_delivery_statuses__redis_unavailable__returns_nothing_and assert log.has("delivery_status.unavailable", level="warning") +def test_get_warehouse_delivery_statuses__cluster_unreachable_on_connect__returns_nothing_and_logs( + mocker: MockerFixture, + settings: SettingsWrapper, + log: StructuredLogCapture, +) -> None: + # Given a cluster client that finds no reachable node as it is created + settings.INGESTION_REDIS_URL = "rediss://ingestion:6379" + mocker.patch( + "experimentation.warehouse_delivery_sync_service.get_client", + side_effect=RedisClusterException("Redis Cluster cannot be connected."), + ) + + # When + statuses = warehouse_delivery_sync_service.get_warehouse_delivery_statuses([42]) + + # Then the caller falls back to the stored status rather than failing + assert statuses == {} + assert log.has("delivery_status.unavailable", level="warning") + + def test_get_warehouse_delivery_statuses__redis_not_configured__returns_nothing( mocker: MockerFixture, settings: SettingsWrapper, diff --git a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md index 27358427b732..d0c84eed61ef 100644 --- a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md @@ -187,7 +187,7 @@ Attributes: ### `experimentation.delivery_status.unavailable` Logged at `warning` from: - - `api/experimentation/warehouse_delivery_sync_service.py:86` + - `api/experimentation/warehouse_delivery_sync_service.py:79` Attributes: - `exc_info` @@ -195,7 +195,7 @@ Attributes: ### `experimentation.delivery_status.unreadable` Logged at `warning` from: - - `api/experimentation/warehouse_delivery_sync_service.py:101` + - `api/experimentation/warehouse_delivery_sync_service.py:94` Attributes: - `connection.id` @@ -212,7 +212,7 @@ Attributes: ### `experimentation.exposures.compute_failed` Logged at `error` from: - - `api/experimentation/tasks.py:145` + - `api/experimentation/tasks.py:151` Attributes: - `environment.id` @@ -224,7 +224,7 @@ Attributes: ### `experimentation.results.compute_failed` Logged at `error` from: - - `api/experimentation/tasks.py:183` + - `api/experimentation/tasks.py:189` Attributes: - `environment.id`