Repository: saltstack/salt
Pull request: Adding retry functionality when parallel is set to True
### What does this PR do?
Making the retry state system feature available when parallel is set to True.

### What issues does this PR fix or reference?
Fixes: #61630 #61439

### Previous Behavior
When the retry options are set in a state and parallel is set to True, the job is rerun regardless of the original state results and the subsequent runs are not done in parallel.

### New Behavior
Retry functionality is properly honored with `parallel: True`

### Merge requirements satisfied?
**[NOTICE] Bug fixes or features added to Salt require tests.**
<!-- Please review the [test documentation](https://docs.saltproject.io/en/master/topics/tutorials/writing_tests.html) for details on how to implement tests into Salt's test suite. -->
- [ ] Docs
- [x] Changelog - https://docs.saltproject.io/en/master/topics/development/changelog.html
- [x] Tests written/updated

### Commits signed with GPG?
Yes

Please review [Salt's Contributing Guide](https://docs.saltproject.io/en/master/topics/development/contributing.html) for best practices.

See GitHub's [page on GPG signing](https://help.github.com/articles/signing-commits-using-gpg/) for more information about signing commits with GPG.


Problem statement:
[BUG] retry state option unavailble when parallel is True
**Description**
When the `retry` options are set in a state and parallel is set to True, the job is rerun regardless of the original state results and the subsequent runs are not done in parallel.

**Setup**
Basic state that includes `parallel: True` and `retry` options:

```
{% set parallel = True %}
{% for i in range(1,5) %}
run {{ i }}:
  cmd.run:
    - name: /tmp/cmd.sh
    - parallel: {{ parallel }}
    - retry:
        attempts: 2
        until: True
        interval: 5
{% endfor %}
```

```
randomNumber=$(( $RANDOM % 2 ))

if [[ $randomNumber == 1 ]]; then
    echo "False"
    exit 1
else
    echo "True"
    exit 0
fi
```

Please be as specific as possible and give set-up details.

- [ ] on-prem machine
- [ ] VM (Virtualbox, KVM, etc. please specify)
- [ ] VM running on a cloud service, please be explicit and add details
- [ ] container (Kubernetes, Docker, containerd, etc. please specify)
- [ ] or a combination, please be explicit
- [ ] jails if it is FreeBSD

```
PASTE HERE
```
</details>

**Additional context**
Add any other context about the problem here.

Hints from issue discussion:

Commit message: Making the retry state system feature available when parallel is set to True.

Patch to review:
diff --git a/changelog/61630.fixed b/changelog/61630.fixed
new file mode 100644
index 000000000000..c7472bd8f7fc
--- /dev/null
+++ b/changelog/61630.fixed
@@ -0,0 +1 @@
+Making the retry state system feature available when parallel is set to True.
diff --git a/salt/state.py b/salt/state.py
index 9deee48cef7d..e864fa09e7ba 100644
--- a/salt/state.py
+++ b/salt/state.py
@@ -2016,11 +2016,74 @@ def _call_parallel_target(self, 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
 
+        if "retry" in low:
+            retries = 1
+            low["retry"] = self.verify_retry_data(low["retry"])
+            if not sys.modules[self.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,
+                    )
+                    self.functions["test.sleep"](interval)
+                    retry_ret = self.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(self.opts["cachedir"], self.jid)
         tfile = os.path.join(troot, salt.utils.hashutils.sha1_digest(tag))
         if not os.path.isdir(troot):
@@ -2263,7 +2326,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"]:
diff --git a/tests/pytests/functional/modules/state/test_state.py b/tests/pytests/functional/modules/state/test_state.py
index 736e7ca2b3f7..03d830b6e119 100644
--- a/tests/pytests/functional/modules/state/test_state.py
+++ b/tests/pytests/functional/modules/state/test_state.py
@@ -702,6 +702,41 @@ def test_retry_option_success(state, state_tree, tmp_path):
             assert "Attempt 2" not in state_return.comment
 
 
+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.is_windows():
+        duration = 16
+
+    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 +782,55 @@ def create_testfile(testfile1, testfile2):
             assert "Attempt 5" not in state_return.comment
 
 
+@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):
     """


Code review comments:
diff_hunk: @@ -2016,11 +2016,74 @@ def _call_parallel_target(self, 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
 
+        if "retry" in low:
+            retries = 1
+            low["retry"] = self.verify_retry_data(low["retry"])
+            if not sys.modules[self.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,
+                    )
+                    self.functions["test.sleep"](interval)
path: salt/state.py
text: @user1:
Why are we calling the `test.sleep` function here instead of just using python's time's lib with `time.sleep`?

@author:
No particular reason, most of this code was copied from the existing `retry` functionality.  In this instance it probably makes more sense to use `time.sleep` as this is running inside it's own thread/process.  I'll update it.