CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-azure-mgmt-cosmosdb

Microsoft Azure Cosmos DB Management Client Library for Python

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview
Eval results
Files

monitoring-metrics.mddocs/

Monitoring and Metrics

Comprehensive monitoring capabilities including percentile metrics, usage statistics, and performance monitoring across regions, databases, containers, and partitions. These operations provide detailed insights into Cosmos DB performance, consumption, and health.

Capabilities

Percentile Metrics

Get percentile-based performance metrics for detailed performance analysis and SLA monitoring.

def list_metrics(
    self,
    resource_group_name: str,
    account_name: str,
    filter: str
) -> ItemPaged[PercentileMetric]:
    """
    Get percentile metrics for a database account.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - account_name: Name of the Cosmos DB account
    - filter: OData filter for metrics (timespan, metric names, aggregation)
    
    Returns:
    ItemPaged[PercentileMetric]: Paginated list of percentile metrics
    """

def list_source_target_metrics(
    self,
    resource_group_name: str,
    account_name: str,
    source_region: str,
    target_region: str,
    filter: str
) -> ItemPaged[PercentileMetric]:
    """
    Get source-target percentile metrics for replication monitoring.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - account_name: Name of the Cosmos DB account
    - source_region: Source region for replication metrics
    - target_region: Target region for replication metrics
    - filter: OData filter for metrics
    
    Returns:
    ItemPaged[PercentileMetric]: Paginated list of replication percentile metrics
    """

def list_target_metrics(
    self,
    resource_group_name: str,
    account_name: str,
    target_region: str,
    filter: str
) -> ItemPaged[PercentileMetric]:
    """
    Get target percentile metrics for a specific region.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - account_name: Name of the Cosmos DB account
    - target_region: Target region for metrics  
    - filter: OData filter for metrics
    
    Returns:
    ItemPaged[PercentileMetric]: Paginated list of target region percentile metrics
    """

Database-Level Monitoring

Monitor database-level metrics, usage, and performance characteristics.

def list_metrics(
    self,
    resource_group_name: str,
    account_name: str,
    database_rid: str,
    filter: str
) -> ItemPaged[Metric]:
    """
    Get metrics for a specific database.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - account_name: Name of the Cosmos DB account
    - database_rid: Resource identifier of the database
    - filter: OData filter for metrics (timespan, metric names)
    
    Returns:
    ItemPaged[Metric]: Paginated list of database metrics
    """

def list_usages(
    self,
    resource_group_name: str,
    account_name: str,
    database_rid: str,
    filter: Optional[str] = None
) -> ItemPaged[Usage]:
    """
    Get usage statistics for a database.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - account_name: Name of the Cosmos DB account
    - database_rid: Resource identifier of the database
    - filter: Optional OData filter for usage data
    
    Returns:
    ItemPaged[Usage]: Paginated list of database usage statistics
    """

def list_metric_definitions(
    self,
    resource_group_name: str,
    account_name: str,
    database_rid: str
) -> ItemPaged[MetricDefinition]:
    """
    Get available metric definitions for a database.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - account_name: Name of the Cosmos DB account
    - database_rid: Resource identifier of the database
    
    Returns:
    ItemPaged[MetricDefinition]: Available metric definitions for the database
    """

Container-Level Monitoring

Monitor container-level metrics, usage, and partition-specific performance data.

def list_metrics(
    self,
    resource_group_name: str,
    account_name: str,
    database_rid: str,
    collection_rid: str,
    filter: str
) -> ItemPaged[Metric]:
    """
    Get metrics for a specific container.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - account_name: Name of the Cosmos DB account
    - database_rid: Resource identifier of the database
    - collection_rid: Resource identifier of the container
    - filter: OData filter for metrics
    
    Returns:
    ItemPaged[Metric]: Paginated list of container metrics
    """

def list_usages(
    self,
    resource_group_name: str,
    account_name: str,
    database_rid: str,
    collection_rid: str,
    filter: Optional[str] = None
) -> ItemPaged[Usage]:
    """
    Get usage statistics for a container.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - account_name: Name of the Cosmos DB account
    - database_rid: Resource identifier of the database
    - collection_rid: Resource identifier of the container
    - filter: Optional OData filter for usage data
    
    Returns:
    ItemPaged[Usage]: Paginated list of container usage statistics
    """

def list_partition_metrics(
    self,
    resource_group_name: str,
    account_name: str,
    database_rid: str,
    collection_rid: str,
    filter: str
) -> ItemPaged[PartitionMetric]:
    """
    Get partition-level metrics for a container.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - account_name: Name of the Cosmos DB account
    - database_rid: Resource identifier of the database
    - collection_rid: Resource identifier of the container
    - filter: OData filter for metrics
    
    Returns:
    ItemPaged[PartitionMetric]: Paginated list of partition-level metrics
    """

Regional Monitoring

Monitor region-specific metrics and cross-region performance data.

def list_region_metrics(
    self,
    resource_group_name: str,
    account_name: str,
    region: str,
    database_rid: str,
    collection_rid: str,
    filter: str
) -> ItemPaged[Metric]:
    """
    Get region-specific metrics for a container.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - account_name: Name of the Cosmos DB account
    - region: Azure region name
    - database_rid: Resource identifier of the database
    - collection_rid: Resource identifier of the container
    - filter: OData filter for metrics
    
    Returns:
    ItemPaged[Metric]: Paginated list of region-specific container metrics
    """

def list_partition_region_metrics(
    self,
    resource_group_name: str,
    account_name: str,
    region: str,
    database_rid: str,
    collection_rid: str,
    filter: str
) -> ItemPaged[PartitionMetric]:
    """
    Get region-specific partition metrics for a container.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - account_name: Name of the Cosmos DB account
    - region: Azure region name
    - database_rid: Resource identifier of the database
    - collection_rid: Resource identifier of the container
    - filter: OData filter for metrics
    
    Returns:
    ItemPaged[PartitionMetric]: Paginated list of region-specific partition metrics
    """

def list_database_account_region_metrics(
    self,
    resource_group_name: str,
    account_name: str,
    region: str,
    filter: str
) -> ItemPaged[Metric]:
    """
    Get region-specific metrics for a database account.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - account_name: Name of the Cosmos DB account
    - region: Azure region name
    - filter: OData filter for metrics
    
    Returns:
    ItemPaged[Metric]: Paginated list of region-specific account metrics
    """

Partition Key Range Monitoring

Monitor partition key range performance and distribution metrics.

def list_metrics(
    self,
    resource_group_name: str,
    account_name: str,
    database_rid: str,
    collection_rid: str,
    partition_key_range_id: str,
    filter: str
) -> ItemPaged[PartitionMetric]:
    """
    Get metrics for a specific partition key range.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - account_name: Name of the Cosmos DB account
    - database_rid: Resource identifier of the database
    - collection_rid: Resource identifier of the container
    - partition_key_range_id: Identifier of the partition key range
    - filter: OData filter for metrics
    
    Returns:
    ItemPaged[PartitionMetric]: Paginated list of partition key range metrics
    """

def list_region_metrics(
    self,
    resource_group_name: str,
    account_name: str,
    region: str,
    database_rid: str,
    collection_rid: str,
    partition_key_range_id: str,
    filter: str
) -> ItemPaged[PartitionMetric]:
    """
    Get region-specific metrics for a partition key range.
    
    Parameters:
    - resource_group_name: Name of the resource group
    - account_name: Name of the Cosmos DB account
    - region: Azure region name
    - database_rid: Resource identifier of the database
    - collection_rid: Resource identifier of the container
    - partition_key_range_id: Identifier of the partition key range
    - filter: OData filter for metrics
    
    Returns:
    ItemPaged[PartitionMetric]: Paginated list of region-specific partition key range metrics
    """

Usage Examples

Getting Account-Level Percentile Metrics

from azure.mgmt.cosmosdb import CosmosDBManagementClient
from azure.identity import DefaultAzureCredential
from datetime import datetime, timedelta

client = CosmosDBManagementClient(DefaultAzureCredential(), "subscription-id")

# Get percentile metrics for the last 24 hours
end_time = datetime.utcnow()
start_time = end_time - timedelta(hours=24)

# Format time filter for OData
time_filter = f"(name.value eq 'ServerSideLatency') and timeGrain eq duration'PT1M' and startTime eq {start_time.isoformat()}Z and endTime eq {end_time.isoformat()}Z"

percentile_metrics = client.percentile.list_metrics(
    "my-resource-group",
    "my-cosmos-account", 
    time_filter
)

for metric in percentile_metrics:
    print(f"Metric: {metric.name.value}")
    print(f"Time Grain: {metric.time_grain}")
    for data_point in metric.data:
        print(f"  Time: {data_point.time_stamp}")
        print(f"  P50: {data_point.p50} ms")
        print(f"  P95: {data_point.p95} ms")
        print(f"  P99: {data_point.p99} ms")

Monitoring Database Usage and Metrics

# Get database resource ID (typically in format "dbs/{database-name}")
database_rid = "dbs/products"

# Get database metrics for request units consumed
ru_filter = f"(name.value eq 'TotalRequestUnits') and timeGrain eq duration'PT1H' and startTime eq {start_time.isoformat()}Z and endTime eq {end_time.isoformat()}Z"

database_metrics = client.database.list_metrics(
    "my-resource-group",
    "my-cosmos-account",
    database_rid,
    ru_filter
)

for metric in database_metrics:
    print(f"Database Metric: {metric.name.value}")
    for data_point in metric.data:
        print(f"  Time: {data_point.time_stamp}")
        print(f"  Total RU: {data_point.total}")
        print(f"  Average RU/s: {data_point.average}")

# Get database usage statistics
usage_stats = client.database.list_usages(
    "my-resource-group", 
    "my-cosmos-account",
    database_rid
)

for usage in usage_stats:
    print(f"Usage Metric: {usage.name.value}")
    print(f"Current Value: {usage.current_value}")
    print(f"Limit: {usage.limit}")
    print(f"Unit: {usage.unit}")

Container Partition Monitoring

# Monitor partition-level metrics for hot partition detection
container_rid = "dbs/products/colls/items"

partition_filter = f"(name.value eq 'NormalizedRUConsumption') and timeGrain eq duration'PT1M' and startTime eq {start_time.isoformat()}Z and endTime eq {end_time.isoformat()}Z"

partition_metrics = client.collection_partition.list_metrics(
    "my-resource-group",
    "my-cosmos-account", 
    database_rid,
    container_rid,
    partition_filter
)

for metric in partition_metrics:
    print(f"Partition Metric: {metric.name.value}")
    for data_point in metric.data:
        print(f"  Partition ID: {data_point.partition_id}")
        print(f"  Time: {data_point.time_stamp}")
        print(f"  Normalized RU%: {data_point.maximum}%")

Cross-Region Replication Monitoring

# Monitor replication latency between regions
source_region = "East US"
target_region = "West Europe"

replication_filter = f"(name.value eq 'ReplicationLatency') and timeGrain eq duration'PT5M' and startTime eq {start_time.isoformat()}Z and endTime eq {end_time.isoformat()}Z"

replication_metrics = client.percentile_source_target.list_metrics(
    "my-resource-group",
    "my-cosmos-account",
    source_region,
    target_region, 
    replication_filter
)

for metric in replication_metrics:
    print(f"Replication Metric: {metric.name.value}")
    print(f"Source: {source_region} -> Target: {target_region}")
    for data_point in metric.data:
        print(f"  Time: {data_point.time_stamp}")
        print(f"  P50 Latency: {data_point.p50} ms")
        print(f"  P99 Latency: {data_point.p99} ms")

Available Metrics Discovery

# Discover what metrics are available for a container
metric_definitions = client.collection.list_metric_definitions(
    "my-resource-group",
    "my-cosmos-account",
    database_rid,
    container_rid
)

print("Available Container Metrics:")
for definition in metric_definitions:
    print(f"  Name: {definition.name.value}")
    print(f"  Display Name: {definition.name.localized_value}")
    print(f"  Unit: {definition.unit}")
    print(f"  Primary Aggregation: {definition.primary_aggregation_type}")
    print(f"  Supported Aggregations: {definition.supported_aggregation_types}")
    print("  ---")

Key Types

class PercentileMetric:
    """Percentile-based performance metric data."""
    name: MetricName  # Metric name and display name
    unit: str  # Metric unit (e.g. "Milliseconds", "Count")
    time_grain: str  # Time grain for aggregation (e.g. "PT1M", "PT1H")
    start_time: str  # Metric period start time
    end_time: str  # Metric period end time
    data: List[PercentileMetricValue]  # Time series data points

class PercentileMetricValue:
    """Individual percentile metric data point."""
    time_stamp: str  # Timestamp for this data point
    p10: float  # 10th percentile value
    p25: float  # 25th percentile value  
    p50: float  # 50th percentile value (median)
    p75: float  # 75th percentile value
    p90: float  # 90th percentile value
    p95: float  # 95th percentile value
    p99: float  # 99th percentile value

class Metric:
    """Standard metric data."""
    name: MetricName  # Metric name and display name
    unit: str  # Metric unit
    time_grain: str  # Time grain for aggregation
    start_time: str  # Metric period start time
    end_time: str  # Metric period end time
    data: List[MetricValue]  # Time series data points

class MetricValue:
    """Individual metric data point."""
    time_stamp: str  # Timestamp for this data point
    total: float  # Total/sum value
    count: float  # Count of samples
    average: float  # Average value
    minimum: float  # Minimum value
    maximum: float  # Maximum value

class PartitionMetric:
    """Partition-specific metric data."""
    name: MetricName  # Metric name and display name
    unit: str  # Metric unit
    time_grain: str  # Time grain for aggregation
    start_time: str  # Metric period start time
    end_time: str  # Metric period end time
    data: List[PartitionMetricValue]  # Time series data points

class PartitionMetricValue:
    """Individual partition metric data point."""
    time_stamp: str  # Timestamp for this data point
    partition_id: str  # Partition identifier
    partition_key_range_id: str  # Partition key range identifier
    total: float  # Total/sum value
    count: float  # Count of samples  
    average: float  # Average value
    minimum: float  # Minimum value
    maximum: float  # Maximum value

class Usage:
    """Resource usage statistics."""
    name: MetricName  # Usage metric name
    unit: str  # Usage unit (e.g. "Bytes", "Count")
    current_value: int  # Current usage value
    limit: int  # Maximum allowed value (-1 for unlimited)
    quota_period: str  # Period for quota enforcement

class MetricDefinition:
    """Definition of an available metric."""
    name: MetricName  # Metric name and display name
    unit: str  # Metric unit
    primary_aggregation_type: str  # Primary aggregation method
    supported_aggregation_types: List[str]  # All supported aggregation methods
    metric_availabilities: List[MetricAvailability]  # Available time grains and retention
    fill_gap_with_zero: bool  # Whether to fill gaps with zero values

class MetricName:
    """Metric name information."""
    value: str  # Programmatic metric name
    localized_value: str  # Human-readable display name

class MetricAvailability:
    """Metric availability and retention information."""
    time_grain: str  # Time grain (e.g. "PT1M", "PT1H", "P1D")
    retention: str  # Data retention period (e.g. "P30D", "P90D")

Install with Tessl CLI

npx tessl i tessl/pypi-azure-mgmt-cosmosdb

docs

database-accounts.md

index.md

mongodb-resources.md

monitoring-metrics.md

multi-api-resources.md

networking-security.md

restore-operations.md

service-management.md

sql-resources.md

tile.json