Файловый менеджер - Редактировать - /snap/core18/2846/usr/lib/python3/dist-packages/cloudinit/config/modules.py
Назад
# Copyright (C) 2008-2022 Canonical Ltd. # Copyright (C) 2012 Hewlett-Packard Development Company, L.P. # # Author: Chuck Short <chuck.short@canonical.com> # Author: Juerg Haefliger <juerg.haefliger@hp.com> # # This file is part of cloud-init. See LICENSE file for license information. import copy from types import ModuleType from typing import Dict, List, NamedTuple, Optional from cloudinit import config, importer from cloudinit import log as logging from cloudinit import type_utils, util from cloudinit.distros import ALL_DISTROS from cloudinit.helpers import ConfigMerger from cloudinit.reporting.events import ReportEventStack from cloudinit.settings import FREQUENCIES from cloudinit.stages import Init LOG = logging.getLogger(__name__) # This prefix is used to make it less # of a chance that when importing # we will not find something else with the same # name in the lookup path... MOD_PREFIX = "cc_" class ModuleDetails(NamedTuple): module: ModuleType name: str frequency: str run_args: List[str] def form_module_name(name): canon_name = name.replace("-", "_") if canon_name.lower().endswith(".py"): canon_name = canon_name[0 : (len(canon_name) - 3)] canon_name = canon_name.strip() if not canon_name: return None if not canon_name.startswith(MOD_PREFIX): canon_name = "%s%s" % (MOD_PREFIX, canon_name) return canon_name def validate_module(mod, name): if ( not hasattr(mod, "meta") or "frequency" not in mod.meta or "distros" not in mod.meta ): raise ValueError( f"Module '{mod}' with name '{name}' MUST have a 'meta' attribute " "of type 'MetaSchema'." ) if mod.meta["frequency"] not in FREQUENCIES: raise ValueError( f"Module '{mod}' with name '{name}' has an invalid frequency " f"{mod.meta['frequency']}." ) if hasattr(mod, "schema"): raise ValueError( f"Module '{mod}' with name '{name}' has a JSON 'schema' attribute " "defined. Please define schema in cloud-init-schema,json." ) def _is_active(module_details: ModuleDetails, cfg: dict) -> bool: activate_by_schema_keys_keys = frozenset( module_details.module.meta.get("activate_by_schema_keys", {}) ) if not activate_by_schema_keys_keys: return True if not activate_by_schema_keys_keys.intersection(cfg.keys()): return False return True class Modules: def __init__(self, init: Init, cfg_files=None, reporter=None): self.init = init self.cfg_files = cfg_files # Created on first use self._cached_cfg: Optional[config.Config] = None if reporter is None: reporter = ReportEventStack( name="module-reporter", description="module-desc", reporting_enabled=False, ) self.reporter = reporter @property def cfg(self) -> config.Config: # None check to avoid empty case causing re-reading if self._cached_cfg is None: merger = ConfigMerger( paths=self.init.paths, datasource=self.init.datasource, additional_fns=self.cfg_files, base_cfg=self.init.cfg, ) self._cached_cfg = merger.cfg # Only give out a copy so that others can't modify this... return copy.deepcopy(self._cached_cfg) def _read_modules(self, name) -> List[Dict]: """Read the modules from the config file given the specified name. Returns a list of module definitions. E.g., [ { "mod": "bootcmd", "freq": "always", "args": "some_arg", } ] Note that in the default case, only "mod" will be set. """ module_list: List[dict] = [] if name not in self.cfg: return module_list cfg_mods = self.cfg.get(name) if not cfg_mods: return module_list for item in cfg_mods: if not item: continue if isinstance(item, str): module_list.append( { "mod": item.strip(), } ) elif isinstance(item, (list)): contents = {} # Meant to fall through... if len(item) >= 1: contents["mod"] = item[0].strip() if len(item) >= 2: contents["freq"] = item[1].strip() if len(item) >= 3: contents["args"] = item[2:] if contents: module_list.append(contents) elif isinstance(item, (dict)): contents = {} valid = False if "name" in item: contents["mod"] = item["name"].strip() valid = True if "frequency" in item: contents["freq"] = item["frequency"].strip() if "args" in item: contents["args"] = item["args"] or [] if contents and valid: module_list.append(contents) else: raise TypeError( "Failed to read '%s' item in config, unknown type %s" % (item, type_utils.obj_name(item)) ) return module_list def _fixup_modules(self, raw_mods) -> List[ModuleDetails]: """Convert list of returned from _read_modules() into new format. Invalid modules and arguments are ignored. Also ensures that the module has the required meta fields. """ mostly_mods = [] for raw_mod in raw_mods: raw_name = raw_mod["mod"] freq = raw_mod.get("freq") run_args = raw_mod.get("args") or [] mod_name = form_module_name(raw_name) if not mod_name: continue if freq and freq not in FREQUENCIES: LOG.warning( "Config specified module %s has an unknown frequency %s", raw_name, freq, ) # Misconfigured in /etc/cloud/cloud.cfg. Reset so cc_* module # default meta attribute "frequency" value is used. freq = None mod_locs, looked_locs = importer.find_module( mod_name, ["", type_utils.obj_name(config)], ["handle"] ) if not mod_locs: LOG.warning( "Could not find module named %s (searched %s)", mod_name, looked_locs, ) continue mod = importer.import_module(mod_locs[0]) validate_module(mod, raw_name) if freq is None: # Use cc_* module default setting since no cloud.cfg overrides freq = mod.meta["frequency"] mostly_mods.append( ModuleDetails( module=mod, name=raw_name, frequency=freq, run_args=run_args, ) ) return mostly_mods def _run_modules(self, mostly_mods: List[ModuleDetails]): cc = self.init.cloudify() # Return which ones ran # and which ones failed + the exception of why it failed failures = [] which_ran = [] for (mod, name, freq, args) in mostly_mods: try: LOG.debug( "Running module %s (%s) with frequency %s", name, mod, freq ) # Use the configs logger and not our own # TODO(harlowja): possibly check the module # for having a LOG attr and just give it back # its own logger? func_args = [name, self.cfg, cc, LOG, args] # Mark it as having started running which_ran.append(name) # This name will affect the semaphore name created run_name = f"config-{name}" desc = "running %s with frequency %s" % (run_name, freq) myrep = ReportEventStack( name=run_name, description=desc, parent=self.reporter ) with myrep: ran, _r = cc.run( run_name, mod.handle, func_args, freq=freq ) if ran: myrep.message = "%s ran successfully" % run_name else: myrep.message = "%s previously ran" % run_name except Exception as e: util.logexc(LOG, "Running module %s (%s) failed", name, mod) failures.append((name, e)) return (which_ran, failures) def run_single(self, mod_name, args=None, freq=None): # Form the users module 'specs' mod_to_be = { "mod": mod_name, "args": args, "freq": freq, } # Now resume doing the normal fixups and running raw_mods = [mod_to_be] mostly_mods = self._fixup_modules(raw_mods) return self._run_modules(mostly_mods) def run_section(self, section_name): """Runs all modules in the given section. section_name - One of the modules lists as defined in /etc/cloud/cloud.cfg. One of: - cloud_init_modules - cloud_config_modules - cloud_final_modules """ raw_mods = self._read_modules(section_name) mostly_mods = self._fixup_modules(raw_mods) distro_name = self.init.distro.name skipped = [] forced = [] overridden = self.cfg.get("unverified_modules", []) inapplicable_mods = [] active_mods = [] for module_details in mostly_mods: (mod, name, _freq, _args) = module_details if mod is None: continue worked_distros = mod.meta["distros"] if not _is_active(module_details, self.cfg): inapplicable_mods.append(name) continue # Skip only when the following conditions are all met: # - distros are defined in the module != ALL_DISTROS # - the current d_name isn't in distros # - and the module is unverified and not in the unverified_modules # override list if worked_distros and worked_distros != [ALL_DISTROS]: if distro_name not in worked_distros: if name not in overridden: skipped.append(name) continue forced.append(name) active_mods.append([mod, name, _freq, _args]) if inapplicable_mods: LOG.info( "Skipping modules '%s' because no applicable config " "is provided.", ",".join(inapplicable_mods), ) if skipped: LOG.info( "Skipping modules '%s' because they are not verified " "on distro '%s'. To run anyway, add them to " "'unverified_modules' in config.", ",".join(skipped), distro_name, ) if forced: LOG.info("running unverified_modules: '%s'", ", ".join(forced)) return self._run_modules(active_mods)
| ver. 1.4 |
Github
|
.
| PHP 7.4.3-4ubuntu2.24 | Генерация страницы: 0 |
proxy
|
phpinfo
|
Настройка