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
|
# Copyright (c) Meta Platforms, Inc. and affiliates.
# SPDX-License-Identifier: LGPL-2.1-or-later
import fnmatch
from importlib import import_module
import logging
import os
import runpy
import sys
from types import SimpleNamespace
from typing import Any, Callable, Dict, List, Tuple
logger = logging.getLogger("drgn.plugins")
_plugins = None
_hooks: Dict[str, List[Tuple[str, Callable[..., Any]]]] = {}
def _load_plugins() -> List[Tuple[str, object]]:
plugins = []
# Mapping from plugin name requested with DRGN_PLUGINS to whether we found
# an entry point with that name.
enabled_entry_points = {}
env = os.getenv("DRGN_PLUGINS")
if env:
for item in env.split(","):
if not item:
# Ignore empty items for convenience.
continue
name, sep, value = item.partition("=")
if sep:
try:
if "/" in value:
plugin: object = SimpleNamespace(**runpy.run_path(value))
else:
plugin = import_module(value)
except Exception:
logger.warning("failed to load %r:", item, exc_info=True)
else:
plugins.append((name, plugin))
logger.debug("loaded %r", item)
else:
enabled_entry_points[name] = False
env = os.getenv("DRGN_DISABLE_PLUGINS")
# If all plugins are disabled, avoid the entry point machinery entirely.
if env != "*" or enabled_entry_points:
disable_plugins = env.split(",") if env else []
import importlib.metadata
group = "drgn.plugins"
if sys.version_info >= (3, 10):
entry_points = importlib.metadata.entry_points(group=group)
else:
entry_points = importlib.metadata.entry_points().get(group, ())
for entry_point in entry_points:
if entry_point.name in enabled_entry_points:
enabled_entry_points[entry_point.name] = True
elif any(
fnmatch.fnmatch(entry_point.name, disable)
for disable in disable_plugins
):
continue
try:
plugin = entry_point.load()
except Exception:
logger.warning(
"failed to load %r:",
f"{entry_point.name} = {entry_point.value}",
exc_info=True,
)
else:
plugins.append((entry_point.name, plugin))
logger.debug(
"loaded entry point %r",
f"{entry_point.name} = {entry_point.value}",
)
missing_entry_points = [
key for key, value in enabled_entry_points.items() if not value
]
if missing_entry_points:
missing_entry_points.sort()
logger.warning(
"not found: %s",
", ".join([repr(name) for name in missing_entry_points]),
)
return plugins
def _load_hook(hook_name: str) -> List[Tuple[str, Callable[..., Any]]]:
global _plugins
if _plugins is None:
_plugins = _load_plugins()
hooks = []
for name, plugin in _plugins:
try:
hook = getattr(plugin, hook_name)
except AttributeError:
continue
hooks.append((name, hook))
hooks.sort(key=lambda hook: (getattr(hook[1], "drgn_priority", 50), hook[0]))
return hooks
def call_plugins(hook_name: str, *args: object) -> None:
try:
hooks = _hooks[hook_name]
except KeyError:
_hooks[hook_name] = hooks = _load_hook(hook_name)
for name, hook in hooks:
try:
hook(*args)
except Exception:
logger.warning("%r %s failed:", name, hook_name, exc_info=True)
|
