framework_test.py 36 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882
  1. from benchmark.fortune_html_parser import FortuneHTMLParser
  2. from setup.linux import setup_util
  3. from benchmark.test_types import *
  4. import importlib
  5. import os
  6. import subprocess
  7. import time
  8. import re
  9. from pprint import pprint
  10. import sys
  11. import traceback
  12. import json
  13. import logging
  14. import csv
  15. import shlex
  16. import math
  17. from collections import OrderedDict
  18. from threading import Thread
  19. from threading import Event
  20. from utils import header
  21. from datetime import datetime
  22. from datetime import timedelta
  23. from threading import Thread
  24. from Queue import Queue, Empty
  25. class NonBlockingStreamReader:
  26. def __init__(self, stream):
  27. '''
  28. stream: the stream to read from.
  29. Usually a process' stdout or stderr.
  30. '''
  31. self._s = stream
  32. self._q = Queue()
  33. def _populateQueue(stream, queue):
  34. '''
  35. Collect lines from 'stream' and put them in 'quque'.
  36. '''
  37. while True:
  38. line = stream.readline()
  39. if line:
  40. queue.put(line)
  41. else:
  42. raise UnexpectedEndOfStream
  43. self._t = Thread(target = _populateQueue,
  44. args = (self._s, self._q))
  45. self._t.daemon = True
  46. self._t.start() #start collecting lines from the stream
  47. def readline(self, timeout = None):
  48. try:
  49. return self._q.get(block = timeout is not None,
  50. timeout = timeout)
  51. except Empty:
  52. return None
  53. class UnexpectedEndOfStream(Exception): pass
  54. class FrameworkTest:
  55. headers_template = "-H 'Host: localhost' -H '{accept}' -H 'Connection: keep-alive'"
  56. # Used for test types that require no pipelining or query string params.
  57. concurrency_template = """
  58. echo ""
  59. echo "---------------------------------------------------------"
  60. echo " Running Primer {name}"
  61. echo " {wrk} {headers} -d 5 -c 8 --timeout 8 -t 8 \"http://{server_host}:{port}{url}\""
  62. echo "---------------------------------------------------------"
  63. echo ""
  64. {wrk} {headers} -d 5 -c 8 --timeout 8 -t 8 "http://{server_host}:{port}{url}"
  65. sleep 5
  66. echo ""
  67. echo "---------------------------------------------------------"
  68. echo " Running Warmup {name}"
  69. echo " {wrk} {headers} -d {duration} -c {max_concurrency} --timeout {max_concurrency} -t {max_threads} \"http://{server_host}:{port}{url}\""
  70. echo "---------------------------------------------------------"
  71. echo ""
  72. {wrk} {headers} -d {duration} -c {max_concurrency} --timeout {max_concurrency} -t {max_threads} "http://{server_host}:{port}{url}"
  73. sleep 5
  74. echo ""
  75. echo "---------------------------------------------------------"
  76. echo " Synchronizing time"
  77. echo "---------------------------------------------------------"
  78. echo ""
  79. ntpdate -s pool.ntp.org
  80. for c in {levels}
  81. do
  82. echo ""
  83. echo "---------------------------------------------------------"
  84. echo " Concurrency: $c for {name}"
  85. echo " {wrk} {headers} -d {duration} -c $c --timeout $c -t $(($c>{max_threads}?{max_threads}:$c)) \"http://{server_host}:{port}{url}\""
  86. echo "---------------------------------------------------------"
  87. echo ""
  88. STARTTIME=$(date +"%s")
  89. {wrk} {headers} -d {duration} -c $c --timeout $c -t "$(($c>{max_threads}?{max_threads}:$c))" http://{server_host}:{port}{url}
  90. echo "STARTTIME $STARTTIME"
  91. echo "ENDTIME $(date +"%s")"
  92. sleep 2
  93. done
  94. """
  95. # Used for test types that require pipelining.
  96. pipeline_template = """
  97. echo ""
  98. echo "---------------------------------------------------------"
  99. echo " Running Primer {name}"
  100. echo " {wrk} {headers} -d 5 -c 8 --timeout 8 -t 8 \"http://{server_host}:{port}{url}\""
  101. echo "---------------------------------------------------------"
  102. echo ""
  103. {wrk} {headers} -d 5 -c 8 --timeout 8 -t 8 "http://{server_host}:{port}{url}"
  104. sleep 5
  105. echo ""
  106. echo "---------------------------------------------------------"
  107. echo " Running Warmup {name}"
  108. echo " {wrk} {headers} -d {duration} -c {max_concurrency} --timeout {max_concurrency} -t {max_threads} \"http://{server_host}:{port}{url}\""
  109. echo "---------------------------------------------------------"
  110. echo ""
  111. {wrk} {headers} -d {duration} -c {max_concurrency} --timeout {max_concurrency} -t {max_threads} "http://{server_host}:{port}{url}"
  112. sleep 5
  113. echo ""
  114. echo "---------------------------------------------------------"
  115. echo " Synchronizing time"
  116. echo "---------------------------------------------------------"
  117. echo ""
  118. ntpdate -s pool.ntp.org
  119. for c in {levels}
  120. do
  121. echo ""
  122. echo "---------------------------------------------------------"
  123. echo " Concurrency: $c for {name}"
  124. echo " {wrk} {headers} -d {duration} -c $c --timeout $c -t $(($c>{max_threads}?{max_threads}:$c)) \"http://{server_host}:{port}{url}\" -s ~/pipeline.lua -- {pipeline}"
  125. echo "---------------------------------------------------------"
  126. echo ""
  127. STARTTIME=$(date +"%s")
  128. {wrk} {headers} -d {duration} -c $c --timeout $c -t "$(($c>{max_threads}?{max_threads}:$c))" http://{server_host}:{port}{url} -s ~/pipeline.lua -- {pipeline}
  129. echo "STARTTIME $STARTTIME"
  130. echo "ENDTIME $(date +"%s")"
  131. sleep 2
  132. done
  133. """
  134. # Used for test types that require a database -
  135. # These tests run at a static concurrency level and vary the size of
  136. # the query sent with each request
  137. query_template = """
  138. echo ""
  139. echo "---------------------------------------------------------"
  140. echo " Running Primer {name}"
  141. echo " wrk {headers} -d 5 -c 8 --timeout 8 -t 8 \"http://{server_host}:{port}{url}2\""
  142. echo "---------------------------------------------------------"
  143. echo ""
  144. wrk {headers} -d 5 -c 8 --timeout 8 -t 8 "http://{server_host}:{port}{url}2"
  145. sleep 5
  146. echo ""
  147. echo "---------------------------------------------------------"
  148. echo " Running Warmup {name}"
  149. echo " wrk {headers} -d {duration} -c {max_concurrency} --timeout {max_concurrency} -t {max_threads} \"http://{server_host}:{port}{url}2\""
  150. echo "---------------------------------------------------------"
  151. echo ""
  152. wrk {headers} -d {duration} -c {max_concurrency} --timeout {max_concurrency} -t {max_threads} "http://{server_host}:{port}{url}2"
  153. sleep 5
  154. echo ""
  155. echo "---------------------------------------------------------"
  156. echo " Synchronizing time"
  157. echo "---------------------------------------------------------"
  158. echo ""
  159. ntpdate -s pool.ntp.org
  160. for c in {levels}
  161. do
  162. echo ""
  163. echo "---------------------------------------------------------"
  164. echo " Queries: $c for {name}"
  165. echo " wrk {headers} -d {duration} -c {max_concurrency} --timeout {max_concurrency} -t {max_threads} \"http://{server_host}:{port}{url}$c\""
  166. echo "---------------------------------------------------------"
  167. echo ""
  168. STARTTIME=$(date +"%s")
  169. wrk {headers} -d {duration} -c {max_concurrency} --timeout {max_concurrency} -t {max_threads} "http://{server_host}:{port}{url}$c"
  170. echo "STARTTIME $STARTTIME"
  171. echo "ENDTIME $(date +"%s")"
  172. sleep 2
  173. done
  174. """
  175. ############################################################
  176. # start(benchmarker)
  177. # Start the test using it's setup file
  178. ############################################################
  179. def start(self, out, err):
  180. # Load profile for this installation
  181. profile="$FWROOT/config/benchmark_profile"
  182. # Setup variables for TROOT and IROOT
  183. setup_util.replace_environ(config=profile,
  184. command='export TROOT=%s && export IROOT=%s && export DBHOST=%s && export MAX_THREADS=%s && export OUT=%s && export ERR=%s' %
  185. (self.directory, self.install_root, self.database_host, self.benchmarker.threads, os.path.join(self.fwroot, out.name), os.path.join(self.fwroot, err.name)))
  186. # Run the module start (inside parent of TROOT)
  187. # - we use the parent as a historical accident - a lot of tests
  188. # use subprocess's cwd argument already
  189. previousDir = os.getcwd()
  190. os.chdir(os.path.dirname(self.troot))
  191. logging.info("Running setup module start (cwd=%s)", self.directory)
  192. # Run the start script for the test as the "testrunner" user.
  193. # This requires superuser privs, so `sudo` is necessary.
  194. # -u [username] The username
  195. # -E Preserves the current environment variables
  196. # -H Forces the home var (~) to be reset to the user specified
  197. # -e Force bash to exit on first error
  198. # -x Turn on bash tracing e.g. print commands before running
  199. # Note: check_call is a blocking call, so any startup scripts
  200. # run by the framework that need to continue (read: server has
  201. # started and needs to remain that way), then they should be
  202. # executed in the background.
  203. command = 'sudo -u %s -E -H bash -ex %s.sh' % (self.benchmarker.runner_user, self.setup_file)
  204. debug_command = '''\
  205. export FWROOT=%s && \\
  206. export TROOT=%s && \\
  207. export IROOT=%s && \\
  208. export DBHOST=%s && \\
  209. export MAX_THREADS=%s && \\
  210. export OUT=%s && \\
  211. export ERR=%s && \\
  212. cd %s && \\
  213. %s''' % (self.fwroot,
  214. self.directory,
  215. self.install_root,
  216. self.database_host,
  217. self.benchmarker.threads,
  218. os.path.join(self.fwroot, out.name),
  219. os.path.join(self.fwroot, err.name),
  220. self.directory,
  221. command)
  222. logging.info("To run framework manually, copy/paste this:\n%s", debug_command)
  223. def tee_output(prefix, line):
  224. # Log to current terminal
  225. # Needs to be one atomic write, so we join because
  226. # list operations are faster than string concat
  227. sys.stdout.write(u''.encode('utf-8').join([prefix, line]))
  228. sys.stdout.flush()
  229. # logging.error("".join([prefix, line]))
  230. # Goal: Stream output of both benchmark toolset and
  231. # server to the console and to a file
  232. # Problem: Capturing output of subprocess and children
  233. # Solution: Use pipes provided by python
  234. # Future-proof: Add unit tests that ensure this code works in all situations
  235. #
  236. # https://blogs.gnome.org/markmc/2013/06/04/async-io-and-python/
  237. # http://eyalarubas.com/python-subproc-nonblock.html
  238. p = subprocess.Popen(command, cwd=self.directory,
  239. shell=True, stdout=subprocess.PIPE, bufsize=1,
  240. stderr=subprocess.STDOUT)
  241. nbsr = NonBlockingStreamReader(p.stdout)
  242. timeout = datetime.now() + timedelta(minutes = 10)
  243. time_remaining = timeout - datetime.now()
  244. # Flush output until setup.sh process is finished. This is
  245. # either a) when setup.sh exits b) when the port is bound
  246. # c) when we run out of time
  247. #
  248. # Note: child processes forked using & will still be alive
  249. # and directing their output to the pipes. E.g. even after
  250. # this loop dies the pipes are used to capture stdout/err from
  251. # the running server
  252. #
  253. # Explicitly set our prefix encoding
  254. prefix = (u"Setup %s: " % self.name).encode('utf-8')
  255. while not (p.poll()
  256. or self.benchmarker.is_port_bound(self.port)
  257. or time_remaining.total_seconds() < 0):
  258. line = nbsr.readline(0.1)
  259. if line:
  260. tee_output(prefix, line)
  261. time_remaining = timeout - datetime.now()
  262. # Were we timed out?
  263. if time_remaining.total_seconds() < 0:
  264. print "Setup.sh timed out!!"
  265. p.kill()
  266. return 1
  267. # If setup.sh exited, use the return code
  268. # Else return 0 if the port was bound
  269. retcode = (p.poll() or 0 if self.benchmarker.is_port_bound(self.port) else 1)
  270. # Before we return control to the benchmarker, spin up a
  271. # thread to keep an eye on the pipes in case the server
  272. # spits anything to stdout/stderr
  273. # TODO add exit condition
  274. def watch_child_pipes(nbsr, prefix):
  275. while True:
  276. line = nbsr.readline(0.1)
  277. if line:
  278. tee_output(prefix, line)
  279. prefix = (u"Server %s: " % self.name).encode('utf-8')
  280. watch_thread = Thread(target = watch_child_pipes,
  281. args = (nbsr, prefix))
  282. watch_thread.daemon = True
  283. watch_thread.start()
  284. logging.info("Executed %s.sh, returning %s", self.setup_file, retcode)
  285. os.chdir(previousDir)
  286. return retcode
  287. ############################################################
  288. # End start
  289. ############################################################
  290. ############################################################
  291. # verify_urls
  292. # Verifys each of the URLs for this test. THis will sinply
  293. # curl the URL and check for it's return status.
  294. # For each url, a flag will be set on this object for whether
  295. # or not it passed
  296. # Returns True if all verifications succeeded
  297. ############################################################
  298. def verify_urls(self, out, err):
  299. result = True
  300. def verify_type(test_type):
  301. test = self.runTests[test_type]
  302. test.setup_out_err(out, err)
  303. out.write(header("VERIFYING %s" % test_type.upper()))
  304. base_url = "http://%s:%s" % (self.benchmarker.server_host, self.port)
  305. try:
  306. results = test.verify(base_url)
  307. except Exception as e:
  308. results = [('fail',"""Caused Exception in TFB
  309. This almost certainly means your return value is incorrect,
  310. but also that you have found a bug. Please submit an issue
  311. including this message: %s\n%s""" % (e, traceback.format_exc()),
  312. base_url)]
  313. logging.warning("Verifying test %s for %s caused an exception: %s", test_type, self.name, e)
  314. traceback.format_exc()
  315. test.failed = any(result is 'fail' for (result, reason, url) in results)
  316. test.warned = any(result is 'warn' for (result, reason, url) in results)
  317. test.passed = all(result is 'pass' for (result, reason, url) in results)
  318. def output_result(result, reason, url):
  319. out.write(" %s for %s\n" % (result.upper(), url))
  320. print " %s for %s" % (result.upper(), url)
  321. if reason is not None and len(reason) != 0:
  322. for line in reason.splitlines():
  323. out.write(" " + line + '\n')
  324. print " " + line
  325. [output_result(r1,r2,url) for (r1, r2, url) in results]
  326. if test.failed:
  327. self.benchmarker.report_verify_results(self, test_type, 'fail')
  328. elif test.warned:
  329. self.benchmarker.report_verify_results(self, test_type, 'warn')
  330. elif test.passed:
  331. self.benchmarker.report_verify_results(self, test_type, 'pass')
  332. else:
  333. raise Exception("Unknown error - test did not pass,warn,or fail")
  334. result = True
  335. for test_type in self.runTests:
  336. verify_type(test_type)
  337. if self.runTests[test_type].failed:
  338. result = False
  339. return result
  340. ############################################################
  341. # End verify_urls
  342. ############################################################
  343. ############################################################
  344. # benchmark
  345. # Runs the benchmark for each type of test that it implements
  346. # JSON/DB/Query.
  347. ############################################################
  348. def benchmark(self, out, err):
  349. def benchmark_type(test_type):
  350. out.write("BENCHMARKING %s ... " % test_type.upper())
  351. test = self.runTests[test_type]
  352. test.setup_out_err(out, err)
  353. output_file = self.benchmarker.output_file(self.name, test_type)
  354. if not os.path.exists(output_file):
  355. # Open to create the empty file
  356. with open(output_file, 'w'):
  357. pass
  358. if not test.failed:
  359. if test_type == 'plaintext': # One special case
  360. remote_script = self.__generate_pipeline_script(test.get_url(), self.port, test.accept_header)
  361. elif test_type == 'query' or test_type == 'update':
  362. remote_script = self.__generate_query_script(test.get_url(), self.port, test.accept_header)
  363. else:
  364. remote_script = self.__generate_concurrency_script(test.get_url(), self.port, test.accept_header)
  365. # Begin resource usage metrics collection
  366. self.__begin_logging(test_type)
  367. # Run the benchmark
  368. with open(output_file, 'w') as raw_file:
  369. p = subprocess.Popen(self.benchmarker.client_ssh_string.split(" "), stdin=subprocess.PIPE, stdout=raw_file, stderr=err)
  370. p.communicate(remote_script)
  371. err.flush()
  372. # End resource usage metrics collection
  373. self.__end_logging()
  374. results = self.__parse_test(test_type)
  375. print "Benchmark results:"
  376. pprint(results)
  377. self.benchmarker.report_benchmark_results(framework=self, test=test_type, results=results['results'])
  378. out.write( "Complete\n" )
  379. out.flush()
  380. for test_type in self.runTests:
  381. benchmark_type(test_type)
  382. ############################################################
  383. # End benchmark
  384. ############################################################
  385. ############################################################
  386. # parse_all
  387. # Method meant to be run for a given timestamp
  388. ############################################################
  389. def parse_all(self):
  390. for test_type in self.runTests:
  391. if os.path.exists(self.benchmarker.get_output_file(self.name, test_type)):
  392. results = self.__parse_test(test_type)
  393. self.benchmarker.report_benchmark_results(framework=self, test=test_type, results=results['results'])
  394. ############################################################
  395. # __parse_test(test_type)
  396. ############################################################
  397. def __parse_test(self, test_type):
  398. try:
  399. results = dict()
  400. results['results'] = []
  401. stats = []
  402. if os.path.exists(self.benchmarker.get_output_file(self.name, test_type)):
  403. with open(self.benchmarker.output_file(self.name, test_type)) as raw_data:
  404. is_warmup = True
  405. rawData = None
  406. for line in raw_data:
  407. if "Queries:" in line or "Concurrency:" in line:
  408. is_warmup = False
  409. rawData = None
  410. continue
  411. if "Warmup" in line or "Primer" in line:
  412. is_warmup = True
  413. continue
  414. if not is_warmup:
  415. if rawData == None:
  416. rawData = dict()
  417. results['results'].append(rawData)
  418. #if "Requests/sec:" in line:
  419. # m = re.search("Requests/sec:\s+([0-9]+)", line)
  420. # rawData['reportedResults'] = m.group(1)
  421. # search for weighttp data such as succeeded and failed.
  422. if "Latency" in line:
  423. m = re.findall("([0-9]+\.*[0-9]*[us|ms|s|m|%]+)", line)
  424. if len(m) == 4:
  425. rawData['latencyAvg'] = m[0]
  426. rawData['latencyStdev'] = m[1]
  427. rawData['latencyMax'] = m[2]
  428. # rawData['latencyStdevPercent'] = m[3]
  429. #if "Req/Sec" in line:
  430. # m = re.findall("([0-9]+\.*[0-9]*[k|%]*)", line)
  431. # if len(m) == 4:
  432. # rawData['requestsAvg'] = m[0]
  433. # rawData['requestsStdev'] = m[1]
  434. # rawData['requestsMax'] = m[2]
  435. # rawData['requestsStdevPercent'] = m[3]
  436. #if "requests in" in line:
  437. # m = re.search("requests in ([0-9]+\.*[0-9]*[ms|s|m|h]+)", line)
  438. # if m != None:
  439. # # parse out the raw time, which may be in minutes or seconds
  440. # raw_time = m.group(1)
  441. # if "ms" in raw_time:
  442. # rawData['total_time'] = float(raw_time[:len(raw_time)-2]) / 1000.0
  443. # elif "s" in raw_time:
  444. # rawData['total_time'] = float(raw_time[:len(raw_time)-1])
  445. # elif "m" in raw_time:
  446. # rawData['total_time'] = float(raw_time[:len(raw_time)-1]) * 60.0
  447. # elif "h" in raw_time:
  448. # rawData['total_time'] = float(raw_time[:len(raw_time)-1]) * 3600.0
  449. if "requests in" in line:
  450. m = re.search("([0-9]+) requests in", line)
  451. if m != None:
  452. rawData['totalRequests'] = int(m.group(1))
  453. if "Socket errors" in line:
  454. if "connect" in line:
  455. m = re.search("connect ([0-9]+)", line)
  456. rawData['connect'] = int(m.group(1))
  457. if "read" in line:
  458. m = re.search("read ([0-9]+)", line)
  459. rawData['read'] = int(m.group(1))
  460. if "write" in line:
  461. m = re.search("write ([0-9]+)", line)
  462. rawData['write'] = int(m.group(1))
  463. if "timeout" in line:
  464. m = re.search("timeout ([0-9]+)", line)
  465. rawData['timeout'] = int(m.group(1))
  466. if "Non-2xx" in line:
  467. m = re.search("Non-2xx or 3xx responses: ([0-9]+)", line)
  468. if m != None:
  469. rawData['5xx'] = int(m.group(1))
  470. if "STARTTIME" in line:
  471. m = re.search("[0-9]+", line)
  472. rawData["startTime"] = int(m.group(0))
  473. if "ENDTIME" in line:
  474. m = re.search("[0-9]+", line)
  475. rawData["endTime"] = int(m.group(0))
  476. test_stats = self.__parse_stats(test_type, rawData["startTime"], rawData["endTime"], 1)
  477. # rawData["averageStats"] = self.__calculate_average_stats(test_stats)
  478. stats.append(test_stats)
  479. with open(self.benchmarker.stats_file(self.name, test_type) + ".json", "w") as stats_file:
  480. json.dump(stats, stats_file, indent=2)
  481. return results
  482. except IOError:
  483. return None
  484. ############################################################
  485. # End benchmark
  486. ############################################################
  487. ##########################################################################################
  488. # Private Methods
  489. ##########################################################################################
  490. ############################################################
  491. # __generate_concurrency_script(url, port)
  492. # Generates the string containing the bash script that will
  493. # be run on the client to benchmark a single test. This
  494. # specifically works for the variable concurrency tests (JSON
  495. # and DB)
  496. ############################################################
  497. def __generate_concurrency_script(self, url, port, accept_header, wrk_command="wrk"):
  498. headers = self.headers_template.format(accept=accept_header)
  499. return self.concurrency_template.format(max_concurrency=max(self.benchmarker.concurrency_levels),
  500. max_threads=self.benchmarker.threads, name=self.name, duration=self.benchmarker.duration,
  501. levels=" ".join("{}".format(item) for item in self.benchmarker.concurrency_levels),
  502. server_host=self.benchmarker.server_host, port=port, url=url, headers=headers, wrk=wrk_command)
  503. ############################################################
  504. # __generate_pipeline_script(url, port)
  505. # Generates the string containing the bash script that will
  506. # be run on the client to benchmark a single pipeline test.
  507. ############################################################
  508. def __generate_pipeline_script(self, url, port, accept_header, wrk_command="wrk"):
  509. headers = self.headers_template.format(accept=accept_header)
  510. return self.pipeline_template.format(max_concurrency=16384,
  511. max_threads=self.benchmarker.threads, name=self.name, duration=self.benchmarker.duration,
  512. levels=" ".join("{}".format(item) for item in [256,1024,4096,16384]),
  513. server_host=self.benchmarker.server_host, port=port, url=url, headers=headers, wrk=wrk_command,
  514. pipeline=16)
  515. ############################################################
  516. # __generate_query_script(url, port)
  517. # Generates the string containing the bash script that will
  518. # be run on the client to benchmark a single test. This
  519. # specifically works for the variable query tests (Query)
  520. ############################################################
  521. def __generate_query_script(self, url, port, accept_header):
  522. headers = self.headers_template.format(accept=accept_header)
  523. return self.query_template.format(max_concurrency=max(self.benchmarker.concurrency_levels),
  524. max_threads=self.benchmarker.threads, name=self.name, duration=self.benchmarker.duration,
  525. levels=" ".join("{}".format(item) for item in self.benchmarker.query_levels),
  526. server_host=self.benchmarker.server_host, port=port, url=url, headers=headers)
  527. ############################################################
  528. # Returns True if any test type this this framework test will use a DB
  529. ############################################################
  530. def requires_database(self):
  531. '''Returns True/False if this test requires a database'''
  532. return any(tobj.requires_db for (ttype,tobj) in self.runTests.iteritems())
  533. ############################################################
  534. # __begin_logging
  535. # Starts a thread to monitor the resource usage, to be synced with the client's time
  536. # TODO: MySQL and InnoDB are possible. Figure out how to implement them.
  537. ############################################################
  538. def __begin_logging(self, test_type):
  539. output_file = "{file_name}".format(file_name=self.benchmarker.get_stats_file(self.name, test_type))
  540. dstat_string = "dstat -afilmprsT --aio --fs --ipc --lock --raw --socket --tcp \
  541. --raw --socket --tcp --udp --unix --vm --disk-util \
  542. --rpc --rpcd --output {output_file}".format(output_file=output_file)
  543. cmd = shlex.split(dstat_string)
  544. dev_null = open(os.devnull, "w")
  545. self.subprocess_handle = subprocess.Popen(cmd, stdout=dev_null)
  546. ##############################################################
  547. # Begin __end_logging
  548. # Stops the logger thread and blocks until shutdown is complete.
  549. ##############################################################
  550. def __end_logging(self):
  551. self.subprocess_handle.terminate()
  552. self.subprocess_handle.communicate()
  553. ##############################################################
  554. # Begin __parse_stats
  555. # For each test type, process all the statistics, and return a multi-layered dictionary
  556. # that has a structure as follows:
  557. # (timestamp)
  558. # | (main header) - group that the stat is in
  559. # | | (sub header) - title of the stat
  560. # | | | (stat) - the stat itself, usually a floating point number
  561. ##############################################################
  562. def __parse_stats(self, test_type, start_time, end_time, interval):
  563. stats_dict = dict()
  564. stats_file = self.benchmarker.stats_file(self.name, test_type)
  565. with open(stats_file) as stats:
  566. while(stats.next() != "\n"): # dstat doesn't output a completely compliant CSV file - we need to strip the header
  567. pass
  568. stats_reader = csv.reader(stats)
  569. main_header = stats_reader.next()
  570. sub_header = stats_reader.next()
  571. time_row = sub_header.index("epoch")
  572. int_counter = 0
  573. for row in stats_reader:
  574. time = float(row[time_row])
  575. int_counter+=1
  576. if time < start_time:
  577. continue
  578. elif time > end_time:
  579. return stats_dict
  580. if int_counter % interval != 0:
  581. continue
  582. row_dict = dict()
  583. for nextheader in main_header:
  584. if nextheader != "":
  585. row_dict[nextheader] = dict()
  586. header = ""
  587. for item_num, column in enumerate(row):
  588. if(len(main_header[item_num]) != 0):
  589. header = main_header[item_num]
  590. row_dict[header][sub_header[item_num]] = float(column) # all the stats are numbers, so we want to make sure that they stay that way in json
  591. stats_dict[time] = row_dict
  592. return stats_dict
  593. ##############################################################
  594. # End __parse_stats
  595. ##############################################################
  596. def __getattr__(self, name):
  597. """For backwards compatibility, we used to pass benchmarker
  598. as the argument to the setup.sh files"""
  599. try:
  600. x = getattr(self.benchmarker, name)
  601. except AttributeError:
  602. print "AttributeError: %s not a member of FrameworkTest or Benchmarker" % name
  603. print "This is probably a bug"
  604. raise
  605. return x
  606. ##############################################################
  607. # Begin __calculate_average_stats
  608. # We have a large amount of raw data for the statistics that
  609. # may be useful for the stats nerds, but most people care about
  610. # a couple of numbers. For now, we're only going to supply:
  611. # * Average CPU
  612. # * Average Memory
  613. # * Total network use
  614. # * Total disk use
  615. # More may be added in the future. If they are, please update
  616. # the above list.
  617. # Note: raw_stats is directly from the __parse_stats method.
  618. # Recall that this consists of a dictionary of timestamps,
  619. # each of which contain a dictionary of stat categories which
  620. # contain a dictionary of stats
  621. ##############################################################
  622. def __calculate_average_stats(self, raw_stats):
  623. raw_stat_collection = dict()
  624. for timestamp, time_dict in raw_stats.items():
  625. for main_header, sub_headers in time_dict.items():
  626. item_to_append = None
  627. if 'cpu' in main_header:
  628. # We want to take the idl stat and subtract it from 100
  629. # to get the time that the CPU is NOT idle.
  630. item_to_append = sub_headers['idl'] - 100.0
  631. elif main_header == 'memory usage':
  632. item_to_append = sub_headers['used']
  633. elif 'net' in main_header:
  634. # Network stats have two parts - recieve and send. We'll use a tuple of
  635. # style (recieve, send)
  636. item_to_append = (sub_headers['recv'], sub_headers['send'])
  637. elif 'dsk' or 'io' in main_header:
  638. # Similar for network, except our tuple looks like (read, write)
  639. item_to_append = (sub_headers['read'], sub_headers['writ'])
  640. if item_to_append is not None:
  641. if main_header not in raw_stat_collection:
  642. raw_stat_collection[main_header] = list()
  643. raw_stat_collection[main_header].append(item_to_append)
  644. # Simple function to determine human readable size
  645. # http://stackoverflow.com/questions/1094841/reusable-library-to-get-human-readable-version-of-file-size
  646. def sizeof_fmt(num):
  647. # We'll assume that any number we get is convertable to a float, just in case
  648. num = float(num)
  649. for x in ['bytes','KB','MB','GB']:
  650. if num < 1024.0 and num > -1024.0:
  651. return "%3.1f%s" % (num, x)
  652. num /= 1024.0
  653. return "%3.1f%s" % (num, 'TB')
  654. # Now we have our raw stats in a readable format - we need to format it for display
  655. # We need a floating point sum, so the built in sum doesn't cut it
  656. display_stat_collection = dict()
  657. for header, values in raw_stat_collection.items():
  658. display_stat = None
  659. if 'cpu' in header:
  660. display_stat = sizeof_fmt(math.fsum(values) / len(values))
  661. elif main_header == 'memory usage':
  662. display_stat = sizeof_fmt(math.fsum(values) / len(values))
  663. elif 'net' in main_header:
  664. receive, send = zip(*values) # unzip
  665. display_stat = {'receive': sizeof_fmt(math.fsum(receive)), 'send': sizeof_fmt(math.fsum(send))}
  666. else: # if 'dsk' or 'io' in header:
  667. read, write = zip(*values) # unzip
  668. display_stat = {'read': sizeof_fmt(math.fsum(read)), 'write': sizeof_fmt(math.fsum(write))}
  669. display_stat_collection[header] = display_stat
  670. return display_stat
  671. ###########################################################################################
  672. # End __calculate_average_stats
  673. #########################################################################################
  674. ##########################################################################################
  675. # Constructor
  676. ##########################################################################################
  677. def __init__(self, name, directory, benchmarker, runTests, args):
  678. self.name = name
  679. self.directory = directory
  680. self.benchmarker = benchmarker
  681. self.runTests = runTests
  682. self.fwroot = benchmarker.fwroot
  683. self.approach = ""
  684. self.classification = ""
  685. self.database = ""
  686. self.framework = ""
  687. self.language = ""
  688. self.orm = ""
  689. self.platform = ""
  690. self.webserver = ""
  691. self.os = ""
  692. self.database_os = ""
  693. self.display_name = ""
  694. self.notes = ""
  695. self.versus = ""
  696. # setup logging
  697. logging.basicConfig(stream=sys.stderr, level=logging.INFO)
  698. self.install_root="%s/%s" % (self.fwroot, "installs")
  699. if benchmarker.install_strategy is 'pertest':
  700. self.install_root="%s/pertest/%s" % (self.install_root, name)
  701. # Used in setup.sh scripts for consistency with
  702. # the bash environment variables
  703. self.troot = self.directory
  704. self.iroot = self.install_root
  705. self.__dict__.update(args)
  706. ############################################################
  707. # End __init__
  708. ############################################################
  709. ############################################################
  710. # End FrameworkTest
  711. ############################################################
  712. ##########################################################################################
  713. # Static methods
  714. ##########################################################################################
  715. ##############################################################
  716. # parse_config(config, directory, benchmarker)
  717. # parses a config file and returns a list of FrameworkTest
  718. # objects based on that config file.
  719. ##############################################################
  720. def parse_config(config, directory, benchmarker):
  721. tests = []
  722. # This sort ordering is set up specifically to return the length
  723. # of the test name. There were SO many problems involved with
  724. # 'plaintext' being run first (rather, just not last) that we
  725. # needed to ensure that it was run last for every framework.
  726. def testOrder(type_name):
  727. return len(type_name)
  728. # The config object can specify multiple tests
  729. # Loop over them and parse each into a FrameworkTest
  730. for test in config['tests']:
  731. names = [name for (name,keys) in test.iteritems()]
  732. if "default" not in names:
  733. logging.warn("Framework %s does not define a default test in benchmark_config", config['framework'])
  734. for test_name, test_keys in test.iteritems():
  735. # Prefix all test names with framework except 'default' test
  736. if test_name == 'default':
  737. test_name = config['framework']
  738. else:
  739. test_name = "%s-%s" % (config['framework'], test_name)
  740. # Ensure FrameworkTest.framework is available
  741. if not test_keys['framework']:
  742. test_keys['framework'] = config['framework']
  743. #if test_keys['framework'].lower() != config['framework'].lower():
  744. # print Exception("benchmark_config for test %s is invalid - test framework '%s' must match benchmark_config framework '%s'" %
  745. # (test_name, test_keys['framework'], config['framework']))
  746. # Confirm required keys are present
  747. # TODO have a TechEmpower person confirm this list - I don't know what the website requires....
  748. required = ['language','webserver','classification','database','approach','orm','framework','os','database_os']
  749. if not all (key in test_keys for key in required):
  750. raise Exception("benchmark_config for test %s is invalid - missing required keys" % test_name)
  751. # Map test type to a parsed FrameworkTestType object
  752. runTests = dict()
  753. for type_name, type_obj in benchmarker.types.iteritems():
  754. try:
  755. runTests[type_name] = type_obj.copy().parse(test_keys)
  756. except AttributeError as ae:
  757. # This is quite common - most tests don't support all types
  758. # Quitely log it and move on (debug logging is on in travis and this causes
  759. # ~1500 lines of debug, so I'm totally ignoring it for now
  760. # logging.debug("Missing arguments for test type %s for framework test %s", type_name, test_name)
  761. pass
  762. # We need to sort by test_type to run
  763. sortedTestKeys = sorted(runTests.keys(), key=testOrder)
  764. sortedRunTests = OrderedDict()
  765. for sortedTestKey in sortedTestKeys:
  766. sortedRunTests[sortedTestKey] = runTests[sortedTestKey]
  767. # By passing the entire set of keys, each FrameworkTest will have a member for each key
  768. tests.append(FrameworkTest(test_name, directory, benchmarker, sortedRunTests, test_keys))
  769. return tests
  770. ##############################################################
  771. # End parse_config
  772. ##############################################################