tests/utils/lttngtest/lttngctl.py

   1 #!/usr/bin/env python3
   2 #
   3 # Copyright (C) 2022 Jérémie Galarneau <jeremie.galarneau@efficios.com>
   4 #
   5 # SPDX-License-Identifier: GPL-2.0-only
   6
   7 import abc
   8 import random
   9 import string
  10 import pathlib
  11 import enum
  12 from typing import Optional, Type, Union, List
  13
  14 """
  15 Defines an abstract interface to control LTTng tracing.
  16
  17 The various control concepts are defined by this module. You can use them with a
  18 Controller to interact with a session daemon.
  19
  20 This interface is not comprehensive; it currently provides a subset of the
  21 control functionality that is used by tests.
  22 """
  23
  24
  25 def _generate_random_string(length: int) -> str:
  26     return "".join(
  27         random.choice(string.ascii_lowercase + string.digits) for _ in range(length)
  28     )
  29
  30
  31 class ContextType(abc.ABC):
  32     """Base class representing a tracing context field."""
  33
  34     pass
  35
  36
  37 class VpidContextType(ContextType):
  38     """Application's virtual process id."""
  39
  40     pass
  41
  42
  43 class VuidContextType(ContextType):
  44     """Application's virtual user id."""
  45
  46     pass
  47
  48
  49 class VgidContextType(ContextType):
  50     """Application's virtual group id."""
  51
  52     pass
  53
  54
  55 class JavaApplicationContextType(ContextType):
  56     """A java application-specific context field is a piece of state which the application provides."""
  57
  58     def __init__(self, retriever_name: str, field_name: str):
  59         self._retriever_name: str = retriever_name
  60         self._field_name: str = field_name
  61
  62     @property
  63     def retriever_name(self) -> str:
  64         return self._retriever_name
  65
  66     @property
  67     def field_name(self) -> str:
  68         return self._field_name
  69
  70
  71 @enum.unique
  72 class TracingDomain(enum.Enum):
  73     """Tracing domain."""
  74
  75     User = "User space tracing domain"
  76     Kernel = "Linux kernel tracing domain."
  77     Log4j = "Log4j tracing back-end."
  78     JUL = "Java Util Logging tracing back-end."
  79     Python = "Python logging module tracing back-end."
  80
  81     def __repr__(self):
  82         return "<%s.%s>" % (self.__class__.__name__, self.name)
  83
  84
  85 class EventRule(abc.ABC):
  86     """Event rule base class, see LTTNG-EVENT-RULE(7)."""
  87
  88     pass
  89
  90
  91 class LogLevelRule:
  92     pass
  93
  94
  95 class LogLevelRuleAsSevereAs(LogLevelRule):
  96     def __init__(self, level: int):
  97         self._level = level
  98
  99     @property
 100     def level(self) -> int:
 101         return self._level
 102
 103
 104 class LogLevelRuleExactly(LogLevelRule):
 105     def __init__(self, level: int):
 106         self._level = level
 107
 108     @property
 109     def level(self) -> int:
 110         return self._level
 111
 112
 113 class TracepointEventRule(EventRule):
 114     def __init__(
 115         self,
 116         name_pattern: Optional[str] = None,
 117         filter_expression: Optional[str] = None,
 118         log_level_rule: Optional[LogLevelRule] = None,
 119         name_pattern_exclusions: Optional[List[str]] = None,
 120     ):
 121         self._name_pattern: Optional[str] = name_pattern
 122         self._filter_expression: Optional[str] = filter_expression
 123         self._log_level_rule: Optional[LogLevelRule] = log_level_rule
 124         self._name_pattern_exclusions: Optional[List[str]] = name_pattern_exclusions
 125
 126     @property
 127     def name_pattern(self) -> Optional[str]:
 128         return self._name_pattern
 129
 130     @property
 131     def filter_expression(self) -> Optional[str]:
 132         return self._filter_expression
 133
 134     @property
 135     def log_level_rule(self) -> Optional[LogLevelRule]:
 136         return self._log_level_rule
 137
 138     @property
 139     def name_pattern_exclusions(self) -> Optional[List[str]]:
 140         return self._name_pattern_exclusions
 141
 142
 143 class UserTracepointEventRule(TracepointEventRule):
 144     def __init__(
 145         self,
 146         name_pattern: Optional[str] = None,
 147         filter_expression: Optional[str] = None,
 148         log_level_rule: Optional[LogLevelRule] = None,
 149         name_pattern_exclusions: Optional[List[str]] = None,
 150     ):
 151         TracepointEventRule.__init__(**locals())
 152
 153
 154 class KernelTracepointEventRule(TracepointEventRule):
 155     def __init__(
 156         self,
 157         name_pattern: Optional[str] = None,
 158         filter_expression: Optional[str] = None,
 159         log_level_rule: Optional[LogLevelRule] = None,
 160         name_pattern_exclusions: Optional[List[str]] = None,
 161     ):
 162         TracepointEventRule.__init__(**locals())
 163
 164
 165 class Channel(abc.ABC):
 166     """
 167     A channel is an object which is responsible for a set of ring buffers. It is
 168     associated to a domain and
 169     """
 170
 171     @staticmethod
 172     def _generate_name() -> str:
 173         return "channel_{random_id}".format(random_id=_generate_random_string(8))
 174
 175     @abc.abstractmethod
 176     def add_context(self, context_type: ContextType) -> None:
 177         pass
 178
 179     @property
 180     @abc.abstractmethod
 181     def domain(self) -> TracingDomain:
 182         pass
 183
 184     @property
 185     @abc.abstractmethod
 186     def name(self) -> str:
 187         pass
 188
 189     @abc.abstractmethod
 190     def add_recording_rule(self, rule: Type[EventRule]) -> None:
 191         pass
 192
 193
 194 class SessionOutputLocation(abc.ABC):
 195     pass
 196
 197
 198 class LocalSessionOutputLocation(SessionOutputLocation):
 199     def __init__(self, trace_path: pathlib.Path):
 200         self._path = trace_path
 201
 202     @property
 203     def path(self) -> pathlib.Path:
 204         return self._path
 205
 206
 207 class ProcessAttributeTracker(abc.ABC):
 208     """
 209     Process attribute tracker used to filter before the evaluation of event
 210     rules.
 211
 212     Note that this interface is currently limited as it doesn't allow changing
 213     the tracking policy. For instance, it is not possible to set the tracking
 214     policy back to "all" once it has transitioned to "include set".
 215     """
 216
 217     @enum.unique
 218     class TrackingPolicy(enum.Enum):
 219         INCLUDE_ALL = """
 220             Track all possible process attribute value of a given type (i.e. no filtering).
 221             This is the default state of a process attribute tracker.
 222             """
 223         EXCLUDE_ALL = "Exclude all possible process attribute values of a given type."
 224         INCLUDE_SET = "Track a set of specific process attribute values."
 225
 226         def __repr__(self):
 227             return "<%s.%s>" % (self.__class__.__name__, self.name)
 228
 229     def __init__(self, policy: TrackingPolicy):
 230         self._policy = policy
 231
 232     @property
 233     def tracking_policy(self) -> TrackingPolicy:
 234         return self._policy
 235
 236
 237 class ProcessIDProcessAttributeTracker(ProcessAttributeTracker):
 238     @abc.abstractmethod
 239     def track(self, pid: int) -> None:
 240         pass
 241
 242     @abc.abstractmethod
 243     def untrack(self, pid: int) -> None:
 244         pass
 245
 246
 247 class VirtualProcessIDProcessAttributeTracker(ProcessAttributeTracker):
 248     @abc.abstractmethod
 249     def track(self, vpid: int) -> None:
 250         pass
 251
 252     @abc.abstractmethod
 253     def untrack(self, vpid: int) -> None:
 254         pass
 255
 256
 257 class UserIDProcessAttributeTracker(ProcessAttributeTracker):
 258     @abc.abstractmethod
 259     def track(self, uid: Union[int, str]) -> None:
 260         pass
 261
 262     @abc.abstractmethod
 263     def untrack(self, uid: Union[int, str]) -> None:
 264         pass
 265
 266
 267 class VirtualUserIDProcessAttributeTracker(ProcessAttributeTracker):
 268     @abc.abstractmethod
 269     def track(self, vuid: Union[int, str]) -> None:
 270         pass
 271
 272     @abc.abstractmethod
 273     def untrack(self, vuid: Union[int, str]) -> None:
 274         pass
 275
 276
 277 class GroupIDProcessAttributeTracker(ProcessAttributeTracker):
 278     @abc.abstractmethod
 279     def track(self, gid: Union[int, str]) -> None:
 280         pass
 281
 282     @abc.abstractmethod
 283     def untrack(self, gid: Union[int, str]) -> None:
 284         pass
 285
 286
 287 class VirtualGroupIDProcessAttributeTracker(ProcessAttributeTracker):
 288     @abc.abstractmethod
 289     def track(self, vgid: Union[int, str]) -> None:
 290         pass
 291
 292     @abc.abstractmethod
 293     def untrack(self, vgid: Union[int, str]) -> None:
 294         pass
 295
 296
 297 class Session(abc.ABC):
 298     @staticmethod
 299     def _generate_name() -> str:
 300         return "session_{random_id}".format(random_id=_generate_random_string(8))
 301
 302     @property
 303     @abc.abstractmethod
 304     def name(self) -> str:
 305         pass
 306
 307     @property
 308     @abc.abstractmethod
 309     def output(self) -> Optional[Type[SessionOutputLocation]]:
 310         pass
 311
 312     @abc.abstractmethod
 313     def add_channel(
 314         self, domain: TracingDomain, channel_name: Optional[str] = None
 315     ) -> Channel:
 316         """Add a channel with default attributes to the session."""
 317         pass
 318
 319     @abc.abstractmethod
 320     def start(self) -> None:
 321         pass
 322
 323     @abc.abstractmethod
 324     def stop(self) -> None:
 325         pass
 326
 327     @abc.abstractmethod
 328     def destroy(self) -> None:
 329         pass
 330
 331     @abc.abstractproperty
 332     def kernel_pid_process_attribute_tracker(
 333         self,
 334     ) -> Type[ProcessIDProcessAttributeTracker]:
 335         raise NotImplementedError
 336
 337     @abc.abstractproperty
 338     def kernel_vpid_process_attribute_tracker(
 339         self,
 340     ) -> Type[VirtualProcessIDProcessAttributeTracker]:
 341         raise NotImplementedError
 342
 343     @abc.abstractproperty
 344     def user_vpid_process_attribute_tracker(
 345         self,
 346     ) -> Type[VirtualProcessIDProcessAttributeTracker]:
 347         raise NotImplementedError
 348
 349     @abc.abstractproperty
 350     def kernel_gid_process_attribute_tracker(
 351         self,
 352     ) -> Type[GroupIDProcessAttributeTracker]:
 353         raise NotImplementedError
 354
 355     @abc.abstractproperty
 356     def kernel_vgid_process_attribute_tracker(
 357         self,
 358     ) -> Type[VirtualGroupIDProcessAttributeTracker]:
 359         raise NotImplementedError
 360
 361     @abc.abstractproperty
 362     def user_vgid_process_attribute_tracker(
 363         self,
 364     ) -> Type[VirtualGroupIDProcessAttributeTracker]:
 365         raise NotImplementedError
 366
 367     @abc.abstractproperty
 368     def kernel_uid_process_attribute_tracker(
 369         self,
 370     ) -> Type[UserIDProcessAttributeTracker]:
 371         raise NotImplementedError
 372
 373     @abc.abstractproperty
 374     def kernel_vuid_process_attribute_tracker(
 375         self,
 376     ) -> Type[VirtualUserIDProcessAttributeTracker]:
 377         raise NotImplementedError
 378
 379     @abc.abstractproperty
 380     def user_vuid_process_attribute_tracker(
 381         self,
 382     ) -> Type[VirtualUserIDProcessAttributeTracker]:
 383         raise NotImplementedError
 384
 385
 386 class ControlException(RuntimeError):
 387     """Base type for exceptions thrown by a controller."""
 388
 389     def __init__(self, msg: str):
 390         super().__init__(msg)
 391
 392
 393 class Controller(abc.ABC):
 394     """
 395     Interface of a top-level control interface. A control interface can be, for
 396     example, the LTTng client or a wrapper around liblttng-ctl. It is used to
 397     create and manage top-level objects of a session daemon instance.
 398     """
 399
 400     @abc.abstractmethod
 401     def create_session(
 402         self, name: Optional[str] = None, output: Optional[SessionOutputLocation] = None
 403     ) -> Session:
 404         """
 405         Create a session with an output. Don't specify an output
 406         to create a session without an output.
 407         """
 408         pass