fire event at job start

salt/metaproxy/deltaproxy.py

@@ -51,6 +51,7 @@
     SaltInvocationError,
     SaltSystemExit,
 )
+from salt.utils.decorators.start_event import fire_started_event_job_wrapper
 from salt.minion import ProxyMinion
 from salt.utils.event import tagify
 from salt.utils.process import SignalHandlingProcess, default_signals
@@ -609,6 +610,7 @@ def target(cls, minion_instance, opts, data, connected, creds_map):
             ProxyMinion._thread_return(minion_instance, opts, data)
 
 
+@fire_started_event_job_wrapper
 def thread_return(cls, minion_instance, opts, data):
     """
     This method should be used as a threading target, start the actual
@@ -858,6 +860,7 @@ def thread_return(cls, minion_instance, opts, data):
                 log.exception("The return failed for job %s: %s", data["jid"], exc)
 
 
+@fire_started_event_job_wrapper
 def thread_multi_return(cls, minion_instance, opts, data):
     """
     This method should be used as a threading target, start the actual

salt/metaproxy/proxy.py

@@ -50,6 +50,7 @@
     SaltSystemExit,
 )
 from salt.minion import ProxyMinion
+from salt.utils.decorators.start_event import fire_started_event_job_wrapper
 from salt.utils.event import tagify
 from salt.utils.process import SignalHandlingProcess, default_signals
 
@@ -393,6 +394,7 @@ def target(cls, minion_instance, opts, data, connected, creds_map):
             ProxyMinion._thread_return(minion_instance, opts, data)
 
 
+@fire_started_event_job_wrapper
 def thread_return(cls, minion_instance, opts, data):
     """
     This method should be used as a threading target, start the actual
@@ -632,6 +634,7 @@ def thread_return(cls, minion_instance, opts, data):
                 log.exception("The return failed for job %s: %s", data["jid"], exc)
 
 
+@fire_started_event_job_wrapper
 def thread_multi_return(cls, minion_instance, opts, data):
     """
     This method should be used as a threading target, start the actual

salt/minion.py

@@ -72,6 +72,7 @@
 from salt.template import SLS_ENCODING
 from salt.utils.ctx import RequestContext
 from salt.utils.debug import enable_sigusr1_handler
+from salt.utils.decorators.start_event import fire_started_event_job_wrapper
 from salt.utils.event import tagify
 from salt.utils.network import parse_host_port
 from salt.utils.odict import OrderedDict
@@ -1906,6 +1907,7 @@ def _execute_job_function(
         return None
 
     @classmethod
+    @fire_started_event_job_wrapper
     def _thread_return(cls, minion_instance, opts, data):
         """
         This method should be used as a threading target, start the actual
@@ -2107,6 +2109,7 @@ def _thread_return(cls, minion_instance, opts, data):
                     log.exception("The return failed for job %s: %s", data["jid"], exc)
 
     @classmethod
+    @fire_started_event_job_wrapper
     def _thread_multi_return(cls, minion_instance, opts, data):
         """
         This method should be used as a threading target, start the actual

salt/utils/decorators/start_event.py

@@ -0,0 +1,31 @@
+import os
+import logging
+from functools import wraps
+
+log = logging.getLogger(__name__)
+
+
+def fire_started_event_job_wrapper(function):
+    """
+    Fire a custom/job/<jid>/started/<minion_id> event when starting a job
+    """
+
+    @wraps(function)
+    def wrapped(*args, **kwargs):
+        """We wraps for *args and **kwargs, to be safe, but we expect: (cls, minion_instance, opts, data) as args"""
+
+        try:
+            sdata = {"pid": os.getpid()}
+            sdata.update(args[3])
+
+            jid = sdata["jid"]
+            minion_id = args[1].opts.get("id", "undefined")
+
+            args[1]._fire_master(sdata, f"custom/job/{jid}/started/{minion_id}")
+
+        except Exception as exc:
+            log.warning(f"Exception raised during started job event wrapper: {exc}")
+
+        return function(*args, **kwargs)
+
+    return wrapped