Change Point Analysis

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

@@ -17,10 +17,10 @@
     under the License.
 -->
 
-<h1>Performance alerts for Beam Python performance and load tests</h1>
+# Performance alerts for Beam Python performance and load tests
 
 
-<h2> Alerts </h2>
+##  Alerts 
 
 Performance regressions or improvements detected with the [Change Point Analysis](https://en.wikipedia.org/wiki/Change_detection) using [edivisive](https://github.com/apache/beam/blob/0a91d139dea4276dc46176c4cdcdfce210fc50c4/.test-infra/jenkins/job_InferenceBenchmarkTests_Python.groovy#L30)
 analyzer are automatically filed as Beam GitHub issues with a label `perf-alert`.
@@ -34,7 +34,7 @@ If a performance alert is created on a test, a GitHub issue will be created and
 URL, issue number along with the change point value and timestamp are exported to BigQuery. This data will be used to analyze the next change point observed on the same test to
 update already created GitHub issue or ignore performance alert by not creating GitHub issue to avoid duplicate issue creation.
 
-<h2> Config file structure </h2>
+##  Config file structure 
 The config file defines the structure to run change point analysis on a given test. To add a test to the config file,
 please follow the below structure.
 
@@ -55,11 +55,13 @@ test_1:
   num_runs_in_change_point_window: 7
 ```
 
+**NOTE**: `test_name` should be in the format `apache_beam.foo.bar`. It should point to a single test target. 
+
 **Note**: If the source is **BigQuery**, the metrics_dataset, metrics_table, project and metric_name should match with the values defined for performance/load tests.
 The above example uses this [test configuration](https://github.com/apache/beam/blob/0a91d139dea4276dc46176c4cdcdfce210fc50c4/.test-infra/jenkins/job_InferenceBenchmarkTests_Python.groovy#L30)
 to fill up the values required to fetch the data from source.
 
-<h3>Different ways to avoid false positive change points</h3>
+### Different ways to avoid false positive change points
 
 **min_runs_between_change_points**: As the metric data moves across the runs, the change point analysis can place the
 change point in a slightly different place. These change points refer to the same regression and are just noise.
@@ -72,9 +74,10 @@ Sometimes, the change point found might be way back in time and could be irrelev
 reported only when it was observed in the last 7 runs from the current run,
 setting `num_runs_in_change_point_window=7` will achieve it.
 
-<h2> Register a test for performance alerts. </h2>
+##  Register a test for performance alerts. 
 
 If a new test needs to be registered for the performance alerting tool, please add the required test parameters to the
 config file.
 
+[//]: # (TODO : Add triaging section)
 

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

@@ -66,7 +66,3 @@
            }, {
                'name': _ISSUE_URL, 'field_type': 'STRING', 'mode': 'REQUIRED'
            }]
-
-_TITLE_TEMPLATE = """
-  Performance Regression or Improvement: {}:{}
-"""

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

@@ -44,11 +44,16 @@
     "X-GitHub-Api-Version": "2022-11-28"
 }
 
-# Fill the GitHub issue description with the below variables.
-_ISSUE_DESCRIPTION_HEADER = """
-  Affected metric: `{}`
+_ISSUE_TITLE_TEMPLATE = """
+  Performance Regression or Improvement: {}:{}
 """
-_METRIC_INFO = "timestamp: {}, metric_value: `{}`"
+
+# TODO: Provide a better debugging tool for the user to visualize metrics.
+# For example, a link to dashboards or query to get recent data to analyze,etc.
+_ISSUE_DESCRIPTION_TEMPLATE = """
+  Performance change found in the test: `{}` for the metric: `{}`.
+"""
+_METRIC_INFO_TEMPLATE = "timestamp: {}, metric_value: `{}`"
 _AWAITING_TRIAGE_LABEL = 'awaiting triage'
 _PERF_ALERT_LABEL = 'perf-alert'
 
@@ -96,8 +101,8 @@ def comment_on_issue(issue_number: int,
     comment_description: If an issue with issue_number is open,
       then comment on the issue with the using comment_description.
   Returns:
-    Boolean, indicating a comment was added to issue, and URL directing to
-     the comment.
+    tuple[bool, Optional[str]] indicating if a comment was added to
+      issue, and the comment URL.
   """
   url = 'https://api.github.com/repos/{}/{}/issues/{}'.format(
       _BEAM_GITHUB_REPO_OWNER, _BEAM_GITHUB_REPO_NAME, issue_number)
@@ -122,14 +127,15 @@ def comment_on_issue(issue_number: int,
   return False, None
 
 
-def add_awaiting_triage_label_to_issue(issue_number: int):
+def add_awaiting_triage_label(issue_number: int):
   url = 'https://api.github.com/repos/{}/{}/issues/{}/labels'.format(
       _BEAM_GITHUB_REPO_OWNER, _BEAM_GITHUB_REPO_NAME, issue_number)
   requests.post(
       url, json.dumps({'labels': [_AWAITING_TRIAGE_LABEL]}), headers=_HEADERS)
 
 
 def get_issue_description(
+    test_name: str,
     metric_name: str,
     timestamps: List[pd.Timestamp],
     metric_values: List,
@@ -139,8 +145,8 @@ def get_issue_description(
   Args:
    metric_name: Metric name used for the Change Point Analysis.
    timestamps: Timestamps of the metrics when they were published to the
-    Database.
-   metric_values: Values of the metric for the previous runs.
+    Database. Timestamps are expected in ascending order.
+   metric_values: metric values for the previous runs.
    change_point_index: Index for the change point. The element in the
     index of the metric_values would be the change point.
    max_results_to_display: Max number of results to display from the change
@@ -151,18 +157,19 @@ def get_issue_description(
   """
 
   # TODO: Add mean and median before and after the changepoint index.
-  upper_bound = min(
-      change_point_index + max_results_to_display + 1, len(metric_values))
-  lower_bound = max(0, change_point_index - max_results_to_display)
+  max_timestamp_index = min(
+      change_point_index + max_results_to_display, len(metric_values) - 1)
+  min_timestamp_index = max(0, change_point_index - max_results_to_display)
 
-  description = _ISSUE_DESCRIPTION_HEADER.format(metric_name) + 2 * '\n'
+  description = _ISSUE_DESCRIPTION_TEMPLATE.format(
+      test_name, metric_name) + 2 * '\n'
 
   runs_to_display = [
-      _METRIC_INFO.format(timestamps[i].ctime(), metric_values[i])
-      for i in range(lower_bound, upper_bound)
+      _METRIC_INFO_TEMPLATE.format(timestamps[i].ctime(), metric_values[i])
+      for i in range(max_timestamp_index, min_timestamp_index - 1, -1)
   ]
 
-  runs_to_display[change_point_index - lower_bound] += " <---- Anomaly"
+  runs_to_display[change_point_index - min_timestamp_index] += " <---- Anomaly"
   return description + '\n'.join(runs_to_display)
 
 
@@ -182,6 +189,6 @@ def report_change_point_on_issues(
           comment_description=description
           )
     if commented_on_issue:
-      add_awaiting_triage_label_to_issue(issue_number=existing_issue_number)
+      add_awaiting_triage_label(issue_number=existing_issue_number)
       return existing_issue_number, issue_url
   return create_issue(title=title, description=description, labels=labels)

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

@@ -76,14 +76,17 @@ def run_change_point_analysis(params, test_id, big_query_metrics_fetcher):
   if not change_point_index:
     return
 
-  if not is_change_point_in_valid_window(num_runs_in_change_point_window,
-                                         change_point_index):
+  if not is_change_point_in_valid_window(
+      num_runs_in_change_point_window,
+      len(timestamps) - 1 - change_point_index):
     logging.info(
         'Performance regression/improvement found for the test: %s. '
-        'Since the change point index %s '
+        'Since the change point run %s '
         'lies outside the num_runs_in_change_point_window distance: %s, '
-        'alert is not raised.' %
-        (test_name, change_point_index, num_runs_in_change_point_window))
+        'alert is not raised.' % (
+            params['test_name'],
+            len(timestamps) - 1 - change_point_index,
+            num_runs_in_change_point_window))
     return
 
   is_alert = True
@@ -103,13 +106,14 @@ def run_change_point_analysis(params, test_id, big_query_metrics_fetcher):
         timestamps=timestamps,
         min_runs_between_change_points=min_runs_between_change_points)
 
+  # TODO: remove before merging.
   logging.info("Performance alert is %s for test %s" % (is_alert, test_name))
   if is_alert:
     issue_number, issue_url = create_performance_alert(
-     metric_name, test_name, timestamps,
-     metric_values, change_point_index,
-     params.get('labels', None),
-      last_reported_issue_number)
+    metric_name, params['test_name'], timestamps,
+    metric_values, change_point_index,
+    params.get('labels', None),
+    last_reported_issue_number)
 
     issue_metadata = GitHubIssueMetaData(
         issue_timestamp=pd.Timestamp(

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

@@ -31,8 +31,7 @@
 from google.api_core import exceptions
 
 from apache_beam.testing.analyzers import constants
-from apache_beam.testing.analyzers.github_issues_utils import get_issue_description
-from apache_beam.testing.analyzers.github_issues_utils import report_change_point_on_issues
+from apache_beam.testing.analyzers import github_issues_utils
 from apache_beam.testing.load_tests import load_test_metrics_utils
 from apache_beam.testing.load_tests.load_test_metrics_utils import BigQueryMetricsFetcher
 from apache_beam.testing.load_tests.load_test_metrics_utils import BigQueryMetricsPublisher
@@ -58,8 +57,6 @@ class GitHubIssueMetaData:
 
 def is_change_point_in_valid_window(
     num_runs_in_change_point_window: int, change_point_index: int) -> bool:
-  # If the change point is more than N runs behind the most recent run,
-  # Ignore the change point and don't raise an alert for it.
   return num_runs_in_change_point_window >= change_point_index
 
 
@@ -94,14 +91,15 @@ def is_perf_alert(
   Search the previous_change_point_timestamps with current observed
   change point sibling window and determine if it is a duplicate
   change point or not.
+  timestamps are expected to be in ascending order.
 
   Return False if the current observed change point is a duplicate of
   already reported change points else return True.
   """
-  sibling_change_point_min_timestamp = timestamps[min(
-      change_point_index + min_runs_between_change_points, len(timestamps) - 1)]
-  sibling_change_point_max_timestamp = timestamps[max(
+  sibling_change_point_min_timestamp = timestamps[max(
       0, change_point_index - min_runs_between_change_points)]
+  sibling_change_point_max_timestamp = timestamps[min(
+      change_point_index + min_runs_between_change_points, len(timestamps) - 1)]
   # Search a list of previous change point timestamps and compare it with
   # current change point timestamp. We do this in case, if a current change
   # point is already reported in the past.
@@ -129,6 +127,15 @@ def validate_config(keys):
 def fetch_metric_data(
     params: Dict[str, Any], big_query_metrics_fetcher: BigQueryMetricsFetcher
 ) -> Tuple[List[Union[int, float]], List[pd.Timestamp]]:
+  """
+  Args:
+   params: Dict containing keys required to fetch data from a data source.
+   big_query_metrics_fetcher: A BigQuery metrics fetcher for fetch metrics.
+  Returns:
+    Tuple[List[Union[int, float]], List[pd.Timestamp]]: Tuple containing list
+    of metric_values and list of timestamps. Both are sorted in ascending
+    order wrt timestamps.
+  """
   query = f"""
       SELECT *
       FROM {params['project']}.{params['metrics_dataset']}.{params['metrics_table']}
@@ -137,18 +144,26 @@ def fetch_metric_data(
       LIMIT {constants._NUM_DATA_POINTS_TO_RUN_CHANGE_POINT_ANALYSIS}
     """
   metric_data: pd.DataFrame = big_query_metrics_fetcher.fetch(query=query)
+  metric_data.sort_values(
+      by=[load_test_metrics_utils.SUBMIT_TIMESTAMP_LABEL], inplace=True)
   return (
-      metric_data[load_test_metrics_utils.VALUE_LABEL],
-      metric_data[load_test_metrics_utils.SUBMIT_TIMESTAMP_LABEL])
+      metric_data[load_test_metrics_utils.VALUE_LABEL].tolist(),
+      metric_data[load_test_metrics_utils.SUBMIT_TIMESTAMP_LABEL].tolist())
 
 
 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
   # Consider the latest change point.
-  change_points_idx.sort(reverse=True)
-  change_point_index = change_points_idx[0]
+  change_points_idx.sort()
+  change_point_index = change_points_idx[-1]
   return change_point_index
 
 
@@ -179,16 +194,19 @@ def create_performance_alert(
   Creates performance alert on GitHub issues and returns GitHub issue
   number and issue URL.
   """
-  description = get_issue_description(
+  description = github_issues_utils.get_issue_description(
+      test_name=test_name,
       metric_name=metric_name,
       timestamps=timestamps,
       metric_values=metric_values,
       change_point_index=change_point_index,
       max_results_to_display=(
           constants._NUM_RESULTS_TO_DISPLAY_ON_ISSUE_DESCRIPTION))
 
-  issue_number, issue_url = report_change_point_on_issues(
-        title=constants._TITLE_TEMPLATE.format(test_name, metric_name),
+  issue_number, issue_url = github_issues_utils.report_change_point_on_issues(
+        title=github_issues_utils._ISSUE_TITLE_TEMPLATE.format(
+          test_name, metric_name
+        ),
         description=description,
         labels=labels,
         existing_issue_number=existing_issue_number)

sdks/python/apache_beam/testing/analyzers/tests_config.yaml

@@ -16,15 +16,16 @@
 #
 
 test_1:
-  test_name: apache_beam.testing.benchmarks.inference.pytorch_image_classification_benchmarks
+  test_name: apache_beam.testing.benchmarks.inference.pytorch_image_classification_benchmarks_16
   metrics_dataset: beam_run_inference
   metrics_table: torch_inference_imagenet_results_resnet152
   project: apache-beam-testing
   metric_name: mean_load_model_latency_milli_secs
   labels:
     - run-inference
+  # TODO: Remove before merging.
   min_runs_between_change_points: 5
-  num_runs_in_change_point_window: 7
+  num_runs_in_change_point_window: 14
 
 test_2:
   test_name: apache_beam.testing.benchmarks.inference.pytorch_image_classification_benchmarks
@@ -34,5 +35,3 @@ test_2:
   metric_name: mean_load_model_latency_milli_secs
   labels:
     - run-inference
-  min_runs_between_change_points: 5
-  num_runs_in_change_point_window: 7