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 pathlib
  10 import os
  11 from typing import Callable, Optional, Type, Union
  12 import shlex
  13 import subprocess
  14 import enum
  15 import xml.etree.ElementTree
  16
  17 """
  18 Implementation of the lttngctl interface based on the `lttng` command line client.
  19 """
  20
  21
  22 class Unsupported(lttngctl.ControlException):
  23     def __init__(self, msg):
  24         # type: (str) -> None
  25         super().__init__(msg)
  26
  27
  28 class InvalidMI(lttngctl.ControlException):
  29     def __init__(self, msg):
  30         # type: (str) -> None
  31         super().__init__(msg)
  32
  33
  34 def _get_domain_option_name(domain):
  35     # type: (lttngctl.TracingDomain) -> str
  36     if domain == lttngctl.TracingDomain.User:
  37         return "userspace"
  38     elif domain == lttngctl.TracingDomain.Kernel:
  39         return "kernel"
  40     elif domain == lttngctl.TracingDomain.Log4j:
  41         return "log4j"
  42     elif domain == lttngctl.TracingDomain.JUL:
  43         return "jul"
  44     elif domain == lttngctl.TracingDomain.Python:
  45         return "python"
  46     else:
  47         raise Unsupported("Domain `{domain_name}` is not supported by the LTTng client")
  48
  49
  50 def _get_context_type_name(context):
  51     # type: (lttngctl.ContextType) -> str
  52     if isinstance(context, lttngctl.VgidContextType):
  53         return "vgid"
  54     elif isinstance(context, lttngctl.VuidContextType):
  55         return "vuid"
  56     elif isinstance(context, lttngctl.VpidContextType):
  57         return "vpid"
  58     elif isinstance(context, lttngctl.JavaApplicationContextType):
  59         return "$app.{retriever}:{field}".format(
  60             retriever=context.retriever_name, field=context.field_name
  61         )
  62     else:
  63         raise Unsupported(
  64             "Context `{context_name}` is not supported by the LTTng client".format(
  65                 type(context).__name__
  66             )
  67         )
  68
  69
  70 class _Channel(lttngctl.Channel):
  71     def __init__(
  72         self,
  73         client,  # type: LTTngClient
  74         name,  # type: str
  75         domain,  # type: lttngctl.TracingDomain
  76         session,  # type: _Session
  77     ):
  78         self._client = client  # type: LTTngClient
  79         self._name = name  # type: str
  80         self._domain = domain  # type: lttngctl.TracingDomain
  81         self._session = session  # type: _Session
  82
  83     def add_context(self, context_type):
  84         # type: (lttngctl.ContextType) -> None
  85         domain_option_name = _get_domain_option_name(self.domain)
  86         context_type_name = _get_context_type_name(context_type)
  87         self._client._run_cmd(
  88             "add-context --{domain_option_name} --channel '{channel_name}' --type {context_type_name}".format(
  89                 domain_option_name=domain_option_name,
  90                 channel_name=self.name,
  91                 context_type_name=context_type_name,
  92             )
  93         )
  94
  95     def add_recording_rule(self, rule):
  96         # type: (Type[lttngctl.EventRule]) -> None
  97         client_args = (
  98             "enable-event --session {session_name} --channel {channel_name}".format(
  99                 session_name=self._session.name, channel_name=self.name
 100             )
 101         )
 102         if isinstance(rule, lttngctl.TracepointEventRule):
 103             domain_option_name = (
 104                 "userspace"
 105                 if isinstance(rule, lttngctl.UserTracepointEventRule)
 106                 else "kernel"
 107             )
 108             client_args = client_args + " --{domain_option_name}".format(
 109                 domain_option_name=domain_option_name
 110             )
 111
 112             if rule.name_pattern:
 113                 client_args = client_args + " " + rule.name_pattern
 114             else:
 115                 client_args = client_args + " --all"
 116
 117             if rule.filter_expression:
 118                 client_args = client_args + " " + rule.filter_expression
 119
 120             if rule.log_level_rule:
 121                 if isinstance(rule.log_level_rule, lttngctl.LogLevelRuleAsSevereAs):
 122                     client_args = client_args + " --loglevel {log_level}".format(
 123                         log_level=rule.log_level_rule.level
 124                     )
 125                 elif isinstance(rule.log_level_rule, lttngctl.LogLevelRuleExactly):
 126                     client_args = client_args + " --loglevel-only {log_level}".format(
 127                         log_level=rule.log_level_rule.level
 128                     )
 129                 else:
 130                     raise Unsupported(
 131                         "Unsupported log level rule type `{log_level_rule_type}`".format(
 132                             log_level_rule_type=type(rule.log_level_rule).__name__
 133                         )
 134                     )
 135
 136             if rule.name_pattern_exclusions:
 137                 client_args = client_args + " --exclude "
 138                 for idx, pattern in enumerate(rule.name_pattern_exclusions):
 139                     if idx != 0:
 140                         client_args = client_args + ","
 141                     client_args = client_args + pattern
 142         else:
 143             raise Unsupported(
 144                 "event rule type `{event_rule_type}` is unsupported by LTTng client".format(
 145                     event_rule_type=type(rule).__name__
 146                 )
 147             )
 148
 149         self._client._run_cmd(client_args)
 150
 151     @property
 152     def name(self):
 153         # type: () -> str
 154         return self._name
 155
 156     @property
 157     def domain(self):
 158         # type: () -> lttngctl.TracingDomain
 159         return self._domain
 160
 161
 162 @enum.unique
 163 class _ProcessAttribute(enum.Enum):
 164     PID = "Process ID"
 165     VPID = "Virtual Process ID"
 166     UID = "User ID"
 167     VUID = "Virtual User ID"
 168     GID = "Group ID"
 169     VGID = "Virtual Group ID"
 170
 171     def __repr__(self):
 172         return "<%s.%s>" % (self.__class__.__name__, self.name)
 173
 174
 175 def _get_process_attribute_option_name(attribute):
 176     # type: (_ProcessAttribute) -> str
 177     return {
 178         _ProcessAttribute.PID: "pid",
 179         _ProcessAttribute.VPID: "vpid",
 180         _ProcessAttribute.UID: "uid",
 181         _ProcessAttribute.VUID: "vuid",
 182         _ProcessAttribute.GID: "gid",
 183         _ProcessAttribute.VGID: "vgid",
 184     }[attribute]
 185
 186
 187 class _ProcessAttributeTracker(lttngctl.ProcessAttributeTracker):
 188     def __init__(
 189         self,
 190         client,  # type: LTTngClient
 191         attribute,  # type: _ProcessAttribute
 192         domain,  # type: lttngctl.TracingDomain
 193         session,  # type: _Session
 194     ):
 195         self._client = client  # type: LTTngClient
 196         self._tracked_attribute = attribute  # type: _ProcessAttribute
 197         self._domain = domain  # type: lttngctl.TracingDomain
 198         self._session = session  # type: _Session
 199         if attribute == _ProcessAttribute.PID or attribute == _ProcessAttribute.VPID:
 200             self._allowed_value_types = [int, str]  # type: list[type]
 201         else:
 202             self._allowed_value_types = [int]  # type: list[type]
 203
 204     def _call_client(self, cmd_name, value):
 205         # type: (str, Union[int, str]) -> None
 206         if type(value) not in self._allowed_value_types:
 207             raise TypeError(
 208                 "Value of type `{value_type}` is not allowed for process attribute {attribute_name}".format(
 209                     value_type=type(value).__name__,
 210                     attribute_name=self._tracked_attribute.name,
 211                 )
 212             )
 213
 214         process_attribute_option_name = _get_process_attribute_option_name(
 215             self._tracked_attribute
 216         )
 217         domain_name = _get_domain_option_name(self._domain)
 218         self._client._run_cmd(
 219             "{cmd_name} --session '{session_name}' --{domain_name} --{tracked_attribute_name} {value}".format(
 220                 cmd_name=cmd_name,
 221                 session_name=self._session.name,
 222                 domain_name=domain_name,
 223                 tracked_attribute_name=process_attribute_option_name,
 224                 value=value,
 225             )
 226         )
 227
 228     def track(self, value):
 229         # type: (Union[int, str]) -> None
 230         self._call_client("track", value)
 231
 232     def untrack(self, value):
 233         # type: (Union[int, str]) -> None
 234         self._call_client("untrack", value)
 235
 236
 237 class _Session(lttngctl.Session):
 238     def __init__(
 239         self,
 240         client,  # type: LTTngClient
 241         name,  # type: str
 242         output,  # type: Optional[lttngctl.SessionOutputLocation]
 243     ):
 244         self._client = client  # type: LTTngClient
 245         self._name = name  # type: str
 246         self._output = output  # type: Optional[lttngctl.SessionOutputLocation]
 247
 248     @property
 249     def name(self):
 250         # type: () -> str
 251         return self._name
 252
 253     def add_channel(
 254         self,
 255         domain,
 256         channel_name=None,
 257         buffer_sharing_policy=lttngctl.BufferSharingPolicy.PerUID,
 258     ):
 259         # type: (lttngctl.TracingDomain, Optional[str], lttngctl.BufferSharingPolicy) -> lttngctl.Channel
 260         channel_name = lttngctl.Channel._generate_name()
 261         domain_option_name = _get_domain_option_name(domain)
 262         self._client._run_cmd(
 263             "enable-channel --session '{session_name}' --{domain_name} '{channel_name}' {buffer_sharing_policy}".format(
 264                 session_name=self.name,
 265                 domain_name=domain_option_name,
 266                 channel_name=channel_name,
 267                 buffer_sharing_policy="--buffers-uid"
 268                 if buffer_sharing_policy == lttngctl.BufferSharingPolicy.PerUID
 269                 else "--buffers-pid",
 270             )
 271         )
 272         return _Channel(self._client, channel_name, domain, self)
 273
 274     def add_context(self, context_type):
 275         # type: (lttngctl.ContextType) -> None
 276         pass
 277
 278     @property
 279     def output(self):
 280         # type: () -> "Optional[Type[lttngctl.SessionOutputLocation]]"
 281         return self._output  # type: ignore
 282
 283     def start(self):
 284         # type: () -> None
 285         self._client._run_cmd("start '{session_name}'".format(session_name=self.name))
 286
 287     def stop(self):
 288         # type: () -> None
 289         self._client._run_cmd("stop '{session_name}'".format(session_name=self.name))
 290
 291     def destroy(self):
 292         # type: () -> None
 293         self._client._run_cmd("destroy '{session_name}'".format(session_name=self.name))
 294
 295     def rotate(self, wait=True):
 296         # type: (bool) -> None
 297         self._client.rotate_session_by_name(self.name, wait)
 298
 299     @property
 300     def is_active(self):
 301         # type: () -> bool
 302         list_session_xml = self._client._run_cmd(
 303             "list '{session_name}'".format(session_name=self.name),
 304             LTTngClient.CommandOutputFormat.MI_XML,
 305         )
 306
 307         root = xml.etree.ElementTree.fromstring(list_session_xml)
 308         command_output = LTTngClient._mi_find_in_element(root, "output")
 309         sessions = LTTngClient._mi_find_in_element(command_output, "sessions")
 310         session_mi = LTTngClient._mi_find_in_element(sessions, "session")
 311
 312         enabled_text = LTTngClient._mi_find_in_element(session_mi, "enabled").text
 313         if enabled_text not in ["true", "false"]:
 314             raise InvalidMI(
 315                 "Expected boolean value in element '{}': value='{}'".format(
 316                     session_mi.tag, enabled_text
 317                 )
 318             )
 319
 320         return enabled_text == "true"
 321
 322     @property
 323     def kernel_pid_process_attribute_tracker(self):
 324         # type: () -> Type[lttngctl.ProcessIDProcessAttributeTracker]
 325         return _ProcessAttributeTracker(self._client, _ProcessAttribute.PID, lttngctl.TracingDomain.Kernel, self)  # type: ignore
 326
 327     @property
 328     def kernel_vpid_process_attribute_tracker(self):
 329         # type: () -> Type[lttngctl.VirtualProcessIDProcessAttributeTracker]
 330         return _ProcessAttributeTracker(self._client, _ProcessAttribute.VPID, lttngctl.TracingDomain.Kernel, self)  # type: ignore
 331
 332     @property
 333     def user_vpid_process_attribute_tracker(self):
 334         # type: () -> Type[lttngctl.VirtualProcessIDProcessAttributeTracker]
 335         return _ProcessAttributeTracker(self._client, _ProcessAttribute.VPID, lttngctl.TracingDomain.User, self)  # type: ignore
 336
 337     @property
 338     def kernel_gid_process_attribute_tracker(self):
 339         # type: () -> Type[lttngctl.GroupIDProcessAttributeTracker]
 340         return _ProcessAttributeTracker(self._client, _ProcessAttribute.GID, lttngctl.TracingDomain.Kernel, self)  # type: ignore
 341
 342     @property
 343     def kernel_vgid_process_attribute_tracker(self):
 344         # type: () -> Type[lttngctl.VirtualGroupIDProcessAttributeTracker]
 345         return _ProcessAttributeTracker(self._client, _ProcessAttribute.VGID, lttngctl.TracingDomain.Kernel, self)  # type: ignore
 346
 347     @property
 348     def user_vgid_process_attribute_tracker(self):
 349         # type: () -> Type[lttngctl.VirtualGroupIDProcessAttributeTracker]
 350         return _ProcessAttributeTracker(self._client, _ProcessAttribute.VGID, lttngctl.TracingDomain.User, self)  # type: ignore
 351
 352     @property
 353     def kernel_uid_process_attribute_tracker(self):
 354         # type: () -> Type[lttngctl.UserIDProcessAttributeTracker]
 355         return _ProcessAttributeTracker(self._client, _ProcessAttribute.UID, lttngctl.TracingDomain.Kernel, self)  # type: ignore
 356
 357     @property
 358     def kernel_vuid_process_attribute_tracker(self):
 359         # type: () -> Type[lttngctl.VirtualUserIDProcessAttributeTracker]
 360         return _ProcessAttributeTracker(self._client, _ProcessAttribute.VUID, lttngctl.TracingDomain.Kernel, self)  # type: ignore
 361
 362     @property
 363     def user_vuid_process_attribute_tracker(self):
 364         # type: () -> Type[lttngctl.VirtualUserIDProcessAttributeTracker]
 365         return _ProcessAttributeTracker(self._client, _ProcessAttribute.VUID, lttngctl.TracingDomain.User, self)  # type: ignore
 366
 367
 368 class LTTngClientError(lttngctl.ControlException):
 369     def __init__(
 370         self,
 371         command_args,  # type: str
 372         error_output,  # type: str
 373     ):
 374         self._command_args = command_args  # type: str
 375         self._output = error_output  # type: str
 376
 377
 378 class LTTngClient(logger._Logger, lttngctl.Controller):
 379     """
 380     Implementation of a LTTngCtl Controller that uses the `lttng` client as a back-end.
 381     """
 382
 383     class CommandOutputFormat(enum.Enum):
 384         MI_XML = 0
 385         HUMAN = 1
 386
 387     _MI_NS = "{https://lttng.org/xml/ns/lttng-mi}"
 388
 389     def __init__(
 390         self,
 391         test_environment,  # type: environment._Environment
 392         log,  # type: Optional[Callable[[str], None]]
 393     ):
 394         logger._Logger.__init__(self, log)
 395         self._environment = test_environment  # type: environment._Environment
 396
 397     @staticmethod
 398     def _namespaced_mi_element(property):
 399         # type: (str) -> str
 400         return LTTngClient._MI_NS + property
 401
 402     def _run_cmd(self, command_args, output_format=CommandOutputFormat.MI_XML):
 403         # type: (str, CommandOutputFormat) -> str
 404         """
 405         Invoke the `lttng` client with a set of arguments. The command is
 406         executed in the context of the client's test environment.
 407         """
 408         args = [str(self._environment.lttng_client_path)]  # type: list[str]
 409         if output_format == LTTngClient.CommandOutputFormat.MI_XML:
 410             args.extend(["--mi", "xml"])
 411
 412         args.extend(shlex.split(command_args))
 413
 414         self._log("lttng {command_args}".format(command_args=command_args))
 415
 416         client_env = os.environ.copy()  # type: dict[str, str]
 417         client_env["LTTNG_HOME"] = str(self._environment.lttng_home_location)
 418
 419         process = subprocess.Popen(
 420             args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, env=client_env
 421         )
 422
 423         out = process.communicate()[0]
 424
 425         if process.returncode != 0:
 426             decoded_output = out.decode("utf-8")
 427             for error_line in decoded_output.splitlines():
 428                 self._log(error_line)
 429             raise LTTngClientError(command_args, decoded_output)
 430         else:
 431             return out.decode("utf-8")
 432
 433     def create_session(self, name=None, output=None):
 434         # type: (Optional[str], Optional[lttngctl.SessionOutputLocation]) -> lttngctl.Session
 435         name = name if name else lttngctl.Session._generate_name()
 436
 437         if isinstance(output, lttngctl.LocalSessionOutputLocation):
 438             output_option = "--output {output_path}".format(output_path=output.path)
 439         elif output is None:
 440             output_option = "--no-output"
 441         else:
 442             raise TypeError("LTTngClient only supports local or no output")
 443
 444         self._run_cmd(
 445             "create '{session_name}' {output_option}".format(
 446                 session_name=name, output_option=output_option
 447             )
 448         )
 449         return _Session(self, name, output)
 450
 451     def start_session_by_name(self, name):
 452         # type: (str) -> None
 453         self._run_cmd("start '{session_name}'".format(session_name=name))
 454
 455     def start_session_by_glob_pattern(self, pattern):
 456         # type: (str) -> None
 457         self._run_cmd("start --glob '{pattern}'".format(pattern=pattern))
 458
 459     def start_sessions_all(self):
 460         # type: () -> None
 461         self._run_cmd("start --all")
 462
 463     def stop_session_by_name(self, name):
 464         # type: (str) -> None
 465         self._run_cmd("stop '{session_name}'".format(session_name=name))
 466
 467     def stop_session_by_glob_pattern(self, pattern):
 468         # type: (str) -> None
 469         self._run_cmd("stop --glob '{pattern}'".format(pattern=pattern))
 470
 471     def stop_sessions_all(self):
 472         # type: () -> None
 473         self._run_cmd("stop --all")
 474
 475     def destroy_session_by_name(self, name):
 476         # type: (str) -> None
 477         self._run_cmd("destroy '{session_name}'".format(session_name=name))
 478
 479     def destroy_session_by_glob_pattern(self, pattern):
 480         # type: (str) -> None
 481         self._run_cmd("destroy --glob '{pattern}'".format(pattern=pattern))
 482
 483     def destroy_sessions_all(self):
 484         # type: () -> None
 485         self._run_cmd("destroy --all")
 486
 487     def rotate_session_by_name(self, name, wait=True):
 488         self._run_cmd(
 489             "rotate '{session_name}' {wait_option}".format(
 490                 session_name=name, wait_option="-n" if wait is False else ""
 491             )
 492         )
 493
 494     def schedule_size_based_rotation(self, name, size_bytes):
 495         # type (str, int) -> None
 496         self._run_cmd(
 497             "enable-rotation --session '{session_name}' --size {size}".format(
 498                 session_name=name, size=size_bytes
 499             )
 500         )
 501
 502     def schedule_time_based_rotation(self, name, period_seconds):
 503         # type (str, int) -> None
 504         self._run_cmd(
 505             "enable-rotation --session '{session_name}' --timer {period_seconds}s".format(
 506                 session_name=name, period_seconds=period_seconds
 507             )
 508         )
 509
 510     @staticmethod
 511     def _mi_find_in_element(element, sub_element_name):
 512         # type: (xml.etree.ElementTree.Element, str) -> xml.etree.ElementTree.Element
 513         result = element.find(LTTngClient._namespaced_mi_element(sub_element_name))
 514         if result is None:
 515             raise InvalidMI(
 516                 "Failed to find element '{}' within command MI element '{}'".format(
 517                     element.tag, sub_element_name
 518                 )
 519             )
 520
 521         return result
 522
 523     def list_sessions(self):
 524         # type () -> List[Session]
 525         list_sessions_xml = self._run_cmd(
 526             "list", LTTngClient.CommandOutputFormat.MI_XML
 527         )
 528
 529         root = xml.etree.ElementTree.fromstring(list_sessions_xml)
 530         command_output = self._mi_find_in_element(root, "output")
 531         sessions = self._mi_find_in_element(command_output, "sessions")
 532
 533         ctl_sessions = []  # type: list[lttngctl.Session]
 534
 535         for session_mi in sessions:
 536             name = self._mi_find_in_element(session_mi, "name").text
 537             path = self._mi_find_in_element(session_mi, "path").text
 538
 539             if name is None:
 540                 raise InvalidMI(
 541                     "Invalid empty 'name' element in '{}'".format(session_mi.tag)
 542                 )
 543             if path is None:
 544                 raise InvalidMI(
 545                     "Invalid empty 'path' element in '{}'".format(session_mi.tag)
 546                 )
 547             if not path.startswith("/"):
 548                 raise Unsupported(
 549                     "{} does not support non-local session outputs".format(type(self))
 550                 )
 551
 552             ctl_sessions.append(
 553                 _Session(self, name, lttngctl.LocalSessionOutputLocation(path))
 554             )
 555
 556         return ctl_sessions