Querying Oracle V$SESSION for resource usage #

import asyncio
import oracledb
import logging
from datetime import datetime, timezone
from typing import Dict, List, Tuple, Any

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(module)s | %(message)s"
)
logger = logging.getLogger("oracle_session_metrics")

class OracleSessionCostExtractor:
    def __init__(
        self,
        dsn: str,
        user: str,
        password: str,
        pool_min: int = 2,
        pool_max: int = 5,
        query_timeout: float = 8.0,
        max_retries: int = 3
    ):
        self.dsn = dsn
        self.user = user
        self.password = password
        self.pool_min = pool_min
        self.pool_max = pool_max
        self.query_timeout = query_timeout
        self.max_retries = max_retries
        self._pool = None
        self._baseline: Dict[str, Dict[str, float]] = {}

    async def initialize_pool(self) -> None:
        try:
            self._pool = await oracledb.create_pool_async(
                dsn=self.dsn,
                user=self.user,
                password=self.password,
                min=self.pool_min,
                max=self.pool_max,
                increment=1,
                timeout=30,
                wait_timeout=10
            )
            logger.info("Connection pool initialized successfully.")
        except oracledb.DatabaseError as e:
            logger.critical(f"Failed to initialize connection pool: {e}")
            raise

    async def _fetch_session_metrics(self) -> List[Tuple[str, str, int, int, int, int, int]]:
        query = """
            SELECT
                s.SID || '-' || s.SERIAL# AS SESSION_KEY,
                s.USERNAME,
                NVL(tm.VALUE, 0)        AS CPU_TIME,
                NVL(p.PGA_USED_MEM, 0)  AS PGA_USED_MEM,
                NVL(temp.VALUE, 0)      AS TEMP_SPACE_ALLOCATED,
                NVL(io.CONSISTENT_GETS + io.DB_BLOCK_GETS, 0) AS LOGICAL_READS,
                NVL(io.PHYSICAL_READS, 0) AS PHYSICAL_READS
            FROM V$SESSION s
            JOIN V$PROCESS p
              ON s.PADDR = p.ADDR
            LEFT JOIN V$SESS_IO io
              ON s.SID = io.SID
            LEFT JOIN V$SESS_TIME_MODEL tm
              ON s.SID = tm.SID
             AND tm.STAT_NAME = 'DB CPU'
            LEFT JOIN (
                SELECT st.SID, st.VALUE
                FROM V$SESSTAT st
                JOIN V$STATNAME sn
                  ON st.STATISTIC# = sn.STATISTIC#
                WHERE sn.NAME = 'temp space allocated (bytes)'
            ) temp
              ON s.SID = temp.SID
            WHERE s.TYPE = 'USER'
              AND s.USERNAME IS NOT NULL
              AND s.STATUS = 'ACTIVE'
        """
        async with self._pool.acquire() as conn:
            async with conn.cursor() as cur:
                await cur.execute(query)
                return await cur.fetchall()

    def _compute_deltas(self, current_metrics: List[Tuple[str, str, int, int, int, int, int]]) -> List[Dict[str, Any]]:
        cost_events = []
        for row in current_metrics:
            key, username, cpu, pga, temp, logical, physical = row
            baseline = self._baseline.get(key, {})
            
            # Handle session resets / negative deltas
            cpu_delta = max(0, cpu - baseline.get("cpu", 0))
            pga_delta = max(0, pga - baseline.get("pga", 0))
            temp_delta = max(0, temp - baseline.get("temp", 0))
            logical_delta = max(0, logical - baseline.get("logical", 0))
            physical_delta = max(0, physical - baseline.get("physical", 0))

            # FinOps cost mapping (example: normalized compute units)
            cost_units = (
                (cpu_delta / 1_000_000 * 0.002) +
                (logical_delta / 10_000 * 0.0005) +
                (physical_delta / 1_000 * 0.005) +
                (temp_delta / 1073741824 * 0.01)
            )

            cost_events.append({
                "timestamp": datetime.now(timezone.utc).isoformat(),
                "session_key": key,
                "username": username,
                "cpu_us": cpu_delta,
                "pga_bytes": pga_delta,
                "temp_bytes": temp_delta,
                "logical_reads": logical_delta,
                "physical_reads": physical_delta,
                "cost_units": round(cost_units, 6)
            })

            # Update baseline
            self._baseline[key] = {
                "cpu": cpu, "pga": pga, "temp": temp,
                "logical": logical, "physical": physical
            }
        return cost_events

    def _enforce_quotas(self, cost_events: List[Dict[str, Any]], quota_limit: float = 1.0) -> List[Dict[str, Any]]:
        for event in cost_events:
            if event["cost_units"] > quota_limit:
                logger.warning(
                    f"Quota breach detected: {event['session_key']} "
                    f"consumed {event['cost_units']:.4f} units (limit: {quota_limit})"
                )
                event["quota_exceeded"] = True
            else:
                event["quota_exceeded"] = False
        return cost_events

    async def extract_and_process(self, quota_limit: float = 1.0) -> List[Dict[str, Any]]:
        if not self._pool:
            await self.initialize_pool()

        retries = 0
        while retries <= self.max_retries:
            try:
                metrics = await self._fetch_session_metrics()
                deltas = self._compute_deltas(metrics)
                return self._enforce_quotas(deltas, quota_limit)
            except oracledb.DatabaseError as e:
                retries += 1
                backoff = min(2 ** retries, 30)
                logger.warning(f"Query failed (attempt {retries}/{self.max_retries}). Retrying in {backoff}s. Error: {e}")
                await asyncio.sleep(backoff)
            except Exception as e:
                logger.error(f"Unexpected extraction failure: {e}")
                break
        logger.error("Max retries exceeded. Returning empty dataset.")
        return []