"""
Defines the base class for other plugins to use.
"""
from ruxit.api.exceptions import ConfigException
from ruxit.utils.select_helper import check_snapshot_entry
[docs]class BasePlugin:
"""
Base class for plugins.
"""
def __init__(self, **kwargs):
"""
__init__ should not be overridden without a call to super.
Args:
config (Dict[str, str]): plugin configuration
results_builder (ResultsBuilder): instance for this plugin,
json_config: parsed plugin.json contents,
plugin_info (PluginInfo): json data of the plugin exposed in a more convenient way,
associated_entity: monitored entity associated with the plugin, if any (deprecated).
activation_context: activation context for the plugin,
process_snapshot (ProcessSnapshot): latest available at method call,
entity_resolver (EntityResolver): exposes additional methods for querying snapshot
logger: plugin specific logger for plugin development, set to None only in unit tests
"""
self.plugin_info = kwargs.get('plugin_info', None)
self.results_builder = kwargs['results_builder']
self.query_snapshot = kwargs.get('process_snapshot', None)
self.logger = kwargs.get('logger', None)
self.initialize(**kwargs)
[docs] def initialize(self, **kwargs):
"""
Abstract method for custom plugin initialization.
Called after __init__. Override if additional initialization
(e.g.: aquiring connections once instead of on every query method execution) is needed.
Args:
**kwargs: see :class:`~ruxit.api.base_plugin.BasePlugin` for keyword args reference
"""
pass
def _query_internal(self, **kwargs):
self.query_snapshot = kwargs.get('process_snapshot', None)
return self.query(**kwargs)
[docs] def query(self, **kwargs):
"""
Abstract method called to gather measurements. Every plugin should override this method.
Args:
**kwargs: see :class:`~ruxit.api.base_plugin.BasePlugin` for keyword args reference
"""
pass
[docs] def close(self, **kwargs):
"""
Abstract method called when plugin is deactivated.
Args:
**kwargs: see :class:`~ruxit.api.base_plugin.BasePlugin` for keyword args reference
"""
pass
[docs] def find_all_process_groups(self, predicate):
"""
Searches the latest process snapshot for all process groups matching the predicate
Args:
predicate: One parameter function returning bool if snapshot entry meets predicate conditions
Returns:
list of process snapshot entries
"""
return [entry for entry in self.query_snapshot.entries if predicate(entry)]
[docs] def find_single_process_group(self, predicate):
"""
Searches the latest process snapshot for a process group matching the predicate
Args:
predicate: One parameter function returning bool if snapshot entry meets predicate conditions
Returns:
One instance of process snapshot entry
Raises:
ruxit.api.exceptions.ConfigException: if not exactly 1 matching pgi found
"""
ret = [entry for entry in self.query_snapshot.entries if predicate(entry)]
if len(ret) != 1:
raise ConfigException("Expected exactly 1 pgi to match predicate found: %s" % len(ret))
return ret[0]
[docs] def find_all_processes(self, process_predicate):
"""
Searches the latest process snapshot for a process and associated process group that match the criteria passed
in arguments.
Only processes of types associated with plugin are being searched.
Args:
process_predicate: one parameter function which takes process info as input, and returns bool
Returns:
list of tuple(PGI, process) that matched predicate
"""
return self._internal_find_process(process_predicate)
[docs] def find_single_process(self, process_predicate):
"""
Searches the latest process snapshot for a process and associated process group that match the criteria passed
in arguments.
If more than one process matching the criteria is found, exception is raised.
Only processes of types associated with plugin are being searched.
Args:
process_predicate: one parameter function which takes process info as input, and returns bool
Returns:
tuple(PGI, process)
Raises:
ruxit.api.exceptions.ConfigException: if not exactly 1 matching process found
"""
ret = self._internal_find_process(process_predicate)
_ln = len(ret)
if _ln != 1:
raise ConfigException("Expected exactly 1 process matching predicate, found: %s" % _ln)
return ret[0]
def _internal_find_process(self, process_predicate):
assoc_process_types = self.plugin_info.associated_process_types
assoc_technologies = self.plugin_info.associated_technologies
name_pattern = self.plugin_info.name_pattern
def pgi_predicate(entry):
return check_snapshot_entry (entry, assoc_process_types, assoc_technologies, name_pattern)
ret = self._find_process_and_pgi(
self.query_snapshot,
pgi_predicate=pgi_predicate,
process_predicate=process_predicate
)
return ret
def _find_process_and_pgi(self, snapshot, pgi_predicate, process_predicate):
entries = (entry for entry in snapshot.entries if pgi_predicate(entry))
return self._find_process(entries, process_predicate)
def _find_process(self, entries, process_predicate):
return [(pgi, process) for pgi in entries for process in pgi.processes if process_predicate(process)]
