Extracting pg_stat_activity for cost tracking #

SELECT 
    pid,
    usename,
    application_name,
    client_addr,
    state,
    backend_type,
    wait_event_type,
    wait_event,
    EXTRACT(EPOCH FROM (now() - query_start)) AS active_seconds,
    current_setting('app.cost_center', true) AS cost_center_tag
FROM pg_stat_activity
WHERE backend_type = 'client backend'
  AND state != 'idle'
  AND pid != pg_backend_pid()
  AND query_start IS NOT NULL;

import asyncio
import asyncpg
import psycopg2
import psycopg2.extras
import logging
from datetime import datetime, timezone
from typing import List, Dict, Any, Optional
import time
import os

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s")
logger = logging.getLogger("pg_cost_extractor")

class ActivityRecordValidator:
    """Lightweight schema validation for billing pipeline ingestion."""
    REQUIRED_FIELDS = {"pid", "usename", "application_name", "state", "active_seconds"}
    
    @classmethod
    def validate(cls, records: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        valid_records = []
        for record in records:
            if cls.REQUIRED_FIELDS.issubset(record.keys()):
                # Enforce numeric type for active_seconds
                try:
                    record["active_seconds"] = float(record["active_seconds"])
                    valid_records.append(record)
                except (ValueError, TypeError):
                    logger.warning("Dropping record with invalid active_seconds: pid=%s", record.get("pid"))
            else:
                logger.warning("Schema validation failed for record: %s", record)
        return valid_records

async def fetch_activity_async(pool: asyncpg.Pool) -> List[Dict[str, Any]]:
    """Execute optimized pg_stat_activity query with asyncpg."""
    query = """
    SELECT 
        pid, usename, application_name, client_addr, state, backend_type,
        wait_event_type, wait_event,
        EXTRACT(EPOCH FROM (now() - query_start)) AS active_seconds,
        current_setting('app.cost_center', true) AS cost_center_tag
    FROM pg_stat_activity
    WHERE backend_type = 'client backend'
      AND state != 'idle'
      AND pid != pg_backend_pid()
      AND query_start IS NOT NULL;
    """
    async with pool.acquire() as conn:
        rows = await conn.fetch(query)
        return [dict(row) for row in rows]

def fetch_activity_sync(dsn: str) -> List[Dict[str, Any]]:
    """Synchronous psycopg2 fallback for degraded async environments."""
    query = """
    SELECT 
        pid, usename, application_name, client_addr, state, backend_type,
        wait_event_type, wait_event,
        EXTRACT(EPOCH FROM (now() - query_start)) AS active_seconds,
        current_setting('app.cost_center', true) AS cost_center_tag
    FROM pg_stat_activity
    WHERE backend_type = 'client backend'
      AND state != 'idle'
      AND pid != pg_backend_pid()
      AND query_start IS NOT NULL;
    """
    with psycopg2.connect(dsn) as conn:
        with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
            cur.execute(query)
            return cur.fetchall()

async def extract_with_retry(dsn: str, max_retries: int = 3, base_delay: float = 1.0) -> List[Dict[str, Any]]:
    """Production extractor with exponential backoff and driver fallback."""
    pool = None
    for attempt in range(max_retries):
        try:
            if pool is None:
                pool = await asyncpg.create_pool(dsn, min_size=2, max_size=5, timeout=10)
            records = await fetch_activity_async(pool)
            logger.info("Async extraction successful: %d records", len(records))
            return ActivityRecordValidator.validate(records)
        except (asyncpg.exceptions.ConnectionError, asyncio.TimeoutError) as e:
            logger.warning("Async attempt %d failed: %s", attempt + 1, e)
            if attempt == max_retries - 1:
                logger.info("Falling back to synchronous psycopg2 extraction")
                sync_records = fetch_activity_sync(dsn)
                return ActivityRecordValidator.validate(sync_records)
            delay = base_delay * (2 ** attempt)
            await asyncio.sleep(delay)
    return []

async def main():
    dsn = os.getenv("PG_DSN", "postgresql://user:pass@localhost:5432/db")
    records = await extract_with_retry(dsn)
    logger.info("Validated payload ready for billing routing: %d records", len(records))
    # Downstream routing to Kafka, S3, or FinOps aggregator would occur here

if __name__ == "__main__":
    asyncio.run(main())