Regression Performance Cut 2

tests/performance/agents/metrics/__init__.py

@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-""" Customised system and mms process metrics for monitoring and pass-fail criteria in taurus"""
+""" Customised system and Model Server process metrics for monitoring and pass-fail criteria in taurus"""
 
 # Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # Licensed under the Apache License, Version 2.0 (the "License").
@@ -10,6 +10,7 @@
 # on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
 # express or implied. See the License for the specific language governing
 # permissions and limitations under the License.
+# pylint: disable=redefined-builtin, redefined-outer-name, broad-except, unused-variable
 
 from enum import Enum
 from statistics import mean
@@ -19,7 +20,7 @@
 
 
 class ProcessType(Enum):
-    """ Type of MMS processes to compute metrics on """
+    """ Type of Server processes to compute metrics on """
     FRONTEND = 1
     WORKER = 2
     ALL = 3
@@ -64,7 +65,8 @@ class ProcessType(Enum):
 misc_metrics = {
     'total_processes': None,
     'total_workers': None,
-    'orphans': None
+    'orphans': None,
+    'zombies': None
 }
 
 AVAILABLE_METRICS = list(system_metrics) + list(misc_metrics)
@@ -85,6 +87,7 @@ class ProcessType(Enum):
                 AVAILABLE_METRICS.append('{}_{}_{}'.format(op, PNAME, metric))
 
 children = set()
+zombie_children = set()
 
 
 def get_metrics(server_process, child_processes, logger):
@@ -116,22 +119,28 @@ def update_metric(metric_name, proc_type, stats):
     try:
         # as_dict() gets all stats in one shot
         processes_stats.append({'type': ProcessType.FRONTEND, 'stats': server_process.as_dict()})
-    except:
+    except Exception as e:
         pass
-    for child in children:
+
+    for child in children | zombie_children:
         try:
             child_cmdline = child.cmdline()
             if psutil.pid_exists(child.pid) and len(child_cmdline) >= 2 and WORKER_NAME in child_cmdline[1]:
                 processes_stats.append({'type': ProcessType.WORKER, 'stats': child.as_dict()})
             else:
                 reclaimed_pids.append(child)
                 logger.debug('child {0} no longer available'.format(child.pid))
-        except (NoSuchProcess, ZombieProcess):
+        except ZombieProcess:
+            zombie_children.add(child)
+        except NoSuchProcess:
             reclaimed_pids.append(child)
             logger.debug('child {0} no longer available'.format(child.pid))
 
     for p in reclaimed_pids:
-        children.remove(p)
+        if p in children:
+            children.remove(p)
+        if p in zombie_children:
+            zombie_children.remove(p)
 
     ### PROCESS METRICS ###
     worker_stats = list(map(lambda x: x['stats'], \
@@ -147,10 +156,11 @@ def update_metric(metric_name, proc_type, stats):
 
     # Total processes
     result['total_processes'] = len(worker_stats) + 1
-    result['total_workers'] = max(len(worker_stats) - 1, 0)
+    result['total_workers'] = len(worker_stats)
     result['orphans'] = len(list(filter(lambda p: p['ppid'] == 1, worker_stats)))
+    result['zombies'] = len(zombie_children)
 
-    ### SYSTEM METRICS ###
+    # ###SYSTEM METRICS ###
     result['system_disk_used'] = psutil.disk_usage('/').used
     result['system_memory_percent'] = psutil.virtual_memory().percent
     system_disk_io_counters = psutil.disk_io_counters()
@@ -160,3 +170,22 @@ def update_metric(metric_name, proc_type, stats):
     result['system_write_bytes'] = system_disk_io_counters.write_bytes
 
     return result
+
+
+if __name__ == "__main__":
+    import logging
+    import sys
+    from agents.utils.process import *
+    from agents import configuration
+
+    logger = logging.getLogger(__name__)
+    logging.basicConfig(stream=sys.stdout, format="%(message)s", level=logging.INFO)
+
+    PID_FILE = configuration.get('server', 'pid_file', 'model_server.pid')
+    server_pid = get_process_pid_from_file(get_server_pidfile(PID_FILE))
+    server_process = get_server_processes(server_pid)
+    children = get_child_processes(server_process)
+
+    metrics = get_metrics(server_process, children, logger)
+
+    print(metrics)

tests/performance/agents/utils/process.py

@@ -17,7 +17,6 @@
 
 import os
 import tempfile
-
 import psutil
 
 
@@ -56,9 +55,7 @@ def get_child_processes(process):
 
 
 def get_server_processes(server_process_pid):
-    """ It caches the main server and child processes at module level.
-    Ensure that you call this process so that MMS process
-    """
+    """get psutil Process object from process id """
     try:
         server_process = psutil.Process(server_process_pid)
     except Exception as e:
@@ -68,4 +65,5 @@ def get_server_processes(server_process_pid):
 
 
 def get_server_pidfile(file):
+    """get temp server pid file"""
     return os.path.join(tempfile.gettempdir(), ".{}".format(file))

tests/performance/requirements.txt

@@ -8,4 +8,5 @@ awscli==1.18.80
 click==7.1.2
 tabulate==0.8.7
 pandas==1.0.3
-termcolor==1.1.0
+termcolor==1.1.0
+bzt== 1.14.2

tests/performance/run_performance_suite.py

@@ -1,5 +1,3 @@
-
-
 # Copyright 2020 Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # Licensed under the Apache License, Version 2.0 (the "License").
 # You may not use this file except in compliance with the License.
@@ -13,20 +11,20 @@
 """
 Run Performance Regression Test Cases and Generate Reports
 """
-# pylint: disable=redefined-builtin, no-value-for-parameter
+# pylint: disable=redefined-builtin, no-value-for-parameter, unused-argument
 
 import logging
 import os
 import subprocess
 import sys
 import time
+import pathlib
 
 import click
-import pathlib
-from runs.context import ExecutionEnv
-from runs.taurus import get_taurus_options, x2junit, update_taurus_metric_files
 from tqdm import tqdm
 
+from runs.context import ExecutionEnv
+from runs.taurus import get_taurus_options, x2junit, update_taurus_metric_files
 from utils import run_process, Timer, get_sub_dirs
 
 logger = logging.getLogger(__name__)
@@ -71,8 +69,9 @@ def validate_env(ctx, param, value):
 @click.option('--monit/--no-monit', help='Start Monitoring server', default=True)
 @click.option('--compare-local/--no-compare-local', help='Compare with previous run with files stored'
                                                          ' in artifacts directory', default=True)
+@click.option('-c', '--compare-with', help='Compare with commit id, branch, tag, HEAD~N.', default="HEAD~1")
 def run_test_suite(artifacts_dir, test_dir, pattern, exclude_pattern,
-                   jmeter_path, env_name, monit, compare_local):
+                   jmeter_path, env_name, monit, compare_local, compare_with):
     """Collect test suites, run them and generate reports"""
 
     logger.info("Artifacts will be stored in directory %s", artifacts_dir)
@@ -84,8 +83,8 @@ def run_test_suite(artifacts_dir, test_dir, pattern, exclude_pattern,
     else:
         logger.info("Collected tests %s", test_dirs)
 
-    with ExecutionEnv(MONITORING_AGENT, artifacts_dir, env_name, compare_local, monit) as prt:
-        pre_command = 'export PYTHONPATH={}:$PYTHONPATH;'.format(os.path.join(str(ROOT_PATH), "agents"))
+    with ExecutionEnv(MONITORING_AGENT, artifacts_dir, env_name, compare_local, compare_with, monit) as prt:
+        pre_command = 'export PYTHONPATH={}:$PYTHONPATH;'.format(os.path.join(str(ROOT_PATH), "runs", "taurus", "override"))
         for suite_name in tqdm(test_dirs, desc="Test Suites"):
             with Timer("Test suite {} execution time".format(suite_name)) as t:
                 suite_artifacts_dir = os.path.join(artifacts_dir, suite_name)
@@ -95,10 +94,13 @@ def run_test_suite(artifacts_dir, test_dir, pattern, exclude_pattern,
                 test_file = os.path.join(test_dir, suite_name, "{}.yaml".format(suite_name))
                 with x2junit.X2Junit(suite_name, suite_artifacts_dir, prt.reporter, t, env_name) as s:
                     s.code, s.err = run_process("{} bzt {} {} {} {}".format(pre_command, options_str,
-                                                                            test_file, env_yaml_path,
-                                                                            GLOBAL_CONFIG_PATH))
+                                                                            GLOBAL_CONFIG_PATH, test_file,
+                                                                            env_yaml_path))
+
+                    update_taurus_metric_files(suite_artifacts_dir)
+
+    sys.exit(prt.exit_code)
 
-                    update_taurus_metric_files(suite_artifacts_dir, test_file)
 
 if __name__ == "__main__":
     run_test_suite()

tests/performance/runs/compare.py

@@ -15,9 +15,6 @@
 """
 # pylint: disable=redefined-builtin, self-assigning-variable, broad-except
 
-
-import csv
-import glob
 import logging
 import sys
 import os
@@ -34,16 +31,18 @@
 
 
 class CompareReportGenerator():
+    """Wrapper class to generate the compare report"""
 
-    def __init__(self, path, env_name, local_run):
+    def __init__(self, path, env_name, local_run, compare_with):
         self.artifacts_dir = path
         self.current_run_name = os.path.basename(path)
         self.env_name = env_name
+        self.comare_with = compare_with
         storage_class = LocalStorage if local_run else S3Storage
-        self.storage = storage_class(self.artifacts_dir, self.env_name)
+        self.storage = storage_class(self.artifacts_dir, self.env_name, compare_with)
         self.junit_reporter = None
         self.pandas_result = None
-        self.pass_fail =  True
+        self.pass_fail = True
 
     def gen(self):
         """Driver method to get comparison directory, do the comparison of it with current run directory
@@ -52,10 +51,11 @@ def gen(self):
         compare_dir, compare_run_name = self.storage.get_dir_to_compare()
         if compare_run_name:
             self.junit_reporter, self.pandas_result = compare_artifacts(self.storage.artifacts_dir, compare_dir,
-                                       self.storage.current_run_name, compare_run_name)
+                                                                        self.storage.current_run_name, compare_run_name)
             self.pandas_result.to_csv(os.path.join(self.artifacts_dir, "comparison_result.csv"))
         else:
-            logger.warning("The latest run not found for env.")
+            logger.info("The latest run for comparison was not found for env='%s' and commit_id='%s'.",
+                        self.env_name, self.comare_with)
 
         self.storage.store_results()
         return self.junit_reporter
@@ -87,7 +87,7 @@ def add_test_case(self, name, msg, type):
 
 def get_log_file(dir, sub_dir):
     """Get metric monitoring log files"""
-    metrics_file = os.path.join(dir, sub_dir, "metrics.csv")
+    metrics_file = os.path.join(dir, sub_dir, "metrics_agg.csv")
     return metrics_file if os.path.exists(metrics_file) else None
 
 
@@ -102,11 +102,21 @@ def get_aggregate_val(df, agg_func, col):
     return val
 
 
+def get_centile_val(df, agg_func, col):
+    """Get aggregate values of a pandas dataframe coulmn for given aggregate function"""
+
+    val = None
+    if "metric_name" in df and agg_func in df:
+        val = df[df["metric_name"] == col][agg_func]
+        val = val.iloc[0] if len(val) >= 1 else None
+    return val
+
+
 def compare_values(val1, val2, diff_percent, run_name1, run_name2):
     """ Compare percentage diff values of val1 and val2 """
     if pd.isna(val1) or pd.isna(val2):
-        msg = "Either of the value can not be determined. The run1 value is '{}' and " \
-              "run2 value is {}.".format(val1, val2)
+        msg = "Either of the value can not be determined. run1_value='{}' and " \
+              "run2_value='{}'.".format(val1, val2)
         pass_fail, diff, msg = "error", "NA", msg
     else:
         try:
@@ -116,15 +126,15 @@ def compare_values(val1, val2, diff_percent, run_name1, run_name2):
                 if diff < float(diff_percent):
                     pass_fail, diff, msg = "pass", diff, "passed"
                 else:
-                    msg = "The diff_percent criteria has failed. The expected diff_percent is '{}' and actual " \
-                          "diff percent is '{}' and the '{}' run value is '{}' and '{}' run value is '{}'. ". \
+                    msg = "The diff_percent criteria has failed. Expected='{}', actual='{}' " \
+                          "run1='{}', run1_value='{}', run2='{}', run2_value='{}' ". \
                         format(diff_percent, diff, run_name1, val1, run_name2, val2)
 
                     pass_fail, diff, msg = "fail", diff, msg
             else:  # special case of 0
                 pass_fail, diff, msg = "pass", 0, ""
         except Exception as e:
-            msg = "error while calculating the diff for val1={} and val2={}." \
+            msg = "error while calculating the diff for val1='{}' and val2='{}'." \
                   "Error is: {}".format(val1, val2, str(e))
             logger.info(msg)
             pass_fail, diff, msg = "pass", "NA", msg
@@ -139,7 +149,7 @@ def compare_artifacts(dir1, dir2, run_name1, run_name2):
     sub_dirs_1 = get_sub_dirs(dir1)
 
     over_all_pass = True
-    aggregates = ["mean", "max", "min"]
+    aggregates = ["first_value", "last_value"]
     header = ["run_name1", "run_name2", "test_suite", "metric", "run1", "run2",
               "percentage_diff", "expected_diff", "result", "message"]
     rows = [header]
@@ -161,14 +171,18 @@ def compare_artifacts(dir1, dir2, run_name1, run_name2):
 
             metrics_from_file1 = pd.read_csv(metrics_file1)
             metrics_from_file2 = pd.read_csv(metrics_file2)
-            metrics, diff_percents = taurus_reader.get_compare_metric_list(dir1, sub_dir1)
+            metrics = taurus_reader.get_compare_metric_list(dir1, sub_dir1)
 
-            for col, diff_percent in zip(metrics, diff_percents):
+            for metric_values in metrics:
+                col = metric_values[0]
+                diff_percent = metric_values[1]
+                if diff_percent is None:
+                    continue
                 for agg_func in aggregates:
                     name = "{}_{}".format(agg_func, str(col))
 
-                    val1 = get_aggregate_val(metrics_from_file1, agg_func, col)
-                    val2 = get_aggregate_val(metrics_from_file2, agg_func, col)
+                    val1 = get_centile_val(metrics_from_file1, agg_func, col)
+                    val2 = get_centile_val(metrics_from_file2, agg_func, col)
 
                     diff, pass_fail, msg = compare_values(val1, val2, diff_percent, run_name1, run_name2)
 
@@ -188,3 +202,10 @@ def compare_artifacts(dir1, dir2, run_name1, run_name2):
     dataframe = pd.DataFrame(rows[1:], columns=rows[0])
     return reporter, dataframe
 
+
+if __name__ == "__main__":
+    compare_artifacts(
+        "./run_artifacts/xlarge__5c35d98__1594819866",
+        "./run_artifacts/xlarge__f386038__1594819700",
+        "xlarge__5c35d98__1594819866", "xlarge__f386038__1594819700"
+    )