Change Point Analysis

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

@@ -17,11 +17,9 @@
     under the License.
 -->
 
-<h1>Performance alerts for Beam Python performance and load tests</h1>
-
-
-<h2> Alerts </h2>
+# Performance alerts for Beam Python performance and load tests
 
+##  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 +32,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 +53,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 +72,61 @@ 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.
 
+## Triage performance alert issues
+
+All the performance/load tests metrics defined at [beam/.test-infra/jenkins](https://github.com/apache/beam/tree/master/.test-infra/jenkins) are imported to [Grafana dashboards](http://104.154.241.245/d/1/getting-started?orgId=1) for visualization. Please 
+find the alerted test dashboard to find a spike in the metric values.
+
+For example, for the below configuration,
+* test: `apache_beam.testing.benchmarks.inference.pytorch_image_classification_benchmarks`
+* metric_name: `mean_load_model_latency_milli_secs`
+
+Grafana dashboard can be found at http://104.154.241.245/d/ZpS8Uf44z/python-ml-runinference-benchmarks?orgId=1&viewPanel=7
+
+If the dashboard for a test is not found, you can use the 
+code below to generate a plot for the given test, metric_name. 
+
+If the performance/load test store the results in BigQuery using this [schema](https://github.com/apache/beam/blob/83679216cce2d52dbeb7e837f06ca1d57b31d509/sdks/python/apache_beam/testing/load_tests/load_test_metrics_utils.py#L66),
+then use the following code to fetch the metric_values for a `metric_name` for the last `30` runs and display a plot using matplotlib.
+
+**NOTE**: Install matplotlib and pandas using `pip install matplotlib pandas`.
+```
+import pandas as pd
+import matplotlib.pyplot as plt
+
+from apache_beam.testing.load_tests import load_test_metrics_utils
+from apache_beam.testing.load_tests.load_test_metrics_utils import BigQueryMetricsFetcher
+
+bq_project = 'apache-beam-testing'
+bq_dataset = '<bq-dataset-name>'
+bq_table = '<bq-table>'
+metric_name = '<perf-alerted-metric-name>'
+
+query = f"""
+      SELECT *
+      FROM {bq_project}.{bq_dataset}.{bq_table}
+      WHERE CONTAINS_SUBSTR(({load_test_metrics_utils.METRICS_TYPE_LABEL}), '{metric_name}')
+      ORDER BY {load_test_metrics_utils.SUBMIT_TIMESTAMP_LABEL} DESC
+      LIMIT 30
+    """
+
+big_query_metrics_fetcher = BigQueryMetricsFetcher()
+metric_data: pd.DataFrame = big_query_metrics_fetcher.fetch(query=query)
+# sort the data to view it in chronological order.
+metric_data.sort_values(
+      by=[load_test_metrics_utils.SUBMIT_TIMESTAMP_LABEL], inplace=True)
+
+metric_data.plot(x=load_test_metrics_utils.SUBMIT_TIMESTAMP_LABEL,
+                 y=load_test_metrics_utils.VALUE_LABEL)
+plt.show()
+```
+
+If you confirm there is a change in the pattern of the values for a test, find the timestamp of when that change happened
+and use that timestamp to find possible culprit commit. 
 
+If the performance alert is a `false positive`, close the issue as `Close as not planned`.

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,20 @@
     "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: `{}`.
+
+  For more information on how to triage the alerts, please look at
+   `Triage performance alert issues` section of the README.
+"""
+_METRIC_INFO_TEMPLATE = "timestamp: {}, metric_value: `{}`"
 _AWAITING_TRIAGE_LABEL = 'awaiting triage'
 _PERF_ALERT_LABEL = 'perf-alert'
 
@@ -78,14 +87,14 @@ def create_issue(
       'labels': [_AWAITING_TRIAGE_LABEL, _PERF_ALERT_LABEL]
   }
   if labels:
-    data['labels'] += labels
+    data['labels'].extend(labels)
   response = requests.post(
       url=url, data=json.dumps(data), headers=_HEADERS).json()
   return response['number'], response['html_url']
 
 
 def comment_on_issue(issue_number: int,
-                     comment_description: str) -> Tuple[bool, Optional[str]]:
+                     comment_description: str) -> Tuple[bool, str]:
   """
   This method looks for an issue with provided issue_number. If an open
   issue is found, comment on the open issue with provided description else
@@ -96,8 +105,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)
@@ -119,17 +128,18 @@ def comment_on_issue(issue_number: int,
     response = requests.post(
         open_issue_response['comments_url'], json.dumps(data), headers=_HEADERS)
     return True, response.json()['html_url']
-  return False, None
+  return False, ''
 
 
-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 +149,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 +161,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 +193,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

@@ -21,7 +21,6 @@
 # regressions for benchmark/load/performance test.
 
 import argparse
-
 import logging
 import os
 import uuid
@@ -34,11 +33,11 @@
 import pandas as pd
 
 from apache_beam.testing.analyzers import constants
+from apache_beam.testing.analyzers.perf_analysis_utils import GitHubIssueMetaData
 from apache_beam.testing.analyzers.perf_analysis_utils import create_performance_alert
 from apache_beam.testing.analyzers.perf_analysis_utils import fetch_metric_data
-from apache_beam.testing.analyzers.perf_analysis_utils import get_existing_issues_data
 from apache_beam.testing.analyzers.perf_analysis_utils import find_latest_change_point_index
-from apache_beam.testing.analyzers.perf_analysis_utils import GitHubIssueMetaData
+from apache_beam.testing.analyzers.perf_analysis_utils import get_existing_issues_data
 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 publish_issue_metadata_to_big_query
@@ -76,14 +75,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 +105,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_test.py

@@ -14,6 +14,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
+# pytype: skip-file
 
 import logging
 import os