Adding retry functionality when parallel is set to True

changelog/61630.fixed

@@ -0,0 +1 @@
+Making the retry state system feature available when parallel is set to True.

salt/state.py

@@ -2030,12 +2030,77 @@ def _call_parallel_target(cls, instance, init_kwargs, name, cdata, low):
             }
 
         utc_finish_time = datetime.datetime.utcnow()
+        timezone_delta = datetime.datetime.utcnow() - datetime.datetime.now()
+        local_finish_time = utc_finish_time - timezone_delta
+        local_start_time = utc_start_time - timezone_delta
+        ret["start_time"] = local_start_time.time().isoformat()
         delta = utc_finish_time - utc_start_time
         # duration in milliseconds.microseconds
         duration = (delta.seconds * 1000000 + delta.microseconds) / 1000.0
         ret["duration"] = duration
         ret["__parallel__"] = True
 
+        if "retry" in low:
+            retries = 1
+            low["retry"] = instance.verify_retry_data(low["retry"])
+            if not sys.modules[instance.states[cdata["full"]].__module__].__opts__[
+                "test"
+            ]:
+                while low["retry"]["attempts"] >= retries:
+
+                    if low["retry"]["until"] == ret["result"]:
+                        break
+
+                    interval = low["retry"]["interval"]
+                    if low["retry"]["splay"] != 0:
+                        interval = interval + random.randint(0, low["retry"]["splay"])
+                    log.info(
+                        "State result does not match retry until value, "
+                        "state will be re-run in %s seconds",
+                        interval,
+                    )
+                    time.sleep(interval)
+                    retry_ret = instance.states[cdata["full"]](
+                        *cdata["args"], **cdata["kwargs"]
+                    )
+
+                    utc_start_time = datetime.datetime.utcnow()
+                    utc_finish_time = datetime.datetime.utcnow()
+                    delta = utc_finish_time - utc_start_time
+                    duration = (delta.seconds * 1000000 + delta.microseconds) / 1000.0
+                    retry_ret["duration"] = duration
+
+                    orig_ret = ret
+                    ret = retry_ret
+                    if not ret["comment"]:
+                        _comment = ""
+                    else:
+                        _comment = (
+                            'Attempt {}: Returned a result of "{}", '
+                            'with the following comment: "{}"'.format(
+                                retries, ret["result"], ret["comment"]
+                            )
+                        )
+
+                    ret["comment"] = "\n".join([orig_ret["comment"], _comment])
+                    ret["duration"] = (
+                        ret["duration"] + orig_ret["duration"] + (interval * 1000)
+                    )
+                    if retries == 1:
+                        ret["start_time"] = orig_ret["start_time"]
+                    retries = retries + 1
+
+            else:
+                ret["comment"] = "  ".join(
+                    [
+                        "" if not ret["comment"] else str(ret["comment"]),
+                        "The state would be retried every {interval} seconds "
+                        "(with a splay of up to {splay} seconds) a maximum of "
+                        "{attempts} times or until a result of {until} "
+                        "is returned".format(**low["retry"]),
+                    ]
+                )
+
         troot = os.path.join(instance.opts["cachedir"], instance.jid)
         tfile = os.path.join(troot, salt.utils.hashutils.sha1_digest(tag))
         if not os.path.isdir(troot):
@@ -2283,7 +2348,7 @@ def call(self, low, chunks=None, running=None, retries=1):
             local_finish_time.time().isoformat(),
             duration,
         )
-        if "retry" in low:
+        if "retry" in low and "parallel" not in low:
             low["retry"] = self.verify_retry_data(low["retry"])
             if not sys.modules[self.states[cdata["full"]].__module__].__opts__["test"]:
                 if low["retry"]["until"] != ret["result"]:

tests/pytests/functional/modules/state/test_state.py

@@ -690,7 +690,7 @@ def test_retry_option_success(state, state_tree, tmp_path):
         testfile
     )
     duration = 4
-    if salt.utils.platform.is_windows():
+    if salt.utils.platform.spawning_platform():
         duration = 16
 
     with pytest.helpers.temp_file("retry.sls", sls_contents, state_tree):
@@ -702,6 +702,44 @@ def test_retry_option_success(state, state_tree, tmp_path):
             assert "Attempt 2" not in state_return.comment
 
 
+@pytest.mark.skip_on_windows(
+    reason="Skipped until parallel states can be fixed on Windows"
+)
+def test_retry_option_success_parallel(state, state_tree, tmp_path):
+    """
+    test a state with the retry option that should return True immediately (i.e. no retries)
+    """
+    testfile = tmp_path / "testfile"
+    testfile.touch()
+    sls_contents = """
+    file_test:
+      file.exists:
+        - name: {}
+        - parallel: True
+        - retry:
+            until: True
+            attempts: 5
+            interval: 2
+            splay: 0
+    """.format(
+        testfile
+    )
+    duration = 4
+    if salt.utils.platform.spawning_platform():
+        duration = 30
+
+    with pytest.helpers.temp_file("retry.sls", sls_contents, state_tree):
+        ret = state.sls(
+            "retry", __pub_jid="1"
+        )  # Because these run in parallel we need a fake JID
+
+        for state_return in ret:
+            assert state_return.result is True
+            assert state_return.full_return["duration"] < duration
+            # It should not take 2 attempts
+            assert "Attempt 2" not in state_return.comment
+
+
 @pytest.mark.slow_test
 def test_retry_option_eventual_success(state, state_tree, tmp_path):
     """
@@ -747,6 +785,58 @@ def create_testfile(testfile1, testfile2):
             assert "Attempt 5" not in state_return.comment
 
 
+@pytest.mark.skip_on_windows(
+    reason="Skipped until parallel states can be fixed on Windows"
+)
+@pytest.mark.slow_test
+def test_retry_option_eventual_success_parallel(state, state_tree, tmp_path):
+    """
+    test a state with the retry option that should return True, eventually
+    """
+    testfile1 = tmp_path / "testfile-1"
+    testfile2 = tmp_path / "testfile-2"
+
+    def create_testfile(testfile1, testfile2):
+        while True:
+            if testfile1.exists():
+                break
+        time.sleep(2)
+        testfile2.touch()
+
+    thread = threading.Thread(target=create_testfile, args=(testfile1, testfile2))
+    sls_contents = """
+    file_test_a:
+      file.managed:
+        - name: {}
+        - content: 'a'
+
+    file_test:
+      file.exists:
+        - name: {}
+        - retry:
+            until: True
+            attempts: 5
+            interval: 2
+            splay: 0
+        - parallel: True
+        - require:
+          - file_test_a
+    """.format(
+        testfile1, testfile2
+    )
+    with pytest.helpers.temp_file("retry.sls", sls_contents, state_tree):
+        thread.start()
+        ret = state.sls(
+            "retry", __pub_jid="1"
+        )  # Because these run in parallel we need a fake JID
+        for state_return in ret:
+            log.debug("=== state_return %s ===", state_return)
+            assert state_return.result is True
+            assert state_return.full_return["duration"] > 4
+            # It should not take 5 attempts
+            assert "Attempt 5" not in state_return.comment
+
+
 @pytest.mark.slow_test
 def test_state_non_base_environment(state, state_tree_prod, tmp_path):
     """