Add engine for fluent to master

doc/ref/engines/all/index.rst

@@ -11,6 +11,7 @@ engine modules
     :template: autosummary.rst.tmpl
 
     docker_events
+    fluent
     http_logstash
     ircbot
     junos_syslog

doc/ref/engines/all/salt.engines.fluent.rst

@@ -0,0 +1,6 @@
+============================
+salt.engines.fluent
+============================
+
+.. automodule:: salt.engines.fluent
+    :members:

doc/topics/releases/neon.rst

@@ -155,13 +155,36 @@ Also, slot parsing is now supported inside of nested state data structures (dict
             - "DO NOT OVERRIDE"
           ignore_if_missing: True
 
-
 State Changes
 =============
 
 - Added new :py:func:`ssh_auth.manage <salt.states.ssh_auth.manage>` state to
   ensure only the specified ssh keys are present for the specified user.
 
+Enhancements to Engines
+=======================
+
+- A new :py:func:`fluent engine <salt.engines.salt.engines.fluent>` has been
+  added to export Salt events to fluentd.
+
+  .. code-block:: yaml
+
+    engines:
+      - fluent
+          host: localhost
+          port: 24224
+
+  .. code-block::
+
+    <source>
+      @type forward
+      port 24224
+    </source>
+    <match saltstack.**>
+      @type file
+      path /var/log/td-agent/saltstack
+    </match>
+
 Module Changes
 ==============
 

salt/engines/fluent.py

@@ -0,0 +1,94 @@
+# -*- coding: utf-8 -*-
+'''
+An engine that reads messages from the salt event bus and pushes
+them onto a fluent endpoint.
+
+.. versionadded:: neon
+
+:Configuration:
+
+All arguments are optional
+
+    Example configuration of default settings
+
+    .. code-block:: yaml
+
+        engines:
+          - fluent:
+              host: localhost
+              port: 24224
+              app: engine
+
+    Example fluentd configuration
+
+    .. code-block:: none
+
+        <source>
+            @type forward
+            port 24224
+        </source>
+
+        <match saltstack.**>
+            @type file
+            path /var/log/td-agent/saltstack
+        </match>
+
+:depends: fluent-logger
+'''
+
+# Import python libraries
+from __future__ import absolute_import, print_function, unicode_literals
+import logging
+
+# Import salt libs
+import salt.utils.event
+
+# Import third-party libs
+try:
+    from fluent import sender, event
+except ImportError:
+    sender = None
+
+log = logging.getLogger(__name__)
+
+__virtualname__ = 'fluent'
+
+
+def __virtual__():
+    return __virtualname__ \
+        if sender is not None \
+        else (False, 'fluent-logger not installed')
+
+
+def start(host='localhost', port=24224, app='engine'):
+    '''
+    Listen to salt events and forward them to fluent
+
+    args:
+        host (str): Host running fluentd agent. Default is localhost
+        port (int): Port of fluentd agent. Default is 24224
+        app (str): Text sent as fluentd tag. Default is "engine". This text is appended
+                   to "saltstack." to form a fluentd tag, ex: "saltstack.engine"
+    '''
+    SENDER_NAME = 'saltstack'
+
+    sender.setup(SENDER_NAME, host=host, port=port)
+
+    if __opts__.get('id').endswith('_master'):
+        event_bus = salt.utils.event.get_master_event(
+                __opts__,
+                __opts__['sock_dir'],
+                listen=True)
+    else:
+        event_bus = salt.utils.event.get_event(
+            'minion',
+            transport=__opts__['transport'],
+            opts=__opts__,
+            sock_dir=__opts__['sock_dir'],
+            listen=True)
+    log.info('Fluent engine started')
+
+    while True:
+        salt_event = event_bus.get_event_block()
+        if salt_event:
+            event.Event(app, salt_event)

tests/unit/engines/test_slack.py

@@ -20,7 +20,7 @@
 @skipIf(slack.HAS_SLACKCLIENT is False, 'The SlackClient is not installed')
 class EngineSlackTestCase(TestCase, LoaderModuleMockMixin):
     '''
-    Test cases for salt.engine.sqs_events
+    Test cases for salt.engine.slack
     '''
 
     def setup_loader_modules(self):