Gecko [ goons4good.org ]

Name	Size	Permission	Date
__pycache__	[ DIR ]	drwxr-xr-x	2022-10-11 05:09
__init__.py	5.98 KB	-rw-r--r--	2022-05-16 09:16
appoptics_return.py	6.11 KB	-rw-r--r--	2022-05-16 09:16
carbon_return.py	8.24 KB	-rw-r--r--	2022-05-16 09:16
cassandra_cql_return.py	15.17 KB	-rw-r--r--	2022-05-16 09:16
cassandra_return.py	2.07 KB	-rw-r--r--	2022-05-16 09:16
couchbase_return.py	8.91 KB	-rw-r--r--	2022-05-16 09:16
couchdb_return.py	10.35 KB	-rw-r--r--	2022-05-16 09:16
django_return.py	2.27 KB	-rw-r--r--	2022-05-16 09:16
elasticsearch_return.py	12.18 KB	-rw-r--r--	2022-05-16 09:16
etcd_return.py	6.32 KB	-rw-r--r--	2022-05-16 09:16
highstate_return.py	15.59 KB	-rw-r--r--	2022-05-16 09:16
influxdb_return.py	8.03 KB	-rw-r--r--	2022-05-16 09:16
kafka_return.py	2.12 KB	-rw-r--r--	2022-05-16 09:16
librato_return.py	4.29 KB	-rw-r--r--	2022-05-16 09:16
local.py	541 B	-rw-r--r--	2022-05-16 09:16
local_cache.py	17.1 KB	-rw-r--r--	2022-05-16 09:16
mattermost_returner.py	4.08 KB	-rw-r--r--	2022-05-16 09:16
memcache_return.py	5.72 KB	-rw-r--r--	2022-05-16 09:16
mongo_future_return.py	10.08 KB	-rw-r--r--	2022-05-16 09:16
mongo_return.py	6.11 KB	-rw-r--r--	2022-05-16 09:16
multi_returner.py	2.82 KB	-rw-r--r--	2022-05-16 09:16
mysql.py	18.61 KB	-rw-r--r--	2022-05-16 09:16
nagios_nrdp_return.py	4.92 KB	-rw-r--r--	2022-05-16 09:16
odbc.py	7.77 KB	-rw-r--r--	2022-05-16 09:16
pgjsonb.py	17.18 KB	-rw-r--r--	2022-05-16 09:16
postgres.py	10.2 KB	-rw-r--r--	2022-05-16 09:16
postgres_local_cache.py	10.73 KB	-rw-r--r--	2022-05-16 09:16
pushover_returner.py	6.55 KB	-rw-r--r--	2022-05-16 09:16
rawfile_json.py	2.23 KB	-rw-r--r--	2022-05-16 09:16
redis_return.py	8.52 KB	-rw-r--r--	2022-05-16 09:16
sentry_return.py	5.3 KB	-rw-r--r--	2022-05-16 09:16
slack_returner.py	6.02 KB	-rw-r--r--	2022-05-16 09:16
slack_webhook_return.py	11.22 KB	-rw-r--r--	2022-05-16 09:16
sms_return.py	2.64 KB	-rw-r--r--	2022-05-16 09:16
smtp_return.py	8.2 KB	-rw-r--r--	2022-05-16 09:16
splunk.py	6.69 KB	-rw-r--r--	2022-05-16 09:16
sqlite3_return.py	7.73 KB	-rw-r--r--	2022-05-16 09:16
syslog_return.py	5.26 KB	-rw-r--r--	2022-05-16 09:16
telegram_return.py	1.92 KB	-rw-r--r--	2022-05-16 09:16
xmpp_return.py	4.85 KB	-rw-r--r--	2022-05-16 09:16
zabbix_return.py	2.45 KB	-rw-r--r--	2022-05-16 09:16

Rename

"""
Return data to a Kafka topic

:maintainer: Justin Desilets (justin.desilets@gmail.com)
:maturity: 20181119
:depends: confluent-kafka
:platform: all

To enable this returner install confluent-kafka and enable the following
settings in the minion config:

returner.kafka.bootstrap:
      - "server1:9092"
      - "server2:9092"
      - "server3:9092"

returner.kafka.topic: 'topic'

To use the kafka returner, append `--return kafka` to the Salt command, eg;

salt '*' test.ping --return kafka

"""

import logging

import salt.utils.json

try:
    from confluent_kafka import Producer

HAS_KAFKA = True
except ImportError:
    HAS_KAFKA = False

log = logging.getLogger(__name__)

__virtualname__ = "kafka"

def __virtual__():
    if not HAS_KAFKA:
        return (
            False,
            "Could not import kafka returner; confluent-kafka is not installed.",
        )
    return __virtualname__

def _get_conn():
    """
    Return a kafka connection
    """
    if __salt__["config.option"]("returner.kafka.bootstrap"):
        bootstrap = ",".join(__salt__["config.option"]("returner.kafka.bootstrap"))
    else:
        log.error("Unable to find kafka returner config option: bootstrap")
        return None
    return bootstrap

def _delivery_report(err, msg):
    """Called once for each message produced to indicate delivery result.
    Triggered by poll() or flush()."""
    if err is not None:
        log.error("Message delivery failed: %s", err)
    else:
        log.debug("Message delivered to %s [%s]", msg.topic(), msg.partition())

def returner(ret):
    """
    Return information to a Kafka server
    """
    if __salt__["config.option"]("returner.kafka.topic"):
        topic = __salt__["config.option"]("returner.kafka.topic")

conn = _get_conn()
        producer = Producer({"bootstrap.servers": conn})
        producer.poll(0)
        producer.produce(
            topic,
            salt.utils.json.dumps(ret),
            str(ret).encode("utf-8"),
            callback=_delivery_report,
        )

producer.flush()
    else:
        log.error("Unable to find kafka returner config option: topic")