""" Control the state system on the minion. State Caching ------------- When a highstate is called, the minion automatically caches a copy of the last high data. If you then run a highstate with cache=True it will use that cached highdata and won't hit the fileserver except for ``salt://`` links in the states themselves. """ import logging import os import shutil import sys import tarfile import tempfile import time from collections import OrderedDict import salt.config import salt.defaults.exitcodes import salt.payload import salt.state import salt.utils.args import salt.utils.atomicfile import salt.utils.data import salt.utils.event import salt.utils.files import salt.utils.functools import salt.utils.hashutils import salt.utils.jid import salt.utils.json import salt.utils.msgpack import salt.utils.platform import salt.utils.process import salt.utils.state import salt.utils.stringutils import salt.utils.url import salt.utils.versions from salt.exceptions import CommandExecutionError, SaltInvocationError from salt.loader import _format_cached_grains from salt.runners.state import orchestrate as _orchestrate __proxyenabled__ = ["*"] __outputter__ = { "sls": "highstate", "sls_id": "highstate", "pkg": "highstate", "top": "highstate", "single": "highstate", "highstate": "highstate", "template": "highstate", "template_str": "highstate", "apply_": "highstate", "test": "highstate", "request": "highstate", "check_request": "highstate", "run_request": "highstate", } __func_alias__ = {"apply_": "apply"} log = logging.getLogger(__name__) # Define the module's virtual name __virtualname__ = "state" def __virtual__(): """ Set the virtualname """ # Update global namespace with functions that are cloned in this module global _orchestrate _orchestrate = salt.utils.functools.namespaced_function(_orchestrate, globals()) return __virtualname__ def _filter_running(runnings): """ Filter out the result: True + no changes data """ ret = { tag: value for tag, value in runnings.items() if not value["result"] or value["changes"] } return ret def _set_retcode(ret, highstate=None): """ Set the return code based on the data back from the state system """ # Set default retcode to 0 __context__["retcode"] = salt.defaults.exitcodes.EX_OK if isinstance(ret, list): __context__["retcode"] = salt.defaults.exitcodes.EX_STATE_COMPILER_ERROR return if not __utils__["state.check_result"](ret, highstate=highstate): __context__["retcode"] = salt.defaults.exitcodes.EX_STATE_FAILURE def _get_pillar_errors(kwargs, pillar=None): """ Check pillar for errors. If a pillar is passed, it will be checked. Otherwise, the in-memory pillar will checked instead. Passing kwargs['force'] = True short cuts the check and always returns None, indicating no errors. :param kwargs: dictionary of options :param pillar: pillar :return: None or a list of error messages """ return None if kwargs.get("force") else (pillar or __pillar__).get("_errors") def _wait(jid, max_queue=0): """ Wait for all previously started state jobs to finish running """ if jid is None: jid = salt.utils.jid.gen_jid(__opts__) with salt.utils.state.acquire_queue_lock(__opts__): states = _prior_running_states(jid) if not max_queue or len(states) < max_queue: while states: time.sleep(1) with salt.utils.state.acquire_queue_lock(__opts__): states = _prior_running_states(jid) return True return False def _snapper_pre(opts, jid): """ Create a snapper pre snapshot """ snapper_pre = None try: if not opts["test"] and __opts__.get("snapper_states"): # Run the snapper pre snapshot snapper_pre = __salt__["snapper.create_snapshot"]( config=__opts__.get("snapper_states_config", "root"), snapshot_type="pre", description=f"Salt State run for jid {jid}", __pub_jid=jid, ) except Exception: # pylint: disable=broad-except log.error("Failed to create snapper pre snapshot for jid: %s", jid) return snapper_pre def _snapper_post(opts, jid, pre_num): """ Create the post states snapshot """ try: if not opts["test"] and __opts__.get("snapper_states") and pre_num: # Run the snapper pre snapshot __salt__["snapper.create_snapshot"]( config=__opts__.get("snapper_states_config", "root"), snapshot_type="post", pre_number=pre_num, description=f"Salt State run for jid {jid}", __pub_jid=jid, ) except Exception: # pylint: disable=broad-except log.error("Failed to create snapper pre snapshot for jid: %s", jid) def _get_pause(jid, state_id=None): """ Return the pause information for a given jid """ pause_dir = os.path.join(__opts__["cachedir"], "state_pause") pause_path = os.path.join(pause_dir, jid) if not os.path.exists(pause_dir): try: os.makedirs(pause_dir) except OSError: # File created in the gap pass data = {} if state_id is not None: if state_id not in data: data[state_id] = {} if os.path.exists(pause_path): with salt.utils.files.fopen(pause_path, "rb") as fp_: data = salt.utils.msgpack.loads(fp_.read()) return data, pause_path def get_pauses(jid=None): """ Get a report on all of the currently paused state runs and pause run settings. Optionally send in a jid if you only desire to see a single pause data set. """ ret = {} active = __salt__["saltutil.is_running"]("state.*") pause_dir = os.path.join(__opts__["cachedir"], "state_pause") if not os.path.exists(pause_dir): return ret if jid is None: jids = os.listdir(pause_dir) elif isinstance(jid, list): jids = salt.utils.data.stringify(jid) else: jids = [str(jid)] for scan_jid in jids: is_active = False for active_data in active: if active_data["jid"] == scan_jid: is_active = True if not is_active: try: pause_path = os.path.join(pause_dir, scan_jid) os.remove(pause_path) except OSError: # Already gone pass continue data, pause_path = _get_pause(scan_jid) ret[scan_jid] = data return ret def soft_kill(jid, state_id=None): """ Set up a state run to die before executing the given state id, this instructs a running state to safely exit at a given state id. This needs to pass in the jid of the running state. If a state_id is not passed then the jid referenced will be safely exited at the beginning of the next state run. The given state id is the id got a given state execution, so given a state that looks like this: .. code-block:: yaml vim: pkg.installed: [] The state_id to pass to `soft_kill` is `vim` CLI Examples: .. code-block:: bash salt '*' state.soft_kill 20171130110407769519 salt '*' state.soft_kill 20171130110407769519 vim """ jid = str(jid) if state_id is None: state_id = "__all__" data, pause_path = _get_pause(jid, state_id) data[state_id]["kill"] = True with salt.utils.files.fopen(pause_path, "wb") as fp_: fp_.write(salt.utils.msgpack.dumps(data)) def pause(jid, state_id=None, duration=None): """ Set up a state id pause, this instructs a running state to pause at a given state id. This needs to pass in the jid of the running state and can optionally pass in a duration in seconds. If a state_id is not passed then the jid referenced will be paused at the beginning of the next state run. The given state id is the id got a given state execution, so given a state that looks like this: .. code-block:: yaml vim: pkg.installed: [] The state_id to pass to `pause` is `vim` CLI Examples: .. code-block:: bash salt '*' state.pause 20171130110407769519 salt '*' state.pause 20171130110407769519 vim salt '*' state.pause 20171130110407769519 vim 20 """ jid = str(jid) if state_id is None: state_id = "__all__" data, pause_path = _get_pause(jid, state_id) if duration: data[state_id]["duration"] = int(duration) with salt.utils.files.fopen(pause_path, "wb") as fp_: fp_.write(salt.utils.msgpack.dumps(data)) def resume(jid, state_id=None): """ Remove a pause from a jid, allowing it to continue. If the state_id is not specified then the a general pause will be resumed. The given state_id is the id got a given state execution, so given a state that looks like this: .. code-block:: yaml vim: pkg.installed: [] The state_id to pass to `rm_pause` is `vim` CLI Examples: .. code-block:: bash salt '*' state.resume 20171130110407769519 salt '*' state.resume 20171130110407769519 vim """ jid = str(jid) if state_id is None: state_id = "__all__" data, pause_path = _get_pause(jid, state_id) if state_id in data: data.pop(state_id) if state_id == "__all__": data = {} with salt.utils.files.fopen(pause_path, "wb") as fp_: fp_.write(salt.utils.msgpack.dumps(data)) def orchestrate( mods, saltenv="base", test=None, exclude=None, pillar=None, pillarenv=None ): """ .. versionadded:: 2016.11.0 Execute the orchestrate runner from a masterless minion. .. seealso:: More Orchestrate documentation * :ref:`Full Orchestrate Tutorial ` * Docs for the salt state module :py:mod:`salt.states.saltmod` CLI Examples: .. code-block:: bash salt-call --local state.orchestrate webserver salt-call --local state.orchestrate webserver saltenv=dev test=True salt-call --local state.orchestrate webserver saltenv=dev pillarenv=aws """ return _orchestrate( mods=mods, saltenv=saltenv, test=test, exclude=exclude, pillar=pillar, pillarenv=pillarenv, ) def running(concurrent=False): """ Return a list of strings that contain state return data if a state function is already running. This function is used to prevent multiple state calls from being run at the same time. CLI Example: .. code-block:: bash salt '*' state.running """ ret = [] if concurrent: return ret active = __salt__["saltutil.is_running"]("state.*") # Get the current JID to avoid false positives (self-detection) # This prevents failures when state.apply(queue=False) is called # but the job has a placeholder in the process table. current_jid = __opts__.get("jid") for data in active: # Ignore self if current_jid is not None: try: if int(data.get("jid")) == int(current_jid): continue except (ValueError, TypeError): pass err = ( 'The function "{}" is running as PID {} and was started at {} ' "with jid {}".format( data["fun"], data["pid"], salt.utils.jid.jid_to_time(data["jid"]), data["jid"], ) ) ret.append(err) return ret def _acquire_queue_lock(): """ Acquire the state queue lock """ return salt.utils.state.acquire_queue_lock(__opts__) def _set_queue_flag(jid): """ Set a flag to indicate that the state run is checking the queue """ if jid is None: return queue_dir = salt.utils.state.state_queue_dir(__opts__) queue_path = os.path.join(queue_dir, str(jid)) if not os.path.exists(queue_dir): try: os.makedirs(queue_dir) except OSError: pass with _acquire_queue_lock(): with salt.utils.files.fopen(queue_path, "w+") as fp_: fp_.write(str(os.getpid())) def _clear_queue_flag(jid): """ Clear the queue flag """ if jid is None: return queue_dir = salt.utils.state.state_queue_dir(__opts__) queue_path = os.path.join(queue_dir, str(jid)) with _acquire_queue_lock(): if os.path.exists(queue_path): try: os.remove(queue_path) except OSError: pass def _prior_running_states(jid): """ Return a list of dicts of prior calls to state functions. This function is used to queue state calls so only one is run at a time. """ active = __salt__["saltutil.is_running"]("state.*") return salt.utils.state.check_prior_running_states(__opts__, jid, active) def _check_queue(queue, kwargs): """ Utility function to queue the state run if requested and to check for conflicts in currently running states """ if queue is None: queue = __salt__["config.option"]("state_queue", False) if queue is True: jid = kwargs.get("__pub_jid") if jid is None: # If running locally (salt-call), JID might be in opts or not present. # Fallback to __opts__['jid'] to ensure we have a JID for comparison. jid = __opts__.get("jid") with salt.utils.state.acquire_queue_lock(__opts__): states = _prior_running_states(jid) if states: # Conflict found, queue the job queue_dir = salt.utils.state.state_queue_dir(__opts__) if not os.path.exists(queue_dir): try: os.makedirs(queue_dir) except OSError: pass # Construct payload to persist # We need to save enough info to re-execute the job. # # Preserve the master-assigned JID end-to-end (issue #69386). # Job-tracking infrastructure (returners, the jobs runner, # syndic forwarding) keys on the JID the master published; if # the minion executes under a different JID the master never # sees the return. # # Only mint a new JID when one wasn't supplied — that is the # salt-call / local case, where the minion is both publisher # and executor and no master-side tracking is involved. # Filename uniqueness is provided by the microsecond-precision # timestamp prefix, so we do not need a fresh JID for that. queued_jid = jid if queued_jid is None: queued_jid = salt.utils.jid.gen_jid(__opts__) # Remove 'queue' from kwargs to prevent re-queuing logic when executed kwarg = {k: v for k, v in kwargs.items() if not k.startswith("__pub_")} if "queue" in kwarg: del kwarg["queue"] payload = { "fun": kwargs.get("__pub_fun"), "arg": kwargs.get("__pub_arg", []), "tgt": kwargs.get("__pub_tgt"), "jid": queued_jid, "ret": kwargs.get("__pub_ret", ""), "user": kwargs.get("__pub_user", "root"), "kwarg": kwarg, } # Use timestamp to ensure FIFO ordering # We use microseconds to avoid collisions fn = f"queued_{int(time.time() * 1000000)}_{queued_jid}.p" path = os.path.join(queue_dir, fn) try: tmp_path = path + ".tmp" with salt.utils.files.fopen(tmp_path, "w+b") as fp_: salt.payload.dump(payload, fp_) salt.utils.atomicfile.atomic_rename(tmp_path, path) return { "result": True, "comment": "Job queued for execution", "queued": True, "changes": {}, "__no_return__": True, } except OSError: log.error("Failed to write queue file %s", path) return { "result": False, "comment": "Failed to queue job: unable to write queue file", "changes": {}, } else: # No conflict, we can run. pass else: queue_ret = False if not isinstance(queue, bool) and isinstance(queue, int): jid = kwargs.get("__pub_jid") # For max_queue (int), we retain blocking behavior but use lock _set_queue_flag(jid) try: queue_ret = _wait(jid, max_queue=queue) finally: _clear_queue_flag(jid) if not queue_ret: conflict = running(concurrent=kwargs.get("concurrent", False)) if conflict: __context__["retcode"] = salt.defaults.exitcodes.EX_STATE_COMPILER_ERROR return conflict return def _get_initial_pillar(opts): return ( __pillar__.value() if __opts__.get("__cli", None) == "salt-call" and opts["pillarenv"] == __opts__["pillarenv"] else None ) def low(data, queue=None, **kwargs): """ Execute a single low data call This function is mostly intended for testing the state system and is not likely to be needed in everyday usage. CLI Example: .. code-block:: bash salt '*' state.low '{"state": "pkg", "fun": "installed", "name": "vi"}' """ conflict = _check_queue(queue, kwargs) if conflict is not None: return conflict try: st_ = salt.state.State(__opts__, proxy=__proxy__) except NameError: st_ = salt.state.State(__opts__) err = st_.verify_data(data) if err: __context__["retcode"] = salt.defaults.exitcodes.EX_STATE_COMPILER_ERROR return err ret = st_.call(data) if isinstance(ret, list): __context__["retcode"] = salt.defaults.exitcodes.EX_STATE_COMPILER_ERROR if __utils__["state.check_result"](ret): __context__["retcode"] = salt.defaults.exitcodes.EX_STATE_FAILURE return ret def _get_test_value(test=None, **kwargs): """ Determine the correct value for the test flag. """ ret = True if test is None: if salt.utils.args.test_mode(test=test, **kwargs): ret = True elif __salt__["config.get"]("test", omit_opts=True) is True: ret = True else: ret = __opts__.get("test", None) elif test is False: ret = False return ret def high(data, test=None, queue=None, **kwargs): """ Execute the compound calls stored in a single set of high data This function is mostly intended for testing the state system and is not likely to be needed in everyday usage. CLI Example: .. code-block:: bash salt '*' state.high '{"vim": {"pkg": ["installed"]}}' """ conflict = _check_queue(queue, kwargs) if conflict is not None: return conflict opts = salt.utils.state.get_sls_opts(__opts__, **kwargs) opts["test"] = _get_test_value(test, **kwargs) pillar_override = kwargs.get("pillar") pillar_enc = kwargs.get("pillar_enc") if ( pillar_enc is None and pillar_override is not None and not isinstance(pillar_override, dict) ): raise SaltInvocationError( "Pillar data must be formatted as a dictionary, unless pillar_enc " "is specified." ) try: st_ = salt.state.State( opts, pillar_override, pillar_enc=pillar_enc, proxy=dict(__proxy__), context=dict(__context__), initial_pillar=_get_initial_pillar(opts), ) except NameError: st_ = salt.state.State( opts, pillar_override, pillar_enc=pillar_enc, initial_pillar=_get_initial_pillar(opts), ) ret = st_.call_high(data) _set_retcode(ret, highstate=data) return ret def template(tem, queue=None, **kwargs): """ Execute the information stored in a template file on the minion. This function does not ask a master for a SLS file to render but instead directly processes the file at the provided path on the minion. CLI Example: .. code-block:: bash salt '*' state.template '' """ if "env" in kwargs: # "env" is not supported; Use "saltenv". kwargs.pop("env") conflict = _check_queue(queue, kwargs) if conflict is not None: return conflict opts = salt.utils.state.get_sls_opts(__opts__, **kwargs) try: st_ = salt.state.HighState( opts, context=dict(__context__), proxy=dict(__proxy__), initial_pillar=_get_initial_pillar(opts), ) except NameError: st_ = salt.state.HighState( opts, context=dict(__context__), initial_pillar=_get_initial_pillar(opts) ) with st_: errors = _get_pillar_errors(kwargs, pillar=st_.opts["pillar"]) if errors: __context__["retcode"] = salt.defaults.exitcodes.EX_PILLAR_FAILURE raise CommandExecutionError("Pillar failed to render", info=errors) if not tem.endswith(".sls"): tem = f"{tem}.sls" high_state, errors = st_.render_state( tem, kwargs.get("saltenv", ""), "", None, local=True ) if errors: __context__["retcode"] = salt.defaults.exitcodes.EX_STATE_COMPILER_ERROR return errors ret = st_.state.call_high(high_state) _set_retcode(ret, highstate=high_state) return ret def template_str(tem, queue=None, **kwargs): """ Execute the information stored in a string from an sls template CLI Example: .. code-block:: bash salt '*' state.template_str '