Uptime Check Management

class UptimeCheckServiceClient:
    def list_uptime_check_configs(
        self,
        request=None,
        *,
        parent: str = None,
        retry=None,
        timeout=None,
        metadata=()
    ) -> pagers.ListUptimeCheckConfigsPager:
        """
        Lists the existing valid Uptime check configurations.

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

        Returns:
            Pager for iterating over UptimeCheckConfig objects
        """

    def get_uptime_check_config(
        self,
        request=None,
        *,
        name: str = None,
        retry=None,
        timeout=None,
        metadata=()
    ) -> uptime.UptimeCheckConfig:
        """
        Gets a single Uptime check configuration.

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

        Returns:
            UptimeCheckConfig object
        """

    def create_uptime_check_config(
        self,
        request=None,
        *,
        parent: str = None,
        uptime_check_config=None,
        retry=None,
        timeout=None,
        metadata=()
    ) -> uptime.UptimeCheckConfig:
        """
        Creates a new Uptime check configuration.

        Args:
            request: The request object or dict equivalent
            parent: Required. Project name
            uptime_check_config: Required. The uptime check config to create
            retry: Retry configuration
            timeout: Request timeout in seconds
            metadata: Additional metadata

        Returns:
            Created UptimeCheckConfig object
        """

    def update_uptime_check_config(
        self,
        request=None,
        *,
        uptime_check_config=None,
        retry=None,
        timeout=None,
        metadata=()
    ) -> uptime.UptimeCheckConfig:
        """
        Updates an Uptime check configuration.

        Args:
            request: The request object or dict equivalent
            uptime_check_config: Required. Updated uptime check config
            retry: Retry configuration
            timeout: Request timeout in seconds
            metadata: Additional metadata

        Returns:
            Updated UptimeCheckConfig object
        """

    def delete_uptime_check_config(
        self,
        request=None,
        *,
        name: str = None,
        retry=None,
        timeout=None,
        metadata=()
    ) -> None:
        """
        Deletes an Uptime check configuration.

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

    def list_uptime_check_ips(
        self,
        request=None,
        *,
        retry=None,
        timeout=None,
        metadata=()
    ) -> pagers.ListUptimeCheckIpsPager:
        """
        Returns the list of IP addresses that checkers run from.

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

        Returns:
            Pager for iterating over UptimeCheckIp objects
        """

class UptimeCheckConfig:
    name: str  # Resource name
    display_name: str  # Human-readable name
    monitored_resource: MonitoredResource  # Resource to monitor
    resource_group: UptimeCheckConfig.ResourceGroup  # Resource group to monitor
    http_check: UptimeCheckConfig.HttpCheck  # HTTP check configuration
    tcp_check: UptimeCheckConfig.TcpCheck  # TCP check configuration
    period: Duration  # Check frequency
    timeout: Duration  # Check timeout
    content_matchers: List[UptimeCheckConfig.ContentMatcher]  # Response content validation
    checker_type: UptimeCheckConfig.CheckerType  # Type of checker
    selected_regions: List[UptimeCheckRegion]  # Regions to run checks from
    is_internal: bool  # Whether check is internal
    internal_checkers: List[InternalChecker]  # Internal checker configurations
    user_labels: Dict[str, str]  # User-defined labels

class UptimeCheckConfig.HttpCheck:
    request_method: UptimeCheckConfig.HttpCheck.RequestMethod  # HTTP method
    use_ssl: bool  # Whether to use HTTPS
    path: str  # URL path to check
    port: int  # Port number
    auth_info: UptimeCheckConfig.HttpCheck.BasicAuthentication  # Authentication
    mask_headers: bool  # Whether to mask headers in logs
    headers: Dict[str, str]  # HTTP headers
    content_type: UptimeCheckConfig.HttpCheck.ContentType  # Request content type
    validate_ssl: bool  # Whether to validate SSL certificates
    body: bytes  # Request body
    accepted_response_status_codes: List[UptimeCheckConfig.HttpCheck.ResponseStatusCode]  # Accepted status codes
    ping_config: UptimeCheckConfig.PingConfig  # Ping configuration

class UptimeCheckConfig.TcpCheck:
    port: int  # Port number
    ping_config: UptimeCheckConfig.PingConfig  # Ping configuration

class UptimeCheckConfig.ContentMatcher:
    content: str  # Content to match
    matcher: UptimeCheckConfig.ContentMatcher.ContentMatcherOption  # Matching option
    json_path_matcher: UptimeCheckConfig.ContentMatcher.JsonPathMatcher  # JSON path matcher

class UptimeCheckConfig.ResourceGroup:
    group_id: str  # Group ID
    resource_type: GroupResourceType  # Resource type

class UptimeCheckConfig.CheckerType(enum.Enum):
    CHECKER_TYPE_UNSPECIFIED = 0
    STATIC_IP_CHECKERS = 1
    VPC_CHECKERS = 2

class UptimeCheckIp:
    region: UptimeCheckRegion  # Region of the checker
    location: str  # Location description
    ip_address: str  # IP address

class InternalChecker:
    name: str  # Checker name
    display_name: str  # Human-readable name
    network: str  # Network to check from
    gcp_zone: str  # GCP zone
    peer_project_id: str  # Peer project ID
    state: InternalChecker.State  # Checker state

class InternalChecker.State(enum.Enum):
    UNSPECIFIED = 0
    CREATING = 1
    RUNNING = 2

class CreateUptimeCheckConfigRequest:
    parent: str  # Required. Project name
    uptime_check_config: UptimeCheckConfig  # Required. Config to create

class DeleteUptimeCheckConfigRequest:
    name: str  # Required. Config name to delete

class GetUptimeCheckConfigRequest:
    name: str  # Required. Config name to retrieve

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

class ListUptimeCheckConfigsResponse:
    uptime_check_configs: List[UptimeCheckConfig]  # Uptime check configs
    next_page_token: str  # Next page token
    total_size: int  # Total number of configs

class ListUptimeCheckIpsRequest:
    page_size: int  # Maximum results per page
    page_token: str  # Page token

class ListUptimeCheckIpsResponse:
    uptime_check_ips: List[UptimeCheckIp]  # IP addresses
    next_page_token: str  # Next page token

class UpdateUptimeCheckConfigRequest:
    uptime_check_config: UptimeCheckConfig  # Required. Updated config
    update_mask: FieldMask  # Fields to update

from google.cloud.monitoring import UptimeCheckServiceClient
from google.cloud.monitoring_v3.types import UptimeCheckConfig, UptimeCheckRegion
from google.protobuf.duration_pb2 import Duration

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

# Create HTTP uptime check
config = UptimeCheckConfig()
config.display_name = "Website Health Check"

# Configure HTTP check
http_check = UptimeCheckConfig.HttpCheck()
http_check.use_ssl = True
http_check.path = "/health"
http_check.port = 443
http_check.request_method = UptimeCheckConfig.HttpCheck.RequestMethod.GET

# Add custom headers
http_check.headers["User-Agent"] = "Google-Cloud-Monitoring-Uptime-Check"
http_check.headers["Accept"] = "application/json"

config.http_check = http_check

# Set monitored resource
config.monitored_resource.type = "uptime_url"
config.monitored_resource.labels["project_id"] = project_id
config.monitored_resource.labels["host"] = "example.com"

# Configure check frequency (5 minutes)
period = Duration()
period.seconds = 300
config.period = period

# Configure timeout (10 seconds)
timeout = Duration()
timeout.seconds = 10
config.timeout = timeout

# Select regions for checking
config.selected_regions = [
    UptimeCheckRegion.USA,
    UptimeCheckRegion.EUROPE,
    UptimeCheckRegion.ASIA_PACIFIC
]

# Add content matcher
content_matcher = UptimeCheckConfig.ContentMatcher()
content_matcher.content = "status"
content_matcher.matcher = UptimeCheckConfig.ContentMatcher.ContentMatcherOption.CONTAINS_STRING
config.content_matchers = [content_matcher]

# Add user labels
config.user_labels["environment"] = "production"
config.user_labels["team"] = "ops"

created_config = client.create_uptime_check_config(
    parent=project_name,
    uptime_check_config=config
)
print(f"Created uptime check: {created_config.name}")
print(f"Display name: {created_config.display_name}")

# Create TCP uptime check
tcp_config = UptimeCheckConfig()
tcp_config.display_name = "Database Connection Check"

# Configure TCP check
tcp_check = UptimeCheckConfig.TcpCheck()
tcp_check.port = 5432  # PostgreSQL port

config.tcp_check = tcp_check

# Set monitored resource for TCP check
tcp_config.monitored_resource.type = "gce_instance"
tcp_config.monitored_resource.labels["project_id"] = project_id
tcp_config.monitored_resource.labels["instance_id"] = "database-server-1"
tcp_config.monitored_resource.labels["zone"] = "us-central1-a"

# Configure check frequency (1 minute)
period = Duration()
period.seconds = 60
tcp_config.period = period

# Configure timeout (5 seconds)
timeout = Duration()
timeout.seconds = 5
tcp_config.timeout = timeout

created_tcp_config = client.create_uptime_check_config(
    parent=project_name,
    uptime_check_config=tcp_config
)
print(f"Created TCP uptime check: {created_tcp_config.name}")

# Create HTTP check with basic authentication
auth_config = UptimeCheckConfig()
auth_config.display_name = "Authenticated API Check"

# Configure HTTP check with authentication
auth_http_check = UptimeCheckConfig.HttpCheck()
auth_http_check.use_ssl = True
auth_http_check.path = "/api/v1/status"
auth_http_check.port = 443
auth_http_check.request_method = UptimeCheckConfig.HttpCheck.RequestMethod.GET

# Configure basic authentication
auth_info = UptimeCheckConfig.HttpCheck.BasicAuthentication()
auth_info.username = "monitor"
auth_info.password = "secret_password"
auth_http_check.auth_info = auth_info

auth_config.http_check = auth_http_check

# Set resource
auth_config.monitored_resource.type = "uptime_url" 
auth_config.monitored_resource.labels["project_id"] = project_id
auth_config.monitored_resource.labels["host"] = "api.example.com"

created_auth_config = client.create_uptime_check_config(
    parent=project_name,
    uptime_check_config=auth_config
)
print(f"Created authenticated uptime check: {created_auth_config.name}")

# Create internal uptime check (from VPC)
internal_config = UptimeCheckConfig()
internal_config.display_name = "Internal Service Check"
internal_config.is_internal = True
internal_config.checker_type = UptimeCheckConfig.CheckerType.VPC_CHECKERS

# Configure HTTP check for internal resource
internal_http = UptimeCheckConfig.HttpCheck()
internal_http.path = "/internal/health"
internal_http.port = 8080
internal_config.http_check = internal_http

# Set internal monitored resource
internal_config.monitored_resource.type = "gce_instance"
internal_config.monitored_resource.labels["project_id"] = project_id
internal_config.monitored_resource.labels["instance_id"] = "internal-service-1"
internal_config.monitored_resource.labels["zone"] = "us-central1-a"

# Configure internal checker
internal_checker = InternalChecker()
internal_checker.display_name = "VPC Checker"
internal_checker.network = f"projects/{project_id}/global/networks/default"
internal_checker.gcp_zone = "us-central1-a"
internal_config.internal_checkers = [internal_checker]

created_internal = client.create_uptime_check_config(
    parent=project_name,
    uptime_check_config=internal_config
)
print(f"Created internal uptime check: {created_internal.name}")

# List all uptime check configurations
print("Uptime check configurations:")
for config in client.list_uptime_check_configs(parent=project_name):
    print(f"- {config.display_name}: {config.name}")
    if config.http_check:
        print(f"  HTTP check: {config.monitored_resource.labels.get('host', 'N/A')}")
        print(f"  Path: {config.http_check.path}")
        print(f"  SSL: {config.http_check.use_ssl}")
    elif config.tcp_check:
        print(f"  TCP check on port: {config.tcp_check.port}")
    
    print(f"  Period: {config.period.seconds}s")
    print(f"  Regions: {len(config.selected_regions)}")

# Filter by labels
filter_expr = 'user_labels.environment="production"'
print(f"\nProduction uptime checks:")
for config in client.list_uptime_check_configs(parent=project_name, filter=filter_expr):
    print(f"- {config.display_name}")

from google.protobuf import field_mask_pb2

# Get existing configuration
config_name = f"projects/{project_id}/uptimeCheckConfigs/{config_id}"
config = client.get_uptime_check_config(name=config_name)

# Update configuration
config.display_name = "Updated Website Health Check"
config.http_check.path = "/api/health"

# Update timeout
config.timeout.seconds = 15

# Update user labels
config.user_labels["version"] = "v2.0"

# Create field mask for selective update
update_mask = field_mask_pb2.FieldMask()
update_mask.paths.extend([
    "display_name", 
    "http_check.path", 
    "timeout", 
    "user_labels"
])

updated_config = client.update_uptime_check_config(
    uptime_check_config=config,
    update_mask=update_mask
)
print(f"Updated uptime check: {updated_config.display_name}")

# Create uptime check with JSON path matcher
json_config = UptimeCheckConfig()
json_config.display_name = "API JSON Response Check"

# Configure HTTP check
json_http_check = UptimeCheckConfig.HttpCheck()
json_http_check.use_ssl = True
json_http_check.path = "/api/v1/health"
json_http_check.port = 443
json_config.http_check = json_http_check

# Set resource
json_config.monitored_resource.type = "uptime_url"
json_config.monitored_resource.labels["project_id"] = project_id
json_config.monitored_resource.labels["host"] = "api.example.com"

# Add JSON path content matcher
json_matcher = UptimeCheckConfig.ContentMatcher()
json_path_matcher = UptimeCheckConfig.ContentMatcher.JsonPathMatcher()
json_path_matcher.json_path = "$.status"
json_path_matcher.json_matcher = UptimeCheckConfig.ContentMatcher.ContentMatcherOption.MATCHES_JSON_PATH
json_matcher.json_path_matcher = json_path_matcher
json_matcher.content = "healthy"

json_config.content_matchers = [json_matcher]

created_json_config = client.create_uptime_check_config(
    parent=project_name,
    uptime_check_config=json_config
)
print(f"Created JSON path uptime check: {created_json_config.name}")

# List IP addresses used by uptime checkers
print("Uptime checker IP addresses:")
for ip_info in client.list_uptime_check_ips():
    print(f"- Region: {ip_info.region}")
    print(f"  Location: {ip_info.location}")
    print(f"  IP: {ip_info.ip_address}")

config_name = f"projects/{project_id}/uptimeCheckConfigs/{config_id}"
client.delete_uptime_check_config(name=config_name)
print(f"Deleted uptime check: {config_name}")

import asyncio
from google.cloud.monitoring import UptimeCheckServiceAsyncClient

async def manage_uptime_checks():
    client = UptimeCheckServiceAsyncClient()
    project_name = f"projects/{project_id}"
    
    # List configurations asynchronously
    async for config in await client.list_uptime_check_configs(parent=project_name):
        print(f"Async uptime check: {config.display_name}")
    
    # Get configuration asynchronously
    config = await client.get_uptime_check_config(name=config_name)
    print(f"Retrieved async config: {config.display_name}")

asyncio.run(manage_uptime_checks())

class UptimeCheckServiceClient:
    @staticmethod
    def uptime_check_config_path(project: str, uptime_check_config: str) -> str:
        """Returns a fully-qualified uptime_check_config string."""
        
    @staticmethod
    def parse_uptime_check_config_path(path: str) -> Dict[str, str]:
        """Parses an uptime_check_config path into its component segments."""

class SyntheticMonitorTarget:
    """Describes a Synthetic Monitor to be invoked by Uptime."""
    
    class CloudFunctionV2Target:
        """A Synthetic Monitor deployed to a Cloud Functions V2 instance."""
        name: str  # Required. Fully qualified GCFv2 resource name
        cloud_run_revision: MonitoredResource  # Output only. Associated cloud_run_revision
    
    cloud_function_v2: CloudFunctionV2Target  # Target a Synthetic Monitor GCFv2 instance

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

client = UptimeCheckServiceClient()

try:
    config = client.get_uptime_check_config(name="invalid/path")
except exceptions.NotFound:
    print("Uptime check configuration not found")
except exceptions.InvalidArgument as e:
    print(f"Invalid uptime check configuration: {e}")
except exceptions.PermissionDenied:
    print("Insufficient permissions")
except exceptions.ResourceExhausted as e:
    print(f"Uptime check quota exceeded: {e}")