hammerd wrapper fixes for some loops

hammerdb/Dockerfile

@@ -9,6 +9,7 @@ RUN ACCEPT_EULA=Y dnf -y install --nodocs msodbcsql17 --enablerepo=centos8
 RUN dnf clean all
 RUN pip3 install --upgrade-strategy=only-if-needed "elasticsearch>=6.0.0,<=7.0.2" pyyaml
 COPY . /opt/snafu
+RUN ln -s /usr/bin/python3 /usr/bin/python
 
 # Download and install the hammer suite
 RUN curl -L 'https://downloads.sourceforge.net/project/hammerdb/HammerDB/HammerDB-3.2/HammerDB-3.2-Linux-x86-64-Install?r=https%3A%2F%2Fsourceforge.net%2Fprojects%2Fhammerdb%2Ffiles%2FHammerDB%2FHammerDB-3.2%2FHammerDB-3.2-Linux-x86-64-Install%2Fdownload&ts=1564587940&use_mirror=autoselect' -o hammer_installer

hammerdb/hammerd_wrapper.py

@@ -1,96 +1,106 @@
 #!/usr/bin/env python
 
-import argparse
 import sys
 import subprocess
 import os
-import re
 import elasticsearch
-import time 
-from datetime import datetime
+import time
+
 
 def _run_hammerdb():
     cmd = "cd /hammer; ./hammerdbcli auto /workload/tpcc-workload.tcl"
     process = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE)
-    stdout,stderr = process.communicate()
+    stdout, stderr = process.communicate()
     return stdout.strip().decode("utf-8"), process.returncode
 
+
 def _fake_run():
     with open("hammerdb.log", "r") as input:
         stdout = input.read()
-    return stdout,0
+    return stdout, 0
+
 
 def _parse_stdout(stdout):
     data = []
-    iteration = 0
     for line in stdout.splitlines():
         if "TEST RESULT" in line:
             worker = (line.split(":"))[0]
             tpm = (line.split(" "))[6]
             nopm = (line.split(" "))[-2]
-            entry = [ worker, tpm, nopm ]
+            entry = [worker, tpm, nopm]
             data.append(entry)
     return data
 
+
 def _json_payload(data, uuid, db_server, db_port, db_warehouses, db_num_workers, db_tcp, db_user, transactions, test_type, runtime, rampup, samples, timed_test, timestamp):
     processed = []
-    for i in range(0,len(data)):
-        processed.append({
-            "workload" : "hammerdb",
-            "uuid" : uuid,
-            "db_server" : db_server,
-            "db_port" : db_port,
-            "db_warehouses" : db_warehouses,
-            "db_num_workers" : db_num_workers,
-            "db_tcp": db_tcp,
-            "db_user": db_user,
-            "transactions": transactions,
-            "test_type": test_type,
-            "runtime": runtime,
-            "rampup": rampup,
-            "samples": samples,
-            "timed_test": timed_test,
-            "worker": data[i][0],
-            "tpm": data[i][1],
-            "nopm": data[i][2],
-            "timestamp": timestamp
-            })
+    for current_worker in range(0, int(db_num_workers)):
+        for current_sample in range(0, int(samples)):
+            for i in range(0, len(data)):
+                processed.append({
+                    "workload": "hammerdb",
+                    "uuid": uuid,
+                    "db_server": db_server,
+                    "db_port": db_port,
+                    "db_warehouses": db_warehouses,
+                    "db_num_workers": db_num_workers,
+                    "db_tcp": db_tcp,
+                    "db_user": db_user,
+                    "transactions": transactions,
+                    "test_type": test_type,
+                    "runtime": runtime,
+                    "rampup": rampup,
+                    "samples": samples,
+                    "current_sample": current_sample,
+                    "current_worker": current_worker,
+                    "timed_test": timed_test,
+                    "worker": data[i][0],
+                    "tpm": data[i][1],
+                    "nopm": data[i][2],
+                    "timestamp": timestamp
+                })
     return processed
 
+
 def _summarize_data(data):
-    for i in range(0,len(data)):
-        entry = data[i]
-
-        print("+{} HammerDB Results {}+".format("-"*(50), "-"*(50)))
-        print("HammerDB setup")
-        print("")
-        print("HammerDB results for:")
-        print("UUID: {}".format(entry['uuid']))
-        print("Database server: {}".format(entry['db_server']))
-        print("Database port: {}".format(entry['db_port']))
-        print("Number of database warehouses: {}".format(entry['db_warehouses']))
-        print("Number of workers: {}".format(entry['db_num_workers']))
-        print("TCP connection to the DB: {}".format(entry['db_tcp']))
-        print("Database user: {}".format(entry['db_user']))
-        print("Transactions: {}".format(entry['transactions']))
-        print("Test type: {}".format(entry['test_type']))
-        print("Runtime: {}".format(entry['runtime']))
-        print("Rampup time: {}".format(entry['rampup']))
-        print("Worker: {}".format(entry['worker']))
-        print("Samples: {}".format(entry['samples']))
-        print("Timed test: {}".format(entry['timed_test']))
-        print("HammerDB results (TPM):")
-        print("""
-              TPM: {}""".format(entry['tpm']))
-        print("HammerDB results (NOPM):")
-        print("""
-              NOPM: {}""".format(entry['nopm']))
-        print("Timestamp: {}".format(entry['timestamp']))
-        print("+{}+".format("-"*(115)))
-
-def _index_result(index,es_server,es_port,payload):
+    max_workers = int(data[0]['db_num_workers'])
+    max_samples = int(data[0]['samples'])
+    for current_worker in range(0, max_workers):
+        for current_sample in range(0, max_samples):
+            for i in range(0, len(data)):
+                entry = data[i]
+                print("+{} HammerDB Results {}+".format("-"*(50), "-"*(50)))
+                print("HammerDB setup")
+                print("")
+                print("HammerDB results for:")
+                print("UUID: {}".format(entry['uuid']))
+                print("Database server: {}".format(entry['db_server']))
+                print("Database port: {}".format(entry['db_port']))
+                print("Number of database warehouses: {}".format(entry['db_warehouses']))
+                print("Number of workers: {}".format(entry['db_num_workers']))
+                print("TCP connection to the DB: {}".format(entry['db_tcp']))
+                print("Database user: {}".format(entry['db_user']))
+                print("Transactions: {}".format(entry['transactions']))
+                print("Test type: {}".format(entry['test_type']))
+                print("Runtime: {}".format(entry['runtime']))
+                print("Rampup time: {}".format(entry['rampup']))
+                print("Worker: {}".format(current_worker))
+                print("Samples: {}".format(entry['samples']))
+                print("Current sample {}".format(current_sample))
+                print("Timed test: {}".format(entry['timed_test']))
+                print("HammerDB results (TPM):")
+                print("""
+                      TPM: {}""".format(entry['tpm']))
+                print("HammerDB results (NOPM):")
+                print("""
+                      NOPM: {}""".format(entry['nopm']))
+                print("Timestamp: {}".format(entry['timestamp']))
+                print("+{}+".format("-"*(115)))
+
+
+def _index_result(index, es_server, es_port, payload):
     _es_connection_string = str(es_server) + ':' + str(es_port)
-    es = elasticsearch.Elasticsearch([_es_connection_string],send_get_body_as='POST')
+    es = elasticsearch.Elasticsearch([_es_connection_string], send_get_body_as='POST')
     indexed = True
     processed_count = 0
     total_count = 0
@@ -99,17 +109,16 @@ def _index_result(index,es_server,es_port,payload):
             es.index(index=index, body=result)
             processed_count += 1
         except Exception as e:
-            print (repr(e) + "occured for the json document:")
+            print(repr(e) + "occured for the json document:")
             print(str(result))
             indexed = False
         total_count += 1
     return indexed, processed_count, total_count
 
-def main():
 
+def main():
     es_server = ""
     es_port = ""
-    protocol = ""
     uuid = ""
     db_user = ""
     db_server = ""
@@ -120,9 +129,10 @@ def main():
     runtime = ""
     rampup = ""
     samples = ""
-    iteration = "" 
     test_type = ""
     timestamp = ""
+    db_tcp = ""
+    timed_test = ""
 
     if "es_server" in os.environ:
         es_server = os.environ["es_server"]
@@ -167,11 +177,14 @@ def main():
             exit(1)
     data = _parse_stdout(stdout[0])
     documents = _json_payload(data, uuid, db_server, db_port, db_warehouses, db_num_workers, db_tcp, db_user, transactions, test_type, runtime, rampup, samples, timed_test, timestamp)
+    #print(documents)
+    if len(documents) > 0 :
+        _summarize_data(documents)
     if es_server != "" :
         if len(documents) > 0 :
             _index_result("ripsaw-hammerdb-results", es_server, es_port, documents)
-    if len(documents) > 0 :
-        _summarize_data(documents)
+        else: 
+            raise Exception('Failed to produce hammerdb results document')
 
 
 if __name__ == '__main__':