Database Operations and Performance - GCP Professional Cloud Database Engineer¶

import sqlalchemy
from sqlalchemy.pool import NullPool

# Production connection pool configuration
def create_db_pool():
    return sqlalchemy.create_engine(
        sqlalchemy.engine.url.URL.create(
            drivername="postgresql+pg8000",
            username="db_user",
            password="db_password",
            host="10.0.0.3",  # Private IP
            port=5432,
            database="production_db"
        ),
        # Connection pool settings
        pool_size=5,  # Connections to keep open
        max_overflow=10,  # Additional connections under load
        pool_timeout=30,  # Seconds to wait for connection
        pool_recycle=1800,  # Recycle connections after 30 min
        pool_pre_ping=True,  # Verify connections before use
    )

# For serverless environments (Cloud Functions, Cloud Run)
def create_serverless_pool():
    return sqlalchemy.create_engine(
        connection_url,
        pool_size=1,  # Minimize connections per instance
        max_overflow=0,
        pool_timeout=30,
        pool_pre_ping=True,
        poolclass=NullPool  # No pooling for short-lived instances
    )

from google.cloud import sql_v1

def analyze_query_insights(project_id, instance_id):
    """Retrieve and analyze query insights"""
    client = sql_v1.SqlInstancesServiceClient()

    # Get query insights
    request = sql_v1.ListQueryInsightsRequest(
        project=project_id,
        instance=instance_id,
        start_time="2024-01-01T00:00:00Z",
        end_time="2024-01-02T00:00:00Z"
    )

    insights = client.list_query_insights(request=request)

    # Identify problematic queries
    slow_queries = [
        q for q in insights
        if q.avg_latency_seconds > 1.0  # Queries > 1 second
    ]

    return slow_queries

from google.cloud import spanner_admin_instance_v1
from google.cloud import monitoring_v3
import os

def scale_spanner(event, context):
    """Auto-scale Spanner based on CPU metrics"""
    instance_id = os.environ['INSTANCE_ID']
    project_id = os.environ['GCP_PROJECT']

    # Get current CPU utilization
    client = monitoring_v3.MetricServiceClient()
    project_name = f"projects/{project_id}"

    query = f"""
    fetch spanner_instance
    | metric 'spanner.googleapis.com/instance/cpu/utilization'
    | filter resource.instance_id == '{instance_id}'
    | group_by 1m, [value_utilization_mean: mean(value.utilization)]
    | within 5m
    """

    results = client.query_time_series(request={
        "name": project_name,
        "query": query
    })

    avg_cpu = calculate_average(results)

    # Get current node count
    instance_admin_client = spanner_admin_instance_v1.InstanceAdminClient()
    instance_name = f"{project_name}/instances/{instance_id}"
    instance = instance_admin_client.get_instance(name=instance_name)
    current_nodes = instance.node_count

    # Scaling logic
    if avg_cpu > 65 and current_nodes < MAX_NODES:
        new_nodes = current_nodes + 1
        scale_instance(instance_name, new_nodes)
        print(f"Scaled up from {current_nodes} to {new_nodes} nodes")
    elif avg_cpu < 35 and current_nodes > MIN_NODES:
        new_nodes = current_nodes - 1
        scale_instance(instance_name, new_nodes)
        print(f"Scaled down from {current_nodes} to {new_nodes} nodes")

from google.cloud import spanner_admin_database_v1
from datetime import datetime, timedelta

def create_scheduled_backup(instance_id, database_id, project_id):
    """Create backup with naming convention and retention"""
    client = spanner_admin_database_v1.DatabaseAdminClient()

    # Backup naming: db-YYYYMMDD-HHMM
    backup_id = f"{database_id}-{datetime.now().strftime('%Y%m%d-%H%M')}"

    # Create backup
    instance_path = client.instance_path(project_id, instance_id)
    database_path = client.database_path(project_id, instance_id, database_id)
    backup_path = f"{instance_path}/backups/{backup_id}"

    expire_time = datetime.now() + timedelta(days=30)

    backup = spanner_admin_database_v1.Backup(
        database=database_path,
        expire_time=expire_time,
    )

    operation = client.create_backup(
        parent=instance_path,
        backup_id=backup_id,
        backup=backup
    )

    print(f"Creating backup {backup_id}...")
    backup = operation.result(timeout=3600)
    print(f"Backup created: {backup.name}")

    # Cleanup old backups (retain last 30)
    delete_old_backups(client, instance_path, 30)

def delete_old_backups(client, instance_path, retain_count):
    """Delete backups beyond retention count"""
    backups = client.list_backups(parent=instance_path)
    backup_list = sorted(backups, key=lambda b: b.create_time, reverse=True)

    for backup in backup_list[retain_count:]:
        client.delete_backup(name=backup.name)
        print(f"Deleted old backup: {backup.name}")

from google.cloud import bigtable
from google.cloud.bigtable import row_filters
import hashlib

# Bad: Sequential keys (creates hotspots)
def bad_row_key(user_id, timestamp):
    return f"{timestamp}#{user_id}"  # All writes go to same tablet

# Good: Hash-based key distribution
def good_row_key(user_id, timestamp):
    # Hash prefix distributes writes
    hash_prefix = hashlib.md5(user_id.encode()).hexdigest()[:4]
    return f"{hash_prefix}#{user_id}#{timestamp}"

# Good: Reverse domain for web URLs
def url_row_key(url):
    # com.example.www instead of www.example.com
    parts = url.split('.')
    return '.'.join(reversed(parts))

# Good: Salting for time-series data
def timeseries_row_key(sensor_id, timestamp):
    # Distribute across multiple tablets
    salt = hash(sensor_id) % 100  # 100 buckets
    return f"{salt:02d}#{sensor_id}#{timestamp}"

# Query pattern example
def read_timeseries_data(table, sensor_id, start_time, end_time):
    """Read time-series with salted keys"""
    rows = []
    for salt in range(100):
        prefix = f"{salt:02d}#{sensor_id}#"
        row_range = row_range.StartingRowKey(prefix)
        # Read from each salt bucket
        partial_rows = table.read_rows(row_range=row_range)
        rows.extend(partial_rows)
    return rows

from google.cloud import bigtable

def configure_column_families():
    """Configure column families with appropriate GC policies"""
    client = bigtable.Client(project=PROJECT_ID, admin=True)
    instance = client.instance(INSTANCE_ID)
    table = instance.table(TABLE_ID)

    # Create column families with different GC policies

    # 1. Max versions (keep last N versions)
    cf_versions = table.column_family(
        'profile',
        max_versions_gc_rule=bigtable.column_family.MaxVersionsGCRule(3)
    )

    # 2. Max age (keep data for N days)
    cf_metrics = table.column_family(
        'metrics',
        max_age_gc_rule=bigtable.column_family.MaxAgeGCRule(
            datetime.timedelta(days=7)
        )
    )

    # 3. Union (keep if ANY condition matches)
    union_rule = bigtable.column_family.GCRuleUnion([
        bigtable.column_family.MaxVersionsGCRule(2),
        bigtable.column_family.MaxAgeGCRule(datetime.timedelta(days=30))
    ])
    cf_events = table.column_family('events', gc_rule=union_rule)

    # 4. Intersection (keep only if ALL conditions match)
    intersection_rule = bigtable.column_family.GCRuleIntersection([
        bigtable.column_family.MaxVersionsGCRule(5),
        bigtable.column_family.MaxAgeGCRule(datetime.timedelta(days=90))
    ])
    cf_archive = table.column_family('archive', gc_rule=intersection_rule)

    # Create table with column families
    if not table.exists():
        table.create(column_families={
            'profile': cf_versions,
            'metrics': cf_metrics,
            'events': cf_events,
            'archive': cf_archive
        })

from google.cloud import bigtable
from google.cloud.bigtable import row_filters

def bulk_write_data(table, data_items):
    """Efficiently write large batches of data"""
    rows = []

    for item in data_items:
        row_key = f"{item['id']}#{item['timestamp']}".encode()
        row = table.direct_row(row_key)

        # Set multiple cells
        row.set_cell('cf1', 'data', item['data'])
        row.set_cell('cf1', 'status', item['status'])
        row.set_cell('cf1', 'metadata', item['metadata'])

        rows.append(row)

        # Batch in groups of 100,000 for optimal performance
        if len(rows) >= 100000:
            table.mutate_rows(rows)
            rows = []

    # Write remaining rows
    if rows:
        table.mutate_rows(rows)

def bulk_read_with_filter(table, start_key, end_key):
    """Efficiently read with row filters"""
    # Create row range
    row_range = row_range.RowRange(
        start_key=start_key.encode(),
        end_key=end_key.encode()
    )

    # Apply filters for efficiency
    filter = row_filters.RowFilterChain([
        # Only return latest version
        row_filters.CellsPerColumnLimitFilter(1),
        # Only return specific column
        row_filters.ColumnQualifierRegexFilter(b'data'),
        # Apply value filter
        row_filters.ValueRegexFilter(b'active')
    ])

    # Read with filter
    partial_rows = table.read_rows(
        row_range=row_range,
        filter_=filter
    )

    return list(partial_rows)

from google.cloud import monitoring_v3
from datetime import datetime, timedelta

def monitor_bigtable_metrics(project_id, instance_id):
    """Monitor key Bigtable performance metrics"""
    client = monitoring_v3.MetricServiceClient()
    project_name = f"projects/{project_id}"

    # Define metrics to monitor
    metrics = [
        'bigtable.googleapis.com/cluster/cpu_load',
        'bigtable.googleapis.com/cluster/storage_utilization',
        'bigtable.googleapis.com/server/request_count',
        'bigtable.googleapis.com/server/latencies',
    ]

    end_time = datetime.utcnow()
    start_time = end_time - timedelta(hours=1)

    interval = monitoring_v3.TimeInterval({
        "start_time": {"seconds": int(start_time.timestamp())},
        "end_time": {"seconds": int(end_time.timestamp())},
    })

    for metric_type in metrics:
        results = client.list_time_series(
            request={
                "name": project_name,
                "filter": f'metric.type="{metric_type}" AND resource.instance_id="{instance_id}"',
                "interval": interval,
                "view": monitoring_v3.ListTimeSeriesRequest.TimeSeriesView.FULL,
            }
        )

        for result in results:
            print(f"Metric: {metric_type}")
            for point in result.points:
                print(f"  Value: {point.value.double_value}, Time: {point.interval.end_time}")

from google.cloud import bigtable_admin_v2
from datetime import datetime, timedelta

def create_automated_backup(project_id, instance_id, cluster_id, table_id):
    """Create automated backup with retention management"""
    client = bigtable_admin_v2.BigtableTableAdminClient()

    # Create backup
    cluster_path = client.cluster_path(project_id, instance_id, cluster_id)
    table_path = client.table_path(project_id, instance_id, table_id)

    backup_id = f"{table_id}-{datetime.now().strftime('%Y%m%d-%H%M')}"
    expire_time = datetime.now() + timedelta(days=30)

    backup = bigtable_admin_v2.Backup(
        source_table=table_path,
        expire_time=expire_time
    )

    operation = client.create_backup(
        parent=cluster_path,
        backup_id=backup_id,
        backup=backup
    )

    print(f"Creating backup: {backup_id}")
    result = operation.result(timeout=3600)
    print(f"Backup created: {result.name}")

    # Cleanup old backups
    cleanup_old_backups(client, cluster_path, retain_count=30)

def cleanup_old_backups(client, cluster_path, retain_count):
    """Delete backups beyond retention count"""
    backups = client.list_backups(parent=cluster_path)
    backup_list = sorted(
        backups,
        key=lambda b: b.start_time,
        reverse=True
    )

    for backup in backup_list[retain_count:]:
        client.delete_backup(name=backup.name)
        print(f"Deleted old backup: {backup.name}")

from google.cloud import monitoring_dashboard_v1

def create_database_monitoring_dashboard(project_id):
    """Create unified database monitoring dashboard"""
    client = monitoring_dashboard_v1.DashboardsServiceClient()
    project_name = f"projects/{project_id}"

    dashboard = monitoring_dashboard_v1.Dashboard(
        display_name="Database Operations Dashboard",
        grid_layout=monitoring_dashboard_v1.GridLayout(
            widgets=[
                # Cloud SQL CPU
                monitoring_dashboard_v1.Widget(
                    title="Cloud SQL CPU Utilization",
                    xy_chart=monitoring_dashboard_v1.XyChart(
                        data_sets=[
                            monitoring_dashboard_v1.XyChart.DataSet(
                                time_series_query=monitoring_dashboard_v1.TimeSeriesQuery(
                                    time_series_filter=monitoring_dashboard_v1.TimeSeriesFilter(
                                        filter='metric.type="cloudsql.googleapis.com/database/cpu/utilization"',
                                        aggregation=monitoring_dashboard_v1.Aggregation(
                                            alignment_period={"seconds": 60},
                                            per_series_aligner="ALIGN_MEAN",
                                        ),
                                    ),
                                ),
                            ),
                        ],
                        y_axis=monitoring_dashboard_v1.XyChart.Axis(
                            label="CPU %",
                            scale="LINEAR",
                        ),
                    ),
                ),
                # Spanner CPU
                monitoring_dashboard_v1.Widget(
                    title="Spanner High Priority CPU",
                    xy_chart=monitoring_dashboard_v1.XyChart(
                        data_sets=[
                            monitoring_dashboard_v1.XyChart.DataSet(
                                time_series_query=monitoring_dashboard_v1.TimeSeriesQuery(
                                    time_series_filter=monitoring_dashboard_v1.TimeSeriesFilter(
                                        filter='metric.type="spanner.googleapis.com/instance/cpu/utilization_by_priority" metric.label.priority="high"',
                                        aggregation=monitoring_dashboard_v1.Aggregation(
                                            alignment_period={"seconds": 60},
                                            per_series_aligner="ALIGN_MEAN",
                                        ),
                                    ),
                                ),
                            ),
                        ],
                    ),
                ),
                # Bigtable CPU
                monitoring_dashboard_v1.Widget(
                    title="Bigtable CPU Load",
                    xy_chart=monitoring_dashboard_v1.XyChart(
                        data_sets=[
                            monitoring_dashboard_v1.XyChart.DataSet(
                                time_series_query=monitoring_dashboard_v1.TimeSeriesQuery(
                                    time_series_filter=monitoring_dashboard_v1.TimeSeriesFilter(
                                        filter='metric.type="bigtable.googleapis.com/cluster/cpu_load"',
                                        aggregation=monitoring_dashboard_v1.Aggregation(
                                            alignment_period={"seconds": 60},
                                            per_series_aligner="ALIGN_MEAN",
                                        ),
                                    ),
                                ),
                            ),
                        ],
                    ),
                ),
                # Firestore Operations
                monitoring_dashboard_v1.Widget(
                    title="Firestore Operations/sec",
                    xy_chart=monitoring_dashboard_v1.XyChart(
                        data_sets=[
                            monitoring_dashboard_v1.XyChart.DataSet(
                                time_series_query=monitoring_dashboard_v1.TimeSeriesQuery(
                                    time_series_filter=monitoring_dashboard_v1.TimeSeriesFilter(
                                        filter='metric.type="firestore.googleapis.com/document/read_count"',
                                        aggregation=monitoring_dashboard_v1.Aggregation(
                                            alignment_period={"seconds": 60},
                                            per_series_aligner="ALIGN_RATE",
                                        ),
                                    ),
                                ),
                            ),
                        ],
                    ),
                ),
            ],
        ),
    )

    response = client.create_dashboard(
        parent=project_name,
        dashboard=dashboard
    )

    print(f"Dashboard created: {response.name}")
    return response

from google.cloud import monitoring_v3

def create_database_alerts(project_id):
    """Create critical database alert policies"""
    client = monitoring_v3.AlertPolicyServiceClient()
    project_name = f"projects/{project_id}"

    # Cloud SQL High CPU Alert
    cloudsql_cpu_policy = monitoring_v3.AlertPolicy(
        display_name="Cloud SQL High CPU (>80%)",
        conditions=[
            monitoring_v3.AlertPolicy.Condition(
                display_name="CPU > 80% for 5 minutes",
                condition_threshold=monitoring_v3.AlertPolicy.Condition.MetricThreshold(
                    filter='metric.type="cloudsql.googleapis.com/database/cpu/utilization"',
                    comparison=monitoring_v3.ComparisonType.COMPARISON_GT,
                    threshold_value=0.8,
                    duration={"seconds": 300},
                    aggregations=[
                        monitoring_v3.Aggregation(
                            alignment_period={"seconds": 60},
                            per_series_aligner="ALIGN_MEAN",
                        ),
                    ],
                )
            )
        ],
        alert_strategy=monitoring_v3.AlertPolicy.AlertStrategy(
            auto_close={"seconds": 1800}
        ),
        combiner=monitoring_v3.AlertPolicy.ConditionCombinerType.AND,
    )

    # Spanner High Priority CPU Alert
    spanner_cpu_policy = monitoring_v3.AlertPolicy(
        display_name="Spanner High Priority CPU (>65%)",
        conditions=[
            monitoring_v3.AlertPolicy.Condition(
                display_name="High Priority CPU > 65%",
                condition_threshold=monitoring_v3.AlertPolicy.Condition.MetricThreshold(
                    filter='metric.type="spanner.googleapis.com/instance/cpu/utilization_by_priority" metric.label.priority="high"',
                    comparison=monitoring_v3.ComparisonType.COMPARISON_GT,
                    threshold_value=0.65,
                    duration={"seconds": 300},
                    aggregations=[
                        monitoring_v3.Aggregation(
                            alignment_period={"seconds": 60},
                            per_series_aligner="ALIGN_MEAN",
                        ),
                    ],
                )
            )
        ],
        combiner=monitoring_v3.AlertPolicy.ConditionCombinerType.AND,
    )

    # Cloud SQL Replication Lag Alert
    replication_lag_policy = monitoring_v3.AlertPolicy(
        display_name="Cloud SQL High Replication Lag",
        conditions=[
            monitoring_v3.AlertPolicy.Condition(
                display_name="Replication lag > 60 seconds",
                condition_threshold=monitoring_v3.AlertPolicy.Condition.MetricThreshold(
                    filter='metric.type="cloudsql.googleapis.com/database/replication/replica_lag"',
                    comparison=monitoring_v3.ComparisonType.COMPARISON_GT,
                    threshold_value=60,
                    duration={"seconds": 180},
                ),
            )
        ],
        combiner=monitoring_v3.AlertPolicy.ConditionCombinerType.AND,
    )

    # Bigtable Storage Utilization
    bigtable_storage_policy = monitoring_v3.AlertPolicy(
        display_name="Bigtable High Storage Utilization (>80%)",
        conditions=[
            monitoring_v3.AlertPolicy.Condition(
                display_name="Storage > 80% of capacity",
                condition_threshold=monitoring_v3.AlertPolicy.Condition.MetricThreshold(
                    filter='metric.type="bigtable.googleapis.com/cluster/storage_utilization"',
                    comparison=monitoring_v3.ComparisonType.COMPARISON_GT,
                    threshold_value=0.8,
                    duration={"seconds": 600},
                ),
            )
        ],
        combiner=monitoring_v3.AlertPolicy.ConditionCombinerType.AND,
    )

    # Create alert policies
    policies = [
        cloudsql_cpu_policy,
        spanner_cpu_policy,
        replication_lag_policy,
        bigtable_storage_policy,
    ]

    for policy in policies:
        created = client.create_alert_policy(
            name=project_name,
            alert_policy=policy
        )
        print(f"Created alert policy: {created.display_name}")

from google.cloud import sql_v1
import time

def test_ha_failover(project_id, instance_id):
    """Test HA failover and measure RTO"""
    client = sql_v1.SqlInstancesServiceClient()
    instance_path = client.instance_path(project_id, instance_id)

    print("Initiating failover...")
    start_time = time.time()

    # Trigger failover
    operation = client.failover(instance=instance_path)

    # Wait for completion
    operation.result(timeout=600)

    # Measure RTO
    end_time = time.time()
    rto_seconds = end_time - start_time

    print(f"Failover completed in {rto_seconds:.2f} seconds")

    # Verify instance is healthy
    instance = client.get(name=instance_path)
    print(f"Instance state: {instance.state}")
    print(f"Availability type: {instance.settings.availability_type}")

    return rto_seconds

def resolve_cloudsql_performance(project_id, instance_id):
    """Automated performance issue resolution"""
    from google.cloud import sql_v1, monitoring_v3

    sql_client = sql_v1.SqlInstancesServiceClient()
    monitoring_client = monitoring_v3.MetricServiceClient()

    instance_path = sql_client.instance_path(project_id, instance_id)
    instance = sql_client.get(name=instance_path)

    # Check CPU utilization
    cpu_query = f'''
    fetch cloudsql_database
    | metric cloudsql.googleapis.com/database/cpu/utilization
    | filter resource.database_id == "{project_id}:{instance_id}"
    | within 1h
    '''

    results = monitoring_client.query_time_series(
        name=f"projects/{project_id}",
        query=cpu_query
    )

    avg_cpu = calculate_average_cpu(results)

    # Decision logic
    if avg_cpu > 0.8:
        print("HIGH CPU: Consider scaling up tier or optimizing queries")
        print(f"Current tier: {instance.settings.tier}")
        print("Recommendation: Increase to next tier or add read replicas")

    # Check slow queries via Query Insights
    # Recommend indexes for top slow queries

    return recommendations

# Before (Hotspot):
def old_row_key(user_id, timestamp):
    return f"{timestamp}#{user_id}"  # Sequential timestamps create hotspots

# After (Distributed):
def new_row_key(user_id, timestamp):
    # Add hash prefix for distribution
    import hashlib
    hash_prefix = hashlib.md5(user_id.encode()).hexdigest()[:4]
    # Reverse timestamp for range scans
    reverse_ts = 9999999999 - timestamp
    return f"{hash_prefix}#{user_id}#{reverse_ts}"

# Migration strategy:
# 1. Write to both old and new key formats
# 2. Backfill historical data
# 3. Update read paths to check both formats
# 4. Remove old key format after validation

import sqlalchemy
from sqlalchemy.pool import QueuePool

# Problem: Each Cloud Run instance creates connections
# With 100 instances * 5 connections = 500 connections
# Exceeds Cloud SQL connection limit

# Solution 1: Reduce pool size for serverless
engine = sqlalchemy.create_engine(
    connection_url,
    pool_size=1,  # Minimal pool for serverless
    max_overflow=0,
    pool_timeout=30,
    pool_pre_ping=True
)

# Solution 2: Increase Cloud SQL max_connections
# gcloud sql instances patch INSTANCE \
#   --database-flags=max_connections=500

# Solution 3: Use Cloud SQL Proxy with connection limiting
# ./cloud-sql-proxy --max-connections=400 PROJECT:REGION:INSTANCE

# Solution 4: Implement connection retry logic
def execute_with_retry(query, max_retries=3):
    for attempt in range(max_retries):
        try:
            with engine.connect() as conn:
                return conn.execute(query)
        except sqlalchemy.exc.OperationalError as e:
            if "too many connections" in str(e):
                time.sleep(2 ** attempt)  # Exponential backoff
                continue
            raise