opendev/puppet-log_processor
Code Issues Proposed changes

Handle cases without a build_change field

Browse Source 
Previously the mqtt topic generation always assumed a build_change was
present. However there are some cases where the isn't a build_change in
the metadata, like periodic, post, and release jobs. This commit handles
those edge cases so it uses the build queue in the topic instead of the
build_change. If that doesn't work the topic is just the project.

Change-Id: I26dba76e3475749d00a45b076d981778f885c339
This commit is contained in:
Matthew Treinish 2017-08-03 11:40:20 -04:00
parent 541d226b36
commit 662ae3777c
No known key found for this signature in database
GPG Key ID: FD12A0F214C9E177
1 changed files with 18 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
files/log-gearman-worker.py
View File

@ -192,7 +192,8 @@ class LogRetriever(threading.Thread):
}) })
self.mqtt.publish_single(msg, fields.get('project'), self.mqtt.publish_single(msg, fields.get('project'),
fields.get('build_change'), fields.get('build_change'),
'retrieve_logs') 'retrieve_logs',
fields.get('build_queue'))
except Exception as e: except Exception as e:
logging.exception("Exception handling log event.") logging.exception("Exception handling log event.")
job.sendWorkException(str(e).encode('utf-8')) job.sendWorkException(str(e).encode('utf-8'))
@ -204,7 +205,8 @@ class LogRetriever(threading.Thread):
}) })
self.mqtt.publish_single(msg, fields.get('project'), self.mqtt.publish_single(msg, fields.get('project'),
fields.get('build_change'), fields.get('build_change'),
'retrieve_logs') 'retrieve_logs',
fields.get('build_queue'))
def _retrieve_log(self, source_url, retry): def _retrieve_log(self, source_url, retry):
# TODO (clarkb): This should check the content type instead of file # TODO (clarkb): This should check the content type instead of file
@ -321,7 +323,8 @@ class StdOutLogProcessor(object):
}) })
self.mqtt.publish_single(msg, log.get('project'), self.mqtt.publish_single(msg, log.get('project'),
log.get('build_change'), log.get('build_change'),
'logs_to_logstash') 'logs_to_logstash',
log.get('build_queue'))
class INETLogProcessor(object): class INETLogProcessor(object):
@ -353,7 +356,8 @@ class INETLogProcessor(object):
}) })
self.mqtt.publish_single(msg, log.get('project'), self.mqtt.publish_single(msg, log.get('project'),
log.get('build_change'), log.get('build_change'),
'logs_to_logstash') 'logs_to_logstash',
log.get('build_queue'))
except: except:
logging.exception("Exception sending INET event.") logging.exception("Exception sending INET event.")
# Logstash seems to take about a minute to start again. Wait 90 # Logstash seems to take about a minute to start again. Wait 90
@ -370,7 +374,8 @@ class INETLogProcessor(object):
}) })
self.mqtt.publish_single(msg, log.get('project'), self.mqtt.publish_single(msg, log.get('project'),
log.get('build_change'), log.get('build_change'),
'logs_to_logstash') 'logs_to_logstash',
log.get('build_queue'))
class UDPLogProcessor(INETLogProcessor): class UDPLogProcessor(INETLogProcessor):
@ -397,8 +402,14 @@ class PushMQTT(object):
def _generate_topic(self, project, job_id, action): def _generate_topic(self, project, job_id, action):
return '/'.join([self.base_topic, project, job_id, action]) return '/'.join([self.base_topic, project, job_id, action])
def publish_single(self, msg, project, job_id, action): def publish_single(self, msg, project, job_id, action, build_queue=None):
topic = self._generate_topic(project, job_id, action) if job_id:
topic = self._generate_topic(project, job_id, action)
elif build_queue:
topic = self._generate_topic(project, build_queue, action)
else
topic = self.base_topic + '/' + project
publish.single(topic, msg, hostname=self.hostname, publish.single(topic, msg, hostname=self.hostname,
port=self.port, client_id=self.client_id, port=self.port, client_id=self.client_id,
keepalive=self.keepalive, will=self.will, keepalive=self.keepalive, will=self.will,