diff --git a/Makefile b/Makefile index e756dd4f3dff..6126e859309b 100644 --- a/Makefile +++ b/Makefile @@ -2,8 +2,8 @@ NAME=graylog2-server PREFIX=/usr DESTDIR= -SERVER_W_DEP=target/graylog2-server-0.9.6-jar-with-dependencies.jar -SERVER=target/graylog2-server-0.9.6.jar +SERVER_W_DEP=target/graylog2-server-0.9.6p1-RC2-jar-with-dependencies.jar +SERVER=target/graylog2-server-0.9.6p1-RC2.jar SYSLOG4J=lib/syslog4j-0.9.46-bin.jar INITD=contrib/distro/generic/graylog2-server.init.d CONF=misc/graylog2.conf diff --git a/contrib/distro/commons-daemon-redhat/SOURCES/graylog2-elasticsearch.yml b/contrib/distro/commons-daemon-redhat/SOURCES/graylog2-elasticsearch.yml new file mode 100644 index 000000000000..40215ac8f245 --- /dev/null +++ b/contrib/distro/commons-daemon-redhat/SOURCES/graylog2-elasticsearch.yml @@ -0,0 +1,65 @@ +# this must be the same as for your elasticsearch cluster +cluster.name: graylog2 + +# you could also leave this out, but makes it easier to identify the graylog2 client instance +node.name: "graylog2-server" + +# we don't want the graylog2 client to store any data, or be master node +node.master: false +node.data: false + +# you might need to bind to a certain IP address, do that here +#network.host: 172.24.0.14 +# use a different port if you run multiple elasticsearch nodes on one machine +#transport.tcp.port: 9350 + +# we don't need to run the embedded HTTP server here +http.enabled: false + +# adapt these for discovery to work in your network! multicast can be tricky +#discovery.zen.ping.multicast.address: 172.24.0.14 +#discovery.zen.ping.multicast.group: 224.0.0.1 + + +################################## Discovery ################################## + +# Discovery infrastructure ensures nodes can be found within a cluster +# and master node is elected. Multicast discovery is the default. + +# Set to ensure a node sees N other master eligible nodes to be considered +# operational within the cluster. Set this option to a higher value (2-4) +# for large clusters (>3 nodes): +# +# discovery.zen.minimum_master_nodes: 1 + +# Set the time to wait for ping responses from other nodes when discovering. +# Set this option to a higher value on a slow or congested network +# to minimize discovery failures: +# +# discovery.zen.ping.timeout: 3s + +# See +# for more information. + +# Unicast discovery allows to explicitly control which nodes will be used +# to discover the cluster. It can be used when multicast is not present, +# or to restrict the cluster communication-wise. +# +# 1. Disable multicast discovery (enabled by default): +# +# discovery.zen.ping.multicast.enabled: false +# +# 2. Configure an initial list of master nodes in the cluster +# to perform discovery when new nodes (master or data) are started: +# +# discovery.zen.ping.unicast.hosts: ["host1", "host2:port", "host3[portX-portY]"] + +# EC2 discovery allows to use AWS EC2 API in order to perform discovery. +# +# You have to install the cloud-aws plugin for enabling the EC2 discovery. +# +# See +# for more information. +# +# See +# for a step-by-step tutorial. diff --git a/contrib/distro/commons-daemon-redhat/SOURCES/graylog2-log4j.xml b/contrib/distro/commons-daemon-redhat/SOURCES/graylog2-log4j.xml new file mode 100644 index 000000000000..89ef57af1e49 --- /dev/null +++ b/contrib/distro/commons-daemon-redhat/SOURCES/graylog2-log4j.xml @@ -0,0 +1,24 @@ + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/contrib/distro/commons-daemon-redhat/SOURCES/graylog2.conf b/contrib/distro/commons-daemon-redhat/SOURCES/graylog2.conf new file mode 100644 index 000000000000..563a201a7ffc --- /dev/null +++ b/contrib/distro/commons-daemon-redhat/SOURCES/graylog2.conf @@ -0,0 +1,167 @@ +# If you are running more than one instances of graylog2-server you have to select one of these +# instances as master. The master will perform some periodical tasks that non-masters won't perform. +is_master = true + +# Set plugin directory here (relative or absolute) +plugin_dir = plugin + +# On which port (UDP) should we listen for Syslog messages? (Standard: 514) +syslog_listen_port = 514 +syslog_listen_address = 0.0.0.0 +syslog_enable_udp = true +syslog_enable_tcp = false +# Standard delimiter is LF. You can force using a NUL byte delimiter using this option. +syslog_use_nul_delimiter = false +# The raw syslog message is stored as full_message of the message if not disabled here. +syslog_store_full_message = true + +# Socket receive buffer size (bytes) for UDP syslog and UDP GELF. +udp_recvbuffer_sizes = 1048576 + +# Embedded elasticsearch configuration file +# pay attention to the working directory of the server, maybe use an absolute path here +elasticsearch_config_file = /etc/graylog2-server/elasticsearch.yml +elasticsearch_max_docs_per_index = 20000000 + +elasticsearch_index_prefix = graylog2 + +# How many indices do you want to keep? If the number of indices exceeds this number, older indices will be dropped. +# elasticsearch_max_number_of_indices*elasticsearch_max_docs_per_index=total number of messages in your setup +elasticsearch_max_number_of_indices = 20 + +# How many ElasticSearch shards and replicas should be used per index? Note that this only applies to newly created indices. +elasticsearch_shards = 4 +elasticsearch_replicas = 0 + +# Analyzer (tokenizer) to use for message and full_message field. The "standard" filter usually is a good idea. +# All supported analyzers are: standard, simple, whitespace, stop, keyword, pattern, language, snowball, custom +# ElasticSearch documentation: http://www.elasticsearch.org/guide/reference/index-modules/analysis/ +# Note that this setting only takes effect on newly created indices. +elasticsearch_analyzer = standard + +# How many minutes of messages do you want to keep in the recent index? This index lives in memory only and is used to build the overview and stream pages. Raise this value if you want to see more messages in the overview pages. This is not affecting for example searches which are always targeting *all* indices. +recent_index_ttl_minutes = 60 + +# Storage type of recent index. Allowed values: niofs, simplefs, mmapfs, memory +# Standard: niofs - Set to memory for best speed but keep in mind that the whole recent index has to fit into the memory of your ElasticSearch machines. Set recent_index_ttl_minutes to a reasonable amount that will let the messages fit into memory. +recent_index_store_type = niofs + +# Always try a reverse DNS lookup instead of parsing hostname from syslog message? +force_syslog_rdns = false +# Set time to NOW if parsing date/time from syslog message failed instead of rejecting it? +allow_override_syslog_date = true + +# Batch size for all outputs. This is the maximum (!) number of messages an output module will get at once. +# For example, if this is set to 5000 (default), the ElasticSearch Output will not index more than 5000 messages +# at once. After that index operation is performed, the next batch will be indexed. If there is only 1 message +# waiting, it will only index that single message. It is important to raise this parameter if you send in so +# many messages that it is not enough to index 5000 messages at once. (Only at *really* high message rates) +output_batch_size = 5000 + +# The number of parallel running processors. +# Raise this number if your buffers are filling up. +processbuffer_processors = 5 +outputbuffer_processors = 5 + +# Wait strategy describing how buffer processors wait on a cursor sequence. (default: sleeping) +# Possible types: +# - yielding +# Compromise between performance and CPU usage. +# - sleeping +# Compromise between performance and CPU usage. Latency spikes can occur after quiet periods. +# - blocking +# High throughput, low latency, higher CPU usage. +# - busy_spinning +# Avoids syscalls which could introduce latency jitter. Best when threads can be bound to specific CPU cores. +processor_wait_strategy = sleeping + +# Size of internal ring buffers. Raise this if raising outputbuffer_processors does not help anymore. +# For optimum performance your LogMessage objects in the ring buffer should fit in your CPU L3 cache. +# Start server with --statistics flag to see buffer utilization. +# Must be a power of 2. (512, 1024, 2048, ...) +ring_size = 1024 + +# MongoDB Configuration +mongodb_useauth = true +mongodb_user = grayloguser +mongodb_password = 123 +mongodb_host = 127.0.0.1 +#mongodb_replica_set = localhost:27017,localhost:27018,localhost:27019 +mongodb_database = graylog2 +mongodb_port = 27017 + +# Raise this according to the maximum connections your MongoDB server can handle if you encounter MongoDB connection problems. +mongodb_max_connections = 100 + +# Number of threads allowed to be blocked by MongoDB connections multiplier. Default: 5 +# If mongodb_max_connections is 100, and mongodb_threads_allowed_to_block_multiplier is 5, then 500 threads can block. More than that and an exception will be thrown. +# http://api.mongodb.org/java/current/com/mongodb/MongoOptions.html#threadsAllowedToBlockForConnectionMultiplier +mongodb_threads_allowed_to_block_multiplier = 5 + +# Graylog Extended Log Format (GELF) +use_gelf = true +gelf_listen_address = 0.0.0.0 +gelf_listen_port = 12201 + +# Drools Rule File (Use to rewrite incoming log messages) +# rules_file = /etc/graylog2.d/rules/graylog2.drl + +# AMQP +amqp_enabled = false +amqp_host = localhost +amqp_port = 5672 +amqp_username = guest +amqp_password = guest +amqp_virtualhost = / + +# HTTP input +# the server will accept PUT requests to /gelf or /gelf/raw +# /gelf can process all standard GELF messages containing the two header bytes +# /gelf/raw can only process uncompressed GELF messages without any header bytes. +# the HTTP server allows keep-alive connections and supports compression. +http_enabled = false +http_listen_address = 0.0.0.0 +http_listen_port = 12202 + +# Email transport +transport_email_enabled = false +transport_email_hostname = mail.example.com +transport_email_port = 587 +transport_email_use_auth = true +transport_email_use_tls = true +transport_email_auth_username = you@example.com +transport_email_auth_password = secret +transport_email_subject_prefix = [graylog2] +transport_email_from_email = graylog2@example.com +transport_email_from_name = Graylog2 + +# Jabber/XMPP transport +transport_jabber_enabled = false +transport_jabber_hostname = jabber.example.com +transport_jabber_port = 5222 +transport_jabber_use_sasl_auth = true +transport_jabber_allow_selfsigned_certs = false +transport_jabber_auth_username = your_user +transport_jabber_auth_password = secret +transport_jabber_message_prefix = [graylog2] + +# Filters +# Enable the filter that tries to extract additional fields from k=v values in the log message? +enable_tokenizer_filter = true + +# Additional modules +# Graphite +#enable_graphite_output = false +#graphite_carbon_host = 127.0.0.1 +#graphite_carbon_tcp_port = 2003 +#graphite_prefix = logs + +# Librato Metrics (http://support.torch.sh/help/kb/graylog2-server/using-librato-metrics-with-graylog2) +#enable_libratometrics_output = false +#enable_libratometrics_system_metrics = false +#libratometrics_api_user = you@example.com +#libratometrics_api_token = abcdefg12345 +#libratometrics_prefix = gl2- +#libratometrics_interval = 60 +#libratometrics_stream_filter = +#libratometrics_host_filter = diff --git a/contrib/distro/commons-daemon-redhat/SOURCES/graylog2.drl b/contrib/distro/commons-daemon-redhat/SOURCES/graylog2.drl new file mode 100644 index 000000000000..3794d3b039a5 --- /dev/null +++ b/contrib/distro/commons-daemon-redhat/SOURCES/graylog2.drl @@ -0,0 +1,26 @@ +# Example Drools file. +#import org.graylog2.messagehandlers.gelf.GELFMessage +# +#rule "Overwrite localhost host" +# when +# m : GELFMessage( host == "localhost" && version == "1.0" ) +# then +# m.setHost( "localhost.example.com" ); +# System.out.println( "[Overwrite localhost rule fired] : " + m.toString() ); +#end +# +#rule "Drop all messages from www1 facility" +# when +# m : GELFMessage( host == "www1" && facility == "graylog2-test" ) +# then +# m.setFilterOut(true); +# System.out.println( "[Drop all messages from www1 facility rule fired] : " + m.toString() ); +#end +# +#rule "Drop UDP and ICMP Traffic from firewall" +# when +# m : GELFMessage( fullMessage matches "(?i).*(ICMP|UDP) Packet(.|\n|\r)*" ) +# then +# m.setFilterOut(true); +# System.out.println("[Drop all syslog ICMP and UDP traffic] : " + m.toString() ); +#end diff --git a/contrib/distro/commons-daemon-redhat/SOURCES/graylog2.init b/contrib/distro/commons-daemon-redhat/SOURCES/graylog2.init new file mode 100755 index 000000000000..03a88439a32b --- /dev/null +++ b/contrib/distro/commons-daemon-redhat/SOURCES/graylog2.init @@ -0,0 +1,150 @@ +#!/bin/bash +# +# graylog2-server - This script starts and stops the graylog2-server daemon +# +# chkconfig: - 85 15 +# description: graylog2 is a syslog implementation that stores logs in Elasticsearch + +### BEGIN INIT INFO +# Provides: graylog2-server +# Required-Start: $network +# Required-Stop: $network +# Default-Stop: 0 1 2 6 +# Short-Description: Start/stop the graylog2-server daemon +# Description: A syslog implementation that stores logs in Elasticsearch +### END INIT INFO + +# Source function library. +. /etc/rc.d/init.d/functions + +# Check networking +. /etc/sysconfig/network +[[ $NETWORKING == no ]] && exit 0 + +NAME=graylog2-server +PID_FILE=${PIDFILE:-/var/run/${NAME}/${NAME}.pid} +LOCK_FILE=${LOCKFILE:-/var/lock/subsys/${NAME}} + +GRAYLOG2_HOME=/opt/graylog2-server +GRAYLOG2_USER=graylog2 +GRAYLOG2_CONFIG=/etc/${NAME}/graylog2.conf +GRAYLOG2_OUT=/var/log/${NAME}/${NAME}.log +LOG4J_CONFIG=file:/etc/${NAME}/log4j.xml + +GRAYLOG2_OPTS="-Xms256m -Xmx256m" +GRAYLOG2_OPTS="$GRAYLOG2_OPTS -Dcom.sun.management.jmxremote.ssl=false" +GRAYLOG2_OPTS="$GRAYLOG2_OPTS -Dcom.sun.management.jmxremote.authenticate=false" +GRAYLOG2_OPTS="$GRAYLOG2_OPTS -Dcom.sun.management.jmxremote.port=12203" + +CLASSPATH="$GRAYLOG2_HOME/$NAME.jar" + +JSVC_EXEC=`which jsvc` +JAVA_HOME=${JAVA_HOME:-/usr/java/default} + +case `echo "testing\c"`,`echo -n testing` in + *c*,-n*) echo_n= echo_c= ;; + *c*,*) echo_n=-n echo_c= ;; + *) echo_n= echo_c='\c' ;; +esac + +start() { + echo -n $"Starting ${NAME}: " + # check if graylog2 is already booted + __pids_var_run "${JAVA_HOME}/bin/java" "$PID_FILE" + RC=$? + if [ -z "$PID_FILE" -a -z "$pid" ]; then + pid="$(__pids_pidof "$1")" + fi + if [ -n "$pid" ]; then + echo_success + echo + return 0 + fi + rm -f $PID_FILE + + $JSVC_EXEC \ + -user $GRAYLOG2_USER \ + -home $JAVA_HOME \ + -server \ + $GRAYLOG2_OPTS \ + -wait 30 \ + -pidfile $PID_FILE \ + -outfile $GRAYLOG2_OUT \ + -errfile '&1' \ + -cp $CLASSPATH \ + -Dlog4j.configuration=$LOG4J_CONFIG \ + org.graylog2.ServerDaemon --configfile ${GRAYLOG2_CONFIG} + exitValue=$? + + if [ $exitValue -eq 0 ]; then + echo_success + else + echo_failure + fi + echo +} + +stop() { + echo -n $"Shutting down ${NAME}: " + + $JSVC_EXEC \ + -stop -pidfile $PID_FILE \ + -user $GRAYLOG2_USER \ + -home $JAVA_HOME \ + org.graylog2.ServerDaemon --configfile ${GRAYLOG2_CONFIG} + if [ $? -eq 0 ]; then + echo_success + else + echo_failure + fi + echo +} + +status() { + __pids_var_run "${JAVA_HOME}/bin/java" "$PID_FILE" + RC=$? + if [ -z "$PID_FILE" -a -z "$pid" ]; then + pid="$(__pids_pidof "$1")" + fi + if [ -n "$pid" ]; then + echo $"${NAME} (pid $pid) is running..." + return 0 + fi + + case "$RC" in + 0) + echo $"${NAME} (pid $pid) is running..." + return 0 + ;; + 1) + echo $"${NAME} dead but pid file exists" + return 1 + ;; + esac + echo $"${NAME} is stopped" + return 2 +} + +case "$1" in + start) + start + ;; + stop) + stop + ;; + status) + status + RETVAL=$? + ;; + restart|force-reload) + stop + start + ;; + *) + N=/etc/init.d/${NAME} + echo "Usage: $N {start|stop|restart|force-reload}" >&2 + RETVAL=2 + ;; +esac + +exit 0 diff --git a/contrib/distro/commons-daemon-redhat/SPECS/graylog2-server.spec b/contrib/distro/commons-daemon-redhat/SPECS/graylog2-server.spec new file mode 100644 index 000000000000..16164beb418c --- /dev/null +++ b/contrib/distro/commons-daemon-redhat/SPECS/graylog2-server.spec @@ -0,0 +1,128 @@ +%define _prefix /opt +%define _docdir /usr/share/doc + +Name: graylog2-server +Version: 0.10.0.rc.1 +Release: 2%{?dist} +Summary: Graylog2 Server + +Group: System Environment/Daemons +License: GPLv3 +URL: http://graylog2.org/ +Source0: https://github.com/downloads/Graylog2/graylog2-server/graylog2-server-0.10.0-rc.1.tar.gz +Source1: graylog2.drl +Source2: graylog2.conf +Source3: graylog2.init +Source4: graylog2-elasticsearch.yml +Source5: graylog2-log4j.xml +BuildRoot: %(mktemp -ud %{_tmppath}/%{name}-server-%{version}-%{release}-XXXXXX) + +Requires: java-1.6.0-openjdk +Requires: apache-commons-daemon-jsvc + +%description +Graylog2 is an open source syslog implementation that stores your logs in ElasticSearch. It consists of a server written in Java that accepts your syslog messages via TCP or UDP and stores it in the database. The second part is a Ruby on Rails web interface that allows you to view the log messages. + +%prep +%setup -q -n graylog2-server-0.10.0-rc.1 + + +%build +true + +%install +rm -rf $RPM_BUILD_ROOT + +# Base +%{__mkdir} -p %{buildroot}%{_prefix}/%{name} +%{__install} -p -D -m 0644 build_date %{buildroot}%{_prefix}/%{name} +%{__install} -p -D -m 0644 COPYING %{buildroot}%{_prefix}/%{name} +%{__install} -p -D -m 0644 graylog2.conf.example %{buildroot}%{_prefix}/%{name} +%{__install} -p -D -m 0644 graylog2-server.jar %{buildroot}%{_prefix}/%{name} +%{__install} -p -D -m 0644 README.markdown %{buildroot}%{_prefix}/%{name} +%{__mkdir} -p %{buildroot}%{_prefix}/%{name}/bin +%{__install} -p -D -m 754 bin/* %{buildroot}%{_prefix}/%{name}/bin +%{__mkdir} -p %{buildroot}%{_prefix}/%{name}/plugin +%{__mkdir} -p %{buildroot}%{_prefix}/%{name}/plugin/alarm_callbacks +%{__mkdir} -p %{buildroot}%{_prefix}/%{name}/plugin/filters +%{__mkdir} -p %{buildroot}%{_prefix}/%{name}/plugin/initializers +%{__mkdir} -p %{buildroot}%{_prefix}/%{name}/plugin/inputs +%{__mkdir} -p %{buildroot}%{_prefix}/%{name}/plugin/outputs +%{__mkdir} -p %{buildroot}%{_prefix}/%{name}/plugin/transports + +# config +%{__mkdir} -p %{buildroot}%{_sysconfdir}/%{name}/rules +%{__install} -p -D -m 0644 %{SOURCE2} %{buildroot}%{_sysconfdir}/%{name}/graylog2.conf +%{__install} -p -D -m 0644 %{SOURCE4} %{buildroot}%{_sysconfdir}/%{name}/elasticsearch.yml +%{__install} -p -D -m 0644 %{SOURCE5} %{buildroot}%{_sysconfdir}/%{name}/log4j.xml +%{__install} -p -D -m 0644 %{SOURCE1} %{buildroot}%{_sysconfdir}/%{name}/rules/graylog2.drl + +# logs +%{__mkdir} -p %{buildroot}%{_localstatedir}/log/%{name} + +# sysconfig and init +%{__mkdir} -p %{buildroot}%{_sysconfdir}/rc.d/init.d +%{__install} -m 755 %{SOURCE3} %{buildroot}%{_sysconfdir}/rc.d/init.d/%{name} + +%{__mkdir} -p %{buildroot}%{_localstatedir}/run/%{name} +%{__mkdir} -p %{buildroot}%{_localstatedir}/lock/subsys/%{name} + +%pre +# create graylog2 group +if ! getent group graylog2 >/dev/null; then + /usr/sbin/groupadd -r graylog2 +fi + +# create graylog2 user +if ! getent passwd graylog2 >/dev/null; then + /usr/sbin/useradd -r -g graylog2 -d %{_prefix}/%{name} \ + -s /sbin/nologin -c "log aggregator" graylog2 +fi + +%post +/sbin/chkconfig --add %{name} + + +%preun +if [[ $1 -ge 1 ]] +then + /sbin/service %{name} stop > /dev/null 2>&1 + /sbin/chkconfig --del %{name} +fi + + +%clean +rm -rf $RPM_BUILD_ROOT + + +%files +%defattr(-,root,root,-) +%{_sysconfdir}/rc.d/init.d/%{name} +%dir %{_sysconfdir}/%{name} +%dir %{_sysconfdir}/%{name}/rules +%config(noreplace) %{_sysconfdir}/%{name}/*.conf +%config(noreplace) %{_sysconfdir}/%{name}/*.yml +%config(noreplace) %{_sysconfdir}/%{name}/*.xml +%config(noreplace) %{_sysconfdir}/%{name}/rules/*.drl + +%defattr(-,graylog2,graylog2,-) +%{_prefix}/%{name} +%{_prefix}/%{name}/bin +%{_prefix}/%{name}/plugin +%{_prefix}/%{name}/plugin/alarm_callbacks +%{_prefix}/%{name}/plugin/filters +%{_prefix}/%{name}/plugin/initializers +%{_prefix}/%{name}/plugin/inputs +%{_prefix}/%{name}/plugin/outputs +%{_prefix}/%{name}/plugin/transports +%dir %{_localstatedir}/run/%{name} +%dir %{_localstatedir}/log/%{name} + + +%changelog +* Wed Dec 26 2012 kbrockhoff@codekaizen.org 0.10.0-rc.1 +- Update to latest graylog2 release + +* Wed Oct 24 2012 kbrockhoff@codekaizen.org 0.9.6p1-2 +- Initial definition + diff --git a/pom.xml b/pom.xml index 7ffca6cacef2..fe44c9bc5a60 100644 --- a/pom.xml +++ b/pom.xml @@ -168,6 +168,11 @@ high-scale-lib 1.1.4 + + commons-daemon + commons-daemon + 1.0.10 + diff --git a/src/main/java/org/graylog2/Core.java b/src/main/java/org/graylog2/Core.java index 44d3c2073aeb..9f4197336bfc 100644 --- a/src/main/java/org/graylog2/Core.java +++ b/src/main/java/org/graylog2/Core.java @@ -39,6 +39,8 @@ import org.graylog2.streams.StreamCache; import com.google.common.collect.Lists; + +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import com.google.common.collect.Maps; import java.util.Map; @@ -115,6 +117,8 @@ public class Core implements GraylogServer { private OutputBuffer outputBuffer; private AtomicInteger outputBufferWatermark = new AtomicInteger(); private AtomicInteger processBufferWatermark = new AtomicInteger(); + private AtomicBoolean componentsInitialized = new AtomicBoolean(false); + private AtomicBoolean processing = new AtomicBoolean(false); private Deflector deflector; @@ -201,10 +205,28 @@ public void registerTransport(Transport transport) { public void registerAlarmCallback(AlarmCallback alarmCallback) { this.alarmCallbacks.add(alarmCallback); } + + public boolean isProcessing() { + return processing.get(); + } + + public void setProcessing(boolean processing) { + this.processing.set(processing); + } @Override public void run() { + if (!componentsInitialized.get()) { + initializeComponents(); + } + + while (isProcessing()) { + try { Thread.sleep(1000); } catch (InterruptedException e) { /* lol, i don't care */ } + } + + } + void initializeComponents() { gelfChunkManager.start(); BlacklistCache.initialize(this); StreamCache.initialize(this); @@ -326,13 +348,9 @@ public void run() { } } + componentsInitialized.set(true); activityWriter.write(new Activity("Started up.", GraylogServer.class)); LOG.info("Graylog2 up and running."); - - while (true) { - try { Thread.sleep(1000); } catch (InterruptedException e) { /* lol, i don't care */ } - } - } private void loadPlugins(Class type, String subDirectory) { diff --git a/src/main/java/org/graylog2/ServerDaemon.java b/src/main/java/org/graylog2/ServerDaemon.java new file mode 100644 index 000000000000..43663a1bea8f --- /dev/null +++ b/src/main/java/org/graylog2/ServerDaemon.java @@ -0,0 +1,215 @@ +/** + * Copyright 2010, 2011, 2012 Lennart Koopmann + * + * This file is part of Graylog2. + * + * Graylog2 is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * Graylog2 is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with Graylog2. If not, see . + * + */ + +package org.graylog2; + +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +import org.apache.commons.daemon.Daemon; +import org.apache.commons.daemon.DaemonContext; +import org.apache.commons.daemon.DaemonInitException; +import org.apache.log4j.Level; +import org.graylog2.activities.Activity; +import org.graylog2.alarms.transports.EmailTransport; +import org.graylog2.alarms.transports.JabberTransport; +import org.graylog2.filters.BlacklistFilter; +import org.graylog2.filters.CounterUpdateFilter; +import org.graylog2.filters.RewriteFilter; +import org.graylog2.filters.StreamMatcherFilter; +import org.graylog2.filters.TokenizerFilter; +import org.graylog2.initializers.AMQPSyncInitializer; +import org.graylog2.initializers.AlarmScannerInitializer; +import org.graylog2.initializers.AnonymousInformationCollectorInitializer; +import org.graylog2.initializers.BufferWatermarkInitializer; +import org.graylog2.initializers.DeflectorThreadsInitializer; +import org.graylog2.initializers.DroolsInitializer; +import org.graylog2.initializers.GraphiteInitializer; +import org.graylog2.initializers.HostCounterCacheWriterInitializer; +import org.graylog2.initializers.IndexRetentionInitializer; +import org.graylog2.initializers.LibratoMetricsInitializer; +import org.graylog2.initializers.MessageCounterInitializer; +import org.graylog2.initializers.ServerValueWriterInitializer; +import org.graylog2.initializers.StatisticsPrinterInitializer; +import org.graylog2.inputs.amqp.AMQPInput; +import org.graylog2.inputs.gelf.GELFTCPInput; +import org.graylog2.inputs.gelf.GELFUDPInput; +import org.graylog2.inputs.http.GELFHttpInput; +import org.graylog2.inputs.syslog.SyslogTCPInput; +import org.graylog2.inputs.syslog.SyslogUDPInput; +import org.graylog2.plugin.Tools; +import org.graylog2.outputs.ElasticSearchOutput; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.beust.jcommander.JCommander; +import com.github.joschi.jadconfig.JadConfig; +import com.github.joschi.jadconfig.RepositoryException; +import com.github.joschi.jadconfig.ValidationException; +import com.github.joschi.jadconfig.repositories.PropertiesRepository; + +/** + * Wrapper for use with Apache Commons Daemon. + * + * @author $Author: kbrockhoff $ + * @version $Revision: 201129 $, $Date: 2013-01-03 08:26:44 -0600 (Thu, 03 Jan 2013) $ + */ +public class ServerDaemon implements Daemon { + + private static final String[] EMPTY_ARGS = {}; + private static final Logger LOG = LoggerFactory.getLogger(ServerDaemon.class); + + private String[] args = EMPTY_ARGS; + private Core server; + private ExecutorService executorService = Executors.newSingleThreadExecutor(); + + public ServerDaemon() { + super(); + } + + @Override + public void init(final DaemonContext context) throws DaemonInitException { + args = context.getArguments(); + } + + @Override + public void start() throws RepositoryException, ValidationException { + + final CommandLineArguments commandLineArguments = new CommandLineArguments(); + final JCommander jCommander = new JCommander(commandLineArguments, args); + jCommander.setProgramName("graylog2"); + + // Are we in debug mode? + if (commandLineArguments.isDebug()) { + LOG.info("Running in Debug mode"); + org.apache.log4j.Logger.getRootLogger().setLevel(Level.ALL); + org.apache.log4j.Logger.getLogger(Main.class.getPackage().getName()).setLevel(Level.ALL); + } + + LOG.info("Graylog2 {} starting up. (JRE: {})", Core.GRAYLOG2_VERSION, Tools.getSystemInformation()); + + String configFile = commandLineArguments.getConfigFile(); + LOG.info("Using config file: {}", configFile); + + final Configuration configuration = new Configuration(); + JadConfig jadConfig = new JadConfig(new PropertiesRepository(configFile), configuration); + + LOG.info("Loading configuration"); + jadConfig.process(); + + // Le server object. This is where all the magic happens. + server = new Core(); + server.initialize(configuration); + + // Could it be that there is another master instance already? + if (configuration.isMaster() && server.cluster().masterCountExcept(server.getServerId()) != 0) { + // All devils here. + String what = "Detected other master node in the cluster! Starting as non-master! " + + "This is a mis-configuration you should fix."; + LOG.warn(what); + server.getActivityWriter().write(new Activity(what, Main.class)); + + configuration.setIsMaster(false); + } + + // Enable local mode? + if (commandLineArguments.isLocal() || commandLineArguments.isDebug()) { + // In local mode, systemstats are sent to localhost for example. + LOG.info("Running in local mode"); + server.setLocalMode(true); + } + + // Are we in stats mode? + if (commandLineArguments.isStats()) { + LOG.info("Printing system utilization information."); + server.setStatsMode(true); + } + + // Register transports. + if (configuration.isTransportEmailEnabled()) { server.registerTransport(new EmailTransport()); } + if (configuration.isTransportJabberEnabled()) { server.registerTransport(new JabberTransport()); } + + // Register initializers. + server.registerInitializer(new ServerValueWriterInitializer()); + server.registerInitializer(new DroolsInitializer()); + server.registerInitializer(new HostCounterCacheWriterInitializer()); + server.registerInitializer(new MessageCounterInitializer()); + server.registerInitializer(new AlarmScannerInitializer()); + if (configuration.isEnableGraphiteOutput()) { server.registerInitializer(new GraphiteInitializer()); } + if (configuration.isEnableLibratoMetricsOutput()) { server.registerInitializer(new LibratoMetricsInitializer()); } + server.registerInitializer(new DeflectorThreadsInitializer()); + server.registerInitializer(new AnonymousInformationCollectorInitializer()); + if (configuration.performRetention() && commandLineArguments.performRetention()) { + server.registerInitializer(new IndexRetentionInitializer()); + } + if (configuration.isAmqpEnabled()) { + server.registerInitializer(new AMQPSyncInitializer()); + } + server.registerInitializer(new BufferWatermarkInitializer()); + if (commandLineArguments.isStats()) { server.registerInitializer(new StatisticsPrinterInitializer()); } + + // Register inputs. + if (configuration.isUseGELF()) { + server.registerInput(new GELFUDPInput()); + server.registerInput(new GELFTCPInput()); + } + + if (configuration.isSyslogUdpEnabled()) { server.registerInput(new SyslogUDPInput()); } + if (configuration.isSyslogTcpEnabled()) { server.registerInput(new SyslogTCPInput()); } + + if (configuration.isAmqpEnabled()) { server.registerInput(new AMQPInput()); } + + if (configuration.isHttpEnabled()) { server.registerInput(new GELFHttpInput()); } + + // Register message filters. + server.registerFilter(new RewriteFilter()); + server.registerFilter(new BlacklistFilter()); + if (configuration.isEnableTokenizerFilter()) { server.registerFilter(new TokenizerFilter()); } + server.registerFilter(new StreamMatcherFilter()); + server.registerFilter(new CounterUpdateFilter()); + + // Register outputs. + server.registerOutput(new ElasticSearchOutput()); + + // initialize the components + server.initializeComponents(); + + // Set running + executorService.execute(server); + } + + @Override + public void stop() { + if (server != null) { + LOG.info("Graylog2 {} exiting.", Core.GRAYLOG2_VERSION); + server.setProcessing(false); + executorService.shutdown(); + } + } + + @Override + public void destroy() { + if (!executorService.isShutdown()) { + stop(); + } + server = null; + } + +} diff --git a/src/main/java/org/graylog2/database/MongoConnection.java b/src/main/java/org/graylog2/database/MongoConnection.java index 599f6a5f1e35..8e7f931bc8ee 100644 --- a/src/main/java/org/graylog2/database/MongoConnection.java +++ b/src/main/java/org/graylog2/database/MongoConnection.java @@ -112,7 +112,7 @@ public DB getDatabase() { * * @return The messages collection */ - public DBCollection getMessageCountsColl() { + public synchronized DBCollection getMessageCountsColl() { if (this.messageCountsCollection != null) { return this.messageCountsCollection; } diff --git a/src/main/java/org/graylog2/inputs/amqp/AMQPConsumer.java b/src/main/java/org/graylog2/inputs/amqp/AMQPConsumer.java index 9155e3c151f3..52ed17382915 100644 --- a/src/main/java/org/graylog2/inputs/amqp/AMQPConsumer.java +++ b/src/main/java/org/graylog2/inputs/amqp/AMQPConsumer.java @@ -20,6 +20,7 @@ package org.graylog2.inputs.amqp; import com.rabbitmq.client.AMQP; +import com.rabbitmq.client.AlreadyClosedException; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; @@ -31,6 +32,7 @@ import java.io.IOException; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import com.google.common.util.concurrent.ThreadFactoryBuilder; @@ -58,6 +60,7 @@ public class AMQPConsumer implements Runnable { Connection connection; Channel channel; + ExecutorService executor; private final Meter handledMessages = Metrics.newMeter(AMQPConsumer.class, "HandledAMQPMessages", "messages", TimeUnit.SECONDS); private final Meter handledSyslogMessages = Metrics.newMeter(AMQPConsumer.class, "HandledAMQPSyslogMessages", "messages", TimeUnit.SECONDS); @@ -139,9 +142,24 @@ public void disconnect() { AMQPInput.getConsumers().remove(queueConfig.getId()); channel.close(); - connection.close(); + } catch (AlreadyClosedException ignore) { + // do nothing } catch(IOException e) { LOG.error("Could not disconnect from AMQP broker!", e); + } finally { + if (executor != null) { + executor.shutdownNow(); + executor = null; + } + try { + if (connection != null && connection.isOpen()) { + connection.close(); + } + } catch (AlreadyClosedException ignore) { + // do nothing + } catch (IOException e) { + LOG.error("Could not disconnect from AMQP broker!", e); + } } } @@ -159,11 +177,11 @@ private Channel connect() throws IOException { factory.setHost(server.getConfiguration().getAmqpHost()); factory.setPort(server.getConfiguration().getAmqpPort()); - connection = factory.newConnection(Executors.newCachedThreadPool( - new ThreadFactoryBuilder() - .setNameFormat("amqp-consumer-" + queueConfig.getId() + "-%d") - .build() - )); + executor = Executors.newCachedThreadPool( + new ThreadFactoryBuilder() + .setNameFormat("amqp-consumer-" + queueConfig.getId() + "-%d") + .build()); + connection = factory.newConnection(executor); return connection.createChannel(); }