[3006.x] Fix parallel state execution with Salt-SSH

changelog/66514.fixed.md

@@ -0,0 +1 @@
+Fixed parallel state execution with Salt-SSH

salt/state.py

@@ -39,6 +39,7 @@
 import salt.utils.files
 import salt.utils.hashutils
 import salt.utils.immutabletypes as immutabletypes
+import salt.utils.jid
 import salt.utils.msgpack
 import salt.utils.platform
 import salt.utils.process
@@ -757,7 +758,21 @@ def __init__(
         loader="states",
         initial_pillar=None,
         file_client=None,
+        __invocation_id=None,
     ):
+        """
+        When instantiating an object of this class, do not pass
+        ``__invocation_id``. It is an internal field for tracking
+        parallel executions where no jid is available (Salt-SSH) and
+        only exposed as an init argument to work on spawning platforms.
+        """
+        if jid is not None:
+            __invocation_id = jid
+        if __invocation_id is None:
+            # For salt-ssh parallel states, we need a unique identifier
+            # for a single execution. self.jid should not be set there
+            # since it's used for other purposes as well.
+            __invocation_id = salt.utils.jid.gen_jid(opts)
         self._init_kwargs = {
             "opts": opts,
             "pillar_override": pillar_override,
@@ -768,6 +783,7 @@ def __init__(
             "mocked": mocked,
             "loader": loader,
             "initial_pillar": initial_pillar,
+            "__invocation_id": __invocation_id,
         }
         self.states_loader = loader
         if "grains" not in opts:
@@ -814,6 +830,7 @@ def __init__(
         self.pre = {}
         self.__run_num = 0
         self.jid = jid
+        self.invocation_id = __invocation_id
         self.instance_id = str(id(self))
         self.inject_globals = {}
         self.mocked = mocked
@@ -2236,7 +2253,7 @@ def _call_parallel_target(cls, instance, init_kwargs, name, cdata, low):
                     ]
                 )
 
-        troot = os.path.join(instance.opts["cachedir"], instance.jid)
+        troot = os.path.join(instance.opts["cachedir"], instance.invocation_id)
         tfile = os.path.join(troot, salt.utils.hashutils.sha1_digest(tag))
         if not os.path.isdir(troot):
             try:
@@ -2820,7 +2837,7 @@ def reconcile_procs(self, running):
                 if not proc.is_alive():
                     ret_cache = os.path.join(
                         self.opts["cachedir"],
-                        self.jid,
+                        self.invocation_id,
                         salt.utils.hashutils.sha1_digest(tag),
                     )
                     if not os.path.isfile(ret_cache):

tests/pytests/integration/ssh/state/test_parallel.py

@@ -0,0 +1,61 @@
+"""
+Verify salt-ssh states support ``parallel``.
+"""
+
+import pytest
+
+pytestmark = [
+    pytest.mark.skip_on_windows(reason="salt-ssh not available on Windows"),
+    pytest.mark.slow_test,
+]
+
+
+@pytest.fixture(scope="module", autouse=True)
+def state_tree_parallel(base_env_state_tree_root_dir):
+    top_file = """
+    base:
+      'localhost':
+        - parallel
+      '127.0.0.1':
+        - parallel
+    """
+    state_file = """
+    {%- for i in range(5) %}
+    This runs in parallel {{ i }}:
+      cmd.run:
+        - name: sleep 0.{{ i }}
+        - parallel: true
+    {%- endfor %}
+    """
+    top_tempfile = pytest.helpers.temp_file(
+        "top.sls", top_file, base_env_state_tree_root_dir
+    )
+    state_tempfile = pytest.helpers.temp_file(
+        "parallel.sls", state_file, base_env_state_tree_root_dir
+    )
+    with top_tempfile, state_tempfile:
+        yield
+
+
+@pytest.mark.parametrize(
+    "args",
+    (
+        pytest.param(("state.sls", "parallel"), id="sls"),
+        pytest.param(("state.highstate",), id="highstate"),
+        pytest.param(("state.top", "top.sls"), id="top"),
+    ),
+)
+def test_it(salt_ssh_cli, args):
+    """
+    Ensure states with ``parallel: true`` do not cause a crash.
+    This does not check that they were actually run in parallel
+    since that would result either in a long-running or flaky test.
+    """
+    ret = salt_ssh_cli.run(*args)
+    assert ret.returncode == 0
+    assert isinstance(ret.data, dict)
+    for i in range(5):
+        key = f"cmd_|-This runs in parallel {i}_|-sleep 0.{i}_|-run"
+        assert key in ret.data
+        assert "pid" in ret.data[key]["changes"]
+        assert ret.data[key]["changes"]["retcode"] == 0