diff --git a/codecarbon/emissions_tracker.py b/codecarbon/emissions_tracker.py index 4bd0ea99d..4116a68c6 100644 --- a/codecarbon/emissions_tracker.py +++ b/codecarbon/emissions_tracker.py @@ -267,6 +267,8 @@ def _resolve_output_methods( self._output_methods.append(OutputMethod.LOGFIRE) def _initialize_runtime_state(self) -> None: + self._api_output = None + self.run_id = uuid.uuid4() self._start_time: Optional[float] = None self._last_measured_time: float = time.perf_counter() self._total_energy: Energy = Energy.from_energy(kWh=0) @@ -615,7 +617,6 @@ def _init_output_methods(self, *, api_key: str = None): methods = set(self._output_methods) if self._output_methods else set() if not methods and not self._emissions_endpoint: - self.run_id = uuid.uuid4() return from codecarbon.output_methods.boamps import BoAmpsOutput @@ -648,10 +649,8 @@ def _init_output_methods(self, *, api_key: str = None): api_key=api_key, conf=self._conf, ) - self.run_id = cc_api__out.run_id + self._api_output = cc_api__out self._output_handlers.append(cc_api__out) - else: - self.run_id = uuid.uuid4() if OutputMethod.PROMETHEUS in methods: self._output_handlers.append( @@ -713,6 +712,18 @@ def start(self) -> None: return self._ensure_hardware_ready() + + if self._api_output is not None: + # Create the run now (blocking HTTP call), so every record carries + # the API run id. The API client already logs the failure itself. + try: + self._api_output._ensure_api_run() + self.run_id = self._api_output.run_id or self.run_id + except Exception as e: + logger.warning( + f"Could not create the API run ({e}), using local run_id {self.run_id}" + ) + self._last_measured_time = self._start_time = time.perf_counter() # Clear utilization history for fresh measurements @@ -1004,11 +1015,25 @@ def _update_emissions(self) -> None: self._total_emissions += delta_emissions self._last_energy_covered = self._total_energy + def _sync_api_run_id(self) -> None: + """ + Adopt the API run id once the run exists, e.g. when the API was down + at start() and the run got created later by a successful emission. + """ + api_run_id = self._api_output.run_id if self._api_output else None + if api_run_id and api_run_id != self.run_id: + logger.warning( + f"API run created late: switching run_id from {self.run_id} to" + f" {api_run_id}. Earlier local records used {self.run_id}." + ) + self.run_id = api_run_id + def _prepare_emissions_data(self) -> EmissionsData: """ Prepare the emissions data to be sent to the API or written to a file. :return: EmissionsData object with the total emissions data. """ + self._sync_api_run_id() self._update_emissions() self._ensure_cloud_conf() cloud = self._get_cloud_metadata() diff --git a/docs/how-to/cloud-api.md b/docs/how-to/cloud-api.md index f7bd4888d..2a831813f 100644 --- a/docs/how-to/cloud-api.md +++ b/docs/how-to/cloud-api.md @@ -34,6 +34,8 @@ if __name__ =="__main__": The decorator will automatically send your emissions data to the dashboard. You can also specify additional options in `@track_emissions()` or in `.codecarbon.config`. +`start()` registers the run on the API with a blocking HTTP call, and `tracker.run_id` then switches from a locally generated id to the API run id, so every output shares it. If the API is unreachable, tracking continues with the local id and switches to the API run id once the run is created. + ## Create Projects & Experiments By default, `codecarbon login` creates a default experiment in your first project. If you want to organize runs by experiment, you can specify an `experiment_id` explicitly. Set the experiment ID in two ways: diff --git a/tests/test_emissions_tracker.py b/tests/test_emissions_tracker.py index a2d6b5966..4a94fc6ad 100644 --- a/tests/test_emissions_tracker.py +++ b/tests/test_emissions_tracker.py @@ -356,6 +356,138 @@ def test_output_methods_boamps_adds_boamps_output_handler( ) ) + def test_run_id_with_api_output_is_never_none( + 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, + ): + with ( + mock.patch( + "codecarbon.output_methods.http.ApiClient._create_run" + ) as mock_create_run, + mock.patch("codecarbon.output_methods.http.ApiClient.add_emission"), + ): + tracker = EmissionsTracker( + output_dir=self.temp_path, + output_handlers=[], + output_methods=[OutputMethod.CSV, OutputMethod.API], + experiment_id="test-experiment-id", + api_key="test-api-key", + ) + api_output = next( + handler + for handler in tracker._output_handlers + if isinstance(handler, CodeCarbonAPIOutput) + ) + + def create_run(experiment_id): + api_output.api.run_id = "run-created" + return "run-created" + + mock_create_run.side_effect = create_run + + self.assertIsNotNone(tracker.run_id) + tracker.run_id = "caller-provided" + self.assertEqual(tracker.run_id, "caller-provided") + + tracker.start() + heavy_computation(1) + tracker.stop() + + self.assertEqual(tracker.run_id, "run-created") + # ...and it is what got persisted, instead of the string "None". + emissions_df = pd.read_csv(self.emissions_file_path) + self.assertEqual(emissions_df["run_id"].iloc[0], "run-created") + + def test_run_id_falls_back_to_uuid_when_api_run_creation_fails( + 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, + ): + with ( + mock.patch( + "codecarbon.output_methods.http.ApiClient._create_run", + side_effect=Exception("API is down"), + ), + mock.patch("codecarbon.output_methods.http.ApiClient.add_emission"), + ): + tracker = EmissionsTracker( + output_dir=self.temp_path, + output_handlers=[], + output_methods=[OutputMethod.CSV, OutputMethod.API], + experiment_id="test-experiment-id", + api_key="test-api-key", + ) + local_run_id = tracker.run_id + + with self.assertLogs("codecarbon", level="WARNING") as logs: + tracker.start() + self.assertTrue(any("API is down" in line for line in logs.output)) + self.assertIsNotNone(tracker._start_time) + + heavy_computation(1) + tracker.stop() + + self.assertEqual(tracker.run_id, local_run_id) + emissions_df = pd.read_csv(self.emissions_file_path) + self.assertEqual(emissions_df["run_id"].iloc[0], str(local_run_id)) + + def test_run_id_syncs_when_api_recovers_after_start( + 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, + ): + with ( + mock.patch( + "codecarbon.output_methods.http.ApiClient._create_run" + ) as mock_create_run, + mock.patch("codecarbon.output_methods.http.ApiClient.add_emission"), + ): + tracker = EmissionsTracker( + output_dir=self.temp_path, + output_handlers=[], + output_methods=[OutputMethod.CSV, OutputMethod.API], + experiment_id="test-experiment-id", + api_key="test-api-key", + ) + api_output = tracker._api_output + + # API down at start(), back up at the first emission. + def create_run(experiment_id): + if mock_create_run.call_count == 1: + raise Exception("API is down") + api_output.api.run_id = "run-created" + return "run-created" + + mock_create_run.side_effect = create_run + tracker.start() + local_run_id = tracker.run_id + self.assertNotEqual(local_run_id, "run-created") + + tracker.flush() + self.assertEqual(api_output.run_id, "run-created") + heavy_computation(1) + tracker.stop() + + self.assertEqual(tracker.run_id, "run-created") + emissions_df = pd.read_csv(self.emissions_file_path) + self.assertEqual(emissions_df["run_id"].iloc[-1], "run-created") + def test_default_output_methods_is_csv( self, mock_cli_setup,