Change Point Analysis

sdks/python/apache_beam/testing/analyzers/README.md

@@ -49,8 +49,8 @@ test_1:
   metric_name: mean_load_model_latency_milli_secs
   labels:
     - run-inference
-  min_runs_between_change_points: 5
-  num_runs_in_change_point_window: 7
+  min_runs_between_change_points: 5 # optional parameter
+  num_runs_in_change_point_window: 7 # optional parameter
 ```
 
 **NOTE**: `test_name` should be in the format `apache_beam.foo.bar`. It should point to a single test target.

sdks/python/apache_beam/testing/analyzers/github_issues_utils.py

@@ -53,7 +53,7 @@
   test: `{}` for the metric: `{}`.
 
   For more information on how to triage the alerts, please look at
-  `Triage performance alert issues` section of the [README](https://github.com/apache/beam/blob/3a453cd9fe2883eebed4b326d800b4cf8b2069e4/sdks/python/apache_beam/testing/analyzers/README.md).
+  `Triage performance alert issues` section of the [README](https://github.com/apache/beam/tree/master/sdks/python/apache_beam/testing/analyzers/README.md#triage-performance-alert-issues).
 """
 _METRIC_INFO_TEMPLATE = "timestamp: {}, metric_value: `{}`"
 _AWAITING_TRIAGE_LABEL = 'awaiting triage'
@@ -85,7 +85,7 @@ def create_issue(
       'labels': [_AWAITING_TRIAGE_LABEL, _PERF_ALERT_LABEL]
   }
   if labels:
-    data['labels'].extend(labels)
+    data['labels'].extend(labels)  # type: ignore
   response = requests.post(
       url=url, data=json.dumps(data), headers=_HEADERS).json()
   return response['number'], response['html_url']

sdks/python/apache_beam/testing/analyzers/perf_analysis.py

@@ -73,7 +73,7 @@ def run_change_point_analysis(params, test_id, big_query_metrics_fetcher):
   change_point_index = find_latest_change_point_index(
       metric_values=metric_values)
   if not change_point_index:
-    return
+    return False
   # since timestamps are ordered in ascending order and
   # num_runs_in_change_point_window refers to the latest runs,
   # latest_change_point_run can help determine if the change point
@@ -89,7 +89,7 @@ def run_change_point_analysis(params, test_id, big_query_metrics_fetcher):
             params['test_name'],
             latest_change_point_run,
             num_runs_in_change_point_window))
-    return
+    return False
 
   is_alert = True
   last_reported_issue_number = None
@@ -131,6 +131,8 @@ def run_change_point_analysis(params, test_id, big_query_metrics_fetcher):
     publish_issue_metadata_to_big_query(
         issue_metadata=issue_metadata, test_name=test_name)
 
+  return is_alert
+
 
 def run(config_file_path: Optional[str] = None) -> None:
   """

sdks/python/apache_beam/testing/analyzers/perf_analysis_test.py

@@ -21,6 +21,9 @@
 import time
 import unittest
 
+import mock
+import pandas as pd
+
 # pylint: disable=ungrouped-imports
 try:
   import apache_beam.testing.analyzers.perf_analysis as analysis
@@ -30,7 +33,30 @@
   from apache_beam.testing.analyzers.perf_analysis_utils import e_divisive
   from apache_beam.testing.analyzers.perf_analysis_utils import validate_config
 except ImportError as e:
-  analysis = None
+  analysis = None  # type: ignore
+
+
+# mock methods.
+def get_fake_data_with_no_change_point(**kwargs):
+  num_samples = 20
+  metric_values = [1 for i in range(num_samples)]
+  timestamps = [i for i in range(num_samples)]
+  return metric_values, timestamps
+
+
+def get_fake_data_with_change_point(**kwargs):
+  num_samples = 20
+  metric_values = [0] * (num_samples // 2) + [1] * (num_samples // 2)
+  timestamps = [i for i in range(num_samples)]
+  return metric_values, timestamps
+
+
+def get_existing_issue_data(**kwargs):
+  # change point found at index 10. So passing 10 in the
+  # existing issue data in mock method.
+  return pd.DataFrame([{
+      constants._CHANGE_POINT_TIMESTAMP_LABEL: 10, constants._ISSUE_NUMBER: 0
+  }])
 
 
 @unittest.skipIf(
@@ -43,6 +69,15 @@ def setUp(self) -> None:
     self.multiple_change_point_series = self.single_change_point_series + [
         2
     ] * 20
+    self.timestamps = [time.time() + 5 for i in range(5)]
+    self.params = {
+        'test_name': 'fake_test',
+        'metrics_dataset': 'fake_dataset',
+        'metrics_table': 'fake_table',
+        'project': 'fake_project',
+        'metric_name': 'fake_metric_name'
+    }
+    self.test_id = 'fake_id'
 
   def test_edivisive_means(self):
     change_point_indexes = e_divisive(self.single_change_point_series)
@@ -51,23 +86,20 @@ def test_edivisive_means(self):
     self.assertEqual(sorted(change_point_indexes), [10, 20])
 
   def test_is_changepoint_in_valid_window(self):
-
     changepoint_to_recent_run_window = 19
     change_point_index = 14
 
     is_valid = is_change_point_in_valid_window(
         changepoint_to_recent_run_window, change_point_index)
     self.assertEqual(is_valid, True)
 
-    changepoint_to_recent_run_window = 13
-    is_valid = is_change_point_in_valid_window(
-        changepoint_to_recent_run_window, change_point_index)
-    self.assertEqual(is_valid, False)
+  def test_is_change_point_in_invalid_window(self):
+    changepoint_to_recent_run_window = 12
+    change_point_index = 14
 
-    changepoint_to_recent_run_window = 14
     is_valid = is_change_point_in_valid_window(
         changepoint_to_recent_run_window, change_point_index)
-    self.assertEqual(is_valid, True)
+    self.assertEqual(is_valid, False)
 
   def test_validate_config(self):
     test_keys = {
@@ -80,34 +112,91 @@ def test_validate_config(self):
     self.assertEqual(test_keys, constants._PERF_TEST_KEYS)
     self.assertTrue(validate_config(test_keys))
 
-  def test_is_perf_alert(self):
-    timestamp_1 = time.time()
-    timestamps = [timestamp_1 + i for i in range(4, -1, -1)]
-
+  def test_duplicate_change_point(self):
     change_point_index = 2
     min_runs_between_change_points = 1
-
     is_alert = is_perf_alert(
-        previous_change_point_timestamps=[timestamps[3]],
-        timestamps=timestamps,
+        previous_change_point_timestamps=[self.timestamps[0]],
+        timestamps=self.timestamps,
         change_point_index=change_point_index,
         min_runs_between_change_points=min_runs_between_change_points)
-    self.assertFalse(is_alert)
+    self.assertTrue(is_alert)
 
+  def test_not_duplicate_change_point(self):
+    change_point_index = 2
+    min_runs_between_change_points = 1
     is_alert = is_perf_alert(
-        previous_change_point_timestamps=[timestamps[0]],
-        timestamps=timestamps,
+        previous_change_point_timestamps=[self.timestamps[3]],
+        timestamps=self.timestamps,
         change_point_index=change_point_index,
         min_runs_between_change_points=min_runs_between_change_points)
-    self.assertTrue(is_alert)
+    self.assertFalse(is_alert)
 
     is_alert = is_perf_alert(
-        previous_change_point_timestamps=[timestamps[0], timestamps[3]],
-        timestamps=timestamps,
+        previous_change_point_timestamps=[
+            self.timestamps[0], self.timestamps[3]
+        ],
+        timestamps=self.timestamps,
         change_point_index=change_point_index,
         min_runs_between_change_points=min_runs_between_change_points)
     self.assertFalse(is_alert)
 
+  @mock.patch(
+      'apache_beam.testing.analyzers.perf_analysis.fetch_metric_data',
+      get_fake_data_with_no_change_point)
+  def test_alert_on_data_with_no_change_point(self):
+    is_alert = analysis.run_change_point_analysis(
+        params=self.params,
+        test_id=self.test_id,
+        big_query_metrics_fetcher=None)
+    self.assertFalse(is_alert)
+
+  @mock.patch(
+      'apache_beam.testing.analyzers.perf_analysis.fetch_metric_data',
+      get_fake_data_with_change_point)
+  @mock.patch(
+      'apache_beam.testing.analyzers.perf_analysis.get_existing_issues_data',
+      lambda *args,
+      **kwargs: None)
+  @mock.patch(
+      'apache_beam.testing.analyzers.perf_analysis.'
+      'publish_issue_metadata_to_big_query',
+      lambda *args,
+      **kwargs: None)
+  @mock.patch(
+      'apache_beam.testing.analyzers.perf_analysis'
+      '.create_performance_alert',
+      lambda *args,
+      **kwargs: (0, ''))
+  def test_alert_on_data_with_change_point(self):
+    is_alert = analysis.run_change_point_analysis(
+        params=self.params,
+        test_id=self.test_id,
+        big_query_metrics_fetcher=None)
+    self.assertTrue(is_alert)
+
+  @mock.patch(
+      'apache_beam.testing.analyzers.perf_analysis.fetch_metric_data',
+      get_fake_data_with_change_point)
+  @mock.patch(
+      'apache_beam.testing.analyzers.perf_analysis.get_existing_issues_data',
+      get_existing_issue_data)
+  @mock.patch(
+      'apache_beam.testing.analyzers.perf_analysis.'
+      'publish_issue_metadata_to_big_query',
+      lambda *args,
+      **kwargs: None)
+  @mock.patch(
+      'apache_beam.testing.analyzers.perf_analysis.create_performance_alert',
+      lambda *args,
+      **kwargs: (0, ''))
+  def test_alert_on_data_with_reported_change_point(self):
+    is_alert = analysis.run_change_point_analysis(
+        params=self.params,
+        test_id=self.test_id,
+        big_query_metrics_fetcher=None)
+    self.assertFalse(is_alert)
+
 
 if __name__ == '__main__':
   logging.getLogger().setLevel(logging.DEBUG)