Workbooks & KQL Monitoring Dashboard

{
  "version": "Notebook/1.0",
  "items": [
    {
      "type": 1,
      "content": {
        "json": "## Business Funnel Analysis\n\nMonitor conversion rates and user behavior patterns in real-time."
      }
    },
    {
      "type": 3,
      "content": {
        "version": "KqlItem/1.0",
        "query": "customEvents\n| where timestamp >= ago(24h)\n| where name in (\"view_event\", \"cart_event\", \"purchase_event\")\n| extend event_type = case(\n    name == \"view_event\", \"view\",\n    name == \"cart_event\", \"cart\", \n    name == \"purchase_event\", \"purchase\",\n    \"unknown\"\n)\n| summarize \n    view_count = countif(event_type == \"view\"),\n    cart_count = countif(event_type == \"cart\"),\n    purchase_count = countif(event_type == \"purchase\")\n    by bin(timestamp, 1h)\n| extend \n    cart_rate = cart_count / view_count,\n    purchase_rate = purchase_count / view_count\n| render timechart",
        "size": 0,
        "title": "Event Distribution Over Time",
        "queryType": 0,
        "resourceType": "microsoft.insights/components"
      }
    }
  ]
}

{
  "type": 3,
  "content": {
    "version": "KqlItem/1.0",
    "query": "customEvents\n| where timestamp >= ago(24h)\n| where name == \"data_validation\"\n| extend \n    validation_error = tostring(customDimensions.error_type),\n    error_count = toint(customDimensions.error_count)\n| summarize \n    total_errors = sum(error_count),\n    error_types = dcount(validation_error)\n    by validation_error, bin(timestamp, 1h)\n| render barchart",
    "size": 0,
    "title": "Data Quality Issues",
    "queryType": 0,
    "resourceType": "microsoft.insights/components"
  }
}

{
  "type": 3,
  "content": {
    "version": "KqlItem/1.0",
    "query": "customEvents\n| where timestamp >= ago(24h)\n| where name == \"feature_distribution\"\n| extend \n    price = todouble(customDimensions.price),\n    category = tostring(customDimensions.category),\n    brand = tostring(customDimensions.brand)\n| summarize \n    price_mean = avg(price),\n    price_std = stdev(price),\n    category_diversity = dcount(category),\n    brand_diversity = dcount(brand)\n    by bin(timestamp, 1h)\n| render timechart",
    "size": 0,
    "title": "Feature Distribution Drift",
    "queryType": 0,
    "resourceType": "microsoft.insights/components"
  }
}

{
  "type": 3,
  "content": {
    "version": "KqlItem/1.0",
    "query": "requests\n| where timestamp >= ago(24h)\n| where name contains \"retail-forecast\"\n| summarize \n    p50 = percentile(duration, 50),\n    p95 = percentile(duration, 95),\n    p99 = percentile(duration, 99),\n    error_rate = countif(success == false) * 100.0 / count()\n    by bin(timestamp, 5m)\n| render timechart",
    "size": 0,
    "title": "Latency and Error Rates",
    "queryType": 0,
    "resourceType": "microsoft.insights/components"
  }
}

# etl_telemetry.py - Add to ETL pipeline
from opencensus.ext.azure.log_exporter import AzureLogHandler
from opencensus.ext.azure.trace_exporter import AzureExporter
from opencensus.trace import config_integration
from opencensus.trace.samplers import ProbabilitySampler
from opencensus.trace.tracer import Tracer
import logging

def setup_telemetry():
    """Setup Application Insights telemetry"""
    # Configure logging
    logger = logging.getLogger(__name__)
    logger.addHandler(AzureLogHandler(connection_string="your-connection-string"))
    
    # Configure tracing
    config_integration.trace_integrations(['requests'])
    tracer = Tracer(
        exporter=AzureExporter(connection_string="your-connection-string"),
        sampler=ProbabilitySampler(rate=1.0)
    )
    
    return logger, tracer

def log_business_event(event_type, properties):
    """Log business events for funnel analysis"""
    logger, tracer = setup_telemetry()
    
    with tracer.span(name=f"{event_type}_event"):
        logger.info(f"Business event: {event_type}", extra={
            "custom_dimensions": {
                "event_type": event_type,
                "user_id": properties.get("user_id"),
                "product_id": properties.get("product_id"),
                "category_code": properties.get("category_code"),
                "brand": properties.get("brand"),
                "price": properties.get("price")
            }
        })

def log_data_quality_metrics(metrics):
    """Log data quality metrics"""
    logger, _ = setup_telemetry()
    
    logger.info("Data quality metrics", extra={
        "custom_dimensions": {
            "total_records": metrics["total_records"],
            "invalid_records": metrics["invalid_records"],
            "missing_values": metrics["missing_values"],
            "schema_errors": metrics["schema_errors"],
            "quality_score": metrics["quality_score"]
        }
    })

def log_feature_distribution(features):
    """Log feature distribution for drift detection"""
    logger, _ = setup_telemetry()
    
    logger.info("Feature distribution", extra={
        "custom_dimensions": {
            "price_mean": features["price_mean"],
            "price_std": features["price_std"],
            "category_distribution": features["category_dist"],
            "brand_distribution": features["brand_dist"]
        }
    })

# inference_telemetry.py - Add to inference service
import time
from functools import wraps

def track_inference_metrics(func):
    """Decorator to track inference metrics"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        start_time = time.time()
        
        try:
            result = func(*args, **kwargs)
            duration = time.time() - start_time
            
            # Log successful inference
            logger.info("Inference completed", extra={
                "custom_dimensions": {
                    "duration_ms": duration * 1000,
                    "success": True,
                    "model_version": kwargs.get("model_version", "unknown"),
                    "input_size": len(str(kwargs.get("input_data", "")))
                }
            })
            
            return result
            
        except Exception as e:
            duration = time.time() - start_time
            
            # Log failed inference
            logger.error("Inference failed", extra={
                "custom_dimensions": {
                    "duration_ms": duration * 1000,
                    "success": False,
                    "error_type": type(e).__name__,
                    "error_message": str(e)
                }
            })
            
            raise
    
    return wrapper

# Usage in FastAPI endpoint
from fastapi import FastAPI
import logging

app = FastAPI()
logger = logging.getLogger(__name__)

@app.post("/predict")
@track_inference_metrics
async def predict(input_data: dict):
    # Your inference logic here
    return {"prediction": "result"}

#!/bin/bash
# deploy-workbook.sh

# Configuration
SUBSCRIPTION_ID="your-subscription-id"
RESOURCE_GROUP="retail-dev-rg"
APP_INSIGHTS_NAME="retail-ml-insights"
WORKBOOK_FILE="retail-forecast-monitoring.json"

echo "🚀 Deploying Monitoring Workbook..."

# Login to Azure
az login

# Set subscription
az account set --subscription $SUBSCRIPTION_ID

# Get Application Insights resource ID
APP_INSIGHTS_ID=$(az monitor app-insights component show \
  --app retail-ml-insights \
  --resource-group $RESOURCE_GROUP \
  --query id -o tsv)

# Deploy workbook
az monitor workbook create \
  --resource-group $RESOURCE_GROUP \
  --name "Retail-Forecast-Monitoring" \
  --display-name "Retail Forecast Monitoring" \
  --source-id $APP_INSIGHTS_ID \
  --file $WORKBOOK_FILE

echo "✅ Workbook deployed successfully!"

#!/bin/bash
# validate-kql-scripts.sh

KQL_DIR="azure/monitor/kql"
SCRIPTS=("event_dist.kql" "latency.kql" "errors.kql" "drift.kql" "funnel.kql")

echo "🔍 Validating KQL Scripts..."

for script in "${SCRIPTS[@]}"; do
    if [ -f "$KQL_DIR/$script" ]; then
        echo "✅ $script exists"
        
        # Basic syntax validation
        if grep -q "render" "$KQL_DIR/$script"; then
            echo "✅ $script contains render statement"
        else
            echo "❌ $script missing render statement"
        fi
        
        if grep -q "where timestamp" "$KQL_DIR/$script"; then
            echo "✅ $script contains time filtering"
        else
            echo "❌ $script missing time filtering"
        fi
        
    else
        echo "❌ $script not found"
    fi
done

echo "📊 KQL Script validation completed!"

# setup_telemetry.py
from azure.monitor.opentelemetry import configure_azure_monitor
from opentelemetry import trace
from opentelemetry.exporter.azure_monitor import AzureMonitorTraceExporter
import logging

def configure_monitoring():
    """Configure Azure Monitor telemetry"""
    
    # Configure Azure Monitor
    configure_azure_monitor(
        connection_string="your-connection-string",
        enable_live_metrics=True
    )
    
    # Setup logging
    logging.basicConfig(level=logging.INFO)
    logger = logging.getLogger(__name__)
    
    # Setup tracing
    tracer = trace.get_tracer(__name__)
    
    return logger, tracer

def create_custom_events():
    """Create custom events for monitoring"""
    logger, tracer = configure_monitoring()
    
    # Business events
    with tracer.start_as_current_span("business_event"):
        logger.info("Business event logged", extra={
            "custom_dimensions": {
                "event_type": "view",
                "user_id": "user123",
                "product_id": "prod456"
            }
        })
    
    # Data quality events
    with tracer.start_as_current_span("data_quality"):
        logger.info("Data quality check", extra={
            "custom_dimensions": {
                "quality_score": 0.95,
                "invalid_records": 5,
                "total_records": 1000
            }
        })
    
    print("✅ Telemetry configured successfully!")

if __name__ == "__main__":
    create_custom_events()

# Check Application Insights connection
# Verify custom events are being logged
# Check time range settings
# Validate KQL query syntax

# Optimize KQL queries with proper indexing
# Use appropriate time ranges
# Consider data sampling for large datasets
# Check Application Insights performance

# Verify telemetry configuration
# Check connection string
# Validate custom event logging code
# Monitor Application Insights ingestion