Tests: python: enum.auto() introduced in python 3.6
[lttng-tools.git] / 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
This page took 0.039794 seconds and 5 git commands to generate.