Add relative median threshold filtering to perf alert tool (#28237)

* Add relative median threshold filtering

* Revert tests_config

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

@@ -29,14 +29,17 @@
 # pylint: disable=ungrouped-imports
 try:
   import apache_beam.testing.analyzers.perf_analysis as analysis
+  from apache_beam.io.filesystems import FileSystems
   from apache_beam.testing.analyzers import constants
   from apache_beam.testing.analyzers import github_issues_utils
   from apache_beam.testing.analyzers.perf_analysis_utils import is_change_point_in_valid_window
   from apache_beam.testing.analyzers.perf_analysis_utils import is_perf_alert
   from apache_beam.testing.analyzers.perf_analysis_utils import e_divisive
+  from apache_beam.testing.analyzers.perf_analysis_utils import filter_change_points_by_median_threshold
+  from apache_beam.testing.analyzers.perf_analysis_utils import find_change_points
   from apache_beam.testing.analyzers.perf_analysis_utils import find_latest_change_point_index
   from apache_beam.testing.analyzers.perf_analysis_utils import validate_config
-
+  from apache_beam.testing.load_tests import load_test_metrics_utils
 except ImportError as e:
   analysis = None  # type: ignore
 
@@ -222,6 +225,22 @@ def test_change_point_has_anomaly_marker_in_gh_description(self):
     match = re.search(pattern, runs_info)
     self.assertTrue(match)
 
+  def test_change_point_on_noisy_data(self):
+    def read_csv(path):
+      with FileSystems.open(path) as fp:
+        return pd.read_csv(fp)
+
+    metric_data = read_csv(
+        'gs://apache-beam-ml/testing/inputs/test_data_with_noise.csv')
+    metric_values = metric_data[load_test_metrics_utils.VALUE_LABEL].tolist()
+    change_points = find_change_points(metric_values)
+    self.assertEqual(change_points[0], 20)
+
+    # filter the noise.
+    valid_points = filter_change_points_by_median_threshold(
+        metric_values, change_points)
+    self.assertEqual(len(valid_points), 0)
+
 
 if __name__ == '__main__':
   logging.getLogger().setLevel(logging.DEBUG)

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

@@ -17,6 +17,7 @@
 import logging
 from dataclasses import asdict
 from dataclasses import dataclass
+from statistics import median
 from typing import Any
 from typing import Dict
 from typing import List
@@ -36,7 +37,7 @@
 from signal_processing_algorithms.energy_statistics.energy_statistics import e_divisive
 
 
-@dataclass
+@dataclass(frozen=True)
 class GitHubIssueMetaData:
   """
   This class holds metadata that needs to be published to the
@@ -149,19 +150,27 @@ def fetch_metric_data(
       metric_data[load_test_metrics_utils.SUBMIT_TIMESTAMP_LABEL].tolist())
 
 
+def find_change_points(metric_values: List[Union[float, int]]):
+  return e_divisive(metric_values)
+
+
 def find_latest_change_point_index(metric_values: List[Union[float, int]]):
   """
   Args:
    metric_values: Metric values used to run change point analysis.
   Returns:
    int: Right most change point index observed on metric_values.
   """
-  change_points_idx = e_divisive(metric_values)
-  if not change_points_idx:
-    return None
+  change_points_indices = find_change_points(metric_values)
+  # reduce noise in the change point analysis by filtering out
+  # the change points that are not significant enough.
+  change_points_indices = filter_change_points_by_median_threshold(
+      metric_values, change_points_indices)
   # Consider the latest change point.
-  change_points_idx.sort()
-  return change_points_idx[-1]
+  if not change_points_indices:
+    return None
+  change_points_indices.sort()
+  return change_points_indices[-1]
 
 
 def publish_issue_metadata_to_big_query(issue_metadata, table_name):
@@ -214,3 +223,33 @@ def create_performance_alert(
       'Performance regression/improvement is alerted on issue #%s. Link '
       ': %s' % (issue_number, issue_url))
   return issue_number, issue_url
+
+
+def filter_change_points_by_median_threshold(
+    data: List[Union[int, float]],
+    change_points: List[int],
+    threshold: float = 0.05,
+):
+  """
+  Reduces the number of change points by filtering out the ones that are
+  not significant enough based on the relative median threshold. Default
+  value of threshold is 0.05.
+  """
+  valid_change_points = []
+  epsilon = 1e-10  # needed to avoid division by zero.
+
+  for idx in change_points:
+    if idx == 0 or idx == len(data):
+      continue
+
+    left_segment = data[:idx]
+    right_segment = data[idx:]
+
+    left_value = median(left_segment)
+    right_value = median(right_segment)
+
+    relative_change = abs(right_value - left_value) / (left_value + epsilon)
+
+    if relative_change > threshold:
+      valid_change_points.append(idx)
+  return valid_change_points