diff --git a/api/core/fields.py b/api/core/fields.py index 90051714cfd3..4206c6f3060a 100644 --- a/api/core/fields.py +++ b/api/core/fields.py @@ -1,17 +1,9 @@ -import base64 -import hashlib -import json -from typing import Any, TypeVar +from typing import TypeVar -import structlog -from cryptography.fernet import Fernet, InvalidToken -from django.conf import settings from django.db import models from core.validators import validate_http_url_scheme, validate_no_internal_address -logger = structlog.get_logger("core") - _ST = TypeVar("_ST") _GT = TypeVar("_GT") @@ -34,38 +26,3 @@ class NoSSRFURLField(models.URLField[_ST, _GT]): validate_http_url_scheme, validate_no_internal_address, ] - - -def _get_fernet() -> Fernet: - secret: str = settings.WAREHOUSE_CREDENTIALS_SECRET - digest = hashlib.sha256(secret.encode()).digest() - return Fernet(base64.urlsafe_b64encode(digest)) - - -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() - - def from_db_value( - self, - value: str | None, - expression: object, - connection: object, - ) -> Any: - if value is None: - return None - try: - plaintext = _get_fernet().decrypt(value.encode()) - 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": - raise NotImplementedError( - "EncryptedJSONField only supports isnull lookups." - ) - return super().get_lookup(lookup_name) diff --git a/api/experimentation/dataclasses.py b/api/experimentation/dataclasses.py index 528d0b4e76f8..2d52c54f358f 100644 --- a/api/experimentation/dataclasses.py +++ b/api/experimentation/dataclasses.py @@ -125,3 +125,10 @@ 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: + connection_id: int + status: str + detail: str | None 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_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 734e4fe8f0bb..c8a9f85ec746 100644 --- a/api/experimentation/ingestion_sync_service.py +++ b/api/experimentation/ingestion_sync_service.py @@ -1,10 +1,8 @@ from __future__ import annotations -from functools import lru_cache from typing import TYPE_CHECKING -from django.conf import settings -from redis.cluster import RedisCluster +from experimentation.ingestion_redis import get_client if TYPE_CHECKING: from datetime import datetime @@ -13,18 +11,6 @@ INGESTION_ENVIRONMENT_DESTINATION_PREFIX = "experimentation:environment_destinations:" -SOCKET_TIMEOUT = 1 - - -@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( key: str, *, @@ -32,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, @@ -41,14 +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) + get_client().delete(redis_key) 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 8f352061bdcd..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 @@ -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/services.py b/api/experimentation/services.py index ff8069b2a483..287f468b4e7d 100644 --- a/api/experimentation/services.py +++ b/api/experimentation/services.py @@ -22,7 +22,10 @@ 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_delivery_sync_service, + warehouse_verification_service, +) from experimentation.constants import ( CONTROL_VARIANT_KEY, EXPERIMENT_FLAG, @@ -1413,15 +1416,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"]) @@ -1461,6 +1464,29 @@ 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. 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, @@ -1501,7 +1527,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: @@ -1539,7 +1565,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 6875bc16e7aa..4e3e992f0350 100644 --- a/api/experimentation/tasks.py +++ b/api/experimentation/tasks.py @@ -6,12 +6,13 @@ 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, ExperimentExposures, ExperimentResults, + WarehouseConnection, WarehouseType, ) from experimentation.services import ( @@ -47,14 +48,40 @@ 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) + # 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) + warehouse_delivery_sync_service.delete_warehouse_delivery_statuses( + list(connection_ids) + ) 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) + 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( + environment.api_key, + connection_id=connection.id, + warehouse_type=connection.warehouse_type, + 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/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_credentials.py b/api/experimentation/warehouse_credentials.py new file mode 100644 index 000000000000..ca59d2840ac2 --- /dev/null +++ b/api/experimentation/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/warehouse_delivery_sync_service.py b/api/experimentation/warehouse_delivery_sync_service.py new file mode 100644 index 000000000000..5e8f9a1fef3a --- /dev/null +++ b/api/experimentation/warehouse_delivery_sync_service.py @@ -0,0 +1,99 @@ +import json +from collections.abc import Iterable, Sequence +from typing import cast + +import structlog +from django.conf import settings +from redis.exceptions import RedisClusterException, RedisError + +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, overwriting the previous one. +WAREHOUSE_DELIVERY_STATUS_KEY = "experimentation:warehouse_delivery_status" + +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: + 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: + get_client().hdel(WAREHOUSE_DELIVERY_STATUS_KEY, *fields) + + +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, RedisClusterException): + 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") + 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", + connection__id=connection_id, + exc_info=True, + ) + 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/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/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_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_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 7d4b6cd61de8..a26cb4766949 100644 --- a/api/tests/unit/experimentation/test_ingestion_sync_service.py +++ b/api/tests/unit/experimentation/test_ingestion_sync_service.py @@ -8,35 +8,13 @@ from experimentation import ingestion_sync_service -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( mocker: MockerFixture, ) -> None: # Given mock_client = mocker.Mock() mocker.patch( - "experimentation.ingestion_sync_service._get_client", + "experimentation.ingestion_sync_service.get_client", return_value=mock_client, ) @@ -60,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) @@ -86,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, ) @@ -105,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, ) @@ -128,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, ) @@ -148,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, ) @@ -167,7 +145,7 @@ 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, ) 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_services.py b/api/tests/unit/experimentation/test_services.py index aea1eac543fc..56f3febbb914 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, ) @@ -258,7 +259,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 @@ -311,7 +312,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",)] @@ -2678,7 +2679,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" @@ -2741,7 +2742,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 = [ @@ -2772,7 +2773,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 {}), @@ -2814,7 +2815,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 @@ -3849,3 +3850,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 d02f5f4937f6..3c685f5d50e8 100644 --- a/api/tests/unit/experimentation/test_tasks.py +++ b/api/tests/unit/experimentation/test_tasks.py @@ -56,20 +56,32 @@ 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) - # 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.set_ingestion_key( + mocker.call.ingestion.delete_ingestion_destination(environment.api_key), + 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, 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, @@ -77,25 +89,43 @@ 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, ) -> 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) - # 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, 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.set_ingestion_destination( + 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.delivery.delete_warehouse_delivery_statuses( + [clickhouse_connection.id] + ), + 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, ), @@ -116,7 +146,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) @@ -124,10 +161,14 @@ 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.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), + mocker.call.delivery.delete_warehouse_delivery_statuses( + [clickhouse_connection.id] + ), ] @@ -138,16 +179,27 @@ 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) - # 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.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.delete_warehouse_delivery_statuses( + [clickhouse_connection.id] + ), ] @@ -156,7 +208,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) diff --git a/api/tests/unit/experimentation/test_views.py b/api/tests/unit/experimentation/test_views.py index 37cb91463405..de4033fb0a60 100644 --- a/api/tests/unit/experimentation/test_views.py +++ b/api/tests/unit/experimentation/test_views.py @@ -12,7 +12,11 @@ from audit.models import AuditLog from audit.related_object_type import RelatedObjectType from environments.models import Environment -from experimentation.dataclasses import WarehouseEventNames, WarehouseEventStats +from experimentation.dataclasses import ( + WarehouseDeliveryStatus, + WarehouseEventNames, + WarehouseEventStats, +) from experimentation.models import ( WarehouseConnection, WarehouseConnectionStatus, @@ -1224,7 +1228,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 @@ -1521,7 +1525,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 @@ -1574,7 +1578,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", @@ -1609,7 +1613,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", @@ -1634,7 +1638,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", @@ -1670,7 +1674,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:" @@ -1741,7 +1745,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( @@ -1920,3 +1924,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_credentials.py b/api/tests/unit/experimentation/test_warehouse_credentials.py new file mode 100644 index 000000000000..e7a17dd2ef2c --- /dev/null +++ b/api/tests/unit/experimentation/test_warehouse_credentials.py @@ -0,0 +1,34 @@ +import pytest +from cryptography.fernet import InvalidToken +from pytest_django.fixtures import SettingsWrapper + +from experimentation.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_warehouse_delivery_sync_service.py b/api/tests/unit/experimentation/test_warehouse_delivery_sync_service.py new file mode 100644 index 000000000000..b2d54041a2af --- /dev/null +++ b/api/tests/unit/experimentation/test_warehouse_delivery_sync_service.py @@ -0,0 +1,217 @@ +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 RedisClusterException, 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" + + +@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=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 + 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 + 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) + 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( + redis_client: Mock, +) -> None: + # Given a warehouse type the API stores no credentials for + + # 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 = redis_client.set.call_args.args + assert json.loads(raw)["credentials"] is None + + +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") + + # 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_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.delete_warehouse_delivery_statuses([]) + + # Then + 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}', + None, + b'{"status": "connected", "detail": null, "at": 1758000001.0}', + ] + + # When + statuses = warehouse_delivery_sync_service.get_warehouse_delivery_statuses( + [42, 43, 7] + ) + + # 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." + ), + 7: WarehouseDeliveryStatus(connection_id=7, status="connected", detail=None), + } + + +def test_get_warehouse_delivery_statuses__unreadable_outcome__skipped_and_logged( + redis_client: Mock, + log: StructuredLogCapture, +) -> None: + # Given one value that is not JSON next to a good one + redis_client.hmget.return_value = [b"not json", b'{"status": "connected"}'] + + # 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_get_warehouse_delivery_statuses__redis_unavailable__returns_nothing_and_logs( + redis_client: Mock, + log: StructuredLogCapture, +) -> None: + # 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__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, +) -> 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.get_warehouse_delivery_statuses([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/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 85d8001426f9..b019955bc6b1 100644 --- a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md @@ -175,27 +175,44 @@ Attributes: - `error.message` - `source` -### `core.encrypted_field.decrypt_failed` +### `dynamodb.environment_document_compressed` + +Logged at `info` from: + - `api/environments/dynamodb/wrappers/environment_wrapper.py:93` + +Attributes: + - `environment_api_key` + - `environment_id` + +### `experimentation.delivery_status.unavailable` Logged at `warning` from: - - `api/core/fields.py:62` + - `api/experimentation/warehouse_delivery_sync_service.py:79` Attributes: - `exc_info` -### `dynamodb.environment_document_compressed` +### `experimentation.delivery_status.unreadable` -Logged at `info` from: - - `api/environments/dynamodb/wrappers/environment_wrapper.py:93` +Logged at `warning` from: + - `api/experimentation/warehouse_delivery_sync_service.py:94` Attributes: - - `environment_api_key` - - `environment_id` + - `connection.id` + - `exc_info` + +### `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: - - `api/experimentation/tasks.py:124` + - `api/experimentation/tasks.py:151` Attributes: - `environment.id` @@ -207,7 +224,7 @@ Attributes: ### `experimentation.results.compute_failed` Logged at `error` from: - - `api/experimentation/tasks.py:162` + - `api/experimentation/tasks.py:189` Attributes: - `environment.id` @@ -218,7 +235,7 @@ Attributes: ### `experimentation.rollout.applied` Logged at `info` from: - - `api/experimentation/services.py:1281` + - `api/experimentation/services.py:1284` Attributes: - `audience.match` @@ -845,7 +862,7 @@ Attributes: ### `warehouse.connection.connected` Logged at `info` from: - - `api/experimentation/services.py:1456` + - `api/experimentation/services.py:1459` Attributes: - `environment.id` @@ -854,8 +871,8 @@ Attributes: ### `warehouse.connection.event_names_failed` Logged at `warning` from: - - `api/experimentation/services.py:267` - - `api/experimentation/services.py:1556` + - `api/experimentation/services.py:270` + - `api/experimentation/services.py:1582` Attributes: - `environment.id` @@ -865,7 +882,7 @@ Attributes: ### `warehouse.connection.event_stats_failed` Logged at `warning` from: - - `api/experimentation/services.py:1519` + - `api/experimentation/services.py:1545` Attributes: - `environment.id` @@ -874,7 +891,7 @@ Attributes: ### `warehouse.connection.test_event_sent` Logged at `info` from: - - `api/experimentation/services.py:1396` + - `api/experimentation/services.py:1399` Attributes: - `environment.id` @@ -883,7 +900,7 @@ Attributes: ### `warehouse.connection.verification_failed` Logged at `warning` from: - - `api/experimentation/services.py:1431` + - `api/experimentation/services.py:1434` Attributes: - `environment.id` @@ -893,7 +910,7 @@ Attributes: ### `warehouse.connection.verification_succeeded` Logged at `info` from: - - `api/experimentation/services.py:1441` + - `api/experimentation/services.py:1444` Attributes: - `environment.id` @@ -902,7 +919,7 @@ Attributes: ### `warehouse.srm.overallocated` Logged at `error` from: - - `api/experimentation/services.py:635` + - `api/experimentation/services.py:638` Attributes: - `environment.id` @@ -912,7 +929,7 @@ Attributes: ### `warehouse.srm.unkeyed_variant` Logged at `error` from: - - `api/experimentation/services.py:621` + - `api/experimentation/services.py:624` Attributes: - `environment.id`