org.thales.punch.libraries.storm.spout

Class HttpInput

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<org.thales.punch.netty.impl.NettyHttpTuple>

org.thales.punch.libraries.storm.spout.HttpInput

All Implemented Interfaces:

Serializable, org.apache.storm.spout.ISpout, org.apache.storm.topology.IComponent, org.apache.storm.topology.IRichSpout

public class HttpInput
extends AbstractSocketInput<org.thales.punch.netty.impl.NettyHttpTuple>

The Http spout receives http requests and forwards them as tuple.

Overview

The Http spout reads body of an Http request. You can configure the Http 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.

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 Http spout relies on a blocking queue in a way to slow down 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.

By default the HttpSpout sends the http response directly to the client after Storm aknowledges the tuple. Another mode is available: instead of sending the response to the client once the tuple is acknowledged, the Http spout subscribes to a whiteboard, specifying a unique id corresponding to the tuple and a callback to use when a WhiteBoardBolt produces on the whiteboard. This is the callback's job to respond to the client. You can enable this mode by setting direct_response parameter to false.

Spout level settings

The listen section has the following parameters:

property	mandatory	type	default	comment
host	yes	string	-	the listening host address. Accepted values are dotted number IP addresses, or hostname. Use ""0.0.0.0" to listen to all interfaces.
port	yes	int	-	The listening port.
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
multipart_body_size	no	Integer	16*1024*1024	the maximum size allowed for decoding multipart content. If you set it to 0, multiparts will not be decoded at all.

Besides the "listen" section, the spout configuration section has the following parameter:

property	mandatory	type	default	comment
rcv_queue_size	no	int	10000	The per spout receive queue size. If the topology is slower to process the received requests, it will end up stop reqding the incpming sockets.
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.

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
  
 

Here is an example configuration :

 "spouts" : [
   {
     "type" : "syslog_spout",
     "spout_settings" : {
       "listen" : {
         "proto" : "tcp",
         "host" : "localhost",
         "port" : 9999,
         "direct_response": true
         "ssl" : true,
         "ssl_private_key" : "/opt/keys/punchplatform.key.pkcs8",
         "ssl_certificate" : "/opt/keys/punchplatform.crt"
         },
     }
   }
 ]

 

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
HttpInput(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.

Method Summary

Modifier and Type	Method and Description
void	ack(Object o)
void	close() Close callback, notified by Storm runtime at topology deactivation.
void	deactivate()
void	fail(Object o)
void	nextTuple()
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

getSocketHostAddress

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

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

HttpInput

public HttpInput(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 punchplatform spout configuration

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<org.thales.punch.netty.impl.NettyHttpTuple>

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()

ack

public void ack(Object o)

Specified by:

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

Overrides:

ack in class AbstractSocketInput<org.thales.punch.netty.impl.NettyHttpTuple>

fail

public void fail(Object o)

Specified by:

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

Overrides:

fail in class AbstractSocketInput<org.thales.punch.netty.impl.NettyHttpTuple>

deactivate

public void deactivate()

Specified by:

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

Overrides:

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