Coverage for cli / commands / costs_cmd.py: 87%
160 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-30 21:47 +0000
1"""Cost tracking commands."""
3import sys
4from typing import Any
6import click
8from ..config import GCOConfig, _load_cdk_json
9from ..output import get_output_formatter
11pass_config = click.make_pass_decorator(GCOConfig, ensure=True)
14def _get_deployment_regions(config: GCOConfig) -> list[str]:
15 """Get the list of regional deployment regions from cdk.json or fallback to default."""
16 cdk_regions = _load_cdk_json()
17 if cdk_regions and "regional" in cdk_regions:
18 regional = cdk_regions["regional"]
19 if isinstance(regional, list) and all(isinstance(r, str) for r in regional):
20 return regional
21 return [config.default_region]
24@click.group()
25@pass_config
26def costs(config: Any) -> None:
27 """View cost breakdowns and estimates for GCO resources."""
28 pass
31@costs.command("summary")
32@click.option(
33 "--days", "-d", default=30, type=int, help="Number of days to look back (default: 30)"
34)
35@click.option(
36 "--all", "show_all", is_flag=True, help="Show all account costs (not filtered by GCO tag)"
37)
38@pass_config
39def costs_summary(config: Any, days: Any, show_all: Any) -> None:
40 """Show total GCO spend by service.
42 Examples:
43 gco costs summary
44 gco costs summary --days 7
45 gco costs summary --all # All account costs (useful before tags propagate)
46 """
47 from ..costs import get_cost_tracker
49 formatter = get_output_formatter(config)
51 try:
52 tracker = get_cost_tracker(config)
53 summary = tracker.get_cost_summary(days=days, unfiltered=show_all)
54 label = "Account" if show_all else "GCO"
56 if config.output_format != "table":
57 formatter.print(
58 {
59 "total": summary.total,
60 "currency": summary.currency,
61 "period_start": summary.period_start,
62 "period_end": summary.period_end,
63 "by_service": [
64 {"service": s.service, "amount": s.amount} for s in summary.by_service
65 ],
66 }
67 )
68 return
70 print(f"\n {label} Cost Summary ({summary.period_start} to {summary.period_end})")
71 print(" " + "-" * 75)
72 print(f" {'SERVICE':<50} {'COST':>12}")
73 print(" " + "-" * 75)
75 for svc in summary.by_service:
76 print(f" {svc.service:<50} ${svc.amount:>10.2f}")
78 print(" " + "-" * 75)
79 print(f" {'TOTAL':<50} ${summary.total:>10.2f}")
80 print()
82 except Exception as e:
83 formatter.print_error(f"Failed to get cost summary: {e}")
84 sys.exit(1)
87@costs.command("regions")
88@click.option(
89 "--days", "-d", default=30, type=int, help="Number of days to look back (default: 30)"
90)
91@pass_config
92def costs_regions(config: Any, days: Any) -> None:
93 """Show cost breakdown by region.
95 Examples:
96 gco costs regions
97 gco costs regions --days 7
98 """
99 from ..costs import get_cost_tracker
101 formatter = get_output_formatter(config)
103 try:
104 tracker = get_cost_tracker(config)
105 by_region = tracker.get_cost_by_region(days=days)
107 if config.output_format != "table":
108 formatter.print(by_region)
109 return
111 total = sum(by_region.values())
112 print(f"\n GCO Cost by Region (last {days} days)")
113 print(" " + "-" * 50)
114 print(f" {'REGION':<30} {'COST':>12}")
115 print(" " + "-" * 50)
117 for region, amount in by_region.items():
118 pct = (amount / total * 100) if total > 0 else 0
119 print(f" {region:<30} ${amount:>10.2f} ({pct:.0f}%)")
121 print(" " + "-" * 50)
122 print(f" {'TOTAL':<30} ${total:>10.2f}")
123 print()
125 except Exception as e:
126 formatter.print_error(f"Failed to get regional costs: {e}")
127 sys.exit(1)
130@costs.command("trend")
131@click.option("--days", "-d", default=14, type=int, help="Number of days (default: 14)")
132@click.option(
133 "--all", "show_all", is_flag=True, help="Show all account costs (not filtered by GCO tag)"
134)
135@pass_config
136def costs_trend(config: Any, days: Any, show_all: Any) -> None:
137 """Show daily cost trend.
139 Examples:
140 gco costs trend
141 gco costs trend --days 7
142 gco costs trend --all
143 """
144 from ..costs import get_cost_tracker
146 formatter = get_output_formatter(config)
148 try:
149 tracker = get_cost_tracker(config)
150 trend = tracker.get_daily_trend(days=days, unfiltered=show_all)
151 label = "Account" if show_all else "GCO"
153 if config.output_format != "table":
154 formatter.print(trend)
155 return
157 print(f"\n Daily Cost Trend — {label} (last {days} days)")
158 print(" " + "-" * 45)
159 print(f" {'DATE':<15} {'COST':>10} {'CHART'}")
160 print(" " + "-" * 45)
162 max_amount = max((d["amount"] for d in trend), default=1) or 1
163 for day in trend:
164 bar_len = int(day["amount"] / max_amount * 25)
165 bar = "█" * bar_len
166 print(f" {day['date']:<15} ${day['amount']:>8.2f} {bar}")
168 total = sum(d["amount"] for d in trend)
169 avg = total / len(trend) if trend else 0
170 print(" " + "-" * 45)
171 print(f" Total: ${total:.2f} | Avg/day: ${avg:.2f}")
172 print()
174 except Exception as e:
175 formatter.print_error(f"Failed to get cost trend: {e}")
176 sys.exit(1)
179@costs.command("workloads")
180@click.option("--region", "-r", help="Region to check (default: all deployment regions)")
181@pass_config
182def costs_workloads(config: Any, region: Any) -> None:
183 """Estimate costs for running workloads (jobs and inference endpoints).
185 Examples:
186 gco costs workloads
187 gco costs workloads -r us-east-1
188 """
189 from ..costs import get_cost_tracker
191 formatter = get_output_formatter(config)
193 try:
194 tracker = get_cost_tracker(config)
196 regions = [region] if region else _get_deployment_regions(config)
197 all_workloads = []
199 for r in regions:
200 workloads = tracker.estimate_running_workloads(r)
201 all_workloads.extend(workloads)
203 if config.output_format != "table":
204 formatter.print(
205 [
206 {
207 "name": w.name,
208 "type": w.workload_type,
209 "instance_type": w.instance_type,
210 "gpu_count": w.gpu_count,
211 "hourly_rate": w.hourly_rate,
212 "runtime_hours": w.runtime_hours,
213 "estimated_cost": w.estimated_cost,
214 "region": w.region,
215 }
216 for w in all_workloads
217 ]
218 )
219 return
221 if not all_workloads: 221 ↛ 225line 221 didn't jump to line 225 because the condition on line 221 was always true
222 formatter.print_info("No running workloads found")
223 return
225 print(f"\n Running Workload Costs ({len(all_workloads)} workloads)")
226 print(" " + "-" * 95)
227 print(
228 f" {'NAME':<30} {'TYPE':<10} {'INSTANCE':<15} {'GPU':>3} {'$/HR':>8} {'HOURS':>7} {'COST':>10}"
229 )
230 print(" " + "-" * 95)
232 total = 0.0
233 for w in sorted(all_workloads, key=lambda x: x.estimated_cost, reverse=True):
234 name = w.name[:29]
235 print(
236 f" {name:<30} {w.workload_type:<10} {w.instance_type:<15} "
237 f"{w.gpu_count:>3} ${w.hourly_rate:>7.3f} {w.runtime_hours:>7.1f} ${w.estimated_cost:>9.4f}"
238 )
239 total += w.estimated_cost
241 total_hourly = sum(w.hourly_rate for w in all_workloads)
242 print(" " + "-" * 95)
243 print(
244 f" {'TOTAL':<30} {'':10} {'':15} {'':>3} ${total_hourly:>7.3f} {'':>7} ${total:>9.4f}"
245 )
246 print()
248 except Exception as e:
249 formatter.print_error(f"Failed to estimate workload costs: {e}")
250 sys.exit(1)
253@costs.command("forecast")
254@click.option("--days", "-d", default=30, type=int, help="Days to forecast (default: 30)")
255@pass_config
256def costs_forecast(config: Any, days: Any) -> None:
257 """Forecast GCO costs for the next N days.
259 Examples:
260 gco costs forecast
261 gco costs forecast --days 60
262 """
263 from ..costs import get_cost_tracker
265 formatter = get_output_formatter(config)
267 try:
268 tracker = get_cost_tracker(config)
269 forecast = tracker.get_forecast(days_ahead=days)
271 if "error" in forecast:
272 formatter.print_error(f"Forecast unavailable: {forecast['error']}")
273 formatter.print_info("Cost Explorer needs 14+ days of data to generate forecasts")
274 return
276 if config.output_format != "table": 276 ↛ 277line 276 didn't jump to line 277 because the condition on line 276 was never true
277 formatter.print(forecast)
278 return
280 total = forecast.get("forecast_total", 0)
281 print(f"\n Cost Forecast ({forecast['period_start']} to {forecast['period_end']})")
282 print(" " + "-" * 40)
283 print(f" Projected spend: ${total:>10.2f}")
284 print(f" Daily average: ${total / days:>10.2f}")
285 print()
287 except Exception as e:
288 formatter.print_error(f"Failed to get forecast: {e}")
289 sys.exit(1)