Add an Event Hubs fraud detection sample

.gitignore

@@ -223,10 +223,18 @@ cython_debug/
 .idea/
 ## Terraform
 .terraform/
+.terraform.lock.hcl
 *.tfstate
 *.tfstate.backup
 tfplan
 override.tf
 override.tf.json
 *_override.tf
 *_override.tf.json
+
+## Sample deployment artifacts
+# Written by a sample's deploy.sh and read back by its other scripts. It holds live
+# connection strings, so it must never be committed.
+.deployment-env
+# Application packages the deploy scripts build on the fly before publishing them.
+*.zip

README.md

@@ -39,6 +39,7 @@ This repository contains comprehensive sample projects demonstrating how to deve
 | [Web App with Custom Docker Image](./samples/web-app-custom-image/python/README.md) | Azure Web App running a custom Docker image |
 | [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 |
 
 ## Sample Structure
 

requirements-runtime.txt

@@ -6,6 +6,10 @@ azure-identity
 azure-storage-blob
 azure-mgmt-cosmosdb
 azure-core
+azure-eventhub
+confluent-kafka
+fastavro
+requests
 python-dotenv
 localstack
 azlocal

run-samples.sh

@@ -32,6 +32,7 @@ PURGE_DOCKER="${PURGE_DOCKER:-0}"
 # 1. Define Samples (placed before tool checks so --list works without dependencies)
 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/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=(
 # 1a. Define Terraform Samples
 TERRAFORM_SAMPLES=(
   "samples/servicebus/java/terraform|bash deploy.sh"
+  "samples/eventhubs/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"
@@ -62,6 +64,7 @@ TERRAFORM_SAMPLES=(
 # 1b. Define Bicep Samples
 BICEP_SAMPLES=(
   "samples/servicebus/java/bicep|bash deploy.sh"
+  "samples/eventhubs/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"
@@ -86,15 +89,19 @@ if [[ "${1:-}" == "--list" ]]; then
     IFS='|' read -r path _ _ <<< "${ALL_SAMPLES[$i]}"
 
     if [[ "$path" == */terraform || "$path" == */bicep ]]; then
-      watch=("$path" "$(dirname "$path")/src")
+      # An IaC sample deploys from its own folder, but its test command may run the
+      # sample's shared scripts, so a change there has to re-run it too. Folders that do
+      # not exist simply never match a changed file.
+      watch=("$path" "$(dirname "$path")/src" "$(dirname "$path")/scripts")
       name="${path#samples/}"
     else
       watch=("$path/scripts" "$path/src")
       name="${path#samples/}/scripts"
     fi
 
-    printf '  {"shard":%d,"splits":%d,"name":"%s","watch_folders":["%s","%s"]}' \
-      $((i+1)) "$TOTAL" "$name" "${watch[0]}" "${watch[1]}"
+    folders=$(printf '"%s",' "${watch[@]}")
+    printf '  {"shard":%d,"splits":%d,"name":"%s","watch_folders":[%s]}' \
+      $((i+1)) "$TOTAL" "$name" "${folders%,}"
     (( i < TOTAL-1 )) && echo "," || echo ""
   done
   echo "]"

samples/eventhubs/python/README.md

@@ -0,0 +1,200 @@
+# Real-time payment fraud detection with Azure Event Hubs
+
+A streaming platform in miniature, running entirely on LocalStack for Azure. Payments
+arrive from three different kinds of producer, a serverless processor scores them for
+fraud in real time, alerts flow onto a second stream, and every event is archived to a
+data lake without a line of code being written for it.
+
+The sample is built around the properties that make Event Hubs different from a queue:
+**one log, many protocols, many independent readers, and a replayable history**.
+
+## Architecture
+
+```
+  Producers (outside Azure)                LocalStack for Azure
+  ┌───────────────────────┐        ┌──────────────────────────────────────────────┐
+  │ POS terminals         │  AMQP  │  Event Hubs namespace (Standard, Kafka on)   │
+  │ producer_amqp.py      ├───────►│                                              │
+  │                       │        │   payments hub ─ 4 partitions ─── Capture ───┼──► Blob Storage
+  │ Legacy gateway        │ Kafka  │     consumer groups:                         │    Avro archives
+  │ producer_kafka.py     ├───────►│       fraud-detector ──┐  analytics  audit   │    (cold path)
+  │                       │        │                        │                     │
+  │ ATM / IoT devices     │ HTTPS  │                        ▼                     │
+  │ producer_http.py      ├───────►│              Function App (Python)           │
+  └───────────────────────┘        │              Event Hubs trigger              │
+                                   │              fraud rules ──────┐             │
+                                   │                                ▼             │
+                                   │   fraud-alerts hub ─ 2 partitions            │
+                                   │                                ▲             │
+                                   │   Key Vault: connection strings │            │
+                                   │   Schema Registry: Avro contract│            │
+                                   │   App Insights + Log Analytics  │            │
+                                   │                                 │            │
+                                   │   Web App ── operations dashboard            │
+                                   └──────────────────────────────────────────────┘
+```
+
+**Deployment flow.** The deploy script creates Log Analytics and Application Insights, a
+storage account with the Capture container, the Event Hubs namespace with both hubs, three
+consumer groups, least-privilege authorization rules and a Schema Registry group; stores
+the connection strings in Key Vault; then deploys the Function App and the dashboard Web
+App.
+
+**At runtime.** Producers publish payments over AMQP, Kafka and HTTPS to the same hub. The
+Function App's Event Hubs trigger reads the `fraud-detector` consumer group in batches,
+applies the fraud rules, and writes alerts to the `fraud-alerts` hub through an output
+binding. Event Hubs Capture independently archives everything to Blob Storage as Avro. The
+dashboard reads the runtime state of both hubs, the checkpoints, the archives and the
+schema registry.
+
+## What this demonstrates
+
+| Capability | Where to see it |
+|------------|-----------------|
+| **Multi-protocol ingestion** - AMQP 1.0, Kafka and HTTPS into one hub | `src/producers/`, step 2 of `run-pipeline.sh` |
+| **Kafka compatibility** - an existing librdkafka app with no Azure SDK | `src/producers/producer_kafka.py` |
+| **Partition keys and ordering** - all of an account's payments on one partition | `producer_amqp.py`, the velocity rule in `function_app.py` |
+| **Consumer groups** - three independent readers of the same stream | `deploy.sh` step 6 |
+| **Serverless stream processing** - Event Hubs trigger, batched | `src/functions/function_app.py` |
+| **Checkpointing and recovery** - restart resumes where it left off | Step 4 of `run-pipeline.sh` |
+| **Event Hubs Capture** - Avro archives to Blob Storage, no consumer code | Step 5 of `run-pipeline.sh`, `scripts/read_capture.py` |
+| **Schema Registry** - a versioned Avro contract | `src/producers/schema_register.py` |
+| **Least-privilege SAS** - send-only and listen-only rules | `deploy.sh` step 7, `validate.sh` check 5 |
+| **Auto-inflate** - throughput units scale with ingress | `deploy.sh` step 4 |
+| **Geo-disaster recovery** - namespace pairing behind an alias | `validate.sh` check 11 |
+| **Application groups** - client throttling policies | `validate.sh` check 10 |
+| **Service integration** - Event Hubs with Storage, Key Vault, Functions, App Service, Monitor | the whole sample |
+
+## Prerequisites
+
+- [LocalStack](https://docs.localstack.cloud/getting-started/installation/) with a valid
+  `LOCALSTACK_AUTH_TOKEN`
+- [Docker](https://docs.docker.com/get-docker/)
+- [Azure CLI](https://learn.microsoft.com/en-us/cli/azure/install-azure-cli) and
+  [azlocal](https://pypi.org/project/azlocal/) (`pip install azlocal`)
+- Python 3.12+ with `src/producers/requirements.txt` installed
+- `jq` and `zip`
+- [Terraform](https://developer.hashicorp.com/terraform/downloads) or the Bicep CLI, for
+  the alternative deployments
+
+## Quick start
+
+```bash
+# Start the LocalStack Azure emulator
+export LOCALSTACK_AUTH_TOKEN=<your_auth_token>
+IMAGE_NAME=localstack/localstack-azure localstack start -d
+localstack wait -t 120
+
+# Route the Azure CLI to the emulator
+azlocal start-interception
+
+cd samples/eventhubs/python
+pip install -r src/producers/requirements.txt
+
+bash scripts/deploy.sh        # create the pipeline
+bash scripts/validate.sh      # verify every capability
+bash scripts/run-pipeline.sh  # run the end-to-end demo
+```
+
+Then open the dashboard URL printed at the end of the deployment.
+
+> **Note**
+> The first deployment downloads the Azure Functions build image and core tools inside the
+> emulator, which can take several minutes. Later runs reuse them.
+
+## Alternative deployments
+
+```bash
+bash bicep/deploy.sh        # Bicep
+bash terraform/deploy.sh    # Terraform
+```
+
+Both provision the same topology and then publish the same application packages.
+
+## The demo, step by step
+
+`scripts/run-pipeline.sh` walks through the whole story:
+
+1. **Publish the contract.** The Avro payment schema is registered in the Schema Registry
+   and read back by id, the way a serializer resolves one at runtime.
+2. **Ingest over three protocols.** 40 payments over AMQP (keyed by account, plus a burst
+   that trips the velocity rule), 15 over Kafka, 8 over HTTPS. The partition counts show
+   they all landed in the same log.
+3. **Detect fraud.** The Function App picks up each batch, applies the amount and velocity
+   rules, and emits alerts to `fraud-alerts`.
+4. **Prove checkpointing.** The function app is restarted while new payments arrive. It
+   resumes from its checkpoint: nothing is lost and nothing is processed twice.
+5. **Read the archive.** Once a Capture window flushes, the newest Avro blob is decoded
+   and its records printed - the cold path, with no consumer code.
+6. **Point at the dashboard** for the live view.
+
+Each step asserts its own result rather than reporting it: the script exits non-zero (and
+prints the Function App's container logs) if the hub does not grow by the events that were
+published, if no alert arrives, if the processor replays instead of resuming, or if Capture
+writes nothing decodable. A run that completes is evidence the pipeline works, which is
+what makes it usable as a CI test.
+
+## Application
+
+### Producers (`src/producers/`)
+
+| File | Role |
+|------|------|
+| `common.py` | Shared config, the Avro contract, the payment generator |
+| `producer_amqp.py` | POS terminals - official SDK, batches, partition keys |
+| `producer_kafka.py` | Legacy gateway - confluent-kafka, bootstrap derived from the connection string |
+| `producer_http.py` | ATM/IoT - HTTP runtime API with a hand-built SAS token |
+| `schema_register.py` | Registers and resolves the Avro contract |
+
+### Processor (`src/functions/`)
+
+`function_app.py` is an Event Hubs-triggered function with an Event Hubs output binding.
+Two rules: an amount threshold, and a per-account velocity window that is only correct
+because producers key events by account, which pins each account to one partition.
+
+### Dashboard (`src/dashboard/`)
+
+A Flask app that polls the Event Hubs runtime API, the checkpoint blobs, the alerts hub,
+the Capture container and the Schema Registry, and refreshes every five seconds.
+
+| Route | Description |
+|-------|-------------|
+| `/` | The dashboard |
+| `/api/overview` | JSON snapshot of the whole pipeline |
+| `/health` | Health check |
+
+## Scripts
+
+| Script | Description |
+|--------|-------------|
+| `scripts/deploy.sh` | Creates the full topology and deploys both workloads |
+| `scripts/validate.sh` | Numbered checks over the control plane, data plane and bonus capabilities |
+| `scripts/run-pipeline.sh` | The end-to-end demo |
+| `scripts/cleanup.sh` | Deletes the resource group and local artifacts |
+| `bicep/deploy.sh`, `terraform/deploy.sh` | The alternative deployment paths |
+
+## Cleanup
+
+```bash
+bash scripts/cleanup.sh
+```
+
+## Emulator notes
+
+- Kafka is served in plaintext by the emulator; Azure requires `SASL_SSL`. The producer
+  shows the SASL settings needed against Azure in a comment.
+- Data-plane authorization is existence-checked by the emulator, so the send-only and
+  listen-only rules demonstrate the *deployment* pattern rather than an enforced boundary.
+  Both rules are used exactly as they would be against Azure.
+- The emulator serves the HTTP runtime API on the gateway port; `common.py` derives it
+  from the connection string so the same code works against Azure unchanged.
+
+## References
+
+- [Azure Event Hubs documentation](https://learn.microsoft.com/en-us/azure/event-hubs/event-hubs-about)
+- [Event Hubs Capture](https://learn.microsoft.com/en-us/azure/event-hubs/event-hubs-capture-overview)
+- [Event Hubs for Apache Kafka](https://learn.microsoft.com/en-us/azure/event-hubs/azure-event-hubs-apache-kafka-overview)
+- [Azure Schema Registry](https://learn.microsoft.com/en-us/azure/event-hubs/schema-registry-overview)
+- [Azure Functions Event Hubs trigger](https://learn.microsoft.com/en-us/azure/azure-functions/functions-bindings-event-hubs-trigger)
+- [Event Hubs quotas and limits](https://learn.microsoft.com/en-us/azure/event-hubs/event-hubs-quotas)
+- [LocalStack for Azure](https://docs.localstack.cloud/azure/)

samples/eventhubs/python/bicep/README.md

@@ -0,0 +1,64 @@
+# Bicep deployment
+
+Provisions the payment fraud detection pipeline with a Bicep template, then publishes the
+application code with the Azure CLI.
+
+## Prerequisites
+
+- The LocalStack Azure emulator is running and `azlocal start-interception` has been run.
+- The Bicep CLI (`az bicep install`) and `jq` are available.
+
+## Usage
+
+```bash
+cd samples/eventhubs/python/bicep
+bash deploy.sh
+```
+
+`deploy.sh` creates the resource group, validates and deploys `main.bicep`, reads the
+outputs, wires the connection strings, zip-deploys the Function App and the Web App, and
+verifies the payments hub.
+
+### Why the connection strings are set by the script, not the template
+
+Anything a template emits as an output is retained in the deployment history, so this sample
+keeps keys out of the template entirely: `deploy.sh` reads them with
+`az ... authorization-rule keys list` once the resources exist and writes them straight into
+Key Vault and the app settings. The storage connection string is assembled from the
+account's own endpoints, because the Azure Functions host cannot parse the `EndpointSuffix`
+form when the suffix carries a port.
+
+## Template layout
+
+| File | Contents |
+|------|----------|
+| `main.bicep` | Parameters, names, and the module graph |
+| `main.bicepparam` | Parameter values for the sample |
+| `modules/storage.bicep` | Storage account and the Capture container |
+| `modules/event-hubs.bicep` | Namespace, both hubs, Capture, consumer groups, authorization rules, schema group |
+| `modules/key-vault.bicep` | Key vault (the secrets themselves are written by `deploy.sh`) |
+| `modules/monitoring.bicep` | Log Analytics workspace and Application Insights |
+| `modules/workloads.bicep` | App Service plans, the Function App and the Web App |
+
+## Parameters
+
+| Parameter | Default | Notes |
+|-----------|---------|-------|
+| `prefix` / `suffix` | `local` / `payments` | Compose every resource name |
+| `eventHubSkuName` | `Standard` | Capture and Kafka need Standard or higher |
+| `partitionCount` | `4` | Partitions are the unit of parallelism and of ordering |
+| `retentionTimeInHours` | `24` | Standard tier allows up to 7 days |
+| `captureIntervalInSeconds` | `60` | Minimum allowed by Azure |
+| `fraudAmountThreshold` | `5000` | A single payment at or above this is flagged |
+| `fraudVelocityCount` | `5` | Payments per account per window before flagging |
+
+## Outputs
+
+`main.bicep` returns the resource names and the dashboard URL:
+
+```bash
+az deployment group show \
+  --name eventhubs-fraud-detection \
+  --resource-group local-eventhubs-rg \
+  --query properties.outputs
+```