Skip to content

deep.api.plugin

Load and handle plugins.

DEEP_PLUGINS = ['deep.api.plugin.otel.OTelPlugin', 'deep.api.plugin.python.PythonPlugin', 'deep.api.plugin.metric.prometheus_metrics.PrometheusPlugin', 'deep.api.plugin.metric.otel_metrics.OTelMetrics'] module-attribute

System provided default plugins.

DidNotEnable

Bases: Exception

Raised when failed to load plugin.

The plugin could not be enabled due to a trivial user error like otel not being installed for the OTelPlugin.

Source code in deep/api/plugin/__init__.py 
183
184
185
186
187
188
189
class DidNotEnable(Exception):
    """
    Raised when failed to load plugin.

    The plugin could not be enabled due to a trivial user error like
    `otel` not being installed for the `OTelPlugin`.
    """

Plugin

Bases: ABC

A deep Plugin.

This type defines a plugin for deep, these plugins allow for extensions to how deep decorates data.

Source code in deep/api/plugin/__init__.py 
 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
class Plugin(abc.ABC):
    """
    A deep Plugin.

    This type defines a plugin for deep, these plugins allow for extensions to how deep decorates data.
    """

    def __init__(self, name: str = None, config: 'ConfigService' = None):
        """
        Create a new plugin.

        :param name: the name of the plugin (default to class name)
        :param config: the deep config service
        """
        super(Plugin, self).__init__()
        self.config = config
        if name is None:
            self._name = self.__class__.__name__
        else:
            self._name = name

    @property
    def name(self):
        """The name of the plugin."""
        return self._name

    def is_active(self) -> bool:
        """
        Is the plugin active.

        Check the value of the config element plugin_{name}. If it is set to
        'False' this plugin is not active.
        """
        attr = getattr(self.config, f'plugin_{self.name}'.upper(), 'True')
        if attr is None:
            return True
        return str2bool(attr)

    def shutdown(self):
        """Clean up and shutdown the plugin."""
        pass

    def order(self) -> int:
        """
        Order of precedence when multiple versions of providers are available.

        Order=1 will run after a provider with order=0.

        :return: the provider order
        """
        return 0

name property

The name of the plugin.

__init__(name=None, config=None)

Create a new plugin.

:param name: the name of the plugin (default to class name) :param config: the deep config service

Source code in deep/api/plugin/__init__.py 
85
86
87
88
89
90
91
92
93
94
95
96
97
def __init__(self, name: str = None, config: 'ConfigService' = None):
    """
    Create a new plugin.

    :param name: the name of the plugin (default to class name)
    :param config: the deep config service
    """
    super(Plugin, self).__init__()
    self.config = config
    if name is None:
        self._name = self.__class__.__name__
    else:
        self._name = name

is_active()

Is the plugin active.

Check the value of the config element plugin_{name}. If it is set to 'False' this plugin is not active.

Source code in deep/api/plugin/__init__.py 
104
105
106
107
108
109
110
111
112
113
114
def is_active(self) -> bool:
    """
    Is the plugin active.

    Check the value of the config element plugin_{name}. If it is set to
    'False' this plugin is not active.
    """
    attr = getattr(self.config, f'plugin_{self.name}'.upper(), 'True')
    if attr is None:
        return True
    return str2bool(attr)

order()

Order of precedence when multiple versions of providers are available.

Order=1 will run after a provider with order=0.

:return: the provider order

Source code in deep/api/plugin/__init__.py 
120
121
122
123
124
125
126
127
128
def order(self) -> int:
    """
    Order of precedence when multiple versions of providers are available.

    Order=1 will run after a provider with order=0.

    :return: the provider order
    """
    return 0

shutdown()

Clean up and shutdown the plugin.

Source code in deep/api/plugin/__init__.py 
116
117
118
def shutdown(self):
    """Clean up and shutdown the plugin."""
    pass

ResourceProvider

Bases: Plugin, ABC

Implement this to have the plugin provide resource attributes to Deep.

Source code in deep/api/plugin/__init__.py 
134
135
136
137
138
139
140
141
142
143
144
class ResourceProvider(Plugin, abc.ABC):
    """Implement this to have the plugin provide resource attributes to Deep."""

    @abc.abstractmethod
    def resource(self) -> Optional[Resource]:
        """
        Provide resource.

        :return: the provided resource
        """
        pass

resource() abstractmethod

Provide resource.

:return: the provided resource

Source code in deep/api/plugin/__init__.py 
137
138
139
140
141
142
143
144
@abc.abstractmethod
def resource(self) -> Optional[Resource]:
    """
    Provide resource.

    :return: the provided resource
    """
    pass

SnapshotDecorator

Bases: Plugin, ABC

Implement this to decorate collected snapshots with attributes.

Source code in deep/api/plugin/__init__.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
class SnapshotDecorator(Plugin, abc.ABC):
    """Implement this to decorate collected snapshots with attributes."""

    @abc.abstractmethod
    def decorate(self, snapshot_id: str, context: ActionContext) -> Optional[BoundedAttributes]:
        """
        Decorate a snapshot with additional data.

        :param snapshot_id: the id of the collected snapshot
        :param context: the action context for this action

        :return: the additional attributes to attach
        """
        pass

decorate(snapshot_id, context) abstractmethod

Decorate a snapshot with additional data.

:param snapshot_id: the id of the collected snapshot :param context: the action context for this action

:return: the additional attributes to attach

Source code in deep/api/plugin/__init__.py 
150
151
152
153
154
155
156
157
158
159
160
@abc.abstractmethod
def decorate(self, snapshot_id: str, context: ActionContext) -> Optional[BoundedAttributes]:
    """
    Decorate a snapshot with additional data.

    :param snapshot_id: the id of the collected snapshot
    :param context: the action context for this action

    :return: the additional attributes to attach
    """
    pass

TracepointLogger

Bases: Plugin, ABC

This defines how a tracepoint logger should interact with Deep.

This can be registered with Deep to provide customization to the way Deep will log dynamic log messages injected via tracepoints.

Source code in deep/api/plugin/__init__.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
class TracepointLogger(Plugin, abc.ABC):
    """
    This defines how a tracepoint logger should interact with Deep.

    This can be registered with Deep to provide customization to the way Deep will log dynamic log
    messages injected via tracepoints.
    """

    @abc.abstractmethod
    def log_tracepoint(self, log_msg: str, tp_id: str, ctx_id: str):
        """
        Log the dynamic log message.

        :param (str) log_msg: the log message to log
        :param (str) tp_id:  the id of the tracepoint that generated this log
        :param (str) ctx_id: the id of the context that was created by this tracepoint
        """
        pass

log_tracepoint(log_msg, tp_id, ctx_id) abstractmethod

Log the dynamic log message.

:param (str) log_msg: the log message to log :param (str) tp_id: the id of the tracepoint that generated this log :param (str) ctx_id: the id of the context that was created by this tracepoint

Source code in deep/api/plugin/__init__.py 
171
172
173
174
175
176
177
178
179
180
@abc.abstractmethod
def log_tracepoint(self, log_msg: str, tp_id: str, ctx_id: str):
    """
    Log the dynamic log message.

    :param (str) log_msg: the log message to log
    :param (str) tp_id:  the id of the tracepoint that generated this log
    :param (str) ctx_id: the id of the context that was created by this tracepoint
    """
    pass

load_plugins(config, custom=None)

Load all the deep plugins.

Attempt to load each plugin, if successful merge an attributes list of each plugin.

:return: the loaded plugins and attributes.

Source code in deep/api/plugin/__init__.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def load_plugins(config: 'ConfigService', custom=None) -> List['Plugin']:
    """
    Load all the deep plugins.

    Attempt to load each plugin, if successful merge an attributes list of each plugin.

    :return: the loaded plugins and attributes.
    """
    if custom is None:
        custom = []
    loaded = []
    for plugin in __plugin_generator(DEEP_PLUGINS + custom):
        try:
            plugin_instance = plugin(config=config)
            if not plugin_instance.is_active():
                logging.debug("Plugin %s is not active.", plugin_instance.name)
                continue
            loaded.append(plugin_instance)
        except Exception as e:
            logging.debug("Could not load plugin %s: %s", plugin, e)

    loaded.sort(key=lambda pl: pl.order() or 0)
    return loaded