Fix CPU usage calculation logic for live metrics

sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md

@@ -7,6 +7,8 @@
 ### Breaking Changes
 
 ### Bugs Fixed
+- Fix CPU usage calculation logic for live metrics
+  ([#45005](https://github.com/Azure/azure-sdk-for-python/pull/45005))
 
 ### Other Changes
 - Fix Ingestion-Side Sampling Disk Persist Behavior

sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_quickpulse/_cpu.py

@@ -41,15 +41,17 @@ def _get_process_time_normalized_old(options: CallbackOptions) -> Iterable[Obser
         # total process time is user + system in s
         total_time_s = cpu_times.user + cpu_times.system
         process_time_s = total_time_s - _get_quickpulse_last_process_time()
-        _set_quickpulse_last_process_time(process_time_s)
+        _set_quickpulse_last_process_time(total_time_s)
         # Find elapsed time in s since last collection
         current_time = datetime.now()
         elapsed_time_s = (current_time - _get_quickpulse_process_elapsed_time()).total_seconds()
         _set_quickpulse_process_elapsed_time(current_time)
         # Obtain cpu % by dividing by elapsed time
         cpu_percentage = process_time_s / elapsed_time_s
         # Normalize by dividing by amount of logical cpus
-        normalized_cpu_percentage = cpu_percentage / NUM_CPUS
+        normalized_cpu_percentage = (cpu_percentage / NUM_CPUS) * 100
+        # Cap at 100% to avoid edge cases where the CPU usage goes over 100%
+        normalized_cpu_percentage = min(normalized_cpu_percentage, 100)
         _set_quickpulse_last_process_cpu(normalized_cpu_percentage)
     except (psutil.NoSuchProcess, psutil.AccessDenied, ZeroDivisionError):
         pass

sdk/monitor/azure-monitor-opentelemetry-exporter/tests/quickpulse/test_cpu.py

@@ -48,7 +48,32 @@ def test_process_time(self, process_mock, process_time_mock, elapsed_time_mock):
             time = _get_process_time_normalized_old(None)
         obs = next(time)
         num_cpus = psutil.cpu_count()
-        self.assertAlmostEqual(obs.value, 1.2 / num_cpus, delta=1)
+        expected_value = (1.2 / num_cpus) * 100
+        self.assertAlmostEqual(obs.value, expected_value, delta=1)
+
+    @mock.patch("azure.monitor.opentelemetry.exporter._quickpulse._cpu._get_quickpulse_process_elapsed_time")
+    @mock.patch("azure.monitor.opentelemetry.exporter._quickpulse._cpu._get_quickpulse_last_process_time")
+    @mock.patch("azure.monitor.opentelemetry.exporter._quickpulse._cpu.PROCESS")
+    def test_process_time_capped_at_100(self, process_mock, process_time_mock, elapsed_time_mock):
+        """Test that CPU percentage is capped at 100% even when raw calculation exceeds it."""
+        current = datetime.now()
+        cpu = collections.namedtuple("cpu", ["user", "system"])
+        # Simulate extreme CPU usage that would calculate > 100%
+        # Total CPU time: 200 seconds
+        cpu_times = cpu(user=100.0, system=100.0)
+        process_mock.cpu_times.return_value = cpu_times
+        # Previous total was 0 (first measurement)
+        process_time_mock.return_value = 0.0
+        # Only 1 second elapsed
+        elapsed_time_mock.return_value = current - timedelta(seconds=1)
+        with mock.patch("datetime.datetime") as datetime_mock:
+            datetime_mock.now.return_value = current
+            time = _get_process_time_normalized_old(None)
+        obs = next(time)
+        num_cpus = psutil.cpu_count()
+        # Without cap: (200 / 1 / num_cpus) * 100 = (200 / 16) * 100 = 1250%
+        # With cap: Should be 100%
+        self.assertEqual(obs.value, 100)
 
 
 # cSpell:enable