From d72a34dbe2e167d78e7f7baf79438ceff5f4aa17 Mon Sep 17 00:00:00 2001 From: Harshit28j Date: Sat, 14 Mar 2026 22:19:56 +0530 Subject: [PATCH] add docs and export fixes --- docs/my-website/docs/observability/vantage.md | 148 ++++++++++++++++++ .../focus/destinations/vantage_destination.py | 102 ++++++++++-- litellm/integrations/focus/export_engine.py | 27 ++++ litellm/integrations/focus/focus_logger.py | 27 +++- 4 files changed, 288 insertions(+), 16 deletions(-) create mode 100644 docs/my-website/docs/observability/vantage.md diff --git a/docs/my-website/docs/observability/vantage.md b/docs/my-website/docs/observability/vantage.md new file mode 100644 index 0000000000..31b43a76c3 --- /dev/null +++ b/docs/my-website/docs/observability/vantage.md @@ -0,0 +1,148 @@ +import Tabs from '@theme/Tabs'; +import TabItem from '@theme/TabItem'; + +# Vantage Integration + +LiteLLM can export proxy spend data to [Vantage](https://vantage.sh) as [FOCUS 1.2](https://focus.finops.org/) formatted cost reports. This lets you visualize LLM spend alongside your cloud infrastructure costs in the Vantage dashboard. + +## Overview + +| Property | Details | +|----------|---------| +| Destination | Export LiteLLM usage data to Vantage Custom Provider | +| Data format | FOCUS CSV (automatically transformed from LiteLLM spend data) | +| Supported operations | Manual export, automatic scheduled export (hourly/daily/interval) | +| Authentication | Vantage API key + Custom Provider token | + +## Prerequisites + +You need two credentials from the [Vantage console](https://console.vantage.sh): + +1. **API Key** — Go to **Settings → API Access Tokens** → Create a token with **Write** scope. The token looks like `vntg_tkn_...`. +2. **Custom Provider Token** — Go to **Settings → Integrations** → Create a **Custom Provider** integration → Copy the Provider ID (looks like `accss_crdntl_...`). + +## Setup via API + +The recommended setup uses the proxy admin endpoints. No config file changes needed. + +### 1. Initialize credentials + +```bash +curl -X POST http://localhost:4000/vantage/init \ + -H "Authorization: Bearer $LITELLM_ADMIN_KEY" \ + -H "Content-Type: application/json" \ + -d '{ + "api_key": "vntg_tkn_YOUR_VANTAGE_API_KEY", + "integration_token": "accss_crdntl_YOUR_PROVIDER_TOKEN" + }' +``` + +Credentials are encrypted and stored in the proxy database. + +### 2. Preview data (dry run) + +```bash +curl -X POST http://localhost:4000/vantage/dry-run \ + -H "Authorization: Bearer $LITELLM_ADMIN_KEY" \ + -H "Content-Type: application/json" \ + -d '{"limit": 10}' +``` + +This returns FOCUS-transformed data without sending anything to Vantage. Use it to verify the pipeline works and inspect the data mapping. + +### 3. Export to Vantage + +```bash +curl -X POST http://localhost:4000/vantage/export \ + -H "Authorization: Bearer $LITELLM_ADMIN_KEY" \ + -H "Content-Type: application/json" \ + -d '{}' +``` + +Optional parameters: +- `limit` — Max number of records to export +- `start_time_utc` / `end_time_utc` — Filter by time range (must be provided together) + +### 4. Verify in Vantage + +Go to **Settings → Integrations → your Custom Provider → Import Costs** tab to see uploaded CSVs. Once the status changes from "Importing and Processing" to "Stable", costs appear in **Cost Reporting → All Resources**. + +## Setup via Environment Variables + +For automatic scheduled exports, configure via environment variables and proxy config: + +### Environment variables + +| Variable | Required | Description | +|----------|----------|-------------| +| `VANTAGE_API_KEY` | Yes | Vantage API access token | +| `VANTAGE_INTEGRATION_TOKEN` | Yes | Custom Provider token from Vantage dashboard | +| `VANTAGE_BASE_URL` | No | API URL override (default: `https://api.vantage.sh`) | +| `VANTAGE_EXPORT_FREQUENCY` | No | `hourly` (default), `daily`, or `interval` | +| `VANTAGE_EXPORT_INTERVAL_SECONDS` | No | Seconds between exports when frequency is `interval` | + +### Proxy config + +```yaml +model_list: + - model_name: gpt-4o + litellm_params: + model: openai/gpt-4o + api_key: sk-your-key + +litellm_settings: + callbacks: ["vantage"] +``` + +```bash +export VANTAGE_API_KEY="vntg_tkn_..." +export VANTAGE_INTEGRATION_TOKEN="accss_crdntl_..." +litellm --config /path/to/config.yaml +``` + +The proxy registers a background job that exports data on the configured schedule. + +## API Endpoints + +All endpoints require admin authentication. + +| Method | Endpoint | Description | +|--------|----------|-------------| +| `POST` | `/vantage/init` | Store Vantage credentials (encrypted) | +| `GET` | `/vantage/settings` | View current config (credentials masked) | +| `PUT` | `/vantage/settings` | Update credentials or base URL | +| `POST` | `/vantage/dry-run` | Preview FOCUS data without uploading | +| `POST` | `/vantage/export` | Upload cost data to Vantage | +| `DELETE` | `/vantage/delete` | Remove credentials and stop scheduled exports | + +## FOCUS Field Mapping + +LiteLLM spend data is transformed into the FOCUS 1.2 schema: + +| LiteLLM Field | FOCUS Column | Description | +|---------------|-------------|-------------| +| `spend` | BilledCost, EffectiveCost | Cost of the usage | +| `model` | ChargeDescription, ResourceId | Model identifier | +| `model_group` | ServiceName | Model group / deployment | +| `custom_llm_provider` | ProviderName, PublisherName | Provider (openai, anthropic, etc.) | +| `api_key` | BillingAccountId | Hashed API key | +| `api_key_alias` | BillingAccountName | Human-readable key alias | +| `team_id` | SubAccountId | Team identifier | +| `team_alias` | SubAccountName | Team name | + +Additional metadata (user_id, model_group, etc.) is included in the `Tags` column as JSON. + +## Upload Limits + +Vantage enforces per-upload limits. LiteLLM handles these automatically: + +- **10,000 rows** per upload — large exports are split into batches +- **2 MB** per upload — oversized batches are further split by size +- **Unsupported columns** are stripped before upload + +## Related Links + +- [Vantage](https://vantage.sh) +- [Vantage Custom Providers](https://docs.vantage.sh/connecting_custom_providers) +- [FOCUS Specification](https://focus.finops.org/) +- [Focus Export (S3/Parquet)](./focus.md) diff --git a/litellm/integrations/focus/destinations/vantage_destination.py b/litellm/integrations/focus/destinations/vantage_destination.py index 8275c0690b..9e6028900f 100644 --- a/litellm/integrations/focus/destinations/vantage_destination.py +++ b/litellm/integrations/focus/destinations/vantage_destination.py @@ -2,6 +2,8 @@ from __future__ import annotations +import csv +import io from typing import Any, Optional import httpx @@ -14,6 +16,77 @@ from .base import FocusDestination, FocusTimeWindow VANTAGE_MAX_ROWS_PER_UPLOAD = 10_000 VANTAGE_MAX_BYTES_PER_UPLOAD = 2 * 1024 * 1024 # 2 MB +# Columns that Vantage actually supports for custom provider CSV uploads. +# See: https://docs.vantage.sh/connecting_custom_providers +# Columns not in this set are silently dropped before upload so Vantage +# does not reject the file. +VANTAGE_SUPPORTED_COLUMNS = { + # Required + "ChargeCategory", + "ChargePeriodStart", + "BilledCost", + "ServiceName", + # Optional + "BillingCurrency", + "BillingAccountId", + "BillingAccountName", + "ChargePeriodEnd", + "ChargeDescription", + "ChargeFrequency", + "ConsumedQuantity", + "ConsumedUnit", + "ContractedCost", + "EffectiveCost", + "ListCost", + "RegionId", + "RegionName", + "ResourceId", + "ResourceName", + "ResourceType", + "ServiceCategory", + "ServiceSubcategory", + "SubAccountId", + "SubAccountName", + "Tags", +} + + +def _strip_unsupported_columns(csv_bytes: bytes) -> bytes: + """Remove CSV columns not in VANTAGE_SUPPORTED_COLUMNS. + + Parses the header row, identifies column indices to keep, and + rebuilds the CSV with only those columns. + """ + lines = csv_bytes.split(b"\n") + if not lines: + return csv_bytes + + header_cols = lines[0].decode("utf-8").split(",") + keep_indices = [ + i + for i, col in enumerate(header_cols) + if col.strip('"') in VANTAGE_SUPPORTED_COLUMNS + ] + + # If all columns are supported, return as-is + if len(keep_indices) == len(header_cols): + return csv_bytes + + dropped = [col for i, col in enumerate(header_cols) if i not in keep_indices] + verbose_logger.debug( + "Vantage destination: dropping unsupported columns: %s", dropped + ) + + output = io.StringIO() + writer = csv.writer(output) + reader = csv.reader(io.StringIO(csv_bytes.decode("utf-8"))) + for row in reader: + if not row: + continue + writer.writerow([row[i] for i in keep_indices]) + + return output.getvalue().encode("utf-8") + class FocusVantageDestination(FocusDestination): """Upload FOCUS CSV exports to the Vantage cost-import API.""" @@ -39,9 +112,7 @@ class FocusVantageDestination(FocusDestination): ) self.api_key = api_key self.integration_token = integration_token - self.base_url = config.get( - "base_url", "https://api.vantage.sh" - ) + self.base_url = config.get("base_url", "https://api.vantage.sh") self.prefix = prefix async def deliver( @@ -56,6 +127,10 @@ class FocusVantageDestination(FocusDestination): verbose_logger.debug("Vantage destination: empty content, skipping upload") return + # Strip columns that Vantage does not support to avoid silent + # rejection (e.g. InvoiceIssuerName, ProviderName, PublisherName). + content = _strip_unsupported_columns(content) + # Reuse a single HTTP client for the entire deliver() call async with httpx.AsyncClient(timeout=60.0) as client: # Check both size and row-count limits before single-shot upload @@ -75,10 +150,7 @@ class FocusVantageDestination(FocusDestination): async def _upload_csv( self, client: httpx.AsyncClient, csv_bytes: bytes, filename: str ) -> None: - url = ( - f"{self.base_url}/v2/integrations/" - f"{self.integration_token}/costs.csv" - ) + url = f"{self.base_url}/v2/integrations/" f"{self.integration_token}/costs.csv" headers = { "Authorization": f"Bearer {self.api_key}", } @@ -86,15 +158,16 @@ class FocusVantageDestination(FocusDestination): response = await client.post( url, headers=headers, - files={"file": (filename, csv_bytes, "text/csv")}, + files={"csv": (filename, csv_bytes, "text/csv")}, ) try: response.raise_for_status() except httpx.HTTPStatusError as e: verbose_logger.error( - "Vantage destination: upload failed for %s — %s", + "Vantage destination: upload failed for %s — %s — response body: %s", filename, e, + response.text, ) raise @@ -174,7 +247,10 @@ class FocusVantageDestination(FocusDestination): ) continue - if current_size + line_size > VANTAGE_MAX_BYTES_PER_UPLOAD and current_chunk: + if ( + current_size + line_size > VANTAGE_MAX_BYTES_PER_UPLOAD + and current_chunk + ): batch_csv = header + b"\n" + b"\n".join(current_chunk) + b"\n" batch_filename = f"{filename}.part{batch_offset}_{sub_batch}" try: @@ -182,7 +258,8 @@ class FocusVantageDestination(FocusDestination): except Exception as e: verbose_logger.error( "Vantage destination: sub-batch %s failed: %s", - batch_filename, e, + batch_filename, + e, ) if first_error is None: first_error = e @@ -200,7 +277,8 @@ class FocusVantageDestination(FocusDestination): except Exception as e: verbose_logger.error( "Vantage destination: sub-batch %s failed: %s", - batch_filename, e, + batch_filename, + e, ) if first_error is None: first_error = e diff --git a/litellm/integrations/focus/export_engine.py b/litellm/integrations/focus/export_engine.py index 8b77fad9e1..37da18a0eb 100644 --- a/litellm/integrations/focus/export_engine.py +++ b/litellm/integrations/focus/export_engine.py @@ -67,6 +67,33 @@ class FocusExportEngine: "summary": summary, } + async def export_all( + self, + *, + limit: Optional[int], + ) -> None: + """Export all available data without time-window filtering.""" + data = await self._database.get_usage_data(limit=limit) + if data.is_empty(): + verbose_logger.debug("Focus export: no usage data available") + return + + normalized = self._transformer.transform(data) + if normalized.is_empty(): + verbose_logger.debug("Focus export: normalized data empty") + return + + # Build a window spanning the full data range for the filename + from datetime import datetime, timezone + + now = datetime.now(timezone.utc) + window = FocusTimeWindow( + start_time=now.replace(hour=0, minute=0, second=0, microsecond=0), + end_time=now, + frequency="all", + ) + await self._serialize_and_upload(normalized, window) + async def export_window( self, *, diff --git a/litellm/integrations/focus/focus_logger.py b/litellm/integrations/focus/focus_logger.py index f461f8c9fa..083b0e1463 100644 --- a/litellm/integrations/focus/focus_logger.py +++ b/litellm/integrations/focus/focus_logger.py @@ -64,7 +64,9 @@ class FocusLogger(CustomLogger): ) env_prefix = os.getenv("FOCUS_PREFIX") self.prefix: str = ( - prefix if prefix is not None else (env_prefix if env_prefix else "focus_exports") + prefix + if prefix is not None + else (env_prefix if env_prefix else "focus_exports") ) self._destination_config = destination_config @@ -90,7 +92,13 @@ class FocusLogger(CustomLogger): start_time_utc: Optional[datetime] = None, end_time_utc: Optional[datetime] = None, ) -> None: - """Public hook to trigger export immediately.""" + """Public hook to trigger export immediately. + + When called without time bounds (manual /vantage/export with no + start/end), exports **all** available data instead of the last + scheduled window. The hourly/daily window only applies to + automatic scheduler runs. + """ if bool(start_time_utc) ^ bool(end_time_utc): raise ValueError( "start_time_utc and end_time_utc must be provided together" @@ -102,9 +110,10 @@ class FocusLogger(CustomLogger): end_time=end_time_utc, frequency=self.frequency, ) + await self._export_window(window=window, limit=limit) else: - window = self._compute_time_window(datetime.now(timezone.utc)) - await self._export_window(window=window, limit=limit) + # No time bounds → export all available data + await self._export_all(limit=limit) async def dry_run_export_usage_data( self, limit: Optional[int] = DEFAULT_DRY_RUN_LIMIT @@ -190,6 +199,15 @@ class FocusLogger(CustomLogger): window = self._compute_time_window(datetime.now(timezone.utc)) await self._export_window(window=window, limit=None) + async def _export_all( + self, + *, + limit: Optional[int], + ) -> None: + """Export all available data without a time window filter.""" + engine = self._ensure_engine() + await engine.export_all(limit=limit) + async def _export_window( self, *, @@ -220,4 +238,5 @@ class FocusLogger(CustomLogger): frequency=self.frequency, ) + __all__ = ["FocusLogger"]