billing/main.py

"""ConnectVM Billing API — aggregates CloudKitty rating data.

Billing model: ALLOCATION-BASED. CloudKitty collects openstack_nova_limits_*
/ cinder_limits_* metrics, which track quota allocation, not power state.
A stopped (SHUTOFF) VM is therefore billed in full until it is shelved
(offloaded) or deleted. See /api/v1/billing/policy.
"""
import asyncio
from contextlib import asynccontextmanager
from datetime import datetime
from pathlib import Path

from fastapi import FastAPI, HTTPException, Query
from fastapi.responses import FileResponse, HTMLResponse
from pydantic import BaseModel

import aggregator
import mailer
import pricing
import reporting
from cloudkitty_client import CloudKittyClient
from config import get_settings
from openstack_client import OpenStackClient

settings = get_settings()
ck: CloudKittyClient | None = None
osx: OpenStackClient | None = None
STATIC_DIR = Path(__file__).parent / "static"

BILLING_POLICY = {
    "model": "allocation-based (quota), hourly collection windows",
    "source_metrics": [
        "openstack_nova_limits_vcpus_used",
        "openstack_nova_limits_memory_used",
        "openstack_nova_limits_instances_used",
        "openstack_nova_server_local_gb",
        "openstack_cinder_limits_volume_used_gb",
        "openstack_glance_image_bytes",
        "openstack_neutron_floating_ip",
    ],
    "vm_state_rules": [
        {
            "state": "ACTIVE",
            "billed": True,
            "detail": "Full rate: vCPU, RAM, disk, volumes, floating IPs.",
        },
        {
            "state": "SHUTOFF (stopped) — e.g. VM down 5 days",
            "billed": True,
            "detail": (
                "STILL FULLY BILLED. Nova keeps the vCPU/RAM/disk quota "
                "allocated (capacity reserved), so openstack_nova_limits_* "
                "metrics do not change on power-off. Billing only stops when "
                "the allocation is released."
            ),
        },
        {
            "state": "PAUSED / SUSPENDED",
            "billed": True,
            "detail": "RAM stays reserved on the host; quota unchanged.",
        },
        {
            "state": "SHELVED_OFFLOADED",
            "billed": "partial",
            "detail": (
                "vCPU/RAM freed from quota -> compute not billed. Root/data "
                "volumes in Cinder keep billing; image resumes on unshelve."
            ),
        },
        {
            "state": "DELETED",
            "billed": "until deletion",
            "detail": (
                "CloudKitty rates hourly windows, so a VM deleted mid-month "
                "is billed only for the hours it existed."
            ),
        },
    ],
    "how_to_stop_billing_a_vm": (
        "Shelve it (openstack server shelve, then it offloads) or delete it. "
        "Stopping/shutting off the VM does NOT stop billing."
    ),
}


@asynccontextmanager
async def lifespan(app: FastAPI):
    global ck, osx
    ck = CloudKittyClient(settings)
    osx = OpenStackClient(settings)
    yield
    await ck.close()
    await osx.close()


app = FastAPI(
    title="ConnectVM Billing API",
    version="1.0.0",
    description="Aggregated tenant billing on top of CloudKitty.",
    lifespan=lifespan,
)


@app.get("/health")
async def health():
    return {"status": "ok"}


@app.get("/admin", response_class=HTMLResponse)
async def admin_ui():
    """Admin billing dashboard."""
    return FileResponse(STATIC_DIR / "admin.html")


async def _tenant_resources(tenant: str, year: int | None, month: int | None):
    """Shared loader: period, inventory, INR cost rows, report data."""
    begin, end = aggregator.month_range(year, month, settings.billing_tz)
    now = datetime.now(begin.tzinfo)

    projects = await osx.projects()
    project_id = projects.get(tenant)
    if not project_id:
        raise HTTPException(404, f"keystone project '{tenant}' not found")

    total_rows, servers, volumes, fips = await asyncio.gather(
        ck.summary(begin=begin, end=end, groupby=("tenant",)),
        osx.servers(project_id),
        osx.volumes(project_id),
        osx.floatingips(project_id),
    )
    mine = [r for r in total_rows if (r.get("tenant") or "") == tenant]
    ck_total = (mine[0].get("rate", 0) or 0) if mine else 0.0

    vms = reporting.vm_rows(servers, begin, end, now)
    vols = reporting.volume_rows(volumes, begin, end, now)
    fip_list = reporting.fip_rows(fips, begin, end, now)
    report = reporting.build_report(tenant, begin, end, ck_total, vms, vols, fip_list)
    report["pricing"] = pricing.RATE_CARD_INFO
    return report


@app.get("/api/v1/billing/tenants/{tenant}/resources")
async def tenant_resources(tenant: str, year: int | None = None, month: int | None = None):
    """Admin drill-down: every VM / volume / FIP with hours and INR cost."""
    return await _tenant_resources(tenant, year, month)


@app.get("/api/v1/billing/summary-inr")
async def summary_inr(year: int | None = None, month: int | None = None):
    """All tenants: INR bill-to-date and monthly run-rate (rate card)."""
    begin, end = aggregator.month_range(year, month, settings.billing_tz)
    now = datetime.now(begin.tzinfo)
    total_rows, projects = await asyncio.gather(
        ck.summary(begin=begin, end=end, groupby=("tenant",)),
        osx.projects(),
    )
    sem = asyncio.Semaphore(6)

    async def one_tenant(tname: str):
        pid = projects.get(tname)
        base = {"tenant": tname}
        if not pid:
            return {**base, "error": "no keystone project"}
        async with sem:
            servers, volumes, fips = await asyncio.gather(
                osx.servers(pid), osx.volumes(pid), osx.floatingips(pid)
            )
        vms = reporting.vm_rows(servers, begin, end, now)
        vols = reporting.volume_rows(volumes, begin, end, now)
        fip_rows = reporting.fip_rows(fips, begin, end, now)
        all_rows = vms + vols + fip_rows
        return {
            **base,
            "bill_to_date": round(sum(r["est_cost"] for r in all_rows), 2),
            "run_rate_per_month": round(sum(r["monthly_rate"] for r in all_rows), 2),
            "vms": len(vms),
            "volumes": len(vols),
            "floating_ips": len(fip_rows),
            "stopped_vms": sum(1 for r in vms if r["status"] in ("SHUTOFF", "ERROR")),
        }

    names = [(r.get("tenant") or "") for r in total_rows]
    tenants = await asyncio.gather(*(one_tenant(n) for n in names if n))
    tenants.sort(key=lambda t: t.get("bill_to_date") or 0, reverse=True)
    return {
        "period": {"begin": begin.isoformat(), "end": end.isoformat()},
        "currency": "INR",
        "total_bill_to_date": round(
            sum(t.get("bill_to_date") or 0 for t in tenants), 2
        ),
        "tenants": tenants,
    }


@app.get("/api/v1/billing/tenants/{tenant}/report")
async def tenant_report_json(tenant: str, year: int | None = None, month: int | None = None):
    return await _tenant_resources(tenant, year, month)


@app.get("/api/v1/billing/tenants/{tenant}/report.html", response_class=HTMLResponse)
async def tenant_report_html(tenant: str, year: int | None = None, month: int | None = None):
    """User-facing report page (this is what gets emailed to the tenant)."""
    report = await _tenant_resources(tenant, year, month)
    return HTMLResponse(reporting.render_html(report))


class SendRequest(BaseModel):
    to_email: str


@app.post("/api/v1/billing/tenants/{tenant}/send")
async def tenant_report_send(tenant: str, req: SendRequest,
                             year: int | None = None, month: int | None = None):
    """Email the report to the tenant (after admin review in /admin)."""
    report = await _tenant_resources(tenant, year, month)
    html = reporting.render_html(report)
    subject = (
        f"ConnectVM billing report — {tenant} "
        f"({report['period']['begin'][:10]} to {report['period']['end'][:10]})"
    )
    try:
        await mailer.send_report(settings, req.to_email, subject, html)
    except mailer.MailNotConfigured as e:
        raise HTTPException(
            400,
            f"{e}. Review/download instead: "
            f"/api/v1/billing/tenants/{tenant}/report.html",
        )
    return {"sent": True, "to": req.to_email, "subject": subject}


@app.get("/api/v1/billing/policy")
async def policy():
    """How power states affect billing (the 'VM down 5 days' question)."""
    return BILLING_POLICY


@app.get("/api/v1/billing/summary")
async def summary(
    year: int | None = Query(None),
    month: int | None = Query(None),
    groupby: str = Query("tenant", description="comma-separated, e.g. tenant,type"),
):
    """All-tenant totals for a month (defaults to current month)."""
    begin, end = aggregator.month_range(year, month, settings.billing_tz)
    rows = await ck.summary(begin=begin, end=end, groupby=tuple(groupby.split(",")))
    tenants = aggregator.tenant_costs(rows) if "tenant" in groupby else rows
    return {
        "period": {"begin": begin.isoformat(), "end": end.isoformat()},
        "currency_note": "rate unit is whatever CloudKitty hashmap prices define",
        "total_rate": round(sum(t.get("rate", 0) for t in tenants), 2)
        if tenants
        else 0,
        "tenants": tenants,
    }


@app.get("/api/v1/billing/tenants/{tenant}")
async def tenant_detail(tenant: str, year: int | None = None, month: int | None = None):
    """One tenant: month total, per-metric breakdown, month-end projection."""
    begin, end = aggregator.month_range(year, month, settings.billing_tz)

    total_rows, metric_rows = await asyncio.gather(
        ck.summary(begin=begin, end=end, groupby=("tenant",)),
        ck.summary(begin=begin, end=end, groupby=("tenant", "type")),
    )
    mine = [r for r in total_rows if (r.get("tenant") or "") == tenant]
    if not mine:
        raise HTTPException(404, f"no billing data for tenant '{tenant}'")

    rate = mine[0].get("rate", 0) or 0
    return {
        "tenant": tenant,
        "period": {"begin": begin.isoformat(), "end": end.isoformat()},
        "rate_so_far": round(rate, 6),
        "qty_so_far": mine[0].get("qty", 0),
        "metrics": aggregator.per_metric(metric_rows, tenant),
        "projection": aggregator.projection(rate, begin, end, settings.billing_tz),
    }


@app.get("/api/v1/billing/daily")
async def daily(
    tenant: str | None = Query(None, description="filter to one tenant"),
    year: int | None = None,
    month: int | None = None,
):
    """Day-by-day cost rollup, month-to-date."""
    begin, end = aggregator.month_range(year, month, settings.billing_tz)
    days = list(aggregator.days_of_month(begin, end))
    sem = asyncio.Semaphore(4)

    async def one_day(d_begin, d_end):
        async with sem:
            rows = await ck.summary(begin=d_begin, end=d_end, groupby=("tenant",))
            if tenant:
                rows = [r for r in rows if (r.get("tenant") or "") == tenant]
            return {
                "date": d_begin.date().isoformat(),
                "rate": round(sum(r.get("rate", 0) or 0 for r in rows), 6),
                "tenants": aggregator.tenant_costs(rows) if not tenant else None,
            }

    results = await asyncio.gather(*(one_day(b, e) for b, e in days))
    for r in results:
        if r["tenants"] is None:
            del r["tenants"]
    return {
        "period": {"begin": begin.isoformat(), "end": end.isoformat()},
        "tenant": tenant or "ALL",
        "days": results,
    }