🛡️ Tutorial 11: Error Handling and Resilience¶

# File: ErrorHandler/__init__.py

import logging
import json
import os
from datetime import datetime
import azure.functions as func
from azure.eventhub import EventHubProducerClient, EventData

def main(req: func.HttpRequest) -> func.HttpResponse:
    """
    Handle failed events and send to dead-letter queue.
    """
    logging.info('Processing error event')

    try:
        # Get original event data
        req_body = req.get_json()

        # Enrich with error metadata
        error_event = {
            **req_body,
            'error': {
                'timestamp': datetime.utcnow().isoformat(),
                'source': 'StreamAnalytics',
                'reason': req_body.get('errorReason', 'Unknown'),
                'retryCount': req_body.get('retryCount', 0)
            }
        }

        # Send to dead-letter queue
        send_to_dlq(error_event)

        return func.HttpResponse(
            json.dumps({"status": "success", "message": "Event sent to DLQ"}),
            status_code=200,
            mimetype="application/json"
        )

    except Exception as e:
        logging.error(f"Error handling failed event: {str(e)}")
        return func.HttpResponse(
            json.dumps({"status": "error", "message": str(e)}),
            status_code=500,
            mimetype="application/json"
        )


def send_to_dlq(event: dict):
    """Send event to dead-letter queue."""
    connection_string = os.environ.get('STREAM_DLQ_CONN')
    eventhub_name = "dead-letter-queue"

    producer = EventHubProducerClient.from_connection_string(
        conn_str=connection_string,
        eventhub_name=eventhub_name
    )

    try:
        event_data_batch = producer.create_batch()
        event_data_batch.add(EventData(json.dumps(event)))
        producer.send_batch(event_data_batch)
        logging.info(f"Event sent to DLQ: {event.get('deviceId', 'unknown')}")
    finally:
        producer.close()

# File: monitor_dlq.py

import os
from azure.eventhub import EventHubConsumerClient

def on_event(partition_context, event):
    """Process dead-letter event."""
    event_data = json.loads(event.body_as_str())

    print(f"\n{'='*60}")
    print(f"Dead-Letter Event Detected")
    print(f"{'='*60}")
    print(f"Device ID: {event_data.get('deviceId')}")
    print(f"Original Timestamp: {event_data.get('timestamp')}")
    print(f"Error Timestamp: {event_data.get('error', {}).get('timestamp')}")
    print(f"Error Reason: {event_data.get('error', {}).get('reason')}")
    print(f"Retry Count: {event_data.get('error', {}).get('retryCount')}")
    print(f"Original Event: {json.dumps(event_data, indent=2)}")
    print(f"{'='*60}\n")

    partition_context.update_checkpoint(event)


def monitor_dlq():
    """Monitor dead-letter queue for failed events."""
    connection_string = os.environ.get("STREAM_DLQ_CONN")
    eventhub_name = "dead-letter-queue"

    consumer = EventHubConsumerClient.from_connection_string(
        conn_str=connection_string,
        consumer_group="$Default",
        eventhub_name=eventhub_name
    )

    print("Monitoring dead-letter queue for failed events...")
    print("Press Ctrl+C to stop\n")

    try:
        with consumer:
            consumer.receive(
                on_event=on_event,
                starting_position="-1"
            )
    except KeyboardInterrupt:
        print("\nMonitoring stopped")


if __name__ == "__main__":
    monitor_dlq()

# File: ResilientFunction/__init__.py

import logging
import time
import json
from typing import Optional
import azure.functions as func

MAX_RETRIES = 3
BASE_DELAY = 1  # seconds

def exponential_backoff_retry(func, *args, **kwargs):
    """
    Retry function with exponential backoff.
    """
    for attempt in range(MAX_RETRIES):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            if attempt == MAX_RETRIES - 1:
                # Last attempt failed
                logging.error(f"All {MAX_RETRIES} attempts failed: {str(e)}")
                raise
            else:
                # Calculate delay: 1s, 2s, 4s
                delay = BASE_DELAY * (2 ** attempt)
                logging.warning(f"Attempt {attempt + 1} failed, retrying in {delay}s: {str(e)}")
                time.sleep(delay)


def send_to_external_api(data: dict) -> Optional[dict]:
    """Send data to external API with retry logic."""
    import requests

    url = "https://api.example.com/events"
    headers = {"Content-Type": "application/json"}

    response = requests.post(url, json=data, headers=headers, timeout=10)
    response.raise_for_status()

    return response.json()


def main(req: func.HttpRequest) -> func.HttpResponse:
    """Process event with retry logic."""
    try:
        req_body = req.get_json()

        # Process with retry
        result = exponential_backoff_retry(send_to_external_api, req_body)

        return func.HttpResponse(
            json.dumps({"status": "success", "result": result}),
            status_code=200,
            mimetype="application/json"
        )

    except Exception as e:
        logging.error(f"Failed to process event after retries: {str(e)}")
        return func.HttpResponse(
            json.dumps({"status": "error", "message": str(e)}),
            status_code=500,
            mimetype="application/json"
        )