perf: 重构 playbook base runner

apps/assets/automations/base/manager.py

@@ -1,7 +1,6 @@
 import os
 import shutil
 import yaml
-from copy import deepcopy
 from collections import defaultdict
 
 from django.conf import settings
@@ -42,33 +41,23 @@ class BasePlaybookManager:
     def method_type(cls):
         raise NotImplementedError
 
+    def get_assets_group_by_platform(self):
+        return self.automation.all_assets_group_by_platform()
+
     @property
     def runtime_dir(self):
         ansible_dir = settings.ANSIBLE_DIR
+        dir_name = '{}_{}'.format(self.automation.name.replace(' ', '_'), self.execution.id)
         path = os.path.join(
-            ansible_dir, self.automation.type,
+            ansible_dir, 'automations', self.automation.type,
-            self.automation.name.replace(' ', '_'),
+            dir_name, timezone.now().strftime('%Y%m%d_%H%M%S')
-            timezone.now().strftime('%Y%m%d_%H%M%S')
         )
+        if not os.path.exists(path):
+            os.makedirs(path, exist_ok=True, mode=0o755)
         return path
 
-    @property
-    def inventory_path(self):
-        return os.path.join(self.runtime_dir, 'inventory', 'hosts.json')
-
-    @property
-    def playbook_path(self):
-        return os.path.join(self.runtime_dir, 'project', 'main.yml')
-
-    def generate(self):
-        self.prepare_playbook_dir()
-        self.generate_inventory()
-        self.generate_playbook()
-
     def prepare_playbook_dir(self):
-        inventory_dir = os.path.dirname(self.inventory_path)
+        for d in [self.runtime_dir]:
-        playbook_dir = os.path.dirname(self.playbook_path)
-        for d in [inventory_dir, playbook_dir]:
             if not os.path.exists(d):
                 os.makedirs(d, exist_ok=True, mode=0o755)
 
@@ -82,85 +71,70 @@ class BasePlaybookManager:
             getattr(automation, method_attr) in self.method_id_meta_mapper
 
         if not method_enabled:
-            host['error'] = _('Change password disabled')
+            host['error'] = _('{} disabled'.format(self.__class__.method_type()))
             return host
-
-        self.method_hosts_mapper[getattr(automation, method_attr)].append(host['name'])
         return host
 
-    def generate_inventory(self):
+    def generate_inventory(self, platformed_assets, inventory_path):
         inventory = JMSInventory(
-            assets=self.automation.get_all_assets(),
+            assets=platformed_assets,
             account_policy=self.ansible_account_policy,
             host_callback=self.host_callback
         )
-        inventory.write_to_file(self.inventory_path)
+        inventory.write_to_file(inventory_path)
-        logger.debug("Generate inventory done: {}".format(self.inventory_path))
 
-    def generate_playbook(self):
+    def generate_playbook(self, platformed_assets, platform, sub_playbook_dir):
-        main_playbook = []
+        method_id = getattr(platform.automation, '{}_method'.format(self.__class__.method_type()))
-        for method_id, host_names in self.method_hosts_mapper.items():
+        method = self.method_id_meta_mapper.get(method_id)
-            method = self.method_id_meta_mapper.get(method_id)
+        if not method:
-            if not method:
+            logger.error("Method not found: {}".format(method_id))
-                logger.error("Method not found: {}".format(method_id))
+            return method
-                continue
+        method_playbook_dir_path = method['dir']
-            method_playbook_dir_path = method['dir']
+        sub_playbook_path = os.path.join(sub_playbook_dir, 'project', 'main.yml')
-            method_playbook_dir_name = os.path.basename(method_playbook_dir_path)
+        shutil.copytree(method_playbook_dir_path, os.path.dirname(sub_playbook_path))
-            sub_playbook_dir = os.path.join(os.path.dirname(self.playbook_path), method_playbook_dir_name)
-            sub_playbook_path = os.path.join(sub_playbook_dir, 'main.yml')
-            shutil.copytree(method_playbook_dir_path, sub_playbook_dir)
 
-            with open(sub_playbook_path, 'r') as f:
+        with open(sub_playbook_path, 'r') as f:
-                host_playbook_play = yaml.safe_load(f)
+            plays = yaml.safe_load(f)
+        for play in plays:
+            play['hosts'] = 'all'
 
-            if isinstance(host_playbook_play, list):
+        with open(sub_playbook_path, 'w') as f:
-                host_playbook_play = host_playbook_play[0]
+            yaml.safe_dump(plays, f)
-
+        return sub_playbook_path
-            hosts_bulked = [host_names[i:i+self.bulk_size] for i in range(0, len(host_names), self.bulk_size)]
-            for i, hosts in enumerate(hosts_bulked):
-                plays = []
-                play = deepcopy(host_playbook_play)
-                play['hosts'] = ':'.join(hosts)
-                plays.append(play)
-
-                playbook_path = os.path.join(sub_playbook_dir, 'part_{}.yml'.format(i))
-                with open(playbook_path, 'w') as f:
-                    yaml.safe_dump(plays, f)
-                self.playbooks.append([playbook_path, hosts])
-
-                main_playbook.append({
-                    'name': method['name'] + ' for part {}'.format(i),
-                    'import_playbook': os.path.join(method_playbook_dir_name, 'part_{}.yml'.format(i))
-                })
-
-        with open(self.playbook_path, 'w') as f:
-            yaml.safe_dump(main_playbook, f)
-
-        logger.debug("Generate playbook done: " + self.playbook_path)
 
     def get_runners(self):
         runners = []
-        for playbook_path in self.playbooks:
+        for platform, assets in self.get_assets_group_by_platform().items():
-            runer = PlaybookRunner(
+            assets_bulked = [assets[i:i+self.bulk_size] for i in range(0, len(assets), self.bulk_size)]
-                self.inventory_path,
+
-                playbook_path,
+            for i, _assets in enumerate(assets_bulked, start=1):
-                self.runtime_dir,
+                sub_dir = '{}_{}'.format(platform.name, i)
-                callback=PlaybookCallback(),
+                playbook_dir = os.path.join(self.runtime_dir, sub_dir)
-            )
+                inventory_path = os.path.join(self.runtime_dir, sub_dir, 'hosts.json')
-            runners.append(runer)
+                self.generate_inventory(_assets, inventory_path)
+                playbook_path = self.generate_playbook(_assets, platform, playbook_dir)
+
+                runer = PlaybookRunner(
+                    inventory_path,
+                    playbook_path,
+                    self.runtime_dir,
+                    callback=PlaybookCallback(),
+                )
+                runners.append(runer)
         return runners
 
-    def on_runner_done(self, runner, cb):
+    def on_runner_success(self, runner, cb):
         raise NotImplementedError
 
     def on_runner_failed(self, runner, e):
         print("Runner failed: {} {}".format(e, self))
 
     def before_runner_start(self, runner):
-        pass
+        print("Start run task: ")
+        print("  inventory: {}".format(runner.inventory))
+        print("  playbook: {}".format(runner.playbook))
 
     def run(self, **kwargs):
-        self.generate()
         runners = self.get_runners()
         if len(runners) > 1:
             print("### 分批次执行开始任务, 总共 {}\n".format(len(runners)))
@@ -173,7 +147,7 @@ class BasePlaybookManager:
             self.before_runner_start(runner)
             try:
                 cb = runner.run(**kwargs)
-                self.on_runner_done(runner, cb)
+                self.on_runner_success(runner, cb)
             except Exception as e:
                 self.on_runner_failed(runner, e)
-            print('\n\n')
+            print('\n')

apps/assets/automations/change_secret/manager.py

@@ -121,8 +121,10 @@ class ChangeSecretManager(BasePlaybookManager):
         ChangeSecretRecord.objects.bulk_create(records)
         return inventory_hosts
 
-    def on_runner_done(self, runner, cb):
+    def on_runner_success(self, runner, cb):
-        summary = runner.summary
+        summary = cb.summary
+        print("Summary: ")
+        print(summary)
 
     def on_runner_failed(self, runner, e):
         pass