feat(experimentation): add per-organisation ingestion infrastructure service#8035
feat(experimentation): add per-organisation ingestion infrastructure service#8035gagantrivedi wants to merge 5 commits into
Conversation
|
The latest updates on your projects. Learn more about Vercel for GitHub.
2 Skipped Deployments
|
📝 WalkthroughWalkthroughAdds an environment setting and immutable return type for ingestion infrastructure. Implements organisation-scoped S3 bucket and AWS Firehose delivery stream provisioning, including lifecycle, access, partitioning, compression, and error-output configuration. Updates production and staging task definitions with the delivery-role ARN, documents creation events, and adds mocked AWS tests covering validation, configuration, headers, logging, and error propagation. Estimated code review effort: 4 (Complex) | ~45 minutes Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
github-actions
Bot
added
api
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #8035 +/- ##
========================================
Coverage 98.64% 98.65%
========================================
Files 1506 1508 +2
Lines 59573 59706 +133
========================================
+ Hits 58767 58900 +133
Misses 806 806 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
e75bdc5 to
ab5cb97
Compare
github-actions
Bot
added
feature
github-actions
Bot
added
feature
Docker builds report
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 24077621-5c07-4d2b-9c18-31d4b76ae2ac
📒 Files selected for processing (9)
api/app/settings/common.pyapi/experimentation/dataclasses.pyapi/experimentation/ingestion_infra_service.pyapi/tests/unit/experimentation/test_ingestion_infra_service.pydocs/docs/deployment-self-hosting/observability/_events-catalogue.mdinfrastructure/aws/production/ecs-task-definition-admin-api.jsoninfrastructure/aws/production/ecs-task-definition-task-processor.jsoninfrastructure/aws/staging/ecs-task-definition-admin-api.jsoninfrastructure/aws/staging/ecs-task-definition-task-processor.json
Playwright Test Results (oss - depot-ubuntu-latest-16)
Details
Failed testsfirefox › tests/onboarding-tests.pw.ts › Onboarding › New user connects via the single-page onboarding flow @oss
Details
Playwright Test Results (private-cloud - depot-ubuntu-latest-16)
Details
|
…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.
Replace the errors-only lifecycle rule with one that expires every object in the bucket after 30 days.
…finitions Wire the per-stack Firehose delivery role ARN into the admin-api and task-processor task definitions for staging and production.
Visual Regression19 screenshots compared. See report for details. |
7fd6fcb to
d9291a6
Compare
github-actions
Bot
added
docs
✅ private-cloud · depot-ubuntu-latest-arm-16 — run #18520 (attempt 1)Playwright Test Results (private-cloud - depot-ubuntu-latest-arm-16)
Details
🗂️ Previous results✅ private-cloud · depot-ubuntu-latest-16 — run #18520 (attempt 1)Playwright Test Results (private-cloud - depot-ubuntu-latest-16)
Details
✅ oss · depot-ubuntu-latest-arm-16 — run #18520 (attempt 1)Playwright Test Results (oss - depot-ubuntu-latest-arm-16)
Details
✅ oss · depot-ubuntu-latest-16 — run #18520 (attempt 1)Playwright Test Results (oss - depot-ubuntu-latest-16)
Details
✅ private-cloud · depot-ubuntu-latest-16 — run #18514 (attempt 1)Playwright Test Results (private-cloud - depot-ubuntu-latest-16)
Details
✅ private-cloud · depot-ubuntu-latest-arm-16 — run #18514 (attempt 1)Playwright Test Results (private-cloud - depot-ubuntu-latest-arm-16)
Details
✅ oss · depot-ubuntu-latest-arm-16 — run #18514 (attempt 1)Playwright Test Results (oss - depot-ubuntu-latest-arm-16)
Details
✅ oss · depot-ubuntu-latest-16 — run #18514 (attempt 1)Playwright Test Results (oss - depot-ubuntu-latest-16)
Details
|
There was a problem hiding this comment.
Actionable comments posted: 2
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: b196d699-26ee-41ce-8f5a-ea241c957100
📒 Files selected for processing (9)
api/app/settings/common.pyapi/experimentation/dataclasses.pyapi/experimentation/ingestion_infra_service.pyapi/tests/unit/experimentation/test_ingestion_infra_service.pydocs/docs/deployment-self-hosting/observability/_events-catalogue.mdinfrastructure/aws/production/ecs-task-definition-admin-api.jsoninfrastructure/aws/production/ecs-task-definition-task-processor.jsoninfrastructure/aws/staging/ecs-task-definition-admin-api.jsoninfrastructure/aws/staging/ecs-task-definition-task-processor.json
| 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" | ||
|
|
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
🧩 Analysis chain
🌐 Web query:
What is the minimum boto3 version that supports the BucketNamespace parameter in S3 create_bucket?
💡 Result:
The BucketNamespace parameter in the S3 create_bucket API requires botocore version 1.42.0 or later to be recognized and supported [1][2][3]. This parameter was introduced to support the account-regional namespaces feature for S3 general purpose buckets, which was officially announced by AWS on March 12, 2026 [4][1][2]. As boto3 relies on botocore for its underlying API definitions and client operations, users must ensure they have a compatible version of botocore installed to utilize this feature [1][2].
Citations:
- 1: Add bucket_namespace support to S3CreateBucketOperator apache/airflow#63622
- 2: S3: Support BucketNamespace parameter for account-regional namespaces getmoto/moto#9870
- 3: https://www.mail-archive.com/commits%40airflow.apache.org/msg503187.html
- 4: https://aws.amazon.com/blogs/aws/introducing-account-regional-namespaces-for-amazon-s3-general-purpose-buckets/
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== api/experimentation/ingestion_infra_service.py ==\n'
nl -ba api/experimentation/ingestion_infra_service.py | sed -n '1,140p'
printf '\n== dependency pins mentioning boto3/botocore ==\n'
rg -n --hidden --glob '!**/.git/**' '\bboto3\b|\bbotocore\b' .Repository: Flagsmith/flagsmith
Length of output: 251
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== api/experimentation/ingestion_infra_service.py ==\n'
cat -n api/experimentation/ingestion_infra_service.py | sed -n '1,160p'
printf '\n== dependency pins mentioning boto3/botocore ==\n'
rg -n --hidden --glob '!**/.git/**' '\bboto3\b|\bbotocore\b' .Repository: Flagsmith/flagsmith
Length of output: 15528
Use the native BucketNamespace parameter once the SDK is upgraded
This shim is needed while pinned to boto3 1.35.95/botocore 1.35.95; when the pin moves to a version that supports BucketNamespace, pass it to create_bucket() directly and drop the raw x-amz-bucket-namespace header mutation.
| 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, | ||
| ) | ||
|
|
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
Firehose destination config (including the delivery role ARN) is never reconciled after first creation.
Unlike _ensure_events_bucket, which reapplies public-access-block and lifecycle settings on every call, _ensure_delivery_stream only sets ExtendedS3DestinationConfiguration at creation time; on ResourceInUseException it does nothing further. If INGESTION_FIREHOSE_DELIVERY_ROLE_ARN (or any other destination setting) is rotated later, already-provisioned organisations keep the stale config indefinitely — re-running provision_ingestion_infrastructure won't update it, and delivery will silently start failing if the old role is later revoked.
Consider calling describe_delivery_stream + update_destination when the current config differs from _expected_destination_configuration, mirroring the reconciliation already done for the bucket.
|
@themis-blindfold review |
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.
github-actions
Bot
added
feature
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
api/experimentation/ingestion_infra_service.py (1)
195-200: 🩺 Stability & Availability | 🔴 Critical | 🏗️ Heavy liftHandle partial provisioning failures to avoid permanently breaking retries.
Because
_create_events_bucketand_create_delivery_streamunconditionally propagate all AWS client exceptions, a transient failure halfway through provisioning (for instance, if_create_delivery_streamfails after the S3 bucket is created) will cause subsequent retries to fail immediately. On the next attempt,s3.create_bucketwill raise aBucketAlreadyOwnedByYou(or similar) exception, which propagates and halts execution before the stream can be created.To ensure the provisioning workflow can safely recover from partial failures, either catch and handle "already exists" exceptions in the helper functions to restore idempotency, or explicitly check for resource existence here before attempting creation.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: f42141b3-6a0e-4f6f-832c-eb7fc8397be9
📒 Files selected for processing (3)
api/experimentation/ingestion_infra_service.pyapi/tests/unit/experimentation/test_ingestion_infra_service.pydocs/docs/deployment-self-hosting/observability/_events-catalogue.md
⚖️ Themis judgement: ✅ Ship itWell-structured service layer for provisioning per-organisation S3 event buckets and Firehose delivery streams. The create-then-reconcile pattern for the bucket (public access block and lifecycle reapplied on every call) is a good security habit, and the deliberate create-only semantics for the stream are documented in the docstring. Error handling correctly distinguishes "resource already exists" from real failures in both S3 and Firehose paths. Tests are thorough — happy path with full destination config verification, idempotency, header injection for account-regional namespace, settings guard, and error propagation for both AWS services.
Test coverage gets a 4 because the main integration test does heavy lifting (full destination config assertion, lifecycle, public access block, log events), but there are no direct unit tests for 📝 Walkthrough
🧪 How to verify
Automate: CI already covers steps 1–2 via the Product take: Infrastructure plumbing for per-organisation experiment event ingestion — no user-visible change yet, but it lays the foundation for the experimentation data pipeline. Solid improvement to the platform's experimentation capabilities once wired up. 🧭 Assumptions & unverified claims
A service that provisions its own buckets and streams — the infrastructure is bootstrapping itself, one org at a time. · reviewed at d9291a6 |
Zaimwa9
left a comment
Zaimwa9
left a comment
There was a problem hiding this comment.
2 comments more than requesting changes. For the create flow, a comment would be satisfying imo.
For the tags, it looks interesting to keep track of the costs precisely
| bucket_name: str, | ||
| organisation_id: int, | ||
| ) -> None: | ||
| _get_firehose_client().create_delivery_stream( |
There was a problem hiding this comment.
I saw that we can tag the streams https://docs.aws.amazon.com/firehose/latest/APIReference/API_CreateDeliveryStream.html
and it would directly reconcile the costs with the given organisation if my understanding is correct. Would it make sense to add them sooner than later ?
It needs to be activated in the console
| return f"{STREAM_NAME_PREFIX}{organisation_id}" | ||
|
|
||
|
|
||
| def _create_events_bucket(bucket_name: str, *, organisation_id: int) -> None: |
There was a problem hiding this comment.
Actually I don't think the comment raised by CodeRabbit has been addressed: #8035 (review)
If indeed create_bucket is create only and not upsert, we could at least skip a BucketAlreadyOwnedByYou error and resume the put operations if there had been failure during the previous creation cycle.
But I agree that we'd also need a retry mechanism somewhere and this can be a follow up. Can we in this case describe the accepted risk as a comment to not forget ?
2 participants
Thanks for submitting a PR! Please check the boxes below:
docs/if required so people know about the feature.Changes
Adds a service layer for provisioning per-organisation experiment events ingestion infrastructure, to be consumed by the events ingestion server and hooked up to an API later.
experimentation/ingestion_infra_service.py:provision_ingestion_infrastructure(organisation_id)idempotently creates an S3 events bucket and a DirectPut Firehose delivery stream per organisation.environment_key, so objects land under per-environment S3 namespaces (events/env_key=.../year=.../...) and are batched per environment.INGESTION_EVENTS_BUCKET_PREFIX,INGESTION_FIREHOSE_DELIVERY_ROLE_ARN(a single shared Firehose delivery role, provisioned out-of-band, scoped by bucket naming convention).Note: with a shared delivery role, per-stream/bucket isolation is enforced by the control plane (only this service sets stream destinations), not IAM; per-organisation roles are a possible v2 hardening.
How did you test this code?
Unit tests with moto covering fresh provisioning (full destination config), idempotent re-runs, missing-settings errors, and AWS client error propagation.
make lint,make typecheck, and the full experimentation unit suite pass locally.