Alert Suppression (Snooze)

class SnoozeServiceClient:
    def create_snooze(
        self,
        request=None,
        *,
        parent: str = None,
        snooze=None,
        retry=None,
        timeout=None,
        metadata=()
    ) -> snooze.Snooze:
        """
        Creates a Snooze that will prevent alerts.

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

        Returns:
            Created Snooze object
        """

    def list_snoozes(
        self,
        request=None,
        *,
        parent: str = None,
        retry=None,
        timeout=None,
        metadata=()
    ) -> pagers.ListSnoozesPager:
        """
        Lists the existing valid Snooze records.

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

        Returns:
            Pager for iterating over Snooze objects
        """

    def get_snooze(
        self,
        request=None,
        *,
        name: str = None,
        retry=None,
        timeout=None,
        metadata=()
    ) -> snooze.Snooze:
        """
        Retrieves a Snooze by name.

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

        Returns:
            Snooze object
        """

    def update_snooze(
        self,
        request=None,
        *,
        snooze=None,
        update_mask=None,
        retry=None,
        timeout=None,
        metadata=()
    ) -> snooze.Snooze:
        """
        Updates a Snooze, identified by its name.

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

        Returns:
            Updated Snooze object
        """

class Snooze:
    name: str  # Resource name
    criteria: Snooze.Criteria  # Criteria for when snooze applies
    interval: TimeInterval  # Time interval when snooze is active
    display_name: str  # Human-readable name

class Snooze.Criteria:
    policies: List[str]  # Alert policy names to snooze

class CreateSnoozeRequest:
    parent: str  # Required. Project name
    snooze: Snooze  # Required. Snooze to create

class GetSnoozeRequest:
    name: str  # Required. Snooze name to retrieve

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

class ListSnoozesResponse:
    snoozes: List[Snooze]  # Snoozes
    next_page_token: str  # Next page token

class UpdateSnoozeRequest:
    snooze: Snooze  # Required. Updated snooze
    update_mask: FieldMask  # Fields to update

from google.cloud.monitoring import SnoozeServiceClient
from google.cloud.monitoring_v3.types import Snooze, TimeInterval
from google.protobuf.timestamp_pb2 import Timestamp
import time

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

# Create snooze for maintenance window
snooze_obj = Snooze()
snooze_obj.display_name = "Weekly Maintenance Window"

# Define snooze criteria
criteria = Snooze.Criteria()
criteria.policies = [
    f"projects/{project_id}/alertPolicies/{alert_policy_id_1}",
    f"projects/{project_id}/alertPolicies/{alert_policy_id_2}"
]
snooze_obj.criteria = criteria

# Define time interval (next Sunday 2-4 AM)
now = time.time()
start_time = now + (6 * 24 * 60 * 60)  # Next Sunday
end_time = start_time + (2 * 60 * 60)  # 2 hours later

interval = TimeInterval()
interval.start_time.seconds = int(start_time)
interval.end_time.seconds = int(end_time)
snooze_obj.interval = interval

created_snooze = client.create_snooze(
    parent=project_name,
    snooze=snooze_obj
)
print(f"Created snooze: {created_snooze.name}")
print(f"Display name: {created_snooze.display_name}")
print(f"Start time: {created_snooze.interval.start_time}")
print(f"End time: {created_snooze.interval.end_time}")

from datetime import datetime, timedelta

# Create emergency snooze to suppress all alerts temporarily
emergency_snooze = Snooze()
emergency_snooze.display_name = "Emergency Alert Suppression"

# Snooze all alert policies in the project
criteria = Snooze.Criteria()
# You would populate this with actual alert policy names
criteria.policies = [
    f"projects/{project_id}/alertPolicies/{policy_id}"
    for policy_id in alert_policy_ids  # List of policy IDs
]
emergency_snooze.criteria = criteria

# Set interval for next 4 hours
now = time.time()
interval = TimeInterval()
interval.start_time.seconds = int(now)
interval.end_time.seconds = int(now + 4 * 60 * 60)  # 4 hours
emergency_snooze.interval = interval

created_emergency = client.create_snooze(
    parent=project_name,
    snooze=emergency_snooze
)
print(f"Created emergency snooze: {created_emergency.name}")
print(f"Suppressing alerts for 4 hours")

# Create snooze for regular maintenance (happening every week)
weekly_snooze = Snooze()
weekly_snooze.display_name = "Database Maintenance - Weekly"

# Define which alert policies to snooze
criteria = Snooze.Criteria()
criteria.policies = [
    f"projects/{project_id}/alertPolicies/{database_alert_policy_id}"
]
weekly_snooze.criteria = criteria

# Schedule for this Sunday 3-5 AM UTC
import datetime
now = datetime.datetime.utcnow()
days_until_sunday = (6 - now.weekday()) % 7
next_sunday = now + datetime.timedelta(days=days_until_sunday)
maintenance_start = next_sunday.replace(hour=3, minute=0, second=0, microsecond=0)
maintenance_end = maintenance_start + datetime.timedelta(hours=2)

interval = TimeInterval()
interval.start_time.seconds = int(maintenance_start.timestamp())
interval.end_time.seconds = int(maintenance_end.timestamp())
weekly_snooze.interval = interval

created_weekly = client.create_snooze(
    parent=project_name,
    snooze=weekly_snooze
)
print(f"Created weekly maintenance snooze: {created_weekly.name}")
print(f"Scheduled for: {maintenance_start} - {maintenance_end} UTC")

# List all snoozes
print("All snoozes:")
for snooze_item in client.list_snoozes(parent=project_name):
    print(f"- {snooze_item.display_name}: {snooze_item.name}")
    
    start_time = datetime.datetime.fromtimestamp(snooze_item.interval.start_time.seconds)
    end_time = datetime.datetime.fromtimestamp(snooze_item.interval.end_time.seconds)
    
    print(f"  Start: {start_time}")
    print(f"  End: {end_time}")
    print(f"  Policies: {len(snooze_item.criteria.policies)}")
    
    # Check if snooze is currently active
    current_time = time.time()
    if (snooze_item.interval.start_time.seconds <= current_time <= 
        snooze_item.interval.end_time.seconds):
        print("  Status: ACTIVE")
    elif snooze_item.interval.start_time.seconds > current_time:
        print("  Status: SCHEDULED")
    else:
        print("  Status: EXPIRED")

# Filter for active snoozes only
current_timestamp = int(time.time())
filter_expr = f'interval.start_time.seconds <= {current_timestamp} AND interval.end_time.seconds >= {current_timestamp}'

print(f"\nCurrently active snoozes:")
for active_snooze in client.list_snoozes(parent=project_name, filter=filter_expr):
    print(f"- {active_snooze.display_name}")

from google.protobuf import field_mask_pb2

# Get existing snooze
snooze_name = f"projects/{project_id}/snoozes/{snooze_id}"
snooze_obj = client.get_snooze(name=snooze_name)

# Update snooze display name
snooze_obj.display_name = "Updated Weekly Maintenance Window"

# Extend the snooze duration by 1 hour
snooze_obj.interval.end_time.seconds += 3600  # Add 1 hour

# Create field mask for selective update
update_mask = field_mask_pb2.FieldMask()
update_mask.paths.extend(["display_name", "interval.end_time"])

updated_snooze = client.update_snooze(
    snooze=snooze_obj,
    update_mask=update_mask
)
print(f"Updated snooze: {updated_snooze.display_name}")
print(f"New end time: {datetime.datetime.fromtimestamp(updated_snooze.interval.end_time.seconds)}")

# Create snooze that targets specific alert policies based on their display names
from google.cloud.monitoring import AlertPolicyServiceClient

# First, get alert policies to snooze
alert_client = AlertPolicyServiceClient()
policies_to_snooze = []

for policy in alert_client.list_alert_policies(name=project_name):
    # Snooze only high CPU alerts during maintenance
    if "High CPU" in policy.display_name:
        policies_to_snooze.append(policy.name)

if policies_to_snooze:
    # Create targeted snooze
    targeted_snooze = Snooze()
    targeted_snooze.display_name = "CPU Alert Maintenance Snooze"
    
    criteria = Snooze.Criteria()
    criteria.policies = policies_to_snooze
    targeted_snooze.criteria = criteria
    
    # Set for next planned maintenance
    now = time.time()
    maintenance_window = 2 * 60 * 60  # 2 hours
    
    interval = TimeInterval()
    interval.start_time.seconds = int(now + 3600)  # Start in 1 hour
    interval.end_time.seconds = int(now + 3600 + maintenance_window)
    targeted_snooze.interval = interval
    
    created_targeted = client.create_snooze(
        parent=project_name,
        snooze=targeted_snooze
    )
    print(f"Created targeted snooze for {len(policies_to_snooze)} CPU alert policies")
    print(f"Snooze name: {created_targeted.name}")

import pytz
from datetime import datetime

# Create snooze with specific timezone considerations
timezone_snooze = Snooze()
timezone_snooze.display_name = "East Coast Maintenance Window"

criteria = Snooze.Criteria()
criteria.policies = [f"projects/{project_id}/alertPolicies/{policy_id}"]
timezone_snooze.criteria = criteria

# Schedule for 2 AM EST (7 AM UTC during standard time)
est = pytz.timezone('US/Eastern')
utc = pytz.UTC

# Define maintenance window in EST
local_start = est.localize(datetime(2024, 1, 15, 2, 0, 0))  # 2 AM EST
local_end = est.localize(datetime(2024, 1, 15, 4, 0, 0))    # 4 AM EST

# Convert to UTC for API
utc_start = local_start.astimezone(utc)
utc_end = local_end.astimezone(utc)

interval = TimeInterval()
interval.start_time.seconds = int(utc_start.timestamp())
interval.end_time.seconds = int(utc_end.timestamp())
timezone_snooze.interval = interval

created_timezone = client.create_snooze(
    parent=project_name,
    snooze=timezone_snooze
)
print(f"Created timezone-aware snooze: {created_timezone.name}")
print(f"Local time: {local_start} - {local_end} EST")
print(f"UTC time: {utc_start} - {utc_end} UTC")

def get_active_snoozes(client, project_name):
    """Get all currently active snoozes."""
    current_time = int(time.time())
    active_snoozes = []
    
    for snooze_item in client.list_snoozes(parent=project_name):
        if (snooze_item.interval.start_time.seconds <= current_time <= 
            snooze_item.interval.end_time.seconds):
            active_snoozes.append(snooze_item)
    
    return active_snoozes

def get_upcoming_snoozes(client, project_name, hours_ahead=24):
    """Get snoozes starting within the next N hours."""
    current_time = int(time.time())
    future_time = current_time + (hours_ahead * 3600)
    upcoming_snoozes = []
    
    for snooze_item in client.list_snoozes(parent=project_name):
        if (current_time < snooze_item.interval.start_time.seconds <= future_time):
            upcoming_snoozes.append(snooze_item)
    
    return upcoming_snoozes

# Usage
active = get_active_snoozes(client, project_name)
print(f"Currently active snoozes: {len(active)}")

upcoming = get_upcoming_snoozes(client, project_name, 48)  # Next 48 hours
print(f"Upcoming snoozes (next 48h): {len(upcoming)}")

for snooze_item in upcoming:
    start_dt = datetime.datetime.fromtimestamp(snooze_item.interval.start_time.seconds)
    print(f"- {snooze_item.display_name} starts at {start_dt}")

import asyncio
from google.cloud.monitoring import SnoozeServiceAsyncClient

async def manage_snoozes():
    client = SnoozeServiceAsyncClient()
    project_name = f"projects/{project_id}"
    
    # List snoozes asynchronously
    async for snooze_item in await client.list_snoozes(parent=project_name):
        print(f"Async snooze: {snooze_item.display_name}")
    
    # Get snooze asynchronously
    snooze_obj = await client.get_snooze(name=snooze_name)
    print(f"Retrieved async snooze: {snooze_obj.display_name}")
    
    # Update snooze asynchronously
    snooze_obj.display_name = "Async Updated Snooze"
    updated = await client.update_snooze(snooze=snooze_obj)
    print(f"Async updated snooze: {updated.display_name}")

asyncio.run(manage_snoozes())

class SnoozeServiceClient:
    @staticmethod
    def alert_policy_path(project: str, alert_policy: str) -> str:
        """Returns a fully-qualified alert_policy string."""
        
    @staticmethod
    def snooze_path(project: str, snooze: str) -> str:
        """Returns a fully-qualified snooze string."""
        
    @staticmethod
    def parse_snooze_path(path: str) -> Dict[str, str]:
        """Parses a snooze path into its component segments."""

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

client = SnoozeServiceClient()

try:
    snooze_obj = client.get_snooze(name="invalid/path")
except exceptions.NotFound:
    print("Snooze not found")
except exceptions.InvalidArgument as e:
    print(f"Invalid snooze configuration: {e}")
    # Common issues: invalid time intervals, non-existent alert policies
except exceptions.PermissionDenied:
    print("Insufficient permissions")
except exceptions.FailedPrecondition as e:
    print(f"Snooze conflict or invalid state: {e}")

def create_maintenance_snooze_with_validation(
    client, 
    project_name, 
    alert_policy_ids, 
    start_time, 
    duration_hours,
    display_name
):
    """Create a maintenance snooze with proper validation."""
    
    # Validate alert policies exist
    alert_client = AlertPolicyServiceClient()
    valid_policies = []
    
    for policy_id in alert_policy_ids:
        policy_name = f"projects/{project_id}/alertPolicies/{policy_id}"
        try:
            alert_client.get_alert_policy(name=policy_name)
            valid_policies.append(policy_name)
        except exceptions.NotFound:
            print(f"Warning: Alert policy {policy_id} not found, skipping")
    
    if not valid_policies:
        raise ValueError("No valid alert policies provided")
    
    # Create snooze
    snooze_obj = Snooze()
    snooze_obj.display_name = display_name
    
    criteria = Snooze.Criteria()
    criteria.policies = valid_policies
    snooze_obj.criteria = criteria
    
    # Set time interval
    end_time = start_time + (duration_hours * 3600)
    interval = TimeInterval()
    interval.start_time.seconds = int(start_time)
    interval.end_time.seconds = int(end_time)
    snooze_obj.interval = interval
    
    return client.create_snooze(parent=project_name, snooze=snooze_obj)

# Usage
maintenance_snooze = create_maintenance_snooze_with_validation(
    client=client,
    project_name=project_name,
    alert_policy_ids=["policy1", "policy2", "policy3"],
    start_time=time.time() + 3600,  # Start in 1 hour
    duration_hours=2,
    display_name="Validated Maintenance Window"
)
print(f"Created validated snooze: {maintenance_snooze.name}")