Gecko [ goons4good.org ]

Name	Size	Permission	Date
__pycache__	[ DIR ]	drwxr-xr-x	2022-10-11 05:09
__init__.py	3.62 KB	-rw-r--r--	2022-05-16 09:16
docker_events.py	2.84 KB	-rw-r--r--	2022-05-16 09:16
fluent.py	2.05 KB	-rw-r--r--	2022-05-16 09:16
http_logstash.py	2.84 KB	-rw-r--r--	2022-05-16 09:16
ircbot.py	10.22 KB	-rw-r--r--	2022-05-16 09:16
junos_syslog.py	13.18 KB	-rw-r--r--	2022-05-16 09:16
libvirt_events.py	21.28 KB	-rw-r--r--	2022-05-16 09:16
logentries.py	5.64 KB	-rw-r--r--	2022-05-16 09:16
logstash_engine.py	1.77 KB	-rw-r--r--	2022-05-16 09:16
napalm_syslog.py	11.59 KB	-rw-r--r--	2022-05-16 09:16
reactor.py	720 B	-rw-r--r--	2022-05-16 09:16
redis_sentinel.py	2.88 KB	-rw-r--r--	2022-05-16 09:16
script.py	3.33 KB	-rw-r--r--	2022-05-16 09:16
slack.py	35.79 KB	-rw-r--r--	2022-05-16 09:16
sqs_events.py	4.65 KB	-rw-r--r--	2022-05-16 09:16
stalekey.py	3.67 KB	-rw-r--r--	2022-05-16 09:16
test.py	932 B	-rw-r--r--	2022-05-16 09:16
thorium.py	314 B	-rw-r--r--	2022-05-16 09:16
webhook.py	2.78 KB	-rw-r--r--	2022-05-16 09:16

Rename

"""
An engine that reads messages from the redis sentinel pubsub and sends reactor
events based on the channels they are subscribed to.

.. versionadded:: 2016.3.0

:configuration:

Example configuration

.. code-block:: yaml

engines:
          - redis_sentinel:
              hosts:
                matching: 'board*'
                port: 26379
                interface: eth2
              channels:
                - '+switch-master'
                - '+odown'
                - '-odown'

:depends: redis
"""

import logging

import salt.client

try:
    import redis
except ImportError:
    redis = None

log = logging.getLogger(__name__)

__virtualname__ = "redis"

def __virtual__():
    return (
        __virtualname__
        if redis is not None
        else (False, "redis python module is not installed")
    )

class Listener:
    def __init__(self, host=None, port=None, channels=None, tag=None):
        if host is None:
            host = "localhost"
        if port is None:
            port = 26379
        if channels is None:
            channels = ["*"]
        if tag is None:
            tag = "salt/engine/redis_sentinel"
        super().__init__()
        self.tag = tag
        self.redis = redis.StrictRedis(host=host, port=port, decode_responses=True)
        self.pubsub = self.redis.pubsub()
        self.pubsub.psubscribe(channels)
        self.fire_master = salt.utils.event.get_master_event(
            __opts__, __opts__["sock_dir"]
        ).fire_event

def work(self, item):
        ret = {"channel": item["channel"]}
        if isinstance(item["data"], int):
            ret["code"] = item["data"]
        elif item["channel"] == "+switch-master":
            ret.update(
                dict(
                    list(
                        zip(
                            ("master", "old_host", "old_port", "new_host", "new_port"),
                            item["data"].split(" "),
                        )
                    )
                )
            )
        elif item["channel"] in ("+odown", "-odown"):
            ret.update(
                dict(list(zip(("master", "host", "port"), item["data"].split(" ")[1:])))
            )
        else:
            ret = {
                "channel": item["channel"],
                "data": item["data"],
            }
        self.fire_master(ret, "{}/{}".format(self.tag, item["channel"]))

def run(self):
        log.debug("Start Listener")
        for item in self.pubsub.listen():
            log.debug("Item: %s", item)
            self.work(item)

def start(hosts, channels, tag=None):
    if tag is None:
        tag = "salt/engine/redis_sentinel"
    with salt.client.LocalClient() as local:
        ips = local.cmd(
            hosts["matching"], "network.ip_addrs", [hosts["interface"]]
        ).values()
    client = Listener(host=ips.pop()[0], port=hosts["port"], channels=channels, tag=tag)
    client.run()