On 03/08/2013 03:16 PM, Don Zickus wrote:
This is the API to beaker. It is split into multiple classes because
some of
us are perverted and like wrapping classes with more classes.
There is the basic beaker proxy API and just passes data from autotest to
beaker. There is my offline proxy that mimics data to/from beaker but stores
everything locally for debugging (and avoiding needless beaker provisioning to
get right). There is the debugging proxy for times the xml is making you
scratch your head. And there is the failing proxy to mimic what happens when
your beaker server goes out to lunch.
Mostly of interest to the beaker folks. Autotest folks can review for style,
syntax and entertainment.
Signed-off-by: Don Zickus <dzickus(a)redhat.com>
---
client/bkr_proxy.py | 337 +++++++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 337 insertions(+), 0 deletions(-)
create mode 100644 client/bkr_proxy.py
diff --git a/client/bkr_proxy.py b/client/bkr_proxy.py
new file mode 100644
index 0000000..a8a9ee9
--- /dev/null
+++ b/client/bkr_proxy.py
@@ -0,0 +1,337 @@
+# bkr_proxy.py
+#
+# Copyright (C) 2011 Jan Stancek <jstancek(a)redhat.com>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
+"""
+bkr_proxy - class used to talk to beaker
+"""
+__author__ = """Copyright Jan Stancek 2011"""
+
+
+import time
+import os
+import sys
+import base64
+import xmlrpclib
+import random
+import logging
+import shutil
+from at_utils import logException
+
+log = logging
+AUTOTEST_CACHE_DIR = '/var/cache/autotest'
+BKR_CACHE = AUTOTEST_CACHE_DIR + '/beaker_job.cache'
+
+if sys.version_info < (2, 5):
+ import md5
+ def getMD5Constructor():
+ return md5.new
+else:
+ import hashlib
+ def getMD5Constructor():
+ return hashlib.md5
^ We have a wrapper for sha1 and md5 functions
autotest.client.shared.utils.hash()
Similar in purpose to this getMD5Constructor function.
+class BkrProxyException(Exception):
+ def __init__(self, text):
+ Exception.__init__(self, text)
+
+class OfflineProxy:
+ '''Locally mimic the beaker server. Useful for debugging and
+ dealing with a flakey server. Sits underneath everything.
+ '''
+
+ OFFLINE_RESULTS = AUTOTEST_CACHE_DIR + '/results'
+
+ def __init__(self):
+ if os.path.isdir(self.OFFLINE_RESULTS):
+ shutil.rmtree(self.OFFLINE_RESULTS)
+ os.makedirs(self.OFFLINE_RESULTS)
+ self.cmd_log = open(self.OFFLINE_RESULTS + '/cmd_log', 'a', 0)
+
+ def task_upload_file(self, tid, remotedir, remotename, sz, digest, offset, data):
+ self.cmd_log.write('task_upload_file: tid(%s) remotedir(%s) remotename(%s)
sz(%d) digest(%d) offset(%d) data(%s)\n' % \
+ (tid, remotedir, remotename, sz, digest, offset, data))
+
+ def result_upload_file(self, tid, remotedir, remotename, sz, digest, offset, data):
+ self.cmd_log.write('result_upload_file: tid(%s) remotedir(%s) remotename(%s)
sz(%d) digest(%d) offset(%d) data(%s)\n' % \
+ (tid, remotedir, remotename, sz, digest, offset, data))
+
+ def recipe_upload_file(self, tid, remotedir, remotename, sz, digest, offset, data):
+ self.cmd_log.write('recipe_upload_file: tid(%s) remotedir(%s) remotename(%s)
sz(%d) digest(%d) offset(%d) data(%s)\n' % \
+ (tid, remotedir, remotename, sz, digest, offset, data))
+
+ def get_my_recipe(self, request):
+ '''Returns the cached recipe. Ignore the request for
now'''
+ self.cmd_log.write('get_my_recipe: request(%s)\n' % request)
+ if not os.path.isfile(BKR_CACHE):
+ raise BkrProxyException("Couldn't find cache %s" % BKR_CACHE)
+ return open(BKR_CACHE, 'r').read()
+
+ def task_result(self, task_id, result_type, result_path,
+ result_score, result_summary):
+ '''No remote server to do anything, so just stub it
out'''
+ self.cmd_log.write('task_result: task_id(%s) result_type(%s) result_path(%s)
result_score(%s) result_summary(%s)\n' % \
+ (task_id, result_type, result_path, result_score,
result_summary))
+
+ def task_start(self, task_id, kill_time=None):
+ '''No remote server to do anything, so just stub it
out'''
+ if not kill_time:
+ ktime = "None"
+ else:
+ ktime = str(kill_time)
+ self.cmd_log.write('task_start: task_id(%s) kill_time(%s)\n' %(task_id,
ktime))
+
+ def task_stop(self, task_id, stop_type, msg):
+ '''No remote server to do anything, so just stub it
out'''
+ self.cmd_log.write('task_stop: task_id(%s) stop_type(%s) msg(%s)\n'
%(task_id, stop_type, msg))
+
+ def recipe_stop(self, recipeset_id, stop_type, msg):
+ '''No remote server to do anything, so just stub it
out'''
+ self.cmd_log.write('Recipe_stop: recipeset_id(%s) stop_type(%s)
msg(%s)\n' %(recipeset_id, stop_type, msg))
+
+ def extend_watchdog(self, task_id, killtime):
+ '''No remote server to kill us, so just stub it out'''
+ self.cmd_log.write('Extend_watchdog: task_id(%s) killtime(%d)\n'
%(task_id, killtime))
+
+class FailingProxy:
+ def __init__(self, server_proxy, failafter=0, failchance=0):
+ self.count = 0
+ self.failafter = failafter
+ self.failchance = failchance
+ self.server_proxy = server_proxy
+
+ def test_fail(self):
+ self.count = self.count + 1
+ if self.count >= self.failafter:
+ self.count = 0
+ raise xmlrpclib.Fault(404, 'XMLRPC failed')
+
+ def test_randfail(self):
+ if not self.failchance:
+ return
+ if random.randint(1, self.failchance) == 1:
+ raise xmlrpclib.Fault(404, 'XMLRPC failed')
+
+ def __getattr__(self, name):
+ self.test_fail()
+ self.test_randfail()
+ def wrapper(*args, **kwargs):
+ fn = getattr(self.server_proxy, name)
+ return fn(*args, **kwargs)
+ return wrapper
+
+class SimpleRepeatingProxy:
+ def __init__(self, server_proxy, delay, retries=0):
+ self.delay = delay
+ self.max_retries = retries
+ self.server_proxy = server_proxy
+ self.attempt = 0
+
+ def __getattr__(self, name):
+ def wrapper(*args, **kwargs):
+ self.attempt = self.max_retries + 1
+ while (self.attempt):
+ try:
+ self.attempt = self.attempt - 1
+ fn = getattr(self.server_proxy, name)
+ return fn(*args, **kwargs)
+ except xmlrpclib.Fault, flt:
+ logException(flt)
+ if (not self.max_retries):
+ raise BkrProxyException(flt.faultString)
+ if self.attempt == 0:
+ raise BkrProxyException('Failed to deliver message after %s
retries' % self.max_retries)
+ log.info('Failed to deliver message: %s', name)
+ log.info('Sleeping for %s', self.delay)
+ time.sleep(self.delay)
+ log.info('Trying again')
+ return wrapper
+
+class DebuggingProxy:
+ def __init__(self, server_proxy):
+ self.server_proxy = server_proxy
+
+ def __getattr__(self, name):
+ def wrapper(*args, **kwargs):
+ def print_args(args):
+ #TODO: hoping to avoid call if not-debug
+ max_param_len = 15
+ short_args = []
+ for arg in args:
+ if (len(str(arg)) > max_param_len):
+ short_args.append(str(arg)[:max_param_len] + '..')
+ else:
+ short_args.append(arg)
+ return short_args
+
+ log.debug('XML: %s(%s)', name, print_args(args))
+ fn = getattr(self.server_proxy, name)
+ ret = fn(*args, **kwargs)
+ log.debug(' --> %s', ret)
+ return ret
+ return wrapper
+
+class BkrProxy:
+ def __init__(self, labc_url='', delay=60, retries=0, failafter=0,
failchance=0):
+
+ #why do I love wrappers??
+
+ #labc_url determines local or remote functionality
+ if labc_url:
+ self.labc_url = "http://" + labc_url + ":8000"
+ self.server_proxy = xmlrpclib.Server(self.labc_url)
+ else:
+ log.info("Using OfflineProxy() as remote host")
+ self.server_proxy = OfflineProxy()
+ self.labc_url = ''
+
+ #emulate failures
+ if failafter or failchance:
+ self.server_proxy = FailingProxy(self.server_proxy, failafter=failafter,
failchance=failchance)
+
+ #deal with retries
+ self.server_proxy = SimpleRepeatingProxy(self.server_proxy, delay=delay,
retries=retries)
+
+ #debugging
+ self.server_proxy = DebuggingProxy(self.server_proxy)
+
+ def task_upload_file(self, localfile, id, remotename, remotedir):
+ return self._uploadWrapper('task_upload_file', localfile, id,
+ remotename, remotedir)
+
+ def result_upload_file(self, localfile, id, remotename, remotedir):
+ return self._uploadWrapper('result_upload_file', localfile, id,
+ remotename, remotedir)
+
+ def recipe_upload_file(self, localfile, id, remotename, remotedir):
+ return self._uploadWrapper('recipe_upload_file', localfile, id,
+ remotename, remotedir)
+
+ def get_recipe(self, hostname):
+ ret = self.server_proxy.get_my_recipe({'system_name': hostname})
+ #cache this locally, in case the server goes down
+ if not os.path.isdir(os.path.dirname(BKR_CACHE)):
+ os.makedirs(os.path.dirname(BKR_CACHE))
+ open(BKR_CACHE, 'w').write("%s" % ret)
+ return ret
+
+ def task_result(self, task_id, result_type, result_path,
+ result_score, result_summary):
+ return self.server_proxy.task_result(task_id, result_type,
+ result_path, result_score, result_summary)
+
+ def task_start(self, task_id, kill_time=None):
+ if (kill_time):
+ return self.server_proxy.task_start(task_id, kill_time)
+ else:
+ return self.server_proxy.task_start(task_id)
+
+ def task_stop(self, task_id, stop_type, msg):
+ return self.server_proxy.task_stop(task_id, stop_type, msg)
+
+ def recipe_stop(self, recipeset_id, stop_type, msg):
+ return self.server_proxy.recipe_stop(recipeset_id, stop_type, msg)
+
+ def extend_watchdog(self, task_id, killtime):
+ return self.server_proxy.extend_watchdog(task_id, killtime)
+
+ def _uploadWrapper(self, rpc_method, localfile, tid, remotename=None,
+ remotedir=None, callback=None, blocksize=262144, start=0):
+ """
+ upload a file in chunks - taken over from rhts-test-env and modified
+ rpc_method - *_upload_file, e.g. task_upload_file
+ localfile - filepath to file on local filesystem
+ tid - ID of task, result or recipe
+ remotename - name of file on remote side
+ remotedir - relative path on remote side
+ """
+ started = time.time()
+ ret = False
+ hashlibMD5 = getMD5Constructor()
+ fn = getattr(self.server_proxy, rpc_method)
+
+ if remotename == None:
+ remotename = os.path.basename(localfile)
+
+ if remotedir == None:
+ remotedir = '.'
+
+ fo = file(localfile, "r") #specify bufsize?
+ totalsize = os.path.getsize(localfile)
+ ofs = start
+ if ofs != 0:
+ fo.seek(ofs)
+
+ digestor = hashlibMD5()
+
+ debug = False
+ if callback:
+ callback(0, totalsize, 0, 0, 0)
+ while ofs <= totalsize:
+ if (totalsize - ofs) < blocksize:
+ blocksize = totalsize - ofs
+ lap = time.time()
+ contents = fo.read(blocksize)
+
+ digestor.update(contents)
+ size = len(contents)
+ data = base64.encodestring(contents)
+ if size == 0:
+ # end of file, use offset = -1 to finalize upload
+ offset = -1
+ digest = digestor.hexdigest()
+ sz = ofs
+ else:
+ offset = ofs
+ digest = hashlibMD5(contents).hexdigest()
+ sz = size
+ del contents
+
+ if True:
+ if debug:
+ log.info('uploadFile(%r,%r,%r,%r,%r,...)',
+ tid, remotename, sz, digest, offset)
+
+ ret = fn(tid, remotedir, remotename, sz, digest, offset, data)
+
+ if debug:
+ log.info('ret: ', ret)
+
+ if size == 0:
+ break
+
+ ofs += size
+ now = time.time()
+ t1 = now - lap
+ if t1 <= 0:
+ t1 = 1
+ t2 = now - started
+ if t2 <= 0:
+ t2 = 1
+ if debug:
+ log.info('Uploaded %d bytes in %f seconds (%f kbytes/sec)',
+ size, t1, size / t1 / 1024)
+ log.info('Total: %d bytes in %f seconds (%f kbytes/sec)',
+ ofs, t2, ofs / t2 / 1024)
+ if callback:
+ callback(ofs, totalsize, size, t1, t2)
+ fo.close()
+ return ret
+
+
+if __name__ == '__main__':
+ pass
+