Issue 664 cloud native aws (#666)

* feat: BIP-0042 Cloud-Native Architecture for Apache Burr on AWS (#664)

- Event-driven SQS telemetry: S3 notifications to SQS, near-instant updates
- Buffered S3 persistence: SpooledTemporaryFile fixes seek errors on large files
- Native BedrockAction and BedrockStreamingAction for Bedrock integration
- Terraform module: S3, SQS, IAM with dev/prod tfvars and tutorial

* feat: BIP-0042 Cloud-Native Architecture for Apache Burr on AWS (#664)

* feat: BIP-0042 Cloud-Native Architecture for Apache Burr on AWS (#664)

* BIP-0042 Cloud-Native Architecture for Apache Burr on AWS (#664) fixed build error

* chore: move Bedrock integration to separate PR and review comments fixed

* feat: add Bedrock integration (BIP-0042) as separate PR

* Revert "feat: add Bedrock integration (BIP-0042) as separate PR"

This reverts commit 61673cd.

* fix: add tracking-server-s3 to CI deps, remove Bedrock (in separate PR), fix typo

* fix: lint S3/run.py, SQS multi-record parse, BIP-0042 tests and CI job

---------

Co-authored-by: vaquarkhan <vaquarkhan@users.noreply.github.com>

Author: vaquarkhan

.github/workflows/python-package.yml

@@ -75,6 +75,29 @@ jobs:
         run: |
           python -m pytest tests --ignore=tests/integrations/persisters
 
+  test-tracking-server-s3:
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ['3.9', '3.10', '3.11', '3.12']
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v4
+        with:
+          python-version: ${{ matrix.python-version }}
+
+      - name: Install dependencies
+        run: |
+          python -m pip install -e ".[tests,tracking-client,tracking-server-s3]"
+
+      - name: Run S3 tracking server tests
+        run: |
+          python -m pytest tests/tracking/test_bip0042_s3_buffering.py -v
+
   validate-examples:
     runs-on: ubuntu-latest
     steps:

.gitignore

@@ -193,3 +193,11 @@ burr/tracking/server/build
 examples/*/statemachine
 examples/*/*/statemachine
 .vscode
+
+# Terraform (see also examples/deployment/aws/terraform/.gitignore)
+**/.terraform.lock.hcl
+examples/deployment/aws/terraform/.terraform/
+examples/deployment/aws/terraform/*.tfstate
+examples/deployment/aws/terraform/*.tfstate.*
+examples/deployment/aws/terraform/.terraform.tfstate.lock.info
+examples/deployment/aws/terraform/*.tfplan

burr/tracking/server/backend.py

@@ -162,6 +162,31 @@ def snapshot_interval_milliseconds(self) -> Optional[int]:
         pass
 
 
+class EventDrivenBackendMixin(abc.ABC):
+    """Mixin for backends that support event-driven updates.
+
+    Enables backends to receive real-time notifications instead of polling
+    for new files.
+    """
+
+    @abc.abstractmethod
+    async def start_event_consumer(self):
+        """Start the event consumer for event-driven tracking.
+
+        This method should run indefinitely, processing event notifications
+        from the configured message queue.
+        """
+        pass
+
+    @abc.abstractmethod
+    def is_event_driven(self) -> bool:
+        """Check if this backend is configured for event-driven updates.
+
+        :return: True if event-driven mode is enabled and configured, False otherwise
+        """
+        pass
+
+
 class BackendBase(abc.ABC):
     async def lifespan(self, app: FastAPI):
         """Quick tool to allow plugin to the app's lifecycle.

burr/tracking/server/run.py

@@ -15,6 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 
+import asyncio
 import importlib
 import logging
 import os
@@ -29,6 +30,7 @@
 from burr.tracking.server.backend import (
     AnnotationsBackendMixin,
     BackendBase,
+    EventDrivenBackendMixin,
     IndexingBackendMixin,
     SnapshottingBackendMixin,
 )
@@ -134,9 +136,20 @@ async def lifespan(app: FastAPI):
     await backend.lifespan(app).__anext__()
     await sync_index()  # this will trigger the repeat every N seconds
     await save_snapshot()  # this will trigger the repeat every N seconds
+    # Start event consumer for event-driven tracking when configured
+    event_consumer_task = None
+    if isinstance(backend, EventDrivenBackendMixin) and backend.is_event_driven():
+        event_consumer_task = asyncio.create_task(backend.start_event_consumer())
     global initialized
     initialized = True
     yield
+    # Graceful shutdown: cancel event consumer task
+    if event_consumer_task is not None:
+        event_consumer_task.cancel()
+        try:
+            await event_consumer_task
+        except asyncio.CancelledError:
+            pass
     await backend.lifespan(app).__anext__()
 
 
@@ -159,17 +172,16 @@ def _get_app_spec() -> BackendSpec:
 logger = logging.getLogger(__name__)
 
 if app_spec.indexing:
-    update_interval = backend.update_interval_milliseconds() / 1000 if app_spec.indexing else None
-    sync_index = repeat_every(
-        seconds=backend.update_interval_milliseconds() / 1000,
-        wait_first=True,
-        logger=logger,
-    )(sync_index)
+    # Only use polling when not in event-driven mode
+    _event_driven = isinstance(backend, EventDrivenBackendMixin) and backend.is_event_driven()
+    if not _event_driven:
+        sync_index = repeat_every(
+            seconds=backend.update_interval_milliseconds() / 1000,
+            wait_first=True,
+            logger=logger,
+        )(sync_index)
 
 if app_spec.snapshotting:
-    snapshot_interval = (
-        backend.snapshot_interval_milliseconds() / 1000 if app_spec.snapshotting else None
-    )
     save_snapshot = repeat_every(
         seconds=backend.snapshot_interval_milliseconds() / 1000,
         wait_first=True,

burr/tracking/server/s3/README.md

@@ -59,8 +59,9 @@ This will immediately start indexing your s3 bucket (or pick up from the last sn
 
 To track your data, you use the S3TrackingClient. You pass the tracker to the `ApplicationBuilder`:
 
-
 ```python
+from burr.tracking.s3client import S3TrackingClient
+
 app = (
     ApplicationBuilder()
     .with_graph(graph)