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(self, domain, channel_name=None):
 254         # type: (lttngctl.TracingDomain, Optional[str]) -> lttngctl.Channel
 255         channel_name = lttngctl.Channel._generate_name()
 256         domain_option_name = _get_domain_option_name(domain)
 257         self._client._run_cmd(
 258             "enable-channel --session '{session_name}' --{domain_name} '{channel_name}'".format(
 259                 session_name=self.name,
 260                 domain_name=domain_option_name,
 261                 channel_name=channel_name,
 262             )
 263         )
 264         return _Channel(self._client, channel_name, domain, self)
 265
 266     def add_context(self, context_type):
 267         # type: (lttngctl.ContextType) -> None
 268         pass
 269
 270     @property
 271     def output(self):
 272         # type: () -> "Optional[Type[lttngctl.SessionOutputLocation]]"
 273         return self._output  # type: ignore
 274
 275     def start(self):
 276         # type: () -> None
 277         self._client._run_cmd("start '{session_name}'".format(session_name=self.name))
 278
 279     def stop(self):
 280         # type: () -> None
 281         self._client._run_cmd("stop '{session_name}'".format(session_name=self.name))
 282
 283     def destroy(self):
 284         # type: () -> None
 285         self._client._run_cmd("destroy '{session_name}'".format(session_name=self.name))
 286
 287     @property
 288     def is_active(self):
 289         # type: () -> bool
 290         list_session_xml = self._client._run_cmd(
 291             "list '{session_name}'".format(session_name=self.name),
 292             LTTngClient.CommandOutputFormat.MI_XML,
 293         )
 294
 295         root = xml.etree.ElementTree.fromstring(list_session_xml)
 296         command_output = LTTngClient._mi_find_in_element(root, "output")
 297         sessions = LTTngClient._mi_find_in_element(command_output, "sessions")
 298         session_mi = LTTngClient._mi_find_in_element(sessions, "session")
 299
 300         enabled_text = LTTngClient._mi_find_in_element(session_mi, "enabled").text
 301         if enabled_text not in ["true", "false"]:
 302             raise InvalidMI(
 303                 "Expected boolean value in element '{}': value='{}'".format(
 304                     session_mi.tag, enabled_text
 305                 )
 306             )
 307
 308         return enabled_text == "true"
 309
 310     @property
 311     def kernel_pid_process_attribute_tracker(self):
 312         # type: () -> Type[lttngctl.ProcessIDProcessAttributeTracker]
 313         return _ProcessAttributeTracker(self._client, _ProcessAttribute.PID, lttngctl.TracingDomain.Kernel, self)  # type: ignore
 314
 315     @property
 316     def kernel_vpid_process_attribute_tracker(self):
 317         # type: () -> Type[lttngctl.VirtualProcessIDProcessAttributeTracker]
 318         return _ProcessAttributeTracker(self._client, _ProcessAttribute.VPID, lttngctl.TracingDomain.Kernel, self)  # type: ignore
 319
 320     @property
 321     def user_vpid_process_attribute_tracker(self):
 322         # type: () -> Type[lttngctl.VirtualProcessIDProcessAttributeTracker]
 323         return _ProcessAttributeTracker(self._client, _ProcessAttribute.VPID, lttngctl.TracingDomain.User, self)  # type: ignore
 324
 325     @property
 326     def kernel_gid_process_attribute_tracker(self):
 327         # type: () -> Type[lttngctl.GroupIDProcessAttributeTracker]
 328         return _ProcessAttributeTracker(self._client, _ProcessAttribute.GID, lttngctl.TracingDomain.Kernel, self)  # type: ignore
 329
 330     @property
 331     def kernel_vgid_process_attribute_tracker(self):
 332         # type: () -> Type[lttngctl.VirtualGroupIDProcessAttributeTracker]
 333         return _ProcessAttributeTracker(self._client, _ProcessAttribute.VGID, lttngctl.TracingDomain.Kernel, self)  # type: ignore
 334
 335     @property
 336     def user_vgid_process_attribute_tracker(self):
 337         # type: () -> Type[lttngctl.VirtualGroupIDProcessAttributeTracker]
 338         return _ProcessAttributeTracker(self._client, _ProcessAttribute.VGID, lttngctl.TracingDomain.User, self)  # type: ignore
 339
 340     @property
 341     def kernel_uid_process_attribute_tracker(self):
 342         # type: () -> Type[lttngctl.UserIDProcessAttributeTracker]
 343         return _ProcessAttributeTracker(self._client, _ProcessAttribute.UID, lttngctl.TracingDomain.Kernel, self)  # type: ignore
 344
 345     @property
 346     def kernel_vuid_process_attribute_tracker(self):
 347         # type: () -> Type[lttngctl.VirtualUserIDProcessAttributeTracker]
 348         return _ProcessAttributeTracker(self._client, _ProcessAttribute.VUID, lttngctl.TracingDomain.Kernel, self)  # type: ignore
 349
 350     @property
 351     def user_vuid_process_attribute_tracker(self):
 352         # type: () -> Type[lttngctl.VirtualUserIDProcessAttributeTracker]
 353         return _ProcessAttributeTracker(self._client, _ProcessAttribute.VUID, lttngctl.TracingDomain.User, self)  # type: ignore
 354
 355
 356 class LTTngClientError(lttngctl.ControlException):
 357     def __init__(
 358         self,
 359         command_args,  # type: str
 360         error_output,  # type: str
 361     ):
 362         self._command_args = command_args  # type: str
 363         self._output = error_output  # type: str
 364
 365
 366 class LTTngClient(logger._Logger, lttngctl.Controller):
 367     """
 368     Implementation of a LTTngCtl Controller that uses the `lttng` client as a back-end.
 369     """
 370
 371     class CommandOutputFormat(enum.Enum):
 372         MI_XML = 0
 373         HUMAN = 1
 374
 375     _MI_NS = "{https://lttng.org/xml/ns/lttng-mi}"
 376
 377     def __init__(
 378         self,
 379         test_environment,  # type: environment._Environment
 380         log,  # type: Optional[Callable[[str], None]]
 381     ):
 382         logger._Logger.__init__(self, log)
 383         self._environment = test_environment  # type: environment._Environment
 384
 385     @staticmethod
 386     def _namespaced_mi_element(property):
 387         # type: (str) -> str
 388         return LTTngClient._MI_NS + property
 389
 390     def _run_cmd(self, command_args, output_format=CommandOutputFormat.MI_XML):
 391         # type: (str, CommandOutputFormat) -> str
 392         """
 393         Invoke the `lttng` client with a set of arguments. The command is
 394         executed in the context of the client's test environment.
 395         """
 396         args = [str(self._environment.lttng_client_path)]  # type: list[str]
 397         if output_format == LTTngClient.CommandOutputFormat.MI_XML:
 398             args.extend(["--mi", "xml"])
 399
 400         args.extend(shlex.split(command_args))
 401
 402         self._log("lttng {command_args}".format(command_args=command_args))
 403
 404         client_env = os.environ.copy()  # type: dict[str, str]
 405         client_env["LTTNG_HOME"] = str(self._environment.lttng_home_location)
 406
 407         process = subprocess.Popen(
 408             args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, env=client_env
 409         )
 410
 411         out = process.communicate()[0]
 412
 413         if process.returncode != 0:
 414             decoded_output = out.decode("utf-8")
 415             for error_line in decoded_output.splitlines():
 416                 self._log(error_line)
 417             raise LTTngClientError(command_args, decoded_output)
 418         else:
 419             return out.decode("utf-8")
 420
 421     def create_session(self, name=None, output=None):
 422         # type: (Optional[str], Optional[lttngctl.SessionOutputLocation]) -> lttngctl.Session
 423         name = name if name else lttngctl.Session._generate_name()
 424
 425         if isinstance(output, lttngctl.LocalSessionOutputLocation):
 426             output_option = "--output {output_path}".format(output_path=output.path)
 427         elif output is None:
 428             output_option = "--no-output"
 429         else:
 430             raise TypeError("LTTngClient only supports local or no output")
 431
 432         self._run_cmd(
 433             "create '{session_name}' {output_option}".format(
 434                 session_name=name, output_option=output_option
 435             )
 436         )
 437         return _Session(self, name, output)
 438
 439     def start_session_by_name(self, name):
 440         # type: (str) -> None
 441         self._run_cmd("start '{session_name}'".format(session_name=name))
 442
 443     def start_session_by_glob_pattern(self, pattern):
 444         # type: (str) -> None
 445         self._run_cmd("start --glob '{pattern}'".format(pattern=pattern))
 446
 447     def start_sessions_all(self):
 448         # type: () -> None
 449         self._run_cmd("start --all")
 450
 451     def stop_session_by_name(self, name):
 452         # type: (str) -> None
 453         self._run_cmd("stop '{session_name}'".format(session_name=name))
 454
 455     def stop_session_by_glob_pattern(self, pattern):
 456         # type: (str) -> None
 457         self._run_cmd("stop --glob '{pattern}'".format(pattern=pattern))
 458
 459     def stop_sessions_all(self):
 460         # type: () -> None
 461         self._run_cmd("stop --all")
 462
 463     def destroy_session_by_name(self, name):
 464         # type: (str) -> None
 465         self._run_cmd("destroy '{session_name}'".format(session_name=name))
 466
 467     def destroy_session_by_glob_pattern(self, pattern):
 468         # type: (str) -> None
 469         self._run_cmd("destroy --glob '{pattern}'".format(pattern=pattern))
 470
 471     def destroy_sessions_all(self):
 472         # type: () -> None
 473         self._run_cmd("destroy --all")
 474
 475     @staticmethod
 476     def _mi_find_in_element(element, sub_element_name):
 477         # type: (xml.etree.ElementTree.Element, str) -> xml.etree.ElementTree.Element
 478         result = element.find(LTTngClient._namespaced_mi_element(sub_element_name))
 479         if result is None:
 480             raise InvalidMI(
 481                 "Failed to find element '{}' within command MI element '{}'".format(
 482                     element.tag, sub_element_name
 483                 )
 484             )
 485
 486         return result
 487
 488     def list_sessions(self):
 489         # type () -> List[Session]
 490         list_sessions_xml = self._run_cmd(
 491             "list", LTTngClient.CommandOutputFormat.MI_XML
 492         )
 493
 494         root = xml.etree.ElementTree.fromstring(list_sessions_xml)
 495         command_output = self._mi_find_in_element(root, "output")
 496         sessions = self._mi_find_in_element(command_output, "sessions")
 497
 498         ctl_sessions = []  # type: list[lttngctl.Session]
 499
 500         for session_mi in sessions:
 501             name = self._mi_find_in_element(session_mi, "name").text
 502             path = self._mi_find_in_element(session_mi, "path").text
 503
 504             if name is None:
 505                 raise InvalidMI(
 506                     "Invalid empty 'name' element in '{}'".format(session_mi.tag)
 507                 )
 508             if path is None:
 509                 raise InvalidMI(
 510                     "Invalid empty 'path' element in '{}'".format(session_mi.tag)
 511                 )
 512             if not path.startswith("/"):
 513                 raise Unsupported(
 514                     "{} does not support non-local session outputs".format(type(self))
 515                 )
 516
 517             ctl_sessions.append(
 518                 _Session(self, name, lttngctl.LocalSessionOutputLocation(path))
 519             )
 520
 521         return ctl_sessions