org.thales.punch.libraries.storm.spout

Class SyslogInput<T>

java.lang.Object

org.apache.storm.topology.base.BaseComponent

org.apache.storm.topology.base.BaseRichSpout

org.thales.punch.libraries.storm.api.BaseInputNode

org.thales.punch.libraries.storm.spout.AbstractSocketInput<T>

org.thales.punch.libraries.storm.spout.SyslogInput<T>

All Implemented Interfaces:

Serializable, org.apache.storm.spout.ISpout, org.apache.storm.topology.IComponent, org.apache.storm.topology.IRichSpout, ISyslogReceiver<T>

Direct Known Subclasses:

SnmpInput

public class SyslogInput<T>
extends AbstractSocketInput<T>
implements ISyslogReceiver<T>

The Syslog input reads lines from tcp/udp and emits them as tuples.

Overview

The syslog spout read lines of data, delimited by a carriage return or '\0'. You can configure the Syslog spout in two ways depending if you want it to take the data as is and embed it into an initial Json document, or if you already expect some well-formed Json document, in which case you may want to decode it as json. With the former, the spout emits tuples containing only one field, for each packet received on the socket.

This spout launches a single dedicated thread for running the TCP server. All received messages are enqueued, waiting for Storm to come and get them by invoking the nextTuple() method. In there, if there are logs they are emitted to the topology.

The syslog input relies on a blocking queue in a way to slowdown the receiver thread (hence the reading of the socket) should Storm be slow to empty the queue. The queue is configured to hold a decent number of message (100000), so as to tolerate a few seconds of incoming logs without slowing down.

When you use a SyslogSpout as entry point of your platform, it is handy to make it add extra information to the read data. You can declare one or several of the following special fields to the Storm configuration, in which case the spout will emit interesting information:

field name	comment	type
local_uuid	a short yet unique identifier	string
local_timestamp	a timestamp in unix format	long
local_port	the local receiving socket port number	long
remote_port	the remote receiving socket port number	long
local_host	the local receiving socket ip address	string
remote_host	the remote socket ip address	string

Listen Configuration

The socket level settings are defined in a listen section. It is defined as follows:

 
 "listen" :
            {
               "proto" : "tcp",
               "host" : "target.ip.address",
               "port" : 9999,
               "ssl" : true,
               "ssl_private_key" : "/opt/keys/punchplatform.key.pkcs8",
               "ssl_certificate" : "/opt/keys/punchplatform.crt",
            }
 
 

Spout level settings

property	mandatory	type	default	comment
load_control	no	string	"none"	If set to "rate", the spout will limit the rate to a specified value. You can use this to limit the incoming or outgoing rate of data from/to external systems. Check the "load_control.adaptative" and "load_control.rate" related properties
load_control.rate	no	long	10000	Only relevant if load_control is set to "rate". Limit the rate to this number of message per seconds.
load_control.adaptative	no	long	false	If true, the load control will not limit the traffic to load_control.rate message per second, but to less or more as long as the topology is not overloaded. Overload situation is determined by monitoring the Storm tuple traversal time. If that traversal time increases (which occurs quicly as soon as the topology is overloaded), the load controller will limit the allowed rate. This option makes it easy to protect your topology without limiting its rate to a too conservative rate, as it is difficult if not impossible to estimate the maximum rate.
rcv_queue_size	no	int	100000	To each configured listening address corresponds one running thread, which reads in lines and stores them in a queue, in turn consumed by the spout executor(s). This parameter sets the size of that queue. If the queue is full, the socket will not be read anymore, which in turn will slow down the sender. Use this to give more or less capacity to accept burst without impacting the senders.
multiline	no	boolean	false	If true, the spout will aggregate subsequent log line using a prefix such as "\t". Refer to the other multiline properties.
multiline.regex	no	String	""	set the regex used to determine if a log line is an initial or subsequent line.
multiline.delimiter	no	String	" "	Once completed the aggregated log line is made up from the sequence of each line. You can insert a line delimiter to make further parsing easier.
multiline.timeout	no	long	1000	If a single log line is received, it will be forwarded downstream after this timeout, expressed in milliseconds.
ssl	no	string	false	Set to true to listen on SSL. If true, you must provide a certificate and a private key.
ssl_certificate	no	string	-	An X.509 certificate chain file in PEM format
ssl_private_key	no	string	-	A PKCS#8 private key file in PEM format. This file must be available on the Storm worker host. If you have another format, refer to the explanation below.
ssl_secret	no	string	-	An optional secret associated to the key if it is password protected.
secret	no	string	-	The secret to load a TLS PKCS12 SSL context

Ssl

For testing, you can generate a certificate and keys using :

 
  openssl req -x509 -batch -nodes -newkey rsa:2048 -keyout punchplatform.key -out punchplatform.crt
 
 

The Syslog spout expects private key in PKCS8. Use the following to convert a non PKCS8 to PKCS8 key.

 
   openssl pkcs8 -topk8 -nocrypt -in punchplatform.key -out punchplatform.key.pcks8
  
 

See Also:

Serialized Form

Field Summary

Fields inherited from class org.thales.punch.libraries.storm.spout.AbstractSocketInput

charset

Fields inherited from class org.thales.punch.libraries.storm.api.BaseInputNode

collector, exitCondition, latencyRecordSender, loadController, metricContext, myself, nodeSettings

Constructor Summary

Constructors

Constructor and Description
SyslogInput(org.thales.punch.libraries.storm.api.NodeSettings spoutConfig) A Spout constructor is in charge of setting the fields that will be required later on when the spout will be effectively started and run somewhere in the cluster.
SyslogInput(org.thales.punch.libraries.storm.api.NodeSettings spoutConfig, org.apache.logging.log4j.Logger subLogger) Alternative constructor to use a Logger coming from the subclass

Method Summary

Modifier and Type	Method and Description
void	close() Close callback, notified by Storm runtime at topology deactivation.
void	fail(Object o)
void	nextTuple() Callback to feed the topology with tuples.
void	onReceive(org.thales.punch.netty.impl.NettyTupleImpl<T> log)
void	open(Map stormSettings, org.apache.storm.task.TopologyContext topologyContext, org.apache.storm.spout.SpoutOutputCollector collector) Setup this spout before it starts receiving data and emitting tuples into the topology.
org.thales.punch.libraries.storm.api.ISpout	registerNextTupleCallback(org.thales.punch.libraries.storm.api.INextTupleCallback cb)

Methods inherited from class org.thales.punch.libraries.storm.spout.AbstractSocketInput

ack, getSocketHostAddress

Methods inherited from class org.thales.punch.libraries.storm.api.BaseInputNode

deactivate, declareOutputFields, getPublishedStreams, regulate, sendLatencyRecord

Methods inherited from class org.apache.storm.topology.base.BaseRichSpout

activate

Methods inherited from class org.apache.storm.topology.base.BaseComponent

getComponentConfiguration

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface org.apache.storm.spout.ISpout

activate

Methods inherited from interface org.apache.storm.topology.IComponent

getComponentConfiguration

Constructor Detail

SyslogInput

public SyslogInput(org.thales.punch.libraries.storm.api.NodeSettings spoutConfig)

A Spout constructor is in charge of setting the fields that will be required later on when the spout will be effectively started and run somewhere in the cluster. Each field set here must be serializable. Here we use a simple scheme: the complete settings, passed as Json, are serialized as a string. We only check that the mandatory properties are present.

Parameters:

spoutConfig - the punchplatofrm spout configuration

SyslogInput

public SyslogInput(org.thales.punch.libraries.storm.api.NodeSettings spoutConfig,
                   org.apache.logging.log4j.Logger subLogger)

Alternative constructor to use a Logger coming from the subclass

Parameters:

spoutConfig - the spout configuration

subLogger - the subclass logger

Method Detail

open

public void open(Map stormSettings,
                 org.apache.storm.task.TopologyContext topologyContext,
                 org.apache.storm.spout.SpoutOutputCollector collector)

Setup this spout before it starts receiving data and emitting tuples into the topology.

This is called by the Storm runtime at topology activation.

Specified by:

open in interface org.apache.storm.spout.ISpout

Overrides:

open in class AbstractSocketInput<T>

Parameters:

stormSettings - storm configuration

topologyContext - the storm topology context

collector - the output collector for emitting tuples.

registerNextTupleCallback

public org.thales.punch.libraries.storm.api.ISpout registerNextTupleCallback(org.thales.punch.libraries.storm.api.INextTupleCallback cb)

close

public void close()

Close callback, notified by Storm runtime at topology deactivation. The spout listening endpoint is closed.

Specified by:

close in interface org.apache.storm.spout.ISpout

Overrides:

close in class org.thales.punch.libraries.storm.api.BaseInputNode

nextTuple

public void nextTuple()

Callback to feed the topology with tuples. This is called by Storm. A single tuple is returned. If there is none in the spout queue, this method sleeps for a short delay, in order not to keep Storm in a busy loop. Dimitri : can you confirm the "short delay" is still implemented here ? You might have removed it because it caused problems with Storm own flow-control...

Specified by:

nextTuple in interface org.apache.storm.spout.ISpout

fail

public void fail(Object o)

Specified by:

fail in interface org.apache.storm.spout.ISpout

Overrides:

fail in class AbstractSocketInput<T>

onReceive

public void onReceive(org.thales.punch.netty.impl.NettyTupleImpl<T> log)

Specified by:

onReceive in interface ISyslogReceiver<T>