83 lines
2.7 KiB
Python
83 lines
2.7 KiB
Python
"""Aggregation helpers: month windows, per-tenant / per-metric / daily rollups."""
|
|
from datetime import datetime, timedelta
|
|
from zoneinfo import ZoneInfo
|
|
|
|
|
|
def month_range(year: int | None, month: int | None, tz: str):
|
|
"""Return (begin, end) datetimes for a billing month in the cluster tz.
|
|
|
|
CloudKitty aligns its rating periods to month boundaries in local time,
|
|
e.g. 2026-08-01T00:00:00+05:30 -> 2026-09-01T00:00:00+05:30.
|
|
"""
|
|
zone = ZoneInfo(tz)
|
|
now = datetime.now(zone)
|
|
year = year or now.year
|
|
month = month or now.month
|
|
begin = datetime(year, month, 1, tzinfo=zone)
|
|
end = (
|
|
datetime(year + 1, 1, 1, tzinfo=zone)
|
|
if month == 12
|
|
else datetime(year, month + 1, 1, tzinfo=zone)
|
|
)
|
|
return begin, end
|
|
|
|
|
|
def days_of_month(begin: datetime, end: datetime):
|
|
"""Yield (day_begin, day_end) pairs from month begin until min(today, end)."""
|
|
now = datetime.now(begin.tzinfo)
|
|
cursor = begin
|
|
while cursor < end and cursor < now:
|
|
nxt = min(cursor + timedelta(days=1), end)
|
|
yield cursor, nxt
|
|
cursor = nxt
|
|
|
|
|
|
def tenant_costs(rows: list[dict]) -> list[dict]:
|
|
"""Normalize CloudKitty summary rows into a sorted tenant cost list."""
|
|
out = [
|
|
{
|
|
"tenant": r.get("tenant") or "(unknown)",
|
|
"qty": r.get("qty", 0),
|
|
"rate": round(r.get("rate", 0) or 0, 6),
|
|
}
|
|
for r in rows
|
|
]
|
|
out.sort(key=lambda x: x["rate"], reverse=True)
|
|
return out
|
|
|
|
|
|
def per_metric(rows: list[dict], tenant: str) -> list[dict]:
|
|
"""From groupby=tenant,type rows, extract the metric breakdown of one tenant.
|
|
|
|
'type' is the CloudKitty dataframe field holding the metric name.
|
|
"""
|
|
out = []
|
|
for r in rows:
|
|
if (r.get("tenant") or "") != tenant:
|
|
continue
|
|
out.append(
|
|
{
|
|
"metric": r.get("type", "?"),
|
|
"qty": r.get("qty", 0),
|
|
"rate": round(r.get("rate", 0) or 0, 6),
|
|
}
|
|
)
|
|
out.sort(key=lambda x: x["rate"], reverse=True)
|
|
return out
|
|
|
|
|
|
def projection(rate_so_far: float, begin: datetime, end: datetime, tz: str) -> dict:
|
|
"""Linear month-end projection from the rate accrued so far."""
|
|
now = datetime.now(ZoneInfo(tz))
|
|
total_hours = (end - begin).total_seconds() / 3600
|
|
elapsed_hours = max((now - begin).total_seconds() / 3600, 0.0)
|
|
if elapsed_hours < 1:
|
|
return {"projected_month_end": None, "note": "less than 1h of data"}
|
|
projected = rate_so_far * (total_hours / elapsed_hours)
|
|
return {
|
|
"projected_month_end": round(projected, 2),
|
|
"elapsed_hours": round(elapsed_hours, 1),
|
|
"total_hours": round(total_hours, 1),
|
|
"pct_elapsed": round(100 * elapsed_hours / total_hours, 1),
|
|
}
|