introduced a stomp transport for beaver using stomp.py

Peter Lenderyou · Peter Lenderyou · commit 8a421946555b · 2015-06-04T12:46:48.000+02:00
diff --git a/beaver/config.py b/beaver/config.py
@@ -109,6 +109,11 @@ def __init__(self, args, logger=None):
             'zeromq_address': os.environ.get('ZEROMQ_ADDRESS', 'tcp://localhost:2120'),
             'zeromq_pattern': 'push',
             'zeromq_hwm': os.environ.get('ZEROMQ_HWM', ''),
+            'stomp_host' : 'localhost',
+            'stomp_port' : '61613',
+            'stomp_user' : 'user',
+            'stomp_password' : None,
+            'stomp_queue' : 'queue/logstash',
 
             # exponential backoff
             'respawn_delay': '3',
diff --git a/beaver/transports/stomp_transport.py b/beaver/transports/stomp_transport.py
@@ -0,0 +1,81 @@
+# -*- coding: utf-8 -*-
+import stomp
+
+from beaver.transports.base_transport import BaseTransport
+from beaver.transports.exception import TransportException
+
+
+class StompTransport(BaseTransport):
+
+    def __init__(self, beaver_config, logger=None):
+        """
+        Mosquitto client initilization. Once this this transport is initialized
+        it has invoked a connection to the server
+        """
+        super(StompTransport, self).__init__(beaver_config, logger=logger)
+
+        self.createConnection(beaver_config)
+        self.logger = logger
+
+ 
+    def callback(self, filename, lines, **kwargs):
+        """publishes lines one by one to the given topic"""
+        timestamp = self.get_timestamp(**kwargs)
+        if kwargs.get('timestamp', False):
+            del kwargs['timestamp']
+        
+
+        for line in lines:
+            try:
+                import warnings
+                with warnings.catch_warnings():
+                    warnings.simplefilter('error')
+                    m = self.format(filename, line, timestamp, **kwargs)
+                    self.logger.debug("Sending message " + m)
+                    self.conn.send(destination=self.queue, body=m)    
+
+            except Exception, e:
+                self.logger.error(e)
+                try:
+                    raise TransportException(e)
+                except AttributeError:
+                    raise TransportException('Unspecified exception encountered')
+
+    def createConnection(self, beaver_config):
+        self.host = beaver_config.get('stomp_host')
+        self.port = int(beaver_config.get('stomp_port'))
+        self.userName = beaver_config.get('stomp_user', None)
+        self.password = beaver_config.get('stomp_password', None)
+        self.queue = beaver_config.get('stomp_queue')
+        self.stompConnect()
+
+    def stompConnect(self):
+        try:
+          host_and_ports = (self.host, self.port)
+          self.conn = stomp.Connection([host_and_ports]);
+          self.conn.start()
+          self.conn.connect(self.userName, self.password)
+        except stomp.exception.NotConnectedException, e:
+            try:
+                raise TransportException(e.strerror)
+            except AttributeError:
+                raise TransportException('Unspecified exception encountered')
+
+    def reconnect(self):
+        """Allows reconnection from when a handled
+        TransportException is thrown"""
+        try:
+            self.conn.close()
+        except Exception,e:
+            self.logger.warn(e)
+
+        self.createConnection()
+
+        return True
+
+    def interrupt(self):
+        if self.conn:
+            self.conn.close()
+
+    def unhandled(self):
+        return True
diff --git a/docs/user/usage.rst b/docs/user/usage.rst
@@ -10,7 +10,7 @@ usage::
     beaver [-h] [-c CONFIG] [-C CONFD_PATH] [-d] [-D] [-f FILES [FILES ...]]
            [-F {json,msgpack,raw,rawjson,string}] [-H HOSTNAME] [-m {bind,connect}]
            [-l OUTPUT] [-p PATH] [-P PID]
-           [-t {kafka,mqtt,rabbitmq,redis,sqs,kinesis,stdout,tcp,udp,zmq}] [-v] [--fqdn]
+           [-t {kafka,mqtt,rabbitmq,redis,sqs,kinesis,stdout,tcp,udp,zmq,stomp}] [-v] [--fqdn]
 
 optional arguments::
 
@@ -33,7 +33,7 @@ optional arguments::
                           file to pipe output to (in addition to stdout)
     -p PATH, --path PATH  path to log files
     -P PID, --pid PID     path to pid file
-    -t {kafka,mqtt,rabbitmq,redis,sqs,kinesis,stdout,tcp,udp,zmq}, --transport {kafka,mqtt,rabbitmq,redis,sqs,kinesis,stdout,tcp,udp,zmq}
+    -t {kafka,mqtt,rabbitmq,redis,sqs,kinesis,stdout,tcp,udp,zmq,stomp}, --transport {kafka,mqtt,rabbitmq,redis,sqs,kinesis,stdout,tcp,udp,zmq,stomp}
                           log transport method
     -v, --version         output version and quit
     --fqdn                use the machine's FQDN for source_host
@@ -93,6 +93,12 @@ Beaver can optionally get data from a ``configfile`` using the ``-c`` flag. This
 * zeromq_hwm: Default None. Zeromq HighWaterMark socket option
 * zeromq_bind: Default ``bind``. Whether to bind to zeromq host or simply connect
 * http_url: Default ``None`` http://someserver.com/path
+* stomp_host: Default ``localhost``
+* stomp_port: Default ``61613``
+* stomp_user: Default ``None``
+* stomp_password: Default ``None``
+* stomp_queue: Default ``queue/logstash``
+
 
 The following are used for instances when a TransportException is thrown - Transport dependent
 
@@ -539,6 +545,29 @@ To ship logs directly to a Graylog server, start with this configuration::
 
     # From the commandline
     beaver -c /etc/beaver/conf -f /var/log/somefile -t http -F gelf
+
+Stomp transport using Stomp.py::
+
+    # /etc/beaver/conf
+    [beaver]
+    stomp_host: 'localhost'
+    stomp_port: '61613'
+    stomp_user: 'producer-user'
+    stomp_password : 'password'
+    stomp_queue : 'queue/logstash'
+
+    # logstash indexer config:
+    stomp {
+        user => "consumer-user"
+        password => "consumer-password"
+        destination => "logstash"
+        host => "localhost"
+        port => "61613"
+    }
+    output { stdout { debug => true } }
+
+    # From the commandline
+    beaver -c /etc/beaver/conf -f /var/log/somefile.log -t stomp
     
 Sincedb support using Sqlite3
 *****************************
