diff --git a/codecarbon/emissions_tracker.py b/codecarbon/emissions_tracker.py index 8d0f00a4c..c48682ffc 100644 --- a/codecarbon/emissions_tracker.py +++ b/codecarbon/emissions_tracker.py @@ -968,7 +968,12 @@ def finish_http_request( task_emission_data = dataclasses.replace(emissions_at_stop) request_duration = time.perf_counter() - baseline.started_at - task_emission_data.duration = Time.from_seconds(request_duration).seconds + # compute_delta_emission subtracts previous.duration, so pass the + # absolute elapsed here; the delta is then the request duration and + # emissions_rate is computed against it (not against a negative). + task_emission_data.duration = ( + baseline.duration_at_start + Time.from_seconds(request_duration).seconds + ) task_emission_data.compute_delta_emission(previous) task.emissions_data = task_emission_data @@ -996,6 +1001,22 @@ def persist_completed_task(self, task_name: str) -> None: handler.task_out(task_payload, self._experiment_name) task.uploaded_to_api = True + def discard_task(self, task_name: str) -> None: + """Drop a finished task record so ``_tasks`` stays bounded. + + Long-lived servers create one task per HTTP request; without eviction + ``_tasks`` grows for the process lifetime. Call after + :meth:`persist_completed_task` (data already left the tracker by then). + + Args: + task_name: Name of the task to forget. Unknown names are ignored. + """ + with self._http_task_lock: + task = self._tasks.get(task_name) + if task is None or task.is_active: + return + del self._tasks[task_name] + @suppress(Exception) def flush(self) -> Optional[float]: """ diff --git a/codecarbon/integrations/fastapi/__init__.py b/codecarbon/integrations/fastapi/__init__.py index 8a3aa36c8..59fe0c4ce 100644 --- a/codecarbon/integrations/fastapi/__init__.py +++ b/codecarbon/integrations/fastapi/__init__.py @@ -11,6 +11,13 @@ log_request_complete, shutdown_codecarbon_middleware, ) + from codecarbon.integrations.fastapi.tiers import ( + EndpointTotals, + MeasurementTier, + RequestMeasurement, + TierDetection, + detect_measurement_tier, + ) except ImportError as exc: raise ImportError( "CodeCarbon FastAPI integration requires Starlette (installed with FastAPI). " @@ -19,6 +26,11 @@ __all__ = [ "CodeCarbonMiddleware", + "EndpointTotals", + "MeasurementTier", + "RequestMeasurement", + "TierDetection", + "detect_measurement_tier", "add_codecarbon_middleware", "compose_lifespans", "create_codecarbon_lifespan", diff --git a/codecarbon/integrations/fastapi/middleware.py b/codecarbon/integrations/fastapi/middleware.py index a9c1241ac..255212f45 100644 --- a/codecarbon/integrations/fastapi/middleware.py +++ b/codecarbon/integrations/fastapi/middleware.py @@ -4,6 +4,7 @@ import asyncio import collections +import dataclasses import threading from collections.abc import Awaitable, Callable, Iterable, Sequence from concurrent import futures @@ -21,6 +22,13 @@ build_endpoint_key, should_track_request, ) +from codecarbon.integrations.fastapi.tiers import ( + EndpointTotals, + MeasurementTier, + RequestMeasurement, + TierDetection, + detect_measurement_tier, +) from codecarbon.output_methods.emissions_data import EmissionsData DEFAULT_TRACKER_KWARGS: dict[str, Any] = { @@ -65,21 +73,75 @@ def _resolve_header_fields( def _inject_emission_headers( message: Message, - emissions_data: EmissionsData | None, + measurement: RequestMeasurement | None, fields: Sequence[str], ) -> Message: - if not fields or emissions_data is None: + """Add ``X-CodeCarbon-*`` headers. + + Headers are opt-in (``response_headers``); when enabled they always carry + ``X-CodeCarbon-Tier`` next to the numbers, and report ``unavailable`` + rather than ``0`` when the backend cannot resolve the request. + """ + if not fields or measurement is None: return message headers = list(message.get("headers", [])) + headers.append((b"X-CodeCarbon-Tier", measurement.tier.value.encode("latin-1"))) + emissions_data = measurement.emissions_data for field in fields: - if not hasattr(emissions_data, field): - continue name = _codecarbon_header_name(field) - value = str(getattr(emissions_data, field)) + if emissions_data is None: + value = "unavailable" + elif hasattr(emissions_data, field): + value = str(getattr(emissions_data, field)) + else: + continue headers.append((name.encode("latin-1"), value.encode("latin-1"))) return {**message, "headers": headers} +def _carbon_intensity(tracker: EmissionsTracker | None) -> float: + """Effective kg CO2 per kWh implied by the tracker's running totals.""" + total_energy = getattr(getattr(tracker, "_total_energy", None), "kWh", 0.0) + total_emissions = getattr(tracker, "_total_emissions", 0.0) + try: + if total_energy and total_energy > 0: + return float(total_emissions) / float(total_energy) + except (TypeError, ValueError): + pass + return 0.0 + + +def _estimate_from_duration( + emissions_data: EmissionsData, intensity: float +) -> EmissionsData | None: + """ESTIMATED tier: energy is analytic (``P x elapsed``) at any resolution. + + The sampled delta can still be 0 when the request fell between scheduler + ticks; with a constant power model the honest figure is power x duration + rather than 0. Returns ``None`` when carbon intensity is not yet known, + so the caller reports unavailable instead of a zero. + """ + duration = emissions_data.duration or 0.0 + if duration <= 0 or intensity <= 0: + return None + hours = duration / 3600.0 + cpu = (emissions_data.cpu_power or 0.0) / 1000.0 * hours + gpu = (emissions_data.gpu_power or 0.0) / 1000.0 * hours + ram = (emissions_data.ram_power or 0.0) / 1000.0 * hours + energy = cpu + gpu + ram + if energy <= 0: + return None + return dataclasses.replace( + emissions_data, + cpu_energy=cpu, + gpu_energy=gpu, + ram_energy=ram, + energy_consumed=energy, + emissions=energy * intensity, + emissions_rate=energy * intensity / duration, + ) + + class _TrackerRunner: """Single tracker thread: request-path jobs first, then pending finalization.""" @@ -242,6 +304,90 @@ def __init__( self._app_tracker: EmissionsTracker | None = None self._tracker_init_lock = threading.Lock() self._tracker_runner = _TrackerRunner() + self._tier_detection: TierDetection | None = None + self._totals: dict[str, EndpointTotals] = {} + self._totals_lock = threading.Lock() + + @property + def measurement_tier(self) -> MeasurementTier | None: + """Tier resolved from the tracker's hardware, or ``None`` before first use.""" + return None if self._tier_detection is None else self._tier_detection.tier + + @property + def tier_detection(self) -> TierDetection | None: + """Full detection result (overall tier plus per-component tiers).""" + return self._tier_detection + + def endpoint_totals(self) -> dict[str, EndpointTotals]: + """Per-endpoint aggregates. The only energy output in AGGREGATE_ONLY. + + Idle/baseline power is charged to requests, not subtracted, so these + totals sum to the tracker total. + """ + with self._totals_lock: + return { + key: dataclasses.replace(value) for key, value in self._totals.items() + } + + def _resolve_tier(self, tracker: EmissionsTracker) -> TierDetection: + """Resolve the tier once, from the hardware the tracker actually detected.""" + detection = self._tier_detection + if detection is None: + detection = detect_measurement_tier(getattr(tracker, "_hardware", None)) + self._tier_detection = detection + logger.debug("CodeCarbon measurement tier: %s", detection.describe()) + return detection + + def _build_measurement( + self, + detection: TierDetection, + endpoint: str, + task_name: str, + emissions_data: EmissionsData | None, + tracker: EmissionsTracker | None = None, + ) -> RequestMeasurement: + tier = detection.tier + duration = float(getattr(emissions_data, "duration", 0.0) or 0.0) + if emissions_data is None: + return RequestMeasurement( + tier, task_name, endpoint, duration, None, "no measurement returned" + ) + if tier is MeasurementTier.AGGREGATE_ONLY: + return RequestMeasurement( + tier, + task_name, + endpoint, + duration, + None, + "backend resolution is coarser than a request; use endpoint_totals()", + ) + if tier is MeasurementTier.ESTIMATED and not emissions_data.energy_consumed: + emissions_data = _estimate_from_duration( + emissions_data, _carbon_intensity(tracker) + ) + if emissions_data is None or not emissions_data.energy_consumed: + return RequestMeasurement( + tier, + task_name, + endpoint, + duration, + None, + "request spanned no completed sampling window", + ) + return RequestMeasurement(tier, task_name, endpoint, duration, emissions_data) + + def _record_totals( + self, + detection: TierDetection, + endpoint: str, + emissions_data: EmissionsData | None, + ) -> None: + with self._totals_lock: + totals = self._totals.get(endpoint) + if totals is None: + totals = EndpointTotals(endpoint=endpoint, tier=detection.tier) + self._totals[endpoint] = totals + totals.add(emissions_data) def shutdown_tracker_executor(self, *, wait: bool = True) -> None: """Shut down the tracker background thread (idempotent). @@ -328,7 +474,8 @@ def _finalize_on_worker( response: Response, run_callback: bool, baseline: HttpRequestBaseline | None, - ) -> EmissionsData | None: + ) -> RequestMeasurement: + detection = self._resolve_tier(tracker) if baseline is not None: emissions_data = tracker.finish_http_request(baseline) resolved_task = baseline.task_name @@ -337,9 +484,24 @@ def _finalize_on_worker( resolved_task = active_task if isinstance(active_task, str) else task_name emissions_data = tracker.stop_task(resolved_task) tracker.persist_completed_task(resolved_task) + # Bound tracker._tasks: the record is persisted, keeping it would leak + # one Task per request for the lifetime of the server. + discard = getattr(tracker, "discard_task", None) + if callable(discard): + discard(resolved_task) + self._record_totals(detection, task_name, emissions_data) + measurement = self._build_measurement( + detection, task_name, resolved_task, emissions_data, tracker + ) + try: + request.state.codecarbon = measurement + except Exception: # pragma: no cover - exotic scopes without state + logger.debug("CodeCarbon: could not attach measurement to request.state") if run_callback: - self._run_request_complete(request, response, emissions_data, resolved_task) - return emissions_data + self._run_request_complete( + request, response, measurement.emissions_data, resolved_task + ) + return measurement def _run_request_complete( self, @@ -370,7 +532,7 @@ async def _finalize_after_response( baseline: HttpRequestBaseline | None, *, run_callback: bool, - ) -> EmissionsData | None: + ) -> RequestMeasurement: return await self._run_finalize_tracker( self._finalize_on_worker, tracker, @@ -513,7 +675,7 @@ async def send_wrapper(message: Message) -> None: return status_code = message["status"] response = Response(status_code=status_code) - emissions_data = await self._finalize_after_response( + measurement = await self._finalize_after_response( tracker, task_name, request, @@ -523,7 +685,7 @@ async def send_wrapper(message: Message) -> None: ) finalized = True await send( - _inject_emission_headers(message, emissions_data, self.header_fields) + _inject_emission_headers(message, measurement, self.header_fields) ) error: BaseException | None = None diff --git a/codecarbon/integrations/fastapi/tiers.py b/codecarbon/integrations/fastapi/tiers.py new file mode 100644 index 000000000..fa68c033d --- /dev/null +++ b/codecarbon/integrations/fastapi/tiers.py @@ -0,0 +1,186 @@ +"""Measurement capability tiers for per-request energy reporting. + +A backend can only report a per-request energy figure if it actually resolves +the request. Measured on real hardware: + +* ``intel_rapl``: counter updates ~1 ms, quantum 15.3 uJ, read cost ~10 us. + A 30 ms request is genuinely resolved -> MEASURED. +* ``constant`` (TDP): ``power = tdp * 0.5``, so energy is analytic in duration + and exact at any resolution, but it only restates wall time -> ESTIMATED. +* ``cpu_load``: ``psutil.cpu_percent(interval=None)`` is tick-quantized; 98% of + 5 ms reads return zero load and the ``0.1 + 0.9*(load/100)**3`` factor turns + that into a 10%-of-TDP floor (30 requests on a pegged core all reported + exactly 0.0900 J) -> AGGREGATE_ONLY. +* ``apple_powermetrics``: ``get_details()`` spawns ``sudo powermetrics -n 10 + -i 100`` and blocks ~1 s -> cannot run in a request path -> AGGREGATE_ONLY. +* NVML: sensor period 20 ms (V100) to ~100 ms with a 25 ms averaging window + (A100/H100); sub-100 ms GPU energy is not obtainable. Only usable when + aggregating over >= 1 s. +* ``amdsmi``, ``windows_emi``: update rate undocumented. Treated as unprobed + -> AGGREGATE_ONLY. + +Idle/baseline power is charged to requests, not subtracted, so per-endpoint +totals still sum to the run total. +""" + +from __future__ import annotations + +import dataclasses +from enum import Enum +from typing import Any, Iterable + +from codecarbon.output_methods.emissions_data import EmissionsData + + +class MeasurementTier(str, Enum): + """What the detected hardware can honestly report per request.""" + + MEASURED = "measured" + ESTIMATED = "estimated" + AGGREGATE_ONLY = "aggregate_only" + + +_RANK = { + MeasurementTier.AGGREGATE_ONLY: 0, + MeasurementTier.ESTIMATED: 1, + MeasurementTier.MEASURED: 2, +} + +#: NVML only resolves energy when aggregated over at least this window. +NVML_MIN_AGGREGATION_SECONDS = 1.0 + +CPU_MODE_TIERS: dict[str, MeasurementTier] = { + "intel_rapl": MeasurementTier.MEASURED, + "constant": MeasurementTier.ESTIMATED, + "cpu_load": MeasurementTier.AGGREGATE_ONLY, + "apple_powermetrics": MeasurementTier.AGGREGATE_ONLY, + "intel_power_gadget": MeasurementTier.AGGREGATE_ONLY, + "windows_emi": MeasurementTier.AGGREGATE_ONLY, # unprobed update rate +} + + +def hardware_tier(hardware: Any, window_seconds: float = 0.0) -> MeasurementTier: + """Tier a single hardware component can support over ``window_seconds``. + + Duck-typed on purpose: tests inject stubs instead of depending on what the + test machine happens to have. + """ + mode = getattr(hardware, "_mode", None) + if isinstance(mode, str): + return CPU_MODE_TIERS.get(mode, MeasurementTier.AGGREGATE_ONLY) + + name = type(hardware).__name__ + if name == "GPU" or hasattr(hardware, "devices"): + return _gpu_tier(hardware, window_seconds) + if name == "RAM": + # RAM power is a constant model, so energy is analytic in duration. + return MeasurementTier.ESTIMATED + if name == "AppleSiliconChip": + return MeasurementTier.AGGREGATE_ONLY + return MeasurementTier.AGGREGATE_ONLY + + +def _gpu_tier(hardware: Any, window_seconds: float) -> MeasurementTier: + devices = getattr(getattr(hardware, "devices", None), "devices", None) + if not devices: + return MeasurementTier.AGGREGATE_ONLY + if not all("Nvidia" in type(device).__name__ for device in devices): + return MeasurementTier.AGGREGATE_ONLY # amdsmi: unprobed + if window_seconds >= NVML_MIN_AGGREGATION_SECONDS: + return MeasurementTier.MEASURED + return MeasurementTier.AGGREGATE_ONLY + + +@dataclasses.dataclass(frozen=True) +class TierDetection: + """Resolved tier plus the per-component tiers it was derived from.""" + + tier: MeasurementTier + components: tuple[tuple[str, MeasurementTier], ...] = () + + def describe(self) -> str: + parts = ", ".join(f"{name}={tier.value}" for name, tier in self.components) + return f"{self.tier.value} ({parts})" if parts else self.tier.value + + +def detect_measurement_tier( + hardware: Iterable[Any] | None, window_seconds: float = 0.0 +) -> TierDetection: + """Resolve the reporting tier from the hardware a tracker detected. + + The overall tier is the weakest component tier: a run is only MEASURED if + every energy contributor resolves the window. + """ + try: + components = tuple( + (repr_hardware(hw), hardware_tier(hw, window_seconds)) + for hw in (hardware or ()) + ) + except TypeError: # not iterable (e.g. a bare mock) + components = () + if not components: + return TierDetection(MeasurementTier.AGGREGATE_ONLY) + # RAM is analytic in duration, so it never limits resolution: it is + # reported but does not vote. + voting = [tier for name, tier in components if not name.startswith("RAM")] + if not voting: + voting = [tier for _, tier in components] + tier = min(voting, key=_RANK.__getitem__) + return TierDetection(tier, components) + + +def repr_hardware(hardware: Any) -> str: + mode = getattr(hardware, "_mode", None) + name = type(hardware).__name__ + return f"{name}[{mode}]" if isinstance(mode, str) else name + + +@dataclasses.dataclass(frozen=True) +class RequestMeasurement: + """Per-request result handed to consumers. + + ``emissions_data`` is ``None`` whenever a per-request energy figure would be + a fabrication: the AGGREGATE_ONLY tier, or a request that spanned no + completed sampling window. Zero is never reported in place of unknown. + """ + + tier: MeasurementTier + task_name: str + endpoint: str + duration: float + emissions_data: EmissionsData | None = None + unavailable_reason: str | None = None + + @property + def available(self) -> bool: + return self.emissions_data is not None + + @property + def emissions(self) -> float | None: + return None if self.emissions_data is None else self.emissions_data.emissions + + @property + def energy_consumed(self) -> float | None: + if self.emissions_data is None: + return None + return self.emissions_data.energy_consumed + + +@dataclasses.dataclass +class EndpointTotals: + """Aggregate per endpoint. Valid in every tier, including AGGREGATE_ONLY.""" + + endpoint: str + tier: MeasurementTier + count: int = 0 + duration: float = 0.0 + energy_consumed: float = 0.0 + emissions: float = 0.0 + + def add(self, emissions_data: EmissionsData | None) -> None: + self.count += 1 + if emissions_data is None: + return + self.duration += emissions_data.duration or 0.0 + self.energy_consumed += emissions_data.energy_consumed or 0.0 + self.emissions += emissions_data.emissions or 0.0 diff --git a/docs/how-to/fastapi.md b/docs/how-to/fastapi.md index 8f461ac64..84f9962eb 100644 --- a/docs/how-to/fastapi.md +++ b/docs/how-to/fastapi.md @@ -182,7 +182,7 @@ add_codecarbon_middleware( ## `response_headers`, `include_background_tasks`, `task_name_formatter`, `on_request_complete` -- **`response_headers`** — `True` (emissions only) or a list of field names (`emissions`, `duration`, `energy_consumed`, …). Measures before the response starts and sets `X-CodeCarbon-*` headers. Default `None` / off (deferred, no headers). +- **`response_headers`** — off by default (`None`); opt-in with `True` (emissions only) or a list of field names (`emissions`, `duration`, `energy_consumed`, …). Measures before the response starts and sets `X-CodeCarbon-*` headers, always including `X-CodeCarbon-Tier`. Values read `unavailable` when the tier cannot resolve the request. - **`include_background_tasks`** — default `True`: FastAPI/Starlette `BackgroundTasks` on the response are included. Set `False` to finalize at end of body and exclude post-body background work. - **`task_name_formatter`** — optional `(Request) -> str`; default is `METHOD /route/template`. Concurrent requests on the same route still get unique internal task IDs with `create_codecarbon_lifespan`. - **`on_request_complete`** — optional callback; default logs via `log_request_complete`; `None` disables it. @@ -195,6 +195,60 @@ add_codecarbon_middleware( ) ``` +## Measurement tiers + +A per-request energy figure is only honest if the backend actually resolves a +request. The middleware resolves a **tier** at startup from the hardware the +tracker detected, exposes it on `middleware.measurement_tier`, and puts it in +every result (`request.state.codecarbon`, and `X-CodeCarbon-Tier` when response +headers are enabled). + +| Tier | Backends | Per-request energy | +|---|---|---| +| `measured` | Linux RAPL (`intel_rapl`); NVML only when aggregating over >= 1 s | Reported | +| `estimated` | `constant` / TDP (`power = tdp * 0.5`), RAM | Reported, but analytic in duration: it restates wall time | +| `aggregate_only` | `cpu_load`, macOS `powermetrics`, `amdsmi`, Windows EMI, `intel_power_gadget` | **Not reported at all** — use `middleware.endpoint_totals()` | + +Why: + +- **Linux RAPL** — counter updates ~1 ms, quantum 15.3 uJ, read cost 10.4 us. A 30 ms request is genuinely resolved. +- **NVML** — sensor period 20 ms (V100) to ~100 ms with a 25 ms averaging window (A100/H100); 75-80% of elapsed time is unsampled. Sub-100 ms GPU energy is not obtainable. +- **macOS powermetrics** — `get_details()` spawns `sudo powermetrics -n 10 -i 100` and blocks ~1 s per read; it cannot sit in a request path. +- **`cpu_load`** — `psutil.cpu_percent(interval=None)` is tick-quantized: 98% of 5 ms reads return zero load, and the `0.1 + 0.9*(load/100)**3` factor turns that into a 10%-of-TDP floor. Measured: 30 requests on a core pegged at 100% all reported exactly 0.0900 J. +- **`amdsmi`, Windows EMI** — update rate undocumented; treated as unprobed. + +The overall tier is the weakest contributing component (RAM is analytic and +never limits resolution, so it does not vote). + +### Unavailable is not zero + +A request that spans no completed sampling window reports **unavailable**, never +`0.0`: zero is a factual claim that the request was free. In that case +`emissions_data` is `None`, the callback receives `None`, and headers read +`unavailable`. + +```python +def on_complete(request, response, emissions_data, task_name): + m = request.state.codecarbon # RequestMeasurement + if m.available: + print(m.tier.value, m.emissions, "kg") + else: + print(m.tier.value, "unavailable:", m.unavailable_reason) +``` + +### Per-endpoint totals + +Valid in every tier, and the only energy output in `aggregate_only`: + +```python +for endpoint, totals in app.state.codecarbon_middleware.endpoint_totals().items(): + print(endpoint, totals.count, totals.energy_consumed, totals.emissions) +``` + +Idle/baseline power is **charged to requests, not subtracted**. That keeps the +per-endpoint totals summing to the run total; it also means an idle server +attributes its floor to whatever traffic it did serve. + ## Middleware order Per [FastAPI middleware order](https://fastapi.tiangolo.com/tutorial/middleware/), the **last** middleware added is **outermost** on the request path (runs first on the way in). Add CodeCarbon **after** other middleware so it wraps inner layers and includes work done by inner middleware and route handlers: diff --git a/examples/fastapi_middleware.py b/examples/fastapi_middleware.py index 097c6adb3..45e815a56 100644 --- a/examples/fastapi_middleware.py +++ b/examples/fastapi_middleware.py @@ -34,11 +34,29 @@ async def lifespan(app: FastAPI): yield +def report(request, response, emissions_data, task_name): + """Report per request, honestly: never a zero standing in for unknown.""" + measurement = request.state.codecarbon + if measurement.available: + print( + f"{task_name}: tier={measurement.tier.value} " + f"emissions={measurement.emissions} kg " + f"energy={measurement.energy_consumed} kWh" + ) + else: + print( + f"{task_name}: tier={measurement.tier.value} " + f"emissions=unavailable ({measurement.unavailable_reason})" + ) + + app = FastAPI(title="CodeCarbon FastAPI demo", lifespan=lifespan) add_codecarbon_middleware( app, project_name="fastapi-demo", tracker_kwargs=_tracker_kwargs, + on_request_complete=report, + response_headers=True, # opt-in; adds X-CodeCarbon-Tier ) @@ -47,6 +65,18 @@ def predict(text: str = "hello"): return {"text": text, "label": "demo"} +@app.get("/totals") +def totals(): + """Per-endpoint aggregates: the only energy output in the aggregate_only tier.""" + middleware = app.state.codecarbon_middleware + return { + "tier": middleware.measurement_tier, + "endpoints": { + key: vars(value) for key, value in middleware.endpoint_totals().items() + }, + } + + # Stack other startup with compose_lifespans — see docs/how-to/fastapi.md # Per-request: codecarbon logger (INFO) after each response. # CSV: examples/output/emissions.csv on shutdown. diff --git a/tests/integrations/test_fastapi_middleware.py b/tests/integrations/test_fastapi_middleware.py index 84256ca26..fab1c67b4 100644 --- a/tests/integrations/test_fastapi_middleware.py +++ b/tests/integrations/test_fastapi_middleware.py @@ -3,6 +3,7 @@ from concurrent import futures from contextlib import asynccontextmanager from pathlib import Path +from types import SimpleNamespace from typing import Any from unittest.mock import MagicMock, patch @@ -19,6 +20,12 @@ shutdown_codecarbon_middleware, ) from codecarbon.integrations.fastapi.middleware import log_request_complete +from codecarbon.integrations.fastapi.tiers import MeasurementTier, RequestMeasurement + + +def measured_hardware() -> list[Any]: + """Injected hardware so tier detection never depends on the test machine.""" + return [SimpleNamespace(_mode="intel_rapl")] def _configure_mock_running_tracker( @@ -37,6 +44,7 @@ def mark_started() -> None: tracker_instance._start_time = None tracker_instance.mark_http_request_start.return_value = baseline tracker_instance.finish_http_request.return_value = MagicMock(emissions=emissions) + tracker_instance._hardware = measured_hardware() return baseline @@ -134,6 +142,7 @@ def test_middleware_uses_lifespan_tracker(MockTracker) -> None: application = FastAPI() tracker_instance = MagicMock() tracker_instance._start_time = 1.0 + tracker_instance._hardware = measured_hardware() baseline = MagicMock(task_name="GET /predict") emissions = MagicMock(emissions=0.003) tracker_instance.mark_http_request_start.return_value = baseline @@ -166,6 +175,7 @@ def test_middleware_skips_callback_when_handler_raises(MockTracker) -> None: application = FastAPI() tracker_instance = MagicMock() tracker_instance._start_time = 1.0 + tracker_instance._hardware = measured_hardware() baseline = MagicMock(task_name="GET /fail") tracker_instance.mark_http_request_start.return_value = baseline tracker_instance.finish_http_request.return_value = MagicMock(emissions=0.001) @@ -841,11 +851,19 @@ def test_resolve_header_fields_and_header_names() -> None: emissions = MagicMock(spec=["emissions", "duration"]) emissions.emissions = 0.0012 emissions.duration = 1.5 + measurement = RequestMeasurement( + tier=MeasurementTier.MEASURED, + task_name="GET /predict", + endpoint="GET /predict", + duration=1.5, + emissions_data=emissions, + ) injected = _inject_emission_headers( - message, emissions, ["emissions", "unknown_field", "duration"] + message, measurement, ["emissions", "unknown_field", "duration"] ) header_names = {name.decode() for name, _ in injected["headers"]} assert header_names == { + "X-CodeCarbon-Tier", "X-CodeCarbon-Emissions-kg", "X-CodeCarbon-Duration-s", } diff --git a/tests/integrations/test_fastapi_tiers.py b/tests/integrations/test_fastapi_tiers.py new file mode 100644 index 000000000..8aec7d151 --- /dev/null +++ b/tests/integrations/test_fastapi_tiers.py @@ -0,0 +1,376 @@ +"""Measurement tier tests. Every test injects known hardware/values. + +Nothing here depends on what the test machine has (see PR #1365: a test that +asserted `< 90 W` passed on Apple Silicon and failed on Linux CI at 280 W). +""" + +from __future__ import annotations + +import dataclasses +import threading +from types import SimpleNamespace +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +import codecarbon.integrations.fastapi.middleware as cc_middleware +from codecarbon.integrations.fastapi import ( + CodeCarbonMiddleware, + MeasurementTier, + add_codecarbon_middleware, + detect_measurement_tier, +) +from codecarbon.integrations.fastapi.tiers import ( + NVML_MIN_AGGREGATION_SECONDS, + hardware_tier, +) +from codecarbon.output_methods.emissions_data import EmissionsData + + +def _run_finalize_immediately(coro: Any) -> None: + import asyncio + from concurrent import futures + + def run_in_thread() -> None: + loop = asyncio.new_event_loop() + try: + loop.run_until_complete(coro) + finally: + loop.close() + + futures.ThreadPoolExecutor(max_workers=1).submit(run_in_thread).result() + + +@pytest.fixture +def finalize_deferred_immediately(): + """Deferred finalization runs inline so assertions see the result.""" + with patch.object( + cc_middleware.CodeCarbonMiddleware, + "_schedule_finalize", + side_effect=_run_finalize_immediately, + ): + yield + + +def _cpu(mode: str) -> Any: + return SimpleNamespace(_mode=mode) + + +class GPU: # stand-in for codecarbon.external.hardware.GPU + def __init__(self, devices: list[Any]) -> None: + self.devices = SimpleNamespace(devices=devices) + + +class RAM: # stand-in for codecarbon.external.ram.RAM + pass + + +def _gpu(device_class_name: str, count: int = 1) -> Any: + device_type = type(device_class_name, (), {}) + return GPU([device_type() for _ in range(count)]) + + +def _ram() -> Any: + return RAM() + + +def _emissions_data(**overrides: Any) -> EmissionsData: + fields = {f.name: 0.0 for f in dataclasses.fields(EmissionsData)} + fields.update( + { + "timestamp": "2026-01-01T00:00:00", + "project_name": "test", + "run_id": "run", + "experiment_id": "exp", + "country_name": "France", + "country_iso_code": "FRA", + "region": "", + "cloud_provider": "", + "cloud_region": "", + "os": "test", + "python_version": "3.13", + "codecarbon_version": "test", + "cpu_model": "test", + "gpu_model": "", + } + ) + fields.update(overrides) + return EmissionsData(**fields) + + +@pytest.mark.parametrize( + "mode,expected", + [ + ("intel_rapl", MeasurementTier.MEASURED), + ("constant", MeasurementTier.ESTIMATED), + ("cpu_load", MeasurementTier.AGGREGATE_ONLY), + ("apple_powermetrics", MeasurementTier.AGGREGATE_ONLY), + ("windows_emi", MeasurementTier.AGGREGATE_ONLY), + ("intel_power_gadget", MeasurementTier.AGGREGATE_ONLY), + ("something_new", MeasurementTier.AGGREGATE_ONLY), + ], +) +def test_cpu_mode_tiers(mode: str, expected: MeasurementTier) -> None: + assert hardware_tier(_cpu(mode)) is expected + + +def test_nvml_needs_a_one_second_window() -> None: + gpu = _gpu("NvidiaGPUDevice") + assert hardware_tier(gpu, window_seconds=0.03) is MeasurementTier.AGGREGATE_ONLY + assert ( + hardware_tier(gpu, window_seconds=NVML_MIN_AGGREGATION_SECONDS) + is MeasurementTier.MEASURED + ) + + +def test_amd_gpu_is_unprobed_even_when_aggregating() -> None: + assert hardware_tier(_gpu("AMDGPUDevice"), window_seconds=60) is ( + MeasurementTier.AGGREGATE_ONLY + ) + + +def test_ram_is_estimated_but_does_not_drag_rapl_down() -> None: + assert hardware_tier(_ram()) is MeasurementTier.ESTIMATED + detection = detect_measurement_tier([_cpu("intel_rapl"), _ram()]) + assert detection.tier is MeasurementTier.MEASURED + assert dict(detection.components)["RAM"] is MeasurementTier.ESTIMATED + + +def test_weakest_component_wins() -> None: + assert ( + detect_measurement_tier([_cpu("intel_rapl"), _gpu("NvidiaGPUDevice")]).tier + is MeasurementTier.AGGREGATE_ONLY + ) + assert ( + detect_measurement_tier([_cpu("constant"), _ram()]).tier + is MeasurementTier.ESTIMATED + ) + + +def test_no_or_unknown_hardware_is_aggregate_only() -> None: + assert detect_measurement_tier([]).tier is MeasurementTier.AGGREGATE_ONLY + assert detect_measurement_tier(None).tier is MeasurementTier.AGGREGATE_ONLY + assert detect_measurement_tier(MagicMock()).tier is MeasurementTier.AGGREGATE_ONLY + assert detect_measurement_tier([object()]).tier is MeasurementTier.AGGREGATE_ONLY + + +def _build(middleware: CodeCarbonMiddleware, hardware, data, tracker=None): + detection = detect_measurement_tier(hardware) + return middleware._build_measurement( + detection, "GET /predict", "GET /predict", data, tracker + ) + + +def test_measured_tier_reports_energy() -> None: + middleware = CodeCarbonMiddleware(MagicMock()) + data = _emissions_data(duration=0.03, energy_consumed=1e-8, emissions=4e-9) + measurement = _build(middleware, [_cpu("intel_rapl")], data) + assert measurement.tier is MeasurementTier.MEASURED + assert measurement.available + assert measurement.energy_consumed == 1e-8 + middleware.shutdown_tracker_executor() + + +def test_zero_delta_is_unavailable_not_zero() -> None: + """Zero is a factual claim that the request was free. Never report it.""" + middleware = CodeCarbonMiddleware(MagicMock()) + data = _emissions_data(duration=0.03, energy_consumed=0.0, emissions=0.0) + measurement = _build(middleware, [_cpu("intel_rapl")], data) + assert measurement.tier is MeasurementTier.MEASURED + assert not measurement.available + assert measurement.emissions is None + assert measurement.energy_consumed is None + assert "no completed sampling window" in measurement.unavailable_reason + middleware.shutdown_tracker_executor() + + +def test_aggregate_only_has_no_per_request_energy_field() -> None: + middleware = CodeCarbonMiddleware(MagicMock()) + data = _emissions_data(duration=0.03, energy_consumed=2.5e-8, emissions=1e-8) + measurement = _build(middleware, [_cpu("cpu_load")], data) + assert measurement.tier is MeasurementTier.AGGREGATE_ONLY + assert measurement.emissions_data is None + assert measurement.emissions is None + assert "endpoint_totals()" in measurement.unavailable_reason + middleware.shutdown_tracker_executor() + + +def test_estimated_tier_derives_energy_from_duration() -> None: + middleware = CodeCarbonMiddleware(MagicMock()) + tracker = SimpleNamespace( + _total_energy=SimpleNamespace(kWh=2.0), _total_emissions=1.0 + ) # intensity = 0.5 kg/kWh + data = _emissions_data(duration=3600.0, cpu_power=100.0, ram_power=20.0) + measurement = _build(middleware, [_cpu("constant"), _ram()], data, tracker) + assert measurement.tier is MeasurementTier.ESTIMATED + assert measurement.energy_consumed == pytest.approx(0.12) # 120 W for 1 h + assert measurement.emissions == pytest.approx(0.06) + middleware.shutdown_tracker_executor() + + +def test_estimated_tier_without_known_intensity_is_unavailable() -> None: + middleware = CodeCarbonMiddleware(MagicMock()) + tracker = SimpleNamespace( + _total_energy=SimpleNamespace(kWh=0.0), _total_emissions=0.0 + ) + data = _emissions_data(duration=3600.0, cpu_power=100.0) + measurement = _build(middleware, [_cpu("constant")], data, tracker) + assert not measurement.available + middleware.shutdown_tracker_executor() + + +def _app_with_tier(hardware, emissions_data, **middleware_kwargs): + application = FastAPI() + + @application.get("/predict") + def predict() -> dict[str, bool]: + return {"ok": True} + + tracker = MagicMock() + tracker._start_time = 1.0 + tracker._hardware = hardware + tracker._total_energy = SimpleNamespace(kWh=1.0) + tracker._total_emissions = 0.5 + tracker.mark_http_request_start.return_value = MagicMock(task_name="GET /predict") + tracker.finish_http_request.return_value = emissions_data + application.state.codecarbon_tracker = tracker + add_codecarbon_middleware(application, **middleware_kwargs) + return application, tracker + + +def test_endpoint_totals_accumulate_in_aggregate_only_tier( + finalize_deferred_immediately, +) -> None: + data = _emissions_data(duration=0.03, energy_consumed=1e-6, emissions=5e-7) + application, _ = _app_with_tier([_cpu("cpu_load")], data, on_request_complete=None) + seen = [] + client = TestClient(application) + for _ in range(3): + assert client.get("/predict").status_code == 200 + middleware = application.state.codecarbon_middleware + assert middleware.measurement_tier is MeasurementTier.AGGREGATE_ONLY + totals = middleware.endpoint_totals() + assert list(totals) == ["GET /predict"] + entry = totals["GET /predict"] + assert entry.count == 3 + assert entry.energy_consumed == pytest.approx(3e-6) + assert entry.emissions == pytest.approx(1.5e-6) + assert entry.tier is MeasurementTier.AGGREGATE_ONLY + assert seen == [] + middleware.shutdown_tracker_executor() + + +def test_callback_and_request_state_carry_the_tier( + finalize_deferred_immediately, +) -> None: + seen = [] + data = _emissions_data(duration=0.03, energy_consumed=1e-6, emissions=5e-7) + application, _ = _app_with_tier( + [_cpu("cpu_load")], + data, + on_request_complete=lambda request, response, emissions_data, task: seen.append( + (emissions_data, request.state.codecarbon) + ), + ) + assert TestClient(application).get("/predict").status_code == 200 + emissions_data, measurement = seen[0] + assert emissions_data is None # aggregate-only: no per-request energy + assert measurement.tier is MeasurementTier.AGGREGATE_ONLY + assert measurement.endpoint == "GET /predict" + application.state.codecarbon_middleware.shutdown_tracker_executor() + + +def test_headers_are_off_by_default_and_carry_tier_when_enabled( + finalize_deferred_immediately, +) -> None: + data = _emissions_data(duration=0.03, energy_consumed=1e-6, emissions=5e-7) + application, _ = _app_with_tier( + [_cpu("intel_rapl")], data, on_request_complete=None + ) + response = TestClient(application).get("/predict") + assert "X-CodeCarbon-Emissions-kg" not in response.headers + assert "X-CodeCarbon-Tier" not in response.headers + application.state.codecarbon_middleware.shutdown_tracker_executor() + + application, _ = _app_with_tier( + [_cpu("intel_rapl")], + data, + on_request_complete=None, + response_headers=True, + ) + response = TestClient(application).get("/predict") + assert response.headers["X-CodeCarbon-Emissions-kg"] == "5e-07" + assert response.headers["X-CodeCarbon-Tier"] == "measured" + application.state.codecarbon_middleware.shutdown_tracker_executor() + + +def test_headers_say_unavailable_instead_of_zero( + finalize_deferred_immediately, +) -> None: + data = _emissions_data(duration=0.03, energy_consumed=1e-6, emissions=5e-7) + application, _ = _app_with_tier( + [_cpu("cpu_load")], + data, + on_request_complete=None, + response_headers=True, + ) + response = TestClient(application).get("/predict") + assert response.headers["X-CodeCarbon-Emissions-kg"] == "unavailable" + assert response.headers["X-CodeCarbon-Tier"] == "aggregate_only" + application.state.codecarbon_middleware.shutdown_tracker_executor() + + +def test_tracker_tasks_map_stays_flat_over_10000_requests() -> None: + """Real tracker methods, no scheduler: _tasks must not grow per request.""" + from codecarbon.emissions_tracker import EmissionsTracker + + tracker = object.__new__(EmissionsTracker) + tracker._tasks = {} + tracker._http_task_lock = threading.RLock() + tracker._save_to_api = False + tracker._output_handlers = [] + + sizes = [] + for index in range(10_000): + name = tracker._resolve_http_task_name("GET /predict") + from codecarbon.external.task import Task + + tracker._tasks[name] = Task(task_name=name) + tracker._tasks[name].is_active = False + tracker.persist_completed_task(name) + tracker.discard_task(name) + if index % 1000 == 0: + sizes.append(len(tracker._tasks)) + + assert len(tracker._tasks) == 0 + assert sizes == [0] * len(sizes) + + +def test_middleware_discards_task_after_persisting( + finalize_deferred_immediately, +) -> None: + data = _emissions_data(duration=0.03, energy_consumed=1e-6, emissions=5e-7) + application, tracker = _app_with_tier( + [_cpu("intel_rapl")], data, on_request_complete=None + ) + client = TestClient(application) + for _ in range(5): + client.get("/predict") + assert tracker.persist_completed_task.call_count == 5 + assert tracker.discard_task.call_count == 5 + application.state.codecarbon_middleware.shutdown_tracker_executor() + + +@patch.object(cc_middleware, "EmissionsTracker") +def test_tier_is_resolved_once(MockTracker) -> None: + middleware = CodeCarbonMiddleware(MagicMock()) + tracker = SimpleNamespace(_hardware=[_cpu("intel_rapl")]) + assert middleware.measurement_tier is None + assert middleware._resolve_tier(tracker).tier is MeasurementTier.MEASURED + tracker._hardware = [_cpu("cpu_load")] + assert middleware._resolve_tier(tracker).tier is MeasurementTier.MEASURED + assert middleware.measurement_tier is MeasurementTier.MEASURED + middleware.shutdown_tracker_executor()