Mapping Azure EA Billing to FinOps Tags

import os
import time
import logging
from typing import Dict, List, Optional
from datetime import datetime, timedelta
from azure.identity import DefaultAzureCredential
from azure.mgmt.costmanagement import CostManagementClient
from azure.mgmt.costmanagement.models import QueryDefinition, QueryDataset, QueryGrouping
from azure.mgmt.resourcegraph import ResourceGraphClient
from azure.mgmt.resourcegraph.models import QueryRequest
import pandas as pd
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

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

class AzureBillingTagMapper:
    """
    Production-grade pipeline for mapping Azure EA billing line items
    to FinOps tags via Resource Graph cross-referencing and fallback inheritance.
    """
    def __init__(self, enrollment_account_id: str, chunk_size: int = 1000):
        self.credential = DefaultAzureCredential()
        self.cost_client = CostManagementClient(self.credential)
        self.rg_client = ResourceGraphClient(self.credential)
        self.scope = f"/providers/Microsoft.Billing/billingAccounts/{enrollment_account_id}"
        self.chunk_size = chunk_size
        self.tag_cache: Dict[str, Dict[str, str]] = {}

    @retry(
        stop=stop_after_attempt(4),
        wait=wait_exponential(multiplier=2, min=4, max=30),
        retry=retry_if_exception_type(Exception)
    )
    def _execute_query_with_retry(self, query_func, *args, **kwargs):
        """Wrapper to handle transient API throttling and network instability."""
        return query_func(*args, **kwargs)

    def fetch_daily_usage(self, start_date: str, end_date: str) -> pd.DataFrame:
        """Retrieve paginated daily usage details from Cost Management API."""
        query_def = QueryDefinition(
            type="ActualCost",
            timeframe="Custom",
            time_period={"from": start_date, "to": end_date},
            dataset=QueryDataset(
                granularity="Daily",
                aggregation={"totalCost": {"name": "PreTaxCost", "function": "Sum"}},
                groupings=[
                    QueryGrouping(type="Dimension", name="ResourceId"),
                    QueryGrouping(type="Dimension", name="ResourceType"),
                    QueryGrouping(type="Dimension", name="MeterCategory")
                ]
            )
        )

        rows = []
        query_result = self._execute_query_with_retry(
            self.cost_client.query.usage, self.scope, query_def
        )

        while query_result:
            for row in query_result.rows:
                rows.append({
                    "date": row[0],
                    "resource_id": row[1],
                    "resource_type": row[2],
                    "meter_category": row[3],
                    "cost": row[4]
                })
            if hasattr(query_result, 'next_link') and query_result.next_link:
                query_result = self._execute_query_with_retry(
                    self.cost_client.query.usage_next, query_result.next_link
                )
            else:
                break

        logger.info(f"Fetched {len(rows)} billing rows for {start_date} to {end_date}")
        return pd.DataFrame(rows)

    def _resolve_tags_batch(self, resource_ids: List[str]) -> Dict[str, Dict[str, str]]:
        """Batch query Azure Resource Graph for resource metadata and tags."""
        resolved = {}
        valid_ids = [rid for rid in resource_ids if rid.startswith("/subscriptions/")]
        if not valid_ids:
            return resolved

        for i in range(0, len(valid_ids), self.chunk_size):
            chunk = valid_ids[i:i + self.chunk_size]
            ids_clause = " OR ".join([f"id = '{rid}'" for rid in chunk])
            query = QueryRequest(
                query=f"Resources | where {ids_clause} | project id, tags, resourceGroup, subscriptionId"
            )
            response = self._execute_query_with_retry(self.rg_client.resources, query)
            for item in response.data:
                resolved[item['id']] = item.get('tags', {}) or {}

        return resolved

    def _apply_fallback_inheritance(self, resource_id: str, resource_tags: Dict[str, str]) -> Dict[str, str]:
        """Apply hierarchical tag inheritance: Resource -> Resource Group -> Subscription."""
        if not resource_id.startswith("/subscriptions/"):
            return resource_tags

        parts = resource_id.split("/")
        if len(parts) < 5:
            return resource_tags

        sub_id = f"/subscriptions/{parts[2]}"
        rg_name = parts[4]
        rg_id = f"{sub_id}/resourceGroups/{rg_name}"

        fallback_query = QueryRequest(
            query=f"ResourceContainers | where id in ('{rg_id}', '{sub_id}') | project id, tags"
        )
        containers = self._execute_query_with_retry(self.rg_client.resources, fallback_query)
        container_map = {c['id']: c.get('tags', {}) or {} for c in containers.data}

        final_tags = dict(resource_tags)
        if rg_id in container_map:
            for k, v in container_map[rg_id].items():
                final_tags.setdefault(k, v)
        if sub_id in container_map:
            for k, v in container_map[sub_id].items():
                final_tags.setdefault(k, v)

        return final_tags

    def map_and_normalize(self, start_date: str, end_date: str) -> pd.DataFrame:
        """Orchestrate the billing-to-tag mapping pipeline."""
        usage_df = self.fetch_daily_usage(start_date, end_date)
        unique_resources = usage_df["resource_id"].dropna().unique().tolist()

        logger.info("Resolving tags via Resource Graph...")
        raw_tags = self._resolve_tags_batch(unique_resources)
        self.tag_cache.update(raw_tags)

        enriched_rows = []
        for _, row in usage_df.iterrows():
            rid = row["resource_id"]
            base_tags = self.tag_cache.get(rid, {})
            final_tags = self._apply_fallback_inheritance(rid, base_tags)

            enriched_rows.append({
                "date": row["date"],
                "resource_id": rid,
                "resource_type": row["resource_type"],
                "meter_category": row["meter_category"],
                "cost": row["cost"],
                "tags": final_tags,
                "finops_cost_center": final_tags.get("CostCenter", "Unallocated"),
                "finops_env": final_tags.get("Environment", "Unknown"),
                "finops_owner": final_tags.get("Owner", "Unassigned")
            })

        return pd.DataFrame(enriched_rows)

if __name__ == "__main__":
    ENROLLMENT_ID = os.getenv("AZURE_ENROLLMENT_ACCOUNT_ID", "12345678")
    mapper = AzureBillingTagMapper(enrollment_account_id=ENROLLMENT_ID)
    df = mapper.map_and_normalize(
        start_date=(datetime.now() - timedelta(days=30)).strftime("%Y-%m-%d"),
        end_date=datetime.now().strftime("%Y-%m-%d")
    )
    print(df.head())
    df.to_csv("finops_billing_allocation.csv", index=False)