From 68928225930a16eaa0bcbedc9898a4f48e794230 Mon Sep 17 00:00:00 2001 From: Gagan Trivedi Date: Mon, 27 Jul 2026 16:12:44 +0530 Subject: [PATCH 1/9] feat(experimentation): deliver buffered events to external ClickHouse warehouses Every 10 minutes, stream each environment's gzipped JSONEachRow event objects from the per-organisation S3 events lake into the customer's ClickHouse over HTTPS, byte for byte. Delivered objects move to archive/, objects the warehouse rejects move to failed/, and connection-level failures abort the run and surface on the connection's status. --- api/experimentation/constants.py | 2 + api/experimentation/metrics.py | 17 + api/experimentation/services.py | 18 + api/experimentation/tasks.py | 121 ++++++- .../warehouse_delivery_service.py | 190 ++++++++++ api/pyproject.toml | 1 + api/tests/unit/experimentation/test_tasks.py | 336 ++++++++++++++++++ .../test_warehouse_delivery_service.py | 255 +++++++++++++ api/uv.lock | 154 ++++++++ .../observability/_metrics-catalogue.md | 18 + 10 files changed, 1108 insertions(+), 4 deletions(-) create mode 100644 api/experimentation/warehouse_delivery_service.py create mode 100644 api/tests/unit/experimentation/test_warehouse_delivery_service.py diff --git a/api/experimentation/constants.py b/api/experimentation/constants.py index fa34d7a7ff97..561ca2718041 100644 --- a/api/experimentation/constants.py +++ b/api/experimentation/constants.py @@ -14,6 +14,8 @@ CONTROL_VARIANT_KEY = "control" +DELIVERY_INTERVAL = timedelta(minutes=10) + # Below these per-variant floors a metric shows "collecting data" rather than # inference; sample-ratio is only checked once there is enough traffic to judge. RESULTS_MIN_IDENTITIES_PER_VARIANT = 50 diff --git a/api/experimentation/metrics.py b/api/experimentation/metrics.py index 8c9d14d1771e..2212de5a9c60 100644 --- a/api/experimentation/metrics.py +++ b/api/experimentation/metrics.py @@ -8,3 +8,20 @@ ["result"], ) ) + +flagsmith_experimentation_warehouse_delivery_runs_total = prometheus_client.Counter( + "flagsmith_experimentation_warehouse_delivery_runs_total", + "Outcomes of per-connection runs delivering buffered event objects to " + "customers' own data warehouses. `result` label is either `success` or " + "`failure`; a failed run delivers nothing and is retried on the next tick.", + ["result"], +) + +flagsmith_experimentation_warehouse_delivery_objects_total = prometheus_client.Counter( + "flagsmith_experimentation_warehouse_delivery_objects_total", + "Buffered S3 event objects processed by delivery to customers' own data " + "warehouses. `result` label is `delivered` for objects inserted and " + "archived, or `rejected` for objects the warehouse refused, which are " + "moved aside and never retried.", + ["result"], +) diff --git a/api/experimentation/services.py b/api/experimentation/services.py index b5b0d4ada6d6..8539e72df3c7 100644 --- a/api/experimentation/services.py +++ b/api/experimentation/services.py @@ -803,6 +803,24 @@ def mark_warehouse_pending_connection( return connection +def mark_warehouse_delivery_failed( + connection: WarehouseConnection, + detail: str, +) -> None: + connection.status = WarehouseConnectionStatus.ERRORED + connection.status_detail = detail[:255] + connection.save(update_fields=["status", "status_detail"]) + + +def mark_warehouse_delivery_succeeded(connection: WarehouseConnection) -> None: + if connection.status == WarehouseConnectionStatus.CONNECTED: + return + + connection.status = WarehouseConnectionStatus.CONNECTED + connection.status_detail = None + connection.save(update_fields=["status", "status_detail"]) + + class InternalAddressError(Exception): pass diff --git a/api/experimentation/tasks.py b/api/experimentation/tasks.py index 0776173901b0..832173723ef6 100644 --- a/api/experimentation/tasks.py +++ b/api/experimentation/tasks.py @@ -1,15 +1,34 @@ import structlog 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 environments.models import Environment, EnvironmentAPIKey -from experimentation import ingestion_sync_service -from experimentation.models import Experiment, ExperimentExposures, ExperimentResults +from experimentation import ingestion_sync_service, warehouse_delivery_service +from experimentation.constants import DELIVERY_INTERVAL +from experimentation.metrics import ( + flagsmith_experimentation_warehouse_delivery_objects_total, + flagsmith_experimentation_warehouse_delivery_runs_total, +) +from experimentation.models import ( + Experiment, + ExperimentExposures, + ExperimentResults, + WarehouseConnection, + WarehouseType, +) from experimentation.organisation_ingestion_service import ( disable_ingestion_for_organisation, enable_ingestion_for_organisation, ) -from experimentation.services import compute_exposures_summary, compute_results_summary +from experimentation.services import ( + compute_exposures_summary, + compute_results_summary, + mark_warehouse_delivery_failed, + mark_warehouse_delivery_succeeded, +) logger = structlog.get_logger("experimentation") @@ -106,6 +125,100 @@ def remove_environment_ingestion_key(key: str) -> None: ingestion_sync_service.delete_ingestion_key(key) +@register_recurring_task(run_every=DELIVERY_INTERVAL) +def deliver_events_to_external_warehouses() -> None: + connection_ids = WarehouseConnection.objects.filter( + warehouse_type=WarehouseType.CLICKHOUSE, + ).values_list("id", flat=True) + for connection_id in connection_ids: + deliver_events_for_connection.delay(kwargs={"connection_id": connection_id}) + + +@register_task_handler() +def deliver_events_for_connection(connection_id: int) -> None: + connection = ( + WarehouseConnection.objects.select_related( + "environment__project__organisation__ingestion_infrastructure", + ) + .filter(id=connection_id) + .first() + ) + if connection is None: + return + + organisation = connection.environment.project.organisation + infrastructure = getattr(organisation, "ingestion_infrastructure", None) + if infrastructure is None or not infrastructure.bucket_name: + return + + log = logger.bind( + environment__id=connection.environment_id, + organisation__id=organisation.id, + ) + bucket_name = infrastructure.bucket_name + pending = warehouse_delivery_service.list_pending_objects( + bucket_name, + environment_key=connection.environment.api_key, + ) + if not pending: + return + + delivered_count = rejected_count = rows_count = 0 + try: + with warehouse_delivery_service.delivery_client(connection) as client: + for s3_key in pending: + try: + rows_count += warehouse_delivery_service.deliver_object( + client, + bucket_name, + s3_key, + ) + except warehouse_delivery_service.ObjectRejectedError: + # This object's contents are the problem; the ones behind + # it are still deliverable. + warehouse_delivery_service.move_object( + bucket_name, + s3_key, + to_prefix=warehouse_delivery_service.FAILED_PREFIX, + ) + rejected_count += 1 + flagsmith_experimentation_warehouse_delivery_objects_total.labels( + result="rejected" + ).inc() + log.warning("warehouse_delivery.object_rejected", exc_info=True) + continue + warehouse_delivery_service.move_object( + bucket_name, + s3_key, + to_prefix=warehouse_delivery_service.ARCHIVE_PREFIX, + ) + delivered_count += 1 + flagsmith_experimentation_warehouse_delivery_objects_total.labels( + result="delivered" + ).inc() + except Exception as exc: + # The warehouse itself is unusable; deliver nothing, leave every + # remaining object in place for the next run, and surface the + # breakage on the connection. + mark_warehouse_delivery_failed(connection, detail=str(exc)) + flagsmith_experimentation_warehouse_delivery_runs_total.labels( + result="failure" + ).inc() + log.error("warehouse_delivery.failed", exc_info=exc) + return + + mark_warehouse_delivery_succeeded(connection) + flagsmith_experimentation_warehouse_delivery_runs_total.labels( + result="success" + ).inc() + log.info( + "warehouse_delivery.completed", + objects__count=delivered_count, + objects__rejected_count=rejected_count, + rows__count=rows_count, + ) + + @register_task_handler() def compute_experiment_exposures(experiment_id: int) -> None: experiment = ( diff --git a/api/experimentation/warehouse_delivery_service.py b/api/experimentation/warehouse_delivery_service.py new file mode 100644 index 000000000000..c3cc9ccf22ff --- /dev/null +++ b/api/experimentation/warehouse_delivery_service.py @@ -0,0 +1,190 @@ +import typing +from contextlib import contextmanager +from functools import lru_cache + +import boto3 +import clickhouse_connect +import structlog +from clickhouse_connect.driver.exceptions import DatabaseError + +from core.network import is_internal_address +from experimentation.ingestion_infra_service import AWS_REGION +from experimentation.types import ClickHouseConfig, ClickHouseCredentials + +if typing.TYPE_CHECKING: + from collections.abc import Iterator + from typing import Any + + from clickhouse_connect.driver.client import Client + + from experimentation.models import WarehouseConnection + +logger = structlog.get_logger("experimentation") + +EVENTS_TABLE_NAME = "events" +EVENTS_FORMAT = "JSONEachRow" +EVENTS_COMPRESSION = "gzip" + +# Objects move between these prefixes as they are delivered, so `events/` holds +# exactly the work still outstanding. `errors/` is not available: Firehose +# writes its own delivery failures there. +PENDING_PREFIX = "events/" +ARCHIVE_PREFIX = "archive/" +FAILED_PREFIX = "failed/" + +# The stored port is the native protocol port used by verification. Delivery +# needs the HTTP interface, which listens on a different port; ClickHouse +# exposes the two as a fixed pair, so it is inferred rather than asked for. +NATIVE_TO_HTTP_PORT = {9440: 8443, 9000: 8123} + +CONNECT_TIMEOUT_SECONDS = 10 +INSERT_TIMEOUT_SECONDS = 300 + + +class DeliveryConfigError(Exception): + """The connection's stored configuration cannot be used for delivery.""" + + +class ObjectRejectedError(Exception): + """The warehouse rejected this object's contents. Other objects are + unaffected, so the object is abandoned rather than retried.""" + + +# ClickHouse error codes raised by the object's own bytes: unparseable records, +# values that do not fit the column, constraint violations. Anything else — +# authentication (516), a missing table (60), an unreachable host — would fail +# every object equally, so it aborts the run instead of abandoning good data. +OBJECT_LEVEL_ERROR_CODES = frozenset( + { + 6, # CANNOT_PARSE_TEXT + 26, # CANNOT_PARSE_QUOTED_STRING + 27, # CANNOT_PARSE_INPUT_ASSERTION_FAILED + 38, # CANNOT_PARSE_DATE + 41, # CANNOT_PARSE_DATETIME + 53, # TYPE_MISMATCH + 69, # ARGUMENT_OUT_OF_BOUND + 72, # CANNOT_PARSE_NUMBER + 117, # INCORRECT_DATA + 469, # VIOLATED_CONSTRAINT + } +) + + +@lru_cache(maxsize=1) +def _get_s3_client() -> "Any": + return boto3.client("s3", region_name=AWS_REGION) + + +def get_pending_prefix(environment_key: str) -> str: + return f"{PENDING_PREFIX}env_key={environment_key}/" + + +def list_pending_objects(bucket_name: str, environment_key: str) -> list[str]: + """Return the keys of an environment's undelivered event objects, oldest + first. + + Delivered objects are moved out of `events/`, so everything under the + prefix is outstanding. The partition path is zero-padded, so the keys S3 + returns in lexicographic order are also in chronological order. + """ + paginator = _get_s3_client().get_paginator("list_objects_v2") + pages = paginator.paginate( + Bucket=bucket_name, + Prefix=get_pending_prefix(environment_key), + ) + return [obj["Key"] for page in pages for obj in page.get("Contents", [])] + + +def move_object(bucket_name: str, s3_key: str, *, to_prefix: str) -> str: + """Move a delivered or abandoned object out of `events/`, preserving its + partition path, and return its new key. + + Copy-then-delete is not atomic: an interrupted move leaves the object in + place, so it is delivered again on the next run. Delivery is therefore + at-least-once, as it already is upstream of here. + """ + destination_key = f"{to_prefix}{s3_key.removeprefix(PENDING_PREFIX)}" + s3 = _get_s3_client() + s3.copy_object( + Bucket=bucket_name, + Key=destination_key, + CopySource={"Bucket": bucket_name, "Key": s3_key}, + ) + s3.delete_object(Bucket=bucket_name, Key=s3_key) + return destination_key + + +@contextmanager +def delivery_client(connection: "WarehouseConnection") -> "Iterator[Client]": + """Yield a ClickHouse HTTP client for a connection, reusable across every + object delivered in one run. + + Raises ``DeliveryConfigError`` if the stored configuration cannot be turned + into a usable client. + """ + config = typing.cast(ClickHouseConfig, connection.config or {}) + credentials = typing.cast(ClickHouseCredentials, connection.credentials or {}) + try: + host = config["host"] + port = config["port"] + database = config["database"] + username = config["username"] + secure = config["secure"] + password = credentials["password"] + except KeyError as exc: + raise DeliveryConfigError("Stored connection details are incomplete.") from exc + + # Re-checked immediately before connecting, as verification does: DNS may + # resolve differently than it did at validation time, and delivery streams + # event data to whatever it connects to. + if is_internal_address(host, include_shared=True): + raise DeliveryConfigError( + "Host must not target internal or private network addresses." + ) + + if (http_port := NATIVE_TO_HTTP_PORT.get(port)) is None: + raise DeliveryConfigError( + f"No HTTP port is known for ClickHouse port {port}; " + f"expected one of {sorted(NATIVE_TO_HTTP_PORT)}." + ) + + client = clickhouse_connect.get_client( + host=host, + port=http_port, + username=username, + password=password, + database=database, + secure=secure, + connect_timeout=CONNECT_TIMEOUT_SECONDS, + send_receive_timeout=INSERT_TIMEOUT_SECONDS, + ) + try: + yield client + finally: + client.close() + + +def deliver_object(client: "Client", bucket_name: str, s3_key: str) -> int: + """Stream one gzipped events object into the customer's warehouse and + return the number of rows written. + + The body is passed through untouched — ClickHouse decompresses and parses + it, so neither happens in this process. The deduplication token is what + absorbs a redelivery, but only on a Replicated table or one configured with + a non-replicated deduplication window; otherwise duplicates reach the table + and the distinct-aware aggregates in the results queries account for them. + """ + body = _get_s3_client().get_object(Bucket=bucket_name, Key=s3_key)["Body"] + try: + summary = client.raw_insert( + EVENTS_TABLE_NAME, + insert_block=body, + fmt=EVENTS_FORMAT, + compression=EVENTS_COMPRESSION, + settings={"insert_deduplication_token": s3_key}, + ) + except DatabaseError as exc: + if exc.code in OBJECT_LEVEL_ERROR_CODES: + raise ObjectRejectedError(str(exc)) from exc + raise + return summary.written_rows diff --git a/api/pyproject.toml b/api/pyproject.toml index af3e37f3effe..bb005ea5b304 100644 --- a/api/pyproject.toml +++ b/api/pyproject.toml @@ -47,6 +47,7 @@ dependencies = [ "flagsmith-sql-flag-engine>=0.2.0,<0.3.0", "django-clickhouse-backend>=1.4,<2.0", "clickhouse-driver==0.2.11", + "clickhouse-connect>=1.6.0,<2.0.0", "boto3>=1.35.95,<1.36.0", "smart-open[s3]>=7.0.0,<8.0.0", "slack-sdk>=3.9.0,<3.10.0", diff --git a/api/tests/unit/experimentation/test_tasks.py b/api/tests/unit/experimentation/test_tasks.py index 07e5e36ccb93..2647ecc6a586 100644 --- a/api/tests/unit/experimentation/test_tasks.py +++ b/api/tests/unit/experimentation/test_tasks.py @@ -1,14 +1,21 @@ +from collections.abc import Iterator from dataclasses import asdict from datetime import datetime, timedelta from datetime import timezone as dt_timezone +from typing import Any +import boto3 import pytest +from clickhouse_connect.driver.exceptions import DatabaseError from django.utils import timezone from freezegun import freeze_time +from moto import mock_s3 # type: ignore[import-untyped] +from prometheus_client import REGISTRY from pytest_mock import MockerFixture from pytest_structlog import StructuredLogCapture from environments.models import Environment, EnvironmentAPIKey +from experimentation import warehouse_delivery_service from experimentation.dataclasses import ( ExposuresSummary, ExposuresTimeseries, @@ -24,11 +31,16 @@ ExperimentStatus, IngestionInfrastructureStatus, OrganisationIngestionInfrastructure, + WarehouseConnection, + WarehouseConnectionStatus, + WarehouseType, ) from experimentation.stats import VariantStats from experimentation.tasks import ( compute_experiment_exposures, compute_experiment_results, + deliver_events_for_connection, + deliver_events_to_external_warehouses, provision_external_warehouse_ingestion_infrastructure, remove_environment_ingestion_key, remove_environment_ingestion_keys, @@ -37,6 +49,7 @@ write_environment_ingestion_keys, ) from organisations.models import Organisation +from projects.models import Project def test_write_environment_ingestion_keys__valid_keys__whitelists_client_and_server( @@ -687,3 +700,326 @@ def test_teardown_organisation_ingestion_infrastructure__created_infrastructure_ assert not OrganisationIngestionInfrastructure.objects.filter( organisation=organisation ).exists() + + +DELIVERY_BUCKET_NAME = "flagsmith-events-lake-org-1-123456789012-eu-west-2-an" + + +def _pending_key(environment_key: str, hour: str = "13") -> str: + return ( + f"events/env_key={environment_key}/year=2026/month=07/day=27/" + f"hour={hour}/object.gz" + ) + + +def _delivery_runs_count(result: str) -> float: + return ( + REGISTRY.get_sample_value( + "flagsmith_experimentation_warehouse_delivery_runs_total", + {"result": result}, + ) + or 0.0 + ) + + +def _delivery_objects_count(result: str) -> float: + return ( + REGISTRY.get_sample_value( + "flagsmith_experimentation_warehouse_delivery_objects_total", + {"result": result}, + ) + or 0.0 + ) + + +@pytest.fixture() +def delivery_bucket(aws_credentials: None) -> Iterator[Any]: + warehouse_delivery_service._get_s3_client.cache_clear() + with mock_s3(): + s3 = boto3.client("s3", region_name="eu-west-2") + s3.create_bucket( + Bucket=DELIVERY_BUCKET_NAME, + CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, + ) + yield s3 + warehouse_delivery_service._get_s3_client.cache_clear() + + +@pytest.fixture() +def ingestion_infrastructure( + organisation: Organisation, +) -> OrganisationIngestionInfrastructure: + return OrganisationIngestionInfrastructure.objects.create( + organisation=organisation, + status=IngestionInfrastructureStatus.CREATED, + bucket_name=DELIVERY_BUCKET_NAME, + stream_name="events-ingestion-org-1", + ) + + +@pytest.fixture() +def warehouse_client(mocker: MockerFixture) -> Any: + get_client = mocker.patch( + "experimentation.warehouse_delivery_service.clickhouse_connect.get_client", + ) + get_client.return_value.raw_insert.return_value.written_rows = 100 + return get_client + + +def test_deliver_events_to_external_warehouses__mixed_connections__enqueues_clickhouse_only( + clickhouse_connection: WarehouseConnection, + project: Project, + mocker: MockerFixture, +) -> None: + # Given flagsmith and soft-deleted clickhouse connections alongside the + # active clickhouse one + flagsmith_environment = Environment.objects.create( + name="Flagsmith Warehouse Environment", project=project + ) + WarehouseConnection.objects.create( + environment=flagsmith_environment, + warehouse_type=WarehouseType.FLAGSMITH, + name="Flagsmith Warehouse", + ) + deleted_environment = Environment.objects.create( + name="Deleted Connection Environment", project=project + ) + WarehouseConnection.objects.create( + environment=deleted_environment, + warehouse_type=WarehouseType.CLICKHOUSE, + name="Deleted ClickHouse", + ).delete() + task = mocker.patch("experimentation.tasks.deliver_events_for_connection") + + # When + deliver_events_to_external_warehouses() + + # Then + task.delay.assert_called_once_with( + kwargs={"connection_id": clickhouse_connection.id}, + ) + + +def test_deliver_events_for_connection__missing_connection__does_nothing( + db: None, + warehouse_client: Any, +) -> None: + # When + deliver_events_for_connection(connection_id=404) + + # Then + warehouse_client.assert_not_called() + + +def test_deliver_events_for_connection__no_infrastructure__does_nothing( + clickhouse_connection: WarehouseConnection, + warehouse_client: Any, +) -> None: + # When + deliver_events_for_connection(connection_id=clickhouse_connection.id) + + # Then + warehouse_client.assert_not_called() + + +def test_deliver_events_for_connection__infrastructure_without_bucket__does_nothing( + clickhouse_connection: WarehouseConnection, + organisation: Organisation, + warehouse_client: Any, +) -> None: + # Given + OrganisationIngestionInfrastructure.objects.create( + organisation=organisation, + status=IngestionInfrastructureStatus.PENDING, + ) + + # When + deliver_events_for_connection(connection_id=clickhouse_connection.id) + + # Then + warehouse_client.assert_not_called() + + +def test_deliver_events_for_connection__no_pending_objects__does_not_connect( + clickhouse_connection: WarehouseConnection, + ingestion_infrastructure: OrganisationIngestionInfrastructure, + delivery_bucket: Any, + warehouse_client: Any, +) -> None: + # When + deliver_events_for_connection(connection_id=clickhouse_connection.id) + + # Then + warehouse_client.assert_not_called() + + +def test_deliver_events_for_connection__pending_objects__delivers_archives_and_recovers_status( + clickhouse_connection: WarehouseConnection, + environment: Environment, + ingestion_infrastructure: OrganisationIngestionInfrastructure, + delivery_bucket: Any, + warehouse_client: Any, + log: StructuredLogCapture, +) -> None: + # Given an errored connection with two objects waiting + clickhouse_connection.status = WarehouseConnectionStatus.ERRORED + clickhouse_connection.status_detail = "Authentication failed" + clickhouse_connection.save() + for hour in ("13", "14"): + delivery_bucket.put_object( + Bucket=DELIVERY_BUCKET_NAME, + Key=_pending_key(environment.api_key, hour=hour), + Body=b"gzipped-events", + ) + success_runs_before = _delivery_runs_count("success") + delivered_objects_before = _delivery_objects_count("delivered") + + # When + deliver_events_for_connection(connection_id=clickhouse_connection.id) + + # Then both objects are inserted and archived, oldest first + raw_insert = warehouse_client.return_value.raw_insert + assert raw_insert.call_count == 2 + assert [ + call.kwargs["settings"]["insert_deduplication_token"] + for call in raw_insert.call_args_list + ] == [ + _pending_key(environment.api_key, hour="13"), + _pending_key(environment.api_key, hour="14"), + ] + archived_keys = [ + item["Key"] + for item in delivery_bucket.list_objects_v2( + Bucket=DELIVERY_BUCKET_NAME, Prefix="archive/" + )["Contents"] + ] + assert len(archived_keys) == 2 + assert ( + warehouse_delivery_service.list_pending_objects( + DELIVERY_BUCKET_NAME, + environment_key=environment.api_key, + ) + == [] + ) + + # Then the delivery success resolves the earlier breakage + clickhouse_connection.refresh_from_db() + assert clickhouse_connection.status == WarehouseConnectionStatus.CONNECTED + assert clickhouse_connection.status_detail is None + + assert _delivery_runs_count("success") == success_runs_before + 1 + assert _delivery_objects_count("delivered") == delivered_objects_before + 2 + assert { + "level": "info", + "event": "warehouse_delivery.completed", + "environment__id": environment.id, + "organisation__id": environment.project.organisation_id, + "objects__count": 2, + "objects__rejected_count": 0, + "rows__count": 200, + } in log.events + + +def test_deliver_events_for_connection__rejected_object__moves_to_failed_and_continues( + clickhouse_connection: WarehouseConnection, + environment: Environment, + ingestion_infrastructure: OrganisationIngestionInfrastructure, + delivery_bucket: Any, + warehouse_client: Any, + log: StructuredLogCapture, +) -> None: + # Given the warehouse rejects the first object's contents but accepts + # the second + clickhouse_connection.status = WarehouseConnectionStatus.CONNECTED + clickhouse_connection.save() + for hour in ("13", "14"): + delivery_bucket.put_object( + Bucket=DELIVERY_BUCKET_NAME, + Key=_pending_key(environment.api_key, hour=hour), + Body=b"gzipped-events", + ) + raw_insert = warehouse_client.return_value.raw_insert + raw_insert.side_effect = [ + DatabaseError("Constraint `event_not_empty` violated", code=469), + raw_insert.return_value, + ] + rejected_objects_before = _delivery_objects_count("rejected") + + # When + deliver_events_for_connection(connection_id=clickhouse_connection.id) + + # Then the rejected object is set aside and the run completes + failed_keys = [ + item["Key"] + for item in delivery_bucket.list_objects_v2( + Bucket=DELIVERY_BUCKET_NAME, Prefix="failed/" + )["Contents"] + ] + assert failed_keys == [ + f"failed/env_key={environment.api_key}/year=2026/month=07/day=27/" + f"hour=13/object.gz" + ] + archived_keys = [ + item["Key"] + for item in delivery_bucket.list_objects_v2( + Bucket=DELIVERY_BUCKET_NAME, Prefix="archive/" + )["Contents"] + ] + assert archived_keys == [ + f"archive/env_key={environment.api_key}/year=2026/month=07/day=27/" + f"hour=14/object.gz" + ] + clickhouse_connection.refresh_from_db() + assert clickhouse_connection.status == WarehouseConnectionStatus.CONNECTED + assert _delivery_objects_count("rejected") == rejected_objects_before + 1 + assert log.has("warehouse_delivery.object_rejected", level="warning") + assert { + "level": "info", + "event": "warehouse_delivery.completed", + "environment__id": environment.id, + "organisation__id": environment.project.organisation_id, + "objects__count": 1, + "objects__rejected_count": 1, + "rows__count": 100, + } in log.events + + +def test_deliver_events_for_connection__warehouse_unusable__aborts_and_marks_errored( + clickhouse_connection: WarehouseConnection, + environment: Environment, + ingestion_infrastructure: OrganisationIngestionInfrastructure, + delivery_bucket: Any, + warehouse_client: Any, + log: StructuredLogCapture, +) -> None: + # Given the warehouse rejects every request + for hour in ("13", "14"): + delivery_bucket.put_object( + Bucket=DELIVERY_BUCKET_NAME, + Key=_pending_key(environment.api_key, hour=hour), + Body=b"gzipped-events", + ) + warehouse_client.return_value.raw_insert.side_effect = DatabaseError( + "Authentication failed", + code=516, + ) + failure_runs_before = _delivery_runs_count("failure") + + # When + deliver_events_for_connection(connection_id=clickhouse_connection.id) + + # Then nothing is moved: every object waits for the next run + assert warehouse_delivery_service.list_pending_objects( + DELIVERY_BUCKET_NAME, + environment_key=environment.api_key, + ) == [ + _pending_key(environment.api_key, hour="13"), + _pending_key(environment.api_key, hour="14"), + ] + + # Then the breakage is surfaced on the connection + clickhouse_connection.refresh_from_db() + assert clickhouse_connection.status == WarehouseConnectionStatus.ERRORED + assert clickhouse_connection.status_detail == "Authentication failed" + assert _delivery_runs_count("failure") == failure_runs_before + 1 + assert log.has("warehouse_delivery.failed", level="error") diff --git a/api/tests/unit/experimentation/test_warehouse_delivery_service.py b/api/tests/unit/experimentation/test_warehouse_delivery_service.py new file mode 100644 index 000000000000..ffaa80c758fe --- /dev/null +++ b/api/tests/unit/experimentation/test_warehouse_delivery_service.py @@ -0,0 +1,255 @@ +import gzip +from collections.abc import Iterator +from typing import Any + +import boto3 +import pytest +from clickhouse_connect.driver.exceptions import DatabaseError +from moto import mock_s3 # type: ignore[import-untyped] +from pytest_mock import MockerFixture + +from experimentation import warehouse_delivery_service +from experimentation.models import WarehouseConnection + +BUCKET_NAME = "flagsmith-events-lake-org-42-123456789012-eu-west-2-an" +ENVIRONMENT_KEY = "delivery_env_key" +OBJECT_BODY = gzip.compress( + b'{"environment_key":"delivery_env_key","event":"$flag_exposure",' + b'"timestamp":1753000000000}\n' +) + + +def _event_key(hour: str, environment_key: str = ENVIRONMENT_KEY) -> str: + return ( + f"events/env_key={environment_key}/year=2026/month=07/day=27/" + f"hour={hour}/object.gz" + ) + + +@pytest.fixture() +def aws_backends(aws_credentials: None) -> Iterator[None]: + warehouse_delivery_service._get_s3_client.cache_clear() + with mock_s3(): + yield + warehouse_delivery_service._get_s3_client.cache_clear() + + +@pytest.fixture() +def events_bucket(aws_backends: None) -> Any: + s3 = boto3.client("s3", region_name="eu-west-2") + s3.create_bucket( + Bucket=BUCKET_NAME, + CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, + ) + return s3 + + +def test_list_pending_objects__objects_across_hours__returns_chronological_and_scoped( + events_bucket: Any, +) -> None: + # Given objects for two hours, put out of order, plus another + # environment's object in the same bucket + events_bucket.put_object(Bucket=BUCKET_NAME, Key=_event_key("14"), Body=OBJECT_BODY) + events_bucket.put_object(Bucket=BUCKET_NAME, Key=_event_key("13"), Body=OBJECT_BODY) + events_bucket.put_object( + Bucket=BUCKET_NAME, + Key=_event_key("13", environment_key="other_env_key"), + Body=OBJECT_BODY, + ) + + # When + pending = warehouse_delivery_service.list_pending_objects( + BUCKET_NAME, + environment_key=ENVIRONMENT_KEY, + ) + + # Then + assert pending == [_event_key("13"), _event_key("14")] + + +def test_list_pending_objects__no_objects__returns_empty( + events_bucket: Any, +) -> None: + # When + pending = warehouse_delivery_service.list_pending_objects( + BUCKET_NAME, + environment_key=ENVIRONMENT_KEY, + ) + + # Then + assert pending == [] + + +def test_move_object__to_archive__moves_and_preserves_partition_path( + events_bucket: Any, +) -> None: + # Given + events_bucket.put_object(Bucket=BUCKET_NAME, Key=_event_key("13"), Body=OBJECT_BODY) + + # When + destination_key = warehouse_delivery_service.move_object( + BUCKET_NAME, + _event_key("13"), + to_prefix=warehouse_delivery_service.ARCHIVE_PREFIX, + ) + + # Then + assert destination_key == ( + f"archive/env_key={ENVIRONMENT_KEY}/year=2026/month=07/day=27/hour=13/object.gz" + ) + archived = events_bucket.get_object(Bucket=BUCKET_NAME, Key=destination_key) + assert archived["Body"].read() == OBJECT_BODY + assert ( + warehouse_delivery_service.list_pending_objects( + BUCKET_NAME, + environment_key=ENVIRONMENT_KEY, + ) + == [] + ) + + +def test_delivery_client__incomplete_config__raises_config_error( + clickhouse_connection: WarehouseConnection, +) -> None: + # Given + clickhouse_connection.credentials = None + + # When / Then + with pytest.raises( + warehouse_delivery_service.DeliveryConfigError, + match="incomplete", + ): + with warehouse_delivery_service.delivery_client(clickhouse_connection): + pass # pragma: no cover + + +def test_delivery_client__internal_host__raises_config_error( + clickhouse_connection: WarehouseConnection, +) -> None: + # Given + clickhouse_connection.config["host"] = "10.13.37.1" # type: ignore[index] + + # When / Then + with pytest.raises( + warehouse_delivery_service.DeliveryConfigError, + match="internal or private", + ): + with warehouse_delivery_service.delivery_client(clickhouse_connection): + pass # pragma: no cover + + +def test_delivery_client__unmappable_port__raises_config_error( + clickhouse_connection: WarehouseConnection, +) -> None: + # Given + clickhouse_connection.config["port"] = 1234 # type: ignore[index] + + # When / Then + with pytest.raises( + warehouse_delivery_service.DeliveryConfigError, + match="No HTTP port is known for ClickHouse port 1234", + ): + with warehouse_delivery_service.delivery_client(clickhouse_connection): + pass # pragma: no cover + + +def test_delivery_client__valid_config__yields_http_client_and_closes( + clickhouse_connection: WarehouseConnection, + mocker: MockerFixture, +) -> None: + # Given + get_client = mocker.patch( + "experimentation.warehouse_delivery_service.clickhouse_connect.get_client", + ) + + # When + with warehouse_delivery_service.delivery_client(clickhouse_connection) as client: + # Then the native port is mapped to its HTTP counterpart + assert client is get_client.return_value + get_client.assert_called_once_with( + host="ch.acme-corp.example", + port=8443, + username="acme_svc", + password="hunter2", + database="acme_dwh", + secure=True, + connect_timeout=10, + send_receive_timeout=300, + ) + get_client.return_value.close.assert_not_called() + + get_client.return_value.close.assert_called_once_with() + + +def test_deliver_object__valid_object__streams_body_and_returns_written_rows( + events_bucket: Any, + mocker: MockerFixture, +) -> None: + # Given + events_bucket.put_object(Bucket=BUCKET_NAME, Key=_event_key("13"), Body=OBJECT_BODY) + client = mocker.MagicMock() + client.raw_insert.return_value.written_rows = 203 + + # When + written_rows = warehouse_delivery_service.deliver_object( + client, + BUCKET_NAME, + _event_key("13"), + ) + + # Then + assert written_rows == 203 + client.raw_insert.assert_called_once_with( + "events", + insert_block=mocker.ANY, + fmt="JSONEachRow", + compression="gzip", + settings={"insert_deduplication_token": _event_key("13")}, + ) + # The body reaches ClickHouse as the exact bytes stored in S3 + insert_block = client.raw_insert.call_args.kwargs["insert_block"] + assert insert_block.read() == OBJECT_BODY + + +def test_deliver_object__object_level_error__raises_object_rejected( + events_bucket: Any, + mocker: MockerFixture, +) -> None: + # Given a warehouse that rejects the object's contents (469 = + # VIOLATED_CONSTRAINT) + events_bucket.put_object(Bucket=BUCKET_NAME, Key=_event_key("13"), Body=OBJECT_BODY) + client = mocker.MagicMock() + client.raw_insert.side_effect = DatabaseError( + "Constraint `event_not_empty` violated", + code=469, + ) + + # When / Then + with pytest.raises(warehouse_delivery_service.ObjectRejectedError): + warehouse_delivery_service.deliver_object( + client, + BUCKET_NAME, + _event_key("13"), + ) + + +def test_deliver_object__connection_level_error__reraises( + events_bucket: Any, + mocker: MockerFixture, +) -> None: + # Given a warehouse failure unrelated to the object's contents (516 = + # AUTHENTICATION_FAILED) + events_bucket.put_object(Bucket=BUCKET_NAME, Key=_event_key("13"), Body=OBJECT_BODY) + client = mocker.MagicMock() + client.raw_insert.side_effect = DatabaseError( + "Authentication failed", + code=516, + ) + + # When / Then + with pytest.raises(DatabaseError): + warehouse_delivery_service.deliver_object( + client, + BUCKET_NAME, + _event_key("13"), + ) diff --git a/api/uv.lock b/api/uv.lock index 0518952c041c..71fa69af4063 100644 --- a/api/uv.lock +++ b/api/uv.lock @@ -388,6 +388,76 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/df/73/b6e24bd22e6720ca8ee9a85a0c4a2971af8497d8f3193fa05390cbd46e09/backoff-2.2.1-py3-none-any.whl", hash = "sha256:63579f9a0628e06278f7e47b7d7d5b6ce20dc65c5e96a6f3ca99a6adca0396e8", size = 15148, upload-time = "2022-10-05T19:19:30.546Z" }, ] +[[package]] +name = "backports-zstd" +version = "1.6.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/8e/b5/5a873da082bd08acd6a497f7aae224e94a7c27fa8f24488089cc50a16c84/backports_zstd-1.6.0.tar.gz", hash = "sha256:80a7859ffe70bf239d7a2ce15293bdeb5b4280ff7dc326ffab312b0e254dbb24", size = 1000009, upload-time = "2026-06-14T10:50:58.555Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/c5/90/428dd82228b1b6d62d5a1bf312c29e6c125af6a182fcfd82768ca179dcc7/backports_zstd-1.6.0-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:c4fc41b2df5529cad5ceb230319e82728096d4b353ce8d4df68a2ec37e291bb8", size = 437067, upload-time = "2026-06-14T10:49:28.335Z" }, + { url = "https://files.pythonhosted.org/packages/ef/48/768edf21fe33bae8d874470b1be136681d4d32eb820a32e1c98262ebe39b/backports_zstd-1.6.0-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:83391ef5935cc0f329b1abca414ae20ffe40d335fc21a4b5e664f08a74317d5f", size = 363454, upload-time = "2026-06-14T10:49:29.784Z" }, + { url = "https://files.pythonhosted.org/packages/29/8a/d462c2e5071eb573378f0d26760f6590613086fdf59c2d3c66bdfffb9f41/backports_zstd-1.6.0-cp311-cp311-manylinux2010_i686.manylinux_2_12_i686.manylinux_2_28_i686.whl", hash = "sha256:7d3f64c503af7b60115b97c16feaf75bd191ef2c978d5c0c7725a6682bef63c5", size = 507393, upload-time = "2026-06-14T10:49:31.077Z" }, + { url = "https://files.pythonhosted.org/packages/b9/cb/af58363b0dd0b497282ecef1fa99789b03cc1885a01a41394cad42ceeff6/backports_zstd-1.6.0-cp311-cp311-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:0308990ffc998df3c7ed35276bde049728b5c3956203cae40d80893576a41459", size = 476957, upload-time = "2026-06-14T10:49:32.53Z" }, + { url = "https://files.pythonhosted.org/packages/e4/fd/5fbdf2275cefae95c4b3509f6db2dc372d0587ebafea342d28781d51d932/backports_zstd-1.6.0-cp311-cp311-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:8c298785e2fadeab82342040f2d9ce764ce500e6da6a6d99a2de514e63580b5a", size = 582618, upload-time = "2026-06-14T10:49:33.723Z" }, + { url = "https://files.pythonhosted.org/packages/99/6f/7dd45c53c907ea67f635c3900b58bb3347c01dc2ded441402028aae0ef9c/backports_zstd-1.6.0-cp311-cp311-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:ae106fe16e36efc60ab098d02478d30aa0e31e1420eb4ecf0116459253bc6361", size = 642279, upload-time = "2026-06-14T10:49:34.938Z" }, + { url = "https://files.pythonhosted.org/packages/4d/25/a9e37dd035027565fa0b7e367da50e88a6ab26e7fd413269aa118e25258b/backports_zstd-1.6.0-cp311-cp311-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:7293fefe15f0e5852bdb4ad1e0e26f3cbd4d3e61c19f751ecc4ff34bc1eb237d", size = 492486, upload-time = "2026-06-14T10:49:36.06Z" }, + { url = "https://files.pythonhosted.org/packages/a1/52/659686bf8f7c53ea279e1c44038504b82a6901cee2f5ae83c30bbf581301/backports_zstd-1.6.0-cp311-cp311-manylinux_2_34_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:ece8e7288db5b827ef8c64b2f78519f1a173a8991a625978fce02eccd7654fe9", size = 566440, upload-time = "2026-06-14T10:49:37.536Z" }, + { url = "https://files.pythonhosted.org/packages/d9/1a/c7ea5a0ff607a1a6066bb7c7cb65ae20e2f85da6adc69ab77fd8943e180c/backports_zstd-1.6.0-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:28eef3881164f3c23ce58ed59e4684103bdd279583eb2d299858c9e9b72fde9a", size = 482899, upload-time = "2026-06-14T10:49:38.805Z" }, + { url = "https://files.pythonhosted.org/packages/83/48/bd2b91100ee4fe6bb4d816e3659cbbb0cda5dd32760d2379c54d1752ec25/backports_zstd-1.6.0-cp311-cp311-musllinux_1_2_i686.whl", hash = "sha256:481a1e9bd8f419fdc625307aa20234687f99368c75df511ef589693c5fea4c6f", size = 510826, upload-time = "2026-06-14T10:49:40.062Z" }, + { url = "https://files.pythonhosted.org/packages/25/fe/fa28509d7ce2ad59404e7ce738a2fd858e12dfd9a896629f10330222a7fb/backports_zstd-1.6.0-cp311-cp311-musllinux_1_2_ppc64le.whl", hash = "sha256:3b6713371f8987a1178df93cb36f29eef191f224021e2d656b2f11ce60d26816", size = 586941, upload-time = "2026-06-14T10:49:41.305Z" }, + { url = "https://files.pythonhosted.org/packages/45/28/757daf2399aa71bb37f9f7f48b42ab03fc51c340eccfad2fec92a23f6aa3/backports_zstd-1.6.0-cp311-cp311-musllinux_1_2_riscv64.whl", hash = "sha256:b0ddbcd2866b8ff1a2836e4b0e4d44788f5b992d83fac75a38cda8f9a2bee079", size = 564261, upload-time = "2026-06-14T10:49:42.49Z" }, + { url = "https://files.pythonhosted.org/packages/4e/53/9b9db30cb2c148a69c40ad7647aa787338041f3dc81c5b22113286e590e9/backports_zstd-1.6.0-cp311-cp311-musllinux_1_2_s390x.whl", hash = "sha256:2914abea516704bdafb2090acd3f15b5f9debecfabd15b8dd8285b2ad3b92209", size = 632869, upload-time = "2026-06-14T10:49:43.981Z" }, + { url = "https://files.pythonhosted.org/packages/81/a4/1692fbb88af8aaf900a53619fcc95c9e45d9ff162223a47fd672a9893c8d/backports_zstd-1.6.0-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:dd085eafa2aac6f883afd28210a3231f717f25409a1e44a39bb7b04c8c5b5646", size = 496496, upload-time = "2026-06-14T10:49:45.118Z" }, + { url = "https://files.pythonhosted.org/packages/93/42/c5a66c47320bd12ce84a7341330ea582d67069bdb70214bca0b6bf394cfd/backports_zstd-1.6.0-cp311-cp311-win32.whl", hash = "sha256:b81b4cf3d6e0ad7ac92bef248f49fafc954262c5fb0f7e19d6aac497e5a856b2", size = 291613, upload-time = "2026-06-14T10:49:46.473Z" }, + { url = "https://files.pythonhosted.org/packages/2a/f2/f22c19b4cdde429805ff5ac8dd77a95569a7c4cb8991741b2ff0d538f220/backports_zstd-1.6.0-cp311-cp311-win_amd64.whl", hash = "sha256:10b61850c4112952e05aa6e6cce8c9a5936fbeadb321e154216705cc76a14afa", size = 329078, upload-time = "2026-06-14T10:49:47.71Z" }, + { url = "https://files.pythonhosted.org/packages/ef/dc/e902a3f1eb92c4907b5f47f90cb3c2734ee315c4ff67179fc111343b45ba/backports_zstd-1.6.0-cp311-cp311-win_arm64.whl", hash = "sha256:068ef3d8c18815a2e3a752f766313e19910e7c50939b956923748d9c04ebcb1b", size = 291727, upload-time = "2026-06-14T10:49:48.929Z" }, + { url = "https://files.pythonhosted.org/packages/1e/bb/009af3a9532d4cc66d5385391c512210fae32ab2442605f26aca1d8d2957/backports_zstd-1.6.0-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:0466b14723f3b7697669c00ee66fe16e30e25636b286b0a923fa86fa3d8a753c", size = 437407, upload-time = "2026-06-14T10:49:50.155Z" }, + { url = "https://files.pythonhosted.org/packages/0c/76/f7c02efde81ebb9993586f9e435d2fd1191a6f806f640e4eeb8d004493ed/backports_zstd-1.6.0-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:1d146926e997d2d3de8212bdcbf4985344a2622ca3bec458d8908000a84fd883", size = 363519, upload-time = "2026-06-14T10:49:51.383Z" }, + { url = "https://files.pythonhosted.org/packages/2e/5e/0cf66f12472fe3e082cc4134395a7e0b8746cfb30aabd74251ce8fafa9a7/backports_zstd-1.6.0-cp312-cp312-manylinux2010_i686.manylinux_2_12_i686.manylinux_2_28_i686.whl", hash = "sha256:460fd6b3f338c659507ae36cfd6b58ac9942a2ff233c5cf574416dfec0451a84", size = 507756, upload-time = "2026-06-14T10:49:52.497Z" }, + { url = "https://files.pythonhosted.org/packages/03/95/7ed25c90369360f96f8bfa961540845e063377c32a43b775201af66a588c/backports_zstd-1.6.0-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:7c2b1f4a640c51130caa92cef5bf72bd3c3dbbcfbf814c37403aa0601b1811b0", size = 477578, upload-time = "2026-06-14T10:49:53.887Z" }, + { url = "https://files.pythonhosted.org/packages/e3/75/f16b1d3e33ca396525847c81d96e3de7bc74d2c6f9ca2ddee76b0c450697/backports_zstd-1.6.0-cp312-cp312-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:beb43e9885202c8d4f3762319ed4d5e98e197622afbff8439fbbdd81d08938b9", size = 583029, upload-time = "2026-06-14T10:49:55.132Z" }, + { url = "https://files.pythonhosted.org/packages/e1/2b/a17b111b631e1c79a0e570881c1a266c661b936585afa395435a458b1991/backports_zstd-1.6.0-cp312-cp312-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:fbb746522ebfc11155f1cd688e2c48ef3d74125e38b63eabdaab068a055c3e88", size = 641741, upload-time = "2026-06-14T10:49:56.42Z" }, + { url = "https://files.pythonhosted.org/packages/6b/b2/d17b2722c636d64b4e77ddc68d8d0625719d39f94021be8719a218af4c0a/backports_zstd-1.6.0-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:1a99710fbb225d459d66def4dc2bb2cd4a9a0bdc8b799fc0621cfdd863be9c93", size = 495554, upload-time = "2026-06-14T10:49:57.652Z" }, + { url = "https://files.pythonhosted.org/packages/63/12/2853e8b6c03f03795b6548ea61f82cc104d4f7ff2523a04bc69f46984663/backports_zstd-1.6.0-cp312-cp312-manylinux_2_34_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:f69365ee2b836939137de024a302395a1cb8654fb6dc5ffef6381105259c8f87", size = 570027, upload-time = "2026-06-14T10:49:59.003Z" }, + { url = "https://files.pythonhosted.org/packages/18/aa/83f37b81f3b8c6ea035bf260ec374648bd59372894c02323dc9de3cbdf77/backports_zstd-1.6.0-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:66cf8038893c7708ec345ffb3ac63c775d10f430f323ac2f0334fdb6a397c57c", size = 483594, upload-time = "2026-06-14T10:50:00.49Z" }, + { url = "https://files.pythonhosted.org/packages/f5/6a/d77f8cd2ff642d3b3652c1ccab5b6583114dbf10f8cb0143531357c83998/backports_zstd-1.6.0-cp312-cp312-musllinux_1_2_i686.whl", hash = "sha256:e514c71ca72f3b56bd8fbda1a6a5b7d1100a2764b42a3c74a38841f25f9b00ab", size = 511206, upload-time = "2026-06-14T10:50:01.86Z" }, + { url = "https://files.pythonhosted.org/packages/56/b2/99a60fe4d1aac8053769d2463271d5df37a7c11c387072fdbb0b16aed7f7/backports_zstd-1.6.0-cp312-cp312-musllinux_1_2_ppc64le.whl", hash = "sha256:7741e44f7938ec94f9a52678c8d19b7bc548522ffdc39c9e4481af8db545fa9a", size = 587416, upload-time = "2026-06-14T10:50:03.236Z" }, + { url = "https://files.pythonhosted.org/packages/ec/1e/a9c003fe4d14bd4bf671598d4c7dcc1cef51e3513d9d7111ba1d07b6f07b/backports_zstd-1.6.0-cp312-cp312-musllinux_1_2_riscv64.whl", hash = "sha256:97e8a9674652496c7612b528085dd5a296c052a2edc466ca1bfb7b0b27820413", size = 567615, upload-time = "2026-06-14T10:50:04.524Z" }, + { url = "https://files.pythonhosted.org/packages/ec/b9/955bd604f692c550c7cb66d00bd7691ead5c86df8ebd23d7254eeaa90789/backports_zstd-1.6.0-cp312-cp312-musllinux_1_2_s390x.whl", hash = "sha256:23a793f2fed4dbf0517319759a2cded0b0dd8e8d3797fe30badd5693e320c175", size = 632269, upload-time = "2026-06-14T10:50:05.86Z" }, + { url = "https://files.pythonhosted.org/packages/18/d7/9f61f612f8a4193484c78a1f26db82a50141234189885113ef0085a8a961/backports_zstd-1.6.0-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:b951113113ed4b8d173418a4f155c14b739dace626b3fa3f82be1831958d39e4", size = 500066, upload-time = "2026-06-14T10:50:07.446Z" }, + { url = "https://files.pythonhosted.org/packages/81/a3/19fb8c48d94139481c5ccaf2fb54c31b543fa635fd7bd7399aadd15752ac/backports_zstd-1.6.0-cp312-cp312-win32.whl", hash = "sha256:6430b34a2ae6fcc604672f4f913102563473d9a015bdca1ce8c95041cc1f2677", size = 291825, upload-time = "2026-06-14T10:50:08.762Z" }, + { url = "https://files.pythonhosted.org/packages/58/38/40ba081c6c71f0f22c64d3d54b912ad75a4e6812caa1397cbb15b5693b12/backports_zstd-1.6.0-cp312-cp312-win_amd64.whl", hash = "sha256:08793876172551a930ce4d65c712cd516184d1a97070d4a1193e05bf0cf7040d", size = 329201, upload-time = "2026-06-14T10:50:09.979Z" }, + { url = "https://files.pythonhosted.org/packages/2b/6c/f7116dd2edc6f960545f0d8616939eae3a20031b3b6669697d4f9fd83b2e/backports_zstd-1.6.0-cp312-cp312-win_arm64.whl", hash = "sha256:03b7c59c71f7a597e2bcb3f8368371e9a660a1bdf1c37afc1f1ad1496a013c19", size = 291901, upload-time = "2026-06-14T10:50:11.198Z" }, + { url = "https://files.pythonhosted.org/packages/38/06/c430537d59c55d49bcd15ecf4b1aa965453219caad810a4f2b484816f4be/backports_zstd-1.6.0-cp313-cp313-android_24_arm64_v8a.whl", hash = "sha256:2ace939e4d620e119423606f2d3d7115f8707733bf57f279ad9a9383f875986f", size = 400327, upload-time = "2026-06-14T10:50:12.446Z" }, + { url = "https://files.pythonhosted.org/packages/36/48/2f8323bb0e3ebba88b54877a2979afeb83983fb2ca572f09ad61aae2d3a0/backports_zstd-1.6.0-cp313-cp313-android_24_x86_64.whl", hash = "sha256:4c68a9ed2df0cca51d774c521e68a34d2e3d9ebfc687ef8096adfd4f345b551d", size = 454276, upload-time = "2026-06-14T10:50:13.667Z" }, + { url = "https://files.pythonhosted.org/packages/7c/39/87a665244a65f5b87a06b848c29a8cce07e91d59c5988ee2a32c0293a21c/backports_zstd-1.6.0-cp313-cp313-ios_13_0_arm64_iphoneos.whl", hash = "sha256:30576f49b82328ec8af16c11100efe52ca88526f71bbe100ef6b4e707dc13bf2", size = 357457, upload-time = "2026-06-14T10:50:14.906Z" }, + { url = "https://files.pythonhosted.org/packages/7f/8b/854d4a47bb8b7a48bfb2ed381c7b03a70efb4fc49f0e4a1509b38a2e1727/backports_zstd-1.6.0-cp313-cp313-ios_13_0_arm64_iphonesimulator.whl", hash = "sha256:b4bddfcfb6679215d6f4dc5f79a1f9301af339480d70527a14b57a1f2e6b6cbf", size = 366139, upload-time = "2026-06-14T10:50:16.399Z" }, + { url = "https://files.pythonhosted.org/packages/8f/de/c3af43eb8df6f2581e157e18a3e0121eadb826055b2fde3f91ec188689cb/backports_zstd-1.6.0-cp313-cp313-ios_13_0_x86_64_iphonesimulator.whl", hash = "sha256:65048ed08c5124f05ff9f355ab9703014bb2dbe7f8d9948ce193685b1775f442", size = 446683, upload-time = "2026-06-14T10:50:17.633Z" }, + { url = "https://files.pythonhosted.org/packages/5c/39/87cf3d883d386c10ac52f5322604fb9afdd204229f4c47d4a820a839b8ff/backports_zstd-1.6.0-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:5918fc6b31437208721276964323933cd86077b8d5b469c59c1b3fd2c8220a05", size = 436869, upload-time = "2026-06-14T10:50:19.113Z" }, + { url = "https://files.pythonhosted.org/packages/5e/b6/9479e6f0f18824ad38e8d7dd85161ab0842a198be669421232925bb30960/backports_zstd-1.6.0-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:4b6c8b02ab0ccb2431bb7bc238be91d158b308915e7b07937388e540466fe7e7", size = 363090, upload-time = "2026-06-14T10:50:20.302Z" }, + { url = "https://files.pythonhosted.org/packages/d9/74/a5e98fe108e17c91d9bc590a19e77f5d47d579e34d3f5bc098a949d6c27c/backports_zstd-1.6.0-cp313-cp313-manylinux2010_i686.manylinux_2_12_i686.manylinux_2_28_i686.whl", hash = "sha256:711e6b98f8924e8b4a61ff97ab6321f33de024e1ed6a32f5123763aeda8459be", size = 507070, upload-time = "2026-06-14T10:50:21.536Z" }, + { url = "https://files.pythonhosted.org/packages/69/f5/392bb7dce7363b77bc5403060f418fad438b9cfdd3edd10d65cee7d8fd11/backports_zstd-1.6.0-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:2ba9ac10fc393e5123a08802e0e895a107cb4a66b9973d2844dbd8a343111e59", size = 477200, upload-time = "2026-06-14T10:50:22.91Z" }, + { url = "https://files.pythonhosted.org/packages/e4/4d/dfb665806ba4f74bc48071d32006843b53568c4a17ff627a3061de5eaa09/backports_zstd-1.6.0-cp313-cp313-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:2f723219335387d7546412d8141e0303590600949b4184a1391a0c6a3c756058", size = 582724, upload-time = "2026-06-14T10:50:24.28Z" }, + { url = "https://files.pythonhosted.org/packages/57/b2/beeca7393a8310debd82ee2f0ce5c1801e8d7cb673f7f226f4a0866ca238/backports_zstd-1.6.0-cp313-cp313-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:64b94d7a836568926a3309ff510c7f8261b881b341fd4992cabf4f0998878f8a", size = 643493, upload-time = "2026-06-14T10:50:25.736Z" }, + { url = "https://files.pythonhosted.org/packages/38/26/ce90e9eed6f25aaa4a4fa305a2aaf2d2ad81fd69de8eb248ddd91c80d1e0/backports_zstd-1.6.0-cp313-cp313-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:e39258a09b1c7ca70b5e94a5c5ccfe4700b4250b8077cfeab31d0f79565d4c9b", size = 492190, upload-time = "2026-06-14T10:50:27.205Z" }, + { url = "https://files.pythonhosted.org/packages/17/9b/37b9b146df1f5452419a96071a7017cbac212ec9b137d7a88ca46dc2aa9e/backports_zstd-1.6.0-cp313-cp313-manylinux_2_34_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:15b1aae0f64cd742df4bba1d989d0a09a6ec619202543fdba684640454541fd3", size = 567432, upload-time = "2026-06-14T10:50:28.386Z" }, + { url = "https://files.pythonhosted.org/packages/06/66/81b30991be83237529f36335ac3682bce26409064b906ac6122874575196/backports_zstd-1.6.0-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:25b5ddc789480072551af571a746e9500356b2aff0499861cf2ca07ea7431e68", size = 483021, upload-time = "2026-06-14T10:50:29.654Z" }, + { url = "https://files.pythonhosted.org/packages/49/2a/792c65dcc1e45eb0c1bdc012ee94b84867186bfe27a860d0813bd216f03b/backports_zstd-1.6.0-cp313-cp313-musllinux_1_2_i686.whl", hash = "sha256:a13cfa3410a75e4cb87abdb669aaf79da861cb79299159054ff8f77b9671bc40", size = 510596, upload-time = "2026-06-14T10:50:31.657Z" }, + { url = "https://files.pythonhosted.org/packages/1d/22/01b92a600505620e4cb5f20429e181f30458b7207ca8b52ca5ca6068c35f/backports_zstd-1.6.0-cp313-cp313-musllinux_1_2_ppc64le.whl", hash = "sha256:2ddab55a5f54dec8acfad68ef70f1c704fd21919990ddc238afbd6f496e61c6a", size = 587143, upload-time = "2026-06-14T10:50:32.868Z" }, + { url = "https://files.pythonhosted.org/packages/d8/60/4672f5110b9eb01388cc6225a739e3a5fcd749a63a9c4c1450a04fa27113/backports_zstd-1.6.0-cp313-cp313-musllinux_1_2_riscv64.whl", hash = "sha256:fa305a84087e10d7a85e8a8a3dcba8cdbda4868f2180173b264b7b488fd37c55", size = 565238, upload-time = "2026-06-14T10:50:34.173Z" }, + { url = "https://files.pythonhosted.org/packages/5c/3b/19928d60ea7d25820bf12ef88de74534ca85b56ff7cf13c1b0e74e3a3d7c/backports_zstd-1.6.0-cp313-cp313-musllinux_1_2_s390x.whl", hash = "sha256:df27b57d214a3124fbe4e933ef5a903d4567f154260d9aece8c797a987f2a205", size = 633970, upload-time = "2026-06-14T10:50:35.506Z" }, + { url = "https://files.pythonhosted.org/packages/df/97/c4cecb3e0ff53563ef9819f0395d919ceaae9c5147392ac23bac7afdb20f/backports_zstd-1.6.0-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:28fecd73459d74910ae1987ab84b7bef690d3dd860948430dd5555108b006daf", size = 496539, upload-time = "2026-06-14T10:50:37.015Z" }, + { url = "https://files.pythonhosted.org/packages/eb/f4/46b2f29d2938a80e56e61a19f11ab093f531a9f8cd0ec8eeaac1246bcd99/backports_zstd-1.6.0-cp313-cp313-win32.whl", hash = "sha256:3e689af303df287142770abe3a48bbefd24dab4a09da5807d0e1fa8c75bab026", size = 291451, upload-time = "2026-06-14T10:50:38.518Z" }, + { url = "https://files.pythonhosted.org/packages/d1/ad/b529f92166da61f496621345f95d2dc583c8ca5ac553c084a4ef6c12cd71/backports_zstd-1.6.0-cp313-cp313-win_amd64.whl", hash = "sha256:b067b1ef9c8e41fb0882c828aa37829938b5c0dab067eca72b23fc24c563b9da", size = 329023, upload-time = "2026-06-14T10:50:39.742Z" }, + { url = "https://files.pythonhosted.org/packages/30/d8/6be904d20345fbebec583ca83676e01f30c76118b283eb666d8ec8291ca1/backports_zstd-1.6.0-cp313-cp313-win_arm64.whl", hash = "sha256:a838296f5b84c920172fb579cac894d255c1fc25457c7234613ddcfa385e49b7", size = 291636, upload-time = "2026-06-14T10:50:41.004Z" }, + { url = "https://files.pythonhosted.org/packages/e8/09/898fe2f8196fa7ab825f5fed786c68581fdac7d23a8e20baa0cc01cb2f0b/backports_zstd-1.6.0-pp311-pypy311_pp73-macosx_10_15_x86_64.whl", hash = "sha256:aeef8563b82ed4af328f98e5041c1b4800d86f68f857ffd1577d4d47dc9aa6cd", size = 411023, upload-time = "2026-06-14T10:50:50.286Z" }, + { url = "https://files.pythonhosted.org/packages/6e/ad/6ad9af1596ab5f284bb53954be41396e13d23c81cdfe3d945402e8ee0215/backports_zstd-1.6.0-pp311-pypy311_pp73-macosx_11_0_arm64.whl", hash = "sha256:9cb75e33131946fabd6319061df3b8b1d588fe0963183280e9b5f49f7772fc09", size = 340554, upload-time = "2026-06-14T10:50:51.523Z" }, + { url = "https://files.pythonhosted.org/packages/f0/00/f083d7c8a4ee5d0bb21b4d3144e76de9f655ca4dd0bffcb95baa5bc47a62/backports_zstd-1.6.0-pp311-pypy311_pp73-manylinux2010_i686.manylinux_2_12_i686.manylinux_2_28_i686.whl", hash = "sha256:ef132cfb638e9a86bd5dc07fb4e1cb895bc55bce6bb5e759366e8b160d0747e2", size = 421694, upload-time = "2026-06-14T10:50:52.917Z" }, + { url = "https://files.pythonhosted.org/packages/41/d7/693b20f3ccae2e05d166f98fe55b1657451170b72c804ed9f6b98df520be/backports_zstd-1.6.0-pp311-pypy311_pp73-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:ab70eace272d6f122b121c057e436709b50a28abf30d97aab28433c08f4a4095", size = 395237, upload-time = "2026-06-14T10:50:54.448Z" }, + { url = "https://files.pythonhosted.org/packages/53/a1/484e0f9ec994bd2285d6747e7c8028350f1a177e9210bc57637898042d3b/backports_zstd-1.6.0-pp311-pypy311_pp73-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:17efb3d11137de5166dd51eedab9c36ad633402acba386eee8d715213ea47e49", size = 415201, upload-time = "2026-06-14T10:50:55.854Z" }, + { url = "https://files.pythonhosted.org/packages/3c/56/70860ece85cd49b564305cbc22bf6c4183975427ff6dfe2097e855f5dd5e/backports_zstd-1.6.0-pp311-pypy311_pp73-win_amd64.whl", hash = "sha256:994167ff6551b9c1ce226e0aab16295b98c94507b5701aa60d2c32b7d50796b1", size = 315721, upload-time = "2026-06-14T10:50:57.074Z" }, +] + [[package]] name = "black" version = "26.3.1" @@ -608,6 +678,48 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/1a/70/e63223f8116931d365993d4a6b7ef653a4d920b41d03de7c59499962821f/click-8.1.6-py3-none-any.whl", hash = "sha256:fa244bb30b3b5ee2cae3da8f55c9e5e0c0e86093306301fb418eb9dc40fbded5", size = 97909, upload-time = "2023-07-18T20:05:12.481Z" }, ] +[[package]] +name = "clickhouse-connect" +version = "1.6.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "backports-zstd" }, + { name = "certifi" }, + { name = "lz4" }, + { name = "tzdata" }, + { name = "urllib3" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/ac/31/aeda81546adc2004742933ed7b741716e3581fea5e4fd6c4dd8a196f82cd/clickhouse_connect-1.6.0.tar.gz", hash = "sha256:e4eb8b83b848f4ca4b984899acd62b0a60f310cbaa559d83519b7477968392ed", size = 193965, upload-time = "2026-07-23T18:03:16.783Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/5f/9e/80810656d9d5ce1357222ad5a326a9c569030ebec5b46351e86716227e96/clickhouse_connect-1.6.0-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:effd020411e4669f21d8b9f7f95d2342145ff6a2e32b49eae416fbd733334e53", size = 387549, upload-time = "2026-07-23T18:01:56.54Z" }, + { url = "https://files.pythonhosted.org/packages/a1/85/ec9f6359dfcf0a918c6926b948f079bbda02f95be06a7f6762aa352ecd5c/clickhouse_connect-1.6.0-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:f8957fa2fd400d1f002dded822b3e667b7941e48107d8d2470e1935a12af9937", size = 376692, upload-time = "2026-07-23T18:01:58.112Z" }, + { url = "https://files.pythonhosted.org/packages/f6/e8/12acfd15c40c9c3ac4d1c6fc413eec3ca3782adb70cf00ecf021abc12556/clickhouse_connect-1.6.0-cp311-cp311-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:ca3d4b70d1fa92e62ab130dec495b470de6e0e56a944cd3c8ea049cb0f137c20", size = 1323910, upload-time = "2026-07-23T18:01:59.525Z" }, + { url = "https://files.pythonhosted.org/packages/c2/f1/dda7de7bce2f72f263a725bf3e3cabcf81f13608cd6f1a71152f59d005e5/clickhouse_connect-1.6.0-cp311-cp311-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:ed28787dd00a34b003fe24de4953e15b172058d875279336d4ee5d8f585acc50", size = 1334678, upload-time = "2026-07-23T18:02:01.097Z" }, + { url = "https://files.pythonhosted.org/packages/a4/03/004af313f0edb521b12967f3d80cb8ac465c1a06c964f3ba305297b7872e/clickhouse_connect-1.6.0-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:d56348f1a39c8bd872106c4af92faa532fbdd8d66e72c588e9b4dd676ca2dbbd", size = 1312144, upload-time = "2026-07-23T18:02:02.698Z" }, + { url = "https://files.pythonhosted.org/packages/fe/bd/24b758e548e34a3a0e661ddd43ffb981629b4ca9b1d0bbf89567bb261a75/clickhouse_connect-1.6.0-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:d541a1a75f08d62b7e0059f77efeff26f2b0829fa0d543e1e1edcce270c78a06", size = 1342925, upload-time = "2026-07-23T18:02:04.008Z" }, + { url = "https://files.pythonhosted.org/packages/79/1d/10f1d37e1feb8840e959dc554ca8cfb5c8b2b0eea93c3e2c0b55a38f43a9/clickhouse_connect-1.6.0-cp311-cp311-win32.whl", hash = "sha256:2d5888d76be3bf1ddea5a210522572d66abcaf651b45ff7c5a1014ec08fe7565", size = 347963, upload-time = "2026-07-23T18:02:05.424Z" }, + { url = "https://files.pythonhosted.org/packages/12/e1/ebe765c6d4173f20ccee002e41289d30f2e40203cedee9e3ee0450ee4cd7/clickhouse_connect-1.6.0-cp311-cp311-win_amd64.whl", hash = "sha256:5135f91841e9df5827faa24f3b2436d4ac8242e74a88a8d47922e31a4383b712", size = 368380, upload-time = "2026-07-23T18:02:06.729Z" }, + { url = "https://files.pythonhosted.org/packages/0b/ea/0c1eaafcb78d5e33d13b0ca531f5f4ee0e1101ea2eb8bfd8a22dde2bdd0f/clickhouse_connect-1.6.0-cp311-cp311-win_arm64.whl", hash = "sha256:fe2736f4cb520c4403f1f6b078ed3ed6f41a6ee8972328082240001853d8639e", size = 350903, upload-time = "2026-07-23T18:02:08.17Z" }, + { url = "https://files.pythonhosted.org/packages/23/0e/9c6a165ecfd0708df9bec011b6adedb2e627b5778b041f7e35ca4bf942b9/clickhouse_connect-1.6.0-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:7043480f1459ddd791480e3ea560a1def81dfd6462eab0c9ada54946e72918af", size = 385950, upload-time = "2026-07-23T18:02:09.475Z" }, + { url = "https://files.pythonhosted.org/packages/ea/8d/11521bebc3c89c8f47e94bdaa5abea49b15946c1d0b9d1464e0efd6623d9/clickhouse_connect-1.6.0-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:064144e1a7f6075ade3c13b8ce49517ca92f86c0c4842bd2abb52a475c026025", size = 374386, upload-time = "2026-07-23T18:02:10.959Z" }, + { url = "https://files.pythonhosted.org/packages/ea/1f/66ed0c9fa39be881f6e1b5b0f570564f6a03b5ee938c614a9e7aa967f63b/clickhouse_connect-1.6.0-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:069f105173cd04263d3bb710381edae341dc9a3aa42fb4673017a617a5a1237a", size = 1335168, upload-time = "2026-07-23T18:02:12.247Z" }, + { url = "https://files.pythonhosted.org/packages/c7/bb/78a23f75d3750e10a6740c9643505f4b5c5e2cdb2fee558b3f2af4803b39/clickhouse_connect-1.6.0-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:30bb77e7aa842550d5265c4aaa3e432b01440fd116d63e7e79ae79c380971ade", size = 1358890, upload-time = "2026-07-23T18:02:13.807Z" }, + { url = "https://files.pythonhosted.org/packages/90/91/e809ac0953394009137eba72bb2f727c6a63e55a1c019a07a8f834589ae2/clickhouse_connect-1.6.0-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:4372dbcbd7b1c897c8d7062832468f82af27fa731c7488421ec7f0a882e1580b", size = 1312423, upload-time = "2026-07-23T18:02:15.314Z" }, + { url = "https://files.pythonhosted.org/packages/66/6b/3cf46e7c270852673307f61d5206875beb6a367853b4d0787bd6d97fb01a/clickhouse_connect-1.6.0-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:a5285b9005d5a3ba41d6f148c957e4df8e37a38ad27835a4ad9c31fe21083149", size = 1354991, upload-time = "2026-07-23T18:02:17.216Z" }, + { url = "https://files.pythonhosted.org/packages/12/1e/793672cd06e32ba16d09e11b687651bda05c9e82e8a64c34c5ebb9f0a7a6/clickhouse_connect-1.6.0-cp312-cp312-win32.whl", hash = "sha256:bd8de68d176a807fd16b85c3c5639c25bbdc81c203217f9e985e9560357fecd5", size = 349053, upload-time = "2026-07-23T18:02:18.736Z" }, + { url = "https://files.pythonhosted.org/packages/45/72/e939a153b6266a4ae4ef32820567a5cc5aa4069b3adbc59b02b861fd140f/clickhouse_connect-1.6.0-cp312-cp312-win_amd64.whl", hash = "sha256:2f8b8060a8277509db77b14a01ed70683a0d9becba7a9994a5e59e42e9f98b37", size = 368135, upload-time = "2026-07-23T18:02:20.235Z" }, + { url = "https://files.pythonhosted.org/packages/fe/4e/c5bf9989371204cfeca68b7f1663509b62662fd25134cd31b81be3cae2cc/clickhouse_connect-1.6.0-cp312-cp312-win_arm64.whl", hash = "sha256:cf66e5bcfba8a183e40485efcfc80329ec75104e22dd02485fe73c0ab9073b19", size = 349140, upload-time = "2026-07-23T18:02:21.935Z" }, + { url = "https://files.pythonhosted.org/packages/e0/58/6daab3e2c2e5e12bb3f658a665c8b36b635b45bf2bb879c429ab736455ba/clickhouse_connect-1.6.0-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:8128dfe4a17aef265d5ba2bb5f2f5a0d5fa76e6aca357c71a17da378c1b30ea6", size = 384027, upload-time = "2026-07-23T18:02:23.482Z" }, + { url = "https://files.pythonhosted.org/packages/f5/6e/abc3575964f6f737274619362cdf83a9c1f21d75febff9eda6301167c15e/clickhouse_connect-1.6.0-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:8d4303987b36ca075cbbe2cff3439ad017f56e1cf2bd936f83b2e6be222ae62f", size = 372729, upload-time = "2026-07-23T18:02:24.855Z" }, + { url = "https://files.pythonhosted.org/packages/8a/eb/afc65f349ef9b38a865ffe58b8fcc9e5bcb14e5a0aa1fb663f07e9c71f99/clickhouse_connect-1.6.0-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:a34e72d1ae6f71be491b2d44bd10d082d04c9e2e2a04d4774498e965d933b21d", size = 1306563, upload-time = "2026-07-23T18:02:26.36Z" }, + { url = "https://files.pythonhosted.org/packages/1a/12/3a6d1b11c01bf936af1168868a2d953e72494537ef142d247fe1fa62d1da/clickhouse_connect-1.6.0-cp313-cp313-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:444564bad257b8923005431d661e347e5360cd686aae67091de3796a465cfec1", size = 1330861, upload-time = "2026-07-23T18:02:28.135Z" }, + { url = "https://files.pythonhosted.org/packages/33/0c/31ce83989fa10606c7108c04caf1d9b662048918e9f2ac0aa2a4fa407929/clickhouse_connect-1.6.0-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:794735a62d0a46688d50652876bcb50116c5e912577e9a469ef6ed1aa52b4d7b", size = 1283562, upload-time = "2026-07-23T18:02:29.82Z" }, + { url = "https://files.pythonhosted.org/packages/c5/53/9613b4b70b66d662e1a0cffecd55fce5eedc398c3aefd08422298c3e88ab/clickhouse_connect-1.6.0-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:75390e6ff6b398a88e0cac4907f785763fc9e476314cbafcfb24553bdbd92d28", size = 1326848, upload-time = "2026-07-23T18:02:31.67Z" }, + { url = "https://files.pythonhosted.org/packages/93/1e/6569d51fea552e69af33feec06f3236c450329ef8f9d19b107888ca574fd/clickhouse_connect-1.6.0-cp313-cp313-win32.whl", hash = "sha256:1a84eb6d545e2647a51b3b4ffa0da4a49f308aaaa096e58c44fb749c171a7d4a", size = 348445, upload-time = "2026-07-23T18:02:33.188Z" }, + { url = "https://files.pythonhosted.org/packages/42/8a/0f13f9f02af1c564fab34346c1496a8254eaa5cb81f10111e72550ea5472/clickhouse_connect-1.6.0-cp313-cp313-win_amd64.whl", hash = "sha256:483e34a7841dbf3d2863e784a129503cc37a910a05464f2f48ce3e9b2f0fc007", size = 366668, upload-time = "2026-07-23T18:02:34.663Z" }, + { url = "https://files.pythonhosted.org/packages/20/46/75a37ee92929327870ab371e3186074cd4806ed16cd5ae604cfd0f54f674/clickhouse_connect-1.6.0-cp313-cp313-win_arm64.whl", hash = "sha256:cd4c1dee496ad3a811237ff8066084c18cdaa46a046f9e0eeacce107ac29c83c", size = 348490, upload-time = "2026-07-23T18:02:36.489Z" }, +] + [[package]] name = "clickhouse-driver" version = "0.2.11" @@ -1439,6 +1551,7 @@ dependencies = [ { name = "backoff" }, { name = "boto3" }, { name = "chargebee" }, + { name = "clickhouse-connect" }, { name = "clickhouse-driver" }, { name = "coreapi" }, { name = "dj-database-url" }, @@ -1565,6 +1678,7 @@ requires-dist = [ { name = "boto3", specifier = ">=1.35.95,<1.36.0" }, { name = "boto3-stubs", marker = "extra == 'dev'", specifier = ">=1.36.20,<2.0.0" }, { name = "chargebee", specifier = ">=3.10.0,<4.0.0" }, + { name = "clickhouse-connect", specifier = ">=1.6.0,<2.0.0" }, { name = "clickhouse-driver", specifier = "==0.2.11" }, { name = "coreapi", specifier = ">=2.3.3,<2.4.0" }, { name = "datamodel-code-generator", marker = "extra == 'dev'", specifier = ">=0.25,<0.26.0" }, @@ -2294,6 +2408,46 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/4e/f6/71d6ec9f18da0b2201287ce9db6afb1a1f637dedb3f0703409558981c723/ldap3-2.9.1-py2.py3-none-any.whl", hash = "sha256:5869596fc4948797020d3f03b7939da938778a0f9e2009f7a072ccf92b8e8d70", size = 432192, upload-time = "2021-07-18T06:34:12.905Z" }, ] +[[package]] +name = "lz4" +version = "4.4.5" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/57/51/f1b86d93029f418033dddf9b9f79c8d2641e7454080478ee2aab5123173e/lz4-4.4.5.tar.gz", hash = "sha256:5f0b9e53c1e82e88c10d7c180069363980136b9d7a8306c4dca4f760d60c39f0", size = 172886, upload-time = "2025-11-03T13:02:36.061Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/93/5b/6edcd23319d9e28b1bedf32768c3d1fd56eed8223960a2c47dacd2cec2af/lz4-4.4.5-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:d6da84a26b3aa5da13a62e4b89ab36a396e9327de8cd48b436a3467077f8ccd4", size = 207391, upload-time = "2025-11-03T13:01:36.644Z" }, + { url = "https://files.pythonhosted.org/packages/34/36/5f9b772e85b3d5769367a79973b8030afad0d6b724444083bad09becd66f/lz4-4.4.5-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:61d0ee03e6c616f4a8b69987d03d514e8896c8b1b7cc7598ad029e5c6aedfd43", size = 207146, upload-time = "2025-11-03T13:01:37.928Z" }, + { url = "https://files.pythonhosted.org/packages/04/f4/f66da5647c0d72592081a37c8775feacc3d14d2625bbdaabd6307c274565/lz4-4.4.5-cp311-cp311-manylinux1_i686.manylinux_2_28_i686.manylinux_2_5_i686.whl", hash = "sha256:33dd86cea8375d8e5dd001e41f321d0a4b1eb7985f39be1b6a4f466cd480b8a7", size = 1292623, upload-time = "2025-11-03T13:01:39.341Z" }, + { url = "https://files.pythonhosted.org/packages/85/fc/5df0f17467cdda0cad464a9197a447027879197761b55faad7ca29c29a04/lz4-4.4.5-cp311-cp311-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:609a69c68e7cfcfa9d894dc06be13f2e00761485b62df4e2472f1b66f7b405fb", size = 1279982, upload-time = "2025-11-03T13:01:40.816Z" }, + { url = "https://files.pythonhosted.org/packages/25/3b/b55cb577aa148ed4e383e9700c36f70b651cd434e1c07568f0a86c9d5fbb/lz4-4.4.5-cp311-cp311-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:75419bb1a559af00250b8f1360d508444e80ed4b26d9d40ec5b09fe7875cb989", size = 1368674, upload-time = "2025-11-03T13:01:42.118Z" }, + { url = "https://files.pythonhosted.org/packages/fb/31/e97e8c74c59ea479598e5c55cbe0b1334f03ee74ca97726e872944ed42df/lz4-4.4.5-cp311-cp311-win32.whl", hash = "sha256:12233624f1bc2cebc414f9efb3113a03e89acce3ab6f72035577bc61b270d24d", size = 88168, upload-time = "2025-11-03T13:01:43.282Z" }, + { url = "https://files.pythonhosted.org/packages/18/47/715865a6c7071f417bef9b57c8644f29cb7a55b77742bd5d93a609274e7e/lz4-4.4.5-cp311-cp311-win_amd64.whl", hash = "sha256:8a842ead8ca7c0ee2f396ca5d878c4c40439a527ebad2b996b0444f0074ed004", size = 99491, upload-time = "2025-11-03T13:01:44.167Z" }, + { url = "https://files.pythonhosted.org/packages/14/e7/ac120c2ca8caec5c945e6356ada2aa5cfabd83a01e3170f264a5c42c8231/lz4-4.4.5-cp311-cp311-win_arm64.whl", hash = "sha256:83bc23ef65b6ae44f3287c38cbf82c269e2e96a26e560aa551735883388dcc4b", size = 91271, upload-time = "2025-11-03T13:01:45.016Z" }, + { url = "https://files.pythonhosted.org/packages/1b/ac/016e4f6de37d806f7cc8f13add0a46c9a7cfc41a5ddc2bc831d7954cf1ce/lz4-4.4.5-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:df5aa4cead2044bab83e0ebae56e0944cc7fcc1505c7787e9e1057d6d549897e", size = 207163, upload-time = "2025-11-03T13:01:45.895Z" }, + { url = "https://files.pythonhosted.org/packages/8d/df/0fadac6e5bd31b6f34a1a8dbd4db6a7606e70715387c27368586455b7fc9/lz4-4.4.5-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:6d0bf51e7745484d2092b3a51ae6eb58c3bd3ce0300cf2b2c14f76c536d5697a", size = 207150, upload-time = "2025-11-03T13:01:47.205Z" }, + { url = "https://files.pythonhosted.org/packages/b7/17/34e36cc49bb16ca73fb57fbd4c5eaa61760c6b64bce91fcb4e0f4a97f852/lz4-4.4.5-cp312-cp312-manylinux1_i686.manylinux_2_28_i686.manylinux_2_5_i686.whl", hash = "sha256:7b62f94b523c251cf32aa4ab555f14d39bd1a9df385b72443fd76d7c7fb051f5", size = 1292045, upload-time = "2025-11-03T13:01:48.667Z" }, + { url = "https://files.pythonhosted.org/packages/90/1c/b1d8e3741e9fc89ed3b5f7ef5f22586c07ed6bb04e8343c2e98f0fa7ff04/lz4-4.4.5-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:2c3ea562c3af274264444819ae9b14dbbf1ab070aff214a05e97db6896c7597e", size = 1279546, upload-time = "2025-11-03T13:01:50.159Z" }, + { url = "https://files.pythonhosted.org/packages/55/d9/e3867222474f6c1b76e89f3bd914595af69f55bf2c1866e984c548afdc15/lz4-4.4.5-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:24092635f47538b392c4eaeff14c7270d2c8e806bf4be2a6446a378591c5e69e", size = 1368249, upload-time = "2025-11-03T13:01:51.273Z" }, + { url = "https://files.pythonhosted.org/packages/b2/e7/d667d337367686311c38b580d1ca3d5a23a6617e129f26becd4f5dc458df/lz4-4.4.5-cp312-cp312-win32.whl", hash = "sha256:214e37cfe270948ea7eb777229e211c601a3e0875541c1035ab408fbceaddf50", size = 88189, upload-time = "2025-11-03T13:01:52.605Z" }, + { url = "https://files.pythonhosted.org/packages/a5/0b/a54cd7406995ab097fceb907c7eb13a6ddd49e0b231e448f1a81a50af65c/lz4-4.4.5-cp312-cp312-win_amd64.whl", hash = "sha256:713a777de88a73425cf08eb11f742cd2c98628e79a8673d6a52e3c5f0c116f33", size = 99497, upload-time = "2025-11-03T13:01:53.477Z" }, + { url = "https://files.pythonhosted.org/packages/6a/7e/dc28a952e4bfa32ca16fa2eb026e7a6ce5d1411fcd5986cd08c74ec187b9/lz4-4.4.5-cp312-cp312-win_arm64.whl", hash = "sha256:a88cbb729cc333334ccfb52f070463c21560fca63afcf636a9f160a55fac3301", size = 91279, upload-time = "2025-11-03T13:01:54.419Z" }, + { url = "https://files.pythonhosted.org/packages/2f/46/08fd8ef19b782f301d56a9ccfd7dafec5fd4fc1a9f017cf22a1accb585d7/lz4-4.4.5-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:6bb05416444fafea170b07181bc70640975ecc2a8c92b3b658c554119519716c", size = 207171, upload-time = "2025-11-03T13:01:56.595Z" }, + { url = "https://files.pythonhosted.org/packages/8f/3f/ea3334e59de30871d773963997ecdba96c4584c5f8007fd83cfc8f1ee935/lz4-4.4.5-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:b424df1076e40d4e884cfcc4c77d815368b7fb9ebcd7e634f937725cd9a8a72a", size = 207163, upload-time = "2025-11-03T13:01:57.721Z" }, + { url = "https://files.pythonhosted.org/packages/41/7b/7b3a2a0feb998969f4793c650bb16eff5b06e80d1f7bff867feb332f2af2/lz4-4.4.5-cp313-cp313-manylinux1_i686.manylinux_2_28_i686.manylinux_2_5_i686.whl", hash = "sha256:216ca0c6c90719731c64f41cfbd6f27a736d7e50a10b70fad2a9c9b262ec923d", size = 1292136, upload-time = "2025-11-03T13:02:00.375Z" }, + { url = "https://files.pythonhosted.org/packages/89/d1/f1d259352227bb1c185288dd694121ea303e43404aa77560b879c90e7073/lz4-4.4.5-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:533298d208b58b651662dd972f52d807d48915176e5b032fb4f8c3b6f5fe535c", size = 1279639, upload-time = "2025-11-03T13:02:01.649Z" }, + { url = "https://files.pythonhosted.org/packages/d2/fb/ba9256c48266a09012ed1d9b0253b9aa4fe9cdff094f8febf5b26a4aa2a2/lz4-4.4.5-cp313-cp313-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:451039b609b9a88a934800b5fc6ee401c89ad9c175abf2f4d9f8b2e4ef1afc64", size = 1368257, upload-time = "2025-11-03T13:02:03.35Z" }, + { url = "https://files.pythonhosted.org/packages/a5/6d/dee32a9430c8b0e01bbb4537573cabd00555827f1a0a42d4e24ca803935c/lz4-4.4.5-cp313-cp313-win32.whl", hash = "sha256:a5f197ffa6fc0e93207b0af71b302e0a2f6f29982e5de0fbda61606dd3a55832", size = 88191, upload-time = "2025-11-03T13:02:04.406Z" }, + { url = "https://files.pythonhosted.org/packages/18/e0/f06028aea741bbecb2a7e9648f4643235279a770c7ffaf70bd4860c73661/lz4-4.4.5-cp313-cp313-win_amd64.whl", hash = "sha256:da68497f78953017deb20edff0dba95641cc86e7423dfadf7c0264e1ac60dc22", size = 99502, upload-time = "2025-11-03T13:02:05.886Z" }, + { url = "https://files.pythonhosted.org/packages/61/72/5bef44afb303e56078676b9f2486f13173a3c1e7f17eaac1793538174817/lz4-4.4.5-cp313-cp313-win_arm64.whl", hash = "sha256:c1cfa663468a189dab510ab231aad030970593f997746d7a324d40104db0d0a9", size = 91285, upload-time = "2025-11-03T13:02:06.77Z" }, + { url = "https://files.pythonhosted.org/packages/49/55/6a5c2952971af73f15ed4ebfdd69774b454bd0dc905b289082ca8664fba1/lz4-4.4.5-cp313-cp313t-macosx_10_13_x86_64.whl", hash = "sha256:67531da3b62f49c939e09d56492baf397175ff39926d0bd5bd2d191ac2bff95f", size = 207348, upload-time = "2025-11-03T13:02:08.117Z" }, + { url = "https://files.pythonhosted.org/packages/4e/d7/fd62cbdbdccc35341e83aabdb3f6d5c19be2687d0a4eaf6457ddf53bba64/lz4-4.4.5-cp313-cp313t-macosx_11_0_arm64.whl", hash = "sha256:a1acbbba9edbcbb982bc2cac5e7108f0f553aebac1040fbec67a011a45afa1ba", size = 207340, upload-time = "2025-11-03T13:02:09.152Z" }, + { url = "https://files.pythonhosted.org/packages/77/69/225ffadaacb4b0e0eb5fd263541edd938f16cd21fe1eae3cd6d5b6a259dc/lz4-4.4.5-cp313-cp313t-manylinux1_i686.manylinux_2_28_i686.manylinux_2_5_i686.whl", hash = "sha256:a482eecc0b7829c89b498fda883dbd50e98153a116de612ee7c111c8bcf82d1d", size = 1293398, upload-time = "2025-11-03T13:02:10.272Z" }, + { url = "https://files.pythonhosted.org/packages/c6/9e/2ce59ba4a21ea5dc43460cba6f34584e187328019abc0e66698f2b66c881/lz4-4.4.5-cp313-cp313t-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:e099ddfaa88f59dd8d36c8a3c66bd982b4984edf127eb18e30bb49bdba68ce67", size = 1281209, upload-time = "2025-11-03T13:02:12.091Z" }, + { url = "https://files.pythonhosted.org/packages/80/4f/4d946bd1624ec229b386a3bc8e7a85fa9a963d67d0a62043f0af0978d3da/lz4-4.4.5-cp313-cp313t-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:a2af2897333b421360fdcce895c6f6281dc3fab018d19d341cf64d043fc8d90d", size = 1369406, upload-time = "2025-11-03T13:02:13.683Z" }, + { url = "https://files.pythonhosted.org/packages/02/a2/d429ba4720a9064722698b4b754fb93e42e625f1318b8fe834086c7c783b/lz4-4.4.5-cp313-cp313t-win32.whl", hash = "sha256:66c5de72bf4988e1b284ebdd6524c4bead2c507a2d7f172201572bac6f593901", size = 88325, upload-time = "2025-11-03T13:02:14.743Z" }, + { url = "https://files.pythonhosted.org/packages/4b/85/7ba10c9b97c06af6c8f7032ec942ff127558863df52d866019ce9d2425cf/lz4-4.4.5-cp313-cp313t-win_amd64.whl", hash = "sha256:cdd4bdcbaf35056086d910d219106f6a04e1ab0daa40ec0eeef1626c27d0fddb", size = 99643, upload-time = "2025-11-03T13:02:15.978Z" }, + { url = "https://files.pythonhosted.org/packages/77/4d/a175459fb29f909e13e57c8f475181ad8085d8d7869bd8ad99033e3ee5fa/lz4-4.4.5-cp313-cp313t-win_arm64.whl", hash = "sha256:28ccaeb7c5222454cd5f60fcd152564205bcb801bd80e125949d2dfbadc76bbd", size = 91504, upload-time = "2025-11-03T13:02:17.313Z" }, +] + [[package]] name = "markupsafe" version = "2.1.3" diff --git a/docs/docs/deployment-self-hosting/observability/_metrics-catalogue.md b/docs/docs/deployment-self-hosting/observability/_metrics-catalogue.md index 83d66747740e..82c6e248c446 100644 --- a/docs/docs/deployment-self-hosting/observability/_metrics-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_metrics-catalogue.md @@ -43,6 +43,24 @@ Counter. Outcomes of connection verification attempts against customers' own data warehouses. `result` label is either `success` or `failure`. +Labels: + - `result` + +### `flagsmith_experimentation_warehouse_delivery_objects` + +Counter. + +Buffered S3 event objects processed by delivery to customers' own data warehouses. `result` label is `delivered` for objects inserted and archived, or `rejected` for objects the warehouse refused, which are moved aside and never retried. + +Labels: + - `result` + +### `flagsmith_experimentation_warehouse_delivery_runs` + +Counter. + +Outcomes of per-connection runs delivering buffered event objects to customers' own data warehouses. `result` label is either `success` or `failure`; a failed run delivers nothing and is retried on the next tick. + Labels: - `result` From 8562a8d1ca7a28f502ea3eb00f8f7ca7569a2c22 Mon Sep 17 00:00:00 2001 From: Gagan Trivedi Date: Tue, 28 Jul 2026 11:07:56 +0530 Subject: [PATCH 2/9] fix(experimentation): bound delivery task runtime and sanitise status detail The task processor's default 60-second timeout retries a task while its original thread keeps running, so any backlog drain would race a second run for the same connection; a 9-minute timeout keeps runs inside the 10-minute tick. Failed runs now write a curated message to the user-facing status_detail instead of raw exception text. --- api/experimentation/tasks.py | 9 +++- .../warehouse_delivery_service.py | 24 ++++++++- api/tests/unit/experimentation/test_tasks.py | 2 +- .../test_warehouse_delivery_service.py | 50 ++++++++++++++++++- 4 files changed, 80 insertions(+), 5 deletions(-) diff --git a/api/experimentation/tasks.py b/api/experimentation/tasks.py index 832173723ef6..140771ceaa6a 100644 --- a/api/experimentation/tasks.py +++ b/api/experimentation/tasks.py @@ -1,3 +1,5 @@ +from datetime import timedelta + import structlog from django.utils import timezone from task_processor.decorators import ( @@ -134,7 +136,7 @@ def deliver_events_to_external_warehouses() -> None: deliver_events_for_connection.delay(kwargs={"connection_id": connection_id}) -@register_task_handler() +@register_task_handler(timeout=timedelta(minutes=9)) def deliver_events_for_connection(connection_id: int) -> None: connection = ( WarehouseConnection.objects.select_related( @@ -200,7 +202,10 @@ def deliver_events_for_connection(connection_id: int) -> None: # The warehouse itself is unusable; deliver nothing, leave every # remaining object in place for the next run, and surface the # breakage on the connection. - mark_warehouse_delivery_failed(connection, detail=str(exc)) + mark_warehouse_delivery_failed( + connection, + detail=warehouse_delivery_service.describe_delivery_error(exc), + ) flagsmith_experimentation_warehouse_delivery_runs_total.labels( result="failure" ).inc() diff --git a/api/experimentation/warehouse_delivery_service.py b/api/experimentation/warehouse_delivery_service.py index c3cc9ccf22ff..63a5bbeff923 100644 --- a/api/experimentation/warehouse_delivery_service.py +++ b/api/experimentation/warehouse_delivery_service.py @@ -5,7 +5,10 @@ import boto3 import clickhouse_connect import structlog -from clickhouse_connect.driver.exceptions import DatabaseError +from clickhouse_connect.driver.exceptions import ( + DatabaseError, + OperationalError, +) from core.network import is_internal_address from experimentation.ingestion_infra_service import AWS_REGION @@ -70,6 +73,25 @@ class ObjectRejectedError(Exception): ) +def describe_delivery_error(error: Exception) -> str: + """Return a user-facing description of a failed delivery run, suitable for + the connection's ``status_detail``. Raw exception text stays in the logs: + it can carry internal infrastructure details.""" + if isinstance(error, DeliveryConfigError): + return str(error) + # OperationalError subclasses DatabaseError, so it is matched first. + if isinstance(error, OperationalError): + return "Could not connect to the host." + if isinstance(error, DatabaseError): + # 516 = AUTHENTICATION_FAILED, 60 = UNKNOWN_TABLE + if error.code == 516: + return "Authentication failed." + if error.code == 60: + return f"Table `{EVENTS_TABLE_NAME}` does not exist." + return "The ClickHouse server rejected the request." + return "Delivery failed." + + @lru_cache(maxsize=1) def _get_s3_client() -> "Any": return boto3.client("s3", region_name=AWS_REGION) diff --git a/api/tests/unit/experimentation/test_tasks.py b/api/tests/unit/experimentation/test_tasks.py index 2647ecc6a586..753fa8e7b7e3 100644 --- a/api/tests/unit/experimentation/test_tasks.py +++ b/api/tests/unit/experimentation/test_tasks.py @@ -1020,6 +1020,6 @@ def test_deliver_events_for_connection__warehouse_unusable__aborts_and_marks_err # Then the breakage is surfaced on the connection clickhouse_connection.refresh_from_db() assert clickhouse_connection.status == WarehouseConnectionStatus.ERRORED - assert clickhouse_connection.status_detail == "Authentication failed" + assert clickhouse_connection.status_detail == "Authentication failed." assert _delivery_runs_count("failure") == failure_runs_before + 1 assert log.has("warehouse_delivery.failed", level="error") diff --git a/api/tests/unit/experimentation/test_warehouse_delivery_service.py b/api/tests/unit/experimentation/test_warehouse_delivery_service.py index ffaa80c758fe..2f81c0e8aa4a 100644 --- a/api/tests/unit/experimentation/test_warehouse_delivery_service.py +++ b/api/tests/unit/experimentation/test_warehouse_delivery_service.py @@ -4,7 +4,7 @@ import boto3 import pytest -from clickhouse_connect.driver.exceptions import DatabaseError +from clickhouse_connect.driver.exceptions import DatabaseError, OperationalError from moto import mock_s3 # type: ignore[import-untyped] from pytest_mock import MockerFixture @@ -253,3 +253,51 @@ def test_deliver_object__connection_level_error__reraises( BUCKET_NAME, _event_key("13"), ) + + +@pytest.mark.parametrize( + "error, expected_detail", + [ + pytest.param( + warehouse_delivery_service.DeliveryConfigError( + "Stored connection details are incomplete." + ), + "Stored connection details are incomplete.", + id="config-error", + ), + pytest.param( + OperationalError("HTTPSConnectionPool: Max retries exceeded"), + "Could not connect to the host.", + id="unreachable", + ), + pytest.param( + DatabaseError("Code: 516. DB::Exception: nope", code=516), + "Authentication failed.", + id="bad-auth", + ), + pytest.param( + DatabaseError("Code: 60. DB::Exception: no table", code=60), + "Table `events` does not exist.", + id="missing-table", + ), + pytest.param( + DatabaseError("Code: 241. DB::Exception: memory limit", code=241), + "The ClickHouse server rejected the request.", + id="other-server-error", + ), + pytest.param( + ConnectionResetError("connection reset by peer"), + "Delivery failed.", + id="unexpected-error", + ), + ], +) +def test_describe_delivery_error__known_failures__returns_user_facing_detail( + error: Exception, + expected_detail: str, +) -> None: + # When + detail = warehouse_delivery_service.describe_delivery_error(error) + + # Then + assert detail == expected_detail From 94abf0be47b323a7cb1864001f20f0d77f7e0d34 Mon Sep 17 00:00:00 2001 From: Gagan Trivedi Date: Tue, 28 Jul 2026 11:54:00 +0530 Subject: [PATCH 3/9] refactor(experimentation): move delivery orchestration into services Sibling tasks delegate their business logic to services.py; the delivery task now does too, leaving it as load-and-dispatch. This also gives future warehouse types a single seam. Under the services module's warehouse logger the events become warehouse.delivery.* to match warehouse.connection.*. --- api/experimentation/services.py | 98 ++++++++++++++++++++ api/experimentation/tasks.py | 86 ++--------------- api/tests/unit/experimentation/test_tasks.py | 8 +- 3 files changed, 110 insertions(+), 82 deletions(-) diff --git a/api/experimentation/services.py b/api/experimentation/services.py index 8539e72df3c7..a6bbf333c9f0 100644 --- a/api/experimentation/services.py +++ b/api/experimentation/services.py @@ -21,6 +21,7 @@ from core.dataclasses import AuthorData from core.network import is_internal_address from environments.tasks import rebuild_environment_document +from experimentation import warehouse_delivery_service from experimentation.constants import ( CONTROL_VARIANT_KEY, EXPERIMENT_FLAG, @@ -45,6 +46,8 @@ ) from experimentation.metrics import ( flagsmith_experimentation_warehouse_connection_verifications_total, + flagsmith_experimentation_warehouse_delivery_objects_total, + flagsmith_experimentation_warehouse_delivery_runs_total, ) from experimentation.models import ( VALID_STATUS_TRANSITIONS, @@ -84,6 +87,8 @@ from collections.abc import Sequence from datetime import datetime + from clickhouse_connect.driver.client import Client as ClickHouseHTTPClient + from experimentation.models import Metric, WarehouseConnection from experimentation.types import ExposureGranularity from features.feature_states.models import FeatureValueType @@ -821,6 +826,99 @@ def mark_warehouse_delivery_succeeded(connection: WarehouseConnection) -> None: connection.save(update_fields=["status", "status_detail"]) +def _deliver_pending_objects( + client: ClickHouseHTTPClient, + *, + bucket_name: str, + pending: list[str], + log: structlog.stdlib.BoundLogger, +) -> tuple[int, int, int]: + delivered_count = rejected_count = rows_count = 0 + for s3_key in pending: + try: + rows_count += warehouse_delivery_service.deliver_object( + client, + bucket_name, + s3_key, + ) + except warehouse_delivery_service.ObjectRejectedError: + # This object's contents are the problem; the ones behind it are + # still deliverable. + warehouse_delivery_service.move_object( + bucket_name, + s3_key, + to_prefix=warehouse_delivery_service.FAILED_PREFIX, + ) + rejected_count += 1 + flagsmith_experimentation_warehouse_delivery_objects_total.labels( + result="rejected" + ).inc() + log.warning("delivery.object_rejected", exc_info=True) + continue + warehouse_delivery_service.move_object( + bucket_name, + s3_key, + to_prefix=warehouse_delivery_service.ARCHIVE_PREFIX, + ) + delivered_count += 1 + flagsmith_experimentation_warehouse_delivery_objects_total.labels( + result="delivered" + ).inc() + return delivered_count, rejected_count, rows_count + + +def deliver_warehouse_events( + connection: WarehouseConnection, + *, + bucket_name: str, +) -> None: + """Deliver the environment's pending event objects to the connection's + warehouse, surfacing the outcome on the connection's status.""" + log = logger.bind( + environment__id=connection.environment_id, + organisation__id=connection.environment.project.organisation_id, + ) + pending = warehouse_delivery_service.list_pending_objects( + bucket_name, + environment_key=connection.environment.api_key, + ) + if not pending: + return + + try: + with warehouse_delivery_service.delivery_client(connection) as client: + delivered_count, rejected_count, rows_count = _deliver_pending_objects( + client, + bucket_name=bucket_name, + pending=pending, + log=log, + ) + except Exception as exc: + # The warehouse itself is unusable; deliver nothing, leave every + # remaining object in place for the next run, and surface the + # breakage on the connection. + mark_warehouse_delivery_failed( + connection, + detail=warehouse_delivery_service.describe_delivery_error(exc), + ) + flagsmith_experimentation_warehouse_delivery_runs_total.labels( + result="failure" + ).inc() + log.error("delivery.failed", exc_info=exc) + return + + mark_warehouse_delivery_succeeded(connection) + flagsmith_experimentation_warehouse_delivery_runs_total.labels( + result="success" + ).inc() + log.info( + "delivery.completed", + objects__count=delivered_count, + objects__rejected_count=rejected_count, + rows__count=rows_count, + ) + + class InternalAddressError(Exception): pass diff --git a/api/experimentation/tasks.py b/api/experimentation/tasks.py index 140771ceaa6a..105dae4d4d71 100644 --- a/api/experimentation/tasks.py +++ b/api/experimentation/tasks.py @@ -8,12 +8,8 @@ ) from environments.models import Environment, EnvironmentAPIKey -from experimentation import ingestion_sync_service, warehouse_delivery_service +from experimentation import ingestion_sync_service from experimentation.constants import DELIVERY_INTERVAL -from experimentation.metrics import ( - flagsmith_experimentation_warehouse_delivery_objects_total, - flagsmith_experimentation_warehouse_delivery_runs_total, -) from experimentation.models import ( Experiment, ExperimentExposures, @@ -28,8 +24,7 @@ from experimentation.services import ( compute_exposures_summary, compute_results_summary, - mark_warehouse_delivery_failed, - mark_warehouse_delivery_succeeded, + deliver_warehouse_events, ) logger = structlog.get_logger("experimentation") @@ -148,80 +143,15 @@ def deliver_events_for_connection(connection_id: int) -> None: if connection is None: return - organisation = connection.environment.project.organisation - infrastructure = getattr(organisation, "ingestion_infrastructure", None) - if infrastructure is None or not infrastructure.bucket_name: - return - - log = logger.bind( - environment__id=connection.environment_id, - organisation__id=organisation.id, - ) - bucket_name = infrastructure.bucket_name - pending = warehouse_delivery_service.list_pending_objects( - bucket_name, - environment_key=connection.environment.api_key, + infrastructure = getattr( + connection.environment.project.organisation, + "ingestion_infrastructure", + None, ) - if not pending: - return - - delivered_count = rejected_count = rows_count = 0 - try: - with warehouse_delivery_service.delivery_client(connection) as client: - for s3_key in pending: - try: - rows_count += warehouse_delivery_service.deliver_object( - client, - bucket_name, - s3_key, - ) - except warehouse_delivery_service.ObjectRejectedError: - # This object's contents are the problem; the ones behind - # it are still deliverable. - warehouse_delivery_service.move_object( - bucket_name, - s3_key, - to_prefix=warehouse_delivery_service.FAILED_PREFIX, - ) - rejected_count += 1 - flagsmith_experimentation_warehouse_delivery_objects_total.labels( - result="rejected" - ).inc() - log.warning("warehouse_delivery.object_rejected", exc_info=True) - continue - warehouse_delivery_service.move_object( - bucket_name, - s3_key, - to_prefix=warehouse_delivery_service.ARCHIVE_PREFIX, - ) - delivered_count += 1 - flagsmith_experimentation_warehouse_delivery_objects_total.labels( - result="delivered" - ).inc() - except Exception as exc: - # The warehouse itself is unusable; deliver nothing, leave every - # remaining object in place for the next run, and surface the - # breakage on the connection. - mark_warehouse_delivery_failed( - connection, - detail=warehouse_delivery_service.describe_delivery_error(exc), - ) - flagsmith_experimentation_warehouse_delivery_runs_total.labels( - result="failure" - ).inc() - log.error("warehouse_delivery.failed", exc_info=exc) + if infrastructure is None or not infrastructure.bucket_name: return - mark_warehouse_delivery_succeeded(connection) - flagsmith_experimentation_warehouse_delivery_runs_total.labels( - result="success" - ).inc() - log.info( - "warehouse_delivery.completed", - objects__count=delivered_count, - objects__rejected_count=rejected_count, - rows__count=rows_count, - ) + deliver_warehouse_events(connection, bucket_name=infrastructure.bucket_name) @register_task_handler() diff --git a/api/tests/unit/experimentation/test_tasks.py b/api/tests/unit/experimentation/test_tasks.py index 753fa8e7b7e3..fb1cb8a55c06 100644 --- a/api/tests/unit/experimentation/test_tasks.py +++ b/api/tests/unit/experimentation/test_tasks.py @@ -911,7 +911,7 @@ def test_deliver_events_for_connection__pending_objects__delivers_archives_and_r assert _delivery_objects_count("delivered") == delivered_objects_before + 2 assert { "level": "info", - "event": "warehouse_delivery.completed", + "event": "delivery.completed", "environment__id": environment.id, "organisation__id": environment.project.organisation_id, "objects__count": 2, @@ -972,10 +972,10 @@ def test_deliver_events_for_connection__rejected_object__moves_to_failed_and_con clickhouse_connection.refresh_from_db() assert clickhouse_connection.status == WarehouseConnectionStatus.CONNECTED assert _delivery_objects_count("rejected") == rejected_objects_before + 1 - assert log.has("warehouse_delivery.object_rejected", level="warning") + assert log.has("delivery.object_rejected", level="warning") assert { "level": "info", - "event": "warehouse_delivery.completed", + "event": "delivery.completed", "environment__id": environment.id, "organisation__id": environment.project.organisation_id, "objects__count": 1, @@ -1022,4 +1022,4 @@ def test_deliver_events_for_connection__warehouse_unusable__aborts_and_marks_err assert clickhouse_connection.status == WarehouseConnectionStatus.ERRORED assert clickhouse_connection.status_detail == "Authentication failed." assert _delivery_runs_count("failure") == failure_runs_before + 1 - assert log.has("warehouse_delivery.failed", level="error") + assert log.has("delivery.failed", level="error") From f7959fbc7ddd43cc515353e852c17ebc775fff08 Mon Sep 17 00:00:00 2001 From: Gagan Trivedi Date: Tue, 28 Jul 2026 15:02:55 +0530 Subject: [PATCH 4/9] test(experimentation): add missing Given comments to delivery tests --- api/tests/unit/experimentation/test_tasks.py | 6 ++++++ .../unit/experimentation/test_warehouse_delivery_service.py | 4 ++++ 2 files changed, 10 insertions(+) diff --git a/api/tests/unit/experimentation/test_tasks.py b/api/tests/unit/experimentation/test_tasks.py index fb1cb8a55c06..6859d2cf6518 100644 --- a/api/tests/unit/experimentation/test_tasks.py +++ b/api/tests/unit/experimentation/test_tasks.py @@ -804,6 +804,8 @@ def test_deliver_events_for_connection__missing_connection__does_nothing( db: None, warehouse_client: Any, ) -> None: + # Given no connection with the requested id + # When deliver_events_for_connection(connection_id=404) @@ -815,6 +817,8 @@ def test_deliver_events_for_connection__no_infrastructure__does_nothing( clickhouse_connection: WarehouseConnection, warehouse_client: Any, ) -> None: + # Given a connection whose organisation has no ingestion infrastructure + # When deliver_events_for_connection(connection_id=clickhouse_connection.id) @@ -846,6 +850,8 @@ def test_deliver_events_for_connection__no_pending_objects__does_not_connect( delivery_bucket: Any, warehouse_client: Any, ) -> None: + # Given an events bucket with no objects for the environment + # When deliver_events_for_connection(connection_id=clickhouse_connection.id) diff --git a/api/tests/unit/experimentation/test_warehouse_delivery_service.py b/api/tests/unit/experimentation/test_warehouse_delivery_service.py index 2f81c0e8aa4a..95a90e24ed3b 100644 --- a/api/tests/unit/experimentation/test_warehouse_delivery_service.py +++ b/api/tests/unit/experimentation/test_warehouse_delivery_service.py @@ -70,6 +70,8 @@ def test_list_pending_objects__objects_across_hours__returns_chronological_and_s def test_list_pending_objects__no_objects__returns_empty( events_bucket: Any, ) -> None: + # Given an events bucket with no objects + # When pending = warehouse_delivery_service.list_pending_objects( BUCKET_NAME, @@ -296,6 +298,8 @@ def test_describe_delivery_error__known_failures__returns_user_facing_detail( error: Exception, expected_detail: str, ) -> None: + # Given a parametrised delivery failure + # When detail = warehouse_delivery_service.describe_delivery_error(error) From 6b56a32de7263c1a709418170cebd23a32309044 Mon Sep 17 00:00:00 2001 From: Gagan Trivedi Date: Wed, 29 Jul 2026 12:50:35 +0530 Subject: [PATCH 5/9] fix(experimentation): address delivery review findings - refuse HTTP redirects: a permitted host could otherwise bounce a request, and its event payload, to an address the internal-address guard never saw - bound each run by a time budget so it finishes before the task timeout, which would otherwise abandon a still-running thread and retry alongside it - narrow the failure handler to warehouse errors, so an S3 failure fails the task instead of blaming the customer's connection - treat a run that delivers nothing and rejects everything as a failure: it points at the table's schema, and would otherwise drain into failed/ - drop insert_deduplication_token; it is inert on plain MergeTree and raises on servers that predate it - bind connection and object keys to delivery events --- api/experimentation/services.py | 60 ++++++++++-- .../warehouse_delivery_service.py | 37 ++++++-- api/tests/unit/experimentation/test_tasks.py | 93 ++++++++++++++++++- .../test_warehouse_delivery_service.py | 26 +++++- 4 files changed, 197 insertions(+), 19 deletions(-) diff --git a/api/experimentation/services.py b/api/experimentation/services.py index a6bbf333c9f0..6627316f11f1 100644 --- a/api/experimentation/services.py +++ b/api/experimentation/services.py @@ -1,10 +1,12 @@ from __future__ import annotations +import time import typing from dataclasses import replace from functools import lru_cache import structlog +from clickhouse_connect.driver.exceptions import ClickHouseError from clickhouse_driver import Client from clickhouse_driver import errors as clickhouse_errors from clickhouse_driver.util.helpers import parse_url @@ -105,6 +107,10 @@ _CUSTOMER_EVENT_STATS_UNAVAILABLE = "unavailable" +# A delivery run stops taking on new objects after this long, leaving room for +# the slowest possible in-flight insert to still land inside the task timeout. +DELIVERY_TIME_BUDGET_SECONDS = 210 + def is_warehouse_feature_enabled(organisation: Organisation) -> bool: return get_openfeature_client().get_boolean_value( @@ -831,10 +837,25 @@ def _deliver_pending_objects( *, bucket_name: str, pending: list[str], - log: structlog.stdlib.BoundLogger, + connection: WarehouseConnection, ) -> tuple[int, int, int]: + log = logger.bind( + connection__id=connection.id, + environment__id=connection.environment_id, + organisation__id=connection.environment.project.organisation_id, + ) + # A run that outlives the task timeout is retried while its own thread + # keeps delivering, so it must finish first: whatever is left is picked up + # on the next tick. + deadline = time.monotonic() + DELIVERY_TIME_BUDGET_SECONDS delivered_count = rejected_count = rows_count = 0 - for s3_key in pending: + for index, s3_key in enumerate(pending): + if time.monotonic() > deadline: + log.info( + "delivery.budget_exhausted", + objects__remaining_count=len(pending) - index, + ) + break try: rows_count += warehouse_delivery_service.deliver_object( client, @@ -853,7 +874,11 @@ def _deliver_pending_objects( flagsmith_experimentation_warehouse_delivery_objects_total.labels( result="rejected" ).inc() - log.warning("delivery.object_rejected", exc_info=True) + log.warning( + "delivery.object_rejected", + s3__key=s3_key, + exc_info=True, + ) continue warehouse_delivery_service.move_object( bucket_name, @@ -875,6 +900,7 @@ def deliver_warehouse_events( """Deliver the environment's pending event objects to the connection's warehouse, surfacing the outcome on the connection's status.""" log = logger.bind( + connection__id=connection.id, environment__id=connection.environment_id, organisation__id=connection.environment.project.organisation_id, ) @@ -891,12 +917,14 @@ def deliver_warehouse_events( client, bucket_name=bucket_name, pending=pending, - log=log, + connection=connection, ) - except Exception as exc: + except (warehouse_delivery_service.DeliveryConfigError, ClickHouseError) as exc: # The warehouse itself is unusable; deliver nothing, leave every # remaining object in place for the next run, and surface the - # breakage on the connection. + # breakage on the connection. Anything else — an S3 failure, a bug + # here — is ours, so it propagates and fails the task instead of + # blaming the customer's warehouse. mark_warehouse_delivery_failed( connection, detail=warehouse_delivery_service.describe_delivery_error(exc), @@ -907,6 +935,26 @@ def deliver_warehouse_events( log.error("delivery.failed", exc_info=exc) return + if delivered_count == 0 and rejected_count: + # Records all come from one ingestion pipeline, so every object being + # rejected points at the table's schema rather than the objects. + mark_warehouse_delivery_failed( + connection, + detail=( + f"The warehouse rejected every event object. Check that the " + f"`{warehouse_delivery_service.EVENTS_TABLE_NAME}` table " + f"matches the expected schema." + ), + ) + flagsmith_experimentation_warehouse_delivery_runs_total.labels( + result="failure" + ).inc() + log.error( + "delivery.all_objects_rejected", + objects__rejected_count=rejected_count, + ) + return + mark_warehouse_delivery_succeeded(connection) flagsmith_experimentation_warehouse_delivery_runs_total.labels( result="success" diff --git a/api/experimentation/warehouse_delivery_service.py b/api/experimentation/warehouse_delivery_service.py index 63a5bbeff923..f6c965adeb6f 100644 --- a/api/experimentation/warehouse_delivery_service.py +++ b/api/experimentation/warehouse_delivery_service.py @@ -4,11 +4,12 @@ import boto3 import clickhouse_connect -import structlog +from clickhouse_connect.driver import httputil from clickhouse_connect.driver.exceptions import ( DatabaseError, OperationalError, ) +from urllib3 import PoolManager from core.network import is_internal_address from experimentation.ingestion_infra_service import AWS_REGION @@ -22,8 +23,6 @@ from experimentation.models import WarehouseConnection -logger = structlog.get_logger("experimentation") - EVENTS_TABLE_NAME = "events" EVENTS_FORMAT = "JSONEachRow" EVENTS_COMPRESSION = "gzip" @@ -44,6 +43,29 @@ INSERT_TIMEOUT_SECONDS = 300 +class _NoRedirectPoolManager(PoolManager): + """The internal-address guard validates the host we dial; following a + redirect would let a permitted host bounce the request, and its event + payload, to an address that was never checked.""" + + def urlopen( # type: ignore[override] + self, + method: str, + url: str, + redirect: bool = True, + **kwargs: "Any", + ) -> "Any": + kwargs["redirect"] = False + return super().urlopen(method, url, **kwargs) + + +@lru_cache(maxsize=1) +def _get_pool_manager() -> PoolManager: + # Shared across delivery clients, as clickhouse-connect's own default pool + # is: the manager pools connections per host and is thread-safe. + return _NoRedirectPoolManager(**httputil.get_pool_manager_options()) + + class DeliveryConfigError(Exception): """The connection's stored configuration cannot be used for delivery.""" @@ -179,6 +201,7 @@ def delivery_client(connection: "WarehouseConnection") -> "Iterator[Client]": secure=secure, connect_timeout=CONNECT_TIMEOUT_SECONDS, send_receive_timeout=INSERT_TIMEOUT_SECONDS, + pool_mgr=_get_pool_manager(), ) try: yield client @@ -191,10 +214,9 @@ def deliver_object(client: "Client", bucket_name: str, s3_key: str) -> int: return the number of rows written. The body is passed through untouched — ClickHouse decompresses and parses - it, so neither happens in this process. The deduplication token is what - absorbs a redelivery, but only on a Replicated table or one configured with - a non-replicated deduplication window; otherwise duplicates reach the table - and the distinct-aware aggregates in the results queries account for them. + it, so neither happens in this process. Delivery is at-least-once, so a + redelivered object duplicates rows; the distinct-aware aggregates in the + results queries account for that. """ body = _get_s3_client().get_object(Bucket=bucket_name, Key=s3_key)["Body"] try: @@ -203,7 +225,6 @@ def deliver_object(client: "Client", bucket_name: str, s3_key: str) -> int: insert_block=body, fmt=EVENTS_FORMAT, compression=EVENTS_COMPRESSION, - settings={"insert_deduplication_token": s3_key}, ) except DatabaseError as exc: if exc.code in OBJECT_LEVEL_ERROR_CODES: diff --git a/api/tests/unit/experimentation/test_tasks.py b/api/tests/unit/experimentation/test_tasks.py index 6859d2cf6518..b52243b9ddd7 100644 --- a/api/tests/unit/experimentation/test_tasks.py +++ b/api/tests/unit/experimentation/test_tasks.py @@ -1,3 +1,4 @@ +import itertools from collections.abc import Iterator from dataclasses import asdict from datetime import datetime, timedelta @@ -866,8 +867,10 @@ def test_deliver_events_for_connection__pending_objects__delivers_archives_and_r delivery_bucket: Any, warehouse_client: Any, log: StructuredLogCapture, + mocker: MockerFixture, ) -> None: # Given an errored connection with two objects waiting + move_object_spy = mocker.spy(warehouse_delivery_service, "move_object") clickhouse_connection.status = WarehouseConnectionStatus.ERRORED clickhouse_connection.status_detail = "Authentication failed" clickhouse_connection.save() @@ -886,10 +889,7 @@ def test_deliver_events_for_connection__pending_objects__delivers_archives_and_r # Then both objects are inserted and archived, oldest first raw_insert = warehouse_client.return_value.raw_insert assert raw_insert.call_count == 2 - assert [ - call.kwargs["settings"]["insert_deduplication_token"] - for call in raw_insert.call_args_list - ] == [ + assert [call.args[1] for call in move_object_spy.call_args_list] == [ _pending_key(environment.api_key, hour="13"), _pending_key(environment.api_key, hour="14"), ] @@ -918,6 +918,7 @@ def test_deliver_events_for_connection__pending_objects__delivers_archives_and_r assert { "level": "info", "event": "delivery.completed", + "connection__id": clickhouse_connection.id, "environment__id": environment.id, "organisation__id": environment.project.organisation_id, "objects__count": 2, @@ -982,6 +983,7 @@ def test_deliver_events_for_connection__rejected_object__moves_to_failed_and_con assert { "level": "info", "event": "delivery.completed", + "connection__id": clickhouse_connection.id, "environment__id": environment.id, "organisation__id": environment.project.organisation_id, "objects__count": 1, @@ -1029,3 +1031,86 @@ def test_deliver_events_for_connection__warehouse_unusable__aborts_and_marks_err assert clickhouse_connection.status_detail == "Authentication failed." assert _delivery_runs_count("failure") == failure_runs_before + 1 assert log.has("delivery.failed", level="error") + + +def test_deliver_events_for_connection__every_object_rejected__marks_errored( + clickhouse_connection: WarehouseConnection, + environment: Environment, + ingestion_infrastructure: OrganisationIngestionInfrastructure, + delivery_bucket: Any, + warehouse_client: Any, + log: StructuredLogCapture, +) -> None: + # Given a warehouse that rejects the contents of every object, as a table + # whose schema does not match the payload would + for hour in ("13", "14"): + delivery_bucket.put_object( + Bucket=DELIVERY_BUCKET_NAME, + Key=_pending_key(environment.api_key, hour=hour), + Body=b"gzipped-events", + ) + warehouse_client.return_value.raw_insert.side_effect = DatabaseError( + "Cannot parse DateTime", + code=41, + ) + failure_runs_before = _delivery_runs_count("failure") + + # When + deliver_events_for_connection(connection_id=clickhouse_connection.id) + + # Then the run is a failure rather than a success, so a schema mismatch is + # visible instead of draining silently into failed/ + clickhouse_connection.refresh_from_db() + assert clickhouse_connection.status == WarehouseConnectionStatus.ERRORED + assert clickhouse_connection.status_detail == ( + "The warehouse rejected every event object. Check that the `events` " + "table matches the expected schema." + ) + assert _delivery_runs_count("failure") == failure_runs_before + 1 + assert log.has("delivery.all_objects_rejected", level="error") + + +def test_deliver_events_for_connection__time_budget_exhausted__defers_remaining( + clickhouse_connection: WarehouseConnection, + environment: Environment, + ingestion_infrastructure: OrganisationIngestionInfrastructure, + delivery_bucket: Any, + warehouse_client: Any, + log: StructuredLogCapture, + mocker: MockerFixture, +) -> None: + # Given three objects waiting and a budget that expires after the first, + # so the run cannot outlive its task timeout + for hour in ("13", "14", "15"): + delivery_bucket.put_object( + Bucket=DELIVERY_BUCKET_NAME, + Key=_pending_key(environment.api_key, hour=hour), + Body=b"gzipped-events", + ) + # The clock is shared with pytest and Django teardown, so it must keep + # answering after the budget is spent. + mocker.patch( + "experimentation.services.time.monotonic", + side_effect=itertools.chain([0.0, 0.0], itertools.repeat(1_000.0)), + ) + + # When + deliver_events_for_connection(connection_id=clickhouse_connection.id) + + # Then only the first object is delivered and the rest wait for the next run + assert warehouse_client.return_value.raw_insert.call_count == 1 + assert warehouse_delivery_service.list_pending_objects( + DELIVERY_BUCKET_NAME, + environment_key=environment.api_key, + ) == [ + _pending_key(environment.api_key, hour="14"), + _pending_key(environment.api_key, hour="15"), + ] + assert { + "level": "info", + "event": "delivery.budget_exhausted", + "connection__id": clickhouse_connection.id, + "environment__id": environment.id, + "organisation__id": environment.project.organisation_id, + "objects__remaining_count": 2, + } in log.events diff --git a/api/tests/unit/experimentation/test_warehouse_delivery_service.py b/api/tests/unit/experimentation/test_warehouse_delivery_service.py index 95a90e24ed3b..ec8d3da8c244 100644 --- a/api/tests/unit/experimentation/test_warehouse_delivery_service.py +++ b/api/tests/unit/experimentation/test_warehouse_delivery_service.py @@ -7,6 +7,7 @@ from clickhouse_connect.driver.exceptions import DatabaseError, OperationalError from moto import mock_s3 # type: ignore[import-untyped] from pytest_mock import MockerFixture +from urllib3 import PoolManager from experimentation import warehouse_delivery_service from experimentation.models import WarehouseConnection @@ -177,6 +178,14 @@ def test_delivery_client__valid_config__yields_http_client_and_closes( secure=True, connect_timeout=10, send_receive_timeout=300, + pool_mgr=mocker.ANY, + ) + # Redirects must not be followed: the internal-address guard only + # validates the host being dialled. + pool_manager = get_client.call_args.kwargs["pool_mgr"] + assert isinstance( + pool_manager, + warehouse_delivery_service._NoRedirectPoolManager, ) get_client.return_value.close.assert_not_called() @@ -206,7 +215,6 @@ def test_deliver_object__valid_object__streams_body_and_returns_written_rows( insert_block=mocker.ANY, fmt="JSONEachRow", compression="gzip", - settings={"insert_deduplication_token": _event_key("13")}, ) # The body reaches ClickHouse as the exact bytes stored in S3 insert_block = client.raw_insert.call_args.kwargs["insert_block"] @@ -305,3 +313,19 @@ def test_describe_delivery_error__known_failures__returns_user_facing_detail( # Then assert detail == expected_detail + + +def test_no_redirect_pool_manager__urlopen__refuses_to_follow_redirects( + mocker: MockerFixture, +) -> None: + # 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() + + # When + manager.urlopen("POST", "https://ch.acme-corp.example/", redirect=True) + + # Then the redirect is refused: a permitted host must not be able to bounce + # the request, and its event payload, to an unchecked address + assert urlopen.call_args.kwargs["redirect"] is False From 7b0440d6e1df47cd907748d023843b52ff4e6310 Mon Sep 17 00:00:00 2001 From: Gagan Trivedi Date: Wed, 29 Jul 2026 13:15:00 +0530 Subject: [PATCH 6/9] fix(experimentation): log rejected delivery objects at error level A rejected object means data loss for the customer's warehouse until someone inspects the failed/ prefix, so it warrants an error, not a warning. --- api/experimentation/services.py | 2 +- api/tests/unit/experimentation/test_tasks.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/api/experimentation/services.py b/api/experimentation/services.py index 6627316f11f1..f11f4c448b4c 100644 --- a/api/experimentation/services.py +++ b/api/experimentation/services.py @@ -874,7 +874,7 @@ def _deliver_pending_objects( flagsmith_experimentation_warehouse_delivery_objects_total.labels( result="rejected" ).inc() - log.warning( + log.error( "delivery.object_rejected", s3__key=s3_key, exc_info=True, diff --git a/api/tests/unit/experimentation/test_tasks.py b/api/tests/unit/experimentation/test_tasks.py index b52243b9ddd7..f0115cbb0717 100644 --- a/api/tests/unit/experimentation/test_tasks.py +++ b/api/tests/unit/experimentation/test_tasks.py @@ -979,7 +979,7 @@ def test_deliver_events_for_connection__rejected_object__moves_to_failed_and_con clickhouse_connection.refresh_from_db() assert clickhouse_connection.status == WarehouseConnectionStatus.CONNECTED assert _delivery_objects_count("rejected") == rejected_objects_before + 1 - assert log.has("delivery.object_rejected", level="warning") + assert log.has("delivery.object_rejected", level="error") assert { "level": "info", "event": "delivery.completed", From d8551c534a01604244804da1353abaf64ca49961 Mon Sep 17 00:00:00 2001 From: Gagan Trivedi Date: Wed, 29 Jul 2026 13:17:19 +0530 Subject: [PATCH 7/9] test(experimentation): cover client close on delivery failure --- .../test_warehouse_delivery_service.py | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/api/tests/unit/experimentation/test_warehouse_delivery_service.py b/api/tests/unit/experimentation/test_warehouse_delivery_service.py index ec8d3da8c244..d6eb84dd2921 100644 --- a/api/tests/unit/experimentation/test_warehouse_delivery_service.py +++ b/api/tests/unit/experimentation/test_warehouse_delivery_service.py @@ -192,6 +192,24 @@ def test_delivery_client__valid_config__yields_http_client_and_closes( get_client.return_value.close.assert_called_once_with() +def test_delivery_client__body_raises__still_closes_client( + clickhouse_connection: WarehouseConnection, + mocker: MockerFixture, +) -> None: + # Given + get_client = mocker.patch( + "experimentation.warehouse_delivery_service.clickhouse_connect.get_client", + ) + + # When a delivery inside the block fails + with pytest.raises(RuntimeError, match="boom"): + with warehouse_delivery_service.delivery_client(clickhouse_connection): + raise RuntimeError("boom") + + # Then the pooled HTTP connection is still released + get_client.return_value.close.assert_called_once_with() + + def test_deliver_object__valid_object__streams_body_and_returns_written_rows( events_bucket: Any, mocker: MockerFixture, From eacbf2d4551234d10a9d88e1b0608db07212830c Mon Sep 17 00:00:00 2001 From: Gagan Trivedi Date: Wed, 29 Jul 2026 14:45:36 +0530 Subject: [PATCH 8/9] fix(experimentation): close S3 body after delivery insert --- .../warehouse_delivery_service.py | 3 +++ .../test_warehouse_delivery_service.py | 16 ++++++++++++---- 2 files changed, 15 insertions(+), 4 deletions(-) diff --git a/api/experimentation/warehouse_delivery_service.py b/api/experimentation/warehouse_delivery_service.py index f6c965adeb6f..388281004f6e 100644 --- a/api/experimentation/warehouse_delivery_service.py +++ b/api/experimentation/warehouse_delivery_service.py @@ -230,4 +230,7 @@ def deliver_object(client: "Client", bucket_name: str, s3_key: str) -> int: if exc.code in OBJECT_LEVEL_ERROR_CODES: raise ObjectRejectedError(str(exc)) from exc raise + finally: + # Releases the pooled S3 connection if the insert failed mid-read. + body.close() return summary.written_rows diff --git a/api/tests/unit/experimentation/test_warehouse_delivery_service.py b/api/tests/unit/experimentation/test_warehouse_delivery_service.py index d6eb84dd2921..84c58321848c 100644 --- a/api/tests/unit/experimentation/test_warehouse_delivery_service.py +++ b/api/tests/unit/experimentation/test_warehouse_delivery_service.py @@ -217,7 +217,13 @@ def test_deliver_object__valid_object__streams_body_and_returns_written_rows( # Given events_bucket.put_object(Bucket=BUCKET_NAME, Key=_event_key("13"), Body=OBJECT_BODY) client = mocker.MagicMock() - client.raw_insert.return_value.written_rows = 203 + streamed_bodies: list[bytes] = [] + + def raw_insert(*args: Any, **kwargs: Any) -> Any: + streamed_bodies.append(kwargs["insert_block"].read()) + return mocker.Mock(written_rows=203) + + client.raw_insert.side_effect = raw_insert # When written_rows = warehouse_delivery_service.deliver_object( @@ -234,9 +240,11 @@ def test_deliver_object__valid_object__streams_body_and_returns_written_rows( fmt="JSONEachRow", compression="gzip", ) - # The body reaches ClickHouse as the exact bytes stored in S3 - insert_block = client.raw_insert.call_args.kwargs["insert_block"] - assert insert_block.read() == OBJECT_BODY + # The body reaches ClickHouse as the exact bytes stored in S3, and is + # closed once delivered + assert streamed_bodies == [OBJECT_BODY] + with pytest.raises(ValueError, match="closed file"): + client.raw_insert.call_args.kwargs["insert_block"].read() def test_deliver_object__object_level_error__raises_object_rejected( From 2942db8b25053cab1f9b1e01bb8d552e79250f23 Mon Sep 17 00:00:00 2001 From: "flagsmith-engineering[bot]" Date: Wed, 29 Jul 2026 09:40:57 +0000 Subject: [PATCH 9/9] chore: Update documentation artefacts --- .../observability/_events-catalogue.md | 76 ++++++++++++++++--- 1 file changed, 67 insertions(+), 9 deletions(-) diff --git a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md index c1fd5572e683..8e695cb526d4 100644 --- a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md @@ -91,7 +91,7 @@ Attributes: ### `experimentation.exposures.compute_failed` Logged at `error` from: - - `api/experimentation/tasks.py:133` + - `api/experimentation/tasks.py:181` Attributes: - `environment.id` @@ -159,7 +159,7 @@ Attributes: ### `experimentation.results.compute_failed` Logged at `error` from: - - `api/experimentation/tasks.py:169` + - `api/experimentation/tasks.py:217` Attributes: - `environment.id` @@ -631,7 +631,7 @@ Attributes: ### `warehouse.connection.connected` Logged at `info` from: - - `api/experimentation/services.py:911` + - `api/experimentation/services.py:1075` Attributes: - `environment.id` @@ -640,7 +640,7 @@ Attributes: ### `warehouse.connection.event_stats_failed` Logged at `warning` from: - - `api/experimentation/services.py:970` + - `api/experimentation/services.py:1134` Attributes: - `environment.id` @@ -649,7 +649,7 @@ Attributes: ### `warehouse.connection.test_event_sent` Logged at `info` from: - - `api/experimentation/services.py:798` + - `api/experimentation/services.py:809` Attributes: - `environment.id` @@ -658,7 +658,7 @@ Attributes: ### `warehouse.connection.verification_failed` Logged at `warning` from: - - `api/experimentation/services.py:886` + - `api/experimentation/services.py:1050` Attributes: - `environment.id` @@ -668,16 +668,74 @@ Attributes: ### `warehouse.connection.verification_succeeded` Logged at `info` from: - - `api/experimentation/services.py:896` + - `api/experimentation/services.py:1060` Attributes: - `environment.id` - `organisation.id` +### `warehouse.delivery.all_objects_rejected` + +Logged at `error` from: + - `api/experimentation/services.py:952` + +Attributes: + - `connection.id` + - `environment.id` + - `objects.rejected_count` + - `organisation.id` + +### `warehouse.delivery.budget_exhausted` + +Logged at `info` from: + - `api/experimentation/services.py:854` + +Attributes: + - `connection.id` + - `environment.id` + - `objects.remaining_count` + - `organisation.id` + +### `warehouse.delivery.completed` + +Logged at `info` from: + - `api/experimentation/services.py:962` + +Attributes: + - `connection.id` + - `environment.id` + - `objects.count` + - `objects.rejected_count` + - `organisation.id` + - `rows.count` + +### `warehouse.delivery.failed` + +Logged at `error` from: + - `api/experimentation/services.py:935` + +Attributes: + - `connection.id` + - `environment.id` + - `exc_info` + - `organisation.id` + +### `warehouse.delivery.object_rejected` + +Logged at `error` from: + - `api/experimentation/services.py:877` + +Attributes: + - `connection.id` + - `environment.id` + - `exc_info` + - `organisation.id` + - `s3.key` + ### `warehouse.srm.overallocated` Logged at `error` from: - - `api/experimentation/services.py:420` + - `api/experimentation/services.py:431` Attributes: - `environment.id` @@ -687,7 +745,7 @@ Attributes: ### `warehouse.srm.unkeyed_variant` Logged at `error` from: - - `api/experimentation/services.py:406` + - `api/experimentation/services.py:417` Attributes: - `environment.id`