Monitoring and Deployment - GCP Professional Cloud Developer¶

from google.cloud import monitoring_v3
import time
from typing import Dict, Any

class CloudMonitoring:
    def __init__(self, project_id: str):
        self.project_id = project_id
        self.client = monitoring_v3.MetricServiceClient()
        self.project_name = f"projects/{project_id}"

    def write_custom_metric(self, metric_type: str, value: float,
                           labels: Dict[str, str] = None):
        """Write custom metric to Cloud Monitoring"""
        series = monitoring_v3.TimeSeries()
        series.metric.type = f"custom.googleapis.com/{metric_type}"
        series.resource.type = "global"

        # Add labels
        if labels:
            for key, val in labels.items():
                series.metric.labels[key] = val

        # Add point
        point = series.points.add()
        point.value.double_value = value
        now = time.time()
        point.interval.end_time.seconds = int(now)
        point.interval.end_time.nanos = int((now - int(now)) * 10**9)

        self.client.create_time_series(
            name=self.project_name,
            time_series=[series]
        )

    def write_business_metric(self, metric_name: str, value: float,
                             user_id: str = None, region: str = None):
        """Write business KPI metric"""
        labels = {}
        if user_id:
            labels['user_id'] = user_id
        if region:
            labels['region'] = region

        self.write_custom_metric(
            metric_type=f"business/{metric_name}",
            value=value,
            labels=labels
        )

    def record_request_latency(self, endpoint: str, latency_ms: float,
                               status_code: int):
        """Record API endpoint latency"""
        self.write_custom_metric(
            metric_type="api/request_latency",
            value=latency_ms,
            labels={
                'endpoint': endpoint,
                'status_code': str(status_code)
            }
        )

    def record_queue_depth(self, queue_name: str, depth: int):
        """Record queue depth for monitoring"""
        self.write_custom_metric(
            metric_type="queue/depth",
            value=float(depth),
            labels={'queue_name': queue_name}
        )

# Usage in application
monitoring = CloudMonitoring(project_id="my-project")

# Record business metrics
monitoring.write_business_metric(
    metric_name="order_total",
    value=299.99,
    user_id="user123",
    region="us-central1"
)

# Record API latency
monitoring.record_request_latency(
    endpoint="/api/products",
    latency_ms=125.5,
    status_code=200
)

from google.cloud import monitoring_v3
import psutil
import threading
import time

class ResourceMonitor:
    def __init__(self, project_id: str, interval: int = 60):
        self.project_id = project_id
        self.client = monitoring_v3.MetricServiceClient()
        self.interval = interval
        self.running = False

    def start_monitoring(self):
        """Start background resource monitoring"""
        self.running = True
        thread = threading.Thread(target=self._monitor_loop)
        thread.daemon = True
        thread.start()

    def _monitor_loop(self):
        while self.running:
            self._record_resources()
            time.sleep(self.interval)

    def _record_resources(self):
        # CPU usage
        cpu_percent = psutil.cpu_percent(interval=1)
        self._write_metric("system/cpu_usage", cpu_percent)

        # Memory usage
        memory = psutil.virtual_memory()
        self._write_metric("system/memory_usage", memory.percent)

        # Disk usage
        disk = psutil.disk_usage('/')
        self._write_metric("system/disk_usage", disk.percent)

    def _write_metric(self, metric_type: str, value: float):
        series = monitoring_v3.TimeSeries()
        series.metric.type = f"custom.googleapis.com/{metric_type}"
        series.resource.type = "gce_instance"

        point = series.points.add()
        point.value.double_value = value
        point.interval.end_time.seconds = int(time.time())

        try:
            self.client.create_time_series(
                name=f"projects/{self.project_id}",
                time_series=[series]
            )
        except Exception as e:
            print(f"Failed to write metric: {e}")

# Start monitoring
monitor = ResourceMonitor(project_id="my-project", interval=60)
monitor.start_monitoring()

from google.cloud import trace_v1
from google.cloud.trace_v1 import types
import time
import uuid
from contextlib import contextmanager

class CloudTracer:
    def __init__(self, project_id: str, service_name: str):
        self.project_id = project_id
        self.service_name = service_name
        self.client = trace_v1.TraceServiceClient()
        self.project_name = f"projects/{project_id}"

    @contextmanager
    def span(self, span_name: str, parent_span_id: str = None):
        """Context manager for creating trace spans"""
        trace_id = str(uuid.uuid4().hex)
        span_id = str(uuid.uuid4().int)[:16]

        start_time = time.time()

        try:
            yield {
                'trace_id': trace_id,
                'span_id': span_id
            }
        finally:
            end_time = time.time()
            self._write_span(
                trace_id=trace_id,
                span_id=span_id,
                span_name=span_name,
                start_time=start_time,
                end_time=end_time,
                parent_span_id=parent_span_id
            )

    def _write_span(self, trace_id: str, span_id: str, span_name: str,
                    start_time: float, end_time: float,
                    parent_span_id: str = None):
        """Write span to Cloud Trace"""
        span = types.TraceSpan()
        span.span_id = int(span_id)
        span.name = span_name

        # Set timestamps
        span.start_time.seconds = int(start_time)
        span.start_time.nanos = int((start_time - int(start_time)) * 10**9)
        span.end_time.seconds = int(end_time)
        span.end_time.nanos = int((end_time - int(end_time)) * 10**9)

        if parent_span_id:
            span.parent_span_id = int(parent_span_id)

        trace = types.Trace()
        trace.project_id = self.project_id
        trace.trace_id = trace_id
        trace.spans.append(span)

        try:
            self.client.patch_traces(
                project_id=self.project_id,
                traces=types.Traces(traces=[trace])
            )
        except Exception as e:
            print(f"Failed to write trace: {e}")

# Usage in application
tracer = CloudTracer(project_id="my-project", service_name="api-service")

def process_order(order_id: str):
    with tracer.span("process_order") as parent_span:
        # Validate order
        with tracer.span("validate_order", parent_span['span_id']):
            validate_order_details(order_id)

        # Process payment
        with tracer.span("process_payment", parent_span['span_id']):
            process_payment(order_id)

        # Ship order
        with tracer.span("ship_order", parent_span['span_id']):
            ship_order(order_id)

from opentelemetry import trace
from opentelemetry.exporter.cloud_trace import CloudTraceSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.instrumentation.requests import RequestsInstrumentor
from opentelemetry.instrumentation.flask import FlaskInstrumentor

# Set up Cloud Trace exporter
trace.set_tracer_provider(TracerProvider())
cloud_trace_exporter = CloudTraceSpanExporter()
trace.get_tracer_provider().add_span_processor(
    BatchSpanProcessor(cloud_trace_exporter)
)

# Auto-instrument libraries
RequestsInstrumentor().instrument()

# Flask app instrumentation
from flask import Flask
app = Flask(__name__)
FlaskInstrumentor().instrument_app(app)

# Manual instrumentation
tracer = trace.get_tracer(__name__)

@app.route('/api/products/<product_id>')
def get_product(product_id):
    with tracer.start_as_current_span("get_product") as span:
        span.set_attribute("product_id", product_id)

        # Database call
        with tracer.start_as_current_span("db_query"):
            product = db.get_product(product_id)

        # Cache call
        with tracer.start_as_current_span("cache_check"):
            cached = cache.get(f"product:{product_id}")

        span.set_attribute("cache_hit", cached is not None)
        return product

from google.cloud import error_reporting
import logging
import traceback
import sys

class ErrorReporter:
    def __init__(self, project_id: str, service_name: str):
        self.client = error_reporting.Client(project=project_id)
        self.service_name = service_name

    def report_error(self, error: Exception, user_id: str = None,
                    http_context: dict = None):
        """Report error to Cloud Error Reporting"""
        try:
            self.client.report_exception(
                http_context=http_context,
                user=user_id
            )
        except Exception as e:
            logging.error(f"Failed to report error: {e}")

    def report_with_context(self, error: Exception, **context):
        """Report error with custom context"""
        error_message = f"{type(error).__name__}: {str(error)}\n"
        error_message += f"Context: {context}\n"
        error_message += traceback.format_exc()

        try:
            self.client.report(error_message)
        except Exception as e:
            logging.error(f"Failed to report error: {e}")

# Flask integration
from flask import Flask, request, g
import functools

app = Flask(__name__)
error_reporter = ErrorReporter(
    project_id="my-project",
    service_name="api-service"
)

def report_errors(f):
    """Decorator to automatically report errors"""
    @functools.wraps(f)
    def wrapper(*args, **kwargs):
        try:
            return f(*args, **kwargs)
        except Exception as e:
            http_context = {
                'method': request.method,
                'url': request.url,
                'user_agent': request.headers.get('User-Agent'),
                'remote_ip': request.remote_addr
            }

            error_reporter.report_error(
                error=e,
                user_id=g.get('user_id'),
                http_context=http_context
            )
            raise

    return wrapper

@app.route('/api/orders', methods=['POST'])
@report_errors
def create_order():
    # Application logic
    order = process_order_request(request.json)
    return {'order_id': order.id}

import logging
import json
import uuid
from google.cloud import logging as cloud_logging
from datetime import datetime

class StructuredLogger:
    def __init__(self, project_id: str, service_name: str):
        self.client = cloud_logging.Client(project=project_id)
        self.client.setup_logging()
        self.service_name = service_name
        self.logger = logging.getLogger(service_name)

    def log(self, level: str, message: str, **context):
        """Log structured message with context"""
        log_entry = {
            'timestamp': datetime.utcnow().isoformat(),
            'severity': level.upper(),
            'service': self.service_name,
            'message': message,
            **context
        }

        if level == 'info':
            self.logger.info(json.dumps(log_entry))
        elif level == 'warning':
            self.logger.warning(json.dumps(log_entry))
        elif level == 'error':
            self.logger.error(json.dumps(log_entry))
        elif level == 'debug':
            self.logger.debug(json.dumps(log_entry))

    def log_request(self, method: str, path: str, status_code: int,
                   latency_ms: float, user_id: str = None):
        """Log HTTP request with standard fields"""
        self.log('info', 'HTTP request processed',
                method=method,
                path=path,
                status_code=status_code,
                latency_ms=latency_ms,
                user_id=user_id,
                log_type='http_request')

    def log_business_event(self, event_type: str, **details):
        """Log business event"""
        self.log('info', f'Business event: {event_type}',
                event_type=event_type,
                log_type='business_event',
                **details)

# Usage
logger = StructuredLogger(
    project_id="my-project",
    service_name="api-service"
)

logger.log_request(
    method="POST",
    path="/api/orders",
    status_code=201,
    latency_ms=145.2,
    user_id="user123"
)

logger.log_business_event(
    event_type="order_created",
    order_id="order-456",
    total_amount=299.99,
    currency="USD"
)

import uuid
from flask import Flask, request, g
import logging

app = Flask(__name__)

@app.before_request
def before_request():
    """Generate or extract correlation ID"""
    correlation_id = request.headers.get('X-Correlation-ID')
    if not correlation_id:
        correlation_id = str(uuid.uuid4())

    g.correlation_id = correlation_id
    g.request_start_time = time.time()

@app.after_request
def after_request(response):
    """Add correlation ID to response headers"""
    response.headers['X-Correlation-ID'] = g.correlation_id
    return response

class CorrelationFilter(logging.Filter):
    """Add correlation ID to all logs"""
    def filter(self, record):
        record.correlation_id = getattr(g, 'correlation_id', 'no-correlation-id')
        record.user_id = getattr(g, 'user_id', 'anonymous')
        return True

# Configure logging
logger = logging.getLogger(__name__)
logger.addFilter(CorrelationFilter())

# Custom formatter
formatter = logging.Formatter(
    '{"timestamp":"%(asctime)s","severity":"%(levelname)s",'
    '"correlation_id":"%(correlation_id)s","user_id":"%(user_id)s",'
    '"message":"%(message)s"}'
)

@app.route('/api/orders', methods=['POST'])
def create_order():
    logger.info(f"Processing order creation",
               extra={'order_data': request.json})

    try:
        order = process_order(request.json)
        logger.info(f"Order created successfully",
                   extra={'order_id': order.id})
        return {'order_id': order.id}, 201
    except Exception as e:
        logger.error(f"Order creation failed: {e}",
                    extra={'error_type': type(e).__name__})
        raise

from google.cloud import logging_v2

def create_log_based_metrics(project_id: str):
    client = logging_v2.MetricsServiceV2Client()
    project_name = f"projects/{project_id}"

    # Error count metric
    metric = {
        'name': f'{project_name}/metrics/application_errors',
        'description': 'Count of application errors',
        'filter': 'severity>=ERROR',
        'metric_descriptor': {
            'metric_kind': 'DELTA',
            'value_type': 'INT64',
        }
    }

    client.create_log_metric(parent=project_name, metric=metric)

    # Latency distribution metric
    latency_metric = {
        'name': f'{project_name}/metrics/request_latency',
        'description': 'Request latency distribution',
        'filter': 'jsonPayload.latency_ms>0',
        'value_extractor': 'EXTRACT(jsonPayload.latency_ms)',
        'metric_descriptor': {
            'metric_kind': 'DELTA',
            'value_type': 'DISTRIBUTION',
        }
    }

    client.create_log_metric(parent=project_name, metric=latency_metric)

import googleclouddebugger

# Enable Cloud Debugger
googleclouddebugger.enable(
    project_id='my-project',
    project_number='123456789',
    service='api-service',
    version='v1.0.0'
)

# Application code - debugger allows setting breakpoints
# and logpoints without code changes
def process_payment(order_id: str, amount: float):
    # Can set snapshot here to inspect variables
    payment_method = get_payment_method(order_id)

    # Can set logpoint here to log without redeploying
    charge_result = charge_payment(payment_method, amount)

    return charge_result

import googlecloudprofiler
import logging

def setup_profiler(project_id: str, service_name: str, version: str):
    """Initialize Cloud Profiler"""
    try:
        googlecloudprofiler.start(
            service=service_name,
            service_version=version,
            project_id=project_id,
            verbose=3,  # Set to 3 for debugging
            # Profile CPU and heap
            disable_cpu_profiling=False,
            disable_heap_profiling=False,
            # Sampling configuration
            period_ms=1000,  # Sample every second
        )
        logging.info("Cloud Profiler started successfully")
    except Exception as e:
        logging.error(f"Failed to start Cloud Profiler: {e}")

# Start profiler at application startup
setup_profiler(
    project_id="my-project",
    service_name="api-service",
    version="v1.0.0"
)

# CPU-intensive function that will show in profiler
def calculate_recommendations(user_id: str):
    # This function's performance will be visible in Cloud Profiler
    user_data = fetch_user_data(user_id)
    recommendations = []

    for item in user_data.history:
        similarity_scores = calculate_similarity(item)
        recommendations.extend(similarity_scores)

    return sorted(recommendations, reverse=True)[:10]

# tests/unit/test_order_service.py
import pytest
from unittest.mock import Mock, patch
from app.services.order_service import OrderService

@pytest.fixture
def order_service():
    return OrderService()

@pytest.fixture
def mock_db():
    with patch('app.services.order_service.db') as mock:
        yield mock

def test_create_order_success(order_service, mock_db):
    """Test successful order creation"""
    order_data = {
        'user_id': 'user123',
        'items': [{'sku': 'PROD-001', 'quantity': 2}],
        'total': 299.99
    }

    mock_db.insert.return_value = {'id': 'order-456'}

    result = order_service.create_order(order_data)

    assert result['id'] == 'order-456'
    mock_db.insert.assert_called_once()

def test_create_order_invalid_data(order_service):
    """Test order creation with invalid data"""
    invalid_data = {'user_id': 'user123'}  # Missing required fields

    with pytest.raises(ValueError):
        order_service.create_order(invalid_data)

def test_calculate_order_total(order_service):
    """Test order total calculation"""
    items = [
        {'price': 100.00, 'quantity': 2},
        {'price': 50.00, 'quantity': 1}
    ]

    total = order_service.calculate_total(items)

    assert total == 250.00

# tests/integration/test_order_workflow.py
import pytest
import requests
from google.cloud import firestore, pubsub_v1

@pytest.fixture
def api_base_url():
    return "https://myapp-staging-uc.a.run.app"

@pytest.fixture
def db_client():
    return firestore.Client(project='my-project')

@pytest.fixture
def pubsub_client():
    return pubsub_v1.SubscriberClient()

def test_create_order_workflow(api_base_url, db_client):
    """Test complete order creation workflow"""
    # Create order via API
    order_data = {
        'user_id': 'test-user',
        'items': [{'sku': 'TEST-001', 'quantity': 1}],
        'total': 99.99
    }

    response = requests.post(
        f"{api_base_url}/api/orders",
        json=order_data,
        headers={'Authorization': 'Bearer test-token'}
    )

    assert response.status_code == 201
    order_id = response.json()['order_id']

    # Verify order in database
    doc = db_client.collection('orders').document(order_id).get()
    assert doc.exists
    assert doc.to_dict()['user_id'] == 'test-user'

    # Cleanup
    db_client.collection('orders').document(order_id).delete()

def test_order_notification_flow(api_base_url, pubsub_client):
    """Test order creates notification message"""
    subscription_path = pubsub_client.subscription_path(
        'my-project', 'order-notifications-test'
    )

    # Create order
    order_data = {
        'user_id': 'test-user',
        'items': [{'sku': 'TEST-001', 'quantity': 1}],
        'total': 99.99
    }

    response = requests.post(f"{api_base_url}/api/orders", json=order_data)
    order_id = response.json()['order_id']

    # Pull message from Pub/Sub
    response = pubsub_client.pull(
        subscription=subscription_path,
        max_messages=1,
        timeout=10
    )

    assert len(response.received_messages) == 1
    message = response.received_messages[0]
    assert message.message.data.decode('utf-8') == order_id

    # Acknowledge message
    pubsub_client.acknowledge(
        subscription=subscription_path,
        ack_ids=[message.ack_id]
    )

# tests/e2e/test_order_flow.py
from playwright.sync_api import sync_playwright
import pytest

@pytest.fixture
def browser():
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        yield browser
        browser.close()

def test_complete_order_flow(browser):
    """Test complete user order flow"""
    page = browser.new_page()

    # Navigate to app
    page.goto("https://myapp.example.com")

    # Login
    page.click("text=Login")
    page.fill("input[name=email]", "test@example.com")
    page.fill("input[name=password]", "testpass123")
    page.click("button[type=submit]")

    # Wait for dashboard
    page.wait_for_selector("text=Dashboard")

    # Add product to cart
    page.click("text=Products")
    page.click("button[data-product=PROD-001]")
    page.click("text=Add to Cart")

    # Checkout
    page.click("text=Cart")
    page.click("text=Checkout")

    # Fill payment info
    page.fill("input[name=card_number]", "4242424242424242")
    page.fill("input[name=exp_date]", "12/25")
    page.fill("input[name=cvv]", "123")

    # Submit order
    page.click("button[type=submit]")

    # Verify success
    page.wait_for_selector("text=Order confirmed")
    order_id = page.locator("[data-order-id]").inner_text()

    assert len(order_id) > 0

# tests/load/locustfile.py
from locust import HttpUser, task, between
import random

class OrderUser(HttpUser):
    wait_time = between(1, 5)

    def on_start(self):
        """Login user"""
        response = self.client.post("/api/auth/login", json={
            "email": f"user{random.randint(1, 1000)}@example.com",
            "password": "testpass123"
        })
        self.token = response.json()['token']

    @task(3)
    def browse_products(self):
        """Browse products - most common action"""
        self.client.get("/api/products",
                       headers={"Authorization": f"Bearer {self.token}"})

    @task(2)
    def view_product(self):
        """View specific product"""
        product_id = random.choice(['PROD-001', 'PROD-002', 'PROD-003'])
        self.client.get(f"/api/products/{product_id}",
                       headers={"Authorization": f"Bearer {self.token}"})

    @task(1)
    def create_order(self):
        """Create order - less frequent but important"""
        order_data = {
            "items": [
                {
                    "sku": random.choice(['PROD-001', 'PROD-002']),
                    "quantity": random.randint(1, 3)
                }
            ]
        }
        self.client.post("/api/orders",
                        json=order_data,
                        headers={"Authorization": f"Bearer {self.token}"})

# Run: locust -f locustfile.py --host=https://myapp.example.com

# tests/chaos/test_resilience.py
import requests
import time
import random
from google.cloud import compute_v1

def test_instance_failure_resilience():
    """Test app resilience during instance failures"""
    compute_client = compute_v1.InstancesClient()
    project = 'my-project'
    zone = 'us-central1-a'
    instance_group = 'myapp-ig'

    # Get healthy instance count before chaos
    instances = compute_client.list(project=project, zone=zone)
    healthy_count = len([i for i in instances if i.status == 'RUNNING'])

    # Make baseline requests
    success_count = 0
    for _ in range(100):
        try:
            response = requests.get('https://myapp.example.com/health')
            if response.status_code == 200:
                success_count += 1
        except:
            pass

    baseline_success_rate = success_count / 100
    assert baseline_success_rate > 0.95

    # Inject chaos - delete random instance
    instances_list = list(instances)
    victim = random.choice(instances_list)
    compute_client.delete(project=project, zone=zone, instance=victim.name)

    # Continue making requests during chaos
    chaos_success_count = 0
    for _ in range(100):
        try:
            response = requests.get('https://myapp.example.com/health')
            if response.status_code == 200:
                chaos_success_count += 1
        except:
            pass
        time.sleep(0.5)

    chaos_success_rate = chaos_success_count / 100

    # Assert system remains available (>90% success)
    assert chaos_success_rate > 0.90

    # Verify auto-healing
    time.sleep(60)
    instances = compute_client.list(project=project, zone=zone)
    recovered_count = len([i for i in instances if i.status == 'RUNNING'])
    assert recovered_count == healthy_count

from flask import Flask, jsonify
import psutil

app = Flask(__name__)

@app.route('/healthz')
def liveness():
    """Liveness probe - is the application running?"""
    # Check critical dependencies
    try:
        # Verify app can still process requests
        health_status = {
            'status': 'healthy',
            'timestamp': time.time()
        }
        return jsonify(health_status), 200
    except Exception as e:
        return jsonify({
            'status': 'unhealthy',
            'error': str(e)
        }), 503

@app.route('/healthz/detailed')
def detailed_liveness():
    """Detailed health check for debugging"""
    try:
        cpu_percent = psutil.cpu_percent(interval=1)
        memory = psutil.virtual_memory()

        health = {
            'status': 'healthy',
            'timestamp': time.time(),
            'cpu_percent': cpu_percent,
            'memory_percent': memory.percent,
            'memory_available_mb': memory.available / 1024 / 1024
        }

        # Check if system is under extreme load
        if cpu_percent > 95 or memory.percent > 95:
            health['status'] = 'degraded'
            return jsonify(health), 200

        return jsonify(health), 200
    except Exception as e:
        return jsonify({
            'status': 'unhealthy',
            'error': str(e)
        }), 503