diff --git a/api/app/settings/common.py b/api/app/settings/common.py index 5adde3332224..859ea6c32dae 100644 --- a/api/app/settings/common.py +++ b/api/app/settings/common.py @@ -806,6 +806,11 @@ # Redis Cluster URL used to communicate with the event ingestion server. INGESTION_REDIS_URL = env.str("INGESTION_REDIS_URL", 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..08b5ce95b35a --- /dev/null +++ b/api/experimentation/ingestion_infra_service.py @@ -0,0 +1,201 @@ +from __future__ import annotations + +import typing +from functools import lru_cache + +import boto3 +import structlog +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-" +# 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. +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 +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": + # 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) +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: + return f"{BUCKET_NAME_PREFIX}{organisation_id}-{_get_account_id()}-{AWS_REGION}-an" + + +def get_stream_name(organisation_id: int) -> str: + return f"{STREAM_NAME_PREFIX}{organisation_id}" + + +def _create_events_bucket(bucket_name: str, *, organisation_id: int) -> None: + s3 = _get_s3_client() + 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={ + "BlockPublicAcls": True, + "IgnorePublicAcls": True, + "BlockPublicPolicy": True, + "RestrictPublicBuckets": True, + }, + ) + s3.put_bucket_lifecycle_configuration( + Bucket=bucket_name, + LifecycleConfiguration={ + "Rules": [ + { + "ID": "expire-objects", + "Filter": {"Prefix": ""}, + "Status": "Enabled", + "Expiration": {"Days": 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 _create_delivery_stream( + stream_name: str, + *, + bucket_name: str, + organisation_id: int, +) -> None: + _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: + """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) + + _create_events_bucket(bucket_name, organisation_id=organisation_id) + _create_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..76fc2207b682 --- /dev/null +++ b/api/tests/unit/experimentation/test_ingestion_infra_service.py @@ -0,0 +1,230 @@ +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 ( # 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 + +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_FIREHOSE_DELIVERY_ROLE_ARN = DELIVERY_ROLE_ARN + return settings + + +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() + + +@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( + 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: + # Given + organisation_id = 42 + + # When + result = ingestion_infra_service.provision_ingestion_infrastructure( + organisation_id=organisation_id, + ) + + # Then + assert result == IngestionInfrastructure( + bucket_name="flagsmith-events-lake-org-42-123456789012-eu-west-2-an", + 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-objects", + "Filter": {"Prefix": ""}, + "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-org-42-123456789012-eu-west-2-an" + ) + 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-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-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__bucket_creation_fails__propagates_client_error( + ingestion_infra_settings: SettingsWrapper, + mocker: MockerFixture, +) -> None: + # Given + mocker.patch( + "experimentation.ingestion_infra_service._get_account_id", + return_value="123456789012", + ) + 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_client, + ) + + # 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_account_id", + return_value="123456789012", + ) + 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..e7087f2421bc 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:91` + +Attributes: + - `bucket.name` + - `organisation.id` + +### `experimentation.ingestion_infra.stream_created` + +Logged at `info` from: + - `api/experimentation/ingestion_infra_service.py:175` + +Attributes: + - `bucket.name` + - `organisation.id` + - `stream.name` + ### `experimentation.results.compute_failed` Logged at `error` from: 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": [