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

alert-policies.mddocs/

Alert Policy Management

Comprehensive alert policy management for Google Cloud Monitoring, enabling creation, modification, and monitoring of alerting conditions based on metrics, logs, and resource states. Alert policies define when and how notifications are sent when specified conditions are met.

Capabilities

Alert Policy Operations

Manage the complete lifecycle of alert policies including creation, updates, retrieval, and deletion.

class AlertPolicyServiceClient:
    def list_alert_policies(
        self,
        request=None,
        *,
        name: str = None,
        retry=None,
        timeout=None,
        metadata=()
    ) -> pagers.ListAlertPoliciesPager:
        """
        Lists the existing alerting policies for the workspace.

        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 AlertPolicy objects
        """

    def get_alert_policy(
        self,
        request=None,
        *,
        name: str = None,
        retry=None,
        timeout=None,
        metadata=()
    ) -> alert.AlertPolicy:
        """
        Gets a single alerting policy.

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

        Returns:
            AlertPolicy object
        """

    def create_alert_policy(
        self,
        request=None,
        *,
        name: str = None,
        alert_policy: alert.AlertPolicy = None,
        retry=None,
        timeout=None,
        metadata=()
    ) -> alert.AlertPolicy:
        """
        Creates a new alerting policy.

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

        Returns:
            Created AlertPolicy object
        """

    def update_alert_policy(
        self,
        request=None,
        *,
        update_mask=None,
        alert_policy: alert.AlertPolicy = None,
        retry=None,
        timeout=None,
        metadata=()
    ) -> alert.AlertPolicy:
        """
        Updates an alerting policy.

        Args:
            request: The request object or dict equivalent
            update_mask: Field mask for selective updates
            alert_policy: Required. Updated alert policy
            retry: Retry configuration
            timeout: Request timeout in seconds
            metadata: Additional metadata

        Returns:
            Updated AlertPolicy object
        """

    def delete_alert_policy(
        self,
        request=None,
        *,
        name: str = None,
        retry=None,
        timeout=None,
        metadata=()
    ) -> None:
        """
        Deletes an alerting policy.

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

Async Alert Policy Operations

Asynchronous versions of all alert policy operations for use with async/await patterns.

class AlertPolicyServiceAsyncClient:
    async def list_alert_policies(
        self,
        request=None,
        *,
        name: str = None,
        retry=None,
        timeout=None,
        metadata=()
    ) -> pagers.ListAlertPoliciesAsyncPager:
        """Async version of list_alert_policies."""

    async def get_alert_policy(
        self,
        request=None,
        *,
        name: str = None,
        retry=None,
        timeout=None,
        metadata=()
    ) -> alert.AlertPolicy:
        """Async version of get_alert_policy."""

    async def create_alert_policy(
        self,
        request=None,
        *,
        name: str = None,
        alert_policy: alert.AlertPolicy = None,
        retry=None,
        timeout=None,
        metadata=()
    ) -> alert.AlertPolicy:
        """Async version of create_alert_policy."""

    async def update_alert_policy(
        self,
        request=None,
        *,
        update_mask=None,
        alert_policy: alert.AlertPolicy = None,
        retry=None,
        timeout=None,
        metadata=()
    ) -> alert.AlertPolicy:
        """Async version of update_alert_policy."""

    async def delete_alert_policy(
        self,
        request=None,
        *,
        name: str = None,
        retry=None,
        timeout=None,
        metadata=()
    ) -> None:
        """Async version of delete_alert_policy."""

Data Types

AlertPolicy

Main alert policy configuration containing conditions, notification channels, and policy metadata.

class AlertPolicy:
    name: str  # Resource name
    display_name: str  # Human-readable name
    documentation: AlertPolicy.Documentation  # Alert documentation
    user_labels: Dict[str, str]  # User-defined labels
    conditions: List[AlertPolicy.Condition]  # Alert conditions
    combiner: ConditionCombinerType  # How to combine conditions
    enabled: bool  # Whether policy is enabled
    validity: Status  # Validation status
    notification_channels: List[str]  # Notification channel names
    creation_record: MutationRecord  # Creation metadata
    mutation_record: MutationRecord  # Last modification metadata
    alert_strategy: AlertPolicy.AlertStrategy  # Alert strategy configuration

class AlertPolicy.Condition:
    name: str  # Condition name
    display_name: str  # Human-readable name
    condition_threshold: MetricThreshold  # Threshold condition
    condition_absent: MetricAbsence  # Absence condition
    condition_matched_log: LogMatch  # Log match condition
    condition_monitoring_query_language: MonitoringQueryLanguageCondition  # MQL condition

class AlertPolicy.Documentation:
    content: str  # Documentation content
    mime_type: str  # Content MIME type

class AlertPolicy.AlertStrategy:
    auto_close: Duration  # Auto-close duration
    notification_rate_limit: AlertPolicy.AlertStrategy.NotificationRateLimit  # Rate limiting

Request and Response Types

class CreateAlertPolicyRequest:
    name: str  # Required. Project name
    alert_policy: AlertPolicy  # Required. Alert policy to create

class DeleteAlertPolicyRequest:
    name: str  # Required. Alert policy name to delete

class GetAlertPolicyRequest:
    name: str  # Required. Alert policy name to retrieve

class ListAlertPoliciesRequest:
    name: str  # Required. Project name
    filter: str  # Filter expression
    order_by: str  # Ordering specification
    page_size: int  # Maximum results per page
    page_token: str  # Page token for pagination

class ListAlertPoliciesResponse:
    alert_policies: List[AlertPolicy]  # Alert policies
    next_page_token: str  # Token for next page

class UpdateAlertPolicyRequest:
    update_mask: FieldMask  # Fields to update
    alert_policy: AlertPolicy  # Required. Updated alert policy

Usage Examples

Creating a Basic Threshold Alert Policy

from google.cloud.monitoring import AlertPolicyServiceClient, AlertPolicy
from google.cloud.monitoring_v3.types import ComparisonType, Aggregation

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

# Create alert policy
policy = AlertPolicy()
policy.display_name = "High CPU Usage"
policy.enabled = True

# Create condition
condition = AlertPolicy.Condition()
condition.display_name = "CPU usage exceeds 80%"

# Configure threshold condition
threshold = condition.condition_threshold
threshold.filter = 'resource.type="gce_instance"'
threshold.comparison = ComparisonType.COMPARISON_GREATER
threshold.threshold_value.double_value = 0.8

# Configure aggregation
aggregation = Aggregation()
aggregation.alignment_period.seconds = 300  # 5 minutes
aggregation.per_series_aligner = Aggregation.Aligner.ALIGN_MEAN
threshold.aggregations = [aggregation]

policy.conditions = [condition]
policy.combiner = AlertPolicy.ConditionCombinerType.AND

# Add notification channels
policy.notification_channels = [
    f"projects/{project_id}/notificationChannels/{channel_id}"
]

# Create the policy
created_policy = client.create_alert_policy(
    name=project_name,
    alert_policy=policy
)
print(f"Created alert policy: {created_policy.name}")

Listing and Filtering Alert Policies

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

# List all alert policies
for policy in client.list_alert_policies(name=project_name):
    print(f"Policy: {policy.display_name}, Enabled: {policy.enabled}")

# List with filtering
filter_expr = 'display_name:"High CPU"'
for policy in client.list_alert_policies(name=project_name, filter=filter_expr):
    print(f"Filtered policy: {policy.display_name}")

Updating an Alert Policy

from google.protobuf import field_mask_pb2

client = AlertPolicyServiceClient()

# Get existing policy
policy_name = f"projects/{project_id}/alertPolicies/{policy_id}"
policy = client.get_alert_policy(name=policy_name)

# Modify policy
policy.display_name = "Updated High CPU Alert"
policy.enabled = False

# Update with field mask
update_mask = field_mask_pb2.FieldMask()
update_mask.paths.extend(["display_name", "enabled"])

updated_policy = client.update_alert_policy(
    alert_policy=policy,
    update_mask=update_mask
)
print(f"Updated policy: {updated_policy.display_name}")

Async Alert Policy Management

import asyncio
from google.cloud.monitoring import AlertPolicyServiceAsyncClient

async def manage_alert_policies():
    client = AlertPolicyServiceAsyncClient()
    project_name = f"projects/{project_id}"
    
    # List policies asynchronously
    async for policy in await client.list_alert_policies(name=project_name):
        print(f"Async policy: {policy.display_name}")
    
    # Get policy asynchronously  
    policy = await client.get_alert_policy(name=policy_name)
    print(f"Retrieved policy: {policy.display_name}")

# Run async function
asyncio.run(manage_alert_policies())

Resource Path Helpers

class AlertPolicyServiceClient:
    @staticmethod
    def alert_policy_path(project: str, alert_policy: str) -> str:
        """Returns a fully-qualified alert_policy string."""
        
    @staticmethod
    def alert_policy_condition_path(project: str, alert_policy: str, condition: str) -> str:
        """Returns a fully-qualified alert_policy_condition string."""
        
    @staticmethod
    def parse_alert_policy_path(path: str) -> Dict[str, str]:
        """Parses a alert_policy path into its component segments."""

Error Handling

Alert policy operations can raise specific exceptions:

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

client = AlertPolicyServiceClient()

try:
    policy = client.get_alert_policy(name="invalid/path")
except exceptions.NotFound:
    print("Alert policy not found")
except exceptions.InvalidArgument as e:
    print(f"Invalid request: {e}")
except exceptions.PermissionDenied:
    print("Insufficient permissions")
except exceptions.ResourceExhausted:
    print("Quota exceeded")

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