Schedule Items in Pillar Refresh Fixes (#59104)

* Fixes to storing schedule items in pillar, when refreshing pillar only update the schedule items if something has changed.

* Adding test to pytests for scheduler & pillar changes.

* Adding changelog file

* Suggested changes from s0undt3ch

* Swapping @slowTest for @pytest.mark.slow_test

Co-authored-by: Megan Wilhite <megan.wilhite@gmail.com>

changelog/59104.fixed

@@ -0,0 +1 @@
+Fixes to storing schedule items in pillar, when refreshing pillar only update the schedule items if something has changed.

salt/minion.py

@@ -41,6 +41,7 @@ import salt.utils.args
 import salt.utils.context
 import salt.utils.crypt
 import salt.utils.data
+import salt.utils.dictdiffer
 import salt.utils.dictupdate
 import salt.utils.error
 import salt.utils.event
@@ -2419,6 +2420,38 @@ class Minion(MinionBase):
         log.debug("Refreshing matchers.")
         self.matchers = salt.loader.matchers(self.opts)
 
+    def pillar_schedule_refresh(self, current, new):
+        """
+        Refresh the schedule in pillar
+        """
+        # delete any pillar schedule items not
+        # in the updated pillar values.
+        for item in list(current):
+            if item not in new:
+                del current[item]
+
+        # Update any entries that have changed
+        pillar_schedule = {}
+        for item in current:
+            schedule_item = current[item]
+            # ignore any of the internal keys
+            ignore = [item for item in schedule_item if item.startswith("_")]
+            if item in new:
+                diff = salt.utils.dictdiffer.deep_diff(
+                    schedule_item, new[item], ignore=ignore
+                )
+                if diff.get("new"):
+                    pillar_schedule[item] = new[item]
+                else:
+                    pillar_schedule[item] = schedule_item
+
+        # Add any new entries
+        for item in new:
+            if item not in pillar_schedule:
+                pillar_schedule[item] = new[item]
+
+        return pillar_schedule
+
     # TODO: only allow one future in flight at a time?
     @salt.ext.tornado.gen.coroutine
     def pillar_refresh(self, force_refresh=False):
@@ -2437,13 +2470,20 @@ class Minion(MinionBase):
                 pillarenv=self.opts.get("pillarenv"),
             )
             try:
-                self.opts["pillar"] = yield async_pillar.compile_pillar()
+                new_pillar = yield async_pillar.compile_pillar()
             except SaltClientError:
                 # Do not exit if a pillar refresh fails.
                 log.error(
                     "Pillar data could not be refreshed. "
                     "One or more masters may be down!"
                 )
+            else:
+                current_schedule = self.opts["pillar"].get("schedule", {})
+                new_schedule = new_pillar.get("schedule", {})
+                new_pillar["schedule"] = self.pillar_schedule_refresh(
+                    current_schedule, new_schedule
+                )
+                self.opts["pillar"] = new_pillar
             finally:
                 async_pillar.destroy()
         self.matchers_refresh()
@@ -2463,6 +2503,7 @@ class Minion(MinionBase):
         schedule = data.get("schedule", None)
         where = data.get("where", None)
         persist = data.get("persist", None)
+        fire_event = data.get("fire_event", None)
 
         funcs = {
             "delete": ("delete_job", (name, persist)),
@@ -2479,6 +2520,7 @@ class Minion(MinionBase):
             "list": ("list", (where,)),
             "save_schedule": ("save_schedule", ()),
             "get_next_fire_time": ("get_next_fire_time", (name,)),
+            "job_status": ("job_status", (name, fire_event)),
         }
 
         # Call the appropriate schedule function

salt/modules/schedule.py

@@ -1316,3 +1316,36 @@ def show_next_fire_time(name, **kwargs):
     else:
         ret["comment"] = "next fire time not available."
     return ret
+
+
+def job_status(name):
+    """
+    Show the information for a particular job.
+
+    CLI Example:
+
+    .. code-block:: bash
+
+        salt '*' schedule.job_status
+
+    """
+
+    schedule = {}
+    try:
+        with salt.utils.event.get_event("minion", opts=__opts__) as event_bus:
+            res = __salt__["event.fire"](
+                {"func": "job_status", "name": name, "fire_event": True},
+                "manage_schedule",
+            )
+            if res:
+                event_ret = event_bus.get_event(
+                    tag="/salt/minion/minion_schedule_job_status_complete", wait=30
+                )
+                return event_ret.get("data", {})
+    except KeyError:
+        # Effectively a no-op, since we can't really return without an event system
+        ret = {}
+        ret["comment"] = "Event module not available. Schedule list failed."
+        ret["result"] = True
+        log.debug("Event module not available. Schedule list failed.")
+        return ret

salt/utils/schedule.py

@@ -647,13 +647,26 @@ class Schedule:
                 tag="/salt/minion/minion_schedule_next_fire_time_complete",
             )
 
-    def job_status(self, name):
+    def job_status(self, name, fire_event=False):
         """
         Return the specified schedule item
         """
 
-        schedule = self._get_schedule()
-        return schedule.get(name, {})
+        if fire_event:
+            schedule = self._get_schedule()
+            data = schedule.get(name, {})
+
+            # Fire the complete event back along with updated list of schedule
+            with salt.utils.event.get_event(
+                "minion", opts=self.opts, listen=False
+            ) as evt:
+                evt.fire_event(
+                    {"complete": True, "data": data},
+                    tag="/salt/minion/minion_schedule_job_status_complete",
+                )
+        else:
+            schedule = self._get_schedule()
+            return schedule.get(name, {})
 
     def handle_func(self, multiprocessing_enabled, func, data, jid=None):
         """

tests/pytests/integration/modules/test_pillar.py

@@ -1,5 +1,6 @@
 import pathlib
 import textwrap
+import time
 
 import attr
 import pytest
@@ -53,6 +54,7 @@ def pillar_tree(base_env_pillar_tree_root_dir, salt_minion, salt_call_cli):
         assert ret.json is True
 
 
+@pytest.mark.slow_test
 def test_data(salt_call_cli, pillar_tree):
     """
     pillar.data
@@ -73,6 +75,7 @@ def test_data(salt_call_cli, pillar_tree):
         assert pillar["class"] == "other"
 
 
+@pytest.mark.slow_test
 def test_issue_5449_report_actual_file_roots_in_pillar(
     salt_call_cli, pillar_tree, base_env_state_tree_root_dir
 ):
@@ -91,6 +94,7 @@ def test_issue_5449_report_actual_file_roots_in_pillar(
     ]
 
 
+@pytest.mark.slow_test
 def test_ext_cmd_yaml(salt_call_cli, pillar_tree):
     """
     pillar.data for ext_pillar cmd.yaml
@@ -102,6 +106,7 @@ def test_ext_cmd_yaml(salt_call_cli, pillar_tree):
     assert pillar["ext_spam"] == "eggs"
 
 
+@pytest.mark.slow_test
 def test_issue_5951_actual_file_roots_in_opts(
     salt_call_cli, pillar_tree, base_env_state_tree_root_dir
 ):
@@ -115,6 +120,7 @@ def test_issue_5951_actual_file_roots_in_opts(
     ]
 
 
+@pytest.mark.slow_test
 def test_pillar_items(salt_call_cli, pillar_tree):
     """
     Test to ensure we get expected output
@@ -130,6 +136,7 @@ def test_pillar_items(salt_call_cli, pillar_tree):
     assert pillar_items["knights"] == ["Lancelot", "Galahad", "Bedevere", "Robin"]
 
 
+@pytest.mark.slow_test
 def test_pillar_command_line(salt_call_cli, pillar_tree):
     """
     Test to ensure when using pillar override
@@ -203,6 +210,7 @@ def key_pillar(salt_minion, salt_cli, base_env_pillar_tree_root_dir):
     )
 
 
+@pytest.mark.slow_test
 def test_pillar_refresh_pillar_raw(salt_cli, salt_minion, key_pillar):
     """
     Validate the minion's pillar.raw call behavior for new pillars
@@ -232,6 +240,7 @@ def test_pillar_refresh_pillar_raw(salt_cli, salt_minion, key_pillar):
         assert val is True, repr(val)
 
 
+@pytest.mark.slow_test
 def test_pillar_refresh_pillar_get(salt_cli, salt_minion, key_pillar):
     """
     Validate the minion's pillar.get call behavior for new pillars
@@ -262,6 +271,7 @@ def test_pillar_refresh_pillar_get(salt_cli, salt_minion, key_pillar):
         assert val is True, repr(val)
 
 
+@pytest.mark.slow_test
 def test_pillar_refresh_pillar_item(salt_cli, salt_minion, key_pillar):
     """
     Validate the minion's pillar.item call behavior for new pillars
@@ -295,6 +305,7 @@ def test_pillar_refresh_pillar_item(salt_cli, salt_minion, key_pillar):
         assert val[key] is True
 
 
+@pytest.mark.slow_test
 def test_pillar_refresh_pillar_items(salt_cli, salt_minion, key_pillar):
     """
     Validate the minion's pillar.item call behavior for new pillars
@@ -317,6 +328,7 @@ def test_pillar_refresh_pillar_items(salt_cli, salt_minion, key_pillar):
         assert val[key] is True
 
 
+@pytest.mark.slow_test
 def test_pillar_refresh_pillar_ping(salt_cli, salt_minion, key_pillar):
     """
     Validate the minion's test.ping does not update pillars
@@ -355,3 +367,90 @@ def test_pillar_refresh_pillar_ping(salt_cli, salt_minion, key_pillar):
         val = ret.json
         assert key in val
         assert val[key] is True
+
+
+@pytest.mark.slow_test
+def test_pillar_refresh_pillar_scheduler(salt_cli, salt_minion):
+    """
+    Ensure schedule jobs in pillar are only updated when values change.
+    """
+
+    top_sls = """
+        base:
+          '{}':
+            - test_schedule
+        """.format(
+        salt_minion.id
+    )
+
+    test_schedule_sls = """
+        schedule:
+          first_test_ping:
+            function: test.ping
+            run_on_start: True
+            seconds: 3600
+        """
+
+    test_schedule_sls2 = """
+        schedule:
+          first_test_ping:
+            function: test.ping
+            run_on_start: True
+            seconds: 7200
+        """
+
+    with pytest.helpers.temp_pillar_file("top.sls", top_sls):
+        with pytest.helpers.temp_pillar_file("test_schedule.sls", test_schedule_sls):
+            # Calling refresh_pillar to update in-memory pillars
+            salt_cli.run(
+                "saltutil.refresh_pillar", wait=True, minion_tgt=salt_minion.id
+            )
+
+            # Give the schedule a chance to run the job
+            time.sleep(5)
+
+            # Get the status of the job
+            ret = salt_cli.run(
+                "schedule.job_status", name="first_test_ping", minion_tgt=salt_minion.id
+            )
+            assert "_next_fire_time" in ret.json
+            _next_fire_time = ret.json["_next_fire_time"]
+
+            # Refresh pillar
+            salt_cli.run(
+                "saltutil.refresh_pillar", wait=True, minion_tgt=salt_minion.id
+            )
+
+            # Ensure next_fire_time is the same, job was not replaced
+            ret = salt_cli.run(
+                "schedule.job_status", name="first_test_ping", minion_tgt=salt_minion.id
+            )
+            assert ret.json["_next_fire_time"] == _next_fire_time
+
+        # Ensure job was replaced when seconds changes
+        with pytest.helpers.temp_pillar_file("test_schedule.sls", test_schedule_sls2):
+            # Calling refresh_pillar to update in-memory pillars
+            salt_cli.run(
+                "saltutil.refresh_pillar", wait=True, minion_tgt=salt_minion.id
+            )
+
+            # Give the schedule a chance to run the job
+            time.sleep(5)
+
+            ret = salt_cli.run(
+                "schedule.job_status", name="first_test_ping", minion_tgt=salt_minion.id
+            )
+            assert "_next_fire_time" in ret.json
+            _next_fire_time = ret.json["_next_fire_time"]
+
+            salt_cli.run(
+                "saltutil.refresh_pillar", wait=True, minion_tgt=salt_minion.id
+            )
+
+            ret = salt_cli.run(
+                "schedule.job_status", name="first_test_ping", minion_tgt=salt_minion.id
+            )
+            assert ret.json["_next_fire_time"] == _next_fire_time
+
+    # Refresh pillar once we're done
+    salt_cli.run("saltutil.refresh_pillar", wait=True, minion_tgt=salt_minion.id)

tests/unit/modules/test_schedule.py

@@ -562,3 +562,27 @@ class ScheduleTestCase(TestCase, LoaderModuleMockMixin):
 
                     ret = schedule.is_enabled()
                     self.assertEqual(ret, True)
+
+    # 'job_status' function tests: 1
+
+    def test_job_status(self):
+        """
+        Test is_enabled
+        """
+        job1 = {"function": "salt", "seconds": 3600}
+
+        comm1 = "Modified job: job1 in schedule."
+
+        mock_schedule = {"enabled": True, "job1": job1}
+
+        mock_lst = MagicMock(return_value=mock_schedule)
+
+        with patch.dict(
+            schedule.__opts__, {"schedule": {"job1": job1}, "sock_dir": self.sock_dir}
+        ):
+            mock = MagicMock(return_value=True)
+            with patch.dict(schedule.__salt__, {"event.fire": mock}):
+                _ret_value = {"complete": True, "data": job1}
+                with patch.object(SaltEvent, "get_event", return_value=_ret_value):
+                    ret = schedule.job_status("job1")
+                    self.assertDictEqual(ret, job1)