CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-google-cloud-monitoring

Google Cloud Monitoring API client library for collecting, analyzing, and alerting on metrics, events, and metadata from cloud and on-premise sources.

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview
Eval results
Files

metrics.mddocs/

Metrics and Time Series

Core functionality for working with metrics, time series data, and monitored resources in Google Cloud Monitoring. This service handles metric descriptor management, time series data ingestion and querying, and monitored resource descriptor operations.

Capabilities

Metric Descriptor Management

Manage metric descriptors that define the schema and metadata for custom metrics.

class MetricServiceClient:
    def list_metric_descriptors(
        self,
        request=None,
        *,
        name: str = None,
        retry=None,
        timeout=None,
        metadata=()
    ) -> pagers.ListMetricDescriptorsPager:
        """
        Lists metric descriptors that match a filter.

        Args:
            request: The request object or dict equivalent
            name: Required. Project name in format 'projects/[PROJECT_ID]'
            retry: Retry configuration
            timeout: Request timeout in seconds
            metadata: Additional metadata

        Returns:
            Pager for iterating over metric descriptors
        """

    def get_metric_descriptor(
        self,
        request=None,
        *,
        name: str = None,
        retry=None,
        timeout=None,
        metadata=()
    ) -> metric_pb2.MetricDescriptor:
        """
        Gets a single metric descriptor.

        Args:
            request: The request object or dict equivalent
            name: Required. Metric descriptor name
            retry: Retry configuration
            timeout: Request timeout in seconds
            metadata: Additional metadata

        Returns:
            MetricDescriptor object
        """

    def create_metric_descriptor(
        self,
        request=None,
        *,
        name: str = None,
        metric_descriptor=None,
        retry=None,
        timeout=None,
        metadata=()
    ) -> metric_pb2.MetricDescriptor:
        """
        Creates a user-defined metric descriptor.

        Args:
            request: The request object or dict equivalent
            name: Required. Project name
            metric_descriptor: Required. The metric descriptor to create
            retry: Retry configuration
            timeout: Request timeout in seconds
            metadata: Additional metadata

        Returns:
            Created MetricDescriptor object
        """

    def delete_metric_descriptor(
        self,
        request=None,
        *,
        name: str = None,
        retry=None,
        timeout=None,
        metadata=()
    ) -> None:
        """
        Deletes a metric descriptor.

        Args:
            request: The request object or dict equivalent
            name: Required. Metric descriptor name to delete
            retry: Retry configuration
            timeout: Request timeout in seconds
            metadata: Additional metadata
        """

Time Series Operations

Manage time series data including reading historical data and writing new data points.

class MetricServiceClient:
    def list_time_series(
        self,
        request=None,
        *,
        name: str = None,
        filter: str = None,
        interval=None,
        view=None,
        retry=None,
        timeout=None,
        metadata=()
    ) -> pagers.ListTimeSeriesPager:
        """
        Lists time series that match a filter.

        Args:
            request: The request object or dict equivalent
            name: Required. Project name
            filter: Required. Monitoring filter string
            interval: Required. Time interval for data
            view: Specifies which information is returned
            retry: Retry configuration
            timeout: Request timeout in seconds
            metadata: Additional metadata

        Returns:
            Pager for iterating over TimeSeries objects
        """

    def create_time_series(
        self,
        request=None,
        *,
        name: str = None,
        time_series=None,
        retry=None,
        timeout=None,
        metadata=()
    ) -> None:
        """
        Creates or adds data to one or more time series.

        Args:
            request: The request object or dict equivalent
            name: Required. Project name
            time_series: Required. List of TimeSeries objects
            retry: Retry configuration
            timeout: Request timeout in seconds
            metadata: Additional metadata
        """

    def create_service_time_series(
        self,
        request=None,
        *,
        name: str = None,
        time_series=None,
        retry=None,
        timeout=None,
        metadata=()
    ) -> None:
        """
        Creates or adds data to service time series.

        Args:
            request: The request object or dict equivalent
            name: Required. Project name
            time_series: Required. List of TimeSeries objects
            retry: Retry configuration
            timeout: Request timeout in seconds
            metadata: Additional metadata
        """

Monitored Resource Descriptors

Manage monitored resource descriptors that define resource types and their label schemas.

class MetricServiceClient:
    def list_monitored_resource_descriptors(
        self,
        request=None,
        *,
        name: str = None,
        retry=None,
        timeout=None,
        metadata=()
    ) -> pagers.ListMonitoredResourceDescriptorsPager:
        """
        Lists monitored resource descriptors.

        Args:
            request: The request object or dict equivalent
            name: Required. Project name
            retry: Retry configuration
            timeout: Request timeout in seconds
            metadata: Additional metadata

        Returns:
            Pager for iterating over MonitoredResourceDescriptor objects
        """

    def get_monitored_resource_descriptor(
        self,
        request=None,
        *,
        name: str = None,
        retry=None,
        timeout=None,
        metadata=()
    ) -> monitored_resource_pb2.MonitoredResourceDescriptor:
        """
        Gets a single monitored resource descriptor.

        Args:
            request: The request object or dict equivalent
            name: Required. Monitored resource descriptor name
            retry: Retry configuration
            timeout: Request timeout in seconds
            metadata: Additional metadata

        Returns:
            MonitoredResourceDescriptor object
        """

Data Types

TimeSeries

Represents a collection of data points associated with a metric.

class TimeSeries:
    metric: Metric  # Associated metric
    resource: MonitoredResource  # Associated resource
    metadata: MonitoredResourceMetadata  # Resource metadata
    metric_kind: MetricDescriptor.MetricKind  # Metric kind
    value_type: MetricDescriptor.ValueType  # Value type
    points: List[Point]  # Data points
    unit: str  # Unit of measurement

class Point:
    interval: TimeInterval  # Time interval for the point
    value: TypedValue  # The value of the point

class Metric:
    type: str  # Metric type name
    labels: Dict[str, str]  # Metric labels

class MonitoredResource:
    type: str  # Resource type
    labels: Dict[str, str]  # Resource labels

Query Types

class ListTimeSeriesRequest:
    name: str  # Required. Project name
    filter: str  # Required. Monitoring filter
    interval: TimeInterval  # Required. Time interval
    aggregation: Aggregation  # Aggregation specification
    order_by: str  # Ordering specification
    view: ListTimeSeriesRequest.TimeSeriesView  # View type
    page_size: int  # Maximum results per page
    page_token: str  # Page token

class ListTimeSeriesResponse:
    time_series: List[TimeSeries]  # Time series data
    next_page_token: str  # Next page token
    execution_errors: List[Status]  # Execution errors

class CreateTimeSeriesRequest:
    name: str  # Required. Project name
    time_series: List[TimeSeries]  # Required. Time series data

class CreateTimeSeriesSummary:
    total_point_count: int  # Total points processed
    success_point_count: int  # Successfully written points
    errors: List[CreateTimeSeriesError]  # Errors encountered

class CreateTimeSeriesError:
    time_series: TimeSeries  # Time series with error
    status: Status  # Error status

Metric Descriptor Types

class ListMetricDescriptorsRequest:
    name: str  # Required. Project name
    filter: str  # Filter expression
    page_size: int  # Maximum results per page
    page_token: str  # Page token

class ListMetricDescriptorsResponse:
    metric_descriptors: List[MetricDescriptor]  # Metric descriptors
    next_page_token: str  # Next page token

class CreateMetricDescriptorRequest:
    name: str  # Required. Project name
    metric_descriptor: MetricDescriptor  # Required. Metric descriptor

class GetMetricDescriptorRequest:
    name: str  # Required. Metric descriptor name

class DeleteMetricDescriptorRequest:
    name: str  # Required. Metric descriptor name to delete

class GetMonitoredResourceDescriptorRequest:
    name: str  # Required. Monitored resource descriptor name

class ListMonitoredResourceDescriptorsRequest:
    name: str  # Required. Project name
    filter: str  # Filter expression  
    page_size: int  # Maximum results per page
    page_token: str  # Page token

class ListMonitoredResourceDescriptorsResponse:
    resource_descriptors: List[MonitoredResourceDescriptor]  # Resource descriptors
    next_page_token: str  # Next page token

Usage Examples

Creating a Custom Metric Descriptor

from google.cloud.monitoring import MetricServiceClient
from google.api import metric_pb2
from google.api import label_pb2

client = MetricServiceClient()
project_name = f"projects/{project_id}"

# Create metric descriptor
descriptor = metric_pb2.MetricDescriptor()
descriptor.type = "custom.googleapis.com/my_metric"
descriptor.metric_kind = metric_pb2.MetricDescriptor.MetricKind.GAUGE
descriptor.value_type = metric_pb2.MetricDescriptor.ValueType.DOUBLE
descriptor.description = "My custom metric"
descriptor.display_name = "My Metric"

# Add labels
label = label_pb2.LabelDescriptor()
label.key = "environment"
label.value_type = label_pb2.LabelDescriptor.ValueType.STRING
label.description = "Environment label"
descriptor.labels.append(label)

# Create the descriptor
created_descriptor = client.create_metric_descriptor(
    name=project_name,
    metric_descriptor=descriptor
)
print(f"Created metric descriptor: {created_descriptor.type}")

Writing Time Series Data

from google.cloud.monitoring import MetricServiceClient
from google.cloud.monitoring_v3.types import TimeSeries, Point, TypedValue
from google.cloud.monitoring_v3.types import TimeInterval, Metric, MonitoredResource
from google.protobuf.timestamp_pb2 import Timestamp
import time

client = MetricServiceClient()
project_name = f"projects/{project_id}"

# Create time series
series = TimeSeries()

# Set metric
series.metric.type = "custom.googleapis.com/my_metric"
series.metric.labels["environment"] = "production"

# Set resource  
series.resource.type = "gce_instance"
series.resource.labels["instance_id"] = "my-instance-123"
series.resource.labels["zone"] = "us-central1-a"

# Create data point
now = time.time()
seconds = int(now)
nanos = int((now - seconds) * 10**9)

point = Point()
point.value.double_value = 42.0
point.interval.end_time.seconds = seconds
point.interval.end_time.nanos = nanos

series.points = [point]

# Write the time series
client.create_time_series(
    name=project_name,
    time_series=[series]
)
print("Time series data written successfully")

Querying Time Series Data

from google.cloud.monitoring import MetricServiceClient
from google.cloud.monitoring_v3.types import TimeInterval, Aggregation
from google.protobuf.timestamp_pb2 import Timestamp
import time

client = MetricServiceClient()
project_name = f"projects/{project_id}"

# Define time interval (last hour)
now = time.time()
interval = TimeInterval()
interval.end_time.seconds = int(now)
interval.start_time.seconds = int(now - 3600)  # 1 hour ago

# Query time series
filter_str = 'metric.type="compute.googleapis.com/instance/cpu/utilization"'

for series in client.list_time_series(
    name=project_name,
    filter=filter_str,
    interval=interval
):
    print(f"Metric: {series.metric.type}")
    print(f"Resource: {series.resource.type}")
    for point in series.points:
        print(f"Value: {point.value.double_value} at {point.interval.end_time}")

Aggregating Time Series Data

from google.cloud.monitoring_v3.types import Aggregation, Duration

# Create aggregation specification
aggregation = Aggregation()
aggregation.alignment_period.seconds = 300  # 5 minutes
aggregation.per_series_aligner = Aggregation.Aligner.ALIGN_MEAN
aggregation.cross_series_reducer = Aggregation.Reducer.REDUCE_MEAN
aggregation.group_by_fields = ["resource.label.zone"]

# Query with aggregation
for series in client.list_time_series(
    name=project_name,
    filter=filter_str,
    interval=interval,
    aggregation=aggregation
):
    print(f"Aggregated series: {series.metric.type}")
    for point in series.points:
        print(f"Aggregated value: {point.value.double_value}")

Listing Metric Descriptors

client = MetricServiceClient()
project_name = f"projects/{project_id}"

# List all metric descriptors
for descriptor in client.list_metric_descriptors(name=project_name):
    print(f"Metric: {descriptor.type}")
    print(f"Kind: {descriptor.metric_kind}")
    print(f"Value Type: {descriptor.value_type}")

# Filter custom metrics
filter_expr = 'metric.type=starts_with("custom.googleapis.com/")'
request = {"name": project_name, "filter": filter_expr}

for descriptor in client.list_metric_descriptors(request=request):
    print(f"Custom metric: {descriptor.type}")

Async Operations

import asyncio
from google.cloud.monitoring import MetricServiceAsyncClient

async def manage_metrics():
    client = MetricServiceAsyncClient()
    project_name = f"projects/{project_id}"
    
    # List metric descriptors asynchronously
    async for descriptor in await client.list_metric_descriptors(name=project_name):
        print(f"Async metric: {descriptor.type}")
    
    # Create time series asynchronously
    await client.create_time_series(
        name=project_name,
        time_series=[series]  # series defined earlier
    )
    print("Async time series creation completed")

asyncio.run(manage_metrics())

Resource Path Helpers

class MetricServiceClient:
    @staticmethod
    def metric_descriptor_path(project: str, metric_descriptor: str) -> str:
        """Returns a fully-qualified metric_descriptor string."""
        
    @staticmethod
    def monitored_resource_descriptor_path(
        project: str, 
        monitored_resource_descriptor: str
    ) -> str:
        """Returns a fully-qualified monitored_resource_descriptor string."""
        
    @staticmethod
    def parse_metric_descriptor_path(path: str) -> Dict[str, str]:
        """Parses a metric_descriptor path into its component segments."""

Error Handling

Metric operations can encounter various error conditions:

from google.api_core import exceptions
from google.cloud.monitoring import MetricServiceClient

client = MetricServiceClient()

try:
    # Attempt to create time series
    client.create_time_series(name=project_name, time_series=invalid_series)
except exceptions.InvalidArgument as e:
    print(f"Invalid time series data: {e}")
except exceptions.ResourceExhausted:
    print("Rate limit exceeded")
except exceptions.NotFound:
    print("Metric descriptor not found")
except exceptions.PermissionDenied:
    print("Insufficient permissions to write metrics")

Install with Tessl CLI

npx tessl i tessl/pypi-google-cloud-monitoring

docs

alert-policies.md

groups.md

index.md

metrics.md

notifications.md

services.md

snooze.md

uptime-checks.md

tile.json