Merge pull request #20 from aws/issue-19

gimki · web-flow · commit f49ff7ed8dda · 2021-03-10T11:34:06.000Z
Improve implementation of ProfilerDisabler

Introduce two layer of check: should_stop_sampling and should_stop_profiling


should_stop_sampling is checked whenever we sample (usually every second) and it is responsible to make sure the sampling overhead is low.

should_stop_profiling is checked whenever a profiling cycle end (whenever we submit profile) and it is responsible to make sure the profiling cycle overhead is low. 

The major difference between the two check is the way of calculation of overhead. should_stop_sampling used "average cpu usage for sample / sampling interval" while should_stop_profiling used "total cpu usage for full profiling cycle / profile.get_active_millis_since_start".
diff --git a/codeguru_profiler_agent/local_aggregator.py b/codeguru_profiler_agent/local_aggregator.py
@@ -44,7 +44,7 @@ def __init__(self, reporter, environment=dict()):
         self.memory_limit_bytes = environment["memory_limit_bytes"]
         self.last_report_attempted = current_milli_time(clock=self.clock)
 
-        self._reset()
+        self.reset()
 
     def add(self, sample):
         """
@@ -69,7 +69,7 @@ def _check_memory_limit(self):
                     "Profiler memory usage limit has been reached")
             self.flush(force=True)
 
-    def _reset(self):
+    def reset(self):
         self.profile = self.profile_factory(
             profiling_group_name=self.profiling_group_name,
             sampling_interval_seconds=AgentConfiguration.get().sampling_interval.total_seconds(),
@@ -80,7 +80,7 @@ def _reset(self):
         self.timer.reset()
 
     @with_timer("flush")
-    def flush(self, force=False):
+    def flush(self, force=False, reset=True):
         now = current_milli_time(clock=self.clock)
         reported = False
         if not force and not self._is_over_reporting_interval(now):
@@ -92,8 +92,8 @@ def flush(self, force=False):
             self._report_profile(now)
             reported = True
 
-        if force or reported:
-            self._reset()
+        if force or (reset and reported):
+            self.reset()
         return reported
 
     def refresh_configuration(self):
diff --git a/codeguru_profiler_agent/profiler_disabler.py b/codeguru_profiler_agent/profiler_disabler.py
@@ -2,7 +2,6 @@
 import time
 import logging
 from codeguru_profiler_agent.reporter.agent_configuration import AgentConfiguration
-from codeguru_profiler_agent.utils.time import current_milli_time
 
 logger = logging.getLogger(__name__)
 CHECK_KILLSWITCH_FILE_INTERVAL_SECONDS = 60
@@ -20,13 +19,18 @@ def __init__(self, environment, clock=time.time):
         self.killswitch = KillSwitch(environment['killswitch_filepath'], clock)
         self.memory_limit_bytes = environment['memory_limit_bytes']
 
+    def should_stop_sampling(self, profile=None):
+        return (self.killswitch.is_killswitch_on()
+                or self.cpu_usage_check.is_sampling_cpu_usage_limit_reached(profile)
+                or self._is_memory_limit_reached(profile))
+
     def should_stop_profiling(self, profile=None):
         return (self.killswitch.is_killswitch_on()
-                or self.cpu_usage_check.is_cpu_usage_limit_reached(profile)
-                or profile is not None and self._is_memory_limit_reached(profile))
+                or self.cpu_usage_check.is_overall_cpu_usage_limit_reached(profile)
+                or self._is_memory_limit_reached(profile))
 
     def _is_memory_limit_reached(self, profile):
-        return profile.get_memory_usage_bytes() > self.memory_limit_bytes
+        return False if profile is None else profile.get_memory_usage_bytes() > self.memory_limit_bytes
 
 
 class CpuUsageCheck:
@@ -38,19 +42,45 @@ class CpuUsageCheck:
     def __init__(self, timer):
         self.timer = timer
 
-    def is_cpu_usage_limit_reached(self, profile=None):
+    def is_overall_cpu_usage_limit_reached(self, profile=None):
+        """
+        This function carries out an overall cpu limit check that covers the cpu overhead caused for the full
+        sampling cycle: refresh config -> (sample -> aggregate) * n -> profile submission. We expect this function to
+        be called after profile submission.
+        """
         profiler_metric = self.timer.metrics.get("runProfiler")
-        if not profiler_metric or profiler_metric.counter < MINIMUM_MEASURES_IN_DURATION_METRICS:
+        if not profile or not profiler_metric or profiler_metric.counter < MINIMUM_MEASURES_IN_DURATION_METRICS:
+            return False
+
+        used_time_percentage = 100 * profiler_metric.total/(profile.get_active_millis_since_start()/1000)
+
+        cpu_limit_percentage = AgentConfiguration.get().cpu_limit_percentage
+
+        if used_time_percentage >= cpu_limit_percentage:
+            logger.debug(self.timer.metrics)
+            logger.info(
+                "Profiler overall cpu usage limit reached: {:.2f} % (limit: {:.2f} %), will stop CodeGuru Profiler."
+                .format(used_time_percentage, cpu_limit_percentage))
+            return True
+        else:
+            return False
+
+    def is_sampling_cpu_usage_limit_reached(self, profile=None):
+        sample_and_aggregate_metric = self.timer.metrics.get("sampleAndAggregate")
+        if not sample_and_aggregate_metric or \
+                sample_and_aggregate_metric.counter < MINIMUM_MEASURES_IN_DURATION_METRICS:
             return False
 
         sampling_interval_seconds = self._get_average_sampling_interval_seconds(profile)
-        used_time_percentage = 100 * profiler_metric.average() / sampling_interval_seconds
+        used_time_percentage = 100 * sample_and_aggregate_metric.average() / sampling_interval_seconds
+
+        cpu_limit_percentage = AgentConfiguration.get().cpu_limit_percentage
 
-        if used_time_percentage >= AgentConfiguration.get().cpu_limit_percentage:
+        if used_time_percentage >= cpu_limit_percentage:
             logger.debug(self.timer.metrics)
             logger.info(
-                "Profiler cpu usage limit reached: {:.2f} % (limit: {:.2f} %), will stop CodeGuru Profiler.".format(
-                    used_time_percentage, AgentConfiguration.get().cpu_limit_percentage))
+                "Profiler sampling cpu usage limit reached: {:.2f} % (limit: {:.2f} %), will stop CodeGuru Profiler."
+                .format(used_time_percentage, cpu_limit_percentage))
             return True
         else:
             return False
diff --git a/codeguru_profiler_agent/profiler_runner.py b/codeguru_profiler_agent/profiler_runner.py
@@ -71,27 +71,37 @@ def _profiling_command(self):
             if self._first_execution:
                 self.collector.setup()
                 self._first_execution = False
-            return self._run_profiler()
+            sample_result = self._run_profiler()
+            if sample_result.success and sample_result.is_end_of_cycle:
+                if self.profiler_disabler.should_stop_profiling(profile=self.collector.profile):
+                    return False
+                self.collector.reset()
+                return True
+            return sample_result.success
         except:
             logger.info("An unexpected issue caused the profiling command to terminate.", exc_info=True)
             return False
 
     @with_timer("runProfiler")
     def _run_profiler(self):
-        if self.profiler_disabler.should_stop_profiling(self.collector.profile):
-            return False
+        if self.profiler_disabler.should_stop_sampling(self.collector.profile):
+            return RunProfilerStatus(success=False, is_end_of_cycle=False)
 
         if not self.is_profiling_in_progress:
             self._refresh_configuration()
 
         # after the refresh we may be working on a profile
         if self.is_profiling_in_progress:
-            if self.collector.flush():
+            if self.collector.flush(reset=False):
                 self.is_profiling_in_progress = False
-                return True
-            sample = self.sampler.sample()
-            self.collector.add(sample)
-        return True
+                return RunProfilerStatus(success=True, is_end_of_cycle=True)
+            self._sample_and_aggregate()
+        return RunProfilerStatus(success=True, is_end_of_cycle=False)
+
+    @with_timer("sampleAndAggregate")
+    def _sample_and_aggregate(self):
+        sample = self.sampler.sample()
+        self.collector.add(sample)
 
     def is_running(self):
         return self.scheduler.is_running()
@@ -125,3 +135,9 @@ def pause(self, block=False):
         """
         self.scheduler.pause(block)
         self.collector.profile.pause()
+
+
+class RunProfilerStatus:
+    def __init__(self, success, is_end_of_cycle):
+        self.success = success
+        self.is_end_of_cycle = is_end_of_cycle
diff --git a/test/acceptance/test_cpu_limit.py b/test/acceptance/test_cpu_limit.py
@@ -31,6 +31,6 @@ def test_profiler_terminates(self):
             # With sampling_interval to be 0.01 seconds, having runProfiler as 0.5 seconds should breach
             # the cpu limit. We need to sample 20 times before we check the CPU limit
             for i in range(20):
-                self.timer.record('runProfiler', 0.5)
+                self.timer.record('sampleAndAggregate', 0.5)
 
             assert wait_for(lambda: not self.profiler.is_running(), timeout_seconds=5)
diff --git a/test/unit/test_profiler_disabler.py b/test/unit/test_profiler_disabler.py
@@ -28,10 +28,10 @@ def set_agent_config(sampling_interval_seconds=1, cpu_limit_percentage=DEFAULT_C
 
 
 def assert_config_sampling_interval_used(process_duration_check, profile):
-    assert process_duration_check.is_cpu_usage_limit_reached(profile)
+    assert process_duration_check.is_sampling_cpu_usage_limit_reached(profile)
 
     set_agent_config(sampling_interval_seconds=42, cpu_limit_percentage=80)
-    assert not process_duration_check.is_cpu_usage_limit_reached(profile)
+    assert not process_duration_check.is_sampling_cpu_usage_limit_reached(profile)
 
 
 class TestProfilerDisabler:
@@ -59,28 +59,57 @@ def test_it_sets_all_parameters(self):
         assert AgentConfiguration.get().cpu_limit_percentage == DEFAULT_CPU_LIMIT_PERCENTAGE
 
 
-class TestWhenAnyFails(TestProfilerDisabler):
-    @before
-    def before(self):
-        super().before()
-        self.profiler = Mock()
-        self.disabler.killswitch = Mock()
-        self.disabler.cpu_usage_check = Mock()
-        self.disabler._is_memory_limit_reached = Mock(return_value=False)
-        self.disabler.killswitch.is_killswitch_on = Mock(return_value=False)
-        self.disabler.killswitch.is_process_duration_limit_reached = Mock(return_value=False)
+class TestShouldStopSampling:
+    class TestWhenAnyFails(TestProfilerDisabler):
+        @before
+        def before(self):
+            super().before()
+            self.disabler.killswitch = Mock()
+            self.disabler.cpu_usage_check = Mock()
+            self.disabler.cpu_usage_check.is_sampling_cpu_usage_limit_reached = Mock(return_value=False)
+            self.disabler._is_memory_limit_reached = Mock(return_value=False)
+            self.disabler.killswitch.is_killswitch_on = Mock(return_value=False)
+            self.disabler.killswitch.is_process_duration_limit_reached = Mock(return_value=False)
+            assert not self.disabler.should_stop_sampling()
+
+        def test_it_stops_profiling_if_killswitch_is_on(self):
+            self.disabler.killswitch.is_killswitch_on = Mock(return_value=True)
+            assert self.disabler.should_stop_sampling()
+
+        def test_it_stops_profiling_if_memory_limit_is_reached(self):
+            self.disabler._is_memory_limit_reached = Mock(return_value=True)
+            assert self.disabler.should_stop_sampling()
+
+        def test_it_stops_profiling_if_process_duration_is_reached(self):
+            self.disabler.cpu_usage_check.is_sampling_cpu_usage_limit_reached = Mock(return_value=True)
+            assert self.disabler.should_stop_sampling()
+
+
+class TestShouldStopProfiling:
+    class TestWhenAnyFails(TestProfilerDisabler):
+        @before
+        def before(self):
+            super().before()
+            self.profiler = Mock()
+            self.disabler.killswitch = Mock()
+            self.disabler.cpu_usage_check = Mock()
+            self.disabler.cpu_usage_check.is_overall_cpu_usage_limit_reached = Mock(return_value=False)
+            self.disabler._is_memory_limit_reached = Mock(return_value=False)
+            self.disabler.killswitch.is_killswitch_on = Mock(return_value=False)
+            self.disabler.killswitch.is_process_duration_limit_reached = Mock(return_value=False)
+            assert not self.disabler.should_stop_profiling()
 
-    def test_it_stops_profiling_if_killswitch_is_on(self):
-        self.disabler.killswitch.is_killswitch_on = Mock(return_value=True)
-        assert self.disabler.should_stop_profiling(self.profiler)
+        def test_it_stops_profiling_if_killswitch_is_on(self):
+            self.disabler.killswitch.is_killswitch_on = Mock(return_value=True)
+            assert self.disabler.should_stop_profiling()
 
-    def test_it_stops_profiling_if_memory_limit_is_reached(self):
-        self.disabler._is_memory_limit_reached = Mock(return_value=True)
-        assert self.disabler.should_stop_profiling(self.profiler)
+        def test_it_stops_profiling_if_memory_limit_is_reached(self):
+            self.disabler._is_memory_limit_reached = Mock(return_value=True)
+            assert self.disabler.should_stop_profiling()
 
-    def test_it_stops_profiling_if_process_duration_is_reached(self):
-        self.disabler.cpu_usage_check.is_cpu_usage_limit_reached = Mock(return_value=True)
-        assert self.disabler.should_stop_profiling(self.profiler)
+        def test_it_stops_profiling_if_process_duration_is_reached(self):
+            self.disabler.cpu_usage_check.is_overall_cpu_usage_limit_reached = Mock(return_value=True)
+            assert self.disabler.should_stop_profiling()
 
 
 class TestKillSwitch:
@@ -145,17 +174,17 @@ def test_it_returns_false_after_a_minute(self):
         assert not self.killswitch.is_killswitch_on()
 
 
-class TestCpuUsageCheck:
+class TestSamplingCpuUsageCheck:
     def before(self):
         self.timer = Timer()
         self.profile = Mock(spec=Profile)
         for i in range(20):
-            self.timer.record('runProfiler', 0.5)
+            self.timer.record('sampleAndAggregate', 0.5)
         set_agent_config(sampling_interval_seconds=1, cpu_limit_percentage=10)
         self.process_duration_check = CpuUsageCheck(self.timer)
 
 
-class TestGetAverageSamplingIntervalSeconds(TestCpuUsageCheck):
+class TestGetAverageSamplingIntervalSeconds(TestSamplingCpuUsageCheck):
     @before
     def before(self):
         super().before()
@@ -176,7 +205,7 @@ def test_when_profiler_sample_count_less_than_min_samples_in_profile_it_returns_
         assert CpuUsageCheck._get_average_sampling_interval_seconds(self.profile) == 23
 
 
-class TestIsCpuUsageLimitReached(TestCpuUsageCheck):
+class TestIsSamplingCpuUsageLimitReached(TestSamplingCpuUsageCheck):
     @before
     def before(self):
         super().before()
@@ -187,43 +216,70 @@ def before(self):
             yield
 
     def test_it_calls_get_average_sampling_interval_with_profile(self):
-        self.process_duration_check.is_cpu_usage_limit_reached(self.profile)
+        self.process_duration_check.is_sampling_cpu_usage_limit_reached(self.profile)
         self.get_average_sampling_interval_mock.assert_called_once_with(self.profile)
 
     def test_when_average_duration_exceeds_limit_it_returns_true(self):
         # timer: (0.5/4) * 100= 12.5%
-        assert self.process_duration_check.is_cpu_usage_limit_reached()
+        assert self.process_duration_check.is_sampling_cpu_usage_limit_reached()
 
-    def test_when_average_duragtion_is_below_limit_it_returns_false(self):
+    def test_when_average_duration_is_below_limit_it_returns_false(self):
         # timer: (0.5/4) * 100= 12.5%
         set_agent_config(cpu_limit_percentage=13)
-        assert not self.process_duration_check.is_cpu_usage_limit_reached()
+        assert not self.process_duration_check.is_sampling_cpu_usage_limit_reached()
 
     def test_when_profile_is_none_it_calls_get_average_sampling_interval_without_profile(self):
-        self.process_duration_check.is_cpu_usage_limit_reached()
+        self.process_duration_check.is_sampling_cpu_usage_limit_reached()
         self.get_average_sampling_interval_mock.assert_called_once_with(None)
 
 
-class TestWhenTimerDoesNotHaveTheKey(TestCpuUsageCheck):
+class TestIsOverallCpuUsageLimitReached():
+    @before
+    def before(self):
+        self.timer = Timer()
+        self.profile = Mock(spec=Profile)
+        for i in range(20):
+            self.timer.record('runProfiler', 0.5)
+        set_agent_config(cpu_limit_percentage=9)
+        self.process_duration_check = CpuUsageCheck(self.timer)
+        self.profile.get_active_millis_since_start = Mock(return_value=100*1000)
+
+    def test_when_average_duration_exceeds_limit_it_returns_true(self):
+        # timer: (0.5*20/100) * 100= 10%
+        assert self.process_duration_check.is_overall_cpu_usage_limit_reached(self.profile)
+
+    def test_when_average_duration_is_below_limit_it_returns_false(self):
+        # timer: (0.5*20/100) * 100= 10%
+        set_agent_config(cpu_limit_percentage=11)
+        assert not self.process_duration_check.is_overall_cpu_usage_limit_reached(self.profile)
+
+    def test_when_profile_is_none_it_returns_false(self):
+        assert not self.process_duration_check.is_overall_cpu_usage_limit_reached()
+
+
+class TestWhenTimerDoesNotHaveTheKey(TestSamplingCpuUsageCheck):
     @before
     def before(self):
         super().before()
 
     def test_it_returns_false(self):
         self.process_duration_check.timer = Timer()
-        assert not self.process_duration_check.is_cpu_usage_limit_reached()
+        assert not self.process_duration_check.is_sampling_cpu_usage_limit_reached()
 
 
-class TestWhenTimerDoesNotHaveEnoughMeasures(TestCpuUsageCheck):
+class TestWhenTimerDoesNotHaveEnoughMeasures(TestSamplingCpuUsageCheck):
     @before
     def before(self):
         super().before()
-
-    def test_it_returns_false(self):
         self.timer.reset()
         for i in range(4):
-            self.timer.record('runProfiler', 0.5)
-        assert not self.process_duration_check.is_cpu_usage_limit_reached()
+            self.timer.record('sampleAndAggregate', 0.5)
+
+    def test_sampling_cpu_usage_limit_reached_returns_false(self):
+        assert not self.process_duration_check.is_sampling_cpu_usage_limit_reached()
+
+    def test_overall_cpu_usage_limit_reached_returns_false(self):
+        assert not self.process_duration_check.is_overall_cpu_usage_limit_reached()
 
 
 class TestMemoryLimitCheck:
diff --git a/test/unit/test_profiler_runner.py b/test/unit/test_profiler_runner.py
