diff --git a/api/experimentation/ingestion_infra_service.py b/api/experimentation/ingestion_infra_service.py index f5c556fc2077..889d94fdafc4 100644 --- a/api/experimentation/ingestion_infra_service.py +++ b/api/experimentation/ingestion_infra_service.py @@ -206,3 +206,30 @@ def provision_ingestion_infrastructure( organisation_id=organisation_id, ) return IngestionInfrastructure(bucket_name=bucket_name, stream_name=stream_name) + + +def _delete_events_bucket(bucket_name: str) -> None: + s3 = _get_s3_client() + paginator = s3.get_paginator("list_objects_v2") + for page in paginator.paginate(Bucket=bucket_name): + objects = [{"Key": obj["Key"]} for obj in page.get("Contents", [])] + if objects: + s3.delete_objects(Bucket=bucket_name, Delete={"Objects": objects}) + s3.delete_bucket(Bucket=bucket_name) + + +def deprovision_ingestion_infrastructure(organisation_id: int) -> None: + bucket_name = get_bucket_name(organisation_id) + stream_name = get_stream_name(organisation_id) + + _get_firehose_client().delete_delivery_stream( + DeliveryStreamName=stream_name, + AllowForceDelete=True, + ) + _delete_events_bucket(bucket_name) + logger.info( + "ingestion_infra.deprovisioned", + organisation__id=organisation_id, + bucket__name=bucket_name, + stream__name=stream_name, + ) diff --git a/api/experimentation/migrations/0011_organisation_ingestion_infrastructure.py b/api/experimentation/migrations/0011_organisation_ingestion_infrastructure.py new file mode 100644 index 000000000000..ac5faf942e2b --- /dev/null +++ b/api/experimentation/migrations/0011_organisation_ingestion_infrastructure.py @@ -0,0 +1,59 @@ +# Generated by Django 5.2.16 on 2026-07-21 07:36 + +import django.db.models.deletion +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ("experimentation", "0010_warehouse_connection_credentials_and_status_detail"), + ("organisations", "0058_update_audit_and_history_limits_in_sub_cache"), + ] + + operations = [ + migrations.CreateModel( + name="OrganisationIngestionInfrastructure", + fields=[ + ( + "id", + models.AutoField( + auto_created=True, + primary_key=True, + serialize=False, + verbose_name="ID", + ), + ), + ( + "status", + models.CharField( + choices=[ + ("pending", "Pending"), + ("created", "Created"), + ("errored", "Errored"), + ], + default="pending", + max_length=50, + ), + ), + ( + "bucket_name", + models.CharField(blank=True, max_length=255, null=True), + ), + ( + "stream_name", + models.CharField(blank=True, max_length=255, null=True), + ), + ("created_at", models.DateTimeField(auto_now_add=True)), + ("updated_at", models.DateTimeField(auto_now=True)), + ( + "organisation", + models.OneToOneField( + on_delete=django.db.models.deletion.DO_NOTHING, + related_name="ingestion_infrastructure", + to="organisations.organisation", + ), + ), + ], + ), + ] diff --git a/api/experimentation/models.py b/api/experimentation/models.py index 75501d1d5048..fb9cb9ed3125 100644 --- a/api/experimentation/models.py +++ b/api/experimentation/models.py @@ -78,9 +78,18 @@ class Meta: @hook(AFTER_CREATE) # type: ignore[misc] def sync_to_ingestion_on_create(self) -> None: - from experimentation.tasks import write_environment_ingestion_keys + from experimentation.tasks import ( + provision_external_warehouse_ingestion_infrastructure, + write_environment_ingestion_keys, + ) + + if self.warehouse_type == WarehouseType.FLAGSMITH: + write_environment_ingestion_keys.delay( + kwargs={"environment_id": self.environment_id}, + ) + return - write_environment_ingestion_keys.delay( + provision_external_warehouse_ingestion_infrastructure.delay( kwargs={"environment_id": self.environment_id}, ) @@ -93,6 +102,29 @@ def sync_to_ingestion_on_delete(self) -> None: ) +class IngestionInfrastructureStatus(models.TextChoices): + PENDING = "pending", "Pending" + CREATED = "created", "Created" + ERRORED = "errored", "Errored" + + +class OrganisationIngestionInfrastructure(models.Model): + organisation = models.OneToOneField( + "organisations.Organisation", + on_delete=models.DO_NOTHING, + related_name="ingestion_infrastructure", + ) + status = models.CharField( + max_length=50, + choices=IngestionInfrastructureStatus.choices, + default=IngestionInfrastructureStatus.PENDING, + ) + bucket_name = models.CharField(max_length=255, null=True, blank=True) + stream_name = models.CharField(max_length=255, null=True, blank=True) + created_at = models.DateTimeField(auto_now_add=True) + updated_at = models.DateTimeField(auto_now=True) + + class ExperimentStatus(models.TextChoices): CREATED = "created", "Created" RUNNING = "running", "Running" diff --git a/api/experimentation/organisation_ingestion_service.py b/api/experimentation/organisation_ingestion_service.py new file mode 100644 index 000000000000..069eed855ff8 --- /dev/null +++ b/api/experimentation/organisation_ingestion_service.py @@ -0,0 +1,74 @@ +from __future__ import annotations + +import typing + +import structlog + +from experimentation.ingestion_infra_service import ( + deprovision_ingestion_infrastructure, + provision_ingestion_infrastructure, +) +from experimentation.models import ( + IngestionInfrastructureStatus, + OrganisationIngestionInfrastructure, +) + +if typing.TYPE_CHECKING: + from organisations.models import Organisation + +logger = structlog.get_logger("experimentation") + + +def enable_ingestion_for_organisation( + organisation: "Organisation", +) -> OrganisationIngestionInfrastructure: + infrastructure = OrganisationIngestionInfrastructure.objects.filter( + organisation=organisation + ).first() + if ( + infrastructure is not None + and infrastructure.status == IngestionInfrastructureStatus.CREATED + ): + return infrastructure + if infrastructure is None: + infrastructure = OrganisationIngestionInfrastructure(organisation=organisation) + + try: + result = provision_ingestion_infrastructure(organisation.id) + except Exception as exc: + infrastructure.status = IngestionInfrastructureStatus.ERRORED + infrastructure.save() + logger.error( + "ingestion_infra.provision_failed", + exc_info=exc, + organisation__id=organisation.id, + ) + raise + + infrastructure.bucket_name = result.bucket_name + infrastructure.stream_name = result.stream_name + infrastructure.status = IngestionInfrastructureStatus.CREATED + infrastructure.save() + logger.info( + "ingestion_infra.provisioned", + organisation__id=organisation.id, + bucket__name=result.bucket_name, + stream__name=result.stream_name, + ) + return infrastructure + + +def disable_ingestion_for_organisation(organisation_id: int) -> None: + infrastructure = OrganisationIngestionInfrastructure.objects.filter( + organisation_id=organisation_id + ).first() + if infrastructure is None: + return + + if infrastructure.status == IngestionInfrastructureStatus.CREATED: + deprovision_ingestion_infrastructure(organisation_id) + logger.info( + "ingestion_infra.torn_down", + organisation__id=organisation_id, + ) + infrastructure.delete() diff --git a/api/experimentation/tasks.py b/api/experimentation/tasks.py index 1446db7f31b2..c041e8a01994 100644 --- a/api/experimentation/tasks.py +++ b/api/experimentation/tasks.py @@ -5,6 +5,10 @@ from environments.models import Environment, EnvironmentAPIKey from experimentation import ingestion_sync_service from experimentation.models import Experiment, ExperimentExposures, ExperimentResults +from experimentation.organisation_ingestion_service import ( + disable_ingestion_for_organisation, + enable_ingestion_for_organisation, +) from experimentation.services import compute_exposures_summary, compute_results_summary logger = structlog.get_logger("experimentation") @@ -48,6 +52,25 @@ def remove_environment_ingestion_keys(environment_id: int) -> None: ingestion_sync_service.delete_ingestion_key(api_key.key) +@register_task_handler() +def provision_external_warehouse_ingestion_infrastructure(environment_id: int) -> None: + environment = ( + Environment.objects.select_related("project__organisation") + .filter(id=environment_id) + .first() + ) + if environment is None: + return + + enable_ingestion_for_organisation(environment.project.organisation) + write_environment_ingestion_keys(environment_id) + + +@register_task_handler() +def teardown_organisation_ingestion_infrastructure(organisation_id: int) -> None: + disable_ingestion_for_organisation(organisation_id) + + @register_task_handler() def write_environment_ingestion_key(environment_api_key_id: int) -> None: api_key = ( diff --git a/api/organisations/models.py b/api/organisations/models.py index e0965cd5e096..9cfebb0f0a1f 100644 --- a/api/organisations/models.py +++ b/api/organisations/models.py @@ -10,6 +10,7 @@ from django.utils import timezone from django_lifecycle import ( # type: ignore[import-untyped] AFTER_CREATE, + AFTER_DELETE, AFTER_SAVE, BEFORE_DELETE, LifecycleModelMixin, @@ -150,6 +151,19 @@ def cancel_subscription(self): # type: ignore[no-untyped-def] if self.has_paid_subscription(): self.subscription.prepare_for_cancel() + @hook(AFTER_DELETE) + def teardown_ingestion_infrastructure(self): # type: ignore[no-untyped-def] + if not hasattr(self, "ingestion_infrastructure"): + return + + from experimentation.tasks import ( + teardown_organisation_ingestion_infrastructure, + ) + + teardown_organisation_ingestion_infrastructure.delay( + kwargs={"organisation_id": self.id}, + ) + @hook(AFTER_CREATE) def create_subscription(self): # type: ignore[no-untyped-def] Subscription.objects.create(organisation=self) diff --git a/api/tests/unit/experimentation/conftest.py b/api/tests/unit/experimentation/conftest.py index 04d814c42ce6..0a4985efdc20 100644 --- a/api/tests/unit/experimentation/conftest.py +++ b/api/tests/unit/experimentation/conftest.py @@ -26,6 +26,17 @@ def mock_ingestion_redis_client(mocker: MockerFixture) -> None: mocker.patch("experimentation.ingestion_sync_service.RedisCluster.from_url") +@pytest.fixture(autouse=True) +def mock_provision_external_warehouse_ingestion_infrastructure( + mocker: MockerFixture, +) -> None: + # Creating an external warehouse connection enqueues provisioning, which + # runs synchronously under test; stub it so tests don't reach AWS. + mocker.patch( + "experimentation.tasks.provision_external_warehouse_ingestion_infrastructure", + ) + + @pytest.fixture() def warehouse_connection(environment: Environment) -> WarehouseConnection: connection: WarehouseConnection = WarehouseConnection.objects.create( diff --git a/api/tests/unit/experimentation/test_ingestion_infra_service.py b/api/tests/unit/experimentation/test_ingestion_infra_service.py index 5ecc54fffafb..d0aaf4e32990 100644 --- a/api/tests/unit/experimentation/test_ingestion_infra_service.py +++ b/api/tests/unit/experimentation/test_ingestion_infra_service.py @@ -234,3 +234,57 @@ def test_provision_ingestion_infrastructure__stream_creation_fails__propagates_c # When / Then with pytest.raises(ClientError, match="LimitExceededException"): ingestion_infra_service.provision_ingestion_infrastructure(organisation_id=42) + + +def test_deprovision_ingestion_infrastructure__existing_resources__deletes_bucket_and_stream( + ingestion_infra_settings: SettingsWrapper, + aws_backends: None, + log: StructuredLogCapture, +) -> None: + # Given + result = ingestion_infra_service.provision_ingestion_infrastructure( + organisation_id=42, + ) + + # When + ingestion_infra_service.deprovision_ingestion_infrastructure(organisation_id=42) + + # Then + s3 = boto3.client("s3", region_name="eu-west-2") + bucket_names = [bucket["Name"] for bucket in s3.list_buckets()["Buckets"]] + assert result.bucket_name not in bucket_names + + firehose = boto3.client("firehose", region_name="eu-west-2") + with pytest.raises(ClientError): + firehose.describe_delivery_stream(DeliveryStreamName=result.stream_name) + + assert { + "level": "info", + "event": "ingestion_infra.deprovisioned", + "organisation__id": 42, + "bucket__name": result.bucket_name, + "stream__name": result.stream_name, + } in log.events + + +def test_deprovision_ingestion_infrastructure__bucket_with_objects__empties_and_deletes_bucket( + ingestion_infra_settings: SettingsWrapper, + aws_backends: None, +) -> None: + # Given + result = ingestion_infra_service.provision_ingestion_infrastructure( + organisation_id=42, + ) + s3 = boto3.client("s3", region_name="eu-west-2") + s3.put_object( + Bucket=result.bucket_name, + Key="events/env_key=abc/data.json.gz", + Body=b"{}", + ) + + # When + ingestion_infra_service.deprovision_ingestion_infrastructure(organisation_id=42) + + # Then + bucket_names = [bucket["Name"] for bucket in s3.list_buckets()["Buckets"]] + assert result.bucket_name not in bucket_names diff --git a/api/tests/unit/experimentation/test_models.py b/api/tests/unit/experimentation/test_models.py index a058e872010e..006fe325733d 100644 --- a/api/tests/unit/experimentation/test_models.py +++ b/api/tests/unit/experimentation/test_models.py @@ -47,6 +47,32 @@ def test_warehouse_connection__after_create__enqueues_ingestion_write_task( ) +def test_warehouse_connection__after_create_external_type__enqueues_provision_task( + environment: Environment, + mocker: MockerFixture, +) -> None: + # Given + mock_provision = mocker.patch( + "experimentation.tasks.provision_external_warehouse_ingestion_infrastructure", + ) + mock_write_keys = mocker.patch( + "experimentation.tasks.write_environment_ingestion_keys", + ) + + # When + WarehouseConnection.objects.create( + environment=environment, + warehouse_type=WarehouseType.CLICKHOUSE, + name="external warehouse", + ) + + # Then the per-org infrastructure is provisioned, which chains the key sync + mock_provision.delay.assert_called_once_with( + kwargs={"environment_id": environment.id}, + ) + mock_write_keys.delay.assert_not_called() + + def test_warehouse_connection__after_delete__enqueues_ingestion_remove_task( warehouse_connection: WarehouseConnection, mocker: MockerFixture, diff --git a/api/tests/unit/experimentation/test_organisation_ingestion_service.py b/api/tests/unit/experimentation/test_organisation_ingestion_service.py new file mode 100644 index 000000000000..9b8517e7532e --- /dev/null +++ b/api/tests/unit/experimentation/test_organisation_ingestion_service.py @@ -0,0 +1,171 @@ +import pytest +from pytest_mock import MockerFixture +from pytest_structlog import StructuredLogCapture + +from experimentation import organisation_ingestion_service +from experimentation.dataclasses import IngestionInfrastructure +from experimentation.models import ( + IngestionInfrastructureStatus, + OrganisationIngestionInfrastructure, +) +from organisations.models import Organisation + +INFRASTRUCTURE = IngestionInfrastructure( + bucket_name="flagsmith-events-lake-org-1-123456789012-eu-west-2-an", + stream_name="events-ingestion-org-1", +) + + +def test_enable_ingestion_for_organisation__fresh__provisions_and_records( + organisation: Organisation, + mocker: MockerFixture, + log: StructuredLogCapture, +) -> None: + # Given + provision = mocker.patch( + "experimentation.organisation_ingestion_service" + ".provision_ingestion_infrastructure", + return_value=INFRASTRUCTURE, + ) + + # When + infrastructure = organisation_ingestion_service.enable_ingestion_for_organisation( + organisation + ) + + # Then + provision.assert_called_once_with(organisation.id) + assert infrastructure.status == IngestionInfrastructureStatus.CREATED + assert infrastructure.bucket_name == INFRASTRUCTURE.bucket_name + assert infrastructure.stream_name == INFRASTRUCTURE.stream_name + assert { + "level": "info", + "event": "ingestion_infra.provisioned", + "organisation__id": organisation.id, + "bucket__name": INFRASTRUCTURE.bucket_name, + "stream__name": INFRASTRUCTURE.stream_name, + } in log.events + + +def test_enable_ingestion_for_organisation__already_created__does_not_reprovision( + organisation: Organisation, + mocker: MockerFixture, +) -> None: + # Given + existing = OrganisationIngestionInfrastructure.objects.create( + organisation=organisation, + status=IngestionInfrastructureStatus.CREATED, + bucket_name=INFRASTRUCTURE.bucket_name, + stream_name=INFRASTRUCTURE.stream_name, + ) + provision = mocker.patch( + "experimentation.organisation_ingestion_service" + ".provision_ingestion_infrastructure", + ) + + # When + infrastructure = organisation_ingestion_service.enable_ingestion_for_organisation( + organisation + ) + + # Then + provision.assert_not_called() + assert infrastructure == existing + + +def test_enable_ingestion_for_organisation__provision_fails__marks_errored_and_reraises( + organisation: Organisation, + mocker: MockerFixture, + log: StructuredLogCapture, +) -> None: + # Given + mocker.patch( + "experimentation.organisation_ingestion_service" + ".provision_ingestion_infrastructure", + side_effect=RuntimeError("boom"), + ) + + # When / Then + with pytest.raises(RuntimeError, match="boom"): + organisation_ingestion_service.enable_ingestion_for_organisation(organisation) + + infrastructure = OrganisationIngestionInfrastructure.objects.get( + organisation=organisation + ) + assert infrastructure.status == IngestionInfrastructureStatus.ERRORED + assert any( + event["event"] == "ingestion_infra.provision_failed" for event in log.events + ) + + +def test_disable_ingestion_for_organisation__created__deprovisions_and_deletes_row( + organisation: Organisation, + mocker: MockerFixture, + log: StructuredLogCapture, +) -> None: + # Given + OrganisationIngestionInfrastructure.objects.create( + organisation=organisation, + status=IngestionInfrastructureStatus.CREATED, + bucket_name=INFRASTRUCTURE.bucket_name, + stream_name=INFRASTRUCTURE.stream_name, + ) + deprovision = mocker.patch( + "experimentation.organisation_ingestion_service" + ".deprovision_ingestion_infrastructure", + ) + + # When + organisation_ingestion_service.disable_ingestion_for_organisation(organisation.id) + + # Then + deprovision.assert_called_once_with(organisation.id) + assert not OrganisationIngestionInfrastructure.objects.filter( + organisation=organisation + ).exists() + assert { + "level": "info", + "event": "ingestion_infra.torn_down", + "organisation__id": organisation.id, + } in log.events + + +def test_disable_ingestion_for_organisation__no_infrastructure__does_nothing( + organisation: Organisation, + mocker: MockerFixture, +) -> None: + # Given + deprovision = mocker.patch( + "experimentation.organisation_ingestion_service" + ".deprovision_ingestion_infrastructure", + ) + + # When + organisation_ingestion_service.disable_ingestion_for_organisation(organisation.id) + + # Then + deprovision.assert_not_called() + + +def test_disable_ingestion_for_organisation__errored__deletes_row_without_deprovisioning( + organisation: Organisation, + mocker: MockerFixture, +) -> None: + # Given + OrganisationIngestionInfrastructure.objects.create( + organisation=organisation, + status=IngestionInfrastructureStatus.ERRORED, + ) + deprovision = mocker.patch( + "experimentation.organisation_ingestion_service" + ".deprovision_ingestion_infrastructure", + ) + + # When + organisation_ingestion_service.disable_ingestion_for_organisation(organisation.id) + + # Then + deprovision.assert_not_called() + assert not OrganisationIngestionInfrastructure.objects.filter( + organisation=organisation + ).exists() diff --git a/api/tests/unit/experimentation/test_tasks.py b/api/tests/unit/experimentation/test_tasks.py index e6e1d9ebb0c5..2f7a082cfc01 100644 --- a/api/tests/unit/experimentation/test_tasks.py +++ b/api/tests/unit/experimentation/test_tasks.py @@ -12,6 +12,7 @@ ExposuresSummary, ExposuresTimeseries, ExposuresTimeseriesPoint, + IngestionInfrastructure, MetricResult, ResultsSummary, ) @@ -20,16 +21,21 @@ ExperimentExposures, ExperimentResults, ExperimentStatus, + IngestionInfrastructureStatus, + OrganisationIngestionInfrastructure, ) from experimentation.stats import VariantStats from experimentation.tasks import ( compute_experiment_exposures, compute_experiment_results, + provision_external_warehouse_ingestion_infrastructure, remove_environment_ingestion_key, remove_environment_ingestion_keys, + teardown_organisation_ingestion_infrastructure, write_environment_ingestion_key, write_environment_ingestion_keys, ) +from organisations.models import Organisation def test_write_environment_ingestion_keys__valid_keys__whitelists_client_and_server( @@ -561,3 +567,78 @@ def test_compute_experiment_results__experiment_deleted_after_enqueue__skips( # Then the task exits without raising into the task processor mock_compute.assert_not_called() + + +def test_provision_external_warehouse_ingestion_infrastructure__valid_environment__provisions_and_syncs_keys( + environment: Environment, + mocker: MockerFixture, +) -> None: + # Given + provision = mocker.patch( + "experimentation.organisation_ingestion_service" + ".provision_ingestion_infrastructure", + return_value=IngestionInfrastructure( + bucket_name="flagsmith-events-lake-org-1-123456789012-eu-west-2-an", + stream_name="events-ingestion-org-1", + ), + ) + set_ingestion_key = mocker.patch( + "experimentation.tasks.ingestion_sync_service.set_ingestion_key", + ) + + # When + provision_external_warehouse_ingestion_infrastructure(environment_id=environment.id) + + # Then the org infrastructure is provisioned and the environment keys synced + provision.assert_called_once_with(environment.project.organisation_id) + assert OrganisationIngestionInfrastructure.objects.filter( + organisation=environment.project.organisation, + status=IngestionInfrastructureStatus.CREATED, + ).exists() + set_ingestion_key.assert_called_once_with( + environment.api_key, environment_key=environment.api_key + ) + + +def test_provision_external_warehouse_ingestion_infrastructure__missing_environment__does_nothing( + db: None, + mocker: MockerFixture, +) -> None: + # Given + provision = mocker.patch( + "experimentation.organisation_ingestion_service" + ".provision_ingestion_infrastructure", + ) + + # When + provision_external_warehouse_ingestion_infrastructure(environment_id=999999) + + # Then + provision.assert_not_called() + assert not OrganisationIngestionInfrastructure.objects.exists() + + +def test_teardown_organisation_ingestion_infrastructure__created_infrastructure__deprovisions( + organisation: Organisation, + mocker: MockerFixture, +) -> None: + # Given + OrganisationIngestionInfrastructure.objects.create( + organisation=organisation, + status=IngestionInfrastructureStatus.CREATED, + bucket_name="flagsmith-events-lake-org-1-123456789012-eu-west-2-an", + stream_name="events-ingestion-org-1", + ) + deprovision = mocker.patch( + "experimentation.organisation_ingestion_service" + ".deprovision_ingestion_infrastructure", + ) + + # When + teardown_organisation_ingestion_infrastructure(organisation_id=organisation.id) + + # Then + deprovision.assert_called_once_with(organisation.id) + assert not OrganisationIngestionInfrastructure.objects.filter( + organisation=organisation + ).exists() diff --git a/api/tests/unit/organisations/test_unit_organisations_models.py b/api/tests/unit/organisations/test_unit_organisations_models.py index 578dd67184f7..3c0a0708231f 100644 --- a/api/tests/unit/organisations/test_unit_organisations_models.py +++ b/api/tests/unit/organisations/test_unit_organisations_models.py @@ -9,6 +9,10 @@ from pytest_mock import MockerFixture from environments.models import Environment +from experimentation.models import ( + IngestionInfrastructureStatus, + OrganisationIngestionInfrastructure, +) from organisations.chargebee.metadata import ChargebeeObjMetadata from organisations.models import ( Organisation, @@ -893,3 +897,44 @@ def test_update_plan__valid_plan_id__updates_fields_from_chargebee( assert subscription.max_seats == 5 assert subscription.max_api_calls == 500000 assert subscription.cancellation_date is None + + +def test_organisation__after_delete_without_infrastructure__does_not_deprovision( + organisation: Organisation, + mocker: MockerFixture, +) -> None: + # Given + deprovision = mocker.patch( + "experimentation.organisation_ingestion_service" + ".deprovision_ingestion_infrastructure", + ) + + # When + organisation.delete() + + # Then + deprovision.assert_not_called() + + +def test_organisation__delete_with_created_infrastructure__deprovisions_aws_resources( + organisation: Organisation, + mocker: MockerFixture, +) -> None: + # Given + OrganisationIngestionInfrastructure.objects.create( + organisation=organisation, + status=IngestionInfrastructureStatus.CREATED, + bucket_name="flagsmith-events-lake-org-1-123456789012-eu-west-2-an", + stream_name="events-ingestion-org-1", + ) + deprovision = mocker.patch( + "experimentation.organisation_ingestion_service" + ".deprovision_ingestion_infrastructure", + ) + organisation_id = organisation.id + + # When + organisation.delete() + + # Then + deprovision.assert_called_once_with(organisation_id) diff --git a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md index 76161e5b84f8..195cd614cba6 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:100` + - `api/experimentation/tasks.py:123` Attributes: - `environment.id` @@ -109,6 +109,35 @@ Attributes: - `bucket.name` - `organisation.id` +### `experimentation.ingestion_infra.deprovisioned` + +Logged at `info` from: + - `api/experimentation/ingestion_infra_service.py:230` + +Attributes: + - `bucket.name` + - `organisation.id` + - `stream.name` + +### `experimentation.ingestion_infra.provision_failed` + +Logged at `error` from: + - `api/experimentation/organisation_ingestion_service.py:41` + +Attributes: + - `exc_info` + - `organisation.id` + +### `experimentation.ingestion_infra.provisioned` + +Logged at `info` from: + - `api/experimentation/organisation_ingestion_service.py:52` + +Attributes: + - `bucket.name` + - `organisation.id` + - `stream.name` + ### `experimentation.ingestion_infra.stream_created` Logged at `info` from: @@ -119,10 +148,18 @@ Attributes: - `organisation.id` - `stream.name` +### `experimentation.ingestion_infra.torn_down` + +Logged at `info` from: + - `api/experimentation/organisation_ingestion_service.py:70` + +Attributes: + - `organisation.id` + ### `experimentation.results.compute_failed` Logged at `error` from: - - `api/experimentation/tasks.py:136` + - `api/experimentation/tasks.py:159` Attributes: - `environment.id`