From 2ff4f7957b8c8cabaf11183666b2ef35c58c2e31 Mon Sep 17 00:00:00 2001
From: Ramy Asselin <ramy.asselin@hp.com>
Date: Wed, 28 Jan 2015 12:56:08 -0800
Subject: [PATCH] Split out log_processor
Change-Id: I414e5dadc53b52d8e1c4350b785187951c83a36b
---
modules.env | 1 +
modules/log_processor/files/classify-log.crm | 123 -----
.../files/jenkins-log-client.init | 158 -------
.../log_processor/files/log-gearman-client.py | 214 ---------
.../log_processor/files/log-gearman-worker.py | 430 ------------------
modules/log_processor/manifests/client.pp | 73 ---
modules/log_processor/manifests/init.pp | 94 ----
modules/log_processor/manifests/worker.pp | 67 ---
.../templates/jenkins-log-client.default.erb | 5 -
.../templates/jenkins-log-worker.init.erb | 158 -------
10 files changed, 1 insertion(+), 1322 deletions(-)
delete mode 100755 modules/log_processor/files/classify-log.crm
delete mode 100755 modules/log_processor/files/jenkins-log-client.init
delete mode 100644 modules/log_processor/files/log-gearman-client.py
delete mode 100644 modules/log_processor/files/log-gearman-worker.py
delete mode 100644 modules/log_processor/manifests/client.pp
delete mode 100644 modules/log_processor/manifests/init.pp
delete mode 100644 modules/log_processor/manifests/worker.pp
delete mode 100644 modules/log_processor/templates/jenkins-log-client.default.erb
delete mode 100755 modules/log_processor/templates/jenkins-log-worker.init.erb
diff --git a/modules.env b/modules.env
index 3ac8acedd0..59107696f3 100644
--- a/modules.env
+++ b/modules.env
@@ -69,6 +69,7 @@ INTEGRATION_MODULES["https://git.openstack.org/openstack-infra/puppet-iptables"]
INTEGRATION_MODULES["https://git.openstack.org/openstack-infra/puppet-zuul"]="origin/master"
INTEGRATION_MODULES["https://git.openstack.org/openstack-infra/puppet-kibana"]="origin/master"
INTEGRATION_MODULES["https://git.openstack.org/openstack-infra/puppet-lodgeit"]="origin/master"
+INTEGRATION_MODULES["https://git.openstack.org/openstack-infra/puppet-log_processor"]="origin/master"
INTEGRATION_MODULES["https://git.openstack.org/openstack-infra/puppet-logstash"]="origin/master"
INTEGRATION_MODULES["https://git.openstack.org/openstack-infra/puppet-meetbot"]="origin/master"
INTEGRATION_MODULES["https://git.openstack.org/openstack-infra/puppet-mysql_backup"]="origin/master"
diff --git a/modules/log_processor/files/classify-log.crm b/modules/log_processor/files/classify-log.crm
deleted file mode 100755
index 66b02bc48a..0000000000
--- a/modules/log_processor/files/classify-log.crm
+++ /dev/null
@@ -1,123 +0,0 @@
-#! /usr/bin/crm
-#
-# Copyright 2013 OpenStack Foundation
-#
-# Licensed under the Apache License, Version 2.0 (the "License"); you may
-# not use this file except in compliance with the License. You may obtain
-# a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-# License for the specific language governing permissions and limitations
-# under the License.
-
-# This script trains an OSB (Orthogonal Sparse Bigram) bayesian filter
-# with log lines from test runs and classifies each line according to
-# the likelyhood it indicates an error. Very little experimentation
-# has been done to determine the best classifier and training method;
-# further experimentation may be useful.
-
-# The training method is TET -- Train Every Thing. This is not
-# normally advised as a training method for Bayesian filters. In
-# experiments, it identified about twice as many lines as being
-# associated with errers as were indicated by a TOE (Train On Error)
-# method. Some of them were false positives, but many were not, and
-# of those, it had a much higher (pR ~= 37) confidence in them than
-# TOE. TET seems to give qualitatively better results when filtering
-# for higher pR values.
-
-# Set unbuffered IO
-window
-
-# Base component of path to data files
-isolate (:prefix:) /:*:_arg2:/
-
-# Whether this run is for a SUCCESS or FAILURE result
-isolate (:target:) /:*:_arg3:/
-
-# Train each file on a newline just to make sure it exists
-learn [:_nl:] <osb unique microgroom> (:*:prefix:/SUCCESS.css)
-learn [:_nl:] <osb unique microgroom> (:*:prefix:/FAILURE.css)
-{
- # Iterate over each line
- window <bychar> /\n/ /\n/
- {
- isolate (:stats:)
- isolate (:result:)
- isolate (:prob:)
- isolate (:pr:)
- # Save a copy of this line
- isolate (:line:) /:*:_dw:/
- {
- {
- # Remove things that look like timestamps from the beginning of the line
- match (:timestamp:) /^[-.0-9 |:]+/
- alter (:timestamp:) //
- }
- {
- # Don't treat UUIDs as uniquely special.
- match (:uuidtoken:) /[[:xdigit:]]{8}-[[:xdigit:]]{4}-[[:xdigit:]]{4}-[[:xdigit:]]{4}-[[:xdigit:]]{12}/
- alter (:uuidtoken:) /UUIDTOKEN/
- {
- match (:uuidtoken:) <fromnext> /[[:xdigit:]]{8}-[[:xdigit:]]{4}-[[:xdigit:]]{4}-[[:xdigit:]]{4}-[[:xdigit:]]{12}/
- alter (:uuidtoken:) /UUIDTOKEN/
- # Loop to replace all TOKENS in line
- liaf
- }
- }
- {
- # Don't treat IDs as uniquely special.
- match (:idtoken:) /[[:xdigit:]]{32,40}/
- alter (:idtoken:) /IDTOKEN/
- {
- match (:idtoken:) <fromnext> /[[:xdigit:]]{32,40}/
- alter (:idtoken:) /IDTOKEN/
- # Loop to replace all TOKENS in line
- liaf
- }
- }
- {
- # Don't treat IDs as uniquely special.
- match (:numtoken:) /-[[:digit:]]{7,}/
- alter (:numtoken:) /-NUMTOKEN/
- {
- match (:numtoken:) <fromnext> /-[[:digit:]]{7,}/
- alter (:numtoken:) /-NUMTOKEN/
- # Loop to replace all TOKENS in line
- liaf
- }
- }
- # Train on the line
- learn <osb unique microgroom> (:*:prefix:/:*:target:.css)
- # Classify the line to see if it looks more like a SUCCESS or FAILURE line
- classify <osb unique microgroom> (:*:prefix:/SUCCESS.css :*:prefix:/FAILURE.css) (:stats:)
- {
- # The stats variable looks like:
- # CLASSIFY succeeds; success probability: 1.0000 pR: 304.6527
- # Best match to file #0 (/tmp/crm114/console_html/SUCCESS.css) prob: 0.9933 pR: 2.1720
- # Total features in input file: 20
- # #0 (/tmp/crm114/console_html/SUCCESS.css): features: 3544235, hits: 901854, prob: 9.93e-01, pR: 2.17
- # #1 (/tmp/crm114/console_html/FAILURE.css): features: 1, hits: 0, prob: 6.69e-03, pR: -2.17
- # Pull out the filename, probability, and pR (a kind of logarithmic probability, see CRM docs)
- match [:stats:] <nomultiline> /^Best match to .*\/([A-Za-z]+).css\) prob: ([-.0-9]+) pR: ([-.0-9]+)/ ( :: :result: :prob: :pr: )
- {
- # If this line is classified as FAILURE, negate
- # the pR value (which will always be positive).
- # Do this by prepending a '-' or the empty string.
- {
- match [:result:] /FAILURE/
- alter (:result:) /-/
- } alius {
- alter (:result:) //
- }
- }
- # Output the sign and pR value for this line.
- output /:*:result::*:pr:\n/
- }
- }
- }
- liaf
-}
diff --git a/modules/log_processor/files/jenkins-log-client.init b/modules/log_processor/files/jenkins-log-client.init
deleted file mode 100755
index 04357b3d10..0000000000
--- a/modules/log_processor/files/jenkins-log-client.init
+++ /dev/null
@@ -1,158 +0,0 @@
-#! /bin/sh
-### BEGIN INIT INFO
-# Provides: jenkins-log-client
-# Required-Start: $remote_fs $syslog
-# Required-Stop: $remote_fs $syslog
-# Default-Start: 2 3 4 5
-# Default-Stop: 0 1 6
-# Short-Description: Jenkins Log Client
-# Description: Service to push Jenkins logs into logstash.
-### END INIT INFO
-
-# Do NOT "set -e"
-
-# PATH should only include /usr/* if it runs after the mountnfs.sh script
-PATH=/sbin:/usr/sbin:/bin:/usr/bin
-DESC="Jenkins Log Client"
-NAME=jenkins-log-client
-DAEMON=/usr/local/bin/log-gearman-client.py
-PIDFILE=/var/run/$NAME/$NAME.pid
-DAEMON_ARGS="-c /etc/logstash/jenkins-log-client.yaml -d /var/log/logstash/log-client-debug.log -p $PIDFILE"
-SCRIPTNAME=/etc/init.d/$NAME
-USER=logstash
-
-# Exit if the package is not installed
-[ -x "$DAEMON" ] || exit 0
-
-# Read configuration variable file if it is present
-[ -r /etc/default/$NAME ] && . /etc/default/$NAME
-
-# Load the VERBOSE setting and other rcS variables
-. /lib/init/vars.sh
-
-# Define LSB log_* functions.
-# Depend on lsb-base (>= 3.0-6) to ensure that this file is present.
-. /lib/lsb/init-functions
-
-#
-# Function that starts the daemon/service
-#
-do_start()
-{
- # Return
- # 0 if daemon has been started
- # 1 if daemon was already running
- # 2 if daemon could not be started
-
- mkdir -p /var/run/$NAME
- chown $USER /var/run/$NAME
- start-stop-daemon --start --quiet --pidfile $PIDFILE -c $USER --exec $DAEMON --test > /dev/null \
- || return 1
- start-stop-daemon --start --quiet --pidfile $PIDFILE -c $USER --exec $DAEMON -- \
- $DAEMON_ARGS \
- || return 2
- # Add code here, if necessary, that waits for the process to be ready
- # to handle requests from services started subsequently which depend
- # on this one. As a last resort, sleep for some time.
-}
-
-#
-# Function that stops the daemon/service
-#
-do_stop()
-{
- # Return
- # 0 if daemon has been stopped
- # 1 if daemon was already stopped
- # 2 if daemon could not be stopped
- # other if a failure occurred
- start-stop-daemon --stop --signal 9 --pidfile $PIDFILE
- RETVAL="$?"
- [ "$RETVAL" = 2 ] && return 2
- rm -f /var/run/$NAME/*
- return "$RETVAL"
-}
-
-#
-# Function that stops the daemon/service
-#
-#do_graceful_stop()
-#{
-# PID=`cat $PIDFILE`
-# kill -USR1 $PID
-#
-# # wait until really stopped
-# if [ -n "${PID:-}" ]; then
-# i=0
-# while kill -0 "${PID:-}" 2> /dev/null; do
-# if [ $i -eq '0' ]; then
-# echo -n " ... waiting "
-# else
-# echo -n "."
-# fi
-# i=$(($i+1))
-# sleep 1
-# done
-# fi
-#
-# rm -f /var/run/$NAME/*
-#}
-
-#
-# Function that sends a SIGHUP to the daemon/service
-#
-#do_reload() {
-# #
-# # If the daemon can reload its configuration without
-# # restarting (for example, when it is sent a SIGHUP),
-# # then implement that here.
-# #
-# start-stop-daemon --stop --signal 1 --quiet --pidfile $PIDFILE --name zuul-server
-# return 0
-#}
-
-case "$1" in
- start)
- [ "$VERBOSE" != no ] && log_daemon_msg "Starting $DESC" "$NAME"
- do_start
- case "$?" in
- 0|1) [ "$VERBOSE" != no ] && log_end_msg 0 ;;
- 2) [ "$VERBOSE" != no ] && log_end_msg 1 ;;
- esac
- ;;
- stop)
- [ "$VERBOSE" != no ] && log_daemon_msg "Stopping $DESC" "$NAME"
- do_stop
- case "$?" in
- 0|1) [ "$VERBOSE" != no ] && log_end_msg 0 ;;
- 2) [ "$VERBOSE" != no ] && log_end_msg 1 ;;
- esac
- ;;
- status)
- status_of_proc "$DAEMON" "$NAME" && exit 0 || exit $?
- ;;
-# reload)
-# #
-# # If do_reload() is not implemented then leave this commented out
-# # and leave 'force-reload' as an alias for 'restart'.
-# #
-# log_daemon_msg "Reloading $DESC" "$NAME"
-# do_reload
-# log_end_msg $?
-# ;;
- restart|force-reload)
- #
- # If the "reload" option is implemented then remove the
- # 'force-reload' alias
- #
- log_daemon_msg "Restarting $DESC" "$NAME"
- do_stop
- do_start
- ;;
- *)
- echo "Usage: $SCRIPTNAME {start|stop|status|restart|force-reload}" >&2
- exit 3
- ;;
-esac
-
-:
diff --git a/modules/log_processor/files/log-gearman-client.py b/modules/log_processor/files/log-gearman-client.py
deleted file mode 100644
index 03aa038b51..0000000000
--- a/modules/log_processor/files/log-gearman-client.py
+++ /dev/null
@@ -1,214 +0,0 @@
-#!/usr/bin/python2
-#
-# Copyright 2013 Hewlett-Packard Development Company, L.P.
-#
-# Licensed under the Apache License, Version 2.0 (the "License"); you may
-# not use this file except in compliance with the License. You may obtain
-# a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-# License for the specific language governing permissions and limitations
-# under the License.
-
-import argparse
-import daemon
-import gear
-import json
-import logging
-import os
-import os.path
-import re
-import signal
-import threading
-import yaml
-import zmq
-
-
-try:
- import daemon.pidlockfile as pidfile_mod
-except ImportError:
- import daemon.pidfile as pidfile_mod
-
-
-class EventProcessor(threading.Thread):
- def __init__(self, zmq_address, gearman_client, files, source_url):
- threading.Thread.__init__(self)
- self.files = files
- self.source_url = source_url
- self.gearman_client = gearman_client
- self.zmq_address = zmq_address
- self._connect_zmq()
-
- def run(self):
- while True:
- try:
- self._read_event()
- except:
- # Assume that an error reading data from zmq or deserializing
- # data received from zmq indicates a zmq error and reconnect.
- logging.exception("ZMQ exception.")
- self._connect_zmq()
-
- def _connect_zmq(self):
- logging.debug("Connecting to zmq endpoint.")
- self.context = zmq.Context()
- self.socket = self.context.socket(zmq.SUB)
- event_filter = b"onFinalized"
- self.socket.setsockopt(zmq.SUBSCRIBE, event_filter)
- self.socket.connect(self.zmq_address)
-
- def _read_event(self):
- string = self.socket.recv().decode('utf-8')
- event = json.loads(string.split(None, 1)[1])
- logging.debug("Jenkins event received: " + json.dumps(event))
- for fileopts in self.files:
- output = {}
- source_url, out_event = self._parse_event(event, fileopts)
- job_filter = fileopts.get('job-filter')
- if (job_filter and
- not re.match(job_filter, out_event['fields']['build_name'])):
- continue
- build_queue_filter = fileopts.get('build-queue-filter')
- if (build_queue_filter and
- not re.match(build_queue_filter,
- out_event['fields']['build_queue'])):
- continue
- output['source_url'] = source_url
- output['retry'] = fileopts.get('retry-get', False)
- 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'))
- try:
- self.gearman_client.submitJob(job)
- except:
- logging.exception("Exception submitting job to Gearman.")
-
- def _get_log_dir(self, event):
- parameters = event["build"].get("parameters", {})
- base = parameters.get('LOG_PATH', 'UNKNOWN')
- return base
-
- def _parse_fields(self, event, filename):
- fields = {}
- fields["filename"] = filename
- fields["build_name"] = event.get("name", "UNKNOWN")
- fields["build_status"] = event["build"].get("status", "UNKNOWN")
- fields["build_node"] = event["build"].get("node_name", "UNKNOWN")
- fields["build_master"] = event["build"].get("host_name", "UNKNOWN")
- parameters = event["build"].get("parameters", {})
- fields["project"] = parameters.get("ZUUL_PROJECT", "UNKNOWN")
- # TODO(clarkb) can we do better without duplicated data here?
- fields["build_uuid"] = parameters.get("ZUUL_UUID", "UNKNOWN")
- fields["build_short_uuid"] = fields["build_uuid"][:7]
- fields["build_queue"] = parameters.get("ZUUL_PIPELINE", "UNKNOWN")
- fields["build_ref"] = parameters.get("ZUUL_REF", "UNKNOWN")
- fields["build_branch"] = parameters.get("ZUUL_BRANCH", "UNKNOWN")
- if parameters.get("ZUUL_CHANGE"):
- fields["build_change"] = parameters.get("ZUUL_CHANGE", "UNKNOWN")
- fields["build_patchset"] = parameters.get("ZUUL_PATCHSET",
- "UNKNOWN")
- elif parameters.get("ZUUL_NEWREV"):
- fields["build_newrev"] = parameters.get("ZUUL_NEWREV",
- "UNKNOWN")
- return fields
-
- def _parse_event(self, event, fileopts):
- fields = self._parse_fields(event, fileopts['name'])
- log_dir = self._get_log_dir(event)
- source_url = fileopts.get('source-url', self.source_url) + '/' + \
- os.path.join(log_dir, fileopts['name'])
- fields["log_url"] = source_url
- out_event = {}
- out_event["fields"] = fields
- out_event["tags"] = [os.path.basename(fileopts['name'])] + \
- fileopts.get('tags', [])
- return source_url, out_event
-
-
-class Server(object):
- def __init__(self, config, debuglog):
- # Config init.
- self.config = config
- self.source_url = self.config['source-url']
- # Pythong logging output file.
- self.debuglog = debuglog
- self.processors = []
-
- def setup_logging(self):
- if self.debuglog:
- logging.basicConfig(format='%(asctime)s %(message)s',
- filename=self.debuglog, level=logging.DEBUG)
- else:
- # Prevent leakage into the logstash log stream.
- logging.basicConfig(level=logging.CRITICAL)
- logging.debug("Log pusher starting.")
-
- def setup_processors(self):
- for publisher in self.config['zmq-publishers']:
- gearclient = gear.Client()
- gearclient.addServer('localhost')
- gearclient.waitForServer()
- log_processor = EventProcessor(
- publisher, gearclient,
- self.config['source-files'], self.source_url)
- 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):
- statsd_host = os.environ.get('STATSD_HOST')
- statsd_port = int(os.environ.get('STATSD_PORT', 8125))
- statsd_prefix = os.environ.get('STATSD_PREFIX', 'logstash.geard')
- self.gearserver = gear.Server(
- statsd_host=statsd_host,
- statsd_port=statsd_port,
- statsd_prefix=statsd_prefix)
-
- self.setup_processors()
- for processor in self.processors:
- processor.daemon = True
- processor.start()
- while True:
- signal.pause()
-
-
-def main():
- parser = argparse.ArgumentParser()
- parser.add_argument("-c", "--config", required=True,
- help="Path to yaml config file.")
- parser.add_argument("-d", "--debuglog",
- help="Enable debug log. "
- "Specifies file to write log to.")
- parser.add_argument("--foreground", action='store_true',
- help="Run in the foreground.")
- parser.add_argument("-p", "--pidfile",
- default="/var/run/jenkins-log-pusher/"
- "jenkins-log-gearman-client.pid",
- help="PID file to lock during daemonization.")
- args = parser.parse_args()
-
- with open(args.config, 'r') as config_stream:
- config = yaml.load(config_stream)
- server = Server(config, args.debuglog)
-
- if args.foreground:
- server.setup_logging()
- server.main()
- else:
- pidfile = pidfile_mod.TimeoutPIDLockFile(args.pidfile, 10)
- with daemon.DaemonContext(pidfile=pidfile):
- server.setup_logging()
- server.main()
-
-
-if __name__ == '__main__':
- main()
diff --git a/modules/log_processor/files/log-gearman-worker.py b/modules/log_processor/files/log-gearman-worker.py
deleted file mode 100644
index 66007cb770..0000000000
--- a/modules/log_processor/files/log-gearman-worker.py
+++ /dev/null
@@ -1,430 +0,0 @@
-#!/usr/bin/python2
-#
-# Copyright 2013 Hewlett-Packard Development Company, L.P.
-#
-# Licensed under the Apache License, Version 2.0 (the "License"); you may
-# not use this file except in compliance with the License. You may obtain
-# a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-# License for the specific language governing permissions and limitations
-# under the License.
-
-import argparse
-import cStringIO
-import daemon
-import gear
-import gzip
-import json
-import logging
-import os
-import Queue
-import re
-import select
-import socket
-import subprocess
-import sys
-import threading
-import time
-import urllib2
-import yaml
-
-
-try:
- import daemon.pidlockfile as pidfile_mod
-except ImportError:
- import daemon.pidfile as pidfile_mod
-
-
-def semi_busy_wait(seconds):
- # time.sleep() may return early. If it does sleep() again and repeat
- # until at least the number of seconds specified has elapsed.
- start_time = time.time()
- while True:
- time.sleep(seconds)
- cur_time = time.time()
- seconds = seconds - (cur_time - start_time)
- if seconds <= 0.0:
- return
-
-
-class FilterException(Exception):
- pass
-
-
-class CRM114Filter(object):
- def __init__(self, script, path, build_status):
- self.p = None
- self.script = script
- self.path = path
- self.build_status = build_status
- if build_status not in ['SUCCESS', 'FAILURE']:
- return
- if not os.path.exists(path):
- os.makedirs(path)
- args = [script, path, build_status]
- self.p = subprocess.Popen(args,
- stdout=subprocess.PIPE,
- stderr=subprocess.PIPE,
- stdin=subprocess.PIPE,
- close_fds=True)
-
- def process(self, data):
- if not self.p:
- return
- self.p.stdin.write(data['message'].encode('utf-8') + '\n')
- (r, w, x) = select.select([self.p.stdout], [],
- [self.p.stdin, self.p.stdout], 20)
- if not r:
- self.p.kill()
- raise FilterException('Timeout reading from CRM114')
- r = self.p.stdout.readline()
- if not r:
- err = self.p.stderr.read()
- if err:
- raise FilterException(err)
- else:
- raise FilterException('Early EOF from CRM114')
- r = r.strip()
- data['error_pr'] = float(r)
-
- def _catchOSError(self, method):
- try:
- method()
- except OSError:
- logging.exception("Subprocess cleanup failed.")
-
- def close(self):
- if not self.p:
- return
- # CRM114 should die when its stdinput is closed. Close that
- # fd along with stdout and stderr then return.
- self._catchOSError(self.p.stdin.close)
- self._catchOSError(self.p.stdout.close)
- self._catchOSError(self.p.stderr.close)
- self._catchOSError(self.p.wait)
-
-
-class CRM114FilterFactory(object):
- name = "CRM114"
-
- def __init__(self, script, basepath):
- self.script = script
- self.basepath = basepath
-
- def create(self, fields):
- filename = re.sub('\.', '_', fields['filename'])
- path = os.path.join(self.basepath, filename)
- return CRM114Filter(self.script, path, fields['build_status'])
-
-
-class LogRetriever(threading.Thread):
- def __init__(self, gearman_worker, filters, logq):
- threading.Thread.__init__(self)
- self.gearman_worker = gearman_worker
- self.filters = filters
- self.logq = logq
-
- def run(self):
- while True:
- try:
- self._handle_event()
- except:
- logging.exception("Exception retrieving log event.")
-
- def _handle_event(self):
- job = self.gearman_worker.getJob()
- try:
- arguments = json.loads(job.arguments.decode('utf-8'))
- source_url = arguments['source_url']
- retry = arguments['retry']
- event = arguments['event']
- logging.debug("Handling event: " + json.dumps(event))
- fields = event.get('fields') or event.get('@fields')
- tags = event.get('tags') or event.get('@tags')
- if fields['build_status'] != 'ABORTED':
- # Handle events ignoring aborted builds. These builds are
- # discarded by zuul.
- log_lines = self._retrieve_log(source_url, retry)
-
- try:
- all_filters = []
- for f in self.filters:
- logging.debug("Adding filter: %s" % f.name)
- all_filters.append(f.create(fields))
- filters = all_filters
-
- logging.debug("Pushing " + str(len(log_lines)) +
- " log lines.")
- base_event = {}
- base_event.update(fields)
- base_event["tags"] = tags
- for line in log_lines:
- out_event = base_event.copy()
- out_event["message"] = line
- new_filters = []
- for f in filters:
- try:
- f.process(out_event)
- new_filters.append(f)
- except FilterException:
- logging.exception("Exception filtering event: "
- "%s" % line.encode("utf-8"))
- filters = new_filters
- self.logq.put(out_event)
- finally:
- for f in all_filters:
- f.close()
- job.sendWorkComplete()
- except Exception as e:
- logging.exception("Exception handling log event.")
- job.sendWorkException(str(e).encode('utf-8'))
-
- def _retrieve_log(self, source_url, retry):
- # TODO (clarkb): This should check the content type instead of file
- # extension for determining if gzip was used.
- gzipped = False
- raw_buf = b''
- try:
- gzipped, raw_buf = self._get_log_data(source_url, retry)
- except urllib2.HTTPError as e:
- if e.code == 404:
- logging.info("Unable to retrieve %s: HTTP error 404" %
- source_url)
- else:
- logging.exception("Unable to get log data.")
- except Exception:
- # Silently drop fatal errors when retrieving logs.
- # TODO (clarkb): Handle these errors.
- # Perhaps simply add a log message to raw_buf?
- logging.exception("Unable to get log data.")
- if gzipped:
- logging.debug("Decompressing gzipped source file.")
- raw_strIO = cStringIO.StringIO(raw_buf)
- f = gzip.GzipFile(fileobj=raw_strIO)
- buf = f.read().decode('utf-8')
- raw_strIO.close()
- f.close()
- else:
- logging.debug("Decoding source file.")
- buf = raw_buf.decode('utf-8')
- return buf.splitlines()
-
- def _get_log_data(self, source_url, retry):
- gzipped = False
- try:
- # TODO(clarkb): We really should be using requests instead
- # of urllib2. urllib2 will automatically perform a POST
- # instead of a GET if we provide urlencoded data to urlopen
- # but we need to do a GET. The parameters are currently
- # hardcoded so this should be ok for now.
- logging.debug("Retrieving: " + source_url + ".gz?level=INFO")
- req = urllib2.Request(source_url + ".gz?level=INFO")
- req.add_header('Accept-encoding', 'gzip')
- r = urllib2.urlopen(req)
- except urllib2.URLError:
- try:
- # Fallback on GETting unzipped data.
- logging.debug("Retrieving: " + source_url + "?level=INFO")
- r = urllib2.urlopen(source_url + "?level=INFO")
- except:
- logging.exception("Unable to retrieve source file.")
- raise
- except:
- logging.exception("Unable to retrieve source file.")
- raise
- if ('gzip' in r.info().get('Content-Type', '') or
- 'gzip' in r.info().get('Content-Encoding', '')):
- gzipped = True
-
- raw_buf = r.read()
- # Hack to read all of Jenkins console logs as they upload
- # asynchronously. After each attempt do an exponential backup before
- # the next request for up to 255 seconds total, if we do not
- # retrieve the entire file. Short circuit when the end of file string
- # for console logs, '\n</pre>\n', is read.
- if (retry and not gzipped and
- raw_buf[-8:].decode('utf-8') != '\n</pre>\n'):
- content_len = len(raw_buf)
- backoff = 1
- while backoff < 129:
- # Try for up to 255 seconds to retrieve the complete log file.
- try:
- logging.debug(str(backoff) + " Retrying fetch of: " +
- source_url + "?level=INFO")
- logging.debug("Fetching bytes=" + str(content_len) + '-')
- req = urllib2.Request(source_url + "?level=INFO")
- req.add_header('Range', 'bytes=' + str(content_len) + '-')
- r = urllib2.urlopen(req)
- raw_buf += r.read()
- content_len = len(raw_buf)
- except urllib2.HTTPError as e:
- if e.code == 416:
- logging.exception("Index out of range.")
- else:
- raise
- finally:
- if raw_buf[-8:].decode('utf-8') == '\n</pre>\n':
- break
- semi_busy_wait(backoff)
- backoff += backoff
-
- return gzipped, raw_buf
-
-
-class StdOutLogProcessor(object):
- def __init__(self, logq, pretty_print=False):
- self.logq = logq
- self.pretty_print = pretty_print
-
- def handle_log_event(self):
- log = self.logq.get()
- if self.pretty_print:
- print(json.dumps(log, sort_keys=True,
- indent=4, separators=(',', ': ')))
- else:
- print(json.dumps(log))
- # Push each log event through to keep logstash up to date.
- sys.stdout.flush()
-
-
-class INETLogProcessor(object):
- socket_type = None
-
- def __init__(self, logq, host, port):
- self.logq = logq
- self.host = host
- self.port = port
- self._connect_socket()
-
- def _connect_socket(self):
- logging.debug("Creating socket.")
- self.socket = socket.socket(socket.AF_INET, self.socket_type)
- self.socket.connect((self.host, self.port))
-
- def handle_log_event(self):
- log = self.logq.get()
- try:
- self.socket.sendall((json.dumps(log) + '\n').encode('utf-8'))
- except:
- logging.exception("Exception sending INET event.")
- # Logstash seems to take about a minute to start again. Wait 90
- # seconds before attempting to reconnect. If logstash is not
- # available after 90 seconds we will throw another exception and
- # die.
- semi_busy_wait(90)
- self._connect_socket()
- self.socket.sendall((json.dumps(log) + '\n').encode('utf-8'))
-
-
-class UDPLogProcessor(INETLogProcessor):
- socket_type = socket.SOCK_DGRAM
-
-
-class TCPLogProcessor(INETLogProcessor):
- socket_type = socket.SOCK_STREAM
-
-
-class Server(object):
- def __init__(self, config, debuglog):
- # Config init.
- self.config = config
- self.gearman_host = self.config['gearman-host']
- self.gearman_port = self.config['gearman-port']
- self.output_host = self.config['output-host']
- self.output_port = self.config['output-port']
- self.output_mode = self.config['output-mode']
- # Pythong logging output file.
- self.debuglog = debuglog
- self.retriever = None
- self.logqueue = Queue.Queue(131072)
- self.processor = None
- self.filter_factories = []
- crmscript = self.config.get('crm114-script')
- crmdata = self.config.get('crm114-data')
- if crmscript and crmdata:
- self.filter_factories.append(
- CRM114FilterFactory(crmscript, crmdata))
-
- def setup_logging(self):
- if self.debuglog:
- logging.basicConfig(format='%(asctime)s %(message)s',
- filename=self.debuglog, level=logging.DEBUG)
- else:
- # Prevent leakage into the logstash log stream.
- logging.basicConfig(level=logging.CRITICAL)
- logging.debug("Log pusher starting.")
-
- def setup_retriever(self):
- hostname = socket.gethostname()
- gearman_worker = gear.Worker(hostname + b'-pusher')
- gearman_worker.addServer(self.gearman_host,
- self.gearman_port)
- gearman_worker.registerFunction(b'push-log')
- self.retriever = LogRetriever(gearman_worker, self.filter_factories,
- self.logqueue)
-
- def setup_processor(self):
- if self.output_mode == "tcp":
- self.processor = TCPLogProcessor(self.logqueue,
- self.output_host,
- self.output_port)
- elif self.output_mode == "udp":
- self.processor = UDPLogProcessor(self.logqueue,
- self.output_host,
- self.output_port)
- else:
- # Note this processor will not work if the process is run as a
- # daemon. You must use the --foreground option.
- self.processor = StdOutLogProcessor(self.logqueue)
-
- def main(self):
- self.setup_retriever()
- self.setup_processor()
-
- self.retriever.daemon = True
- self.retriever.start()
-
- while True:
- try:
- self.processor.handle_log_event()
- except:
- logging.exception("Exception processing log event.")
- raise
-
-
-def main():
- parser = argparse.ArgumentParser()
- parser.add_argument("-c", "--config", required=True,
- help="Path to yaml config file.")
- parser.add_argument("-d", "--debuglog",
- help="Enable debug log. "
- "Specifies file to write log to.")
- parser.add_argument("--foreground", action='store_true',
- help="Run in the foreground.")
- parser.add_argument("-p", "--pidfile",
- default="/var/run/jenkins-log-pusher/"
- "jenkins-log-gearman-worker.pid",
- help="PID file to lock during daemonization.")
- args = parser.parse_args()
-
- with open(args.config, 'r') as config_stream:
- config = yaml.load(config_stream)
- server = Server(config, args.debuglog)
-
- if args.foreground:
- server.setup_logging()
- server.main()
- else:
- pidfile = pidfile_mod.TimeoutPIDLockFile(args.pidfile, 10)
- with daemon.DaemonContext(pidfile=pidfile):
- server.setup_logging()
- server.main()
-
-
-if __name__ == '__main__':
- main()
diff --git a/modules/log_processor/manifests/client.pp b/modules/log_processor/manifests/client.pp
deleted file mode 100644
index 8081e57153..0000000000
--- a/modules/log_processor/manifests/client.pp
+++ /dev/null
@@ -1,73 +0,0 @@
-# Copyright 2012-2013 Hewlett-Packard Development Company, L.P.
-# Copyright 2013 OpenStack Foundation
-#
-# Licensed under the Apache License, Version 2.0 (the "License"); you may
-# not use this file except in compliance with the License. You may obtain
-# a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-# License for the specific language governing permissions and limitations
-# under the License.
-
-# == Class: log_processor::client
-#
-class log_processor::client (
- $config_file,
- $statsd_host = '',
-) {
-
- file { '/etc/logstash/jenkins-log-client.yaml':
- ensure => present,
- owner => 'root',
- group => 'root',
- mode => '0555',
- source => $config_file,
- require => File['/etc/logstash'],
- }
-
- file { '/etc/init.d/jenkins-log-client':
- ensure => present,
- owner => 'root',
- group => 'root',
- mode => '0555',
- source => 'puppet:///modules/log_processor/jenkins-log-client.init',
- require => [
- File['/usr/local/bin/log-gearman-client.py'],
- File['/etc/logstash/jenkins-log-client.yaml'],
- File['/etc/default/jenkins-log-client'],
- ],
- }
-
- file { '/etc/default/jenkins-log-client':
- ensure => present,
- owner => 'root',
- group => 'root',
- mode => '0444',
- content => template('log_processor/jenkins-log-client.default.erb'),
- }
-
- service { 'jenkins-log-client':
- enable => true,
- hasrestart => true,
- subscribe => File['/etc/logstash/jenkins-log-client.yaml'],
- require => File['/etc/init.d/jenkins-log-client'],
- }
-
- include logrotate
- logrotate::file { 'log-client-debug.log':
- log => '/var/log/logstash/log-client-debug.log',
- options => [
- 'compress',
- 'copytruncate',
- 'missingok',
- 'rotate 7',
- 'daily',
- 'notifempty',
- ],
- require => Service['jenkins-log-client'],
- }
-}
diff --git a/modules/log_processor/manifests/init.pp b/modules/log_processor/manifests/init.pp
deleted file mode 100644
index 661ca0ed6d..0000000000
--- a/modules/log_processor/manifests/init.pp
+++ /dev/null
@@ -1,94 +0,0 @@
-# Copyright 2012-2013 Hewlett-Packard Development Company, L.P.
-# Copyright 2013 OpenStack Foundation
-#
-# Licensed under the Apache License, Version 2.0 (the "License"); you may
-# not use this file except in compliance with the License. You may obtain
-# a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-# License for the specific language governing permissions and limitations
-# under the License.
-
-# == Class: log_processor
-#
-class log_processor (
-) {
- package { 'python-daemon':
- ensure => present,
- }
-
- package { 'python-zmq':
- ensure => present,
- }
-
- package { 'python-yaml':
- ensure => present,
- }
-
- package { 'crm114':
- ensure => present,
- }
-
- include pip
- package { 'gear':
- ensure => latest,
- provider => 'pip',
- require => Class['pip'],
- }
-
- package { 'statsd':
- ensure => latest,
- provider => 'pip',
- require => Class['pip'],
- }
-
- file { '/var/lib/crm114':
- ensure => directory,
- owner => 'logstash',
- group => 'logstash',
- require => User['logstash'],
- }
-
- file { '/usr/local/bin/classify-log.crm':
- ensure => present,
- owner => 'root',
- group => 'root',
- mode => '0755',
- source => 'puppet:///modules/log_processor/classify-log.crm',
- require => [
- Package['crm114'],
- ],
- }
-
- file { '/usr/local/bin/log-gearman-client.py':
- ensure => present,
- owner => 'root',
- group => 'root',
- mode => '0755',
- source => 'puppet:///modules/log_processor/log-gearman-client.py',
- require => [
- Package['python-daemon'],
- Package['python-zmq'],
- Package['python-yaml'],
- Package['gear'],
- ],
- }
-
- file { '/usr/local/bin/log-gearman-worker.py':
- ensure => present,
- owner => 'root',
- group => 'root',
- mode => '0755',
- source => 'puppet:///modules/log_processor/log-gearman-worker.py',
- require => [
- Package['python-daemon'],
- Package['python-zmq'],
- Package['python-yaml'],
- Package['gear'],
- ],
- }
-}
diff --git a/modules/log_processor/manifests/worker.pp b/modules/log_processor/manifests/worker.pp
deleted file mode 100644
index ec5653bdac..0000000000
--- a/modules/log_processor/manifests/worker.pp
+++ /dev/null
@@ -1,67 +0,0 @@
-# Copyright 2012-2013 Hewlett-Packard Development Company, L.P.
-# Copyright 2013 OpenStack Foundation
-#
-# Licensed under the Apache License, Version 2.0 (the "License"); you may
-# not use this file except in compliance with the License. You may obtain
-# a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-# License for the specific language governing permissions and limitations
-# under the License.
-
-# == Class: log_processor::worker
-#
-define log_processor::worker (
- $config_file,
-) {
- $suffix = "-${name}"
-
- file { "/etc/logstash/jenkins-log-worker${suffix}.yaml":
- ensure => present,
- owner => 'root',
- group => 'root',
- mode => '0555',
- source => $config_file,
- require => Class['logstash::indexer'],
- }
-
- file { "/etc/init.d/jenkins-log-worker${suffix}":
- ensure => present,
- owner => 'root',
- group => 'root',
- mode => '0555',
- content => template('log_processor/jenkins-log-worker.init.erb'),
- require => [
- File['/usr/local/bin/log-gearman-worker.py'],
- File["/etc/logstash/jenkins-log-worker${suffix}.yaml"],
- ],
- }
-
- service { "jenkins-log-worker${suffix}":
- enable => true,
- hasrestart => true,
- subscribe => File["/etc/logstash/jenkins-log-worker${suffix}.yaml"],
- require => [
- Class['logstash::indexer'],
- File["/etc/init.d/jenkins-log-worker${suffix}"],
- ],
- }
-
- include logrotate
- logrotate::file { "log-worker${suffix}-debug.log":
- log => "/var/log/logstash/log-worker${suffix}-debug.log",
- options => [
- 'compress',
- 'copytruncate',
- 'missingok',
- 'rotate 7',
- 'daily',
- 'notifempty',
- ],
- require => Service["jenkins-log-worker${suffix}"],
- }
-}
diff --git a/modules/log_processor/templates/jenkins-log-client.default.erb b/modules/log_processor/templates/jenkins-log-client.default.erb
deleted file mode 100644
index a048fad87d..0000000000
--- a/modules/log_processor/templates/jenkins-log-client.default.erb
+++ /dev/null
@@ -1,5 +0,0 @@
-<% if scope.lookupvar("log_processor::client::statsd_host") != "" %>
-export STATSD_HOST=<%= scope.lookupvar("log_processor::client::statsd_host") %>
-export STATSD_PORT=8125
-export STATSD_PREFIX="logstash.geard"
-<% end %>
diff --git a/modules/log_processor/templates/jenkins-log-worker.init.erb b/modules/log_processor/templates/jenkins-log-worker.init.erb
deleted file mode 100755
index 152a97b1f8..0000000000
--- a/modules/log_processor/templates/jenkins-log-worker.init.erb
+++ /dev/null
@@ -1,158 +0,0 @@
-#! /bin/sh
-### BEGIN INIT INFO
-# Provides: jenkins-log-worker
-# Required-Start: $remote_fs $syslog
-# Required-Stop: $remote_fs $syslog
-# Default-Start: 2 3 4 5
-# Default-Stop: 0 1 6
-# Short-Description: Jenkins Log Worker
-# Description: Service to push Jenkins logs into logstash.
-### END INIT INFO
-
-# Do NOT "set -e"
-
-# PATH should only include /usr/* if it runs after the mountnfs.sh script
-PATH=/sbin:/usr/sbin:/bin:/usr/bin
-DESC="Jenkins Log Worker"
-NAME=jenkins-log-worker<%= suffix %>
-DAEMON=/usr/local/bin/log-gearman-worker.py
-PIDFILE=/var/run/$NAME/$NAME.pid
-DAEMON_ARGS="-c /etc/logstash/jenkins-log-worker<%= suffix %>.yaml -d /var/log/logstash/log-worker<%= suffix %>-debug.log -p $PIDFILE"
-SCRIPTNAME=/etc/init.d/$NAME
-USER=logstash
-
-# Exit if the package is not installed
-[ -x "$DAEMON" ] || exit 0
-
-# Read configuration variable file if it is present
-[ -r /etc/default/$NAME ] && . /etc/default/$NAME
-
-# Load the VERBOSE setting and other rcS variables
-. /lib/init/vars.sh
-
-# Define LSB log_* functions.
-# Depend on lsb-base (>= 3.0-6) to ensure that this file is present.
-. /lib/lsb/init-functions
-
-#
-# Function that starts the daemon/service
-#
-do_start()
-{
- # Return
- # 0 if daemon has been started
- # 1 if daemon was already running
- # 2 if daemon could not be started
-
- mkdir -p /var/run/$NAME
- chown $USER /var/run/$NAME
- start-stop-daemon --start --quiet --pidfile $PIDFILE -c $USER --exec $DAEMON --test > /dev/null \
- || return 1
- start-stop-daemon --start --quiet --pidfile $PIDFILE -c $USER --exec $DAEMON -- \
- $DAEMON_ARGS \
- || return 2
- # Add code here, if necessary, that waits for the process to be ready
- # to handle requests from services started subsequently which depend
- # on this one. As a last resort, sleep for some time.
-}
-
-#
-# Function that stops the daemon/service
-#
-do_stop()
-{
- # Return
- # 0 if daemon has been stopped
- # 1 if daemon was already stopped
- # 2 if daemon could not be stopped
- # other if a failure occurred
- start-stop-daemon --stop --signal 9 --pidfile $PIDFILE
- RETVAL="$?"
- [ "$RETVAL" = 2 ] && return 2
- rm -f /var/run/$NAME/*
- return "$RETVAL"
-}
-
-#
-# Function that stops the daemon/service
-#
-#do_graceful_stop()
-#{
-# PID=`cat $PIDFILE`
-# kill -USR1 $PID
-#
-# # wait until really stopped
-# if [ -n "${PID:-}" ]; then
-# i=0
-# while kill -0 "${PID:-}" 2> /dev/null; do
-# if [ $i -eq '0' ]; then
-# echo -n " ... waiting "
-# else
-# echo -n "."
-# fi
-# i=$(($i+1))
-# sleep 1
-# done
-# fi
-#
-# rm -f /var/run/$NAME/*
-#}
-
-#
-# Function that sends a SIGHUP to the daemon/service
-#
-#do_reload() {
-# #
-# # If the daemon can reload its configuration without
-# # restarting (for example, when it is sent a SIGHUP),
-# # then implement that here.
-# #
-# start-stop-daemon --stop --signal 1 --quiet --pidfile $PIDFILE --name zuul-server
-# return 0
-#}
-
-case "$1" in
- start)
- [ "$VERBOSE" != no ] && log_daemon_msg "Starting $DESC" "$NAME"
- do_start
- case "$?" in
- 0|1) [ "$VERBOSE" != no ] && log_end_msg 0 ;;
- 2) [ "$VERBOSE" != no ] && log_end_msg 1 ;;
- esac
- ;;
- stop)
- [ "$VERBOSE" != no ] && log_daemon_msg "Stopping $DESC" "$NAME"
- do_stop
- case "$?" in
- 0|1) [ "$VERBOSE" != no ] && log_end_msg 0 ;;
- 2) [ "$VERBOSE" != no ] && log_end_msg 1 ;;
- esac
- ;;
- status)
- status_of_proc "$DAEMON" "$NAME" && exit 0 || exit $?
- ;;
-# reload)
-# #
-# # If do_reload() is not implemented then leave this commented out
-# # and leave 'force-reload' as an alias for 'restart'.
-# #
-# log_daemon_msg "Reloading $DESC" "$NAME"
-# do_reload
-# log_end_msg $?
-# ;;
- restart|force-reload)
- #
- # If the "reload" option is implemented then remove the
- # 'force-reload' alias
- #
- log_daemon_msg "Restarting $DESC" "$NAME"
- do_stop
- do_start
- ;;
- *)
- echo "Usage: $SCRIPTNAME {start|stop|status|restart|force-reload}" >&2
- exit 3
- ;;
-esac
-
-: