Script

attention.md

@@ -0,0 +1,6 @@
+Attention
+
+1.	Client/settings.py line6 用gitee git
+2.	所有的tmp改成了raid
+3.	Client/settings.py line38 duration 600改成1
+

client/.gitignore

@@ -1,2 +1,4 @@
 .lock
 settings_local.py
+settings.py
+

client/benchmarks/pgbench.py

@@ -7,6 +7,9 @@
 from numpy import mean, median, std
 
 from multiprocessing import cpu_count
+
+from collectors.scripts import ScriptCollector
+from settings import SCRIPTS_DIR
 from utils.logging import log
 from utils.misc import available_ram, run_cmd
 
@@ -17,9 +20,8 @@ class PgBench(object):
     # TODO allow running custom scripts, not just the default
     #      read-write/read-only tests
     # TODO allow running 'prepared' mode
-
     def __init__(self, bin_path, dbname, runs=3, duration=60, csv=False,
-                 results_dir=None):
+                 results_dir=None, benchmark_options=None):
         '''
         bin_path   - path to PostgreSQL binaries (dropdb, createdb, psql
                      commands)
@@ -40,6 +42,7 @@ def __init__(self, bin_path, dbname, runs=3, duration=60, csv=False,
         self._env['PATH'] = ':'.join([bin_path, self._env['PATH']])
 
         self._results = {}
+        self._benchmark_options = benchmark_options
 
     @staticmethod
     def _configure(cpu_count, ram_mbs):
@@ -76,10 +79,11 @@ def _init(self, scale):
         log("initializing pgbench '%s' with scale %s" % (self._dbname, scale))
         r = run_cmd(['pgbench', '-i', '-s', str(scale), self._dbname],
                     env=self._env, cwd=self._outdir)
-
         # remember the init duration
         self._results['results']['init'] = r[2]
 
+
+
     @staticmethod
     def _parse_results(data):
         'extract results (including parameters) from the pgbench output'
@@ -170,7 +174,8 @@ def _run(self, run, scale, duration, nclients=1, njobs=1, read_only=False,
             rtag = "rw"
         rdir = "%s/pgbench-%s-%d-%d-%s" % (self._outdir, rtag, scale, nclients,
                                            str(run))
-        os.mkdir(rdir)
+        if not(os.path.exists(rdir)):
+         os.mkdir(rdir)
 
         args = ['pgbench', '-c', str(nclients), '-j', str(njobs), '-T',
                 str(duration)]
@@ -205,16 +210,22 @@ def _run(self, run, scale, duration, nclients=1, njobs=1, read_only=False,
 
         return r
 
-    def run_tests(self, csv_queue):
+    def run_tests(self, csv_queue, benchmark_options=None):
         """
         execute the whole benchmark, including initialization, warmup and
         benchmark runs
         """
 
         # derive configuration for the CPU count / RAM size
-        configs = PgBench._configure(cpu_count(), available_ram())
+        # if there is benchmark setting in settings.py, take it and use,
+        # otherwise use the default one
+        configs = []
+        if benchmark_options:
+            configs.append(benchmark_options)
+        else:
+            configs = PgBench._configure(cpu_count(), available_ram())
 
-        results = {'ro': {}, 'rw': {}}
+        results = {'ro': {}, 'rw': {}, 'customeScript': {}}
         j = 0
         for config in configs:
             scale = config['scale']
@@ -223,6 +234,9 @@ def run_tests(self, csv_queue):
                 results['ro'][scale] = {}
             if scale not in results['rw']:
                 results['rw'][scale] = {}
+            # if scale not in results['customeScript']:
+                # results['customeScript'][scale] = {}
+
 
             # init for the dataset scale and warmup
             self._init(scale)
@@ -257,6 +271,33 @@ def run_tests(self, csv_queue):
                         results[tag][scale][clients]['metric'] = mean(tps)
                         results[tag][scale][clients]['median'] = median(tps)
                         results[tag][scale][clients]['std'] = std(tps)
+        # todo add cmmand
+        # args = ['pgbench', '-c', str(nclients), '-j', str(njobs), '-T',
+        #         str(duration)]
+
+        script = ScriptCollector(scriptdirpath=SCRIPTS_DIR,env=self._env, dbname=self._dbname)
+        if script.hasScript():
+            start = time.time()
+            scriptResult =script.run_custem_script()
+
+            print('scriptResult   ')
+            print(scriptResult)
+            end = time.time()
+            r = PgBench._parse_results(scriptResult[1])
+            # r.update({'customeScript': read_only})
+            # results[tag][scale][clients]['results'].append(r)
+            r.update({'start': start, 'end': end})
+
+            results['customeScript']['results']=[]
+            results['customeScript']['results'].append(r)
+            tps = []
+            for result in results['customeScript']['results']:
+                tps.append(float(result['tps']))
+            results['customeScript']['metric'] = mean(tps)
+            results['customeScript']['median'] = median(tps)
+            results['customeScript']['std'] = std(tps)
+
+            results['customeScript']['scriptList'] = script.getScriptListJson()
 
         self._results['pgbench'] = results
         return self._results

client/benchmarks/runner.py

@@ -79,6 +79,7 @@ def _run_config(self, config_name):
         # construct the benchmark class for the given config name
         config = self._configs[config_name]
         bench = self._benchmarks[config['benchmark']]
+        benchmark_options = config['config']['benchmark_options']
 
         # expand the attribute names
         bench = bench(**config['config'])
@@ -91,14 +92,18 @@ def _run_config(self, config_name):
         # if requested output to CSV, create a queue and collector process
         csv_queue = None
         csv_collector = None
+        # somehow get the benchmarking options in settings.py
+        if benchmark_options:
+            options = benchmark_options
+
         if 'csv' in config['config'] and config['config']['csv']:
             csv_queue = Queue()
             csv_collector = Process(target=csv_collect_results,
                                     args=(config_name, csv_queue))
             csv_collector.start()
 
         # run the tests
-        r = bench.run_tests(csv_queue)
+        r = bench.run_tests(csv_queue, options)
 
         # notify the result collector to end and wait for it to terminate
         if csv_queue:
@@ -116,16 +121,16 @@ def _run_config(self, config_name):
         uname = check_output(['uname', '-a'])
 
         r['meta'] = {
-                'benchmark': config['benchmark'],
-                'date': strftime("%Y-%m-%d %H:%M:%S.000000+00", gmtime()),
-                'name': config_name,
-                'uname': uname,
+            'benchmark': config['benchmark'],
+            'date': strftime("%Y-%m-%d %H:%M:%S.000000+00", gmtime()),
+            'name': config_name,
+            'uname': uname,
         }
 
         r['postgres'] = {
-                'branch': config['branch'],
-                'commit': config['commit'],
-                'settings': config['postgres'],
+            'branch': config['branch'],
+            'commit': config['commit'],
+            'settings': config['postgres'],
         }
 
         with open('%s/results.json' % self._output, 'w') as f:
@@ -134,23 +139,25 @@ def _run_config(self, config_name):
         try:
             self._upload_results(r)
         except Exception as e:
-            print (e)
+            print(e)
 
     def _upload_results(self, results):
 
-        postdata = results  
-        post = []  
-        post.append(postdata)
-
-        headers = {'Content-Type': 'application/json; charset=utf-8', 'Authorization': self._secret}
-        r = requests.post(self._url.encode('utf-8'), data=json.dumps(post).encode('utf-8'), headers=headers)
-
+        PATH_URL = 'upload/'
+        url = self._url + PATH_URL
+        headers = {'Authorization': self._secret}
+        files = {
+            'json': (None, json.dumps(results), 'application/json'),
+            'insert.sql': open('scripts/files/insert.sql', 'rb'),
+            'test.sql': open('scripts/files/test.sql', 'rb')
+        }
+        r = requests.post(url.encode('utf-8'), files=files, headers=headers)
 
     def run(self):
         'run all the configured benchmarks'
 
         # Removing the existing directory
-        
+
         try:
             os.mkdir(self._output)
         except OSError as e:
@@ -183,4 +190,4 @@ def csv_collect_results(bench_name, queue):
 
             # otherwise we expect the value to be a list, and we just print it
             results_file.write(bench_name + "\t" + "\t".join(v) + "\n")
-            results_file.flush()
+            results_file.flush()

client/benchmarks/test_upload_json.py

@@ -0,0 +1,13 @@
+import json
+import requests
+
+url = 'http://127.0.0.1:8000/upload/'
+secret = 'e984c3017cd1a0dff0ef9f0c394a5c285e421411'
+headers = {'Content-Type': 'application/json; charset=utf-8', 'Authorization': secret}
+with open("/home/guo/PythonClass/results_need.json", 'r') as load_f:
+    load_data = json.load(load_f)
+    postdata = load_data
+    post = []
+    post.append(postdata)
+
+r = requests.post(url.encode('utf-8'), data=json.dumps(post).encode('utf-8'), headers=headers)

client/benchmarks/test_upload_json_script.py

@@ -0,0 +1,39 @@
+import json
+import requests
+import os
+
+url = 'http://127.0.0.1:8000/upload/'
+secret = 'e984c3017cd1a0dff0ef9f0c394a5c285e421411'
+headers = {'Authorization': secret}
+# headers = {'Content-Type': 'application/json; charset=utf-8', 'Authorization': secret}
+with open("/home/guo/PythonClass/results_script1.json", 'r') as load_f:
+    load_data = json.load(load_f)
+    print(load_data)
+    # postdata = load_data
+    # post = []
+    # post.append(postdata)
+
+# files = {
+#     'json': (None, json.dumps(load_data), 'application/json'),
+#     'file': [
+#         {(os.path.basename('insert.sql'), open('../scripts/files/insert.sql', 'rb'), 'application/octet-stream')},
+#         {(os.path.basename('test.sql'), open('../scripts/files/test.sql', 'rb'), 'application/octet-stream')}]
+# }
+
+files = {
+    'json': (None, json.dumps(load_data), 'application/json'),
+    'insert.sql': open('../scripts/files/insert.sql', 'rb'),
+    'test.sql': open('../scripts/files/test.sql', 'rb')
+}
+# files = {
+#      'json': (None, json.dumps(load_data), 'application/json'),
+#      'file': (os.path.basename('insert.sql'), open('../scripts/files/insert.sql', 'rb'), 'application/octet-stream')
+# }
+# files = {'insert.sql': open('../scripts/files/insert.sql', 'rb'),
+#          'test.sql': open('../scripts/files/test.sql', 'rb')}
+# r = requests.post(url.encode('utf-8'), data=json.dumps(post).encode('utf-8'), files=files, headers=headers)
+# r = requests.post(url.encode('utf-8'), data=load_data, files=files, headers=headers)
+# r = requests.post(url.encode('utf-8'), data=(None, json.dumps(load_data), 'application/json'), files=files, headers=headers)
+# r = requests.post(url.encode('utf-8'), data=json.dumps(load_data), files=files, headers=headers)
+
+r = requests.post(url.encode('utf-8'), files=files, headers=headers)

client/collectors/collectd.py

@@ -3,8 +3,8 @@
 from utils.logging import log
 from utils.misc import run_cmd
 
-COLLECTD_CONFIG = '/tmp/.collectd.conf'
-COLLECTD_PIDFILE = '/tmp/.collectd.pid'
+COLLECTD_CONFIG = '/raid/.collectd.conf'
+COLLECTD_PIDFILE = '/raid/.collectd.pid'
 
 
 class CollectdCollector(object):

client/collectors/postgres.py

@@ -6,6 +6,8 @@
 import time
 
 from multiprocessing import Process, Queue
+
+from settings_local import DATABASES
 from utils.logging import log
 from utils.misc import run_cmd
 
@@ -24,7 +26,7 @@ def __init__(self, outdir, dbname, bin_path):
     def start(self):
         log("saving postgres settings")
         try:
-            conn = psycopg2.connect('host=localhost dbname=%s' % self._dbname)
+            conn = psycopg2.connect('dbname={} user={} password={} host=localhost'.format(DATABASES['default']['NAME'], DATABASES['default']['USER'],DATABASES['default']['PASSWORD']))
             cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
             cur.execute(
                 'SELECT name, setting, source '

client/collectors/scripts.py

@@ -0,0 +1,59 @@
+import glob
+import os.path
+from utils.misc import run_cmd
+import json
+
+
+class ScriptCollector(object):
+    'Collect various Linux-specific statistics (cpuinfo, mounts)'
+
+    def __init__(self, scriptdirpath, env, dbname):
+        self._dirPath = scriptdirpath
+        self._env = env
+        self._dbname=dbname
+        self._scriptFileList = []
+        self.hasScript()
+
+
+
+    def hasScript(self):
+        isHasScript =False
+        if os.path.exists(self._dirPath):
+            self._scriptFileList = glob.glob(self._dirPath + "/*.sql")
+            if(len(self._scriptFileList) > 0):
+                isHasScript = True
+        return  isHasScript
+
+
+    def _collect_scriptDir_info(self):
+
+
+        for script in self.sqlScript:
+            scriptObj = {}
+            scriptObj["scriptName"] = os.path.basename(script)
+
+
+        return scriptObj
+
+    def run_custem_script(self):
+        args = ['pgbench']
+        for script in self._scriptFileList:
+            print(script)
+            args.append('-f')
+            path = os.path.abspath(script)
+            args.append(path)
+        args.append(self._dbname)
+
+        r = run_cmd(args,
+                    env=self._env)
+        return r
+
+    def getScriptListJson(self):
+        scriptList = []
+        for script in self._scriptFileList:
+            scriptObj = {}
+            scriptObj["scriptName"] = os.path.basename(script)
+            scriptList.append(scriptObj)
+        result = json.dumps(scriptList)
+        print("collecting self-design script json: " + result)
+        return scriptList