id: cloud-llm-finops-circuit-breaker
namespace: company.finops
labels:
domain: finops-governance
architecture: circuit-breaker
owner: platform-economics
inputs:
- id: tenant_cluster
type: STRING
defaults: "prod-ai-inference-us-east-1"
description: "Target Kubernetes or AI inference cluster identifier being audited"
- id: daily_budget_limit_usd
type: FLOAT
defaults: 850.00
description: "Maximum allowable daily cloud + LLM spend in USD before hard throttling"
- id: observed_daily_spend_usd
type: FLOAT
defaults: 615.40
description: "Current cumulative 24-hour spend in USD reported by metering collectors"
- id: baseline_daily_mean_usd
type: FLOAT
defaults: 520.00
description: "Rolling 30-day historical mean daily spend in USD for this cluster"
- id: baseline_daily_stddev_usd
type: FLOAT
defaults: 65.00
description: "Rolling 30-day historical standard deviation in USD for Z-score
calculation"
- id: prompt_cache_hit_ratio
type: FLOAT
defaults: 0.68
description: "Observed semantic/prefix prompt cache hit rate across LLM
inference gateways (0.0 to 1.0)"
- id: min_cache_efficiency_target
type: FLOAT
defaults: 0.45
description: "Minimum acceptable prompt cache hit ratio before flagging caching
regression"
- id: z_score_circuit_threshold
type: FLOAT
defaults: 2.50
description: "Statistical Z-score ceiling above which spend velocity is
classified as anomalous"
- id: enforce_hard_throttle
type: BOOL
defaults: true
description: "Whether to automatically apply model-tier downgrade policies when
the circuit breaker trips"
triggers:
- id: hourly_burn_rate_audit
type: io.kestra.plugin.core.trigger.Schedule
cron: "0 * * * *"
- id: billing_anomaly_webhook
type: io.kestra.plugin.core.trigger.Webhook
key: "finops-budget-alert-webhook-key"
tasks:
# ---------------------------------------------------------------------------
# STAGE 1: Audit Context Initialization
# ---------------------------------------------------------------------------
- id: initialize_finops_audit
type: io.kestra.plugin.core.log.Log
description: "Record audit context, target cluster, and active budget guardrails"
level: INFO
message: |
[FINOPS CIRCUIT BREAKER] Initializing burn-rate & token telemetry audit
Cluster: {{ inputs.tenant_cluster }} | Daily Cap: ${{ inputs.daily_budget_limit_usd }}
Baseline Mean: ${{ inputs.baseline_daily_mean_usd }} (StdDev:${{ inputs.baseline_daily_stddev_usd }})
# ---------------------------------------------------------------------------
# STAGE 2: Parallel Metering & Pricing Endpoint Pre-Flight Probes
# ---------------------------------------------------------------------------
- id: parallel_telemetry_preflight
type: io.kestra.plugin.core.flow.Parallel
description: "Concurrently verify pricing feed endpoints and regional telemetry
collectors"
tasks:
- id: check_primary_pricing_api
type: io.kestra.plugin.core.http.Request
description: "Verify primary cloud pricing catalog API availability"
uri: "https://jsonplaceholder.typicode.com/posts/1"
method: GET
- id: check_fallback_metering_api
type: io.kestra.plugin.core.http.Request
description: "Verify LLM gateway token metering collector health"
uri: "https://jsonplaceholder.typicode.com/posts/2"
method: GET
- id: check_quota_controller_health
type: io.kestra.plugin.core.http.Request
description: "Verify cluster rate-limiter and quota controller endpoint"
uri: "https://jsonplaceholder.typicode.com/todos/1"
method: GET
- id: log_collector_readiness
type: io.kestra.plugin.core.log.Log
description: "Emit pre-flight collector synchronization heartbeat"
level: INFO
message: "All regional cost & token metering endpoints responded HTTP 200."
# ---------------------------------------------------------------------------
# STAGE 3: In-Memory DuckDB Multi-CTE Cost Attribution Rollup
# ---------------------------------------------------------------------------
- id: rollup_hourly_cost_ledger
type: io.kestra.plugin.jdbc.duckdb.Query
description: "Execute multi-CTE analytical cost attribution rollup across
compute and LLM workloads"
sql: |
WITH raw_usage_events AS (
SELECT
'svc-rag-indexer' AS workload,
'llm-tokens' AS category,
'tier-1-production' AS cost_center,
1420000 AS consumed_units,
184.60 AS cost_usd,
0.74 AS cache_ratio
UNION ALL
SELECT
'svc-agent-router' AS workload,
'llm-tokens' AS category,
'tier-1-production' AS cost_center,
980000 AS consumed_units,
147.00 AS cost_usd,
0.61 AS cache_ratio
UNION ALL
SELECT
'gpu-vllm-worker-pool' AS workload,
'gpu-compute' AS category,
'tier-1-production' AS cost_center,
48 AS consumed_units,
192.00 AS cost_usd,
1.00 AS cache_ratio
UNION ALL
SELECT
'vector-db-replicas' AS workload,
'storage-iops' AS category,
'tier-2-shared' AS cost_center,
320 AS consumed_units,
64.80 AS cost_usd,
0.89 AS cache_ratio
UNION ALL
SELECT
'nat-egress-gateway' AS workload,
'network' AS category,
'tier-2-shared' AS cost_center,
540 AS consumed_units,
27.00 AS cost_usd,
1.00 AS cache_ratio
),
category_summary AS (
SELECT
category,
COUNT(*) AS active_workloads,
SUM(consumed_units) AS total_units,
ROUND(SUM(cost_usd), 2) AS category_spend_usd,
ROUND(AVG(cache_ratio), 4) AS avg_category_efficiency
FROM raw_usage_events
GROUP BY category
)
SELECT
category,
active_workloads,
total_units,
category_spend_usd,
avg_category_efficiency
FROM category_summary
ORDER BY category_spend_usd DESC;
fetchType: FETCH
# ---------------------------------------------------------------------------
# STAGE 4: Python Statistical Z-Score & Burn-Rate Forecasting Engine
# ---------------------------------------------------------------------------
- id: evaluate_burn_rate_forecast
type: io.kestra.plugin.scripts.python.Script
description: "Run statistical Z-score anomaly detection, runway projection, and
mitigation planning"
beforeCommands:
- pip install kestra
script: |
from kestra import Kestra
# Ingest runtime parameters and DuckDB rollup cardinality
tracked_categories = {{ outputs.rollup_hourly_cost_ledger.size }}
cluster_name = "{{ inputs.tenant_cluster }}"
daily_cap = float({{ inputs.daily_budget_limit_usd }})
current_spend = float({{ inputs.observed_daily_spend_usd }})
hist_mean = float({{ inputs.baseline_daily_mean_usd }})
hist_std = float({{ inputs.baseline_daily_stddev_usd }})
cache_hit = float({{ inputs.prompt_cache_hit_ratio }})
min_cache = float({{ inputs.min_cache_efficiency_target }})
z_limit = float({{ inputs.z_score_circuit_threshold }})
# 1. Calculate statistical Z-score against rolling 30-day baseline
safe_std = hist_std if hist_std > 0 else 1.0
z_score = round((current_spend - hist_mean) / safe_std, 3)
# 2. Compute budget utilization and projected 30-day runway
utilization_pct = round((current_spend / daily_cap) * 100.0, 2)
projected_monthly_usd = round(current_spend * 30.0, 2)
monthly_cap_usd = round(daily_cap * 30.0, 2)
variance_usd = round(projected_monthly_usd - monthly_cap_usd, 2)
# 3. Evaluate multi-factor circuit breaker conditions
violation_reasons = []
if current_spend > daily_cap:
violation_reasons.append(
f"Hard budget cap exceeded (${current_spend:.2f} >${daily_cap:.2f})"
)
if z_score > z_limit:
violation_reasons.append(
f"Anomalous burn velocity (Z-score {z_score} > {z_limit})"
)
if cache_hit < min_cache:
violation_reasons.append(
f"Prompt cache degradation ({cache_hit:.0%} < {min_cache:.0%})"
)
trip_breaker = len(violation_reasons) > 0
# 4. Select governance state and automated remediation policy
if not trip_breaker and utilization_pct < 85.0:
governance_state = "OPTIMAL_WITHIN_BUDGET"
recommended_policy = (
"Maintain standard autoscaling and primary LLM routing tiers."
)
elif not trip_breaker and utilization_pct >= 85.0:
governance_state = "ELEVATED_BURN_WARNING"
recommended_policy = (
"Enable aggressive semantic caching and notify FinOps on-call."
)
else:
governance_state = "CIRCUIT_BREAKER_TRIPPED"
recommended_policy = (
"Downgrade non-critical LLM routes to mini-tier and cap GPU replicas."
)
audit_report = (
f"[{governance_state}] Cluster={cluster_name} | "
f"Categories={tracked_categories} | "
f"Spend=${current_spend:.2f}/${daily_cap:.2f} ({utilization_pct}%) | "
f"Z-Score={z_score} | CacheHit={cache_hit:.0%} | "
f"30dForecast=${projected_monthly_usd:.2f}"
)
print(audit_report)
if violation_reasons:
print("Triggered violations: " + "; ".join(violation_reasons))
Kestra.outputs({
'trip_breaker': trip_breaker,
'governance_state': governance_state,
'utilization_pct': utilization_pct,
'z_score': z_score,
'projected_monthly_usd': projected_monthly_usd,
'variance_usd': variance_usd,
'recommended_policy': recommended_policy,
'audit_report': audit_report
})
# ---------------------------------------------------------------------------
# STAGE 5: Sequential Departmental Chargeback Ledger Synchronization
# ---------------------------------------------------------------------------
- id: sequential_ledger_synchronization
type: io.kestra.plugin.core.flow.Sequential
description: "Sequentially update chargeback watermarks across engineering cost centers"
tasks:
- id: sync_ai_platform_ledger
type: io.kestra.plugin.core.log.Log
description: "Update AI Platform token attribution watermark"
level: INFO
message: |
[CHARGEBACK LEDGER 1/3] AI Platform Cost Center synced -> State: {{ outputs.evaluate_burn_rate_forecast.vars.governance_state }}
- id: sync_core_infra_ledger
type: io.kestra.plugin.core.log.Log
description: "Update Core Kubernetes GPU/Compute utilization ledger"
level: INFO
message: |
[CHARGEBACK LEDGER 2/3] Core Kubernetes Compute synced -> Utilization: {{ outputs.evaluate_burn_rate_forecast.vars.utilization_pct }}%
- id: sync_finance_forecast_table
type: io.kestra.plugin.core.log.Log
description: "Record rolling 30-day run-rate projection for Finance reporting"
level: INFO
message: |
[CHARGEBACK LEDGER 3/3] 30-Day Run-Rate Forecast updated -> Projected: ${{ outputs.evaluate_burn_rate_forecast.vars.projected_monthly_usd }}/mo
# ---------------------------------------------------------------------------
# STAGE 6: Conditional Circuit-Breaker Enforcement & Compliance Routing
# ---------------------------------------------------------------------------
- id: route_governance_enforcement
type: io.kestra.plugin.core.flow.If
description: "Branch execution between automated rate-limiting mitigation and
healthy budget certification"
condition: "{{ outputs.evaluate_burn_rate_forecast.vars.trip_breaker == true }}"
then:
- id: apply_model_tier_downgrade
type: io.kestra.plugin.core.log.Log
description: "Execute automated fallback policy to curb runaway token/compute burn"
level: WARN
message: |
[MITIGATION STEP 1/2] Activating token rate-limiter on {{ inputs.tenant_cluster }}.
Policy Applied: {{ outputs.evaluate_burn_rate_forecast.vars.recommended_policy }}
- id: dispatch_finops_escalation
type: io.kestra.plugin.core.log.Log
description: "Emit high-priority anomaly escalation payload for engineering
leadership"
level: WARN
message: |
[MITIGATION STEP 2/2] Escalating budget anomaly alert to Engineering & Finance leads.
Summary: {{ outputs.evaluate_burn_rate_forecast.vars.audit_report }}
else:
- id: certify_budget_compliance
type: io.kestra.plugin.core.log.Log
description: "Certify that cluster spend velocity and cache efficiency meet SLA
guardrails"
level: INFO
message: |
[COMPLIANCE VERIFIED 1/2] Cluster {{ inputs.tenant_cluster }} operating within FinOps guardrails.
Summary: {{ outputs.evaluate_burn_rate_forecast.vars.audit_report }}
- id: archive_clean_window_snapshot
type: io.kestra.plugin.core.log.Log
description: "Commit verified hourly cost attribution record to governance ledger"
level: INFO
message: |
[COMPLIANCE VERIFIED 2/2] Archived hourly cost attribution snapshot (Z-Score: {{ outputs.evaluate_burn_rate_forecast.vars.z_score }}).
# ---------------------------------------------------------------------------
# STAGE 7: Final Governance Ledger Closure
# ---------------------------------------------------------------------------
- id: close_finops_governance_run
type: io.kestra.plugin.core.log.Log
description: "Finalize audit run and emit execution completion summary"
level: INFO
message: |
Completed FinOps governance execution {{ execution.id }} -> Final Status: {{ outputs.evaluate_burn_rate_forecast.vars.governance_state }}
