feat: 支持 ansible 沙盒运行 (#12953)

* feat: 支持 ansible 沙盒运行

* feat: 修改 receptor sock 默认路径

* feat: 增加 adhoc 执行命令的 local connection 权限

---------

Co-authored-by: Aaron3S <chenyang@fit2cloud.com>
Co-authored-by: Bai <baijiangjie@gmail.com>

apps/ops/ansible/receptor/receptor_runner.py

@@ -0,0 +1,89 @@
+import concurrent.futures
+import queue
+import socket
+
+import ansible_runner
+from receptorctl import ReceptorControl
+
+receptor_ctl = ReceptorControl('control.sock')
+
+
+def init_receptor_ctl(sock_path):
+    global receptor_ctl
+    receptor_ctl = ReceptorControl(sock_path)
+
+
+def nodes():
+    return receptor_ctl.simple_command("status").get("Advertisements", None)
+
+
+def run(**kwargs):
+    receptor_runner = AnsibleReceptorRunner(**kwargs)
+    return receptor_runner.run()
+
+
+class AnsibleReceptorRunner:
+    def __init__(self, **kwargs):
+        self.runner_params = kwargs
+        self.unit_id = None
+
+    def run(self):
+        input, output = socket.socketpair()
+
+        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
+            transmitter_future = executor.submit(self.transmit, input)
+            result = receptor_ctl.submit_work(payload=output.makefile('rb'),
+                                              node='primary', worktype='ansible-runner')
+            input.close()
+            output.close()
+
+            self.unit_id = result['unitid']
+
+        transmitter_future.result()
+
+        result_file = receptor_ctl.get_work_results(self.unit_id, return_sockfile=True)
+
+        stdout_queue = queue.Queue()
+
+        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
+            processor_future = executor.submit(self.processor, result_file, stdout_queue)
+
+            while not processor_future.done() or \
+                    not stdout_queue.empty():
+                msg = stdout_queue.get()
+                if msg is None:
+                    break
+                print(msg)
+
+        return processor_future.result()
+
+    def transmit(self, _socket):
+        try:
+            ansible_runner.run(
+                streamer='transmit',
+                _output=_socket.makefile('wb'),
+                **self.runner_params
+            )
+        finally:
+            _socket.shutdown(socket.SHUT_WR)
+
+    def processor(self, _result_file, stdout_queue):
+        try:
+            original_handler = self.runner_params.pop("event_handler", None)
+
+            def event_handler(data, **kwargs):
+                stdout = data.get('stdout', '')
+                if stdout:
+                    stdout_queue.put(stdout)
+                if original_handler:
+                    original_handler(data, **kwargs)
+
+            return ansible_runner.interface.run(
+                quite=True,
+                streamer='process',
+                _input=_result_file,
+                event_handler=event_handler,
+                **self.runner_params,
+            )
+        finally:
+            stdout_queue.put(None)

apps/ops/ansible/runner.py

@@ -1,3 +1,4 @@
+import logging
 import os
 import shutil
 import uuid
@@ -5,15 +6,35 @@ import uuid
 import ansible_runner
 from django.conf import settings
 from django.utils._os import safe_join
+from django.utils.functional import LazyObject
 
 from .callback import DefaultCallback
+from .receptor import receptor_runner
 from ..utils import get_ansible_log_verbosity
 
+logger = logging.getLogger(__file__)
+
 
 class CommandInBlackListException(Exception):
     pass
 
 
+class AnsibleWrappedRunner(LazyObject):
+    def _setup(self):
+        self._wrapped = self.get_runner()
+
+    @staticmethod
+    def get_runner():
+        if settings.ANSIBLE_RECEPTOR_ENABLE and settings.ANSIBLE_RECEPTOR_SOCK_PATH:
+            logger.info("Ansible receptor enabled, run ansible task via receptor")
+            receptor_runner.init_receptor_ctl(settings.ANSIBLE_RECEPTOR_SOCK_PATH)
+            return receptor_runner
+        return ansible_runner
+
+
+runner = AnsibleWrappedRunner()
+
+
 class AdHocRunner:
     cmd_modules_choices = ('shell', 'raw', 'command', 'script', 'win_shell')
 
@@ -30,6 +51,8 @@ class AdHocRunner:
         self.extra_vars = extra_vars
         self.dry_run = dry_run
         self.timeout = timeout
+        # enable local connection
+        self.extra_vars.update({"LOCAL_CONNECTION_ENABLED": "1"})
 
     def check_module(self):
         if self.module not in self.cmd_modules_choices:
@@ -48,7 +71,7 @@ class AdHocRunner:
         if os.path.exists(private_env):
             shutil.rmtree(private_env)
 
-        ansible_runner.run(
+        runner.run(
             timeout=self.timeout if self.timeout > 0 else None,
             extravars=self.extra_vars,
             host_pattern=self.pattern,
@@ -81,7 +104,7 @@ class PlaybookRunner:
         if os.path.exists(private_env):
             shutil.rmtree(private_env)
 
-        ansible_runner.run(
+        runner.run(
             private_data_dir=self.project_dir,
             inventory=self.inventory,
             playbook=self.playbook,
@@ -112,7 +135,7 @@ class UploadFileRunner:
 
     def run(self, verbosity=0, **kwargs):
         verbosity = get_ansible_log_verbosity(verbosity)
-        ansible_runner.run(
+        runner.run(
             host_pattern="*",
             inventory=self.inventory,
             module='copy',