Split out log_processor
Change-Id: I414e5dadc53b52d8e1c4350b785187951c83a36b
This commit is contained in:
Ramy Asselin
Joshua Hesketh
parent
4763b3e4e4
commit
2ff4f7957b
@ -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"
|
||||
|
||||
@ -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
|
||||
}
|
||||
@ -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
|
||||
|
||||
:
|
||||
@ -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()
|
||||
@ -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()
|
||||
@ -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'],
|
||||
}
|
||||
}
|
||||
@ -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'],
|
||||
],
|
||||
}
|
||||
}
|
||||
@ -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}"],
|
||||
}
|
||||
}
|
||||
@ -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 %>
|
||||
@ -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
|
||||
|
||||
:
|
||||
Loading…
x
Reference in New Issue
Block a user