Fix wrong retcodes from concurrent jobs on a threaded minion

changelog/61830.fixed.md

@@ -0,0 +1 @@
+Fixed threaded minion jobs (``multiprocessing: False``) reporting the wrong retcode, including a failed command returning success. Every job rebuilds and rebinds the shared ``minion_instance.functions`` at its top via ``gen_modules()``; a sibling job's rebind landing between a job's retcode write and its read made ``_thread_return`` read a fresh, empty ``__context__`` and deliver ``EX_OK``. ``gen_modules()`` now returns the loader generation it built, and ``_thread_return``/``_thread_multi_return`` capture it and read the retcode from the loader the job owns instead of the shared attribute. The ``sys.reload_modules`` binding was pointed at a new ``_reload_modules`` wrapper so it keeps returning ``None`` on the wire. The proxy and delta-proxy job runners got the same loader-capture treatment.

salt/metaproxy/deltaproxy.py

@@ -655,7 +655,11 @@ def thread_return(cls, minion_instance, opts, data):
             if f"{executor}.allow_missing_func" in minion_instance.executors
         ]
     )
-    if function_name in minion_instance.functions or allow_missing_funcs is True:
+    # Own the loader for the length of this job so the func lookup, the retcode
+    # reset and the retcode read all resolve against the same generation, even if
+    # a concurrent sys.reload_modules rebinds minion_instance.functions. See #61830.
+    functions = minion_instance.functions
+    if function_name in functions or allow_missing_funcs is True:
         try:
             minion_blackout_violation = False
             if minion_instance.connected and minion_instance.opts["pillar"].get(
@@ -687,15 +691,15 @@ def thread_return(cls, minion_instance, opts, data):
                     "saltutil.refresh_pillar allowed in blackout mode."
                 )
 
-            if function_name in minion_instance.functions:
-                func = minion_instance.functions[function_name]
+            if function_name in functions:
+                func = functions[function_name]
                 args, kwargs = salt.minion.load_args_and_kwargs(func, data["arg"], data)
             else:
                 # only run if function_name is not in minion_instance.functions and allow_missing_funcs is True
                 func = function_name
                 args, kwargs = data["arg"], data
-            minion_instance.functions.pack["__context__"]["retcode"] = 0
-            minion_instance.functions.pack["__opts__"] = opts
+            functions.pack["__context__"]["retcode"] = 0
+            functions.pack["__opts__"] = opts
             if isinstance(executors, str):
                 executors = [executors]
             elif not isinstance(executors, list) or not executors:
@@ -737,7 +741,7 @@ def thread_return(cls, minion_instance, opts, data):
             else:
                 ret["return"] = return_data
 
-            retcode = minion_instance.functions.pack["__context__"].get(
+            retcode = functions.pack["__context__"].get(
                 "retcode", salt.defaults.exitcodes.EX_OK
             )
             if retcode == salt.defaults.exitcodes.EX_OK:
@@ -897,6 +901,10 @@ def thread_multi_return(cls, minion_instance, opts, data):
     else:
         ret = {"return": {}, "retcode": {}, "success": {}}
 
+    # Own the loader for the length of this job so every func's lookup, retcode
+    # reset and retcode read resolve against the same generation, even if a
+    # concurrent sys.reload_modules rebinds minion_instance.functions. See #61830.
+    functions = minion_instance.functions
     for ind in range(0, num_funcs):
         if not multifunc_ordered:
             ret["success"][data["fun"][ind]] = False
@@ -930,15 +938,15 @@ def thread_multi_return(cls, minion_instance, opts, data):
                     "saltutil.refresh_pillar allowed in blackout mode."
                 )
 
-            func = minion_instance.functions[data["fun"][ind]]
+            func = functions[data["fun"][ind]]
 
             args, kwargs = salt.minion.load_args_and_kwargs(
                 func, data["arg"][ind], data
             )
-            minion_instance.functions.pack["__context__"]["retcode"] = 0
+            functions.pack["__context__"]["retcode"] = 0
             key = ind if multifunc_ordered else data["fun"][ind]
             ret["return"][key] = func(*args, **kwargs)
-            retcode = minion_instance.functions.pack["__context__"].get("retcode", 0)
+            retcode = functions.pack["__context__"].get("retcode", 0)
             if retcode == 0:
                 # No nonzero retcode in __context__ dunder. Check if return
                 # is a dictionary with a "result" or "success" key.

salt/metaproxy/proxy.py

@@ -418,7 +418,11 @@ def thread_return(cls, minion_instance, opts, data):
             if f"{executor}.allow_missing_func" in minion_instance.executors
         ]
     )
-    if function_name in minion_instance.functions or allow_missing_funcs is True:
+    # Own the loader for the length of this job so the func lookup, the retcode
+    # reset and the retcode read all resolve against the same generation, even if
+    # a concurrent sys.reload_modules rebinds minion_instance.functions. See #61830.
+    functions = minion_instance.functions
+    if function_name in functions or allow_missing_funcs is True:
         try:
             minion_blackout_violation = False
             if minion_instance.connected and minion_instance.opts["pillar"].get(
@@ -450,14 +454,14 @@ def thread_return(cls, minion_instance, opts, data):
                     "saltutil.refresh_pillar allowed in blackout mode."
                 )
 
-            if function_name in minion_instance.functions:
-                func = minion_instance.functions[function_name]
+            if function_name in functions:
+                func = functions[function_name]
                 args, kwargs = salt.minion.load_args_and_kwargs(func, data["arg"], data)
             else:
                 # only run if function_name is not in minion_instance.functions and allow_missing_funcs is True
                 func = function_name
                 args, kwargs = data["arg"], data
-            minion_instance.functions.pack["__context__"]["retcode"] = 0
+            functions.pack["__context__"]["retcode"] = 0
             if isinstance(executors, str):
                 executors = [executors]
             elif not isinstance(executors, list) or not executors:
@@ -497,7 +501,7 @@ def thread_return(cls, minion_instance, opts, data):
             else:
                 ret["return"] = return_data
 
-            retcode = minion_instance.functions.pack["__context__"].get(
+            retcode = functions.pack["__context__"].get(
                 "retcode", salt.defaults.exitcodes.EX_OK
             )
             if retcode == salt.defaults.exitcodes.EX_OK:
@@ -651,6 +655,10 @@ def thread_multi_return(cls, minion_instance, opts, data):
     else:
         ret = {"return": {}, "retcode": {}, "success": {}}
 
+    # Own the loader for the length of this job so every func's lookup, retcode
+    # reset and retcode read resolve against the same generation, even if a
+    # concurrent sys.reload_modules rebinds minion_instance.functions. See #61830.
+    functions = minion_instance.functions
     for ind in range(0, num_funcs):
         if not multifunc_ordered:
             ret["success"][data["fun"][ind]] = False
@@ -684,15 +692,15 @@ def thread_multi_return(cls, minion_instance, opts, data):
                     "saltutil.refresh_pillar allowed in blackout mode."
                 )
 
-            func = minion_instance.functions[data["fun"][ind]]
+            func = functions[data["fun"][ind]]
 
             args, kwargs = salt.minion.load_args_and_kwargs(
                 func, data["arg"][ind], data
             )
-            minion_instance.functions.pack["__context__"]["retcode"] = 0
+            functions.pack["__context__"]["retcode"] = 0
             key = ind if multifunc_ordered else data["fun"][ind]
             ret["return"][key] = func(*args, **kwargs)
-            retcode = minion_instance.functions.pack["__context__"].get("retcode", 0)
+            retcode = functions.pack["__context__"].get("retcode", 0)
             if retcode == 0:
                 # No nonzero retcode in __context__ dunder. Check if return
                 # is a dictionary with a "result" or "success" key.

salt/minion.py

@@ -512,34 +512,47 @@ def gen_modules(self, initial_load=False, context=None):
             ).compile_pillar()
 
         self.utils = salt.loader.utils(self.opts, context=context)
-        self.functions = salt.loader.minion_mods(
+        functions = salt.loader.minion_mods(
             self.opts, utils=self.utils, context=context
         )
+        self.functions = functions
         self.serializers = salt.loader.serializers(self.opts)
         self.returners = salt.loader.returners(
-            self.opts, functions=self.functions, context=context
+            self.opts, functions=functions, context=context
         )
         self.proxy = salt.loader.proxy(
-            self.opts, functions=self.functions, returners=self.returners
+            self.opts, functions=functions, returners=self.returners
         )
         # TODO: remove
         self.function_errors = {}  # Keep the funcs clean
         self.states = salt.loader.states(
             self.opts,
-            functions=self.functions,
+            functions=functions,
             utils=self.utils,
             serializers=self.serializers,
             context=context,
         )
-        self.rend = salt.loader.render(
-            self.opts, functions=self.functions, context=context
-        )
+        self.rend = salt.loader.render(self.opts, functions=functions, context=context)
         #        self.matcher = Matcher(self.opts, self.functions)
         self.matchers = salt.loader.matchers(self.opts)
-        self.functions["sys.reload_modules"] = self.gen_modules
+        functions["sys.reload_modules"] = self._reload_modules
         self.executors = salt.loader.executors(
-            self.opts, functions=self.functions, proxy=self.proxy, context=context
+            self.opts, functions=functions, proxy=self.proxy, context=context
         )
+        # Return the loader generation this call built so that a threaded job
+        # (multiprocessing=False) can own the exact loader it wrote its retcode
+        # into, rather than re-reading the shared self.functions attribute which
+        # a sibling job's gen_modules() may have rebound. See issue #61830.
+        return functions
+
+    def _reload_modules(self, initial_load=False, context=None):
+        """
+        The ``sys.reload_modules`` execution function. gen_modules() returns the
+        rebuilt loader (a LazyLoader, which is not serializable), so it cannot be
+        bound to ``sys.reload_modules`` directly -- the returned value would be
+        put on the wire. This wrapper reloads the modules and returns None.
+        """
+        self.gen_modules(initial_load=initial_load, context=context)
 
     @staticmethod
     def process_schedule(minion, loop_interval):
@@ -2613,12 +2626,19 @@ def run_func(minion_instance, opts, data):
                 run_func(minion_instance, opts, data)
 
     def _execute_job_function(
-        self, function_name, function_args, executors, opts, data
+        self, function_name, function_args, executors, opts, data, functions=None
     ):
         """
         Executes a function within a job given it's name, the args and the executors.
         It also checks if the function is allowed to run if 'blackout mode' is enabled.
+
+        ``functions`` is the loader generation this job owns (returned by
+        gen_modules()); the func lookup and the retcode reset must use it rather
+        than the shared self.functions, which a sibling job may have rebound. It
+        defaults to self.functions for callers that do not thread a loader.
         """
+        if functions is None:
+            functions = self.functions
         minion_blackout_violation = False
         if self.connected and self.opts["pillar"].get("minion_blackout", False):
             whitelist = self.opts["pillar"].get("minion_blackout_whitelist", [])
@@ -2643,14 +2663,14 @@ def _execute_job_function(
                 "saltutil.refresh_pillar allowed in blackout mode."
             )
 
-        if function_name in self.functions:
-            func = self.functions[function_name]
+        if function_name in functions:
+            func = functions[function_name]
             args, kwargs = load_args_and_kwargs(func, function_args, data)
         else:
             # only run if function_name is not in minion_instance.functions and allow_missing_funcs is True
             func = function_name
             args, kwargs = function_args, data
-        self.functions.pack["__context__"]["retcode"] = 0
+        functions.pack["__context__"]["retcode"] = 0
 
         if isinstance(executors, str):
             executors = [executors]
@@ -2680,7 +2700,7 @@ def _thread_return(cls, minion_instance, opts, data):
         This method should be used as a threading target, start the actual
         minion side execution.
         """
-        minion_instance.gen_modules()
+        functions = minion_instance.gen_modules()
         fn_ = os.path.join(minion_instance.proc_dir, str(data["jid"]))
 
         if opts.get("multiprocessing", True):
@@ -2709,13 +2729,10 @@ def _thread_return(cls, minion_instance, opts, data):
             ]
         )
         try:
-            if (
-                function_name in minion_instance.functions
-                or allow_missing_funcs is True
-            ):
+            if function_name in functions or allow_missing_funcs is True:
                 try:
                     return_data = minion_instance._execute_job_function(
-                        function_name, function_args, executors, opts, data
+                        function_name, function_args, executors, opts, data, functions
                     )
                     log.info(
                         "Job %s execution finished, return_data: %s",
@@ -2743,7 +2760,7 @@ def _thread_return(cls, minion_instance, opts, data):
                     else:
                         ret["return"] = return_data
 
-                    retcode = minion_instance.functions.pack["__context__"].get(
+                    retcode = functions.pack["__context__"].get(
                         "retcode", salt.defaults.exitcodes.EX_OK
                     )
                     if retcode == salt.defaults.exitcodes.EX_OK:
@@ -2924,7 +2941,7 @@ def _thread_multi_return(cls, minion_instance, opts, data):
         This method should be used as a threading target, start the actual
         minion side execution.
         """
-        minion_instance.gen_modules()
+        functions = minion_instance.gen_modules()
         fn_ = os.path.join(minion_instance.proc_dir, str(data["jid"]))
 
         if opts.get("multiprocessing", True):
@@ -2977,14 +2994,12 @@ def _thread_multi_return(cls, minion_instance, opts, data):
                     ret["success"][function_name] = False
                 try:
                     return_data = minion_instance._execute_job_function(
-                        function_name, function_args, executors, opts, data
+                        function_name, function_args, executors, opts, data, functions
                     )
 
                     key = ind if multifunc_ordered else data["fun"][ind]
                     ret["return"][key] = return_data
-                    retcode = minion_instance.functions.pack["__context__"].get(
-                        "retcode", 0
-                    )
+                    retcode = functions.pack["__context__"].get("retcode", 0)
                     if retcode == 0:
                         # No nonzero retcode in __context__ dunder. Check if return
                         # is a dictionary with a "result" or "success" key.