tests/utils/lttngtest/lttng.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
   8 from . import lttngctl, logger, environment
   9 import os
  10 from typing import Callable, Optional, Type, Union, Iterator
  11 import shlex
  12 import subprocess
  13 import enum
  14 import xml.etree.ElementTree
  15
  16 """
  17 Implementation of the lttngctl interface based on the `lttng` command line client.
  18 """
  19
  20
  21 class Unsupported(lttngctl.ControlException):
  22     def __init__(self, msg):
  23         # type: (str) -> None
  24         super().__init__(msg)
  25
  26
  27 class InvalidMI(lttngctl.ControlException):
  28     def __init__(self, msg):
  29         # type: (str) -> None
  30         super().__init__(msg)
  31
  32
  33 class ChannelNotFound(lttngctl.ControlException):
  34     def __init__(self, msg):
  35         # type: (str) -> None
  36         super().__init__(msg)
  37
  38
  39 def _get_domain_option_name(domain):
  40     # type: (lttngctl.TracingDomain) -> str
  41     return {
  42         lttngctl.TracingDomain.User: "userspace",
  43         lttngctl.TracingDomain.Kernel: "kernel",
  44         lttngctl.TracingDomain.Log4j: "log4j",
  45         lttngctl.TracingDomain.Python: "python",
  46         lttngctl.TracingDomain.JUL: "jul",
  47     }[domain]
  48
  49
  50 def _get_domain_xml_mi_name(domain):
  51     # type: (lttngctl.TracingDomain) -> str
  52     return {
  53         lttngctl.TracingDomain.User: "UST",
  54         lttngctl.TracingDomain.Kernel: "KERNEL",
  55         lttngctl.TracingDomain.Log4j: "LOG4J",
  56         lttngctl.TracingDomain.Python: "PYTHON",
  57         lttngctl.TracingDomain.JUL: "JUL",
  58     }[domain]
  59
  60
  61 def _get_context_type_name(context):
  62     # type: (lttngctl.ContextType) -> str
  63     if isinstance(context, lttngctl.VgidContextType):
  64         return "vgid"
  65     elif isinstance(context, lttngctl.VuidContextType):
  66         return "vuid"
  67     elif isinstance(context, lttngctl.VpidContextType):
  68         return "vpid"
  69     elif isinstance(context, lttngctl.JavaApplicationContextType):
  70         return "$app.{retriever}:{field}".format(
  71             retriever=context.retriever_name, field=context.field_name
  72         )
  73     else:
  74         raise Unsupported(
  75             "Context `{context_name}` is not supported by the LTTng client".format(
  76                 type(context).__name__
  77             )
  78         )
  79
  80
  81 def _get_log_level_argument_name(log_level):
  82     # type: (lttngctl.LogLevel) -> str
  83     if isinstance(log_level, lttngctl.UserLogLevel):
  84         return {
  85             lttngctl.UserLogLevel.EMERGENCY: "EMER",
  86             lttngctl.UserLogLevel.ALERT: "ALERT",
  87             lttngctl.UserLogLevel.CRITICAL: "CRIT",
  88             lttngctl.UserLogLevel.ERROR: "ERR",
  89             lttngctl.UserLogLevel.WARNING: "WARNING",
  90             lttngctl.UserLogLevel.NOTICE: "NOTICE",
  91             lttngctl.UserLogLevel.INFO: "INFO",
  92             lttngctl.UserLogLevel.DEBUG_SYSTEM: "DEBUG_SYSTEM",
  93             lttngctl.UserLogLevel.DEBUG_PROGRAM: "DEBUG_PROGRAM",
  94             lttngctl.UserLogLevel.DEBUG_PROCESS: "DEBUG_PROCESS",
  95             lttngctl.UserLogLevel.DEBUG_MODULE: "DEBUG_MODULE",
  96             lttngctl.UserLogLevel.DEBUG_UNIT: "DEBUG_UNIT",
  97             lttngctl.UserLogLevel.DEBUG_FUNCTION: "DEBUG_FUNCTION",
  98             lttngctl.UserLogLevel.DEBUG_LINE: "DEBUG_LINE",
  99             lttngctl.UserLogLevel.DEBUG: "DEBUG",
 100         }[log_level]
 101     elif isinstance(log_level, lttngctl.JULLogLevel):
 102         return {
 103             lttngctl.JULLogLevel.OFF: "OFF",
 104             lttngctl.JULLogLevel.SEVERE: "SEVERE",
 105             lttngctl.JULLogLevel.WARNING: "WARNING",
 106             lttngctl.JULLogLevel.INFO: "INFO",
 107             lttngctl.JULLogLevel.CONFIG: "CONFIG",
 108             lttngctl.JULLogLevel.FINE: "FINE",
 109             lttngctl.JULLogLevel.FINER: "FINER",
 110             lttngctl.JULLogLevel.FINEST: "FINEST",
 111             lttngctl.JULLogLevel.ALL: "ALL",
 112         }[log_level]
 113     elif isinstance(log_level, lttngctl.Log4jLogLevel):
 114         return {
 115             lttngctl.Log4jLogLevel.OFF: "OFF",
 116             lttngctl.Log4jLogLevel.FATAL: "FATAL",
 117             lttngctl.Log4jLogLevel.ERROR: "ERROR",
 118             lttngctl.Log4jLogLevel.WARN: "WARN",
 119             lttngctl.Log4jLogLevel.INFO: "INFO",
 120             lttngctl.Log4jLogLevel.DEBUG: "DEBUG",
 121             lttngctl.Log4jLogLevel.TRACE: "TRACE",
 122             lttngctl.Log4jLogLevel.ALL: "ALL",
 123         }[log_level]
 124     elif isinstance(log_level, lttngctl.PythonLogLevel):
 125         return {
 126             lttngctl.PythonLogLevel.CRITICAL: "CRITICAL",
 127             lttngctl.PythonLogLevel.ERROR: "ERROR",
 128             lttngctl.PythonLogLevel.WARNING: "WARNING",
 129             lttngctl.PythonLogLevel.INFO: "INFO",
 130             lttngctl.PythonLogLevel.DEBUG: "DEBUG",
 131             lttngctl.PythonLogLevel.NOTSET: "NOTSET",
 132         }[log_level]
 133
 134     raise TypeError("Unknown log level type")
 135
 136
 137 def _get_log_level_from_mi_log_level_name(mi_log_level_name):
 138     # type: (str) -> lttngctl.LogLevel
 139     return {
 140         "TRACE_EMERG": lttngctl.UserLogLevel.EMERGENCY,
 141         "TRACE_ALERT": lttngctl.UserLogLevel.ALERT,
 142         "TRACE_CRIT": lttngctl.UserLogLevel.CRITICAL,
 143         "TRACE_ERR": lttngctl.UserLogLevel.ERROR,
 144         "TRACE_WARNING": lttngctl.UserLogLevel.WARNING,
 145         "TRACE_NOTICE": lttngctl.UserLogLevel.NOTICE,
 146         "TRACE_INFO": lttngctl.UserLogLevel.INFO,
 147         "TRACE_DEBUG_SYSTEM": lttngctl.UserLogLevel.DEBUG_SYSTEM,
 148         "TRACE_DEBUG_PROGRAM": lttngctl.UserLogLevel.DEBUG_PROGRAM,
 149         "TRACE_DEBUG_PROCESS": lttngctl.UserLogLevel.DEBUG_PROCESS,
 150         "TRACE_DEBUG_MODULE": lttngctl.UserLogLevel.DEBUG_MODULE,
 151         "TRACE_DEBUG_UNIT": lttngctl.UserLogLevel.DEBUG_UNIT,
 152         "TRACE_DEBUG_FUNCTION": lttngctl.UserLogLevel.DEBUG_FUNCTION,
 153         "TRACE_DEBUG_LINE": lttngctl.UserLogLevel.DEBUG_LINE,
 154         "TRACE_DEBUG": lttngctl.UserLogLevel.DEBUG,
 155         "JUL_OFF": lttngctl.JULLogLevel.OFF,
 156         "JUL_SEVERE": lttngctl.JULLogLevel.SEVERE,
 157         "JUL_WARNING": lttngctl.JULLogLevel.WARNING,
 158         "JUL_INFO": lttngctl.JULLogLevel.INFO,
 159         "JUL_CONFIG": lttngctl.JULLogLevel.CONFIG,
 160         "JUL_FINE": lttngctl.JULLogLevel.FINE,
 161         "JUL_FINER": lttngctl.JULLogLevel.FINER,
 162         "JUL_FINEST": lttngctl.JULLogLevel.FINEST,
 163         "JUL_ALL": lttngctl.JULLogLevel.ALL,
 164         "LOG4J_OFF": lttngctl.Log4jLogLevel.OFF,
 165         "LOG4J_FATAL": lttngctl.Log4jLogLevel.FATAL,
 166         "LOG4J_ERROR": lttngctl.Log4jLogLevel.ERROR,
 167         "LOG4J_WARN": lttngctl.Log4jLogLevel.WARN,
 168         "LOG4J_INFO": lttngctl.Log4jLogLevel.INFO,
 169         "LOG4J_DEBUG": lttngctl.Log4jLogLevel.DEBUG,
 170         "LOG4J_TRACE": lttngctl.Log4jLogLevel.TRACE,
 171         "LOG4J_ALL": lttngctl.Log4jLogLevel.ALL,
 172         "PYTHON_CRITICAL": lttngctl.PythonLogLevel.CRITICAL,
 173         "PYTHON_ERROR": lttngctl.PythonLogLevel.ERROR,
 174         "PYTHON_WARNING": lttngctl.PythonLogLevel.WARNING,
 175         "PYTHON_INFO": lttngctl.PythonLogLevel.INFO,
 176         "PYTHON_DEBUG": lttngctl.PythonLogLevel.DEBUG,
 177         "PYTHON_NOTSET": lttngctl.PythonLogLevel.NOTSET,
 178     }[mi_log_level_name]
 179
 180
 181 def _get_tracepoint_event_rule_class_from_domain_type(domain_type):
 182     # type: (lttngctl.TracingDomain) -> Type[lttngctl.UserTracepointEventRule] | Type[lttngctl.Log4jTracepointEventRule] | Type[lttngctl.JULTracepointEventRule] | Type[lttngctl.PythonTracepointEventRule] | Type[lttngctl.KernelTracepointEventRule]
 183     return {
 184         lttngctl.TracingDomain.User: lttngctl.UserTracepointEventRule,
 185         lttngctl.TracingDomain.JUL: lttngctl.JULTracepointEventRule,
 186         lttngctl.TracingDomain.Log4j: lttngctl.Log4jTracepointEventRule,
 187         lttngctl.TracingDomain.Python: lttngctl.PythonTracepointEventRule,
 188         lttngctl.TracingDomain.Kernel: lttngctl.KernelTracepointEventRule,
 189     }[domain_type]
 190
 191
 192 class _Channel(lttngctl.Channel):
 193     def __init__(
 194         self,
 195         client,  # type: LTTngClient
 196         name,  # type: str
 197         domain,  # type: lttngctl.TracingDomain
 198         session,  # type: _Session
 199     ):
 200         self._client = client  # type: LTTngClient
 201         self._name = name  # type: str
 202         self._domain = domain  # type: lttngctl.TracingDomain
 203         self._session = session  # type: _Session
 204
 205     def add_context(self, context_type):
 206         # type: (lttngctl.ContextType) -> None
 207         domain_option_name = _get_domain_option_name(self.domain)
 208         context_type_name = _get_context_type_name(context_type)
 209         self._client._run_cmd(
 210             "add-context --{domain_option_name} --channel '{channel_name}' --type {context_type_name}".format(
 211                 domain_option_name=domain_option_name,
 212                 channel_name=self.name,
 213                 context_type_name=context_type_name,
 214             )
 215         )
 216
 217     def add_recording_rule(self, rule):
 218         # type: (Type[lttngctl.EventRule]) -> None
 219         client_args = (
 220             "enable-event --session {session_name} --channel {channel_name}".format(
 221                 session_name=self._session.name, channel_name=self.name
 222             )
 223         )
 224         if isinstance(rule, lttngctl.TracepointEventRule):
 225             domain_option_name = (
 226                 "userspace"
 227                 if isinstance(rule, lttngctl.UserTracepointEventRule)
 228                 else "kernel"
 229             )
 230             client_args = client_args + " --{domain_option_name}".format(
 231                 domain_option_name=domain_option_name
 232             )
 233
 234             if rule.name_pattern:
 235                 client_args = client_args + " " + rule.name_pattern
 236             else:
 237                 client_args = client_args + " --all"
 238
 239             if rule.filter_expression:
 240                 client_args = client_args + " " + rule.filter_expression
 241
 242             if rule.log_level_rule:
 243                 if isinstance(rule.log_level_rule, lttngctl.LogLevelRuleAsSevereAs):
 244                     client_args = client_args + " --loglevel {log_level}".format(
 245                         log_level=_get_log_level_argument_name(
 246                             rule.log_level_rule.level
 247                         )
 248                     )
 249                 elif isinstance(rule.log_level_rule, lttngctl.LogLevelRuleExactly):
 250                     client_args = client_args + " --loglevel-only {log_level}".format(
 251                         log_level=_get_log_level_argument_name(
 252                             rule.log_level_rule.level
 253                         )
 254                     )
 255                 else:
 256                     raise Unsupported(
 257                         "Unsupported log level rule type `{log_level_rule_type}`".format(
 258                             log_level_rule_type=type(rule.log_level_rule).__name__
 259                         )
 260                     )
 261
 262             if rule.name_pattern_exclusions:
 263                 client_args = client_args + " --exclude "
 264                 for idx, pattern in enumerate(rule.name_pattern_exclusions):
 265                     if idx != 0:
 266                         client_args = client_args + ","
 267                     client_args = client_args + pattern
 268         else:
 269             raise Unsupported(
 270                 "event rule type `{event_rule_type}` is unsupported by LTTng client".format(
 271                     event_rule_type=type(rule).__name__
 272                 )
 273             )
 274
 275         self._client._run_cmd(client_args)
 276
 277     @property
 278     def name(self):
 279         # type: () -> str
 280         return self._name
 281
 282     @property
 283     def domain(self):
 284         # type: () -> lttngctl.TracingDomain
 285         return self._domain
 286
 287     @property
 288     def recording_rules(self):
 289         # type: () -> Iterator[lttngctl.EventRule]
 290         list_session_xml = self._client._run_cmd(
 291             "list '{session_name}'".format(session_name=self._session.name),
 292             LTTngClient.CommandOutputFormat.MI_XML,
 293         )
 294
 295         root = xml.etree.ElementTree.fromstring(list_session_xml)
 296         command_output = LTTngClient._mi_get_in_element(root, "output")
 297         sessions = LTTngClient._mi_get_in_element(command_output, "sessions")
 298
 299         # The channel's session is supposed to be the only session returned by the command
 300         if len(sessions) != 1:
 301             raise InvalidMI(
 302                 "Only one session expected when listing with an explicit session name"
 303             )
 304         session = sessions[0]
 305
 306         # Look for the channel's domain
 307         target_domain = None
 308         target_domain_mi_name = _get_domain_xml_mi_name(self.domain)
 309         for domain in LTTngClient._mi_get_in_element(session, "domains"):
 310             if (
 311                 LTTngClient._mi_get_in_element(domain, "type").text
 312                 == target_domain_mi_name
 313             ):
 314                 target_domain = domain
 315
 316         if target_domain is None:
 317             raise ChannelNotFound(
 318                 "Failed to find channel `{channel_name}`: no channel in target domain".format(
 319                     channel_name=self.name
 320                 )
 321             )
 322
 323         target_channel = None
 324         for channel in LTTngClient._mi_get_in_element(target_domain, "channels"):
 325             if LTTngClient._mi_get_in_element(channel, "name").text == self.name:
 326                 target_channel = channel
 327                 break
 328
 329         if target_channel is None:
 330             raise ChannelNotFound(
 331                 "Failed to find channel `{channel_name}`: no such channel in target domain".format(
 332                     channel_name=self.name
 333                 )
 334             )
 335
 336         tracepoint_event_rule_class = None
 337
 338         for event in LTTngClient._mi_get_in_element(target_channel, "events"):
 339             # Note that the "enabled" property is ignored as it is not exposed by
 340             # the EventRule interface.
 341             pattern = LTTngClient._mi_get_in_element(event, "name").text
 342             type = LTTngClient._mi_get_in_element(event, "type").text
 343
 344             filter_expression = None
 345             filter_expression_element = LTTngClient._mi_find_in_element(
 346                 event, "filter_expression"
 347             )
 348             if filter_expression_element:
 349                 filter_expression = filter_expression_element.text
 350
 351             exclusions = []
 352             for exclusion in LTTngClient._mi_get_in_element(event, "exclusions"):
 353                 exclusions.append(exclusion.text)
 354
 355             exclusions = exclusions if len(exclusions) > 0 else None
 356
 357             if type != "TRACEPOINT":
 358                 raise Unsupported(
 359                     "Non-tracepoint event rules are not supported by this Controller implementation"
 360                 )
 361
 362             tracepoint_event_rule_class = (
 363                 _get_tracepoint_event_rule_class_from_domain_type(self.domain)
 364             )
 365             event_rule = None
 366             if self.domain != lttngctl.TracingDomain.Kernel:
 367                 log_level_element = LTTngClient._mi_find_in_element(event, "loglevel")
 368                 log_level_type_element = LTTngClient._mi_find_in_element(
 369                     event, "loglevel_type"
 370                 )
 371
 372                 log_level_rule = None
 373                 if log_level_element is not None and log_level_type_element is not None:
 374                     if log_level_element.text is None:
 375                         raise InvalidMI("`loglevel` element of event rule has no text")
 376
 377                     if log_level_type_element.text == "RANGE":
 378                         log_level_rule = lttngctl.LogLevelRuleAsSevereAs(
 379                             _get_log_level_from_mi_log_level_name(
 380                                 log_level_element.text
 381                             )
 382                         )
 383                     elif log_level_type_element.text == "SINGLE":
 384                         log_level_rule = lttngctl.LogLevelRuleExactly(
 385                             _get_log_level_from_mi_log_level_name(
 386                                 log_level_element.text
 387                             )
 388                         )
 389
 390                 yield tracepoint_event_rule_class(
 391                     pattern, filter_expression, log_level_rule, exclusions
 392                 )
 393             else:
 394                 yield tracepoint_event_rule_class(pattern, filter_expression)
 395
 396
 397 @enum.unique
 398 class _ProcessAttribute(enum.Enum):
 399     PID = "Process ID"
 400     VPID = "Virtual Process ID"
 401     UID = "User ID"
 402     VUID = "Virtual User ID"
 403     GID = "Group ID"
 404     VGID = "Virtual Group ID"
 405
 406     def __repr__(self):
 407         return "<%s.%s>" % (self.__class__.__name__, self.name)
 408
 409
 410 def _get_process_attribute_option_name(attribute):
 411     # type: (_ProcessAttribute) -> str
 412     return {
 413         _ProcessAttribute.PID: "pid",
 414         _ProcessAttribute.VPID: "vpid",
 415         _ProcessAttribute.UID: "uid",
 416         _ProcessAttribute.VUID: "vuid",
 417         _ProcessAttribute.GID: "gid",
 418         _ProcessAttribute.VGID: "vgid",
 419     }[attribute]
 420
 421
 422 class _ProcessAttributeTracker(lttngctl.ProcessAttributeTracker):
 423     def __init__(
 424         self,
 425         client,  # type: LTTngClient
 426         attribute,  # type: _ProcessAttribute
 427         domain,  # type: lttngctl.TracingDomain
 428         session,  # type: _Session
 429     ):
 430         self._client = client  # type: LTTngClient
 431         self._tracked_attribute = attribute  # type: _ProcessAttribute
 432         self._domain = domain  # type: lttngctl.TracingDomain
 433         self._session = session  # type: _Session
 434         if attribute == _ProcessAttribute.PID or attribute == _ProcessAttribute.VPID:
 435             self._allowed_value_types = [int, str]  # type: list[type]
 436         else:
 437             self._allowed_value_types = [int]  # type: list[type]
 438
 439     def _call_client(self, cmd_name, value):
 440         # type: (str, Union[int, str]) -> None
 441         if type(value) not in self._allowed_value_types:
 442             raise TypeError(
 443                 "Value of type `{value_type}` is not allowed for process attribute {attribute_name}".format(
 444                     value_type=type(value).__name__,
 445                     attribute_name=self._tracked_attribute.name,
 446                 )
 447             )
 448
 449         process_attribute_option_name = _get_process_attribute_option_name(
 450             self._tracked_attribute
 451         )
 452         domain_name = _get_domain_option_name(self._domain)
 453         self._client._run_cmd(
 454             "{cmd_name} --session '{session_name}' --{domain_name} --{tracked_attribute_name} {value}".format(
 455                 cmd_name=cmd_name,
 456                 session_name=self._session.name,
 457                 domain_name=domain_name,
 458                 tracked_attribute_name=process_attribute_option_name,
 459                 value=value,
 460             )
 461         )
 462
 463     def track(self, value):
 464         # type: (Union[int, str]) -> None
 465         self._call_client("track", value)
 466
 467     def untrack(self, value):
 468         # type: (Union[int, str]) -> None
 469         self._call_client("untrack", value)
 470
 471
 472 class _Session(lttngctl.Session):
 473     def __init__(
 474         self,
 475         client,  # type: LTTngClient
 476         name,  # type: str
 477         output,  # type: Optional[lttngctl.SessionOutputLocation]
 478     ):
 479         self._client = client  # type: LTTngClient
 480         self._name = name  # type: str
 481         self._output = output  # type: Optional[lttngctl.SessionOutputLocation]
 482
 483     @property
 484     def name(self):
 485         # type: () -> str
 486         return self._name
 487
 488     def add_channel(
 489         self,
 490         domain,
 491         channel_name=None,
 492         buffer_sharing_policy=lttngctl.BufferSharingPolicy.PerUID,
 493     ):
 494         # type: (lttngctl.TracingDomain, Optional[str], lttngctl.BufferSharingPolicy) -> lttngctl.Channel
 495         channel_name = lttngctl.Channel._generate_name()
 496         domain_option_name = _get_domain_option_name(domain)
 497         self._client._run_cmd(
 498             "enable-channel --session '{session_name}' --{domain_name} '{channel_name}' {buffer_sharing_policy}".format(
 499                 session_name=self.name,
 500                 domain_name=domain_option_name,
 501                 channel_name=channel_name,
 502                 buffer_sharing_policy=(
 503                     "--buffers-uid"
 504                     if buffer_sharing_policy == lttngctl.BufferSharingPolicy.PerUID
 505                     else "--buffers-pid"
 506                 ),
 507             )
 508         )
 509         return _Channel(self._client, channel_name, domain, self)
 510
 511     def add_context(self, context_type):
 512         # type: (lttngctl.ContextType) -> None
 513         pass
 514
 515     @property
 516     def output(self):
 517         # type: () -> "Optional[Type[lttngctl.SessionOutputLocation]]"
 518         return self._output  # type: ignore
 519
 520     def start(self):
 521         # type: () -> None
 522         self._client._run_cmd("start '{session_name}'".format(session_name=self.name))
 523
 524     def stop(self):
 525         # type: () -> None
 526         self._client._run_cmd("stop '{session_name}'".format(session_name=self.name))
 527
 528     def clear(self):
 529         # type: () -> None
 530         self._client._run_cmd("clear '{session_name}'".format(session_name=self.name))
 531
 532     def destroy(self):
 533         # type: () -> None
 534         self._client._run_cmd("destroy '{session_name}'".format(session_name=self.name))
 535
 536     def rotate(self, wait=True):
 537         # type: (bool) -> None
 538         self._client.rotate_session_by_name(self.name, wait)
 539
 540     @property
 541     def is_active(self):
 542         # type: () -> bool
 543         list_session_xml = self._client._run_cmd(
 544             "list '{session_name}'".format(session_name=self.name),
 545             LTTngClient.CommandOutputFormat.MI_XML,
 546         )
 547
 548         root = xml.etree.ElementTree.fromstring(list_session_xml)
 549         command_output = LTTngClient._mi_get_in_element(root, "output")
 550         sessions = LTTngClient._mi_get_in_element(command_output, "sessions")
 551         session_mi = LTTngClient._mi_get_in_element(sessions, "session")
 552
 553         enabled_text = LTTngClient._mi_get_in_element(session_mi, "enabled").text
 554         if enabled_text not in ["true", "false"]:
 555             raise InvalidMI(
 556                 "Expected boolean value in element '{}': value='{}'".format(
 557                     session_mi.tag, enabled_text
 558                 )
 559             )
 560
 561         return enabled_text == "true"
 562
 563     @property
 564     def kernel_pid_process_attribute_tracker(self):
 565         # type: () -> Type[lttngctl.ProcessIDProcessAttributeTracker]
 566         return _ProcessAttributeTracker(self._client, _ProcessAttribute.PID, lttngctl.TracingDomain.Kernel, self)  # type: ignore
 567
 568     @property
 569     def kernel_vpid_process_attribute_tracker(self):
 570         # type: () -> Type[lttngctl.VirtualProcessIDProcessAttributeTracker]
 571         return _ProcessAttributeTracker(self._client, _ProcessAttribute.VPID, lttngctl.TracingDomain.Kernel, self)  # type: ignore
 572
 573     @property
 574     def user_vpid_process_attribute_tracker(self):
 575         # type: () -> Type[lttngctl.VirtualProcessIDProcessAttributeTracker]
 576         return _ProcessAttributeTracker(self._client, _ProcessAttribute.VPID, lttngctl.TracingDomain.User, self)  # type: ignore
 577
 578     @property
 579     def kernel_gid_process_attribute_tracker(self):
 580         # type: () -> Type[lttngctl.GroupIDProcessAttributeTracker]
 581         return _ProcessAttributeTracker(self._client, _ProcessAttribute.GID, lttngctl.TracingDomain.Kernel, self)  # type: ignore
 582
 583     @property
 584     def kernel_vgid_process_attribute_tracker(self):
 585         # type: () -> Type[lttngctl.VirtualGroupIDProcessAttributeTracker]
 586         return _ProcessAttributeTracker(self._client, _ProcessAttribute.VGID, lttngctl.TracingDomain.Kernel, self)  # type: ignore
 587
 588     @property
 589     def user_vgid_process_attribute_tracker(self):
 590         # type: () -> Type[lttngctl.VirtualGroupIDProcessAttributeTracker]
 591         return _ProcessAttributeTracker(self._client, _ProcessAttribute.VGID, lttngctl.TracingDomain.User, self)  # type: ignore
 592
 593     @property
 594     def kernel_uid_process_attribute_tracker(self):
 595         # type: () -> Type[lttngctl.UserIDProcessAttributeTracker]
 596         return _ProcessAttributeTracker(self._client, _ProcessAttribute.UID, lttngctl.TracingDomain.Kernel, self)  # type: ignore
 597
 598     @property
 599     def kernel_vuid_process_attribute_tracker(self):
 600         # type: () -> Type[lttngctl.VirtualUserIDProcessAttributeTracker]
 601         return _ProcessAttributeTracker(self._client, _ProcessAttribute.VUID, lttngctl.TracingDomain.Kernel, self)  # type: ignore
 602
 603     @property
 604     def user_vuid_process_attribute_tracker(self):
 605         # type: () -> Type[lttngctl.VirtualUserIDProcessAttributeTracker]
 606         return _ProcessAttributeTracker(self._client, _ProcessAttribute.VUID, lttngctl.TracingDomain.User, self)  # type: ignore
 607
 608
 609 class LTTngClientError(lttngctl.ControlException):
 610     def __init__(
 611         self,
 612         command_args,  # type: str
 613         error_output,  # type: str
 614     ):
 615         self._command_args = command_args  # type: str
 616         self._output = error_output  # type: str
 617
 618
 619 class LTTngClient(logger._Logger, lttngctl.Controller):
 620     """
 621     Implementation of a LTTngCtl Controller that uses the `lttng` client as a back-end.
 622     """
 623
 624     class CommandOutputFormat(enum.Enum):
 625         MI_XML = 0
 626         HUMAN = 1
 627
 628     _MI_NS = "{https://lttng.org/xml/ns/lttng-mi}"
 629
 630     def __init__(
 631         self,
 632         test_environment,  # type: environment._Environment
 633         log,  # type: Optional[Callable[[str], None]]
 634     ):
 635         logger._Logger.__init__(self, log)
 636         self._environment = test_environment  # type: environment._Environment
 637
 638     @staticmethod
 639     def _namespaced_mi_element(property):
 640         # type: (str) -> str
 641         return LTTngClient._MI_NS + property
 642
 643     def _run_cmd(self, command_args, output_format=CommandOutputFormat.MI_XML):
 644         # type: (str, CommandOutputFormat) -> str
 645         """
 646         Invoke the `lttng` client with a set of arguments. The command is
 647         executed in the context of the client's test environment.
 648         """
 649         args = [str(self._environment.lttng_client_path)]  # type: list[str]
 650         if output_format == LTTngClient.CommandOutputFormat.MI_XML:
 651             args.extend(["--mi", "xml"])
 652
 653         args.extend(shlex.split(command_args))
 654
 655         self._log("lttng {command_args}".format(command_args=command_args))
 656
 657         client_env = os.environ.copy()  # type: dict[str, str]
 658         client_env["LTTNG_HOME"] = str(self._environment.lttng_home_location)
 659
 660         process = subprocess.Popen(
 661             args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, env=client_env
 662         )
 663
 664         out = process.communicate()[0]
 665
 666         if process.returncode != 0:
 667             decoded_output = out.decode("utf-8")
 668             for error_line in decoded_output.splitlines():
 669                 self._log(error_line)
 670
 671             raise LTTngClientError(command_args, decoded_output)
 672         else:
 673             return out.decode("utf-8")
 674
 675     def create_session(self, name=None, output=None, live=False):
 676         # type: (Optional[str], Optional[lttngctl.SessionOutputLocation], bool) -> lttngctl.Session
 677         name = name if name else lttngctl.Session._generate_name()
 678
 679         if isinstance(output, lttngctl.LocalSessionOutputLocation):
 680             output_option = "--output '{output_path}'".format(output_path=output.path)
 681         elif isinstance(output, lttngctl.NetworkSessionOutputLocation):
 682             output_option = "--set-url '{}'".format(output.url)
 683         elif output is None:
 684             output_option = "--no-output"
 685         else:
 686             raise TypeError("LTTngClient only supports local or no output")
 687
 688         self._run_cmd(
 689             "create '{session_name}' {output_option} {live_option}".format(
 690                 session_name=name,
 691                 output_option=output_option,
 692                 live_option="--live" if live else "",
 693             )
 694         )
 695         return _Session(self, name, output)
 696
 697     def start_session_by_name(self, name):
 698         # type: (str) -> None
 699         self._run_cmd("start '{session_name}'".format(session_name=name))
 700
 701     def start_session_by_glob_pattern(self, pattern):
 702         # type: (str) -> None
 703         self._run_cmd("start --glob '{pattern}'".format(pattern=pattern))
 704
 705     def start_sessions_all(self):
 706         # type: () -> None
 707         self._run_cmd("start --all")
 708
 709     def stop_session_by_name(self, name):
 710         # type: (str) -> None
 711         self._run_cmd("stop '{session_name}'".format(session_name=name))
 712
 713     def stop_session_by_glob_pattern(self, pattern):
 714         # type: (str) -> None
 715         self._run_cmd("stop --glob '{pattern}'".format(pattern=pattern))
 716
 717     def stop_sessions_all(self):
 718         # type: () -> None
 719         self._run_cmd("stop --all")
 720
 721     def destroy_session_by_name(self, name):
 722         # type: (str) -> None
 723         self._run_cmd("destroy '{session_name}'".format(session_name=name))
 724
 725     def destroy_session_by_glob_pattern(self, pattern):
 726         # type: (str) -> None
 727         self._run_cmd("destroy --glob '{pattern}'".format(pattern=pattern))
 728
 729     def destroy_sessions_all(self):
 730         # type: () -> None
 731         self._run_cmd("destroy --all")
 732
 733     def rotate_session_by_name(self, name, wait=True):
 734         self._run_cmd(
 735             "rotate '{session_name}' {wait_option}".format(
 736                 session_name=name, wait_option="-n" if wait is False else ""
 737             )
 738         )
 739
 740     def schedule_size_based_rotation(self, name, size_bytes):
 741         # type (str, int) -> None
 742         self._run_cmd(
 743             "enable-rotation --session '{session_name}' --size {size}".format(
 744                 session_name=name, size=size_bytes
 745             )
 746         )
 747
 748     def schedule_time_based_rotation(self, name, period_seconds):
 749         # type (str, int) -> None
 750         self._run_cmd(
 751             "enable-rotation --session '{session_name}' --timer {period_seconds}s".format(
 752                 session_name=name, period_seconds=period_seconds
 753             )
 754         )
 755
 756     @staticmethod
 757     def _mi_find_in_element(element, sub_element_name):
 758         # type: (xml.etree.ElementTree.Element, str) -> Optional[xml.etree.ElementTree.Element]
 759         return element.find(LTTngClient._namespaced_mi_element(sub_element_name))
 760
 761     @staticmethod
 762     def _mi_get_in_element(element, sub_element_name):
 763         # type: (xml.etree.ElementTree.Element, str) -> xml.etree.ElementTree.Element
 764         result = LTTngClient._mi_find_in_element(element, sub_element_name)
 765         if result is None:
 766             raise InvalidMI(
 767                 "Failed to find element '{}' within command MI element '{}'".format(
 768                     element.tag, sub_element_name
 769                 )
 770             )
 771
 772         return result
 773
 774     def list_sessions(self):
 775         # type () -> List[Session]
 776         list_sessions_xml = self._run_cmd(
 777             "list", LTTngClient.CommandOutputFormat.MI_XML
 778         )
 779
 780         root = xml.etree.ElementTree.fromstring(list_sessions_xml)
 781         command_output = self._mi_get_in_element(root, "output")
 782         sessions = self._mi_get_in_element(command_output, "sessions")
 783
 784         ctl_sessions = []  # type: list[lttngctl.Session]
 785
 786         for session_mi in sessions:
 787             name = self._mi_get_in_element(session_mi, "name").text
 788             path = self._mi_get_in_element(session_mi, "path").text
 789
 790             if name is None:
 791                 raise InvalidMI(
 792                     "Invalid empty 'name' element in '{}'".format(session_mi.tag)
 793                 )
 794             if path is None:
 795                 raise InvalidMI(
 796                     "Invalid empty 'path' element in '{}'".format(session_mi.tag)
 797                 )
 798             if not path.startswith("/"):
 799                 raise Unsupported(
 800                     "{} does not support non-local session outputs".format(type(self))
 801                 )
 802
 803             ctl_sessions.append(
 804                 _Session(self, name, lttngctl.LocalSessionOutputLocation(path))
 805             )
 806
 807         return ctl_sessions