Monitoring and Observability

from google.api.metric_pb2 import MetricDescriptor, Metric
from google.api.label_pb2 import LabelDescriptor

class MetricDescriptor(message.Message):
    """Describes a metric type and its schema."""
    name: str  # Metric type name (e.g., "compute.googleapis.com/instance/cpu/usage")
    type: str  # Metric type identifier
    labels: list[LabelDescriptor]  # Metric labels
    metric_kind: MetricDescriptor.MetricKind  # Kind of measurement
    value_type: MetricDescriptor.ValueType  # Type of metric values
    unit: str  # Unit of measurement
    description: str  # Metric description
    display_name: str  # Human-readable name
    metadata: MetricDescriptor.MetricDescriptorMetadata  # Additional metadata
    launch_stage: LaunchStage  # API maturity level
    monitored_resource_types: list[str]  # Compatible resource types

    class MetricKind(enum.Enum):
        """Kind of measurement."""
        METRIC_KIND_UNSPECIFIED = 0
        GAUGE = 1  # Instantaneous measurement
        DELTA = 2  # Change since last recorded value
        CUMULATIVE = 3  # Cumulative measurement

    class ValueType(enum.Enum):
        """Type of metric values."""
        VALUE_TYPE_UNSPECIFIED = 0
        BOOL = 1  # Boolean values
        INT64 = 2  # 64-bit integers
        DOUBLE = 3  # Double precision floats
        STRING = 4  # String values
        DISTRIBUTION = 5  # Distribution values
        MONEY = 6  # Monetary values

class Metric(message.Message):
    """Single metric measurement."""
    type: str  # Metric type identifier
    labels: dict[str, str]  # Metric label values

class LabelDescriptor(message.Message):
    """Describes a label for metrics and resources."""
    key: str  # Label key
    value_type: LabelDescriptor.ValueType  # Label value type
    description: str  # Label description

    class ValueType(enum.Enum):
        """Label value types."""
        STRING = 0
        BOOL = 1
        INT64 = 2

from google.api.distribution_pb2 import Distribution

class Distribution(message.Message):
    """Statistical distribution of values."""
    count: int  # Number of samples
    mean: float  # Mean of samples
    sum_of_squared_deviation: float  # Sum of squared deviations from mean
    range: Distribution.Range  # Min/max range
    bucket_options: Distribution.BucketOptions  # Bucket configuration
    bucket_counts: list[int]  # Count per bucket
    exemplars: list[Distribution.Exemplar]  # Example values

    class Range(message.Message):
        """Min and max values in the distribution."""
        min: float  # Minimum value
        max: float  # Maximum value

    class BucketOptions(message.Message):
        """Bucket options for histogram."""
        # Union field 'options' (one of):
        linear_buckets: Distribution.BucketOptions.Linear
        exponential_buckets: Distribution.BucketOptions.Exponential
        explicit_buckets: Distribution.BucketOptions.Explicit

        class Linear(message.Message):
            """Linear bucket spacing."""
            num_finite_buckets: int  # Number of finite buckets
            width: float  # Bucket width
            offset: float  # Starting offset

        class Exponential(message.Message):
            """Exponential bucket spacing."""
            num_finite_buckets: int  # Number of finite buckets
            growth_factor: float  # Growth factor between buckets
            scale: float  # Scale factor

        class Explicit(message.Message):
            """Explicit bucket boundaries."""
            bounds: list[float]  # Bucket boundary values

    class Exemplar(message.Message):
        """Example value from the distribution."""
        value: float  # Example value
        timestamp: Timestamp  # When value was recorded
        attachments: list[Any]  # Additional context

from google.api.monitored_resource_pb2 import (
    MonitoredResourceDescriptor, MonitoredResource, MonitoredResourceMetadata
)

class MonitoredResourceDescriptor(message.Message):
    """Describes a monitored resource type."""
    name: str  # Resource descriptor name
    type: str  # Resource type identifier
    display_name: str  # Human-readable name
    description: str  # Resource description
    labels: list[LabelDescriptor]  # Resource labels
    launch_stage: LaunchStage  # API maturity level

class MonitoredResource(message.Message):
    """Monitored resource instance."""
    type: str  # Resource type
    labels: dict[str, str]  # Resource label values

class MonitoredResourceMetadata(message.Message):
    """Additional metadata for monitored resources."""
    system_labels: Struct  # System-defined labels
    user_labels: dict[str, str]  # User-defined labels

from google.logging.type.http_request_pb2 import HttpRequest
from google.logging.type.log_severity_pb2 import LogSeverity

class HttpRequest(message.Message):
    """HTTP request information for logging."""
    request_method: str  # HTTP method (GET, POST, etc.)
    request_url: str  # Request URL
    request_size: int  # Request size in bytes
    status: int  # HTTP status code
    response_size: int  # Response size in bytes
    user_agent: str  # User agent string
    remote_ip: str  # Client IP address
    server_ip: str  # Server IP address
    referer: str  # HTTP referer
    latency: Duration  # Request latency
    cache_lookup: bool  # Cache lookup performed
    cache_hit: bool  # Cache hit occurred
    cache_validated_with_origin_server: bool  # Cache validation occurred
    cache_fill_bytes: int  # Bytes sent to cache
    protocol: str  # Protocol version

class LogSeverity(enum.Enum):
    """Log entry severity levels."""
    DEFAULT = 0  # Default severity
    DEBUG = 100  # Debug information
    INFO = 200  # Informational messages
    NOTICE = 300  # Normal but significant events
    WARNING = 400  # Warning conditions
    ERROR = 500  # Error conditions
    CRITICAL = 600  # Critical conditions
    ALERT = 700  # Action must be taken immediately
    EMERGENCY = 800  # System is unusable

from google.api.metric_pb2 import MetricDescriptor
from google.api.label_pb2 import LabelDescriptor

# Create a CPU usage metric descriptor
cpu_metric = MetricDescriptor()
cpu_metric.name = "projects/my-project/metricDescriptors/compute.googleapis.com/instance/cpu/usage"
cpu_metric.type = "compute.googleapis.com/instance/cpu/usage"
cpu_metric.metric_kind = MetricDescriptor.MetricKind.GAUGE
cpu_metric.value_type = MetricDescriptor.ValueType.DOUBLE
cpu_metric.unit = "1"  # Percentage (0.0-1.0)
cpu_metric.description = "CPU usage percentage"
cpu_metric.display_name = "CPU Usage"

# Add labels
instance_label = LabelDescriptor()
instance_label.key = "instance_name"
instance_label.value_type = LabelDescriptor.ValueType.STRING
instance_label.description = "Name of the VM instance"
cpu_metric.labels.append(instance_label)

zone_label = LabelDescriptor()
zone_label.key = "zone"
zone_label.value_type = LabelDescriptor.ValueType.STRING  
zone_label.description = "GCP zone of the instance"
cpu_metric.labels.append(zone_label)

from google.api.metric_pb2 import Metric

# Record a metric value
metric = Metric()
metric.type = "compute.googleapis.com/instance/cpu/usage"
metric.labels["instance_name"] = "web-server-1"
metric.labels["zone"] = "us-central1-a"

# The actual metric value would be recorded in a TimeSeries
# (TimeSeries is typically part of monitoring service APIs)

from google.api.distribution_pb2 import Distribution

# Create a latency distribution
latency_dist = Distribution()
latency_dist.count = 1000  # 1000 requests
latency_dist.mean = 0.150  # 150ms average
latency_dist.sum_of_squared_deviation = 2.5

# Set range
latency_dist.range.min = 0.010  # 10ms minimum
latency_dist.range.max = 2.000  # 2s maximum

# Configure exponential buckets for latency
bucket_opts = latency_dist.bucket_options.exponential_buckets
bucket_opts.num_finite_buckets = 20
bucket_opts.growth_factor = 2.0
bucket_opts.scale = 0.01  # Start at 10ms

# Set bucket counts (example data)
latency_dist.bucket_counts.extend([
    10,   # < 10ms
    50,   # 10-20ms
    200,  # 20-40ms
    300,  # 40-80ms
    250,  # 80-160ms
    150,  # 160-320ms
    30,   # 320-640ms
    10,   # 640ms-1.28s
    0,    # > 1.28s
])

# Add exemplar
exemplar = latency_dist.exemplars.add()
exemplar.value = 0.095  # 95ms example
exemplar.timestamp.GetCurrentTime()

from google.api.monitored_resource_pb2 import MonitoredResourceDescriptor, MonitoredResource

# Define a monitored resource type
resource_desc = MonitoredResourceDescriptor()
resource_desc.name = "projects/my-project/monitoredResourceDescriptors/gce_instance"
resource_desc.type = "gce_instance"
resource_desc.display_name = "GCE VM Instance"
resource_desc.description = "A Google Compute Engine virtual machine instance"

# Add resource labels
project_label = LabelDescriptor()
project_label.key = "project_id"
project_label.value_type = LabelDescriptor.ValueType.STRING
project_label.description = "GCP project ID"
resource_desc.labels.append(project_label)

instance_label = LabelDescriptor()
instance_label.key = "instance_id"
instance_label.value_type = LabelDescriptor.ValueType.STRING
instance_label.description = "VM instance ID"
resource_desc.labels.append(instance_label)

# Create a monitored resource instance
resource = MonitoredResource()
resource.type = "gce_instance"
resource.labels["project_id"] = "my-project"
resource.labels["instance_id"] = "1234567890123456789"
resource.labels["zone"] = "us-central1-a"

from google.logging.type.http_request_pb2 import HttpRequest
from google.protobuf.duration_pb2 import Duration

# Log an HTTP request
http_request = HttpRequest()
http_request.request_method = "GET"
http_request.request_url = "https://api.example.com/users/123"
http_request.status = 200
http_request.request_size = 1024
http_request.response_size = 2048
http_request.user_agent = "MyApp/1.0"
http_request.remote_ip = "203.0.113.10"
http_request.referer = "https://example.com/dashboard"

# Set latency (250ms)
http_request.latency.seconds = 0
http_request.latency.nanos = 250000000

# Cache information
http_request.cache_lookup = True
http_request.cache_hit = False
http_request.cache_validated_with_origin_server = True

from google.logging.type.log_severity_pb2 import LogSeverity

def get_severity_name(severity: LogSeverity) -> str:
    """Convert severity enum to string."""
    severity_names = {
        LogSeverity.DEFAULT: "DEFAULT",
        LogSeverity.DEBUG: "DEBUG", 
        LogSeverity.INFO: "INFO",
        LogSeverity.NOTICE: "NOTICE",
        LogSeverity.WARNING: "WARNING",
        LogSeverity.ERROR: "ERROR",
        LogSeverity.CRITICAL: "CRITICAL",
        LogSeverity.ALERT: "ALERT",
        LogSeverity.EMERGENCY: "EMERGENCY"
    }
    return severity_names.get(severity, "UNKNOWN")

# Example usage in logging
log_severity = LogSeverity.ERROR
print(f"Log level: {get_severity_name(log_severity)}")

from google.api.metric_pb2 import MetricDescriptor

def create_custom_metric(name: str, description: str, unit: str, 
                        metric_kind: MetricDescriptor.MetricKind,
                        value_type: MetricDescriptor.ValueType) -> MetricDescriptor:
    """Create a custom metric descriptor."""
    metric = MetricDescriptor()
    metric.name = f"projects/my-project/metricDescriptors/{name}"
    metric.type = name
    metric.description = description
    metric.unit = unit
    metric.metric_kind = metric_kind
    metric.value_type = value_type
    metric.display_name = description
    return metric

# Create error rate metric
error_rate = create_custom_metric(
    name="custom.googleapis.com/service/error_rate",
    description="Service Error Rate",
    unit="1",  # Ratio
    metric_kind=MetricDescriptor.MetricKind.GAUGE,
    value_type=MetricDescriptor.ValueType.DOUBLE
)

# Create request count metric
request_count = create_custom_metric(
    name="custom.googleapis.com/service/request_count",
    description="Total Request Count",
    unit="1",  # Count
    metric_kind=MetricDescriptor.MetricKind.CUMULATIVE,
    value_type=MetricDescriptor.ValueType.INT64
)

from google.api.distribution_pb2 import Distribution
import math

def analyze_distribution(dist: Distribution):
    """Analyze distribution statistics."""
    if dist.count == 0:
        print("No data in distribution")
        return
        
    print(f"Sample count: {dist.count}")
    print(f"Mean: {dist.mean:.3f}")
    
    # Calculate standard deviation
    if dist.count > 1:
        variance = dist.sum_of_squared_deviation / (dist.count - 1)
        std_dev = math.sqrt(variance)
        print(f"Standard deviation: {std_dev:.3f}")
    
    print(f"Range: {dist.range.min:.3f} - {dist.range.max:.3f}")
    
    # Analyze bucket distribution
    if dist.bucket_counts:
        print(f"Histogram buckets: {len(dist.bucket_counts)}")
        total_samples = sum(dist.bucket_counts)
        print(f"Samples in buckets: {total_samples}")
        
        # Find mode (bucket with most samples)
        max_count = max(dist.bucket_counts)
        mode_bucket = dist.bucket_counts.index(max_count)
        print(f"Mode bucket: {mode_bucket} (count: {max_count})")

# Example usage
latency_distribution = Distribution()
# ... populate distribution ...
analyze_distribution(latency_distribution)