"""
Selectors available for plugins. The idea of selectors is to abstract away the process snapshot
and allow plugin authors to submit results without explicitly knowing what the entity id is.
"""
import enum
import operator
import collections
import re
[docs]class EntityType(enum.Enum):
ME_TYPE_UNKNOWN = 0
HOST = 1
PROCESS_GROUP_INSTANCE = 2
DOCKER_CONTAINER_GROUP_INSTANCE = 3
PROCESS_GROUP = 4
# Selectors are used as keys in dictorionaries in result_builder - it's best to force them to be immutable
# deriving from tuple also gives correct __eq__ and __hash__ behaviour
# subclassing from namedtuple would mostly work, except un-parametrized selectors would be equal to empty tuple
# and each other - that's what _markers are for
# for serialization, a full immutable class with overriding __setattr__ and __delatrr_ would probably be needed
MeId = collections.namedtuple(
'MeId',
["metype", "meid"]
)
[docs]class HostSelector(tuple):
"""
Selector binding data with host entity id.
"""
__slots__ = []
_marker = object()
def __new__(cls):
return super().__new__(cls, (cls._marker,))
def select_id(self, **kwargs):
snapshot = kwargs["process_snapshot"]
return EntityType.HOST, snapshot.host_id
def __repr__(self):
return "HostSelector()"
[docs]class FromPluginSelector(tuple):
"""
Selector binding data with entity id associated with plugin.
The resulting entity id depends on the plugin type, e.g. it might be process group instance id if the plugin
is associated with one process group instance, it might be host_id if plugin is associated with host.
Some plugins do not have well defined associated entity id (e.g. plugins activated as Singletons)
"""
__slots__ = []
_marker = object()
def __new__(cls):
return super().__new__(cls, (cls._marker,))
def select_id(self, **kwargs):
plugin_engine = kwargs["plugin_engine"]
return plugin_engine.select_id(**kwargs)
def __repr__(self):
return "FromPluginSelector()"
[docs]class ListenPortSelector(tuple):
"""
Selector binding data with entity id of process listening on a specified port.
For improved performance, snapshot entries of a given type (which is usually known by plugin) can be scanned
for the selected port, before proceeding to all process group instances.
"""
__slots__ = []
port_number = property(operator.itemgetter(1))
process_types = property(operator.itemgetter(2))
me_type = property(operator.itemgetter(3))
technologies = property(operator.itemgetter(4))
name_pattern = property(operator.itemgetter(5))
_marker = object()
def __new__(cls, port_number, process_types=None, me_type=EntityType.PROCESS_GROUP_INSTANCE, technologies=None, name_pattern=None):
if process_types is not None:
process_types = frozenset(process_types)
return super().__new__(cls, (cls._marker, port_number, process_types, me_type, technologies, name_pattern))
def select_id(self, **kwargs):
process_type_snapshot_entries = kwargs["process_snapshot"].entries
if not self.process_types:
reduce1 = process_type_snapshot_entries;
else:
snapshot_by_process_type = kwargs["snapshot_by_process_type"]
reduce1 = [snapshot_entry for process_type in self.process_types
for snapshot_entry in snapshot_by_process_type[process_type]]
if not self.technologies:
reduce2 = reduce1
else:
snapshot_by_process_type = kwargs["snapshot_by_technology"]
reduce2 = [snapshot_entry for techno in self.technologies
for snapshot_entry in snapshot_by_process_type[techno]]
if not self.name_pattern:
reduce3 = reduce2
else:
reduce3 = [snapshot_entry for snapshot_entry in reduce2 if re.search(self.name_pattern, snapshot_entry.group_name)]
ret = self._select_from_snapshot_entries(reduce3)
if ret:
return ret
raise LookupError("could not find PGI associated with a given port")
def _select_from_snapshot_entries(self, snapshot_entries):
processes = [(process, se) for se in snapshot_entries for process in se.processes]
for p, se in processes:
listen_ports = p.properties.get("ListeningPorts", "").split()
if str(self.port_number) in listen_ports:
if (self.me_type == EntityType.PROCESS_GROUP_INSTANCE):
return EntityType.PROCESS_GROUP_INSTANCE, se.group_instance_id
elif (self.me_type == EntityType.PROCESS_GROUP):
return EntityType.PROCESS_GROUP, se.group_id
else:
raise ValueError("Attempting to select entity_type of unsupported EntityType")
def __repr__(self):
return "ListenPortSelector(port_number=%s, process_types=%s, technologies=%s)" % (self.port_number, self.process_types, self.technologies)
[docs]class DockerSelector(tuple):
"""
Selector binding data with entity id of docker container with specified docker_id.
"""
__slots__ = []
docker_id = property(operator.itemgetter(1))
_marker = object()
def __new__(cls, docker_id):
return super().__new__(cls, (cls._marker, docker_id))
def select_id(self, **kwargs):
docker_entity_map = kwargs["docker_entity_map"]
return EntityType.DOCKER_CONTAINER_GROUP_INSTANCE, docker_entity_map[self.docker_id]
def __repr__(self):
return "DockerSelector(docker_id=%s)" % self.docker_id
[docs]class ExplicitPgiSelector(tuple):
"""
Selector binding data with explicitly specified entity id.
"""
__slots__ = []
pgi_id = property(operator.itemgetter(1))
me_type = property(operator.itemgetter(2))
_marker = object()
def __new__(cls, pgi_id, me_type=EntityType.PROCESS_GROUP_INSTANCE):
return super().__new__(cls, (cls._marker, pgi_id, me_type))
def select_id(self, **kwargs):
return self.me_type, self.pgi_id
def __repr__(self):
return "ExplicitIdSelector(me_id=%s, me_type=%s)" % (self.pgi_id, self.me_type)
