File manager

File manager - Edit - /opt/saltstack/salt/lib/python3.10/site-packages/salt/transport/zeromq.py

Back
""" Zeromq transport classes """ import datetime import errno import hashlib import logging import os import signal import socket import sys import threading from random import randint import zmq.error import zmq.eventloop.future import zmq.eventloop.zmqstream import salt.ext.tornado import salt.ext.tornado.concurrent import salt.ext.tornado.gen import salt.ext.tornado.ioloop import salt.ext.tornado.locks import salt.ext.tornado.queues import salt.payload import salt.transport.base import salt.utils.files import salt.utils.process import salt.utils.stringutils import salt.utils.zeromq from salt._compat import ipaddress from salt.exceptions import SaltException, SaltReqTimeoutError from salt.utils.zeromq import LIBZMQ_VERSION_INFO, ZMQ_VERSION_INFO, zmq try: import zmq.utils.monitor HAS_ZMQ_MONITOR = True except ImportError: HAS_ZMQ_MONITOR = False log = logging.getLogger(__name__) REQUEST_TIMEOUT = 60 # Payload marker for AsyncReqMessageClient queue: stop _send_recv gracefully. _REQ_QUEUE_SHUTDOWN = object() def _get_master_uri(master_ip, master_port, source_ip=None, source_port=None): """ Return the ZeroMQ URI to connect the Minion to the Master. It supports different source IP / port, given the ZeroMQ syntax: // Connecting using a IP address and bind to an IP address rc = zmq_connect(socket, "tcp://192.168.1.17:5555;192.168.1.1:5555"); assert (rc == 0); Source: http://api.zeromq.org/4-1:zmq-tcp """ from salt.utils.network import ip_bracket master_uri = "tcp://{master_ip}:{master_port}".format( master_ip=ip_bracket(master_ip), master_port=master_port ) if source_ip or source_port: if LIBZMQ_VERSION_INFO >= (4, 1, 6) and ZMQ_VERSION_INFO >= (16, 0, 1): # The source:port syntax for ZeroMQ has been added in libzmq 4.1.6 # which is included in the pyzmq wheels starting with 16.0.1. if source_ip and source_port: master_uri = ( "tcp://{source_ip}:{source_port};{master_ip}:{master_port}".format( source_ip=ip_bracket(source_ip), source_port=source_port, master_ip=ip_bracket(master_ip), master_port=master_port, ) ) elif source_ip and not source_port: master_uri = "tcp://{source_ip}:0;{master_ip}:{master_port}".format( source_ip=ip_bracket(source_ip), master_ip=ip_bracket(master_ip), master_port=master_port, ) elif source_port and not source_ip: ip_any = ( "0.0.0.0" if ipaddress.ip_address(master_ip).version == 4 else ip_bracket("::") ) master_uri = ( "tcp://{ip_any}:{source_port};{master_ip}:{master_port}".format( ip_any=ip_any, source_port=source_port, master_ip=ip_bracket(master_ip), master_port=master_port, ) ) else: log.warning( "Unable to connect to the Master using a specific source IP / port" ) log.warning("Consider upgrading to pyzmq >= 16.0.1 and libzmq >= 4.1.6") log.warning( "Specific source IP / port for connecting to master returner port:" " configuraion ignored" ) return master_uri class PublishClient(salt.transport.base.PublishClient): """ A transport channel backed by ZeroMQ for a Salt Publisher to use to publish commands to connected minions """ ttype = "zeromq" def __init__(self, opts, io_loop, kwargs): super().__init__(opts, io_loop, kwargs) self.opts = opts self.io_loop = io_loop self.hexid = hashlib.sha1( salt.utils.stringutils.to_bytes(self.opts["id"]) ).hexdigest() self._closing = False self.context = zmq.Context() self._socket = self.context.socket(zmq.SUB) if self.opts["zmq_filtering"]: # TODO: constants file for "broadcast" self._socket.setsockopt(zmq.SUBSCRIBE, b"broadcast") if self.opts.get("__role") == "syndic": self._socket.setsockopt(zmq.SUBSCRIBE, b"syndic") else: self._socket.setsockopt( zmq.SUBSCRIBE, salt.utils.stringutils.to_bytes(self.hexid) ) else: self._socket.setsockopt(zmq.SUBSCRIBE, b"") self._socket.setsockopt( zmq.IDENTITY, salt.utils.stringutils.to_bytes(self.opts["id"]) ) # TODO: cleanup all the socket opts stuff if hasattr(zmq, "TCP_KEEPALIVE"): self._socket.setsockopt(zmq.TCP_KEEPALIVE, self.opts["tcp_keepalive"]) self._socket.setsockopt( zmq.TCP_KEEPALIVE_IDLE, self.opts["tcp_keepalive_idle"] ) self._socket.setsockopt( zmq.TCP_KEEPALIVE_CNT, self.opts["tcp_keepalive_cnt"] ) self._socket.setsockopt( zmq.TCP_KEEPALIVE_INTVL, self.opts["tcp_keepalive_intvl"] ) recon_delay = self.opts["recon_default"] if self.opts["recon_randomize"]: recon_delay = randint( self.opts["recon_default"], self.opts["recon_default"] + self.opts["recon_max"], ) log.debug( "Generated random reconnect delay between '%sms' and '%sms' (%s)", self.opts["recon_default"], self.opts["recon_default"] + self.opts["recon_max"], recon_delay, ) log.debug("Setting zmq_reconnect_ivl to '%sms'", recon_delay) self._socket.setsockopt(zmq.RECONNECT_IVL, recon_delay) if hasattr(zmq, "RECONNECT_IVL_MAX"): log.debug( "Setting zmq_reconnect_ivl_max to '%sms'", self.opts["recon_default"] + self.opts["recon_max"], ) self._socket.setsockopt(zmq.RECONNECT_IVL_MAX, self.opts["recon_max"]) if (self.opts["ipv6"] is True or ":" in self.opts["master_ip"]) and hasattr( zmq, "IPV4ONLY" ): # IPv6 sockets work for both IPv6 and IPv4 addresses self._socket.setsockopt(zmq.IPV4ONLY, 0) if HAS_ZMQ_MONITOR and self.opts["zmq_monitor"]: self._monitor = ZeroMQSocketMonitor(self._socket) self._monitor.start_io_loop(self.io_loop) def close(self): if self._closing is True: return self._closing = True if hasattr(self, "_monitor") and self._monitor is not None: self._monitor.stop() self._monitor = None if hasattr(self, "_stream"): self._stream.close(0) elif hasattr(self, "_socket"): self._socket.close(0) if hasattr(self, "context") and self.context.closed is False: self.context.term() # pylint: enable=W1701 def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): self.close() # TODO: this is the time to see if we are connected, maybe use the req channel to guess? @salt.ext.tornado.gen.coroutine def connect(self, publish_port, connect_callback=None, disconnect_callback=None): self._connect_called = True self.publish_port = publish_port log.debug( "Connecting the Minion to the Master publish port, using the URI: %s", self.master_pub, ) log.debug("%r connecting to %s", self, self.master_pub) self._socket.connect(self.master_pub) if connect_callback is not None: connect_callback(True) @property def master_pub(self): """ Return the master publish port """ return _get_master_uri( self.opts["master_ip"], self.publish_port, source_ip=self.opts.get("source_ip"), source_port=self.opts.get("source_publish_port"), ) @salt.ext.tornado.gen.coroutine def _decode_messages(self, messages): """ Take the zmq messages, decrypt/decode them into a payload :param list messages: A list of messages to be decoded """ messages_len = len(messages) # if it was one message, then its old style if messages_len == 1: payload = salt.payload.loads(messages[0]) # 2 includes a header which says who should do it elif messages_len == 2: message_target = salt.utils.stringutils.to_str(messages[0]) if ( self.opts.get("__role") != "syndic" and message_target not in ("broadcast", self.hexid) ) or ( self.opts.get("__role") == "syndic" and message_target not in ("broadcast", "syndic") ): log.debug("Publish received for not this minion: %s", message_target) raise salt.ext.tornado.gen.Return(None) payload = salt.payload.loads(messages[1]) else: raise Exception( "Invalid number of messages ({}) in zeromq pubmessage from master".format( len(messages_len) ) ) # Yield control back to the caller. When the payload has been decoded, assign # the decoded payload to 'ret' and resume operation raise salt.ext.tornado.gen.Return(payload) @property def stream(self): """ Return the current zmqstream, creating one if necessary """ if not hasattr(self, "_stream"): self._stream = zmq.eventloop.zmqstream.ZMQStream( self._socket, io_loop=self.io_loop ) return self._stream def on_recv(self, callback): """ Register a callback for received messages (that we didn't initiate) :param func callback: A function which should be called when data is received """ if callback is None: # Caller wants to clear the callback — pass through directly. try: return self.stream.on_recv(None) except OSError as exc: if str(exc) == "Stream is closed": return raise # Wrap the callback so PyZMQ never sees an Awaitable return value. # Without this, when callback is a @gen.coroutine (e.g. the minion's # _handle_payload), PyZMQ's _run_callback does # `asyncio.ensure_future(callback_result)`, creating asyncio.Tasks on # the asyncio loop which is never driven by Tornado's IOLoop. Those # Tasks (plus their gen.Runner / Future / WeakRef tracking) accumulate # indefinitely. Routing through spawn_callback lets Tornado's own # _run_callback convert the coroutine into a Tornado Future and drive # it to completion natively, returning None to PyZMQ. io_loop = self.io_loop def _dispatch(args, kwargs): io_loop.spawn_callback(callback, args, kwargs) try: return self.stream.on_recv(_dispatch) except OSError as exc: if str(exc) == "Stream is closed": return raise @salt.ext.tornado.gen.coroutine def send(self, msg): self.stream.send(msg, noblock=True) class RequestServer(salt.transport.base.DaemonizedRequestServer): def __init__(self, opts): # pylint: disable=W0231 self.opts = opts self._closing = False self._monitor = None self._w_monitor = None def zmq_device(self): """ Multiprocessing target for the zmq queue device """ self.__setup_signals() # The first argument to zmq.Context is ``io_threads`` -- the # number of background I/O threads libzmq spawns -- not the # number of MWorker processes. Each libzmq I/O thread keeps # its own message-buffer pool that grows under sustained # traffic and is never released, so passing in # ``opts["worker_threads"]`` (typically 5-10) caused the # MWorkerQueue process RSS to climb ~7-8 MB/min indefinitely. # The QUEUE device only proxies two sockets; one I/O thread is # plenty. context = zmq.Context(1) # Prepare the zeromq sockets self.uri = "tcp://{interface}:{ret_port}".format(self.opts) self.clients = context.socket(zmq.ROUTER) # LINGER=-1 ("never discard") combined with the salt CLI's pattern # of one-shot connections (connect, send, recv, disconnect) caused # libzmq to retain undelivered queue slots for every disconnected # peer indefinitely under sustained CLI churn. A small finite # LINGER lets libzmq reap those slots. ROUTER_HANDOVER=1 makes # the router swap a stale peer (same routing-id, new connection) # instead of blocking on the old one -- relevant for minions that # reconnect after a brief network blip. TCP_KEEPALIVE forces # libzmq to notice peers that disappear without sending FIN, so # their queues are reaped instead of leaking until the OS default # 2-hour idle timer fires. self.clients.setsockopt(zmq.LINGER, 1000) if hasattr(zmq, "ROUTER_HANDOVER"): self.clients.setsockopt(zmq.ROUTER_HANDOVER, 1) self.clients.setsockopt(zmq.TCP_KEEPALIVE, 1) self.clients.setsockopt(zmq.TCP_KEEPALIVE_IDLE, 60) self.clients.setsockopt(zmq.TCP_KEEPALIVE_INTVL, 15) self.clients.setsockopt(zmq.TCP_KEEPALIVE_CNT, 3) if self.opts["ipv6"] is True and hasattr(zmq, "IPV4ONLY"): # IPv6 sockets work for both IPv6 and IPv4 addresses self.clients.setsockopt(zmq.IPV4ONLY, 0) self.clients.setsockopt(zmq.BACKLOG, self.opts.get("zmq_backlog", 1000)) self._start_zmq_monitor() self.workers = context.socket(zmq.DEALER) self.workers.setsockopt(zmq.LINGER, 1000) if self.opts["mworker_queue_niceness"] and not salt.utils.platform.is_windows(): log.info( "setting mworker_queue niceness to %d", self.opts["mworker_queue_niceness"], ) os.nice(self.opts["mworker_queue_niceness"]) if self.opts.get("ipc_mode", "") == "tcp": self.w_uri = "tcp://127.0.0.1:{}".format( self.opts.get("tcp_master_workers", 4515) ) else: self.w_uri = "ipc://{}".format( os.path.join(self.opts["sock_dir"], "workers.ipc") ) log.info("Setting up the master communication server") log.info("ReqServer clients %s", self.uri) self.clients.bind(self.uri) log.info("ReqServer workers %s", self.w_uri) self.workers.bind(self.w_uri) if self.opts.get("ipc_mode", "") != "tcp": os.chmod(os.path.join(self.opts["sock_dir"], "workers.ipc"), 0o600) while True: if self.clients.closed or self.workers.closed: break try: zmq.device(zmq.QUEUE, self.clients, self.workers) except zmq.ZMQError as exc: if exc.errno == errno.EINTR: continue raise except (KeyboardInterrupt, SystemExit): break context.term() def close(self): """ Cleanly shutdown the router socket """ if self._closing: return log.info("MWorkerQueue under PID %s is closing", os.getpid()) self._closing = True if getattr(self, "_monitor", None) is not None: self._monitor.stop() self._monitor = None if getattr(self, "_w_monitor", None) is not None: self._w_monitor.stop() self._w_monitor = None if hasattr(self, "clients") and self.clients.closed is False: self.clients.close() if hasattr(self, "workers") and self.workers.closed is False: self.workers.close() if hasattr(self, "stream"): self.stream.close() if hasattr(self, "_socket") and self._socket.closed is False: self._socket.close() if hasattr(self, "context") and self.context.closed is False: self.context.term() def pre_fork(self, process_manager): """ Pre-fork we need to create the zmq router device :param func process_manager: An instance of salt.utils.process.ProcessManager """ process_manager.add_process(self.zmq_device, name="MWorkerQueue") def _start_zmq_monitor(self): """ Starts ZMQ monitor for debugging purposes. :return: """ # Socket monitor shall be used the only for debug # purposes so using threading doesn't look too bad here if HAS_ZMQ_MONITOR and self.opts["zmq_monitor"]: log.debug("Starting ZMQ monitor") self._w_monitor = ZeroMQSocketMonitor(self._socket) threading.Thread(target=self._w_monitor.start_poll).start() log.debug("ZMQ monitor has been started started") def post_fork(self, message_handler, io_loop): """ After forking we need to create all of the local sockets to listen to the router :param func message_handler: A function to called to handle incoming payloads as they are picked up off the wire :param IOLoop io_loop: An instance of a Tornado IOLoop, to handle event scheduling """ context = zmq.Context(1) self._socket = context.socket(zmq.REP) # Linger -1 means we'll never discard messages. self._socket.setsockopt(zmq.LINGER, -1) self._start_zmq_monitor() if self.opts.get("ipc_mode", "") == "tcp": self.w_uri = "tcp://127.0.0.1:{}".format( self.opts.get("tcp_master_workers", 4515) ) else: self.w_uri = "ipc://{}".format( os.path.join(self.opts["sock_dir"], "workers.ipc") ) log.info("Worker binding to socket %s", self.w_uri) self._socket.connect(self.w_uri) if self.opts.get("ipc_mode", "") != "tcp" and os.path.isfile( os.path.join(self.opts["sock_dir"], "workers.ipc") ): os.chmod(os.path.join(self.opts["sock_dir"], "workers.ipc"), 0o600) self.stream = zmq.eventloop.zmqstream.ZMQStream(self._socket, io_loop=io_loop) self.message_handler = message_handler def _dispatch_handle_message(stream, payload): # Drive the coroutine via Tornado's IOLoop rather than returning # it to PyZMQ's _run_callback. PyZMQ wraps any Awaitable return # value with asyncio.ensure_future, creating Tasks on the asyncio # event loop which is never driven in MWorkers — causing permanent # Task accumulation. Routing through spawn_callback lets Tornado's # own _run_callback convert it to a Tornado Future and drive it to # completion without touching asyncio. io_loop.spawn_callback(self.handle_message, stream, payload) self.stream.on_recv_stream(_dispatch_handle_message) @salt.ext.tornado.gen.coroutine def handle_message(self, stream, payload): try: payload = self.decode_payload(payload) except salt.exceptions.SaltDeserializationError: self.stream.send(self.encode_payload({"msg": "bad load"})) return # XXX: Is header really needed? reply = yield self.message_handler(payload) self.stream.send(self.encode_payload(reply)) def encode_payload(self, payload): return salt.payload.dumps(payload) def __setup_signals(self): signal.signal(signal.SIGINT, self._handle_signals) signal.signal(signal.SIGTERM, self._handle_signals) def _handle_signals(self, signum, sigframe): msg = f"{self.__class__.__name__} received a " if signum == signal.SIGINT: msg += "SIGINT" elif signum == signal.SIGTERM: msg += "SIGTERM" msg += ". Exiting" log.debug(msg) self.close() sys.exit(salt.defaults.exitcodes.EX_OK) def decode_payload(self, payload): payload = salt.payload.loads(payload[0]) return payload def _set_tcp_keepalive(zmq_socket, opts): """ Ensure that TCP keepalives are set as specified in "opts". Warning: Failure to set TCP keepalives on the salt-master can result in not detecting the loss of a minion when the connection is lost or when its host has been terminated without first closing the socket. Salt's Presence System depends on this connection status to know if a minion is "present". Warning: Failure to set TCP keepalives on minions can result in frequent or unexpected disconnects! """ if hasattr(zmq, "TCP_KEEPALIVE") and opts: if "tcp_keepalive" in opts: zmq_socket.setsockopt(zmq.TCP_KEEPALIVE, opts["tcp_keepalive"]) if "tcp_keepalive_idle" in opts: zmq_socket.setsockopt(zmq.TCP_KEEPALIVE_IDLE, opts["tcp_keepalive_idle"]) if "tcp_keepalive_cnt" in opts: zmq_socket.setsockopt(zmq.TCP_KEEPALIVE_CNT, opts["tcp_keepalive_cnt"]) if "tcp_keepalive_intvl" in opts: zmq_socket.setsockopt(zmq.TCP_KEEPALIVE_INTVL, opts["tcp_keepalive_intvl"]) # TODO: unit tests! class AsyncReqMessageClient: """ This class wraps the underlying zeromq REQ socket and gives a future-based interface to sending and recieving messages. This works around the primary limitation of serialized send/recv on the underlying socket by queueing the message sends in this class. In the future if we decide to attempt to multiplex we can manage a pool of REQ/REP sockets-- but for now we'll just do them in serial """ def __init__(self, opts, addr, linger=0, io_loop=None): """ Create an asynchronous message client :param dict opts: The salt opts dictionary :param str addr: The interface IP address to bind to :param int linger: The number of seconds to linger on a ZMQ socket. See http://api.zeromq.org/2-1:zmq-setsockopt [ZMQ_LINGER] :param IOLoop io_loop: A Tornado IOLoop event scheduler [tornado.ioloop.IOLoop] """ self.opts = opts self.addr = addr self.linger = linger if io_loop is None: self.io_loop = salt.ext.tornado.ioloop.IOLoop.current() else: self.io_loop = io_loop self.context = zmq.eventloop.future.Context() self.socket = None self._closed = False self._queue = salt.ext.tornado.queues.Queue() self._send_recv_exit_future = None def connect(self): if self.context is None: self.context = zmq.eventloop.future.Context() if hasattr(self, "socket") and self.socket: return # wire up sockets self._init_socket() def _init_socket(self): self.socket = self.context.socket(zmq.REQ) # socket options if hasattr(zmq, "RECONNECT_IVL_MAX"): self.socket.setsockopt(zmq.RECONNECT_IVL_MAX, 5000) # Set a stable ZMQ routing identity so the master's ROUTER socket # reuses an existing slot for this caller (combined with # ROUTER_HANDOVER=1 on the master) rather than allocating a new # entry in its per-peer table for every CLI invocation. Without # this, the master's libzmq peer-id hashtable grows unbounded # under sustained CLI churn (about 6 MB/min in stress). # # Only do this for salt CLI tools (which do NOT set ``__role`` in # opts). All long-lived daemons -- minion, syndic, master -- # open multiple AsyncReqMessageClient instances concurrently from # a single process: the minion at startup for auth + pillar + # file requests, the syndic when relaying multiple downstream # minions' returns upstream, and a master when forwarding to # peer masters. Giving them all the same stable identity would # cause ROUTER_HANDOVER on the upstream ROUTER to silently drop # any reply still in flight to the previous REQ as each new one # arrived, hanging startup and breaking syndic relays. Their # own REQ churn is bounded anyway (one peer per daemon), so they # can keep using libzmq's default per-connection random # routing-ids. if not self.opts.get("__role"): role = self.opts.get("id") or "clir" try: uid = os.getuid() except AttributeError: # Windows uid = 0 identity = "salt-req/{role}/{host}/{uid}/{slot}".format( role=role, host=socket.gethostname(), uid=uid, slot=os.getpid() % 256, ) self.socket.setsockopt(zmq.IDENTITY, identity.encode("utf-8")) _set_tcp_keepalive(self.socket, self.opts) if self.addr.startswith("tcp://["): # Hint PF type if bracket enclosed IPv6 address if hasattr(zmq, "IPV6"): self.socket.setsockopt(zmq.IPV6, 1) elif hasattr(zmq, "IPV4ONLY"): self.socket.setsockopt(zmq.IPV4ONLY, 0) self.socket.setsockopt(zmq.LINGER, self.linger) self.socket.connect(self.addr) self.io_loop.spawn_callback(self._send_recv, self.socket) def _close_zmq_only(self): """Close socket and ZMQ context only (no IOLoop / _send_recv coordination).""" if hasattr(self, "socket") and self.socket is not None: self.socket.close(0) self.socket = None if self.context is not None and self.context.closed is False: self.context.term() self.context = None def close_future(self): """ Return a ``Future`` that completes after ZMQ resources are released. Coroutine callers on the owning I/O loop thread should ``yield`` this future (via ``salt.ext.tornado.gen.convert_yielded``) before allocating another client on that loop—``close()`` schedules teardown asynchronously in that case. """ self._initiate_async_req_close() return self._close_completed_future def close(self): """ Stop the send/recv coroutine and close ZMQ resources. Safe to call more than once. When no I/O loop iteration is active, uses ``run_sync`` around the graceful shutdown coroutine (LocalClient/SyncWrapper and similar). When the owning ``io_loop`` is already running—including on the master event loop thread while short-lived REQ clients are torn down—you cannot call ``run_sync``. In that case the shutdown future is queued with ``add_callback`` / ``add_future``. If ``close()`` is invoked from another thread while the loop is running, we block until shutdown completes. On the owning I/O loop thread, prefer ``yield``-ing ``close_future()`` instead of ``close()`` when you recreate clients immediately afterward. """ cross_thread_evt = self._initiate_async_req_close() if cross_thread_evt is not None: cross_thread_evt.wait(timeout=30) def _mark_teardown_finished(self): fut = getattr(self, "_close_completed_future", None) if fut is not None and not fut.done(): fut.set_result(None) def _initiate_async_req_close(self): """ Start teardown once. Sets ``self._close_completed_future`` and returns ``None``, or an ``threading.Event`` that unblocks after ``finalize()`` when the caller needs to wait from a non-I/O-loop thread. """ if getattr(self, "_close_completed_future", None) is not None: return None self._close_completed_future = salt.ext.tornado.concurrent.Future() cross_thread_evt = None def finalize(): self._send_recv_exit_future = None self._close_zmq_only() self._mark_teardown_finished() if cross_thread_evt is not None: cross_thread_evt.set() def run_shutdown(): return self._graceful_shutdown_coro() if self._closed: finalize() return None self._closed = True if self.socket is None: finalize() return None self._send_recv_exit_future = salt.ext.tornado.concurrent.Future() # _running and _thread_ident are upstream Tornado internals (mirrored in # salt.ext.tornado). Prefer them for a fast path before run_sync; if a # Tornado upgrade breaks this, re-check this branch and the RuntimeError # "already running" fallback below. try: if not getattr(self.io_loop, "_running", False): self.io_loop.run_sync(run_shutdown, timeout=30) finalize() return None except RuntimeError as exc: if "already running" not in str(exc).lower(): log.debug( "REQ client shutdown: run_sync aborted: %s", exc, exc_info=True, ) finalize() return None except salt.ext.tornado.ioloop.TimeoutError: log.debug("Graceful REQ message client shutdown timed out during run_sync") finalize() return None except Exception: # pylint: disable=broad-except log.debug( "Graceful REQ message client shutdown failed during run_sync", exc_info=True, ) finalize() return None try: shutdown_future = salt.ext.tornado.gen.convert_yielded( self._graceful_shutdown_coro() ) except Exception: # pylint: disable=broad-except log.debug( "Could not schedule REQ shutdown coroutine while loop is running", exc_info=True, ) finalize() return None ioloop_thread = getattr(self.io_loop, "_thread_ident", None) same_thread = ioloop_thread == threading.get_ident() cross_thread_evt = None if same_thread else threading.Event() def on_done(future): try: future.result() except Exception: # pylint: disable=broad-except log.debug( "Graceful REQ message client shutdown failed during async teardown", exc_info=True, ) finally: finalize() def schedule(): self.io_loop.add_future(shutdown_future, on_done) try: self.io_loop.add_callback(schedule) except Exception: # pylint: disable=broad-except log.debug( "Graceful REQ message client shutdown failed scheduling callback", exc_info=True, ) finalize() return None return cross_thread_evt @salt.ext.tornado.gen.coroutine def _graceful_shutdown_coro(self): try: self._queue.put_nowait( (salt.ext.tornado.concurrent.Future(), _REQ_QUEUE_SHUTDOWN) ) except Exception: # pylint: disable=broad-except log.debug("Could not queue shutdown sentinel", exc_info=True) if self._send_recv_exit_future and not self._send_recv_exit_future.done(): self._send_recv_exit_future.set_result(None) raise salt.ext.tornado.gen.Return() yield self._send_recv_exit_future @salt.ext.tornado.gen.coroutine def send(self, message, timeout=None, callback=None): """ Return a future which will be completed when the message has a response """ future = salt.ext.tornado.concurrent.Future() message = salt.payload.dumps(message) self._queue.put_nowait((future, message)) send_timeout = None if callback is not None: def handle_future(future): response = future.result() self.io_loop.add_callback(callback, response) future.add_done_callback(handle_future) if self.opts.get("detect_mode") is True: timeout = 1 if timeout is not None: send_timeout = self.io_loop.call_later( timeout, self._timeout_message, future ) try: recv = yield future finally: if send_timeout is not None: self.io_loop.remove_timeout(send_timeout) raise salt.ext.tornado.gen.Return(recv) def _timeout_message(self, future): if not future.done(): future.set_exception(SaltReqTimeoutError("Message timed out")) @salt.ext.tornado.gen.coroutine def _send_recv(self, socket, _TimeoutError=salt.ext.tornado.gen.TimeoutError): """ Long-running send/receive coroutine. This should be started once for each socket created. Once started, the coroutine will run until the socket is closed. A future and message are pulled from the queue. The message is sent and the reply socket is polled for a response while checking the future to see if it was timed out. """ send_recv_running = True # Hold on to the socket so we'll still have a reference to it after the # close method is called. This allows us to fail gracefully once it's # been closed. try: while send_recv_running: try: future, message = yield self._queue.get( timeout=datetime.timedelta(milliseconds=300) ) except _TimeoutError: try: # For some reason yielding here doesn't work becaues the # future always has a result? poll_future = socket.poll(0, zmq.POLLOUT) poll_future.result() except _TimeoutError: # This is what we expect if the socket is still alive pass except zmq.eventloop.future.CancelledError: log.trace("Loop closed while polling send socket.") # The ioloop was closed before polling finished. send_recv_running = False break except zmq.ZMQError: log.trace("Send socket closed while polling.") send_recv_running = False break continue if message is _REQ_QUEUE_SHUTDOWN: send_recv_running = False break try: yield socket.send(message) except zmq.eventloop.future.CancelledError as exc: log.trace("Loop closed while sending.") # The ioloop was closed before polling finished. send_recv_running = False if not future.done(): future.set_exception(exc) break except zmq.ZMQError as exc: if exc.errno in [ zmq.ENOTSOCK, zmq.ETERM, zmq.error.EINTR, ]: log.trace("Send socket closed while sending.") send_recv_running = False if not future.done(): future.set_exception(exc) elif exc.errno == zmq.EFSM: log.error("Socket was found in invalid state.") send_recv_running = False if not future.done(): future.set_exception(exc) else: log.error( "Unhandled Zeromq error durring send/receive: %s", exc ) if not future.done(): future.set_exception(exc) if future.done(): exc = future.exception() if isinstance(exc, SaltReqTimeoutError): log.trace("Request timed out while sending. reconnecting.") else: log.trace( "The request ended with an error while sending. reconnecting." ) # Only reconnect if the client is still active. If close() was # already called externally (context is None), do not create a # new socket/context that would never be cleaned up. _should_reconnect = self.context is not None self._close_zmq_only() if _should_reconnect: self.connect() send_recv_running = False break received = False ready = False while True: if future.done(): break try: # Time is in milliseconds. ready = yield socket.poll(300, zmq.POLLIN) except zmq.eventloop.future.CancelledError as exc: log.trace( "Loop closed while polling receive socket.", exc_info=True ) log.error("Master is unavailable (Connection Cancelled).") send_recv_running = False if not future.done(): future.set_result(None) except zmq.ZMQError as exc: log.trace("Receive socket closed while polling.") send_recv_running = False if not future.done(): future.set_exception(exc) if ready: try: recv = yield socket.recv() received = True except zmq.eventloop.future.CancelledError as exc: log.trace("Loop closed while receiving.") send_recv_running = False if not future.done(): future.set_exception(exc) except zmq.ZMQError as exc: log.trace("Receive socket closed while receiving.") send_recv_running = False if not future.done(): future.set_exception(exc) break elif future.done(): break if future.done(): exc = future.exception() if isinstance(exc, SaltReqTimeoutError): log.trace( "Request timed out while waiting for a response. reconnecting." ) else: log.trace("The request ended with an error. reconnecting.") # Only reconnect if the client is still active. If close() was # already called externally (context is None), do not create a # new socket/context that would never be cleaned up. _should_reconnect = self.context is not None self._close_zmq_only() if _should_reconnect: self.connect() send_recv_running = False elif received: data = salt.payload.loads(recv) if not future.done(): future.set_result(data) finally: if ( self._send_recv_exit_future is not None and not self._send_recv_exit_future.done() ): self._send_recv_exit_future.set_result(None) log.trace("Send and receive coroutine ending %s", socket) class ZeroMQSocketMonitor: __EVENT_MAP = None def __init__(self, socket): """ Create ZMQ monitor sockets More information: http://api.zeromq.org/4-0:zmq-socket-monitor """ self._socket = socket self._monitor_socket = self._socket.get_monitor_socket() self._monitor_stream = None def start_io_loop(self, io_loop): log.trace("Event monitor start!") self._monitor_stream = zmq.eventloop.zmqstream.ZMQStream( self._monitor_socket, io_loop=io_loop ) self._monitor_stream.on_recv(self.monitor_callback) def start_poll(self): log.trace("Event monitor start!") try: while self._monitor_socket is not None and self._monitor_socket.poll(): msg = self._monitor_socket.recv_multipart() self.monitor_callback(msg) except (AttributeError, zmq.error.ContextTerminated): # We cannot log here because we'll get an interrupted system call in trying # to flush the logging buffer as we terminate pass @property def event_map(self): if ZeroMQSocketMonitor.__EVENT_MAP is None: event_map = {} for name in dir(zmq): if name.startswith("EVENT_"): value = getattr(zmq, name) event_map[value] = name ZeroMQSocketMonitor.__EVENT_MAP = event_map return ZeroMQSocketMonitor.__EVENT_MAP def monitor_callback(self, msg): evt = zmq.utils.monitor.parse_monitor_message(msg) evt["description"] = self.event_map[evt["event"]] log.debug("ZeroMQ event: %s", evt) if evt["event"] == zmq.EVENT_MONITOR_STOPPED: self.stop() def stop(self): if self._socket is None: return try: self._socket.disable_monitor() except zmq.Error: pass self._socket = None if self._monitor_stream is not None: self._monitor_stream.close() self._monitor_stream = None if self._monitor_socket is not None: self._monitor_socket.close() self._monitor_socket = None log.trace("Event monitor done!") class PublishServer(salt.transport.base.DaemonizedPublishServer): """ Encapsulate synchronous operations for a publisher channel """ _sock_data = threading.local() def __init__(self, opts): self.opts = opts def connect(self): return salt.ext.tornado.gen.sleep(5) def publish_daemon( self, publish_payload, presence_callback=None, remove_presence_callback=None, ): """ This method represents the Publish Daemon process. It is intended to be run in a thread or process as it creates and runs its own ioloop. """ ioloop = salt.ext.tornado.ioloop.IOLoop() ioloop.make_current() self.io_loop = ioloop context = zmq.Context(1) pub_sock = context.socket(zmq.PUB) monitor = ZeroMQSocketMonitor(pub_sock) monitor.start_io_loop(ioloop) _set_tcp_keepalive(pub_sock, self.opts) self.dpub_sock = pub_sock = zmq.eventloop.zmqstream.ZMQStream(pub_sock) # if 2.1 >= zmq < 3.0, we only have one HWM setting try: pub_sock.setsockopt(zmq.HWM, self.opts.get("pub_hwm", 1000)) # in zmq >= 3.0, there are separate send and receive HWM settings except (AttributeError, zmq.error.ZMQError): # Set the High Water Marks. For more information on HWM, see: # http://api.zeromq.org/4-1:zmq-setsockopt pub_sock.setsockopt(zmq.SNDHWM, self.opts.get("pub_hwm", 1000)) pub_sock.setsockopt(zmq.RCVHWM, self.opts.get("pub_hwm", 1000)) if self.opts["ipv6"] is True and hasattr(zmq, "IPV4ONLY"): # IPv6 sockets work for both IPv6 and IPv4 addresses pub_sock.setsockopt(zmq.IPV4ONLY, 0) pub_sock.setsockopt(zmq.BACKLOG, self.opts.get("zmq_backlog", 1000)) pub_sock.setsockopt(zmq.LINGER, -1) # Prepare minion pull socket pull_sock = context.socket(zmq.PULL) pull_sock = zmq.eventloop.zmqstream.ZMQStream(pull_sock) pull_sock.setsockopt(zmq.LINGER, -1) salt.utils.zeromq.check_ipc_path_max_len(self.pull_uri) # Start the minion command publisher log.info("Starting the Salt Publisher on %s", self.pub_uri) pub_sock.bind(self.pub_uri) # Securely create socket log.info("Starting the Salt Puller on %s", self.pull_uri) with salt.utils.files.set_umask(0o177): pull_sock.bind(self.pull_uri) @salt.ext.tornado.gen.coroutine def on_recv(packages): try: for package in packages: payload = salt.payload.loads(package) yield publish_payload(payload) except Exception as exc: # pylint: disable=broad-except log.error( "Un-handled error in publisher %s", exc, exc_info_on_loglevel=logging.DEBUG, ) def _dispatch_on_recv(packages): # Same fix as in RequestServer: route through Tornado's IOLoop # instead of returning the coroutine to PyZMQ's _run_callback, # which would wrap it with asyncio.ensure_future. ioloop.spawn_callback(on_recv, packages) pull_sock.on_recv(_dispatch_on_recv) try: ioloop.start() except (KeyboardInterrupt, SystemExit): pass finally: pub_sock.close() pull_sock.close() @property def pull_uri(self): if self.opts.get("ipc_mode", "") == "tcp": pull_uri = "tcp://127.0.0.1:{}".format( self.opts.get("tcp_master_publish_pull", 4514) ) else: pull_uri = "ipc://{}".format( os.path.join(self.opts["sock_dir"], "publish_pull.ipc") ) return pull_uri @property def pub_uri(self): return "tcp://{interface}:{publish_port}".format(self.opts) @salt.ext.tornado.gen.coroutine def publish_payload(self, payload, topic_list=None): payload = salt.payload.dumps(payload) if self.opts["zmq_filtering"]: if topic_list: for topic in topic_list: log.trace("Sending filtered data over publisher %s", self.pub_uri) # zmq filters are substring match, hash the topic # to avoid collisions htopic = salt.utils.stringutils.to_bytes( hashlib.sha1(salt.utils.stringutils.to_bytes(topic)).hexdigest() ) yield self.dpub_sock.send_multipart([htopic, payload]) log.trace("Filtered data has been sent") # Syndic broadcast if self.opts.get("order_masters"): log.trace("Sending filtered data to syndic") yield self.dpub_sock.send_multipart([b"syndic", payload]) log.trace("Filtered data has been sent to syndic") # otherwise its a broadcast else: # TODO: constants file for "broadcast" log.trace("Sending broadcasted data over publisher %s", self.pub_uri) yield self.dpub_sock.send_multipart([b"broadcast", payload]) log.trace("Broadcasted data has been sent") else: log.trace("Sending ZMQ-unfiltered data over publisher %s", self.pub_uri) yield self.dpub_sock.send(payload) log.trace("Unfiltered data has been sent") def pre_fork(self, process_manager): """ Do anything necessary pre-fork. Since this is on the master side this will primarily be used to create IPC channels and create our daemon process to do the actual publishing :param func process_manager: A ProcessManager, from salt.utils.process.ProcessManager """ process_manager.add_process( self.publish_daemon, args=(self.publish_payload,), ) @property def pub_sock(self): """ This thread's zmq publisher socket. This socket is stored on the class so that multiple instantiations in the same thread will re-use a single zmq socket. """ try: return self._sock_data.sock except AttributeError: pass def pub_connect(self): """ Create and connect this thread's zmq socket. If a publisher socket already exists "pub_close" is called before creating and connecting a new socket. """ if self.pub_sock: self.pub_close() ctx = zmq.Context() self._sock_data.sock = ctx.socket(zmq.PUSH) self.pub_sock.setsockopt(zmq.LINGER, -1) if self.opts.get("ipc_mode", "") == "tcp": pull_uri = "tcp://127.0.0.1:{}".format( self.opts.get("tcp_master_publish_pull", 4514) ) else: pull_uri = "ipc://{}".format( os.path.join(self.opts["sock_dir"], "publish_pull.ipc") ) log.debug("Connecting to pub server: %s", pull_uri) self.pub_sock.connect(pull_uri) return self._sock_data.sock def pub_close(self): """ Disconnect an existing publisher socket and remove it from the local thread's cache. """ if hasattr(self._sock_data, "sock"): self._sock_data.sock.close() delattr(self._sock_data, "sock") def publish(self, payload, kwargs): """ Publish "load" to minions. This send the load to the publisher daemon process with does the actual sending to minions. :param dict load: A load to be sent across the wire to minions """ if not self.pub_sock: self.pub_connect() serialized = salt.payload.dumps(payload) self.pub_sock.send(serialized) log.debug("Sent payload to publish daemon.") @property def topic_support(self): return self.opts.get("zmq_filtering", False) def close(self): self.pub_close() def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): self.close() class RequestClient(salt.transport.base.RequestClient): ttype = "zeromq" def __init__(self, opts, io_loop): # pylint: disable=W0231 super().__init__(opts, io_loop) self.opts = opts master_uri = self.get_master_uri(opts) self.message_client = AsyncReqMessageClient( self.opts, master_uri, io_loop=io_loop, ) self._closing = False self._connect_called = False @salt.ext.tornado.gen.coroutine def connect(self): self._connect_called = True self.message_client.connect() @salt.ext.tornado.gen.coroutine def send(self, load, timeout=REQUEST_TIMEOUT): yield self.connect() ret = yield self.message_client.send(load, timeout=timeout) raise salt.ext.tornado.gen.Return(ret) def close_future(self): fut = self.message_client.close_future() self._closing = True return fut def close(self): if self._closing: return self._closing = True self.message_client.close() @staticmethod def get_master_uri(opts): if "master_uri" in opts: return opts["master_uri"] if "master_ip" in opts: return _get_master_uri( opts["master_ip"], opts["master_port"], source_ip=opts.get("source_ip"), source_port=opts.get("source_ret_port"), ) # if we've reached here something is very abnormal raise SaltException("ReqChannel: missing master_uri/master_ip in self.opts")

| ver. 1.4 | Github | . | PHP 8.2.30 | Generation time: 0 | proxy | phpinfo | Settings