Method _thread_return

salt/minion.py:2843–3373

This method should be used as a threading target, start the actual minion side execution.

(cls, minion_instance, opts, data)

    @classmethod
    def _thread_return(cls, minion_instance, opts, data):
        """
        This method should be used as a threading target, start the actual
        minion side execution.
        """
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)

        salt.utils.tracing.configure(opts)
        salt.utils.metrics.configure({**opts, "__role": "minion"})
        _exec_trace_ctx = (
            salt.utils.tracing.extract(data) if isinstance(data, dict) else None
        )
        _exec_fun = data.get("fun", "") if isinstance(data, dict) else ""
        _exec_span_cm = salt.utils.tracing.start_span(
            f"salt.minion.exec.{_exec_fun}" if _exec_fun else "salt.minion.exec",
            attributes={
                "salt.fun": _exec_fun,
                "salt.jid": str(data.get("jid", "")) if isinstance(data, dict) else "",
            },
            context=_exec_trace_ctx,
        )
        # The span needs to outlive the existing try/finally below, which
        # makes a plain ``with`` block awkward; enter / exit manually.
        _exec_span_cm.__enter__()  # pylint: disable=unnecessary-dunder-call
        _exec_span_exit_called = False
        # Companion histogram for the same wall-clock window the exec span
        # covers; recorded in the finally below.
        _exec_perf_start = time.perf_counter()

        minion_instance.gen_modules()

        fn_ = os.path.join(minion_instance.proc_dir, str(data["jid"]))

        try:
            if opts.get("multiprocessing", True):
                salt.utils.process.appendproctitle(f"{cls.__name__}._thread_return")

            sdata = {"pid": os.getpid()}
            sdata.update(data)
            log.info("Starting a new job %s with PID %s", data["jid"], sdata["pid"])
            with salt.utils.files.fopen(fn_, "w+b") as fp_:
                fp_.write(salt.payload.dumps(sdata))
            if data.get("start_event"):
                minion_instance._fire_start_event(data)
            ret = {"success": False}
            function_name = data["fun"]
            function_args = data["arg"]
            executors = (
                data.get("module_executors")
                or getattr(minion_instance, "module_executors", [])
                or opts.get("module_executors", ["direct_call"])
            )
            allow_missing_funcs = any(
                [
                    minion_instance.executors[f"{executor}.allow_missing_func"](
                        function_name
                    )