Repository: saltstack/salt
Pull request: [3006.x] deltaproxy, scheduler and event bus fixes
### What does this PR do?
Update all the scheduler functions to include a fire_event argument which will determine whether to fire the completion event onto the event bus.  This event is only used when these functions are called via the schedule execution modules.  Inside deltaproxy, then update all scheudle related functions to include fire_event=False, as the event bus is not available when these functions are called.

### What issues does this PR fix or reference?
Fixes:

### 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] One network device failure caused the restart proxy in loop until manually stopped.
**Description**
One network device failure caused the restart proxy in loop until manually stopped. 

**Setup**
Delta proxy devices loading from MongoDB script. There are 35 devices in one delta proxy and it keep crashing and unable to start.

Please be as specific as possible and give set-up details.
- [ ] VM (Virtualbox, KVM, etc. please specify)

**Steps to Reproduce the behavior**
```
/opt/saltstack/salt/lib/python3.10/site-packages/salt/transport/client.py:79: DeprecationWarning: This module is deprecated. Please use salt.channel.client instead.
  warn_until(
[CRITICAL] Unexpected error while connecting to salt-prod-master-1.infra.vmware.com
Traceback (most recent call last):
  File "/opt/saltstack/salt/lib/python3.10/site-packages/salt/minion.py", line 1135, in _connect_minion
    yield minion.connect_master(failed=failed)
  File "/opt/saltstack/salt/lib/python3.10/site-packages/salt/ext/tornado/gen.py", line 1056, in run
    value = future.result()
  File "/opt/saltstack/salt/lib/python3.10/site-packages/salt/ext/tornado/concurrent.py", line 249, in result
    raise_exc_info(self._exc_info)
  File "<string>", line 4, in raise_exc_info
  File "/opt/saltstack/salt/lib/python3.10/site-packages/salt/ext/tornado/gen.py", line 1064, in run
    yielded = self.gen.throw(*exc_info)
  File "/opt/saltstack/salt/lib/python3.10/site-packages/salt/minion.py", line 1376, in connect_master
    yield self._post_master_init(master)
  File "/opt/saltstack/salt/lib/python3.10/site-packages/salt/ext/tornado/gen.py", line 1056, in run
    value = future.result()
  File "/opt/saltstack/salt/lib/python3.10/site-packages/salt/ext/tornado/concurrent.py", line 249, in result
    raise_exc_info(self._exc_info)
  File "<string>", line 4, in raise_exc_info
  File "/opt/saltstack/salt/lib/python3.10/site-packages/salt/ext/tornado/gen.py", line 1070, in run
    yielded = self.gen.send(value)
  File "/opt/saltstack/salt/lib/python3.10/types.py", line 220, in send
    return self.__wrapped.send(val)
  File "/opt/saltstack/salt/lib/python3.10/site-packages/salt/metaproxy/deltaproxy.py", line 356, in post_master_init
    sub_proxy_data = subproxy_post_master_init(
  File "/opt/saltstack/salt/lib/python3.10/site-packages/salt/metaproxy/deltaproxy.py", line 524, in subproxy_post_master_init
    _proxy_minion.schedule.add_job(
  File "/opt/saltstack/salt/lib/python3.10/site-packages/salt/utils/schedule.py", line 413, in add_job
    evt.fire_event(
  File "/opt/saltstack/salt/lib/python3.10/site-packages/salt/utils/event.py", line 835, in fire_event
    self.pusher.send(msg)
  File "/opt/saltstack/salt/lib/python3.10/site-packages/salt/utils/asynchronous.py", line 125, in wrap
    raise exc_info[1].with_traceback(exc_info[2])
  File "/opt/saltstack/salt/lib/python3.10/site-packages/salt/utils/asynchronous.py", line 131, in _target
    result = io_loop.run_sync(lambda: getattr(self.obj, key)(*args, **kwargs))
  File "/opt/saltstack/salt/lib/python3.10/site-packages/salt/ext/tornado/ioloop.py", line 459, in run_sync
    return future_cell[0].result()
  File "/opt/saltstack/salt/lib/python3.10/site-packages/salt/ext/tornado/concurrent.py", line 249, in result
    raise_exc_info(self._exc_info)
  File "<string>", line 4, in raise_exc_info
  File "/opt/saltstack/salt/lib/python3.10/site-packages/salt/ext/tornado/gen.py", line 1070, in run
    yielded = self.gen.send(value)
  File "/opt/saltstack/salt/lib/python3.10/site-packages/salt/transport/ipc.py", line 456, in send
    yield self.stream.write(pack)
  File "/opt/saltstack/salt/lib/python3.10/site-packages/salt/ext/tornado/iostream.py", line 388, in write
    self._check_closed()
  File "/opt/saltstack/salt/lib/python3.10/site-packages/salt/ext/tornado/iostream.py", line 926, in _check_closed
    raise StreamClosedError(real_error=self.error)
salt.ext.tornado.iostream.StreamClosedError: Stream is closed
/opt/saltstack/salt/lib/python3.10/site-packages/salt/transport/client.py:79: DeprecationWarning: This module is deprecated. Please use salt.channel.client instead.
  warn_until(
```

**Expected behavior**
Proxy should start gracefully


**Versions Report**
Salt Version:
Salt: 3005


[BUG] Add multiple Network Devices without restarting proxy each time per device.  
**Description**
Add multiple Network Devices without restarting proxy each time per device.  

**Setup**
Delta proxy devices loading from MongoDB script. There are 35 devices in one delta proxy and it requires 2-3 mins wait to accept key and it can be proceed only one device at a time.

Please be as specific as possible and give set-up details.
- [ ] VM (Virtualbox, KVM, etc. please specify)

**Steps to Reproduce the behavior**


**Expected behavior**
All devices keys should be accepted in one shot rather than serial acceptance and wait time.

**Screenshots**


**Versions Report**
Salt Version:
Salt: 3005

**Additional context**

Hints from issue discussion:
@vanditpatel619 This is related to  the bug related to using multiple threads when the number of devices configured for a control proxy is greater than 20, this causes an error to occur and the process will restart. If you can divide up your minions so that each control proxy has no more than 20 sub proxies, that should work around the issue until a proper fix can be implemented.
@vanditpatel619 Do you have the `parallel: True` option in the pillar data for the control proxies?  Also, I discovered a bug when the number of devices configured for a control proxy is greater than 20, this causes an error to occur and the process will restart.  If you can divide up your minions so that each control proxy has no more than 20 sub proxies.

Commit message: Adding one more sub proxy.

Patch to review:
diff --git a/changelog/64102.fixed.md b/changelog/64102.fixed.md
new file mode 100644
index 000000000000..64988c2f2e01
--- /dev/null
+++ b/changelog/64102.fixed.md
@@ -0,0 +1,4 @@
+Update all the scheduler functions to include a fire_event argument which will determine whether to fire the completion event onto the event bus.
+This event is only used when these functions are called via the schedule execution modules.
+Inside deltaproxy, then update all scheudle related functions to include fire_event=False,
+as the event bus is not available when these functions are called.
diff --git a/changelog/64103.fixed.md b/changelog/64103.fixed.md
new file mode 100644
index 000000000000..64988c2f2e01
--- /dev/null
+++ b/changelog/64103.fixed.md
@@ -0,0 +1,4 @@
+Update all the scheduler functions to include a fire_event argument which will determine whether to fire the completion event onto the event bus.
+This event is only used when these functions are called via the schedule execution modules.
+Inside deltaproxy, then update all scheudle related functions to include fire_event=False,
+as the event bus is not available when these functions are called.
diff --git a/salt/metaproxy/deltaproxy.py b/salt/metaproxy/deltaproxy.py
index c3003b368f7b..d866d6f4c1de 100644
--- a/salt/metaproxy/deltaproxy.py
+++ b/salt/metaproxy/deltaproxy.py
@@ -231,10 +231,11 @@ def post_master_init(self, master):
                 }
             },
             persist=True,
+            fire_event=False,
         )
         log.info("Added mine.update to scheduler")
     else:
-        self.schedule.delete_job("__mine_interval", persist=True)
+        self.schedule.delete_job("__mine_interval", persist=True, fire_event=False)
 
     # add master_alive job if enabled
     if self.opts["transport"] != "tcp" and self.opts["master_alive_interval"] > 0:
@@ -250,6 +251,7 @@ def post_master_init(self, master):
                 }
             },
             persist=True,
+            fire_event=False,
         )
         if (
             self.opts["master_failback"]
@@ -268,18 +270,24 @@ def post_master_init(self, master):
                     }
                 },
                 persist=True,
+                fire_event=False,
             )
         else:
             self.schedule.delete_job(
-                salt.minion.master_event(type="failback"), persist=True
+                salt.minion.master_event(type="failback"),
+                persist=True,
+                fire_event=False,
             )
     else:
         self.schedule.delete_job(
             salt.minion.master_event(type="alive", master=self.opts["master"]),
             persist=True,
+            fire_event=False,
         )
         self.schedule.delete_job(
-            salt.minion.master_event(type="failback"), persist=True
+            salt.minion.master_event(type="failback"),
+            persist=True,
+            fire_event=False,
         )
 
     # proxy keepalive
@@ -304,10 +312,15 @@ def post_master_init(self, master):
                 }
             },
             persist=True,
+            fire_event=False,
         )
-        self.schedule.enable_schedule()
+        self.schedule.enable_schedule(fire_event=False)
     else:
-        self.schedule.delete_job("__proxy_keepalive", persist=True)
+        self.schedule.delete_job(
+            "__proxy_keepalive",
+            persist=True,
+            fire_event=False,
+        )
 
     #  Sync the grains here so the proxy can communicate them to the master
     self.functions["saltutil.sync_grains"](saltenv="base")
@@ -321,10 +334,11 @@ def post_master_init(self, master):
     self.proxy_context = {}
     self.add_periodic_callback("cleanup", self.cleanup_subprocesses)
 
+    _failed = list()
     if self.opts["proxy"].get("parallel_startup"):
         log.debug("Initiating parallel startup for proxies")
         with concurrent.futures.ThreadPoolExecutor() as executor:
-            futures = [
+            futures = {
                 executor.submit(
                     subproxy_post_master_init,
                     _id,
@@ -332,12 +346,22 @@ def post_master_init(self, master):
                     self.opts,
                     self.proxy,
                     self.utils,
-                )
+                ): _id
                 for _id in self.opts["proxy"].get("ids", [])
-            ]
+            }
 
-        for f in concurrent.futures.as_completed(futures):
-            sub_proxy_data = f.result()
+        for future in concurrent.futures.as_completed(futures):
+            try:
+                sub_proxy_data = future.result()
+            except Exception as exc:  # pylint: disable=broad-except
+                _id = futures[future]
+                log.info(
+                    "An exception occured during initialization for %s, skipping: %s",
+                    _id,
+                    exc,
+                )
+                _failed.append(_id)
+                continue
             minion_id = sub_proxy_data["proxy_opts"].get("id")
 
             if sub_proxy_data["proxy_minion"]:
@@ -347,16 +371,24 @@ def post_master_init(self, master):
                 if self.deltaproxy_opts[minion_id] and self.deltaproxy_objs[minion_id]:
                     self.deltaproxy_objs[
                         minion_id
-                    ].req_channel = salt.transport.client.AsyncReqChannel.factory(
+                    ].req_channel = salt.channel.client.AsyncReqChannel.factory(
                         sub_proxy_data["proxy_opts"], io_loop=self.io_loop
                     )
     else:
         log.debug("Initiating non-parallel startup for proxies")
         for _id in self.opts["proxy"].get("ids", []):
-            sub_proxy_data = subproxy_post_master_init(
-                _id, uid, self.opts, self.proxy, self.utils
-            )
-
+            try:
+                sub_proxy_data = subproxy_post_master_init(
+                    _id, uid, self.opts, self.proxy, self.utils
+                )
+            except Exception as exc:  # pylint: disable=broad-except
+                log.info(
+                    "An exception occured during initialization for %s, skipping: %s",
+                    _id,
+                    exc,
+                )
+                _failed.append(_id)
+                continue
             minion_id = sub_proxy_data["proxy_opts"].get("id")
 
             if sub_proxy_data["proxy_minion"]:
@@ -366,10 +398,12 @@ def post_master_init(self, master):
                 if self.deltaproxy_opts[minion_id] and self.deltaproxy_objs[minion_id]:
                     self.deltaproxy_objs[
                         minion_id
-                    ].req_channel = salt.transport.client.AsyncReqChannel.factory(
+                    ].req_channel = salt.channel.client.AsyncReqChannel.factory(
                         sub_proxy_data["proxy_opts"], io_loop=self.io_loop
                     )
 
+    if _failed:
+        log.info("Following sub proxies failed %s", _failed)
     self.ready = True
 
 
@@ -535,10 +569,13 @@ def subproxy_post_master_init(minion_id, uid, opts, main_proxy, main_utils):
                 }
             },
             persist=True,
+            fire_event=False,
         )
-        _proxy_minion.schedule.enable_schedule()
+        _proxy_minion.schedule.enable_schedule(fire_event=False)
     else:
-        _proxy_minion.schedule.delete_job("__proxy_keepalive", persist=True)
+        _proxy_minion.schedule.delete_job(
+            "__proxy_keepalive", persist=True, fire_event=False
+        )
 
     return {"proxy_minion": _proxy_minion, "proxy_opts": proxyopts}
 
diff --git a/salt/utils/schedule.py b/salt/utils/schedule.py
index 814c2980d4af..6565dda59e6e 100644
--- a/salt/utils/schedule.py
+++ b/salt/utils/schedule.py
@@ -315,7 +315,7 @@ def persist(self):
                 exc_info_on_loglevel=logging.DEBUG,
             )
 
-    def delete_job(self, name, persist=True):
+    def delete_job(self, name, persist=True, fire_event=True):
         """
         Deletes a job from the scheduler. Ignore jobs from pillar
         """
@@ -325,12 +325,15 @@ def delete_job(self, name, persist=True):
         elif name in self._get_schedule(include_opts=False):
             log.warning("Cannot delete job %s, it's in the pillar!", 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, "schedule": self._get_schedule()},
-                tag="/salt/minion/minion_schedule_delete_complete",
-            )
+        if fire_event:
+            # 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, "schedule": self._get_schedule()},
+                    tag="/salt/minion/minion_schedule_delete_complete",
+                )
 
         # remove from self.intervals
         if name in self.intervals:
@@ -349,7 +352,7 @@ def reset(self):
         self.splay = None
         self.opts["schedule"] = {}
 
-    def delete_job_prefix(self, name, persist=True):
+    def delete_job_prefix(self, name, persist=True, fire_event=True):
         """
         Deletes a job from the scheduler. Ignores jobs from pillar
         """
@@ -361,12 +364,15 @@ def delete_job_prefix(self, name, persist=True):
             if job.startswith(name):
                 log.warning("Cannot delete job %s, it's in the pillar!", job)
 
-        # 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, "schedule": self._get_schedule()},
-                tag="/salt/minion/minion_schedule_delete_complete",
-            )
+        if fire_event:
+            # 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, "schedule": self._get_schedule()},
+                    tag="/salt/minion/minion_schedule_delete_complete",
+                )
 
         # remove from self.intervals
         for job in list(self.intervals.keys()):
@@ -376,7 +382,7 @@ def delete_job_prefix(self, name, persist=True):
         if persist:
             self.persist()
 
-    def add_job(self, data, persist=True):
+    def add_job(self, data, persist=True, fire_event=True):
         """
         Adds a new job to the scheduler. The format is the same as required in
         the configuration file. See the docs on how YAML is interpreted into
@@ -410,16 +416,19 @@ def add_job(self, data, persist=True):
             self.opts["schedule"].update(data)
 
         # 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, "schedule": self._get_schedule()},
-                tag="/salt/minion/minion_schedule_add_complete",
-            )
+        if fire_event:
+            with salt.utils.event.get_event(
+                "minion", opts=self.opts, listen=False
+            ) as evt:
+                evt.fire_event(
+                    {"complete": True, "schedule": self._get_schedule()},
+                    tag="/salt/minion/minion_schedule_add_complete",
+                )
 
         if persist:
             self.persist()
 
-    def enable_job(self, name, persist=True):
+    def enable_job(self, name, persist=True, fire_event=True):
         """
         Enable a job in the scheduler. Ignores jobs from pillar
         """
@@ -430,17 +439,20 @@ def enable_job(self, name, persist=True):
         elif name in self._get_schedule(include_opts=False):
             log.warning("Cannot modify job %s, it's in the pillar!", 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, "schedule": self._get_schedule()},
-                tag="/salt/minion/minion_schedule_enabled_job_complete",
-            )
+        if fire_event:
+            # 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, "schedule": self._get_schedule()},
+                    tag="/salt/minion/minion_schedule_enabled_job_complete",
+                )
 
         if persist:
             self.persist()
 
-    def disable_job(self, name, persist=True):
+    def disable_job(self, name, persist=True, fire_event=True):
         """
         Disable a job in the scheduler. Ignores jobs from pillar
         """
@@ -451,23 +463,26 @@ def disable_job(self, name, persist=True):
         elif name in self._get_schedule(include_opts=False):
             log.warning("Cannot modify job %s, it's in the pillar!", name)
 
-        with salt.utils.event.get_event("minion", opts=self.opts, listen=False) as evt:
-            # Fire the complete event back along with updated list of schedule
-            evt.fire_event(
-                {"complete": True, "schedule": self._get_schedule()},
-                tag="/salt/minion/minion_schedule_disabled_job_complete",
-            )
+        if fire_event:
+            with salt.utils.event.get_event(
+                "minion", opts=self.opts, listen=False
+            ) as evt:
+                # Fire the complete event back along with updated list of schedule
+                evt.fire_event(
+                    {"complete": True, "schedule": self._get_schedule()},
+                    tag="/salt/minion/minion_schedule_disabled_job_complete",
+                )
 
         if persist:
             self.persist()
 
-    def modify_job(self, name, schedule, persist=True):
+    def modify_job(self, name, schedule, persist=True, fire_event=True):
         """
         Modify a job in the scheduler. Ignores jobs from pillar
         """
         # ensure job exists, then replace it
         if name in self.opts["schedule"]:
-            self.delete_job(name, persist)
+            self.delete_job(name, persist, fire_event)
         elif name in self._get_schedule(include_opts=False):
             log.warning("Cannot modify job %s, it's in the pillar!", name)
             return
@@ -511,34 +526,40 @@ def run_job(self, name):
             log.info("Running Job: %s", name)
             self._run_job(func, data)
 
-    def enable_schedule(self, persist=True):
+    def enable_schedule(self, persist=True, fire_event=True):
         """
         Enable the scheduler.
         """
         self.opts["schedule"]["enabled"] = True
 
-        # 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, "schedule": self._get_schedule()},
-                tag="/salt/minion/minion_schedule_enabled_complete",
-            )
+        if fire_event:
+            # 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, "schedule": self._get_schedule()},
+                    tag="/salt/minion/minion_schedule_enabled_complete",
+                )
 
         if persist:
             self.persist()
 
-    def disable_schedule(self, persist=True):
+    def disable_schedule(self, persist=True, fire_event=True):
         """
         Disable the scheduler.
         """
         self.opts["schedule"]["enabled"] = False
 
-        # 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, "schedule": self._get_schedule()},
-                tag="/salt/minion/minion_schedule_disabled_complete",
-            )
+        if fire_event:
+            # 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, "schedule": self._get_schedule()},
+                    tag="/salt/minion/minion_schedule_disabled_complete",
+                )
 
         if persist:
             self.persist()
@@ -554,7 +575,7 @@ def reload(self, schedule):
             schedule = schedule["schedule"]
         self.opts.setdefault("schedule", {}).update(schedule)
 
-    def list(self, where):
+    def list(self, where, fire_event=True):
         """
         List the current schedule items
         """
@@ -565,24 +586,32 @@ def list(self, where):
         else:
             schedule = self._get_schedule()
 
-        # Fire the complete event back along with the list of schedule
-        with salt.utils.event.get_event("minion", opts=self.opts, listen=False) as evt:
-            evt.fire_event(
-                {"complete": True, "schedule": schedule},
-                tag="/salt/minion/minion_schedule_list_complete",
-            )
+        if fire_event:
+            # Fire the complete event back along with the list of schedule
+            with salt.utils.event.get_event(
+                "minion", opts=self.opts, listen=False
+            ) as evt:
+                evt.fire_event(
+                    {"complete": True, "schedule": schedule},
+                    tag="/salt/minion/minion_schedule_list_complete",
+                )
 
-    def save_schedule(self):
+    def save_schedule(self, fire_event=True):
         """
         Save the current schedule
         """
         self.persist()
 
-        # Fire the complete event back along with the list of schedule
-        with salt.utils.event.get_event("minion", opts=self.opts, listen=False) as evt:
-            evt.fire_event({"complete": True}, tag="/salt/minion/minion_schedule_saved")
+        if fire_event:
+            # Fire the complete event back along with the list of schedule
+            with salt.utils.event.get_event(
+                "minion", opts=self.opts, listen=False
+            ) as evt:
+                evt.fire_event(
+                    {"complete": True}, tag="/salt/minion/minion_schedule_saved"
+                )
 
-    def postpone_job(self, name, data):
+    def postpone_job(self, name, data, fire_event=True):
         """
         Postpone a job in the scheduler.
         Ignores jobs from pillar
@@ -608,14 +637,17 @@ def postpone_job(self, name, data):
         elif name in self._get_schedule(include_opts=False):
             log.warning("Cannot modify job %s, it's in the pillar!", 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, "schedule": self._get_schedule()},
-                tag="/salt/minion/minion_schedule_postpone_job_complete",
-            )
+        if fire_event:
+            # 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, "schedule": self._get_schedule()},
+                    tag="/salt/minion/minion_schedule_postpone_job_complete",
+                )
 
-    def skip_job(self, name, data):
+    def skip_job(self, name, data, fire_event=True):
         """
         Skip a job at a specific time in the scheduler.
         Ignores jobs from pillar
@@ -634,14 +666,17 @@ def skip_job(self, name, data):
         elif name in self._get_schedule(include_opts=False):
             log.warning("Cannot modify job %s, it's in the pillar!", 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, "schedule": self._get_schedule()},
-                tag="/salt/minion/minion_schedule_skip_job_complete",
-            )
+        if fire_event:
+            # 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, "schedule": self._get_schedule()},
+                    tag="/salt/minion/minion_schedule_skip_job_complete",
+                )
 
-    def get_next_fire_time(self, name, fmt="%Y-%m-%dT%H:%M:%S"):
+    def get_next_fire_time(self, name, fmt="%Y-%m-%dT%H:%M:%S", fire_event=True):
         """
         Return the next fire time for the specified job
         """
@@ -653,12 +688,15 @@ def get_next_fire_time(self, name, fmt="%Y-%m-%dT%H:%M:%S"):
             if _next_fire_time:
                 _next_fire_time = _next_fire_time.strftime(fmt)
 
-        # 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, "next_fire_time": _next_fire_time},
-                tag="/salt/minion/minion_schedule_next_fire_time_complete",
-            )
+        if fire_event:
+            # 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, "next_fire_time": _next_fire_time},
+                    tag="/salt/minion/minion_schedule_next_fire_time_complete",
+                )
 
     def job_status(self, name, fire_event=False):
         """
diff --git a/tests/pytests/functional/cli/test_salt_deltaproxy.py b/tests/pytests/functional/cli/test_salt_deltaproxy.py
new file mode 100644
index 000000000000..5bc7604c84ab
--- /dev/null
+++ b/tests/pytests/functional/cli/test_salt_deltaproxy.py
@@ -0,0 +1,225 @@
+"""
+:codeauthor: Gareth J. Greenaway (ggreenaway@vmware.com)
+"""
+
+import logging
+import os
+import random
+
+import pytest
+from saltfactories.utils import random_string
+
+import salt.defaults.exitcodes
+from tests.support.helpers import PRE_PYTEST_SKIP_REASON
+
+log = logging.getLogger(__name__)
+
+
+pytestmark = [
+    pytest.mark.skip_on_spawning_platform(
+        reason="Deltaproxy minions do not currently work on spawning platforms.",
+    ),
+    pytest.mark.core_test,
+]
+
+
+@pytest.fixture(scope="package")
+def salt_master(salt_factories):
+    config_defaults = {
+        "open_mode": True,
+    }
+    salt_master = salt_factories.salt_master_daemon(
+        "deltaproxy-functional-master", defaults=config_defaults
+    )
+    with salt_master.started():
+        yield salt_master
+
+
+@pytest.fixture(scope="package")
+def salt_cli(salt_master):
+    """
+    The ``salt`` CLI as a fixture against the running master
+    """
+    assert salt_master.is_running()
+    return salt_master.salt_cli(timeout=30)
+
+
+@pytest.fixture(scope="package", autouse=True)
+def skip_on_tcp_transport(request):
+    if request.config.getoption("--transport") == "tcp":
+        pytest.skip("Deltaproxy under the TPC transport is not working. See #61367")
+
+
+@pytest.fixture
+def proxy_minion_id(salt_master):
+    _proxy_minion_id = random_string("proxy-minion-")
+
+    try:
+        yield _proxy_minion_id
+    finally:
+        # Remove stale key if it exists
+        pytest.helpers.remove_stale_minion_key(salt_master, _proxy_minion_id)
+
+
+def clear_proxy_minions(salt_master, proxy_minion_id):
+    for proxy in [proxy_minion_id, "dummy_proxy_one", "dummy_proxy_two"]:
+        pytest.helpers.remove_stale_minion_key(salt_master, proxy)
+
+        cachefile = os.path.join(
+            salt_master.config["cachedir"], "{}.cache".format(proxy)
+        )
+        if os.path.exists(cachefile):
+            os.unlink(cachefile)
+
+
+# Hangs on Windows. You can add a timeout to the proxy.run command, but then
+# it just times out.
+@pytest.mark.skip_on_windows(reason=PRE_PYTEST_SKIP_REASON)
+@pytest.mark.parametrize(
+    "parallel_startup",
+    [True, False],
+    ids=["parallel_startup=True", "parallel_startup=False"],
+)
+def test_exit_status_correct_usage_large_number_of_minions(
+    salt_master,
+    salt_cli,
+    proxy_minion_id,
+    parallel_startup,
+):
+    """
+    Ensure the salt-proxy control proxy starts and
+    is able to respond to test.ping, additionally ensure that
+    the proxies being controlled also respond to test.ping.
+
+    Finally ensure correct exit status when salt-proxy exits correctly.
+
+    Skip on Windows because daemonization not supported
+    """
+
+    config_defaults = {
+        "metaproxy": "deltaproxy",
+    }
+    sub_proxies = [
+        "proxy_one",
+        "proxy_two",
+        "proxy_three",
+        "proxy_four",
+        "proxy_five",
+        "proxy_six",
+        "proxy_seven",
+        "proxy_eight",
+        "proxy_nine",
+        "proxy_ten",
+        "proxy_eleven",
+        "proxy_twelve",
+        "proxy_thirteen",
+        "proxy_fourteen",
+        "proxy_fifteen",
+        "proxy_sixteen",
+        "proxy_seventeen",
+        "proxy_eighteen",
+        "proxy_nineteen",
+        "proxy_twenty",
+        "proxy_twenty_one",
+        "proxy_twenty_two",
+        "proxy_twenty_three",
+        "proxy_twenty_four",
+        "proxy_twenty_five",
+        "proxy_twenty_six",
+        "proxy_twenty_seven",
+        "proxy_twenty_eight",
+        "proxy_twenty_nine",
+        "proxy_thirty",
+        "proxy_thirty_one",
+        "proxy_thirty_two",
+    ]
+
+    top_file = """
+    base:
+      {control}:
+        - controlproxy
+    """.format(
+        control=proxy_minion_id,
+    )
+    controlproxy_pillar_file = """
+    proxy:
+        proxytype: deltaproxy
+        parallel_startup: {}
+        ids:
+    """.format(
+        parallel_startup
+    )
+
+    dummy_proxy_pillar_file = """
+    proxy:
+      proxytype: dummy
+    """
+
+    for minion_id in sub_proxies:
+        top_file += """
+      {minion_id}:
+        - dummy""".format(
+            minion_id=minion_id,
+        )
+
+        controlproxy_pillar_file += """
+            - {}
+        """.format(
+            minion_id,
+        )
+
+    top_tempfile = salt_master.pillar_tree.base.temp_file("top.sls", top_file)
+    controlproxy_tempfile = salt_master.pillar_tree.base.temp_file(
+        "controlproxy.sls", controlproxy_pillar_file
+    )
+    dummy_proxy_tempfile = salt_master.pillar_tree.base.temp_file(
+        "dummy.sls",
+        dummy_proxy_pillar_file,
+    )
+    with top_tempfile, controlproxy_tempfile, dummy_proxy_tempfile:
+        with salt_master.started():
+            assert salt_master.is_running()
+
+            factory = salt_master.salt_proxy_minion_daemon(
+                proxy_minion_id,
+                defaults=config_defaults,
+                extra_cli_arguments_after_first_start_failure=["--log-level=info"],
+                start_timeout=240,
+            )
+
+            for minion_id in [proxy_minion_id] + sub_proxies:
+                factory.before_start(
+                    pytest.helpers.remove_stale_proxy_minion_cache_file,
+                    factory,
+                    minion_id,
+                )
+                factory.after_terminate(
+                    pytest.helpers.remove_stale_minion_key, salt_master, minion_id
+                )
+                factory.after_terminate(
+                    pytest.helpers.remove_stale_proxy_minion_cache_file,
+                    factory,
+                    minion_id,
+                )
+
+            with factory.started():
+                assert factory.is_running()
+
+                # Let's issue a ping the control proxy
+                ret = salt_cli.run("test.ping", minion_tgt=proxy_minion_id)
+                assert ret.returncode == 0
+                assert ret.data is True
+
+                for minion_id in random.sample(sub_proxies, 4):
+                    # Let's issue a ping to one of the controlled proxies
+                    ret = salt_cli.run("test.ping", minion_tgt=minion_id)
+                    assert ret.returncode == 0
+                    assert ret.data is True
+
+        # Terminate the proxy minion
+        ret = factory.terminate()
+        assert ret.returncode == salt.defaults.exitcodes.EX_OK, ret
+
+        # Terminate the salt master
+        ret = salt_master.terminate()
+        assert ret.returncode == salt.defaults.exitcodes.EX_OK, ret


Code review comments:
diff_hunk: @@ -0,0 +1,4 @@
+Update all the scheduler functions to include a fire_event argument which will determine whether to fire the completion event onto the event bus.
+This event is only used when these functions are called via the schedule execution modules.
+Inside deltaproxy, then update all scheudle related functions to include fire_event=False,
path: changelog/64102.fixed.md
text: @user1:
```suggestion
Inside deltaproxy, then update all scheudle related functions to include fire_event=False,
```
Is this sentence complete?

@author:
No.  That does look incomplete.  Thanks for catching that.