Cost Optimization at Scale Questions

# Pseudocode (Python-like)
import time, math, requests, logging
from collections import deque
from datetime import datetime, timedelta

# Config
COST_THRESHOLD = 200.0           # USD/hour considered runaway
SUSTAIN_DURATION = 15*60        # seconds sustained above threshold (15 min)
POLL_INTERVAL = 60              # seconds between cost checks
CHECKPOINT_RETRIES = 5
BACKOFF_BASE = 2                # exponential base
BACKOFF_INITIAL = 5             # seconds
SLACK_WEBHOOK = "https://hooks.slack.com/..."
AUDIT_API = "https://audit.example.com/events"
SAFE_TERMINATION_GRACE = 120    # seconds to let downstream finish

# Helpers
def log_audit(event):
    payload = {"timestamp": now_iso(), **event}
    # best-effort send to audit store (do not block main flow)
    try: requests.post(AUDIT_API, json=payload, timeout=3)
    except: logging.exception("Audit send failed")

def send_slack(msg, channel="#ops"):
    requests.post(SLACK_WEBHOOK, json={"text": msg, "channel": channel})

def now_iso(): return datetime.utcnow().isoformat()+"Z"

# Core monitoring loop for a single job
def monitor_job(job_id, owner_contact):
    window = deque()  # items: (timestamp, cost_rate)
    alerted = False
    while True:
        cost_rate = query_cost_rate(job_id)   # USD/hour from cloud billing / job metrics
        ts = time.time()
        window.append((ts, cost_rate))
        # remove old samples beyond SUSTAIN_DURATION
        cutoff = ts - SUSTAIN_DURATION
        while window and window[0][0] < cutoff:
            window.popleft()

        # compute sustained condition: all samples in window exceed threshold
        sustained = len(window) > 0 and all(r > COST_THRESHOLD for (_, r) in window)
        logging.info(f"{now_iso()} job={job_id} cost_rate={cost_rate} sustained={sustained}")
        log_audit({"job_id": job_id, "event": "cost_sample", "cost_rate": cost_rate})

        if sustained and not alerted:
            msg = f":warning: Runaway detected for job {job_id}: cost_rate={cost_rate:.2f} USD/hr sustained for {SUSTAIN_DURATION//60}m"
            send_slack(msg)
            log_audit({"job_id": job_id, "event": "alert_sent", "message": msg, "owner": owner_contact})
            alerted = True

            # Attempt graceful checkpoint with retries + backoff
            success = False
            for attempt in range(1, CHECKPOINT_RETRIES+1):
                try:
                    resp = requests.post(f"https://api.jobs.example.com/{job_id}/checkpoint", timeout=10)
                    if resp.status_code == 200:
                        success = True
                        log_audit({"job_id": job_id, "event": "checkpoint_success", "attempt": attempt})
                        send_slack(f"Checkpoint succeeded for job {job_id} on attempt {attempt}")
                        break
                    else:
                        logging.warning(f"checkpoint failed code={resp.status_code}")
                        log_audit({"job_id": job_id, "event": "checkpoint_failed", "attempt": attempt, "status": resp.status_code})
                except Exception as e:
                    logging.exception("checkpoint attempt exception")
                    log_audit({"job_id": job_id, "event": "checkpoint_exception", "attempt": attempt, "error": str(e)})

                backoff = BACKOFF_INITIAL * (BACKOFF_BASE ** (attempt-1))
                time.sleep(backoff)

            # If checkpoint succeeded: initiate safe-failover and terminate gracefully
            if success:
                send_slack(f"Initiating safe failover for job {job_id}: draining outputs, snapshotting state.")
                log_audit({"job_id": job_id, "event": "safe_failover_start"})
                perform_safe_drain(job_id)   # e.g., pause input connectors, mark downstream
                time.sleep(SAFE_TERMINATION_GRACE)
                terminate_job(job_id, reason="cost_runaway_checkpointed")
                log_audit({"job_id": job_id, "event": "terminated", "reason": "runaway_after_checkpoint"})
                send_slack(f"Job {job_id} terminated after successful checkpoint.")
            else:
                # If checkpoint failed after retries: escalate and force-terminate to stop billing
                send_slack(f":rotating_light: Checkpoint failed for job {job_id} after {CHECKPOINT_RETRIES} attempts. Forcing termination.")
                log_audit({"job_id": job_id, "event": "checkpoint_failed_final"})
                force_terminate_job(job_id, reason="cost_runaway_no_checkpoint")
                log_audit({"job_id": job_id, "event": "force_terminated"})
                send_slack(f"Job {job_id} force-terminated to prevent cost runaway.")

        # Reset alert if job cost returns to normal for a cooldown period
        if alerted and cost_rate < (COST_THRESHOLD * 0.8):
            alerted = False
            log_audit({"job_id": job_id, "event": "alert_cleared", "cost_rate": cost_rate})
            send_slack(f"Runaway cleared for job {job_id}; cost_rate={cost_rate:.2f}")

        time.sleep(POLL_INTERVAL)

# Placeholder: provider APIs
def query_cost_rate(job_id): pass
def perform_safe_drain(job_id): pass
def terminate_job(job_id, reason): pass
def force_terminate_job(job_id, reason): pass

from typing import List, Optional, Tuple

def estimate_monthly_cost(hourly_bytes: List[int],
                          cost_per_gb: float,
                          hours_in_sample: Optional[int] = None,
                          events_in_sample: Optional[int] = None
                         ) -> Tuple[float, Optional[float]]:
    """
    hourly_bytes: list of bytes observed per hour (can be empty)
    cost_per_gb: USD per GB-month
    hours_in_sample: number of hours the list represents; if None, treated as len(hourly_bytes)
    events_in_sample: number of events corresponding to the sample period; if None, cost per million events returned as None

    Returns: (estimated_monthly_cost_usd, cost_per_million_events_usd_or_None)

    Behavior:
    - Aggregates bytes in the sample, computes average bytes/hour.
    - Scales average to 24*30 hours (30-day month).
    - Converts bytes to GB (1 GB = 10**9 bytes) and multiplies by cost_per_gb.
    - If events_in_sample provided, scales events to monthly and returns cost per million events.
    """
    if not hourly_bytes:
        return 0.0, None

    sample_hours = hours_in_sample if hours_in_sample is not None else len(hourly_bytes)
    if sample_hours <= 0:
        raise ValueError("hours_in_sample must be positive")

    total_bytes = sum(hourly_bytes)
    avg_bytes_per_hour = total_bytes / sample_hours  # supports streaming average if needed

    monthly_hours = 24 * 30
    monthly_bytes = avg_bytes_per_hour * monthly_hours
    monthly_gb = monthly_bytes / 1e9
    monthly_cost = monthly_gb * cost_per_gb

    cost_per_million = None
    if events_in_sample is not None:
        if events_in_sample < 0:
            raise ValueError("events_in_sample must be non-negative")
        avg_events_per_hour = events_in_sample / sample_hours
        monthly_events = avg_events_per_hour * monthly_hours
        if monthly_events == 0:
            cost_per_million = None
        else:
            cost_per_million = monthly_cost / (monthly_events / 1_000_000)

    return monthly_cost, cost_per_million