simple-cost-dashboard/collector/collect.py

#!/usr/bin/env python3
"""
collect.py — Point d'entrée du cron. Collecte le coût de la veille (J-1)
pour chaque tenant défini dans config.json, et l'écrit dans SQLite.

Pourquoi J-1 et pas "aujourd'hui" : les données de facturation AWS/Azure du
jour même sont généralement incomplètes (délai de consolidation de
quelques heures à ~24-48h côté AWS Cost Explorer notamment). Collecter J-1
donne un chiffre stable.

Utilisation (cron quotidien, par exemple à 6h du matin) :
    python collect.py --config /path/to/config.json --db /path/to/cost_dashboard.db

Un échec sur un tenant n'empêche pas la collecte des autres : chaque
tenant est traité indépendamment et les erreurs sont journalisées en base
(status='error') pour que le dashboard puisse l'afficher plutôt que de
silencieusement afficher un trou ou un zéro.
"""

import argparse
import json
import logging
import sys
from datetime import date, datetime, timedelta, timezone
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent))

from db import (
    get_connection, init_db, upsert_collection,
    upsert_resource_costs, set_resource_collection_status,
)
from providers import (
    get_aws_cost, get_azure_cost, convert_result,
    get_aws_resource_costs, get_azure_resource_costs, convert_resource_costs,
    ResourceLevelUnavailable,
)

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
)
log = logging.getLogger("collect")

TARGET_CURRENCY = "EUR"

# Fenêtre du détail par ressource (bucket S3, instance EC2...) : bornée à 14
# jours côté AWS Cost Explorer (limite dure de l'API) ; gardée identique côté
# Azure pour rester cohérent entre providers. Voir providers.py.
RESOURCE_WINDOW_DAYS = 14


def collect_tenant(conn, provider: str, entry: dict, period_start: str, period_end: str) -> bool:
    """Collecte le relevé quotidien (par service) d'un tenant. Retourne True
    si la collecte a réussi — sert à décider si ça vaut la peine de tenter
    le détail par ressource juste après (inutile si l'auth du tenant est
    déjà cassée)."""
    name = entry["name"]
    collected_at = datetime.now(timezone.utc).isoformat()

    try:
        if provider == "aws":
            data = get_aws_cost(entry, period_start, period_end, granularity="DAILY")
        else:
            data = get_azure_cost(
                entry["tenant_id"], entry["client_id"], entry["client_secret"],
                entry["subscription_id"], period_start, period_end,
            )
        data = convert_result(data, TARGET_CURRENCY)

        upsert_collection(
            conn, provider, name, "daily", period_start, period_end,
            collected_at, "ok", data=data,
        )
        log.info("OK   %-6s %-30s %.2f %s", provider, name, data["total"], data["currency"])
        return True

    except Exception as e:
        upsert_collection(
            conn, provider, name, "daily", period_start, period_end,
            collected_at, "error", error_message=str(e),
        )
        log.error("ECHEC %-6s %-30s %s", provider, name, e)
        return False


def collect_tenant_resources(conn, provider: str, entry: dict, period_start: str, period_end: str) -> None:
    """Collecte le détail par ressource (RESOURCE_WINDOW_DAYS derniers
    jours) d'un tenant, pour le panneau de détail du dashboard. Séparé de
    collect_tenant : ça peut échouer pour une raison indépendante (ex:
    "Resource IDs" non activé côté AWS) sans que la collecte quotidienne
    normale soit affectée."""
    name = entry["name"]
    collected_at = datetime.now(timezone.utc).isoformat()

    try:
        if provider == "aws":
            data = get_aws_resource_costs(entry, period_start, period_end)
        else:
            data = get_azure_resource_costs(
                entry["tenant_id"], entry["client_id"], entry["client_secret"],
                entry["subscription_id"], period_start, period_end,
            )
        data = convert_resource_costs(data, TARGET_CURRENCY)

        upsert_resource_costs(conn, provider, name, period_start, period_end, data["services"])
        set_resource_collection_status(conn, provider, name, "ok", collected_at)
        n_resources = sum(len(v) for v in data["services"].values())
        log.info("OK   %-6s %-30s détail ressources (%d ressources)", provider, name, n_resources)

    except ResourceLevelUnavailable as e:
        set_resource_collection_status(conn, provider, name, "unavailable", collected_at, message=str(e))
        log.warning("INDISPO %-6s %-30s détail ressources indisponible : %s", provider, name, e)

    except Exception as e:
        set_resource_collection_status(conn, provider, name, "error", collected_at, message=str(e))
        log.error("ECHEC %-6s %-30s détail ressources : %s", provider, name, e)


def main():
    parser = argparse.ArgumentParser(description="Collecte journalière des coûts (cron)")
    parser.add_argument("--config", default="config.json")
    parser.add_argument("--db", default=None, help="Chemin vers la base SQLite")
    parser.add_argument("--date", help="Jour à collecter (YYYY-MM-DD), défaut: hier")
    args = parser.parse_args()

    if args.date:
        target_day = date.fromisoformat(args.date)
    else:
        target_day = date.today() - timedelta(days=1)

    period_start = target_day.isoformat()
    period_end = (target_day + timedelta(days=1)).isoformat()

    db_path = args.db or None
    if db_path:
        init_db(db_path)
        conn = get_connection(db_path)
    else:
        init_db()
        conn = get_connection()

    with open(args.config) as f:
        config = json.load(f)

    log.info("Collecte pour le %s", period_start)

    resource_period_end = target_day + timedelta(days=1)
    resource_period_start = resource_period_end - timedelta(days=RESOURCE_WINDOW_DAYS)

    for provider, entries in (("aws", config.get("aws", [])), ("azure", config.get("azure", []))):
        for entry in entries:
            ok = collect_tenant(conn, provider, entry, period_start, period_end)
            if ok:
                collect_tenant_resources(
                    conn, provider, entry,
                    resource_period_start.isoformat(), resource_period_end.isoformat(),
                )

    conn.close()
    log.info("Terminé.")


if __name__ == "__main__":
    main()