tests/utils/lttngtest/environment.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 types import FrameType
   9 from typing import Callable, Iterator, Optional, Tuple, List, Generator
  10 import sys
  11 import pathlib
  12 import signal
  13 import subprocess
  14 import shlex
  15 import shutil
  16 import os
  17 import queue
  18 import tempfile
  19 from . import logger
  20 import time
  21 import threading
  22 import contextlib
  23
  24
  25 class TemporaryDirectory:
  26     def __init__(self, prefix):
  27         # type: (str) -> None
  28         self._directory_path = tempfile.mkdtemp(prefix=prefix)
  29
  30     def __del__(self):
  31         shutil.rmtree(self._directory_path, ignore_errors=True)
  32
  33     @property
  34     def path(self):
  35         # type: () -> pathlib.Path
  36         return pathlib.Path(self._directory_path)
  37
  38
  39 class _SignalWaitQueue:
  40     """
  41     Utility class useful to wait for a signal before proceeding.
  42
  43     Simply register the `signal` method as the handler for the signal you are
  44     interested in and call `wait_for_signal` to wait for its reception.
  45
  46     Registering a signal:
  47         signal.signal(signal.SIGWHATEVER, queue.signal)
  48
  49     Waiting for the signal:
  50         queue.wait_for_signal()
  51     """
  52
  53     def __init__(self):
  54         self._queue = queue.Queue()  # type: queue.Queue
  55
  56     def signal(
  57         self,
  58         signal_number,
  59         frame,  # type: Optional[FrameType]
  60     ):
  61         self._queue.put_nowait(signal_number)
  62
  63     def wait_for_signal(self):
  64         self._queue.get(block=True)
  65
  66     @contextlib.contextmanager
  67     def intercept_signal(self, signal_number):
  68         # type: (int) -> Generator[None, None, None]
  69         original_handler = signal.getsignal(signal_number)
  70         signal.signal(signal_number, self.signal)
  71         try:
  72             yield
  73         except:
  74             # Restore the original signal handler and forward the exception.
  75             raise
  76         finally:
  77             signal.signal(signal_number, original_handler)
  78
  79
  80 class WaitTraceTestApplication:
  81     """
  82     Create an application that waits before tracing. This allows a test to
  83     launch an application, get its PID, and get it to start tracing when it
  84     has completed its setup.
  85     """
  86
  87     def __init__(
  88         self,
  89         binary_path,  # type: pathlib.Path
  90         event_count,  # type: int
  91         environment,  # type: Environment
  92         wait_time_between_events_us=0,  # type: int
  93     ):
  94         self._environment = environment  # type: Environment
  95         self._iteration_count = event_count
  96         # File that the application will wait to see before tracing its events.
  97         self._app_start_tracing_file_path = pathlib.Path(
  98             tempfile.mktemp(
  99                 prefix="app_",
 100                 suffix="_start_tracing",
 101                 dir=self._compat_open_path(environment.lttng_home_location),
 102             )
 103         )
 104         self._has_returned = False
 105
 106         test_app_env = os.environ.copy()
 107         test_app_env["LTTNG_HOME"] = str(environment.lttng_home_location)
 108         # Make sure the app is blocked until it is properly registered to
 109         # the session daemon.
 110         test_app_env["LTTNG_UST_REGISTER_TIMEOUT"] = "-1"
 111
 112         # File that the application will create to indicate it has completed its initialization.
 113         app_ready_file_path = tempfile.mktemp(
 114             prefix="app_",
 115             suffix="_ready",
 116             dir=self._compat_open_path(environment.lttng_home_location),
 117         )  # type: str
 118
 119         test_app_args = [str(binary_path)]
 120         test_app_args.extend(
 121             shlex.split(
 122                 "--iter {iteration_count} --sync-application-in-main-touch {app_ready_file_path} --sync-before-first-event {app_start_tracing_file_path} --wait {wait_time_between_events_us}".format(
 123                     iteration_count=self._iteration_count,
 124                     app_ready_file_path=app_ready_file_path,
 125                     app_start_tracing_file_path=self._app_start_tracing_file_path,
 126                     wait_time_between_events_us=wait_time_between_events_us,
 127                 )
 128             )
 129         )
 130
 131         self._process = subprocess.Popen(
 132             test_app_args,
 133             env=test_app_env,
 134         )  # type: subprocess.Popen
 135
 136         # Wait for the application to create the file indicating it has fully
 137         # initialized. Make sure the app hasn't crashed in order to not wait
 138         # forever.
 139         while True:
 140             if os.path.exists(app_ready_file_path):
 141                 break
 142
 143             if self._process.poll() is not None:
 144                 # Application has unexepectedly returned.
 145                 raise RuntimeError(
 146                     "Test application has unexepectedly returned during its initialization with return code `{return_code}`".format(
 147                         return_code=self._process.returncode
 148                     )
 149                 )
 150
 151             time.sleep(0.1)
 152
 153     def trace(self):
 154         # type: () -> None
 155         if self._process.poll() is not None:
 156             # Application has unexepectedly returned.
 157             raise RuntimeError(
 158                 "Test application has unexepectedly before tracing with return code `{return_code}`".format(
 159                     return_code=self._process.returncode
 160                 )
 161             )
 162         open(self._compat_open_path(self._app_start_tracing_file_path), mode="x")
 163
 164     def wait_for_exit(self):
 165         # type: () -> None
 166         if self._process.wait() != 0:
 167             raise RuntimeError(
 168                 "Test application has exit with return code `{return_code}`".format(
 169                     return_code=self._process.returncode
 170                 )
 171             )
 172         self._has_returned = True
 173
 174     @property
 175     def vpid(self):
 176         # type: () -> int
 177         return self._process.pid
 178
 179     @staticmethod
 180     def _compat_open_path(path):
 181         # type: (pathlib.Path) -> pathlib.Path | str
 182         """
 183         The builtin open() in python >= 3.6 expects a path-like object while
 184         prior versions expect a string or bytes object. Return the correct type
 185         based on the presence of the "__fspath__" attribute specified in PEP-519.
 186         """
 187         if hasattr(path, "__fspath__"):
 188             return path
 189         else:
 190             return str(path)
 191
 192     def __del__(self):
 193         if not self._has_returned:
 194             # This is potentially racy if the pid has been recycled. However,
 195             # we can't use pidfd_open since it is only available in python >= 3.9.
 196             self._process.kill()
 197             self._process.wait()
 198
 199
 200 class TraceTestApplication:
 201     """
 202     Create an application that emits events as soon as it is launched. In most
 203     scenarios, it is preferable to use a WaitTraceTestApplication.
 204     """
 205
 206     def __init__(self, binary_path, environment):
 207         # type: (pathlib.Path, Environment)
 208         self._environment = environment  # type: Environment
 209         self._has_returned = False
 210
 211         test_app_env = os.environ.copy()
 212         test_app_env["LTTNG_HOME"] = str(environment.lttng_home_location)
 213         # Make sure the app is blocked until it is properly registered to
 214         # the session daemon.
 215         test_app_env["LTTNG_UST_REGISTER_TIMEOUT"] = "-1"
 216
 217         test_app_args = [str(binary_path)]
 218
 219         self._process = subprocess.Popen(
 220             test_app_args, env=test_app_env
 221         )  # type: subprocess.Popen
 222
 223     def wait_for_exit(self):
 224         # type: () -> None
 225         if self._process.wait() != 0:
 226             raise RuntimeError(
 227                 "Test application has exit with return code `{return_code}`".format(
 228                     return_code=self._process.returncode
 229                 )
 230             )
 231         self._has_returned = True
 232
 233     def __del__(self):
 234         if not self._has_returned:
 235             # This is potentially racy if the pid has been recycled. However,
 236             # we can't use pidfd_open since it is only available in python >= 3.9.
 237             self._process.kill()
 238             self._process.wait()
 239
 240
 241 class ProcessOutputConsumer(threading.Thread, logger._Logger):
 242     def __init__(
 243         self,
 244         process,  # type: subprocess.Popen
 245         name,  # type: str
 246         log,  # type: Callable[[str], None]
 247     ):
 248         threading.Thread.__init__(self)
 249         self._prefix = name
 250         logger._Logger.__init__(self, log)
 251         self._process = process
 252
 253     def run(self):
 254         # type: () -> None
 255         while self._process.poll() is None:
 256             assert self._process.stdout
 257             line = self._process.stdout.readline().decode("utf-8").replace("\n", "")
 258             if len(line) != 0:
 259                 self._log("{prefix}: {line}".format(prefix=self._prefix, line=line))
 260
 261
 262 # Generate a temporary environment in which to execute a test.
 263 class _Environment(logger._Logger):
 264     def __init__(
 265         self,
 266         with_sessiond,  # type: bool
 267         log=None,  # type: Optional[Callable[[str], None]]
 268     ):
 269         super().__init__(log)
 270         signal.signal(signal.SIGTERM, self._handle_termination_signal)
 271         signal.signal(signal.SIGINT, self._handle_termination_signal)
 272
 273         # Assumes the project's hierarchy to this file is:
 274         # tests/utils/python/this_file
 275         self._project_root = (
 276             pathlib.Path(__file__).absolute().parents[3]
 277         )  # type: pathlib.Path
 278         self._lttng_home = TemporaryDirectory(
 279             "lttng_test_env_home"
 280         )  # type: Optional[TemporaryDirectory]
 281
 282         self._sessiond = (
 283             self._launch_lttng_sessiond() if with_sessiond else None
 284         )  # type: Optional[subprocess.Popen[bytes]]
 285
 286     @property
 287     def lttng_home_location(self):
 288         # type: () -> pathlib.Path
 289         if self._lttng_home is None:
 290             raise RuntimeError("Attempt to access LTTng home after clean-up")
 291         return self._lttng_home.path
 292
 293     @property
 294     def lttng_client_path(self):
 295         # type: () -> pathlib.Path
 296         return self._project_root / "src" / "bin" / "lttng" / "lttng"
 297
 298     def create_temporary_directory(self, prefix=None):
 299         # type: (Optional[str]) -> pathlib.Path
 300         # Simply return a path that is contained within LTTNG_HOME; it will
 301         # be destroyed when the temporary home goes out of scope.
 302         assert self._lttng_home
 303         return pathlib.Path(
 304             tempfile.mkdtemp(
 305                 prefix="tmp" if prefix is None else prefix,
 306                 dir=str(self._lttng_home.path),
 307             )
 308         )
 309
 310     # Unpack a list of environment variables from a string
 311     # such as "HELLO=is_it ME='/you/are/looking/for'"
 312     @staticmethod
 313     def _unpack_env_vars(env_vars_string):
 314         # type: (str) -> List[Tuple[str, str]]
 315         unpacked_vars = []
 316         for var in shlex.split(env_vars_string):
 317             equal_position = var.find("=")
 318             # Must have an equal sign and not end with an equal sign
 319             if equal_position == -1 or equal_position == len(var) - 1:
 320                 raise ValueError(
 321                     "Invalid sessiond environment variable: `{}`".format(var)
 322                 )
 323
 324             var_name = var[0:equal_position]
 325             var_value = var[equal_position + 1 :]
 326             # Unquote any paths
 327             var_value = var_value.replace("'", "")
 328             var_value = var_value.replace('"', "")
 329             unpacked_vars.append((var_name, var_value))
 330
 331         return unpacked_vars
 332
 333     def _launch_lttng_sessiond(self):
 334         # type: () -> Optional[subprocess.Popen]
 335         is_64bits_host = sys.maxsize > 2**32
 336
 337         sessiond_path = (
 338             self._project_root / "src" / "bin" / "lttng-sessiond" / "lttng-sessiond"
 339         )
 340         consumerd_path_option_name = "--consumerd{bitness}-path".format(
 341             bitness="64" if is_64bits_host else "32"
 342         )
 343         consumerd_path = (
 344             self._project_root / "src" / "bin" / "lttng-consumerd" / "lttng-consumerd"
 345         )
 346
 347         no_sessiond_var = os.environ.get("TEST_NO_SESSIOND")
 348         if no_sessiond_var and no_sessiond_var == "1":
 349             # Run test without a session daemon; the user probably
 350             # intends to run one under gdb for example.
 351             return None
 352
 353         # Setup the session daemon's environment
 354         sessiond_env_vars = os.environ.get("LTTNG_SESSIOND_ENV_VARS")
 355         sessiond_env = os.environ.copy()
 356         if sessiond_env_vars:
 357             self._log("Additional lttng-sessiond environment variables:")
 358             additional_vars = self._unpack_env_vars(sessiond_env_vars)
 359             for var_name, var_value in additional_vars:
 360                 self._log("  {name}={value}".format(name=var_name, value=var_value))
 361                 sessiond_env[var_name] = var_value
 362
 363         sessiond_env["LTTNG_SESSION_CONFIG_XSD_PATH"] = str(
 364             self._project_root / "src" / "common"
 365         )
 366
 367         assert self._lttng_home is not None
 368         sessiond_env["LTTNG_HOME"] = str(self._lttng_home.path)
 369
 370         wait_queue = _SignalWaitQueue()
 371         with wait_queue.intercept_signal(signal.SIGUSR1):
 372             self._log(
 373                 "Launching session daemon with LTTNG_HOME=`{home_dir}`".format(
 374                     home_dir=str(self._lttng_home.path)
 375                 )
 376             )
 377             process = subprocess.Popen(
 378                 [
 379                     str(sessiond_path),
 380                     consumerd_path_option_name,
 381                     str(consumerd_path),
 382                     "--sig-parent",
 383                 ],
 384                 stdout=subprocess.PIPE,
 385                 stderr=subprocess.STDOUT,
 386                 env=sessiond_env,
 387             )
 388
 389             if self._logging_function:
 390                 self._sessiond_output_consumer = ProcessOutputConsumer(
 391                     process, "lttng-sessiond", self._logging_function
 392                 )  # type: Optional[ProcessOutputConsumer]
 393                 self._sessiond_output_consumer.daemon = True
 394                 self._sessiond_output_consumer.start()
 395
 396             # Wait for SIGUSR1, indicating the sessiond is ready to proceed
 397             wait_queue.wait_for_signal()
 398
 399         return process
 400
 401     def _handle_termination_signal(self, signal_number, frame):
 402         # type: (int, Optional[FrameType]) -> None
 403         self._log(
 404             "Killed by {signal_name} signal, cleaning-up".format(
 405                 signal_name=signal.strsignal(signal_number)
 406             )
 407         )
 408         self._cleanup()
 409
 410     def launch_wait_trace_test_application(self, event_count):
 411         # type: (int) -> WaitTraceTestApplication
 412         """
 413         Launch an application that will wait before tracing `event_count` events.
 414         """
 415         return WaitTraceTestApplication(
 416             self._project_root
 417             / "tests"
 418             / "utils"
 419             / "testapp"
 420             / "gen-ust-events"
 421             / "gen-ust-events",
 422             event_count,
 423             self,
 424         )
 425
 426     def launch_trace_test_constructor_application(self):
 427         # type () -> TraceTestApplication
 428         """
 429         Launch an application that will trace from within constructors.
 430         """
 431         return TraceTestApplication(
 432             self._project_root
 433             / "tests"
 434             / "utils"
 435             / "testapp"
 436             / "gen-ust-events-constructor"
 437             / "gen-ust-events-constructor",
 438             self,
 439         )
 440
 441     # Clean-up managed processes
 442     def _cleanup(self):
 443         # type: () -> None
 444         if self._sessiond and self._sessiond.poll() is None:
 445             # The session daemon is alive; kill it.
 446             self._log(
 447                 "Killing session daemon (pid = {sessiond_pid})".format(
 448                     sessiond_pid=self._sessiond.pid
 449                 )
 450             )
 451
 452             self._sessiond.terminate()
 453             self._sessiond.wait()
 454             if self._sessiond_output_consumer:
 455                 self._sessiond_output_consumer.join()
 456                 self._sessiond_output_consumer = None
 457
 458             self._log("Session daemon killed")
 459             self._sessiond = None
 460
 461         self._lttng_home = None
 462
 463     def __del__(self):
 464         self._cleanup()
 465
 466
 467 @contextlib.contextmanager
 468 def test_environment(with_sessiond, log=None):
 469     # type: (bool, Optional[Callable[[str], None]]) -> Iterator[_Environment]
 470     env = _Environment(with_sessiond, log)
 471     try:
 472         yield env
 473     finally:
 474         env._cleanup()