Azure Cosmos DB RU Optimization¶

from azure.cosmos import CosmosClient
from azure.monitor.query import LogsQueryClient
from datetime import timedelta

def monitor_ru_consumption(workspace_id, resource_id):
    """Monitor RU consumption patterns."""

    query = f"""
    AzureDiagnostics
    | where ResourceId == "{resource_id}"
    | where TimeGenerated > ago(24h)
    | summarize
        TotalRU = sum(todouble(requestCharge_s)),
        AvgRU = avg(todouble(requestCharge_s)),
        MaxRU = max(todouble(requestCharge_s)),
        Count = count()
        by operationType_s, bin(TimeGenerated, 1h)
    | order by TotalRU desc
    """

    # Execute and analyze results
    # Implementation details...

    return query

def optimize_document_size(document):
    """Reduce document size to lower RU consumption."""

    optimized = {}

    # Use shorter property names
    property_mapping = {
        'customerIdentifier': 'cId',
        'orderTimestamp': 'ts',
        'totalAmount': 'amt',
        'shippingAddress': 'addr'
    }

    for original, short in property_mapping.items():
        if original in document:
            optimized[short] = document[original]

    # Remove null/empty values
    optimized = {k: v for k, v in optimized.items() if v is not None and v != ''}

    # Store large properties separately (blob storage)
    if 'largeDescription' in document:
        # Upload to blob, store reference
        optimized['descRef'] = upload_to_blob(document['largeDescription'])

    return optimized

from azure.cosmos import CosmosClient, PartitionKey

def bulk_upsert_items(container, items, batch_size=100):
    """Efficiently upsert items in bulk."""

    from concurrent.futures import ThreadPoolExecutor, as_completed

    def upsert_batch(batch):
        """Upsert a batch of items."""
        operations = []
        for item in batch:
            operations.append(('upsert', (item,), {}))

        # Execute batch
        results = container.execute_item_batch(operations, item['partitionKey'])
        return len(results)

    # Split into batches
    batches = [items[i:i + batch_size] for i in range(0, len(items), batch_size)]

    total_upserted = 0
    with ThreadPoolExecutor(max_workers=5) as executor:
        futures = [executor.submit(upsert_batch, batch) for batch in batches]

        for future in as_completed(futures):
            total_upserted += future.result()

    print(f"✅ Bulk upserted {total_upserted} items")
    return total_upserted

from functools import lru_cache
from datetime import datetime, timedelta

class CosmosCache:
    """Cache frequently accessed documents."""

    def __init__(self, container):
        self.container = container
        self.cache = {}
        self.cache_ttl = {}

    def get_item_cached(self, item_id, partition_key, ttl_minutes=30):
        """Get item with caching."""

        cache_key = f"{partition_key}/{item_id}"

        # Check cache
        if cache_key in self.cache:
            if datetime.now() < self.cache_ttl.get(cache_key, datetime.min):
                print(f"✅ Cache hit for {cache_key}")
                return self.cache[cache_key]

        # Cache miss - fetch from Cosmos
        print(f"⚠️ Cache miss for {cache_key} - fetching from Cosmos DB")
        item = self.container.read_item(
            item=item_id,
            partition_key=partition_key
        )

        # Update cache
        self.cache[cache_key] = item
        self.cache_ttl[cache_key] = datetime.now() + timedelta(minutes=ttl_minutes)

        return item

def calculate_autoscale_savings(container_name, metrics):
    """Calculate potential savings with autoscale."""

    peak_ru = max(metrics['ru_per_hour'])
    average_ru = sum(metrics['ru_per_hour']) / len(metrics['ru_per_hour'])

    # Provisioned cost (always at peak)
    provisioned_monthly_cost = (peak_ru / 100) * 0.008 * 730  # $0.008 per 100 RU/s per hour

    # Autoscale cost (average consumption + 10% overhead)
    autoscale_monthly_cost = (average_ru / 100) * 0.012 * 730  # $0.012 per 100 RU/s per hour

    savings = provisioned_monthly_cost - autoscale_monthly_cost
    savings_percent = (savings / provisioned_monthly_cost) * 100

    print(f"💰 Cost Analysis for {container_name}")
    print(f"   Peak RU/s: {peak_ru:,.0f}")
    print(f"   Average RU/s: {average_ru:,.0f}")
    print(f"   Provisioned Monthly Cost: ${provisioned_monthly_cost:,.2f}")
    print(f"   Autoscale Monthly Cost: ${autoscale_monthly_cost:,.2f}")
    print(f"   Potential Savings: ${savings:,.2f} ({savings_percent:.1f}%)")

    if savings > 0:
        print(f"   ✅ Recommendation: Switch to autoscale")
    else:
        print(f"   ℹ️ Recommendation: Keep provisioned throughput")

    return savings