Upload macrobenchmark results to the metric service. (#2682)

Author: yifanyang

ci/fireci/fireci/uploader.py

@@ -19,13 +19,15 @@
 import subprocess
 import urllib.parse
 
+from . import prow_utils
+
 _logger = logging.getLogger('fireci.uploader')
 
 
-def post_report(test_report, metrics_service_url, access_token, note=''):
+def post_report(test_report, metrics_service_url, access_token, metric='reports', note=''):
   """Post a report to the metrics service backend."""
 
-  endpoint = _construct_request_endpoint(note)
+  endpoint = _construct_request_endpoint(metric, note)
   headers = {'Authorization': f'Bearer {access_token}', 'Content-Type': 'application/json'}
   data = json.dumps(test_report)
 
@@ -39,24 +41,26 @@ def post_report(test_report, metrics_service_url, access_token, note=''):
   _logger.info(f'Response: {result.text}')
 
 
-def _construct_request_endpoint(note):
+def _construct_request_endpoint(metric, note):
   repo_owner = os.getenv('REPO_OWNER')
   repo_name = os.getenv('REPO_NAME')
   branch = os.getenv('PULL_BASE_REF')
   pull_request = os.getenv('PULL_NUMBER')
 
   commit = _get_commit_hash('HEAD@{0}')
+  log = prow_utils.prow_job_log_link()
 
-  endpoint = f'/repos/{repo_owner}/{repo_name}/commits/{commit}/reports'
+  endpoint = f'/repos/{repo_owner}/{repo_name}/commits/{commit}/{metric}?log={log}'
   if pull_request:
-    base_commit = _get_commit_hash('HEAD@{1}')
-    endpoint += f'?pull_request={pull_request}&base_commit={base_commit}'
+    base_commit = os.getenv('PULL_BASE_SHA')
+    head_commit = os.getenv('PULL_PULL_SHA')
+    endpoint += f'&pull_request={pull_request}&base_commit={base_commit}&head_commit={head_commit}'
 
     commit_note = _get_prow_commit_note('HEAD@{0}')
     note += f'\n{commit_note}\n'
     endpoint += f'&note={urllib.parse.quote(note)}'
   else:
-    endpoint += f'?branch={branch}'
+    endpoint += f'&branch={branch}'
 
   return endpoint
 

ci/fireci/fireciplugins/macrobenchmark.py

@@ -20,17 +20,19 @@
 import random
 import re
 import shutil
-import statistics
 import sys
 import uuid
 
 import click
+import numpy
 import pystache
 import yaml
 from google.cloud import storage
 
 from fireci import ci_command
 from fireci.dir_utils import chdir
+from fireci import prow_utils
+from fireci import uploader
 
 _logger = logging.getLogger('fireci.macrobenchmark')
 
@@ -55,9 +57,7 @@ async def _launch_macrobenchmark_test():
     runners = [MacrobenchmarkTest(k, v, artifact_versions) for k, v in config.items()]
     results = await asyncio.gather(*[x.run() for x in runners], return_exceptions=True)
 
-  if any(map(lambda x: isinstance(x, Exception), results)):
-    _logger.error(f'Exceptions: {[x for x in results if (isinstance(x, Exception))]}')
-    raise click.ClickException('Macrobenchmark test failed with above errors.')
+  await _post_processing(results)
 
   _logger.info('Macrobenchmark test finished.')
 
@@ -104,6 +104,23 @@ async def _copy_google_services():
     shutil.copyfile(src, dst)
 
 
+async def _post_processing(results):
+  # Upload successful measurements to the metric service
+  measurements = []
+  for result in results:
+    if not isinstance(result, Exception):
+      measurements.extend(result)
+
+  metrics_service_url = os.getenv('METRICS_SERVICE_URL')
+  access_token = prow_utils.gcloud_identity_token()
+  uploader.post_report(measurements, metrics_service_url, access_token, metric='macrobenchmark')
+
+  # Raise exceptions for failed measurements
+  if any(map(lambda x: isinstance(x, Exception), results)):
+    _logger.error(f'Exceptions: {[x for x in results if isinstance(x, Exception)]}')
+    raise click.ClickException('Macrobenchmark test failed with above errors.')
+
+
 class MacrobenchmarkTest:
   """Builds the test based on configurations and runs the test on FTL."""
   def __init__(
@@ -127,7 +144,7 @@ async def run(self):
     await self._create_benchmark_projects()
     await self._assemble_benchmark_apks()
     await self._execute_benchmark_tests()
-    await self._upload_benchmark_results()
+    return await self._aggregate_benchmark_results()
 
   async def _create_benchmark_projects(self):
     app_name = self.test_app_config['name']
@@ -205,7 +222,7 @@ async def _prepare_mustache_context(self):
 
     return mustache_context
 
-  async def _upload_benchmark_results(self):
+  async def _aggregate_benchmark_results(self):
     results = []
     blobs = self.gcs_client.list_blobs(self.test_results_bucket, prefix=self.test_results_dir)
     files = [x for x in blobs if re.search(r'artifacts/[^/]*\.json', x.name)]
@@ -222,14 +239,13 @@ async def _upload_benchmark_results(self):
           'name': f'{clazz}.{method}',
           'min': min(runs),
           'max': max(runs),
-          'mean': statistics.mean(runs),
-          'median': statistics.median(runs),
-          'stdev': statistics.stdev(runs),
+          'p50': numpy.percentile(runs, 50),
+          'p90': numpy.percentile(runs, 90),
+          'p99': numpy.percentile(runs, 99),
           'unit': 'ms',
         })
     self.logger.info(f'Benchmark results: {results}')
-
-    # TODO(yifany): upload to metric service once it is ready
+    return results
 
   async def _exec_subprocess(self, executable, args):
     command = " ".join([executable, *args])

ci/fireci/setup.py

@@ -27,6 +27,7 @@
     install_requires=[
         'click==7.0',
         'google-cloud-storage==1.38.0',
+        'numpy==1.19.5',
         'PyGithub==1.43.8',
         'pystache==0.5.4',
         'requests==2.23.0',