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 class TracingDomain(enum.Enum):
  72     """Tracing domain."""
  73
  74     User = enum.auto(), "User space tracing domain"
  75     Kernel = enum.auto(), "Linux kernel tracing domain."
  76     Log4j = enum.auto(), "Log4j tracing back-end."
  77     JUL = enum.auto(), "Java Util Logging tracing back-end."
  78     Python = enum.auto(), "Python logging module tracing back-end."
  79
  80
  81 class EventRule(abc.ABC):
  82     """Event rule base class, see LTTNG-EVENT-RULE(7)."""
  83
  84     pass
  85
  86
  87 class LogLevelRule:
  88     pass
  89
  90
  91 class LogLevelRuleAsSevereAs(LogLevelRule):
  92     def __init__(self, level: int):
  93         self._level = level
  94
  95     @property
  96     def level(self) -> int:
  97         return self._level
  98
  99
 100 class LogLevelRuleExactly(LogLevelRule):
 101     def __init__(self, level: int):
 102         self._level = level
 103
 104     @property
 105     def level(self) -> int:
 106         return self._level
 107
 108
 109 class TracepointEventRule(EventRule):
 110     def __init__(
 111         self,
 112         name_pattern: Optional[str] = None,
 113         filter_expression: Optional[str] = None,
 114         log_level_rule: Optional[LogLevelRule] = None,
 115         name_pattern_exclusions: Optional[List[str]] = None,
 116     ):
 117         self._name_pattern: Optional[str] = name_pattern
 118         self._filter_expression: Optional[str] = filter_expression
 119         self._log_level_rule: Optional[LogLevelRule] = log_level_rule
 120         self._name_pattern_exclusions: Optional[List[str]] = name_pattern_exclusions
 121
 122     @property
 123     def name_pattern(self) -> Optional[str]:
 124         return self._name_pattern
 125
 126     @property
 127     def filter_expression(self) -> Optional[str]:
 128         return self._filter_expression
 129
 130     @property
 131     def log_level_rule(self) -> Optional[LogLevelRule]:
 132         return self._log_level_rule
 133
 134     @property
 135     def name_pattern_exclusions(self) -> Optional[List[str]]:
 136         return self._name_pattern_exclusions
 137
 138
 139 class UserTracepointEventRule(TracepointEventRule):
 140     def __init__(
 141         self,
 142         name_pattern: Optional[str] = None,
 143         filter_expression: Optional[str] = None,
 144         log_level_rule: Optional[LogLevelRule] = None,
 145         name_pattern_exclusions: Optional[List[str]] = None,
 146     ):
 147         TracepointEventRule.__init__(**locals())
 148
 149
 150 class KernelTracepointEventRule(TracepointEventRule):
 151     def __init__(
 152         self,
 153         name_pattern: Optional[str] = None,
 154         filter_expression: Optional[str] = None,
 155         log_level_rule: Optional[LogLevelRule] = None,
 156         name_pattern_exclusions: Optional[List[str]] = None,
 157     ):
 158         TracepointEventRule.__init__(**locals())
 159
 160
 161 class Channel(abc.ABC):
 162     """
 163     A channel is an object which is responsible for a set of ring buffers. It is
 164     associated to a domain and
 165     """
 166
 167     @staticmethod
 168     def _generate_name() -> str:
 169         return "channel_{random_id}".format(random_id=_generate_random_string(8))
 170
 171     @abc.abstractmethod
 172     def add_context(self, context_type: ContextType) -> None:
 173         pass
 174
 175     @property
 176     @abc.abstractmethod
 177     def domain(self) -> TracingDomain:
 178         pass
 179
 180     @property
 181     @abc.abstractmethod
 182     def name(self) -> str:
 183         pass
 184
 185     @abc.abstractmethod
 186     def add_recording_rule(self, rule: Type[EventRule]) -> None:
 187         pass
 188
 189
 190 class SessionOutputLocation(abc.ABC):
 191     pass
 192
 193
 194 class LocalSessionOutputLocation(SessionOutputLocation):
 195     def __init__(self, trace_path: pathlib.Path):
 196         self._path = trace_path
 197
 198     @property
 199     def path(self) -> pathlib.Path:
 200         return self._path
 201
 202
 203 class ProcessAttributeTracker(abc.ABC):
 204     """
 205     Process attribute tracker used to filter before the evaluation of event
 206     rules.
 207
 208     Note that this interface is currently limited as it doesn't allow changing
 209     the tracking policy. For instance, it is not possible to set the tracking
 210     policy back to "all" once it has transitioned to "include set".
 211     """
 212
 213     class TrackingPolicy(enum.Enum):
 214         INCLUDE_ALL = (
 215             enum.auto(),
 216             """
 217             Track all possible process attribute value of a given type (i.e. no filtering).
 218             This is the default state of a process attribute tracker.
 219             """,
 220         )
 221         EXCLUDE_ALL = (
 222             enum.auto(),
 223             "Exclude all possible process attribute values of a given type.",
 224         )
 225         INCLUDE_SET = enum.auto(), "Track a set of specific process attribute values."
 226
 227     def __init__(self, policy: TrackingPolicy):
 228         self._policy = policy
 229
 230     @property
 231     def tracking_policy(self) -> TrackingPolicy:
 232         return self._policy
 233
 234
 235 class ProcessIDProcessAttributeTracker(ProcessAttributeTracker):
 236     @abc.abstractmethod
 237     def track(self, pid: int) -> None:
 238         pass
 239
 240     @abc.abstractmethod
 241     def untrack(self, pid: int) -> None:
 242         pass
 243
 244
 245 class VirtualProcessIDProcessAttributeTracker(ProcessAttributeTracker):
 246     @abc.abstractmethod
 247     def track(self, vpid: int) -> None:
 248         pass
 249
 250     @abc.abstractmethod
 251     def untrack(self, vpid: int) -> None:
 252         pass
 253
 254
 255 class UserIDProcessAttributeTracker(ProcessAttributeTracker):
 256     @abc.abstractmethod
 257     def track(self, uid: Union[int, str]) -> None:
 258         pass
 259
 260     @abc.abstractmethod
 261     def untrack(self, uid: Union[int, str]) -> None:
 262         pass
 263
 264
 265 class VirtualUserIDProcessAttributeTracker(ProcessAttributeTracker):
 266     @abc.abstractmethod
 267     def track(self, vuid: Union[int, str]) -> None:
 268         pass
 269
 270     @abc.abstractmethod
 271     def untrack(self, vuid: Union[int, str]) -> None:
 272         pass
 273
 274
 275 class GroupIDProcessAttributeTracker(ProcessAttributeTracker):
 276     @abc.abstractmethod
 277     def track(self, gid: Union[int, str]) -> None:
 278         pass
 279
 280     @abc.abstractmethod
 281     def untrack(self, gid: Union[int, str]) -> None:
 282         pass
 283
 284
 285 class VirtualGroupIDProcessAttributeTracker(ProcessAttributeTracker):
 286     @abc.abstractmethod
 287     def track(self, vgid: Union[int, str]) -> None:
 288         pass
 289
 290     @abc.abstractmethod
 291     def untrack(self, vgid: Union[int, str]) -> None:
 292         pass
 293
 294
 295 class Session(abc.ABC):
 296     @staticmethod
 297     def _generate_name() -> str:
 298         return "session_{random_id}".format(random_id=_generate_random_string(8))
 299
 300     @property
 301     @abc.abstractmethod
 302     def name(self) -> str:
 303         pass
 304
 305     @property
 306     @abc.abstractmethod
 307     def output(self) -> Optional[Type[SessionOutputLocation]]:
 308         pass
 309
 310     @abc.abstractmethod
 311     def add_channel(
 312         self, domain: TracingDomain, channel_name: Optional[str] = None
 313     ) -> Channel:
 314         """Add a channel with default attributes to the session."""
 315         pass
 316
 317     @abc.abstractmethod
 318     def start(self) -> None:
 319         pass
 320
 321     @abc.abstractmethod
 322     def stop(self) -> None:
 323         pass
 324
 325     @abc.abstractmethod
 326     def destroy(self) -> None:
 327         pass
 328
 329     @abc.abstractproperty
 330     def kernel_pid_process_attribute_tracker(
 331         self,
 332     ) -> Type[ProcessIDProcessAttributeTracker]:
 333         raise NotImplementedError
 334
 335     @abc.abstractproperty
 336     def kernel_vpid_process_attribute_tracker(
 337         self,
 338     ) -> Type[VirtualProcessIDProcessAttributeTracker]:
 339         raise NotImplementedError
 340
 341     @abc.abstractproperty
 342     def user_vpid_process_attribute_tracker(
 343         self,
 344     ) -> Type[VirtualProcessIDProcessAttributeTracker]:
 345         raise NotImplementedError
 346
 347     @abc.abstractproperty
 348     def kernel_gid_process_attribute_tracker(
 349         self,
 350     ) -> Type[GroupIDProcessAttributeTracker]:
 351         raise NotImplementedError
 352
 353     @abc.abstractproperty
 354     def kernel_vgid_process_attribute_tracker(
 355         self,
 356     ) -> Type[VirtualGroupIDProcessAttributeTracker]:
 357         raise NotImplementedError
 358
 359     @abc.abstractproperty
 360     def user_vgid_process_attribute_tracker(
 361         self,
 362     ) -> Type[VirtualGroupIDProcessAttributeTracker]:
 363         raise NotImplementedError
 364
 365     @abc.abstractproperty
 366     def kernel_uid_process_attribute_tracker(
 367         self,
 368     ) -> Type[UserIDProcessAttributeTracker]:
 369         raise NotImplementedError
 370
 371     @abc.abstractproperty
 372     def kernel_vuid_process_attribute_tracker(
 373         self,
 374     ) -> Type[VirtualUserIDProcessAttributeTracker]:
 375         raise NotImplementedError
 376
 377     @abc.abstractproperty
 378     def user_vuid_process_attribute_tracker(
 379         self,
 380     ) -> Type[VirtualUserIDProcessAttributeTracker]:
 381         raise NotImplementedError
 382
 383
 384 class ControlException(RuntimeError):
 385     """Base type for exceptions thrown by a controller."""
 386
 387     def __init__(self, msg: str):
 388         super().__init__(msg)
 389
 390
 391 class Controller(abc.ABC):
 392     """
 393     Interface of a top-level control interface. A control interface can be, for
 394     example, the LTTng client or a wrapper around liblttng-ctl. It is used to
 395     create and manage top-level objects of a session daemon instance.
 396     """
 397
 398     @abc.abstractmethod
 399     def create_session(
 400         self, name: Optional[str] = None, output: Optional[SessionOutputLocation] = None
 401     ) -> Session:
 402         """
 403         Create a session with an output. Don't specify an output
 404         to create a session without an output.
 405         """
 406         pass