From 70c69bf1bbd02ea5971fea1f6aa4d0390ded6f2d Mon Sep 17 00:00:00 2001 From: David Berenstein Date: Wed, 12 Aug 2026 17:12:53 +0200 Subject: [PATCH 1/2] feat(api): reuse connections and retry transient API failures ApiClient called module-level requests functions with a hardcoded timeout=2 and no retry policy, so every emission POST opened a fresh connection and any transient failure discarded the measurement. - Add a Session with an HTTPAdapter/urllib3 Retry (backoff + jitter, falling back gracefully on urllib3 < 2.0). POST is deliberately retried: a duplicate telemetry row beats a lost measurement. - Make timeout, retries and backoff constructor arguments, exposed to users as the api_timeout / api_retries config knobs. - Back off run creation after a failure instead of re-attempting it on every measurement tick, which otherwise stampedes the API after an outage. - Close the Session in CodeCarbonAPIOutput.exit() so long-lived processes do not leak sockets. Measured against a loopback stub server: 50 sequential emission POSTs open 1 TCP connection instead of 50, and a row that hit two transient 503s is now delivered instead of dropped. Worst-case time an unreachable API can block the measurement loop is ~16s, below the tracker's own 45s stale-measurement warning. The disk spill buffer from the plan is deliberately not implemented. Co-Authored-By: Claude Opus 5 (1M context) --- codecarbon/core/api_client.py | 123 ++++++++++++--- codecarbon/emissions_tracker.py | 11 ++ codecarbon/output_methods/http.py | 17 +++ tests/test_api_client_retry.py | 241 ++++++++++++++++++++++++++++++ 4 files changed, 374 insertions(+), 18 deletions(-) create mode 100644 tests/test_api_client_retry.py diff --git a/codecarbon/core/api_client.py b/codecarbon/core/api_client.py index 94bb85f01..2dbcf1fdf 100644 --- a/codecarbon/core/api_client.py +++ b/codecarbon/core/api_client.py @@ -1,16 +1,16 @@ """ Based on https://kernelpanic.io/the-modern-way-to-call-apis-in-python - -TODO : use async call to API """ -# from httpx import AsyncClient import dataclasses import json +import time from datetime import datetime, timedelta, tzinfo import requests +from requests.adapters import HTTPAdapter +from urllib3.util.retry import Retry from codecarbon.core.schemas import ( EmissionCreate, @@ -44,6 +44,40 @@ def _measurement_timestamp(carbon_emission: dict) -> str: return get_datetime_with_timezone() +def _build_session(retries: int, backoff: float) -> requests.Session: + """ + A Session so sockets (and the TLS handshake) are reused across calls, with + retry and exponential backoff on the failures that are worth retrying. + + POST is deliberately in `allowed_methods`, unlike urllib3's default. A + duplicate emission row is far less harmful than a lost measurement: this is + telemetry, not billing. Do not "fix" this without an idempotency key on the + row, server side. + """ + retry_kwargs = { + "total": retries, + "connect": retries, + "read": retries, + "status": retries, + "backoff_factor": backoff, + "status_forcelist": (429, 500, 502, 503, 504), + "allowed_methods": frozenset({"GET", "POST", "PATCH", "PUT", "DELETE"}), + "raise_on_status": False, + } + try: + # Spreads the retry storm when a whole fleet reconnects after an outage. + retry = Retry(backoff_jitter=1.0, **retry_kwargs) + except TypeError: + # urllib3 < 2.0 has no backoff_jitter. We are a library in other + # people's environments, so degrade instead of pinning urllib3. + retry = Retry(**retry_kwargs) + adapter = HTTPAdapter(max_retries=retry, pool_maxsize=4) + session = requests.Session() + session.mount("http://", adapter) + session.mount("https://", adapter) + return session + + class ApiClient: # (AsyncClient) """ This class call the Code Carbon API @@ -59,6 +93,9 @@ def __init__( access_token=None, conf=None, create_run_automatically=True, + timeout=(3.05, 10), + retries=2, + backoff=0.5, ): """ :endpoint_url: URL of the API endpoint @@ -67,13 +104,25 @@ def __init__( :access_token: Code Carbon API access token :conf: Metadata of the experiment :create_run_automatically: If False, do not create a run. To use API in read only mode. + :timeout: requests timeout, either seconds or a (connect, read) tuple. + :retries: number of retries after the first attempt. + :backoff: backoff factor between retries, in seconds (0.5 -> 0.5s, 1s, 2s...). + + Note that `timeout` and `retries` multiply: the worst case wall time of a + call is roughly `(connect + read) * (retries + 1)` plus backoff, so raise + one only while looking at the other. """ - # super().__init__(base_url=endpoint_url) # (AsyncClient) self.url = endpoint_url self.experiment_id = experiment_id self.api_key = api_key self.conf = conf self.access_token = access_token + self._timeout = timeout + self._session = _build_session(retries, backoff) + # Run creation is the most expensive write path. When it fails, back off + # instead of re-attempting on every measurement tick. + self._create_run_not_before = 0.0 + self._create_run_backoff = 0.0 if self.experiment_id is not None and create_run_automatically: self._create_run(self.experiment_id) @@ -96,11 +145,15 @@ def _request(self, method, url, payload=None, expected_status=200): :expected_status: the http code the API returns when the call succeeds """ headers = self._get_headers() - response = method(url=url, json=payload, timeout=2, headers=headers) + response = method(url=url, json=payload, timeout=self._timeout, headers=headers) if response.status_code != expected_status: self._raise_api_error(url, payload or {}, response) return response + def close(self): + """Release the pooled sockets. Safe to call more than once.""" + self._session.close() + def set_access_token(self, token: str): """This method sets the access token to be used for the API. Args: @@ -113,14 +166,14 @@ def check_auth(self): Check API access to user account """ url = self.url + "/auth/check" - return self._request(requests.get, url).json() + return self._request(self._session.get, url).json() def get_list_organizations(self): """ List all organizations """ url = self.url + "/organizations" - return self._request(requests.get, url).json() + return self._request(self._session.get, url).json() def check_organization_exists(self, organization_name: str): """ @@ -145,7 +198,7 @@ def create_organization(self, organization: OrganizationCreate): return organization else: return self._request( - requests.post, url, payload=payload, expected_status=201 + self._session.post, url, payload=payload, expected_status=201 ).json() def get_organization(self, organization_id): @@ -153,7 +206,7 @@ def get_organization(self, organization_id): Get an organization """ url = self.url + "/organizations/" + organization_id - return self._request(requests.get, url).json() + return self._request(self._session.get, url).json() def update_organization(self, organization: OrganizationCreate): """ @@ -161,14 +214,14 @@ def update_organization(self, organization: OrganizationCreate): """ payload = dataclasses.asdict(organization) url = self.url + "/organizations/" + organization.id - return self._request(requests.patch, url, payload=payload).json() + return self._request(self._session.patch, url, payload=payload).json() def list_projects_from_organization(self, organization_id): """ List all projects """ url = self.url + "/organizations/" + organization_id + "/projects" - return self._request(requests.get, url).json() + return self._request(self._session.get, url).json() def create_project(self, project: ProjectCreate): """ @@ -177,7 +230,7 @@ def create_project(self, project: ProjectCreate): payload = dataclasses.asdict(project) url = self.url + "/projects" return self._request( - requests.post, url, payload=payload, expected_status=201 + self._session.post, url, payload=payload, expected_status=201 ).json() def get_project(self, project_id): @@ -185,7 +238,7 @@ def get_project(self, project_id): Get a project """ url = self.url + "/projects/" + project_id - return self._request(requests.get, url).json() + return self._request(self._session.get, url).json() def add_emission(self, carbon_emission: dict): assert self.experiment_id is not None @@ -226,7 +279,7 @@ def add_emission(self, carbon_emission: dict): try: payload = dataclasses.asdict(emission) url = self.url + "/emissions" - self._request(requests.post, url, payload=payload, expected_status=201) + self._request(self._session.post, url, payload=payload, expected_status=201) logger.debug(f"ApiClient - Successful upload emission {payload} to {url}") except requests.exceptions.HTTPError: # Already logged by _raise_api_error, do not log it twice. @@ -236,7 +289,39 @@ def add_emission(self, carbon_emission: dict): raise return True + # Bounds on the run-creation retry delay, in seconds. + _CREATE_RUN_BACKOFF_MIN = 30.0 + _CREATE_RUN_BACKOFF_MAX = 900.0 + def _create_run(self, experiment_id: str): + """ + Create a run, backing off after a failure. + + Without the backoff every client in a fleet re-attempts run creation on + every measurement tick for as long as the API is down, which hammers the + most expensive write path exactly when it is least able to take it. + Returns None without calling the API while the backoff is in effect. + """ + if time.monotonic() < self._create_run_not_before: + logger.debug( + "ApiClient run creation is backing off after a previous failure, " + "skipping this attempt." + ) + return None + try: + run_id = self._create_run_once(experiment_id) + except Exception: + self._create_run_backoff = min( + max(self._create_run_backoff * 2, self._CREATE_RUN_BACKOFF_MIN), + self._CREATE_RUN_BACKOFF_MAX, + ) + self._create_run_not_before = time.monotonic() + self._create_run_backoff + raise + self._create_run_backoff = 0.0 + self._create_run_not_before = 0.0 + return run_id + + def _create_run_once(self, experiment_id: str): """ Create the experiment for project_id """ @@ -269,7 +354,9 @@ def _create_run(self, experiment_id: str): ) payload = dataclasses.asdict(run) url = self.url + "/runs" - r = self._request(requests.post, url, payload=payload, expected_status=201) + r = self._request( + self._session.post, url, payload=payload, expected_status=201 + ) self.run_id = r.json()["id"] logger.info( "ApiClient Successfully registered your run on the API.\n\n" @@ -295,7 +382,7 @@ def list_experiments_from_project(self, project_id: str): List all experiments for a project """ url = self.url + "/projects/" + project_id + "/experiments" - return self._request(requests.get, url).json() + return self._request(self._session.get, url).json() def set_experiment(self, experiment_id: str): """ @@ -311,7 +398,7 @@ def add_experiment(self, experiment: ExperimentCreate): payload = dataclasses.asdict(experiment) url = self.url + "/experiments" return self._request( - requests.post, url, payload=payload, expected_status=201 + self._session.post, url, payload=payload, expected_status=201 ).json() def get_experiment(self, experiment_id): @@ -319,7 +406,7 @@ def get_experiment(self, experiment_id): Get an experiment by id """ url = self.url + "/experiments/" + experiment_id - return self._request(requests.get, url).json() + return self._request(self._session.get, url).json() def _raise_api_error(self, url, payload, response): """ diff --git a/codecarbon/emissions_tracker.py b/codecarbon/emissions_tracker.py index 96ed00c91..ce545a672 100644 --- a/codecarbon/emissions_tracker.py +++ b/codecarbon/emissions_tracker.py @@ -392,6 +392,8 @@ def __init__( api_call_interval: Optional[int] = _sentinel, api_endpoint: Optional[str] = _sentinel, api_key: Optional[str] = _sentinel, + api_timeout: Optional[float] = _sentinel, + api_retries: Optional[int] = _sentinel, output_dir: Optional[str] = _sentinel, output_file: Optional[str] = _sentinel, output_methods: Optional[List[OutputMethod]] = _sentinel, @@ -437,6 +439,11 @@ def __init__( :param api_endpoint: Optional URL of Code Carbon API endpoint for sending emissions data. :param api_key: API key for Code Carbon API (mandatory!). + :param api_timeout: Read timeout in seconds for API calls (default: 5). + Connect timeout is fixed at 3.05s. + :param api_retries: Retries after the first failed API attempt (default: 2). + Worst-case time an API call can block a measurement + is about (3.05 + api_timeout) * (api_retries + 1). :param output_dir: Directory path to which the experiment details are logged, defaults to current directory. :param output_file: Name of the output CSV file, defaults to `emissions.csv`. @@ -554,6 +561,8 @@ def __init__( self._set_from_conf(api_call_interval, "api_call_interval", 8, int) self._set_from_conf(api_endpoint, "api_endpoint", "https://api.codecarbon.io") self._set_from_conf(api_key, "api_key", "api_key") + self._set_from_conf(api_timeout, "api_timeout", 5, float) + self._set_from_conf(api_retries, "api_retries", 2, int) self._configure_electricitymaps_token( electricitymaps_api_token, co2_signal_api_token ) @@ -644,6 +653,8 @@ def _init_output_methods(self, *, api_key: str = None): experiment_id=self._experiment_id, api_key=api_key, conf=self._conf, + timeout=(3.05, self._api_timeout), + retries=self._api_retries, ) self.run_id = cc_api__out.run_id self._output_handlers.append(cc_api__out) diff --git a/codecarbon/output_methods/http.py b/codecarbon/output_methods/http.py index e0ff710b1..b560490d8 100644 --- a/codecarbon/output_methods/http.py +++ b/codecarbon/output_methods/http.py @@ -46,7 +46,16 @@ def __init__( experiment_id: str, api_key: str, conf, + timeout=(3.05, 5), + retries: int = 2, + backoff: float = 0.5, ): + """ + The timeout and retry budget here is deliberately tighter than + ApiClient's own default: this send runs inline on the scheduler thread, + so a slow failure delays the next measurement. Worst case is roughly + (3.05 + 5) * 3 plus backoff. + """ self.endpoint_url: str = endpoint_url self.api = ApiClient( endpoint_url=endpoint_url, @@ -54,9 +63,17 @@ def __init__( api_key=api_key, conf=conf, create_run_automatically=False, + timeout=timeout, + retries=retries, + backoff=backoff, ) self.run_id = self.api.run_id + def exit(self) -> None: + # A Session holds sockets; long-lived processes creating many trackers + # would otherwise leak file descriptors. + self.api.close() + def _ensure_api_run(self) -> None: if self.api.run_id is None and self.api.experiment_id is not None: self.api._create_run(self.api.experiment_id) diff --git a/tests/test_api_client_retry.py b/tests/test_api_client_retry.py new file mode 100644 index 000000000..7cc8d4eab --- /dev/null +++ b/tests/test_api_client_retry.py @@ -0,0 +1,241 @@ +""" +Retry, timeout and connection-reuse tests for ApiClient. + +These run against a stdlib HTTP server on loopback rather than requests_mock, +because requests_mock replaces the transport adapter and therefore never +exercises the HTTPAdapter/urllib3 Retry layer that is under test here. +No traffic leaves the machine. +""" + +import socket +import threading +import unittest +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +import requests + +from codecarbon.core.api_client import ApiClient + +CONF = { + "os": "linux", + "python_version": "3.12", + "codecarbon_version": "3.0", + "cpu_count": 8, + "cpu_model": "CPU", + "gpu_count": 0, + "gpu_model": "", + "longitude": 0.0, + "latitude": 0.0, + "region": "EU", + "provider": "none", + "ram_total_size": 16.0, + "tracking_mode": "machine", +} + +EMISSION = { + "duration": 5, + "emissions": 1.0, + "emissions_rate": 1.0, + "cpu_power": 1.0, + "gpu_power": 0.0, + "ram_power": 0.5, + "cpu_energy": 0.1, + "gpu_energy": 0.0, + "ram_energy": 0.1, + "energy_consumed": 0.2, +} + + +class _Handler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" # keep-alive, so pooling is observable + + def log_message(self, *args): + pass + + def _serve(self): + state = self.server.state + state["requests"] += 1 + length = int(self.headers.get("Content-Length", 0) or 0) + if length: + self.rfile.read(length) + plan = state["status_plan"] + code = plan.pop(0) if plan else state["success_status"] + body = b'{"id": "run-1"}' + self.send_response(code) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + do_GET = _serve + do_POST = _serve + + +class _Server(ThreadingHTTPServer): + daemon_threads = True + allow_reuse_address = True + + def __init__(self, state): + self.state = state + super().__init__(("127.0.0.1", 0), _Handler) + + def process_request(self, request, client_address): + self.state["connections"] += 1 + super().process_request(request, client_address) + + +class StubServerTestCase(unittest.TestCase): + """A loopback HTTP server whose responses can be scripted per test.""" + + def setUp(self): + self.state = { + "requests": 0, + "connections": 0, + "status_plan": [], + "success_status": 201, + } + self.server = _Server(self.state) + threading.Thread(target=self.server.serve_forever, daemon=True).start() + self.url = f"http://127.0.0.1:{self.server.server_address[1]}" + self.addCleanup(self.server.server_close) + self.addCleanup(self.server.shutdown) + + def client(self, **kwargs): + kwargs.setdefault("retries", 2) + kwargs.setdefault("backoff", 0) # keep the suite fast + api = ApiClient( + endpoint_url=self.url, + experiment_id="exp-1", + conf=CONF, + create_run_automatically=False, + **kwargs, + ) + self.addCleanup(api.close) + api.run_id = "run-1" + return api + + +class TestRetry(StubServerTestCase): + def test_retries_then_succeeds_on_transient_503(self): + self.state["status_plan"] = [503, 503] + api = self.client() + + self.assertTrue(api.add_emission(dict(EMISSION))) + self.assertEqual(self.state["requests"], 3) + + def test_gives_up_after_configured_retries(self): + self.state["status_plan"] = [503] * 10 + api = self.client(retries=2) + + with self.assertRaises(requests.exceptions.HTTPError): + api.add_emission(dict(EMISSION)) + self.assertEqual(self.state["requests"], 3) # 1 attempt + 2 retries + + def test_no_retry_on_client_error(self): + """Retrying a validation error is pure waste.""" + self.state["status_plan"] = [400] * 10 + api = self.client() + + with self.assertRaises(requests.exceptions.HTTPError): + api.add_emission(dict(EMISSION)) + self.assertEqual(self.state["requests"], 1) + + def test_retries_on_connection_error(self): + """Nothing is listening, so every attempt fails to connect.""" + sock = socket.socket() + sock.bind(("127.0.0.1", 0)) + port = sock.getsockname()[1] + sock.close() + + api = ApiClient( + endpoint_url=f"http://127.0.0.1:{port}", + experiment_id="exp-1", + conf=CONF, + create_run_automatically=False, + retries=2, + backoff=0, + ) + self.addCleanup(api.close) + api.run_id = "run-1" + + with self.assertRaises(requests.exceptions.ConnectionError) as ctx: + api.add_emission(dict(EMISSION)) + # urllib3 reports the exhausted budget in the message. + self.assertIn("max retries exceeded", str(ctx.exception).lower()) + + +class TestSessionReuse(StubServerTestCase): + def test_sequential_calls_reuse_one_connection(self): + api = self.client() + + for _ in range(5): + self.assertTrue(api.add_emission(dict(EMISSION))) + + self.assertEqual(self.state["requests"], 5) + self.assertEqual(self.state["connections"], 1) + + def test_close_releases_the_session(self): + api = self.client() + api.add_emission(dict(EMISSION)) + api.close() + api.close() # idempotent + + +class TestTimeoutConfiguration(unittest.TestCase): + def test_timeout_is_passed_through_to_requests(self): + api = ApiClient( + endpoint_url="http://test.com", + create_run_automatically=False, + timeout=(1.5, 7), + ) + seen = {} + + def fake_get(url, json, timeout, headers): + seen["timeout"] = timeout + return type("R", (), {"status_code": 200, "json": lambda self: {}})() + + api._request(fake_get, "http://test.com/x") + self.assertEqual(seen["timeout"], (1.5, 7)) + + def test_default_timeout_is_not_the_old_hardcoded_two_seconds(self): + api = ApiClient(endpoint_url="http://test.com", create_run_automatically=False) + self.assertEqual(api._timeout, (3.05, 10)) + + +class TestCreateRunBackoff(StubServerTestCase): + def test_failed_run_creation_is_not_retried_on_every_tick(self): + self.state["status_plan"] = [400] * 10 # not retryable, fails fast + api = ApiClient( + endpoint_url=self.url, + experiment_id="exp-1", + conf=CONF, + create_run_automatically=False, + retries=0, + ) + self.addCleanup(api.close) + + with self.assertRaises(requests.exceptions.HTTPError): + api._create_run("exp-1") + self.assertEqual(self.state["requests"], 1) + + # Subsequent attempts are suppressed while the backoff is in effect. + for _ in range(5): + self.assertIsNone(api._create_run("exp-1")) + self.assertEqual(self.state["requests"], 1) + + def test_backoff_clears_after_a_success(self): + api = ApiClient( + endpoint_url=self.url, + experiment_id="exp-1", + conf=CONF, + create_run_automatically=False, + retries=0, + ) + self.addCleanup(api.close) + + self.assertEqual(api._create_run("exp-1"), "run-1") + self.assertEqual(api._create_run_not_before, 0.0) + + +if __name__ == "__main__": + unittest.main() From 863b9c3db7b08ed2d05289b510cad1829f7d65f1 Mon Sep 17 00:00:00 2001 From: David Berenstein Date: Wed, 12 Aug 2026 19:46:43 +0200 Subject: [PATCH 2/2] fix(api): do not replay emission POSTs that may already have committed status_forcelist included 502/503/504 and read=retries applied to POST, so a 504 or a read timeout arriving after carbonserver had already committed the insert made the client send the row again. There is no idempotency key and the dashboard sums emission rows, so that silently inflates a user's reported emissions -- and unlike a dropped row, nothing shows it happened. - Split the retry policy in two. GETs and PATCHes keep the broad policy. POSTs only retry failures that plausibly never reached the application: connection errors, connect timeouts, 429/502/503. Read timeouts, truncated responses, 500 and 504 are no longer retried. Widen only behind a server-side idempotency key. - Replace the 'telemetry, not billing' comment: wrong call for a carbon-accounting product. - Make the timing numbers agree and say which worst case is which. Measured: hung endpoint 5s (was 16.5s), unreachable endpoint 11s, worst case 16s, all under the tracker's 45s stale-measurement warning. - Document api_timeout / api_retries in docs/how-to/configuration.md. Tests: 4 new cases pinning POST no-retry on read timeout and on 504 against POST-still-retries on 503 and GET-still-retries on read timeout, plus one asserting api_timeout/api_retries reach ApiClient from the tracker constructor. All fail if the change is reverted. Co-Authored-By: Claude Opus 5 (1M context) --- codecarbon/core/api_client.py | 68 +++++++++++++++++++++++-------- codecarbon/emissions_tracker.py | 6 ++- codecarbon/output_methods/http.py | 17 +++++++- docs/how-to/configuration.md | 35 ++++++++++++++++ tests/test_api_client_retry.py | 49 ++++++++++++++++++++++ tests/test_emissions_tracker.py | 31 ++++++++++++++ 6 files changed, 186 insertions(+), 20 deletions(-) diff --git a/codecarbon/core/api_client.py b/codecarbon/core/api_client.py index 2dbcf1fdf..7702a1175 100644 --- a/codecarbon/core/api_client.py +++ b/codecarbon/core/api_client.py @@ -44,23 +44,34 @@ def _measurement_timestamp(carbon_emission: dict) -> str: return get_datetime_with_timezone() -def _build_session(retries: int, backoff: float) -> requests.Session: +# Failures where the request plausibly never reached the application, so a +# retry cannot duplicate work: no upstream was reachable (502/503) or we were +# told to slow down before being served (429). +_POST_SAFE_STATUSES = (429, 502, 503) +# Reads are idempotent, so a retry costs at most a wasted round trip. +_READ_SAFE_STATUSES = (429, 500, 502, 503, 504) + + +def _build_session( + retries: int, backoff: float, statuses, retry_read: bool +) -> requests.Session: """ A Session so sockets (and the TLS handshake) are reused across calls, with retry and exponential backoff on the failures that are worth retrying. - POST is deliberately in `allowed_methods`, unlike urllib3's default. A - duplicate emission row is far less harmful than a lost measurement: this is - telemetry, not billing. Do not "fix" this without an idempotency key on the - row, server side. + :statuses: response codes to retry. + :retry_read: whether to retry a read timeout or a truncated response, i.e. + a failure that happened *after* the request reached the server. """ retry_kwargs = { "total": retries, "connect": retries, - "read": retries, + # False, not 0: urllib3 then re-raises the original ReadTimeout instead + # of burning the budget and reporting an exhausted-retries error. + "read": retries if retry_read else False, "status": retries, "backoff_factor": backoff, - "status_forcelist": (429, 500, 502, 503, 504), + "status_forcelist": statuses, "allowed_methods": frozenset({"GET", "POST", "PATCH", "PUT", "DELETE"}), "raise_on_status": False, } @@ -108,9 +119,16 @@ def __init__( :retries: number of retries after the first attempt. :backoff: backoff factor between retries, in seconds (0.5 -> 0.5s, 1s, 2s...). - Note that `timeout` and `retries` multiply: the worst case wall time of a - call is roughly `(connect + read) * (retries + 1)` plus backoff, so raise - one only while looking at the other. + Note that `timeout` and `retries` multiply, so raise one only while + looking at the other. Two different worst cases are worth keeping apart: + + - a *hung* endpoint, which accepts the connection and never answers: + `read * (retries + 1)` plus backoff, since only the read times out. + - the *full* retry chain, where the connection also has to time out: + `(connect + read) * (retries + 1)` plus backoff. + + POSTs do not retry read timeouts (see `_post_session`), so their hung + case is a single `connect + read` and the chain above is a GET bound. """ self.url = endpoint_url self.experiment_id = experiment_id @@ -118,7 +136,22 @@ def __init__( self.conf = conf self.access_token = access_token self._timeout = timeout - self._session = _build_session(retries, backoff) + self._session = _build_session( + retries, backoff, _READ_SAFE_STATUSES, retry_read=True + ) + # POSTs create rows. carbonserver has no idempotency key and the + # dashboard sums emission rows, so a POST replayed after the server + # already committed the insert inflates a user's reported emissions + # with nothing to show for it: a dropped row is visible, a duplicated + # one is not. This session therefore only retries POSTs that + # plausibly never reached the application -- connection errors, + # connect timeouts, 429/502/503. Read timeouts, truncated responses, + # 500 and 504 are *not* retried: the request landed, and the insert + # may well have gone through. Widen this only once the API accepts an + # idempotency key. + self._post_session = _build_session( + retries, backoff, _POST_SAFE_STATUSES, retry_read=False + ) # Run creation is the most expensive write path. When it fails, back off # instead of re-attempting on every measurement tick. self._create_run_not_before = 0.0 @@ -153,6 +186,7 @@ def _request(self, method, url, payload=None, expected_status=200): def close(self): """Release the pooled sockets. Safe to call more than once.""" self._session.close() + self._post_session.close() def set_access_token(self, token: str): """This method sets the access token to be used for the API. @@ -198,7 +232,7 @@ def create_organization(self, organization: OrganizationCreate): return organization else: return self._request( - self._session.post, url, payload=payload, expected_status=201 + self._post_session.post, url, payload=payload, expected_status=201 ).json() def get_organization(self, organization_id): @@ -230,7 +264,7 @@ def create_project(self, project: ProjectCreate): payload = dataclasses.asdict(project) url = self.url + "/projects" return self._request( - self._session.post, url, payload=payload, expected_status=201 + self._post_session.post, url, payload=payload, expected_status=201 ).json() def get_project(self, project_id): @@ -279,7 +313,9 @@ def add_emission(self, carbon_emission: dict): try: payload = dataclasses.asdict(emission) url = self.url + "/emissions" - self._request(self._session.post, url, payload=payload, expected_status=201) + self._request( + self._post_session.post, url, payload=payload, expected_status=201 + ) logger.debug(f"ApiClient - Successful upload emission {payload} to {url}") except requests.exceptions.HTTPError: # Already logged by _raise_api_error, do not log it twice. @@ -355,7 +391,7 @@ def _create_run_once(self, experiment_id: str): payload = dataclasses.asdict(run) url = self.url + "/runs" r = self._request( - self._session.post, url, payload=payload, expected_status=201 + self._post_session.post, url, payload=payload, expected_status=201 ) self.run_id = r.json()["id"] logger.info( @@ -398,7 +434,7 @@ def add_experiment(self, experiment: ExperimentCreate): payload = dataclasses.asdict(experiment) url = self.url + "/experiments" return self._request( - self._session.post, url, payload=payload, expected_status=201 + self._post_session.post, url, payload=payload, expected_status=201 ).json() def get_experiment(self, experiment_id): diff --git a/codecarbon/emissions_tracker.py b/codecarbon/emissions_tracker.py index ce545a672..7b7b47d59 100644 --- a/codecarbon/emissions_tracker.py +++ b/codecarbon/emissions_tracker.py @@ -442,8 +442,10 @@ def __init__( :param api_timeout: Read timeout in seconds for API calls (default: 5). Connect timeout is fixed at 3.05s. :param api_retries: Retries after the first failed API attempt (default: 2). - Worst-case time an API call can block a measurement - is about (3.05 + api_timeout) * (api_retries + 1). + An emission POST blocks a measurement for at most + about 3.05 * api_retries + (3.05 + api_timeout), + ~16s with the defaults: only connect failures are + retried on POST, so read timeouts happen once. :param output_dir: Directory path to which the experiment details are logged, defaults to current directory. :param output_file: Name of the output CSV file, defaults to `emissions.csv`. diff --git a/codecarbon/output_methods/http.py b/codecarbon/output_methods/http.py index b560490d8..c4876e2a4 100644 --- a/codecarbon/output_methods/http.py +++ b/codecarbon/output_methods/http.py @@ -53,8 +53,21 @@ def __init__( """ The timeout and retry budget here is deliberately tighter than ApiClient's own default: this send runs inline on the scheduler thread, - so a slow failure delays the next measurement. Worst case is roughly - (3.05 + 5) * 3 plus backoff. + so a slow failure delays the next measurement. + + With the defaults below, measured against a loopback stub: + + - hung endpoint (connects, never answers): ~5s. POSTs do not retry + read timeouts, see ApiClient._post_session, so this is one read + timeout and not three. + - unreachable endpoint (every connect times out): ~11s, i.e. + 3.05 * 3 plus jittered backoff. + - worst case, connect timeouts until the last attempt connects and + then hangs: ~16s. + + All under the tracker's own 45s stale-measurement warning at + emissions_tracker.py, which is the number that actually matters: this + call blocks the scheduler thread. """ self.endpoint_url: str = endpoint_url self.api = ApiClient( diff --git a/docs/how-to/configuration.md b/docs/how-to/configuration.md index 9f6766aa1..be5b3c95f 100644 --- a/docs/how-to/configuration.md +++ b/docs/how-to/configuration.md @@ -178,6 +178,41 @@ Despite its name, this option applies to every counter-based CPU interface: It has no effect when CodeCarbon falls back to TDP/CPU-load estimation, since that mode models the CPU package only. +## API timeouts and retries + +When `save_to_api` is enabled, each measurement is sent to the API from the +scheduler thread, so a slow or unreachable API delays the next measurement. Two +parameters bound how long that can take: + +- **`api_timeout`** (default: `5`): read timeout in seconds. The connect + timeout is fixed at 3.05s. +- **`api_retries`** (default: `2`): retries after the first failed attempt. + +``` ini +[codecarbon] +api_timeout = 5 +api_retries = 2 +``` + +Or in code: + +``` python +EmissionsTracker(api_timeout=5, api_retries=2) +``` + +With the defaults, an emission upload blocks for at most ~16 seconds, which +stays under CodeCarbon's own 45s stale-measurement warning. Raise one only +while looking at the other: they multiply. + +!!! note "What gets retried" + + Reads (`GET`) retry any transient failure. Emission uploads (`POST`) only + retry failures where the request plausibly never reached the API — + connection errors, connect timeouts, and `429`/`502`/`503`. A read timeout + or a `504` is **not** retried, because the API may already have stored the + row and the dashboard sums emission rows: a duplicated measurement would + silently inflate your reported emissions. + ## Access internet through proxy server If you need a proxy to access internet, which is needed to call a Web diff --git a/tests/test_api_client_retry.py b/tests/test_api_client_retry.py index 7cc8d4eab..a627a6407 100644 --- a/tests/test_api_client_retry.py +++ b/tests/test_api_client_retry.py @@ -9,6 +9,7 @@ import socket import threading +import time import unittest from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer @@ -58,6 +59,10 @@ def _serve(self): length = int(self.headers.get("Content-Length", 0) or 0) if length: self.rfile.read(length) + if state["hang"]: + # Answer far too late, i.e. the client hits its read timeout while + # the server is happily processing the request. + time.sleep(state["hang"]) plan = state["status_plan"] code = plan.pop(0) if plan else state["success_status"] body = b'{"id": "run-1"}' @@ -93,6 +98,7 @@ def setUp(self): "connections": 0, "status_plan": [], "success_status": 201, + "hang": 0, } self.server = _Server(self.state) threading.Thread(target=self.server.serve_forever, daemon=True).start() @@ -164,6 +170,49 @@ def test_retries_on_connection_error(self): self.assertIn("max retries exceeded", str(ctx.exception).lower()) +class TestPostIsNotReplayedAfterTheRequestLanded(StubServerTestCase): + """ + carbonserver has no idempotency key and the dashboard sums emission rows, + so replaying a POST whose response was lost silently inflates a user's + reported emissions. Only failures that plausibly never reached the + application may be retried on POST. + """ + + def test_read_timeout_on_post_is_not_retried(self): + self.state["hang"] = 3 # much longer than the read timeout below + api = self.client(timeout=(3.05, 0.3)) + + with self.assertRaises(requests.exceptions.ReadTimeout): + api.add_emission(dict(EMISSION)) + self.assertEqual(self.state["requests"], 1) + + def test_503_on_post_is_still_retried(self): + """The counterpart: 503 means no upstream took the request.""" + self.state["status_plan"] = [503, 503] + api = self.client(timeout=(3.05, 0.3)) + + self.assertTrue(api.add_emission(dict(EMISSION))) + self.assertEqual(self.state["requests"], 3) + + def test_504_on_post_is_not_retried(self): + """A gateway timeout means the app was reached and may have committed.""" + self.state["status_plan"] = [504] * 10 + api = self.client() + + with self.assertRaises(requests.exceptions.HTTPError): + api.add_emission(dict(EMISSION)) + self.assertEqual(self.state["requests"], 1) + + def test_read_timeout_on_get_is_still_retried(self): + """GETs are idempotent, so they keep the broader policy.""" + self.state["hang"] = 3 + api = self.client(timeout=(3.05, 0.3)) + + with self.assertRaises(requests.exceptions.RequestException): + api.get_list_organizations() + self.assertEqual(self.state["requests"], 3) + + class TestSessionReuse(StubServerTestCase): def test_sequential_calls_reuse_one_connection(self): api = self.client() diff --git a/tests/test_emissions_tracker.py b/tests/test_emissions_tracker.py index 8ab12e5d8..c551ecc30 100644 --- a/tests/test_emissions_tracker.py +++ b/tests/test_emissions_tracker.py @@ -425,6 +425,37 @@ def test_output_methods_overrides_save_to_flags( ) ) + def test_api_timeout_and_retries_reach_the_api_client( + self, + mock_cli_setup, + mock_log_values, + mocked_get_gpu_details, + mocked_env_cloud_details, + mocked_get_gpu_utilization_list, + mocked_is_gpu_details_available, + mocked_is_nvidia_system, + ): + """The knobs are public config, so prove they survive the whole path.""" + tracker = EmissionsTracker( + output_dir=self.temp_path, + output_handlers=[], + output_methods=[OutputMethod.API], + experiment_id="exp-1", + api_key="key", + api_timeout=7, + api_retries=4, + ) + + api_output = next( + handler + for handler in tracker._output_handlers + if isinstance(handler, CodeCarbonAPIOutput) + ) + self.assertEqual(api_output.api._timeout, (3.05, 7)) + for session in (api_output.api._session, api_output.api._post_session): + retry = session.get_adapter("http://x").max_retries + self.assertEqual(retry.total, 4) + def test_output_methods_parsed_from_config_string( self, mock_cli_setup,