frootai/frootai/fai-azure-event-hubs-setup

// infra/event-hubs.bicep
param location string
param namespaceName string
 
resource namespace 'Microsoft.EventHub/namespaces@2023-01-01-preview' = {
  name: namespaceName
  location: location
  sku: {
    name: 'Standard'
    capacity: 1
  }
  identity: { type: 'SystemAssigned' }
  properties: {
    publicNetworkAccess: 'Disabled'
    minimumTlsVersion: '1.2'
    zoneRedundant: true
  }
}
 
// Event Hubs (topics)
resource telemetryHub 'Microsoft.EventHub/namespaces/eventhubs@2023-01-01-preview' = {
  parent: namespace
  name: 'ai-telemetry'
  properties: {
    messageRetentionInDays: 7
    partitionCount: 32        // Auto-scale across partitions
    captureDescription: {
      enabled: true
      encoding: 'Avro'
      destination: {
        name: 'EventHubArchive.AzureBlockBlob'
        properties: {
          storageAccountResourceId: storageAccountId
          blobContainer: 'event-hubs-archive'
          archiveNameFormat: '{Namespace}/{EventHub}/{PartitionId}/{Year}/{Month}/{Day}/{Hour}/{Minute}/{Second}'
        }
      }
    }
  }
}

# Consumer group for real-time metrics processor
az eventhubs eventhub consumer-group create \
  --resource-group $RG_NAME \
  --namespace-name $EH_NAMESPACE \
  --eventhub-name ai-telemetry \
  --name metrics-processor
 
# Consumer group for ML feature extraction
az eventhubs eventhub consumer-group create \
  --resource-group $RG_NAME \
  --namespace-name $EH_NAMESPACE \
  --eventhub-name ai-telemetry \
  --name ml-feature-extractor
 
# Consumer group for audit/compliance logging
az eventhubs eventhub consumer-group create \
  --resource-group $RG_NAME \
  --namespace-name $EH_NAMESPACE \
  --eventhub-name ai-telemetry \
  --name audit-logger

from azure.eventhub import EventHubProducerClient, EventData
from azure.identity import DefaultAzureCredential
import json
import time
producer = EventHubProducerClient(
    fully_qualified_namespace=f"{EH_NAMESPACE}.servicebus.windows.net",
    eventhub_name="ai-telemetry",
    credential=DefaultAzureCredential(),
)
def send_completion_event(request_id: str, latency_ms: int, tokens: int):
    event_data = EventData(
        json.dumps({
            "type": "completion",
            "request_id": request_id,
            "timestamp": time.time(),
            "latency_ms": latency_ms,
            "tokens": tokens,
        })
    )
    
    # Batch send for throughput
    with producer:
        batch = producer.create_batch()
        batch.add(event_data)
        producer.send_batch(batch)

-- Azure Stream Analytics job query
SELECT
    System.Timestamp('10 second') as window_end,
    COUNT(*) as event_count,
    AVG(latency_ms) as avg_latency,
    MAX(latency_ms) as max_latency,
    SUM(tokens) as total_tokens
INTO
    output_metrics_table
FROM
    event_hub_input TIMESTAMP BY timestamp
GROUP BY
    TumblingWindow(second, 10)

from azure.eventhub import EventHubConsumerClient
from azure.identity import DefaultAzureCredential
consumer = EventHubConsumerClient(
    fully_qualified_namespace=f"{EH_NAMESPACE}.servicebus.windows.net",
    consumer_group="metrics-processor",
    eventhub_name="ai-telemetry",
    credential=DefaultAzureCredential(),
)
def on_event(partition_context, event):
    print(f"Received event from partition {partition_context.partition_id}: {event.body_as_json()}")
    partition_context.update_latest_offset_and_checkpoint(event)
def on_error(partition_context, error):
    print(f"Error in partition {partition_context.partition_id}: {error}")
# Start consuming
consumer.receive(on_event=on_event, on_error=on_error, starting_position="@latest")

from azure.eventhub.aio import EventHubConsumerClient
import asyncio
async def check_consumer_lag():
    client = EventHubConsumerClient(
        fully_qualified_namespace=f"{EH_NAMESPACE}.servicebus.windows.net",
        consumer_group="metrics-processor",
        eventhub_name="ai-telemetry",
        credential=DefaultAzureCredential(),
    )
    
    partition_ids = await client.get_partition_ids()
    
    async with client:
        for pid in partition_ids:
            properties = await client.get_partition_properties(pid)
            last_offset = properties.last_enqueued_offset
            # Consumer lag = (last_offset - consumer_checkpoint) / last_offset
            print(f"Partition {pid}: {properties.last_enqueued_event_number} events")