SaltStack State Event TCP Returner Module

tcp_returner.py

# -*- coding: utf-8 -*-
'''
Returns state event data for state.sls and state.highstate execution only using a tcp socket, this method of
returning data can be used for Splunk or ELK.

Each event sent represents a single state executed.

It is strongly recommended to use the ``event_return_whitelist`` so not all
events call this returner, for example:

..code-block:: yaml

   event_return_whitelist:
      - salt/job/*/ret/*

.. versionadded:: Boron

Add the following to the master configuration file:

..code-block:: yaml

   returner.tcp_return.host:<recieving server hostname or ip>
   returner.tcp_return.port: <recieving server listening port>

For events return set the event_return to tcp_return

This is NOT a job cache returner, it was designed to send state.sls and state.highstate events to a log aggregator like Splunk
or ELK.

'''

from __future__ import absolute_import

# Import python libs
import json
import socket
import logging

# Import Salt libs
import salt.utils.jid
import salt.returners

log = logging.getLogger(__name__)

# Define virtual name
__virtualname__ = 'tcp_return'


def __virtual__():
    return __virtualname__


def _get_options(ret=None):
    attrs = {'host': 'host',
             'port': 'port'}
    _options = salt.returners.get_returner_options('returner.{0}'.format
                                                   (__virtualname__),
                                                   ret,
                                                   attrs,
                                                   __salt__=__salt__,
                                                   __opts__=__opts__)
    return _options


def _return_states(data, host, port):
    if data.get('fun') == 'state.sls' or data.get('fun') == 'state.highstate':
        log.info('{0}: Sending event_return to host "{1}" on port "{2}"'.format(__virtualname__, host, port))
        log.debug('{0}: {1}'.format(__virtualname__, data))
        for state_name, state in data.get('return').iteritems():
            connection = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            connection.settimeout(5.0)
            connection.connect((host, port))
            # Add extra data to state event
            state.update({'state_name': state_name,
                          'state_id': state_name.split('_|-')[1],
                          'minion_id': data.get('id'),
                          'jid': data.get('jid'),
                          'fun': data.get('fun'),
                          'fun_args': data.get('fun_args')})
            connection.sendall(json.dumps(state))
            connection.shutdown(socket.SHUT_RDWR)
            connection.close()


def event_return(events):
    _options = _get_options()
    host = _options.get('host')
    port = _options.get('port')
    for event in events:
        data = event.get('data', {})
        _return_states(data, host, port)