1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
|
# 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 inspect import signature
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
)
# 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
)
func_args = {
"name": name,
"cfg": self.cfg,
"cloud": cc,
"args": args,
}
with myrep:
func_signature = signature(mod.handle)
func_params = func_signature.parameters
if len(func_params) == 5:
util.deprecate(
deprecated="Config modules with a `log` parameter",
deprecated_version="23.2",
)
func_args.update({"log": LOG})
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)
|