From 006e8592df222d1e9bf2cd1779633196094c80ec Mon Sep 17 00:00:00 2001 From: Gagan Trivedi Date: Fri, 17 Jul 2026 15:22:20 +0530 Subject: [PATCH 1/7] feat(experimentation): add per-organisation ingestion infrastructure service --- api/app/settings/common.py | 8 + api/experimentation/dataclasses.py | 6 + .../ingestion_infra_service.py | 187 +++++++++++++++ .../test_ingestion_infra_service.py | 223 ++++++++++++++++++ .../observability/_events-catalogue.md | 19 ++ 5 files changed, 443 insertions(+) create mode 100644 api/experimentation/ingestion_infra_service.py create mode 100644 api/tests/unit/experimentation/test_ingestion_infra_service.py diff --git a/api/app/settings/common.py b/api/app/settings/common.py index 5adde3332224..0972a41fd9eb 100644 --- a/api/app/settings/common.py +++ b/api/app/settings/common.py @@ -806,6 +806,14 @@ # Redis Cluster URL used to communicate with the event ingestion server. INGESTION_REDIS_URL = env.str("INGESTION_REDIS_URL", default="") +# Name prefix for per-organisation experiment events S3 buckets. +INGESTION_EVENTS_BUCKET_PREFIX = env.str("INGESTION_EVENTS_BUCKET_PREFIX", default="") + +# ARN of the IAM role Firehose assumes to deliver experiment events to S3. +INGESTION_FIREHOSE_DELIVERY_ROLE_ARN = env.str( + "INGESTION_FIREHOSE_DELIVERY_ROLE_ARN", default="" +) + CACHES = { "default": { "BACKEND": "django.core.cache.backends.locmem.LocMemCache", diff --git a/api/experimentation/dataclasses.py b/api/experimentation/dataclasses.py index d8db1624ed34..1766ea5aeb44 100644 --- a/api/experimentation/dataclasses.py +++ b/api/experimentation/dataclasses.py @@ -80,3 +80,9 @@ class MetricResult: class ResultsSummary: srm_p_value: float | None metrics: list[MetricResult] + + +@dataclass(frozen=True) +class IngestionInfrastructure: + bucket_name: str + stream_name: str diff --git a/api/experimentation/ingestion_infra_service.py b/api/experimentation/ingestion_infra_service.py new file mode 100644 index 000000000000..30006b2f88c8 --- /dev/null +++ b/api/experimentation/ingestion_infra_service.py @@ -0,0 +1,187 @@ +from __future__ import annotations + +import typing +from functools import lru_cache + +import boto3 +import structlog +from botocore.exceptions import ClientError +from django.conf import settings +from django.core.exceptions import ImproperlyConfigured + +from experimentation.dataclasses import IngestionInfrastructure + +if typing.TYPE_CHECKING: + from typing import Any + +logger = structlog.get_logger("experimentation") + +AWS_REGION = "eu-west-2" +STREAM_NAME_PREFIX = "events-ingestion-org-" + +# S3 object keys are namespaced per environment via Firehose dynamic +# partitioning on each record's environment_key field. +EVENTS_PREFIX = ( + "events/env_key=!{partitionKeyFromQuery:env_key}/" + "year=!{timestamp:yyyy}/month=!{timestamp:MM}/" + "day=!{timestamp:dd}/hour=!{timestamp:HH}/" +) +ERRORS_PREFIX = ( + "errors/!{firehose:error-output-type}/" + "year=!{timestamp:yyyy}/month=!{timestamp:MM}/" + "day=!{timestamp:dd}/hour=!{timestamp:HH}/" +) + +# Firehose requires buffers of at least 64 MiB when dynamic partitioning is +# enabled. +BUFFERING_SIZE_MB = 64 +BUFFERING_INTERVAL_SECONDS = 300 +DYNAMIC_PARTITIONING_RETRY_SECONDS = 300 +ERROR_OBJECT_EXPIRATION_DAYS = 30 + + +@lru_cache(maxsize=1) +def _get_s3_client() -> "Any": + return boto3.client("s3", region_name=AWS_REGION) + + +@lru_cache(maxsize=1) +def _get_firehose_client() -> "Any": + return boto3.client("firehose", region_name=AWS_REGION) + + +def get_bucket_name(organisation_id: int) -> str: + if not settings.INGESTION_EVENTS_BUCKET_PREFIX: + raise ImproperlyConfigured("INGESTION_EVENTS_BUCKET_PREFIX is not set") + return f"{settings.INGESTION_EVENTS_BUCKET_PREFIX}-org-{organisation_id}" + + +def get_stream_name(organisation_id: int) -> str: + return f"{STREAM_NAME_PREFIX}{organisation_id}" + + +def _ensure_events_bucket(bucket_name: str, *, organisation_id: int) -> None: + s3 = _get_s3_client() + try: + s3.create_bucket( + Bucket=bucket_name, + CreateBucketConfiguration={"LocationConstraint": AWS_REGION}, + ) + except ClientError as exc: + if exc.response["Error"]["Code"] != "BucketAlreadyOwnedByYou": + raise + else: + logger.info( + "ingestion_infra.bucket_created", + organisation__id=organisation_id, + bucket__name=bucket_name, + ) + s3.put_public_access_block( + Bucket=bucket_name, + PublicAccessBlockConfiguration={ + "BlockPublicAcls": True, + "IgnorePublicAcls": True, + "BlockPublicPolicy": True, + "RestrictPublicBuckets": True, + }, + ) + s3.put_bucket_lifecycle_configuration( + Bucket=bucket_name, + LifecycleConfiguration={ + "Rules": [ + { + "ID": "expire-delivery-errors", + "Filter": {"Prefix": "errors/"}, + "Status": "Enabled", + "Expiration": {"Days": ERROR_OBJECT_EXPIRATION_DAYS}, + } + ] + }, + ) + + +def _expected_destination_configuration(bucket_name: str) -> dict[str, Any]: + return { + "RoleARN": settings.INGESTION_FIREHOSE_DELIVERY_ROLE_ARN, + "BucketARN": f"arn:aws:s3:::{bucket_name}", + "Prefix": EVENTS_PREFIX, + "ErrorOutputPrefix": ERRORS_PREFIX, + "BufferingHints": { + "SizeInMBs": BUFFERING_SIZE_MB, + "IntervalInSeconds": BUFFERING_INTERVAL_SECONDS, + }, + "CompressionFormat": "GZIP", + "DynamicPartitioningConfiguration": { + "Enabled": True, + "RetryOptions": {"DurationInSeconds": DYNAMIC_PARTITIONING_RETRY_SECONDS}, + }, + "ProcessingConfiguration": { + "Enabled": True, + "Processors": [ + { + "Type": "MetadataExtraction", + "Parameters": [ + { + "ParameterName": "MetadataExtractionQuery", + "ParameterValue": "{env_key:.environment_key}", + }, + { + "ParameterName": "JsonParsingEngine", + "ParameterValue": "JQ-1.6", + }, + ], + }, + { + "Type": "AppendDelimiterToRecord", + "Parameters": [ + {"ParameterName": "Delimiter", "ParameterValue": "\\n"}, + ], + }, + ], + }, + } + + +def _ensure_delivery_stream( + stream_name: str, + *, + bucket_name: str, + organisation_id: int, +) -> None: + try: + _get_firehose_client().create_delivery_stream( + DeliveryStreamName=stream_name, + DeliveryStreamType="DirectPut", + ExtendedS3DestinationConfiguration=_expected_destination_configuration( + bucket_name + ), + ) + except ClientError as exc: + if exc.response["Error"]["Code"] != "ResourceInUseException": + raise + else: + logger.info( + "ingestion_infra.stream_created", + organisation__id=organisation_id, + stream__name=stream_name, + bucket__name=bucket_name, + ) + + +def provision_ingestion_infrastructure( + organisation_id: int, +) -> IngestionInfrastructure: + """Idempotently create the organisation's events S3 bucket and Firehose + delivery stream; existing resources are left untouched.""" + if not settings.INGESTION_FIREHOSE_DELIVERY_ROLE_ARN: + raise ImproperlyConfigured("INGESTION_FIREHOSE_DELIVERY_ROLE_ARN is not set") + bucket_name = get_bucket_name(organisation_id) + stream_name = get_stream_name(organisation_id) + + _ensure_events_bucket(bucket_name, organisation_id=organisation_id) + _ensure_delivery_stream( + stream_name, + bucket_name=bucket_name, + organisation_id=organisation_id, + ) + return IngestionInfrastructure(bucket_name=bucket_name, stream_name=stream_name) diff --git a/api/tests/unit/experimentation/test_ingestion_infra_service.py b/api/tests/unit/experimentation/test_ingestion_infra_service.py new file mode 100644 index 000000000000..dee3686c338d --- /dev/null +++ b/api/tests/unit/experimentation/test_ingestion_infra_service.py @@ -0,0 +1,223 @@ +from collections.abc import Iterator +from typing import Any + +import boto3 +import pytest +from botocore.exceptions import ClientError +from django.core.exceptions import ImproperlyConfigured +from moto import mock_firehose, mock_s3 # type: ignore[import-untyped] +from pytest_django.fixtures import SettingsWrapper +from pytest_mock import MockerFixture +from pytest_structlog import StructuredLogCapture + +from experimentation import ingestion_infra_service +from experimentation.dataclasses import IngestionInfrastructure + +DELIVERY_ROLE_ARN = "arn:aws:iam::123456789012:role/firehose-events-delivery" + + +@pytest.fixture() +def ingestion_infra_settings(settings: SettingsWrapper) -> SettingsWrapper: + settings.INGESTION_EVENTS_BUCKET_PREFIX = "flagsmith-events-lake-test" + settings.INGESTION_FIREHOSE_DELIVERY_ROLE_ARN = DELIVERY_ROLE_ARN + return settings + + +@pytest.fixture() +def aws_backends(aws_credentials: None) -> Iterator[None]: + ingestion_infra_service._get_s3_client.cache_clear() + ingestion_infra_service._get_firehose_client.cache_clear() + with mock_s3(), mock_firehose(): + yield + ingestion_infra_service._get_s3_client.cache_clear() + ingestion_infra_service._get_firehose_client.cache_clear() + + +def test_provision_ingestion_infrastructure__no_bucket_prefix__raises_improperly_configured( + ingestion_infra_settings: SettingsWrapper, +) -> None: + # Given + ingestion_infra_settings.INGESTION_EVENTS_BUCKET_PREFIX = "" + + # When / Then + with pytest.raises(ImproperlyConfigured): + ingestion_infra_service.provision_ingestion_infrastructure(organisation_id=42) + + +def test_provision_ingestion_infrastructure__no_delivery_role_arn__raises_improperly_configured( + ingestion_infra_settings: SettingsWrapper, +) -> None: + # Given + ingestion_infra_settings.INGESTION_FIREHOSE_DELIVERY_ROLE_ARN = "" + + # When / Then + with pytest.raises(ImproperlyConfigured): + ingestion_infra_service.provision_ingestion_infrastructure(organisation_id=42) + + +def test_provision_ingestion_infrastructure__fresh_account__creates_bucket_and_stream( + ingestion_infra_settings: SettingsWrapper, + aws_backends: None, + log: StructuredLogCapture, +) -> None: + # When + result = ingestion_infra_service.provision_ingestion_infrastructure( + organisation_id=42, + ) + + # Then + assert result == IngestionInfrastructure( + bucket_name="flagsmith-events-lake-test-org-42", + stream_name="events-ingestion-org-42", + ) + + s3 = boto3.client("s3", region_name="eu-west-2") + public_access_block = s3.get_public_access_block(Bucket=result.bucket_name) + assert public_access_block["PublicAccessBlockConfiguration"] == { + "BlockPublicAcls": True, + "IgnorePublicAcls": True, + "BlockPublicPolicy": True, + "RestrictPublicBuckets": True, + } + lifecycle = s3.get_bucket_lifecycle_configuration(Bucket=result.bucket_name) + assert lifecycle["Rules"] == [ + { + "ID": "expire-delivery-errors", + "Filter": {"Prefix": "errors/"}, + "Status": "Enabled", + "Expiration": {"Days": 30}, + } + ] + + firehose = boto3.client("firehose", region_name="eu-west-2") + stream = firehose.describe_delivery_stream(DeliveryStreamName=result.stream_name)[ + "DeliveryStreamDescription" + ] + assert stream["DeliveryStreamType"] == "DirectPut" + destination = stream["Destinations"][0]["ExtendedS3DestinationDescription"] + assert destination["RoleARN"] == DELIVERY_ROLE_ARN + assert destination["BucketARN"] == "arn:aws:s3:::flagsmith-events-lake-test-org-42" + assert destination["Prefix"] == ( + "events/env_key=!{partitionKeyFromQuery:env_key}/" + "year=!{timestamp:yyyy}/month=!{timestamp:MM}/" + "day=!{timestamp:dd}/hour=!{timestamp:HH}/" + ) + assert destination["ErrorOutputPrefix"] == ( + "errors/!{firehose:error-output-type}/" + "year=!{timestamp:yyyy}/month=!{timestamp:MM}/" + "day=!{timestamp:dd}/hour=!{timestamp:HH}/" + ) + assert destination["CompressionFormat"] == "GZIP" + assert destination["BufferingHints"] == { + "SizeInMBs": 64, + "IntervalInSeconds": 300, + } + assert destination["DynamicPartitioningConfiguration"] == { + "Enabled": True, + "RetryOptions": {"DurationInSeconds": 300}, + } + assert destination["ProcessingConfiguration"] == { + "Enabled": True, + "Processors": [ + { + "Type": "MetadataExtraction", + "Parameters": [ + { + "ParameterName": "MetadataExtractionQuery", + "ParameterValue": "{env_key:.environment_key}", + }, + { + "ParameterName": "JsonParsingEngine", + "ParameterValue": "JQ-1.6", + }, + ], + }, + { + "Type": "AppendDelimiterToRecord", + "Parameters": [ + {"ParameterName": "Delimiter", "ParameterValue": "\\n"}, + ], + }, + ], + } + + assert log.events == [ + { + "level": "info", + "event": "ingestion_infra.bucket_created", + "organisation__id": 42, + "bucket__name": "flagsmith-events-lake-test-org-42", + }, + { + "level": "info", + "event": "ingestion_infra.stream_created", + "organisation__id": 42, + "stream__name": "events-ingestion-org-42", + "bucket__name": "flagsmith-events-lake-test-org-42", + }, + ] + + +def test_provision_ingestion_infrastructure__already_provisioned__is_idempotent( + ingestion_infra_settings: SettingsWrapper, + aws_backends: None, + log: StructuredLogCapture, +) -> None: + # Given + first = ingestion_infra_service.provision_ingestion_infrastructure( + organisation_id=42, + ) + events_after_first_run = list(log.events) + + # When + second = ingestion_infra_service.provision_ingestion_infrastructure( + organisation_id=42, + ) + + # Then + assert second == first + assert log.events == events_after_first_run + + +def test_provision_ingestion_infrastructure__bucket_creation_fails__propagates_client_error( + ingestion_infra_settings: SettingsWrapper, + mocker: MockerFixture, +) -> None: + # Given + mock_s3 = mocker.Mock() + mock_s3.create_bucket.side_effect = ClientError( + {"Error": {"Code": "AccessDenied", "Message": "nope"}}, + "CreateBucket", + ) + mocker.patch( + "experimentation.ingestion_infra_service._get_s3_client", + return_value=mock_s3, + ) + + # When / Then + with pytest.raises(ClientError, match="AccessDenied"): + ingestion_infra_service.provision_ingestion_infrastructure(organisation_id=42) + + +def test_provision_ingestion_infrastructure__stream_creation_fails__propagates_client_error( + ingestion_infra_settings: SettingsWrapper, + mocker: MockerFixture, +) -> None: + # Given + mocker.patch( + "experimentation.ingestion_infra_service._get_s3_client", + return_value=mocker.Mock(), + ) + mock_firehose_client: Any = mocker.Mock() + mock_firehose_client.create_delivery_stream.side_effect = ClientError( + {"Error": {"Code": "LimitExceededException", "Message": "too many"}}, + "CreateDeliveryStream", + ) + mocker.patch( + "experimentation.ingestion_infra_service._get_firehose_client", + return_value=mock_firehose_client, + ) + + # When / Then + with pytest.raises(ClientError, match="LimitExceededException"): + ingestion_infra_service.provision_ingestion_infrastructure(organisation_id=42) diff --git a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md index 40fe1dd98bb8..68a1c2f5e546 100644 --- a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md @@ -92,6 +92,25 @@ Attributes: - `feature.id` - `organisation.id` +### `experimentation.ingestion_infra.bucket_created` + +Logged at `info` from: + - `api/experimentation/ingestion_infra_service.py:74` + +Attributes: + - `bucket.name` + - `organisation.id` + +### `experimentation.ingestion_infra.stream_created` + +Logged at `info` from: + - `api/experimentation/ingestion_infra_service.py:163` + +Attributes: + - `bucket.name` + - `organisation.id` + - `stream.name` + ### `experimentation.results.compute_failed` Logged at `error` from: From 47df807a34fa4b0bd59bb1ef633179c9917b3224 Mon Sep 17 00:00:00 2001 From: Gagan Trivedi Date: Mon, 20 Jul 2026 12:10:17 +0530 Subject: [PATCH 2/7] feat(experimentation): create events buckets in the account regional namespace Derive bucket names as {prefix}-{account_id}-{region}-an and opt in via the x-amz-bucket-namespace header, replacing the INGESTION_EVENTS_BUCKET_PREFIX setting. --- api/app/settings/common.py | 3 - .../ingestion_infra_service.py | 31 ++++++-- .../test_ingestion_infra_service.py | 72 +++++++++++++------ .../observability/_events-catalogue.md | 4 +- 4 files changed, 79 insertions(+), 31 deletions(-) diff --git a/api/app/settings/common.py b/api/app/settings/common.py index 0972a41fd9eb..859ea6c32dae 100644 --- a/api/app/settings/common.py +++ b/api/app/settings/common.py @@ -806,9 +806,6 @@ # Redis Cluster URL used to communicate with the event ingestion server. INGESTION_REDIS_URL = env.str("INGESTION_REDIS_URL", default="") -# Name prefix for per-organisation experiment events S3 buckets. -INGESTION_EVENTS_BUCKET_PREFIX = env.str("INGESTION_EVENTS_BUCKET_PREFIX", default="") - # ARN of the IAM role Firehose assumes to deliver experiment events to S3. INGESTION_FIREHOSE_DELIVERY_ROLE_ARN = env.str( "INGESTION_FIREHOSE_DELIVERY_ROLE_ARN", default="" diff --git a/api/experimentation/ingestion_infra_service.py b/api/experimentation/ingestion_infra_service.py index 30006b2f88c8..0f140f35c8e9 100644 --- a/api/experimentation/ingestion_infra_service.py +++ b/api/experimentation/ingestion_infra_service.py @@ -18,6 +18,11 @@ AWS_REGION = "eu-west-2" STREAM_NAME_PREFIX = "events-ingestion-org-" +# Buckets are created in the account regional namespace, so the name only has +# to be unique within our account and must follow the +# {prefix}-{account_id}-{region}-an convention. +BUCKET_NAME_PREFIX = "flagsmith-events-lake-org-" +BUCKET_NAMESPACE_HEADER = "x-amz-bucket-namespace" # S3 object keys are namespaced per environment via Firehose dynamic # partitioning on each record's environment_key field. @@ -40,9 +45,23 @@ ERROR_OBJECT_EXPIRATION_DAYS = 30 +def _add_account_regional_namespace_header( + params: dict[str, Any], + **kwargs: Any, +) -> None: + params["headers"][BUCKET_NAMESPACE_HEADER] = "account-regional" + + @lru_cache(maxsize=1) def _get_s3_client() -> "Any": - return boto3.client("s3", region_name=AWS_REGION) + # The pinned boto3 version predates the CreateBucket BucketNamespace + # parameter, so the corresponding header is injected directly. + client = boto3.client("s3", region_name=AWS_REGION) + client.meta.events.register( + "before-call.s3.CreateBucket", + _add_account_regional_namespace_header, + ) + return client @lru_cache(maxsize=1) @@ -50,10 +69,14 @@ def _get_firehose_client() -> "Any": return boto3.client("firehose", region_name=AWS_REGION) +@lru_cache(maxsize=1) +def _get_account_id() -> str: + sts = boto3.client("sts", region_name=AWS_REGION) + return sts.get_caller_identity()["Account"] # type: ignore[no-any-return] + + def get_bucket_name(organisation_id: int) -> str: - if not settings.INGESTION_EVENTS_BUCKET_PREFIX: - raise ImproperlyConfigured("INGESTION_EVENTS_BUCKET_PREFIX is not set") - return f"{settings.INGESTION_EVENTS_BUCKET_PREFIX}-org-{organisation_id}" + return f"{BUCKET_NAME_PREFIX}{organisation_id}-{_get_account_id()}-{AWS_REGION}-an" def get_stream_name(organisation_id: int) -> str: diff --git a/api/tests/unit/experimentation/test_ingestion_infra_service.py b/api/tests/unit/experimentation/test_ingestion_infra_service.py index dee3686c338d..a36827139593 100644 --- a/api/tests/unit/experimentation/test_ingestion_infra_service.py +++ b/api/tests/unit/experimentation/test_ingestion_infra_service.py @@ -5,7 +5,11 @@ import pytest from botocore.exceptions import ClientError from django.core.exceptions import ImproperlyConfigured -from moto import mock_firehose, mock_s3 # type: ignore[import-untyped] +from moto import ( # type: ignore[import-untyped] + mock_firehose, + mock_s3, + mock_sts, +) from pytest_django.fixtures import SettingsWrapper from pytest_mock import MockerFixture from pytest_structlog import StructuredLogCapture @@ -18,30 +22,22 @@ @pytest.fixture() def ingestion_infra_settings(settings: SettingsWrapper) -> SettingsWrapper: - settings.INGESTION_EVENTS_BUCKET_PREFIX = "flagsmith-events-lake-test" settings.INGESTION_FIREHOSE_DELIVERY_ROLE_ARN = DELIVERY_ROLE_ARN return settings -@pytest.fixture() -def aws_backends(aws_credentials: None) -> Iterator[None]: - ingestion_infra_service._get_s3_client.cache_clear() - ingestion_infra_service._get_firehose_client.cache_clear() - with mock_s3(), mock_firehose(): - yield +def _clear_client_caches() -> None: ingestion_infra_service._get_s3_client.cache_clear() ingestion_infra_service._get_firehose_client.cache_clear() + ingestion_infra_service._get_account_id.cache_clear() -def test_provision_ingestion_infrastructure__no_bucket_prefix__raises_improperly_configured( - ingestion_infra_settings: SettingsWrapper, -) -> None: - # Given - ingestion_infra_settings.INGESTION_EVENTS_BUCKET_PREFIX = "" - - # When / Then - with pytest.raises(ImproperlyConfigured): - ingestion_infra_service.provision_ingestion_infrastructure(organisation_id=42) +@pytest.fixture() +def aws_backends(aws_credentials: None) -> Iterator[None]: + _clear_client_caches() + with mock_s3(), mock_firehose(), mock_sts(): + yield + _clear_client_caches() def test_provision_ingestion_infrastructure__no_delivery_role_arn__raises_improperly_configured( @@ -60,14 +56,17 @@ def test_provision_ingestion_infrastructure__fresh_account__creates_bucket_and_s aws_backends: None, log: StructuredLogCapture, ) -> None: + # Given + organisation_id = 42 + # When result = ingestion_infra_service.provision_ingestion_infrastructure( - organisation_id=42, + organisation_id=organisation_id, ) # Then assert result == IngestionInfrastructure( - bucket_name="flagsmith-events-lake-test-org-42", + bucket_name="flagsmith-events-lake-org-42-123456789012-eu-west-2-an", stream_name="events-ingestion-org-42", ) @@ -96,7 +95,10 @@ def test_provision_ingestion_infrastructure__fresh_account__creates_bucket_and_s assert stream["DeliveryStreamType"] == "DirectPut" destination = stream["Destinations"][0]["ExtendedS3DestinationDescription"] assert destination["RoleARN"] == DELIVERY_ROLE_ARN - assert destination["BucketARN"] == "arn:aws:s3:::flagsmith-events-lake-test-org-42" + assert ( + destination["BucketARN"] + == "arn:aws:s3:::flagsmith-events-lake-org-42-123456789012-eu-west-2-an" + ) assert destination["Prefix"] == ( "events/env_key=!{partitionKeyFromQuery:env_key}/" "year=!{timestamp:yyyy}/month=!{timestamp:MM}/" @@ -146,18 +148,36 @@ def test_provision_ingestion_infrastructure__fresh_account__creates_bucket_and_s "level": "info", "event": "ingestion_infra.bucket_created", "organisation__id": 42, - "bucket__name": "flagsmith-events-lake-test-org-42", + "bucket__name": "flagsmith-events-lake-org-42-123456789012-eu-west-2-an", }, { "level": "info", "event": "ingestion_infra.stream_created", "organisation__id": 42, "stream__name": "events-ingestion-org-42", - "bucket__name": "flagsmith-events-lake-test-org-42", + "bucket__name": "flagsmith-events-lake-org-42-123456789012-eu-west-2-an", }, ] +def test_provision_ingestion_infrastructure__bucket_creation__sends_account_regional_namespace_header( + ingestion_infra_settings: SettingsWrapper, + aws_backends: None, +) -> None: + # Given + captured_headers: list[dict[str, str]] = [] + ingestion_infra_service._get_s3_client().meta.events.register( + "before-call.s3.CreateBucket", + lambda params, **kwargs: captured_headers.append(params["headers"]), + ) + + # When + ingestion_infra_service.provision_ingestion_infrastructure(organisation_id=42) + + # Then + assert captured_headers[0]["x-amz-bucket-namespace"] == "account-regional" + + def test_provision_ingestion_infrastructure__already_provisioned__is_idempotent( ingestion_infra_settings: SettingsWrapper, aws_backends: None, @@ -184,6 +204,10 @@ def test_provision_ingestion_infrastructure__bucket_creation_fails__propagates_c mocker: MockerFixture, ) -> None: # Given + mocker.patch( + "experimentation.ingestion_infra_service._get_account_id", + return_value="123456789012", + ) mock_s3 = mocker.Mock() mock_s3.create_bucket.side_effect = ClientError( {"Error": {"Code": "AccessDenied", "Message": "nope"}}, @@ -204,6 +228,10 @@ def test_provision_ingestion_infrastructure__stream_creation_fails__propagates_c mocker: MockerFixture, ) -> None: # Given + mocker.patch( + "experimentation.ingestion_infra_service._get_account_id", + return_value="123456789012", + ) mocker.patch( "experimentation.ingestion_infra_service._get_s3_client", return_value=mocker.Mock(), diff --git a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md index 68a1c2f5e546..86610f2332e5 100644 --- a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md @@ -95,7 +95,7 @@ Attributes: ### `experimentation.ingestion_infra.bucket_created` Logged at `info` from: - - `api/experimentation/ingestion_infra_service.py:74` + - `api/experimentation/ingestion_infra_service.py:97` Attributes: - `bucket.name` @@ -104,7 +104,7 @@ Attributes: ### `experimentation.ingestion_infra.stream_created` Logged at `info` from: - - `api/experimentation/ingestion_infra_service.py:163` + - `api/experimentation/ingestion_infra_service.py:186` Attributes: - `bucket.name` From b456e8563336f41cd725b093f7c3015e125b2cee Mon Sep 17 00:00:00 2001 From: Gagan Trivedi Date: Mon, 20 Jul 2026 14:46:38 +0530 Subject: [PATCH 3/7] feat(experimentation): expire all events bucket objects after 30 days Replace the errors-only lifecycle rule with one that expires every object in the bucket after 30 days. --- api/experimentation/ingestion_infra_service.py | 8 ++++---- .../unit/experimentation/test_ingestion_infra_service.py | 4 ++-- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/api/experimentation/ingestion_infra_service.py b/api/experimentation/ingestion_infra_service.py index 0f140f35c8e9..7c931a677533 100644 --- a/api/experimentation/ingestion_infra_service.py +++ b/api/experimentation/ingestion_infra_service.py @@ -42,7 +42,7 @@ BUFFERING_SIZE_MB = 64 BUFFERING_INTERVAL_SECONDS = 300 DYNAMIC_PARTITIONING_RETRY_SECONDS = 300 -ERROR_OBJECT_EXPIRATION_DAYS = 30 +OBJECT_EXPIRATION_DAYS = 30 def _add_account_regional_namespace_header( @@ -113,10 +113,10 @@ def _ensure_events_bucket(bucket_name: str, *, organisation_id: int) -> None: LifecycleConfiguration={ "Rules": [ { - "ID": "expire-delivery-errors", - "Filter": {"Prefix": "errors/"}, + "ID": "expire-objects", + "Filter": {"Prefix": ""}, "Status": "Enabled", - "Expiration": {"Days": ERROR_OBJECT_EXPIRATION_DAYS}, + "Expiration": {"Days": OBJECT_EXPIRATION_DAYS}, } ] }, diff --git a/api/tests/unit/experimentation/test_ingestion_infra_service.py b/api/tests/unit/experimentation/test_ingestion_infra_service.py index a36827139593..8bd781c09ded 100644 --- a/api/tests/unit/experimentation/test_ingestion_infra_service.py +++ b/api/tests/unit/experimentation/test_ingestion_infra_service.py @@ -81,8 +81,8 @@ def test_provision_ingestion_infrastructure__fresh_account__creates_bucket_and_s lifecycle = s3.get_bucket_lifecycle_configuration(Bucket=result.bucket_name) assert lifecycle["Rules"] == [ { - "ID": "expire-delivery-errors", - "Filter": {"Prefix": "errors/"}, + "ID": "expire-objects", + "Filter": {"Prefix": ""}, "Status": "Enabled", "Expiration": {"Days": 30}, } From d9291a6da7912f28f0aa13cac9c0cce2ce07b17b Mon Sep 17 00:00:00 2001 From: Gagan Trivedi Date: Mon, 20 Jul 2026 16:09:20 +0530 Subject: [PATCH 4/7] chore(infra): set INGESTION_FIREHOSE_DELIVERY_ROLE_ARN in ECS task definitions Wire the per-stack Firehose delivery role ARN into the admin-api and task-processor task definitions for staging and production. --- .../aws/production/ecs-task-definition-admin-api.json | 4 ++++ .../aws/production/ecs-task-definition-task-processor.json | 4 ++++ infrastructure/aws/staging/ecs-task-definition-admin-api.json | 4 ++++ .../aws/staging/ecs-task-definition-task-processor.json | 4 ++++ 4 files changed, 16 insertions(+) diff --git a/infrastructure/aws/production/ecs-task-definition-admin-api.json b/infrastructure/aws/production/ecs-task-definition-admin-api.json index 32605f19ac9c..b1b59483f9cf 100644 --- a/infrastructure/aws/production/ecs-task-definition-admin-api.json +++ b/infrastructure/aws/production/ecs-task-definition-admin-api.json @@ -232,6 +232,10 @@ { "name": "INGESTION_REDIS_URL", "value": "rediss://serverless-redis-cache-c4q8sw.serverless.euw2.cache.amazonaws.com:6379" + }, + { + "name": "INGESTION_FIREHOSE_DELIVERY_ROLE_ARN", + "value": "arn:aws:iam::084060095745:role/experiment-events-firehose-delivery-external-warehouse" } ], "secrets": [ diff --git a/infrastructure/aws/production/ecs-task-definition-task-processor.json b/infrastructure/aws/production/ecs-task-definition-task-processor.json index a476a1bf12e4..20d340322ab2 100644 --- a/infrastructure/aws/production/ecs-task-definition-task-processor.json +++ b/infrastructure/aws/production/ecs-task-definition-task-processor.json @@ -197,6 +197,10 @@ { "name": "INGESTION_REDIS_URL", "value": "rediss://serverless-redis-cache-c4q8sw.serverless.euw2.cache.amazonaws.com:6379" + }, + { + "name": "INGESTION_FIREHOSE_DELIVERY_ROLE_ARN", + "value": "arn:aws:iam::084060095745:role/experiment-events-firehose-delivery-external-warehouse" } ], "secrets": [ diff --git a/infrastructure/aws/staging/ecs-task-definition-admin-api.json b/infrastructure/aws/staging/ecs-task-definition-admin-api.json index 7afb69722b3b..2282854a091d 100644 --- a/infrastructure/aws/staging/ecs-task-definition-admin-api.json +++ b/infrastructure/aws/staging/ecs-task-definition-admin-api.json @@ -233,6 +233,10 @@ { "name": "INGESTION_REDIS_URL", "value": "rediss://serverless-redis-cache-7u3xil.serverless.euw2.cache.amazonaws.com:6379" + }, + { + "name": "INGESTION_FIREHOSE_DELIVERY_ROLE_ARN", + "value": "arn:aws:iam::302456015006:role/experiment-events-firehose-delivery-external-warehouse" } ], "secrets": [ diff --git a/infrastructure/aws/staging/ecs-task-definition-task-processor.json b/infrastructure/aws/staging/ecs-task-definition-task-processor.json index 7c1ed3f07f94..a81d37ee4d98 100644 --- a/infrastructure/aws/staging/ecs-task-definition-task-processor.json +++ b/infrastructure/aws/staging/ecs-task-definition-task-processor.json @@ -183,6 +183,10 @@ { "name": "INGESTION_REDIS_URL", "value": "rediss://serverless-redis-cache-7u3xil.serverless.euw2.cache.amazonaws.com:6379" + }, + { + "name": "INGESTION_FIREHOSE_DELIVERY_ROLE_ARN", + "value": "arn:aws:iam::302456015006:role/experiment-events-firehose-delivery-external-warehouse" } ], "secrets": [ From 7e3bb42b57c9dea7a9f31e51308ccec7428affe9 Mon Sep 17 00:00:00 2001 From: Gagan Trivedi Date: Mon, 20 Jul 2026 17:12:15 +0530 Subject: [PATCH 5/7] refactor(experimentation): make ingestion provisioning create-once Rename the _ensure_* helpers to _create_*, drop the ClientError swallow so existing-resource errors propagate to the caller, and clarify the docstring. The caller is responsible for tracking whether provisioning has run. --- .../ingestion_infra_service.py | 69 ++++++++----------- .../test_ingestion_infra_service.py | 27 +------- .../observability/_events-catalogue.md | 4 +- 3 files changed, 35 insertions(+), 65 deletions(-) diff --git a/api/experimentation/ingestion_infra_service.py b/api/experimentation/ingestion_infra_service.py index 7c931a677533..08b5ce95b35a 100644 --- a/api/experimentation/ingestion_infra_service.py +++ b/api/experimentation/ingestion_infra_service.py @@ -5,7 +5,6 @@ import boto3 import structlog -from botocore.exceptions import ClientError from django.conf import settings from django.core.exceptions import ImproperlyConfigured @@ -83,22 +82,17 @@ def get_stream_name(organisation_id: int) -> str: return f"{STREAM_NAME_PREFIX}{organisation_id}" -def _ensure_events_bucket(bucket_name: str, *, organisation_id: int) -> None: +def _create_events_bucket(bucket_name: str, *, organisation_id: int) -> None: s3 = _get_s3_client() - try: - s3.create_bucket( - Bucket=bucket_name, - CreateBucketConfiguration={"LocationConstraint": AWS_REGION}, - ) - except ClientError as exc: - if exc.response["Error"]["Code"] != "BucketAlreadyOwnedByYou": - raise - else: - logger.info( - "ingestion_infra.bucket_created", - organisation__id=organisation_id, - bucket__name=bucket_name, - ) + s3.create_bucket( + Bucket=bucket_name, + CreateBucketConfiguration={"LocationConstraint": AWS_REGION}, + ) + logger.info( + "ingestion_infra.bucket_created", + organisation__id=organisation_id, + bucket__name=bucket_name, + ) s3.put_public_access_block( Bucket=bucket_name, PublicAccessBlockConfiguration={ @@ -165,44 +159,41 @@ def _expected_destination_configuration(bucket_name: str) -> dict[str, Any]: } -def _ensure_delivery_stream( +def _create_delivery_stream( stream_name: str, *, bucket_name: str, organisation_id: int, ) -> None: - try: - _get_firehose_client().create_delivery_stream( - DeliveryStreamName=stream_name, - DeliveryStreamType="DirectPut", - ExtendedS3DestinationConfiguration=_expected_destination_configuration( - bucket_name - ), - ) - except ClientError as exc: - if exc.response["Error"]["Code"] != "ResourceInUseException": - raise - else: - logger.info( - "ingestion_infra.stream_created", - organisation__id=organisation_id, - stream__name=stream_name, - bucket__name=bucket_name, - ) + _get_firehose_client().create_delivery_stream( + DeliveryStreamName=stream_name, + DeliveryStreamType="DirectPut", + ExtendedS3DestinationConfiguration=_expected_destination_configuration( + bucket_name + ), + ) + logger.info( + "ingestion_infra.stream_created", + organisation__id=organisation_id, + stream__name=stream_name, + bucket__name=bucket_name, + ) def provision_ingestion_infrastructure( organisation_id: int, ) -> IngestionInfrastructure: - """Idempotently create the organisation's events S3 bucket and Firehose - delivery stream; existing resources are left untouched.""" + """Create the organisation's events S3 bucket and Firehose delivery + stream. Both are created unconditionally; if a resource already exists + the underlying client error propagates to the caller, which is + responsible for tracking whether provisioning has already run.""" if not settings.INGESTION_FIREHOSE_DELIVERY_ROLE_ARN: raise ImproperlyConfigured("INGESTION_FIREHOSE_DELIVERY_ROLE_ARN is not set") bucket_name = get_bucket_name(organisation_id) stream_name = get_stream_name(organisation_id) - _ensure_events_bucket(bucket_name, organisation_id=organisation_id) - _ensure_delivery_stream( + _create_events_bucket(bucket_name, organisation_id=organisation_id) + _create_delivery_stream( stream_name, bucket_name=bucket_name, organisation_id=organisation_id, diff --git a/api/tests/unit/experimentation/test_ingestion_infra_service.py b/api/tests/unit/experimentation/test_ingestion_infra_service.py index 8bd781c09ded..76fc2207b682 100644 --- a/api/tests/unit/experimentation/test_ingestion_infra_service.py +++ b/api/tests/unit/experimentation/test_ingestion_infra_service.py @@ -178,27 +178,6 @@ def test_provision_ingestion_infrastructure__bucket_creation__sends_account_regi assert captured_headers[0]["x-amz-bucket-namespace"] == "account-regional" -def test_provision_ingestion_infrastructure__already_provisioned__is_idempotent( - ingestion_infra_settings: SettingsWrapper, - aws_backends: None, - log: StructuredLogCapture, -) -> None: - # Given - first = ingestion_infra_service.provision_ingestion_infrastructure( - organisation_id=42, - ) - events_after_first_run = list(log.events) - - # When - second = ingestion_infra_service.provision_ingestion_infrastructure( - organisation_id=42, - ) - - # Then - assert second == first - assert log.events == events_after_first_run - - def test_provision_ingestion_infrastructure__bucket_creation_fails__propagates_client_error( ingestion_infra_settings: SettingsWrapper, mocker: MockerFixture, @@ -208,14 +187,14 @@ def test_provision_ingestion_infrastructure__bucket_creation_fails__propagates_c "experimentation.ingestion_infra_service._get_account_id", return_value="123456789012", ) - mock_s3 = mocker.Mock() - mock_s3.create_bucket.side_effect = ClientError( + mock_s3_client = mocker.Mock() + mock_s3_client.create_bucket.side_effect = ClientError( {"Error": {"Code": "AccessDenied", "Message": "nope"}}, "CreateBucket", ) mocker.patch( "experimentation.ingestion_infra_service._get_s3_client", - return_value=mock_s3, + return_value=mock_s3_client, ) # When / Then diff --git a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md index 86610f2332e5..e7087f2421bc 100644 --- a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md @@ -95,7 +95,7 @@ Attributes: ### `experimentation.ingestion_infra.bucket_created` Logged at `info` from: - - `api/experimentation/ingestion_infra_service.py:97` + - `api/experimentation/ingestion_infra_service.py:91` Attributes: - `bucket.name` @@ -104,7 +104,7 @@ Attributes: ### `experimentation.ingestion_infra.stream_created` Logged at `info` from: - - `api/experimentation/ingestion_infra_service.py:186` + - `api/experimentation/ingestion_infra_service.py:175` Attributes: - `bucket.name` From 51d767fa5bc6518c51b4f944ae6498eeb2064112 Mon Sep 17 00:00:00 2001 From: Gagan Trivedi Date: Tue, 21 Jul 2026 10:58:57 +0530 Subject: [PATCH 6/7] feat(experimentation): tag events bucket and stream with organisation id Add an organisation_id cost-allocation tag to the per-organisation S3 bucket and Firehose delivery stream so AWS billing can be attributed per organisation. --- api/experimentation/ingestion_infra_service.py | 11 +++++++++++ .../experimentation/test_ingestion_infra_service.py | 6 ++++++ .../observability/_events-catalogue.md | 4 ++-- 3 files changed, 19 insertions(+), 2 deletions(-) diff --git a/api/experimentation/ingestion_infra_service.py b/api/experimentation/ingestion_infra_service.py index 08b5ce95b35a..f524ad64c2bb 100644 --- a/api/experimentation/ingestion_infra_service.py +++ b/api/experimentation/ingestion_infra_service.py @@ -43,6 +43,10 @@ DYNAMIC_PARTITIONING_RETRY_SECONDS = 300 OBJECT_EXPIRATION_DAYS = 30 +# Cost-allocation tag applied to per-organisation resources so AWS billing can +# be attributed per organisation. +ORGANISATION_TAG_KEY = "organisation_id" + def _add_account_regional_namespace_header( params: dict[str, Any], @@ -115,6 +119,12 @@ def _create_events_bucket(bucket_name: str, *, organisation_id: int) -> None: ] }, ) + s3.put_bucket_tagging( + Bucket=bucket_name, + Tagging={ + "TagSet": [{"Key": ORGANISATION_TAG_KEY, "Value": str(organisation_id)}] + }, + ) def _expected_destination_configuration(bucket_name: str) -> dict[str, Any]: @@ -171,6 +181,7 @@ def _create_delivery_stream( ExtendedS3DestinationConfiguration=_expected_destination_configuration( bucket_name ), + Tags=[{"Key": ORGANISATION_TAG_KEY, "Value": str(organisation_id)}], ) logger.info( "ingestion_infra.stream_created", diff --git a/api/tests/unit/experimentation/test_ingestion_infra_service.py b/api/tests/unit/experimentation/test_ingestion_infra_service.py index 76fc2207b682..5ecc54fffafb 100644 --- a/api/tests/unit/experimentation/test_ingestion_infra_service.py +++ b/api/tests/unit/experimentation/test_ingestion_infra_service.py @@ -87,6 +87,8 @@ def test_provision_ingestion_infrastructure__fresh_account__creates_bucket_and_s "Expiration": {"Days": 30}, } ] + tagging = s3.get_bucket_tagging(Bucket=result.bucket_name) + assert tagging["TagSet"] == [{"Key": "organisation_id", "Value": "42"}] firehose = boto3.client("firehose", region_name="eu-west-2") stream = firehose.describe_delivery_stream(DeliveryStreamName=result.stream_name)[ @@ -142,6 +144,10 @@ def test_provision_ingestion_infrastructure__fresh_account__creates_bucket_and_s }, ], } + stream_tags = firehose.list_tags_for_delivery_stream( + DeliveryStreamName=result.stream_name + ) + assert stream_tags["Tags"] == [{"Key": "organisation_id", "Value": "42"}] assert log.events == [ { diff --git a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md index e7087f2421bc..e64964d8dd6f 100644 --- a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md @@ -95,7 +95,7 @@ Attributes: ### `experimentation.ingestion_infra.bucket_created` Logged at `info` from: - - `api/experimentation/ingestion_infra_service.py:91` + - `api/experimentation/ingestion_infra_service.py:95` Attributes: - `bucket.name` @@ -104,7 +104,7 @@ Attributes: ### `experimentation.ingestion_infra.stream_created` Logged at `info` from: - - `api/experimentation/ingestion_infra_service.py:175` + - `api/experimentation/ingestion_infra_service.py:186` Attributes: - `bucket.name` From e2124c318f0ed650f89f5c9c02a480ed902f2bcc Mon Sep 17 00:00:00 2001 From: Gagan Trivedi Date: Tue, 21 Jul 2026 11:12:48 +0530 Subject: [PATCH 7/7] refactor(experimentation): drop redundant provisioning docstring --- api/experimentation/ingestion_infra_service.py | 4 ---- 1 file changed, 4 deletions(-) diff --git a/api/experimentation/ingestion_infra_service.py b/api/experimentation/ingestion_infra_service.py index f524ad64c2bb..f5c556fc2077 100644 --- a/api/experimentation/ingestion_infra_service.py +++ b/api/experimentation/ingestion_infra_service.py @@ -194,10 +194,6 @@ def _create_delivery_stream( def provision_ingestion_infrastructure( organisation_id: int, ) -> IngestionInfrastructure: - """Create the organisation's events S3 bucket and Firehose delivery - stream. Both are created unconditionally; if a resource already exists - the underlying client error propagates to the caller, which is - responsible for tracking whether provisioning has already run.""" if not settings.INGESTION_FIREHOSE_DELIVERY_ROLE_ARN: raise ImproperlyConfigured("INGESTION_FIREHOSE_DELIVERY_ROLE_ARN is not set") bucket_name = get_bucket_name(organisation_id)