Add support for ReBenchDB API Version 2 (#236)

rebench/__init__.py

@@ -1 +1 @@
-__version__ = "1.2.0"
+__version__ = "1.2.1.dev1"

rebench/model/data_point.py

@@ -77,5 +77,49 @@ def measurements_as_dict(self, criteria):
             'm': data
         }
 
+    def add_measurements_api_v20(self, criteria, data):
+        num_measurements = 0
+        iteration = -1
+        invocation = -1
+
+        # data contains a list of hashes
+        # with {in: n, m: [[], [], ...]}
+        # where m is a list of values for each criterion
+
+        ms = None
+
+        for m in self._measurements:
+            if iteration == -1:
+                iteration = m.iteration
+                invocation = m.invocation
+                for d in data:
+                    if d['in'] == invocation:
+                        ms = d['m']
+                        break
+                if ms is None:
+                    ms = []
+                    data.append({'in': invocation, 'm': ms})
+                    for _ in criteria:
+                        ms.append([])
+            else:
+                assert iteration == m.iteration
+                assert invocation == m.invocation
+                assert ms is not None
+            criterion = (m.criterion, m.unit)
+            if criterion not in criteria:
+                criteria[criterion] = len(criteria)
+
+            c_idx = criteria[criterion]
+            if len(ms) <= c_idx:
+                ms.append([])
+
+            while len(ms[c_idx]) + 1 < iteration:
+                ms[c_idx].append(None)
+            ms[c_idx].append(m.value)
+            num_measurements += 1
+
+        assert self.invocation == invocation
+        return num_measurements
+
     def __repr__(self):
         return "DataPoint(" + str(self.run_id) + ", " + str(self._measurements) + ")"

rebench/persistence.py

@@ -432,12 +432,11 @@ def _send_data_and_empty_cache(self):
             if self._send_data(self._cache):
                 self._cache = {}
 
-    def _send_data(self, cache):
-        self.ui.debug_output_info("ReBenchDB: Prepare data for sending\n")
+    def convert_data_to_api_format(self, data):
         num_measurements = 0
         all_data = []
         criteria = {}
-        for run_id, data_points in cache.items():
+        for run_id, data_points in data.items():
             dp_data = []
             for dp in data_points:
                 measurements = dp.measurements_as_dict(criteria)
@@ -452,6 +451,34 @@ def _send_data(self, cache):
         for c, idx in criteria.items():
             criteria_index.append({'c': c[0], 'u': c[1], 'i': idx})
 
+        return all_data, criteria_index, num_measurements
+
+    def convert_data_to_api_20_format(self, data):
+        num_measurements = 0
+        all_data = []
+        criteria = {}
+        for run_id, data_points in data.items():
+            dp_data = []
+            for dp in data_points:
+                num_measurements += dp.add_measurements_api_v20(criteria, dp_data)
+            all_data.append({
+                'runId': run_id.as_dict(),
+                'd': dp_data
+            })
+
+        criteria_index = []
+        for c, idx in criteria.items():
+            criteria_index.append({'c': c[0], 'u': c[1], 'i': idx})
+
+        return all_data, criteria_index, num_measurements
+
+    def _send_data(self, cache):
+        self.ui.debug_output_info("ReBenchDB: Prepare data for sending\n")
+        if self._rebench_db.is_api_v2():
+            all_data, criteria_index, num_measurements = self.convert_data_to_api_20_format(cache)
+        else:
+            all_data, criteria_index, num_measurements = self.convert_data_to_api_format(cache)
+
         self.ui.debug_output_info(
             "ReBenchDB: Sending {num_m} measures. startTime: {st}\n",
             num_m=num_measurements, st=self._start_time)

rebench/rebenchdb.py

@@ -3,7 +3,7 @@
 from time import sleep
 
 from http.client import HTTPException
-from urllib.request import urlopen, Request as PutRequest
+from urllib.request import urlopen, Request as HttpRequest
 
 from .ui import UIError
 
@@ -33,6 +33,20 @@ def __init__(self, server_base_url, project_name, experiment_name, ui):
         self._server_base_url = server_base_url
         self._project_name = project_name
         self._experiment_name = experiment_name
+        self._api_v2 = None
+
+    def is_api_v2(self):
+        if self._api_v2 is None:
+            api_version = self._get_api_version()
+            if api_version:
+                major = api_version.split('.')[0]
+                if int(major) == 2:
+                    self._api_v2 = True
+
+            if self._api_v2 is None:
+                self._api_v2 = False
+
+        return self._api_v2
 
     def send_results(self, benchmark_data, num_items):
         success, response = self._send_to_rebench_db(benchmark_data, '/results')
@@ -60,18 +74,32 @@ def send_completion(self, end_time):
 
     @staticmethod
     def _send_payload(payload, url):
-        req = PutRequest(url, payload,
+        req = HttpRequest(url, payload,
                          {'Content-Type': 'application/json'}, method='PUT')
         with urlopen(req) as socket:
             response = socket.read()
             return response
 
+    def _get_api_version(self):
+        url = self._server_base_url + '/results'
+        req = HttpRequest(url, method='OPTIONS')
+        try:
+            with urlopen(req) as socket:
+                response = socket.read()
+                return socket.getheader('X-ReBenchDB-Result-API-Version')
+        except:
+            # some error, so no API version available
+            return None
+
+    def convert_data_to_json(self, data):
+        return json.dumps(data, separators=(',', ':'), ensure_ascii=True)
+
     def _send_to_rebench_db(self, payload_data, operation):
         payload_data['projectName'] = self._project_name
         payload_data['experimentName'] = self._experiment_name
         url = self._server_base_url + operation
 
-        payload = json.dumps(payload_data, separators=(',', ':'), ensure_ascii=True)
+        payload = self.convert_data_to_json(payload_data)
 
         # self.ui.output("Saving JSON Payload of size: %d\n" % len(payload))
         with open("payload.json", "w") as text_file:  # pylint: disable=unspecified-encoding
@@ -92,10 +120,11 @@ def _send_with_retries(self, payload_bytes, url):
                                + "{ind}{ind}" + str(te) + "\n")
                 return False, None
             except (IOError, HTTPException) as error:
-                if attempts > 0:
+                is_client_error = hasattr(error, 'status') and 400 <= error.status < 500
+                if not is_client_error and attempts > 0:
                     # let's retry, the benchmark server might just time out, as usual
                     # but let it breath a little
-                    self.ui.verbose_output_info(
+                    self.ui.warning(
                         "ReBenchDB: had issue reporting data. Trying again after "
                         + str(wait_sec) + "seconds.\n"
                         + "{ind}{ind}" + str(error) + "\n")

rebench/tests/mock_http_server.py

@@ -7,16 +7,26 @@
 
 class _RequestHandler(BaseHTTPRequestHandler):
     def do_GET(self):
-        self.send_response(200)
+        self.send_response(self.server.status_code)
         self.end_headers()
         self.send_header("Content-Length", 0)
+        self.server.get_requests += 1
 
     def do_PUT(self):
-        self.send_response(200)
+        self.send_response(self.server.status_code)
         self.send_header("Content-Length", 0)
         self.end_headers()
         self.server.put_requests += 1
 
+    def do_OPTIONS(self):
+        self.send_response(self.server.status_code)
+        if self.server.api_v2:
+            self.send_header("X-ReBenchDB-Result-API-Version", "2.0.0")
+        self.send_header("Allow", "PUT")
+        self.send_header("Content-Length", 0)
+        self.end_headers()
+        self.server.options_requests += 1
+
     def log_request(self, code='-', size='-'):
         pass
 
@@ -25,15 +35,21 @@ class HTTPServerWithCounter(HTTPServer):
     def __init__(self, *args, **kwargs):
         super(HTTPServerWithCounter, self).__init__(*args, **kwargs)
         self.put_requests = 0
+        self.get_requests = 0
+        self.options_requests = 0
+        self.api_v2 = None
+        self.status_code = 200
 
 
 class MockHTTPServer(object):
 
-    def __init__(self):
+    def __init__(self, api_v2 = True, test_error_handling = False):
         self._port = -1
         self._server = None
         self._thread = None
         self._is_shutdown = False
+        self.api_v2 = api_v2
+        self._test_error_handling = test_error_handling
 
     def get_free_port(self):
         s = socket.socket(socket.AF_INET, type=socket.SOCK_STREAM)
@@ -46,6 +62,9 @@ def get_free_port(self):
 
     def start(self):
         self._server = HTTPServerWithCounter(('localhost', self._port), _RequestHandler)
+        self._server.api_v2 = self.api_v2
+        if self._test_error_handling:
+            self._server.status_code = 400
 
         self._thread = Thread(target=self._server.serve_forever)
         self._thread.daemon = True