opendev/puppet-log_processor
Code Issues Proposed changes

Add subunit2sql gearman workers

Browse Source 
This adds a new gearman worker to process the subunit files from
the gate job runs. It will use subunit2sql to connect to a sql
server and process the data from the subunit file. The
log-gearman-client is modified to allow for pushing subunit jobs
to gearman, and the worker model for processsing logs is borrowed
to process the subunit files.

Change-Id: I83103eb6afc22d91f916583c36c0e956c23a64b3
This commit is contained in:
Matthew Treinish 2014-07-18 08:26:52 -04:00
parent 742c92e537
commit e5fbd6ca48
1 changed files with 11 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
files/log-gearman-client.py
View File

@ -75,6 +75,10 @@ class EventProcessor(threading.Thread):
output['source_url'] = source_url output['source_url'] = source_url
output['retry'] = fileopts.get('retry-get', False) output['retry'] = fileopts.get('retry-get', False)
output['event'] = out_event output['event'] = out_event
if 'subunit' in fileopts.get('name'):
job = gear.Job(b'push-subunit',
json.dumps(output).encode('utf8'))
else:
job = gear.Job(b'push-log', json.dumps(output).encode('utf8')) job = gear.Job(b'push-log', json.dumps(output).encode('utf8'))
try: try:
self.gearman_client.submitJob(job) self.gearman_client.submitJob(job)
@ -146,10 +150,14 @@ class Server(object):
gearclient = gear.Client() gearclient = gear.Client()
gearclient.addServer('localhost') gearclient.addServer('localhost')
gearclient.waitForServer() gearclient.waitForServer()
processor = EventProcessor( log_processor = EventProcessor(
publisher, gearclient, publisher, gearclient,
self.config['source-files'], self.source_url) self.config['source-files'], self.source_url)
self.processors.append(processor) subunit_processor = EventProcessor(
publisher, gearclient,
self.config['subunit-files'], self.source_url)
self.processors.append(log_processor)
self.processors.append(subunit_processor)
def main(self): def main(self):
statsd_host = os.environ.get('STATSD_HOST') statsd_host = os.environ.get('STATSD_HOST')