localstack · DrisDary · Jul 31, 2026 · Jul 29, 2026 · Jul 30, 2026
@@ -40,6 +40,7 @@ This repository contains comprehensive sample projects demonstrating how to deve
 | [ACI and Blob Storage](./samples/aci-blob-storage/python/README.md) | Azure Container Instances with ACR, Key Vault, and Blob Storage |
 | [Azure Service Bus with Spring Boot](./samples/servicebus/java/README.md) | Azure Service Bus used by a Spring Boot application |
 | [Event Hubs Fraud Detection Pipeline](./samples/eventhubs/python/README.md) | Real-time payment stream processing with Event Hubs (AMQP, Kafka and HTTPS ingestion, Capture, Schema Registry), an Event Hubs-triggered Function App, Key Vault, Storage and a Web App dashboard |
+| [Event Hubs Cold-Path Automation](./samples/eventhubs-eventgrid/python/README.md) | Event Hubs Capture raises `Microsoft.EventHub.CaptureFileCreated` to an Event Grid system topic, a subscription delivers it into a second event hub, and an Event Hubs-triggered Function App decodes each Avro archive and writes per-device summaries |
 
 ## Sample Structure
 

@@ -31,6 +31,7 @@ PURGE_DOCKER="${PURGE_DOCKER:-0}"
 SAMPLES=(
   "samples/servicebus/java|bash scripts/deploy.sh"
   "samples/eventhubs/python|bash scripts/deploy.sh|bash scripts/validate.sh && bash scripts/run-pipeline.sh"
+  "samples/eventhubs-eventgrid/python|bash scripts/deploy.sh|bash scripts/validate.sh && bash scripts/run-pipeline.sh"
   "samples/function-app-front-door/python|bash scripts/deploy_all.sh --name-prefix testafd|"
   "samples/function-app-managed-identity/python|bash scripts/user-managed-identity.sh|bash scripts/validate.sh && bash scripts/test.sh"
   "samples/function-app-service-bus/dotnet|bash scripts/deploy.sh|bash scripts/validate.sh && bash scripts/call-http-trigger.sh"
@@ -48,6 +49,7 @@ SAMPLES=(
 TERRAFORM_SAMPLES=(
   "samples/servicebus/java/terraform|bash deploy.sh"
   "samples/eventhubs/python/terraform|bash deploy.sh|bash ../scripts/validate.sh"
+  "samples/eventhubs-eventgrid/python/terraform|bash deploy.sh|bash ../scripts/validate.sh"
   "samples/function-app-managed-identity/python/terraform|bash deploy.sh"
   "samples/function-app-service-bus/dotnet/terraform|bash deploy.sh"
   "samples/function-app-storage-http/dotnet/terraform|bash deploy.sh"
@@ -63,6 +65,7 @@ TERRAFORM_SAMPLES=(
 BICEP_SAMPLES=(
   "samples/servicebus/java/bicep|bash deploy.sh"
   "samples/eventhubs/python/bicep|bash deploy.sh|bash ../scripts/validate.sh"
+  "samples/eventhubs-eventgrid/python/bicep|bash deploy.sh|bash ../scripts/validate.sh"
   #"samples/web-app-sql-database/python/bicep|bash deploy.sh"
   "samples/function-app-managed-identity/python/bicep|bash deploy.sh"
   "samples/function-app-service-bus/dotnet/bicep|bash deploy.sh"

@@ -0,0 +1,165 @@
+# Cold-path automation: Event Hubs Capture → Event Grid → Functions
+
+A cold-chain monitoring pipeline where **Azure tells you when a batch is ready**, instead of your
+code polling for it.
+
+Devices publish temperature readings into an event hub. Event Hubs Capture archives the stream to
+Blob Storage as Avro. The moment an archive lands, Event Hubs raises
+`Microsoft.EventHub.CaptureFileCreated` to Event Grid, an Event Grid subscription delivers that
+notification **into a second event hub**, and a Function App triggered on that hub downloads the
+archive, aggregates it per device, and writes summaries to a curated hub.
+
+```
+ devices ──▶ telemetry hub ──── Capture (60s window) ────▶ Avro archive in Blob Storage
+                  │                                                    │
+                  │                        Microsoft.EventHub.CaptureFileCreated
+                  │                                                    ▼
+                  │                                    Event Grid system topic
+                  │                                                    │
+                  │                          subscription, EventHub destination
+                  │                                                    ▼
+                  │                                      capture-notifications hub
+                  │                                                    │
+                  │                                    Event Hubs trigger
+                  │                                                    ▼
+                  └───────────── archive read back ────────────  Function App
+                                                                       │
+                                                     Event Hubs output binding
+                                                                       ▼
+                                                                  curated hub
+```
+
+## Why this shape
+
+The obvious way to process archives is a timer that lists a container and looks for new blobs.
+That is a poll: it is late by up to its interval, it re-lists everything each run, and it needs
+its own bookkeeping to avoid double-processing.
+
+Here the platform does the telling:
+
+| Concern | How the pipeline handles it |
+|---|---|
+| **When is a batch ready?** | Event Hubs says so — `CaptureFileCreated` carries the file URL, the partition, and the exact sequence range it contains |
+| **How does the notification reach the processor?** | Event Grid delivers it into an **event hub**, so there is no public webhook to expose and no HTTP endpoint to keep available |
+| **What if the processor is down?** | The notification sits in the hub for the retention window; the processor resumes from its checkpoint and catches up |
+| **Is the aggregation correct?** | Readings are keyed by device, so a device's readings share a partition — and therefore share an archive, in order |
+| **Can it be re-run?** | The archive is durable in Blob Storage; replaying is reading a file, not replaying a stream |
+
+That last column is what makes it a *cold* path: it runs on whole, closed windows rather than on
+whatever happened to be in the last batch.
+
+## What it demonstrates
+
+| Capability | Where |
+|---|---|
+| **Event Hubs Capture** to Avro in Blob Storage | `deploy.sh` step 4, `bicep/modules/event-hubs.bicep` |
+| **`Microsoft.EventHub.CaptureFileCreated`** system event | Raised by Event Hubs; consumed in `src/functions/function_app.py` |
+| **Event Grid system topic** over a namespace | `deploy.sh` step 7, `bicep/modules/event-grid.bicep` |
+| **Event Grid subscription with an EventHub destination** | Same — this is what makes the notification a stream |
+| **Event Hubs trigger** with `Cardinality.MANY` | `function_app.py` — batched delivery |
+| **Event Hubs output binding** | `function_app.py` — writes the curated summaries |
+| **Partition keys for correctness** | `telemetry_producer.py` — a device's readings stay together |
+| **Consumer groups** | The processor reads the notifications hub with its own group |
+| **Least-privilege SAS** | Send-only for producers, namespace-wide Listen-only for the demo scripts |
+| **Avro decoding from Blob Storage** | `function_app.py` `_read_archive` |
+
+## Prerequisites
+
+- The LocalStack Azure emulator running, and `lstk az start-interception` run.
+- [lstk](https://github.com/localstack/lstk) (`brew install localstack/tap/lstk` or
+  `npm install -g @localstack/lstk`), which routes the Azure CLI to the emulator.
+- Python 3.12+ with `src/producers/requirements.txt` installed.
+- `jq` and `zip`.
+
+> **The emulator must be able to resolve `cdn.functions.azure.com` from the function container.**
+> The Event Hubs trigger lives in the Functions extension bundle, which the host downloads on
+> every cold container. If the emulator's DNS server cannot forward external queries, the host
+> aborts and the trigger silently never fires. Starting the emulator with `DNS_ADDRESS=0`
+> sidesteps it.
+
+## Usage
+
+```bash
+cd samples/eventhubs-eventgrid/python
+
+bash scripts/deploy.sh          # nine steps: storage, hubs, Capture, Event Grid, Function App
+bash scripts/validate.sh        # numbered checks over every capability above
+bash scripts/run-pipeline.sh    # the end-to-end demo
+
+bash scripts/cleanup.sh         # when you are done
+```
+
+`deploy.sh` writes `scripts/.deployment-env` with the resource names and connection strings;
+the other scripts source it, and so can you:
+
+```bash
+source scripts/.deployment-env
+cd src/producers && python telemetry_producer.py --count 200 --devices 8
+```
+
+### Infrastructure as code
+
+The same topology through either IaC path; both then publish the application code with the CLI,
+and both write the same `scripts/.deployment-env`, so the demo runs after either one:
+
+```bash
+cd bicep     && bash deploy.sh    # or:
+cd terraform && bash deploy.sh
+
+bash ../scripts/validate.sh
+bash ../scripts/run-pipeline.sh
+```
+
+## The demo, step by step
+
+`scripts/run-pipeline.sh` asserts each stage rather than narrating it — it exits non-zero, and
+prints the processor's container logs, the moment the chain breaks:
+
+1. **Publish telemetry.** 120 readings across 6 devices, keyed by device id. One device
+   (`DEV-0003` by default) also emits readings above the temperature limit.
+2. **Wait for Capture.** Polls the blob container until a new archive appears. Nothing downstream
+   can happen until a window closes, so this sets the tempo.
+3. **Show the notification.** Reads `capture-notifications` — the hub Event Grid delivered into.
+4. **Read the curated summaries.** Polls the `curated` hub for the per-device aggregates the
+   processor produced, and fails if the excursion device is missing from them.
+5. **Recap** the path an event took.
+
+## Application
+
+### Producer (`src/producers/telemetry_producer.py`)
+
+Publishes cold-chain readings with the device id as the **partition key**, one batch per device.
+That is not a throughput trick: it is what guarantees a device's readings land in one archive, in
+order, so the archive-level aggregation is exact.
+
+```bash
+python telemetry_producer.py [--count 120] [--devices 6] [--excursion-device DEV-0003]
+```
+
+### Processor (`src/functions/function_app.py`)
+
+An Event Hubs-triggered function over `capture-notifications`. For each notification it:
+
+1. parses the Event Grid **batch array** (Event Grid always delivers an array — "an array with a
+   single event" by default),
+2. downloads the archive named by `data.fileUrl`,
+3. decodes the Avro records,
+4. aggregates per device (count, min/mean/max, excursions), carrying the archive's sequence range
+   through so a downstream consumer can spot a gap between windows,
+5. writes one summary per device via the **output binding**.
+
+Two details that are easy to get wrong, both commented in the code:
+
+- The trigger declares `cardinality=func.Cardinality.MANY`. The default delivers a bare
+  `EventHubEvent` and the handler fails with *"'EventHubEvent' object is not iterable"*.
+- The handler annotates `List[func.EventHubEvent]` from `typing`. The builtin `list[...]` makes
+  the Functions worker reject the function as an *"invalid non-type annotation"*, and it never
+  loads at all.
+
+## Notes
+
+- The first `deploy.sh` run pulls the Functions build image and can take several minutes.
+- `deploy.sh` is idempotent: every resource is created only when it does not already exist.
+- Capture's minimum interval is 60 seconds, so the demo always takes at least that long.
+- `skip_empty_archives` is on, so a window with no readings raises no event — otherwise the
+  processor would be woken for empty files.
@@ -0,0 +1,165 @@
+#!/bin/bash
+
+# =============================================================================
+# Deploys the cold-path pipeline with Bicep, then publishes the application code.
+# Bicep provisions the infrastructure; the Function App package is pushed with the Azure CLI, as
+# in the other samples.
+# =============================================================================
+
+PREFIX='local'
+LOCATION='westeurope'
+RESOURCE_GROUP_NAME="${PREFIX}-ehgrid-rg"
+DEPLOYMENT_NAME='eventhubs-eventgrid-coldpath'
+FUNCTION_ZIP='capture_processor.zip'
+
+CURRENT_DIR="$(cd "$(dirname "$0")" && pwd)"
+SRC_DIR="$(cd "$CURRENT_DIR/../src" && pwd)"
+cd "$CURRENT_DIR" || exit 1
+
+fail() {
+	echo "ERROR: $1"
+	exit 1
+}
+
+echo "=== Creating the resource group ==="
+az group create --name "$RESOURCE_GROUP_NAME" --location "$LOCATION" \
+	--only-show-errors 1>/dev/null || fail "could not create the resource group"
+echo "Resource group [$RESOURCE_GROUP_NAME] ready."
+
+echo ""
+echo "=== Validating the Bicep template ==="
+az deployment group validate \
+	--resource-group "$RESOURCE_GROUP_NAME" \
+	--template-file main.bicep \
+	--parameters main.bicepparam \
+	--only-show-errors 1>/dev/null || fail "the Bicep template did not pass validation"
+
+echo ""
+echo "=== Deploying the template ==="
+az deployment group create \
+	--name "$DEPLOYMENT_NAME" \
+	--resource-group "$RESOURCE_GROUP_NAME" \
+	--template-file main.bicep \
+	--parameters main.bicepparam \
+	--only-show-errors 1>/dev/null || fail "the Bicep deployment failed"
+echo "Template deployed."
+
+echo ""
+echo "=== Reading the deployment outputs ==="
+outputs=$(az deployment group show --name "$DEPLOYMENT_NAME" \
+	--resource-group "$RESOURCE_GROUP_NAME" --query properties.outputs \
+	--output json --only-show-errors)
+
+EVENTHUB_NAMESPACE_NAME=$(echo "$outputs" | jq -r '.eventHubNamespaceName.value')
+TELEMETRY_HUB_NAME=$(echo "$outputs" | jq -r '.telemetryHubName.value')
+NOTIFICATION_HUB_NAME=$(echo "$outputs" | jq -r '.notificationHubName.value')
+CURATED_HUB_NAME=$(echo "$outputs" | jq -r '.curatedHubName.value')
+CAPTURE_CONSUMER_GROUP=$(echo "$outputs" | jq -r '.captureConsumerGroup.value')
+STORAGE_ACCOUNT_NAME=$(echo "$outputs" | jq -r '.storageAccountName.value')
+CAPTURE_CONTAINER_NAME=$(echo "$outputs" | jq -r '.captureContainerName.value')
+SYSTEM_TOPIC_NAME=$(echo "$outputs" | jq -r '.systemTopicName.value')
+SUBSCRIPTION_NAME=$(echo "$outputs" | jq -r '.eventSubscriptionName.value')
+FUNCTION_APP_NAME=$(echo "$outputs" | jq -r '.functionAppName.value')
+TEMPERATURE_LIMIT=$(echo "$outputs" | jq -r '.temperatureLimit.value')
+
+[[ -n "$FUNCTION_APP_NAME" && "$FUNCTION_APP_NAME" != "null" ]] || fail "could not read the deployment outputs"
+echo "Namespace:    $EVENTHUB_NAMESPACE_NAME"
+echo "Function App: $FUNCTION_APP_NAME"
+echo "System topic: $SYSTEM_TOPIC_NAME -> $SUBSCRIPTION_NAME"
+
+# -----------------------------------------------------------------------------
+echo ""
+echo "=== Wiring the connection strings ==="
+# -----------------------------------------------------------------------------
+# Done here rather than in the template: connection strings are built from keys that only exist
+# once the resources do, and anything a template emits as an output is kept in the deployment
+# history. Reading them with the CLI keeps the secrets out of both.
+STORAGE_KEY=$(az storage account keys list --account-name "$STORAGE_ACCOUNT_NAME" \
+	--resource-group "$RESOURCE_GROUP_NAME" --query "[0].value" -o tsv --only-show-errors)
+STORAGE_BLOB=$(az storage account show -n "$STORAGE_ACCOUNT_NAME" -g "$RESOURCE_GROUP_NAME" \
+	--query primaryEndpoints.blob -o tsv --only-show-errors)
+STORAGE_QUEUE=$(az storage account show -n "$STORAGE_ACCOUNT_NAME" -g "$RESOURCE_GROUP_NAME" \
+	--query primaryEndpoints.queue -o tsv --only-show-errors)
+STORAGE_TABLE=$(az storage account show -n "$STORAGE_ACCOUNT_NAME" -g "$RESOURCE_GROUP_NAME" \
+	--query primaryEndpoints.table -o tsv --only-show-errors)
+# Explicit endpoints, not EndpointSuffix: the Functions host cannot parse a suffix with a port.
+STORAGE_CONNECTION_STRING="DefaultEndpointsProtocol=https;AccountName=${STORAGE_ACCOUNT_NAME};AccountKey=${STORAGE_KEY};BlobEndpoint=${STORAGE_BLOB};QueueEndpoint=${STORAGE_QUEUE};TableEndpoint=${STORAGE_TABLE}"
+
+SEND_CONNECTION=$(az eventhubs eventhub authorization-rule keys list --name telemetry-send \
+	--eventhub-name "$TELEMETRY_HUB_NAME" --namespace-name "$EVENTHUB_NAMESPACE_NAME" \
+	--resource-group "$RESOURCE_GROUP_NAME" --query primaryConnectionString -o tsv --only-show-errors)
+LISTEN_CONNECTION=$(az eventhubs namespace authorization-rule keys list --name pipeline-listen \
+	--namespace-name "$EVENTHUB_NAMESPACE_NAME" --resource-group "$RESOURCE_GROUP_NAME" \
+	--query primaryConnectionString -o tsv --only-show-errors)
+# Namespace-level for the bindings: an entity-level string carries EntityPath, which would pin the
+# output binding to the notifications hub instead of the curated one.
+NAMESPACE_CONNECTION=$(az eventhubs namespace authorization-rule keys list \
+	--name RootManageSharedAccessKey --namespace-name "$EVENTHUB_NAMESPACE_NAME" \
+	--resource-group "$RESOURCE_GROUP_NAME" --query primaryConnectionString -o tsv --only-show-errors)
+
+[[ -n "$SEND_CONNECTION" ]] || fail "could not read the send connection string"
+[[ -n "$LISTEN_CONNECTION" ]] || fail "could not read the listen connection string"
+[[ -n "$NAMESPACE_CONNECTION" ]] || fail "could not read the namespace connection string"
+
+az functionapp config appsettings set --name "$FUNCTION_APP_NAME" \
+	--resource-group "$RESOURCE_GROUP_NAME" \
+	--settings \
+	"AzureWebJobsStorage=$STORAGE_CONNECTION_STRING" \
+	"WEBSITE_CONTENTAZUREFILECONNECTIONSTRING=$STORAGE_CONNECTION_STRING" \
+	"STORAGE_CONNECTION_STRING=$STORAGE_CONNECTION_STRING" \
+	"EVENTHUB_NOTIFICATION_CONNECTION=$NAMESPACE_CONNECTION" \
+	"EVENTHUB_CURATED_CONNECTION=$NAMESPACE_CONNECTION" \
+	"ENABLE_ORYX_BUILD=true" \
+	"SCM_DO_BUILD_DURING_DEPLOYMENT=true" \
+	--only-show-errors 1>/dev/null || fail "could not set the function app settings"
+echo "Configured the processor."
+
+# -----------------------------------------------------------------------------
+echo ""
+echo "=== Publishing the application code ==="
+# -----------------------------------------------------------------------------
+cd "$SRC_DIR/functions" || fail "missing src/functions"
+rm -f "$CURRENT_DIR/$FUNCTION_ZIP"
+zip -r "$CURRENT_DIR/$FUNCTION_ZIP" function_app.py host.json requirements.txt 1>/dev/null
+cd "$CURRENT_DIR" || exit 1
+
+echo "Deploying the capture processor (the first run can take several minutes)..."
+az functionapp deploy \
+	--resource-group "$RESOURCE_GROUP_NAME" \
+	--name "$FUNCTION_APP_NAME" \
+	--src-path "$FUNCTION_ZIP" \
+	--type zip 1>/dev/null || fail "could not deploy the function app"
+rm -f "$FUNCTION_ZIP"
+
+# -----------------------------------------------------------------------------
+echo ""
+echo "=== Verifying the deployment ==="
+# -----------------------------------------------------------------------------
+az eventhubs eventhub show --name "$TELEMETRY_HUB_NAME" \
+	--namespace-name "$EVENTHUB_NAMESPACE_NAME" --resource-group "$RESOURCE_GROUP_NAME" \
+	--query "{name:name, partitions:partitionCount, capture:captureDescription.enabled}" \
+	--output table --only-show-errors || fail "the telemetry hub is not readable"
+
+# Mirror scripts/deploy.sh so the demo runs after a Bicep deployment too.
+cat >"$CURRENT_DIR/../scripts/.deployment-env" <<EOF
+export RESOURCE_GROUP_NAME='$RESOURCE_GROUP_NAME'
+export EVENTHUB_NAMESPACE_NAME='$EVENTHUB_NAMESPACE_NAME'
+export TELEMETRY_HUB_NAME='$TELEMETRY_HUB_NAME'
+export NOTIFICATION_HUB_NAME='$NOTIFICATION_HUB_NAME'
+export CURATED_HUB_NAME='$CURATED_HUB_NAME'
+export CAPTURE_CONSUMER_GROUP='$CAPTURE_CONSUMER_GROUP'
+export CAPTURE_CONTAINER_NAME='$CAPTURE_CONTAINER_NAME'
+export STORAGE_ACCOUNT_NAME='$STORAGE_ACCOUNT_NAME'
+export SYSTEM_TOPIC_NAME='$SYSTEM_TOPIC_NAME'
+export SUBSCRIPTION_NAME='$SUBSCRIPTION_NAME'
+export FUNCTION_APP_NAME='$FUNCTION_APP_NAME'
+export TEMPERATURE_LIMIT='$TEMPERATURE_LIMIT'
+export EVENTHUB_SEND_CONNECTION_STRING='$SEND_CONNECTION'
+export EVENTHUB_LISTEN_CONNECTION_STRING='$LISTEN_CONNECTION'
+export EVENTHUB_NAMESPACE_CONNECTION_STRING='$NAMESPACE_CONNECTION'
+export STORAGE_CONNECTION_STRING='$STORAGE_CONNECTION_STRING'
+EOF
+
+echo ""
+echo "Deployment complete. Run 'bash ../scripts/validate.sh' to exercise every capability,"
+echo "then 'bash ../scripts/run-pipeline.sh' for the end-to-end demo."