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         wait_before_exit=False,  # type: bool
  94         wait_before_exit_file_path=None,  # type: Optional[pathlib.Path]
  95     ):
  96         self._environment = environment  # type: Environment
  97         self._iteration_count = event_count
  98         # File that the application will wait to see before tracing its events.
  99         self._app_start_tracing_file_path = pathlib.Path(
 100             tempfile.mktemp(
 101                 prefix="app_",
 102                 suffix="_start_tracing",
 103                 dir=self._compat_open_path(environment.lttng_home_location),
 104             )
 105         )
 106         # File that the application will create when all events have been emitted.
 107         self._app_tracing_done_file_path = pathlib.Path(
 108             tempfile.mktemp(
 109                 prefix="app_",
 110                 suffix="_done_tracing",
 111                 dir=self._compat_open_path(environment.lttng_home_location),
 112             )
 113         )
 114
 115         if wait_before_exit and wait_before_exit_file_path is None:
 116             wait_before_exit_file_path = pathlib.Path(
 117                 tempfile.mktemp(
 118                     prefix="app_",
 119                     suffix="_exit",
 120                     dir=self._compat_open_path(environment.lttng_home_location),
 121                 )
 122             )
 123
 124         self._has_returned = False
 125
 126         test_app_env = os.environ.copy()
 127         test_app_env["LTTNG_HOME"] = str(environment.lttng_home_location)
 128         # Make sure the app is blocked until it is properly registered to
 129         # the session daemon.
 130         test_app_env["LTTNG_UST_REGISTER_TIMEOUT"] = "-1"
 131
 132         # File that the application will create to indicate it has completed its initialization.
 133         app_ready_file_path = tempfile.mktemp(
 134             prefix="app_",
 135             suffix="_ready",
 136             dir=self._compat_open_path(environment.lttng_home_location),
 137         )  # type: str
 138
 139         test_app_args = [str(binary_path)]
 140         test_app_args.extend(["--iter", str(event_count)])
 141         test_app_args.extend(
 142             ["--sync-application-in-main-touch", str(app_ready_file_path)]
 143         )
 144         test_app_args.extend(
 145             ["--sync-before-first-event", str(self._app_start_tracing_file_path)]
 146         )
 147         test_app_args.extend(
 148             ["--sync-before-exit-touch", str(self._app_tracing_done_file_path)]
 149         )
 150         if wait_time_between_events_us != 0:
 151             test_app_args.extend(["--wait", str(wait_time_between_events_us)])
 152
 153         self._process = subprocess.Popen(
 154             test_app_args,
 155             env=test_app_env,
 156             stdout=subprocess.PIPE,
 157             stderr=subprocess.STDOUT,
 158         )  # type: subprocess.Popen
 159
 160         # Wait for the application to create the file indicating it has fully
 161         # initialized. Make sure the app hasn't crashed in order to not wait
 162         # forever.
 163         self._wait_for_file_to_be_created(pathlib.Path(app_ready_file_path))
 164
 165     def _wait_for_file_to_be_created(self, sync_file_path):
 166         # type: (pathlib.Path) -> None
 167         while True:
 168             if os.path.exists(sync_file_path):
 169                 break
 170
 171             if self._process.poll() is not None:
 172                 # Application has unexepectedly returned.
 173                 raise RuntimeError(
 174                     "Test application has unexepectedly returned while waiting for synchronization file to be created: sync_file=`{sync_file}`, return_code=`{return_code}`".format(
 175                         sync_file=sync_file_path, return_code=self._process.returncode
 176                     )
 177                 )
 178
 179             time.sleep(0.001)
 180
 181     def trace(self):
 182         # type: () -> None
 183         if self._process.poll() is not None:
 184             # Application has unexepectedly returned.
 185             raise RuntimeError(
 186                 "Test application has unexepectedly before tracing with return code `{return_code}`".format(
 187                     return_code=self._process.returncode
 188                 )
 189             )
 190         open(self._compat_open_path(self._app_start_tracing_file_path), mode="x")
 191
 192     def wait_for_tracing_done(self):
 193         # type: () -> None
 194         self._wait_for_file_to_be_created(self._app_tracing_done_file_path)
 195
 196     def wait_for_exit(self):
 197         # type: () -> None
 198         if self._process.wait() != 0:
 199             raise RuntimeError(
 200                 "Test application has exit with return code `{return_code}`".format(
 201                     return_code=self._process.returncode
 202                 )
 203             )
 204         self._has_returned = True
 205
 206     @property
 207     def vpid(self):
 208         # type: () -> int
 209         return self._process.pid
 210
 211     @staticmethod
 212     def _compat_open_path(path):
 213         # type: (pathlib.Path) -> pathlib.Path | str
 214         """
 215         The builtin open() in python >= 3.6 expects a path-like object while
 216         prior versions expect a string or bytes object. Return the correct type
 217         based on the presence of the "__fspath__" attribute specified in PEP-519.
 218         """
 219         if hasattr(path, "__fspath__"):
 220             return path
 221         else:
 222             return str(path)
 223
 224     def __del__(self):
 225         if not self._has_returned:
 226             # This is potentially racy if the pid has been recycled. However,
 227             # we can't use pidfd_open since it is only available in python >= 3.9.
 228             self._process.kill()
 229             self._process.wait()
 230
 231
 232 class WaitTraceTestApplicationGroup:
 233     def __init__(
 234         self,
 235         environment,  # type: Environment
 236         application_count,  # type: int
 237         event_count,  # type: int
 238         wait_time_between_events_us=0,  # type: int
 239         wait_before_exit=False,  # type: bool
 240     ):
 241         self._wait_before_exit_file_path = (
 242             pathlib.Path(
 243                 tempfile.mktemp(
 244                     prefix="app_group_",
 245                     suffix="_exit",
 246                     dir=_WaitTraceTestApplication._compat_open_path(
 247                         environment.lttng_home_location
 248                     ),
 249                 )
 250             )
 251             if wait_before_exit
 252             else None
 253         )
 254
 255         self._apps = []
 256         self._consumers = []
 257         for i in range(application_count):
 258             new_app = environment.launch_wait_trace_test_application(
 259                 event_count,
 260                 wait_time_between_events_us,
 261                 wait_before_exit,
 262                 self._wait_before_exit_file_path,
 263             )
 264
 265             # Attach an output consumer to log the application's error output (if any).
 266             if environment._logging_function:
 267                 app_output_consumer = ProcessOutputConsumer(
 268                     new_app._process,
 269                     "app-{}".format(str(new_app.vpid)),
 270                     environment._logging_function,
 271                 )  # type: Optional[ProcessOutputConsumer]
 272                 app_output_consumer.daemon = True
 273                 app_output_consumer.start()
 274                 self._consumers.append(app_output_consumer)
 275
 276             self._apps.append(new_app)
 277
 278     def trace(self):
 279         # type: () -> None
 280         for app in self._apps:
 281             app.trace()
 282
 283     def exit(
 284         self, wait_for_apps=False  # type: bool
 285     ):
 286         if self._wait_before_exit_file_path is None:
 287             raise RuntimeError(
 288                 "Can't call exit on an application group created with `wait_before_exit=False`"
 289             )
 290
 291         # Wait for apps to have produced all of their events so that we can
 292         # cause the death of all apps to happen within a short time span.
 293         for app in self._apps:
 294             app.wait_for_tracing_done()
 295
 296         open(
 297             _WaitTraceTestApplication._compat_open_path(
 298                 self._wait_before_exit_file_path
 299             ),
 300             mode="x",
 301         )
 302         # Performed in two passes to allow tests to stress the unregistration of many applications.
 303         # Waiting for each app to exit turn-by-turn would defeat the purpose here.
 304         if wait_for_apps:
 305             for app in self._apps:
 306                 app.wait_for_exit()
 307
 308
 309 class _TraceTestApplication:
 310     """
 311     Create an application that emits events as soon as it is launched. In most
 312     scenarios, it is preferable to use a WaitTraceTestApplication.
 313     """
 314
 315     def __init__(self, binary_path, environment):
 316         # type: (pathlib.Path, Environment)
 317         self._environment = environment  # type: Environment
 318         self._has_returned = False
 319
 320         test_app_env = os.environ.copy()
 321         test_app_env["LTTNG_HOME"] = str(environment.lttng_home_location)
 322         # Make sure the app is blocked until it is properly registered to
 323         # the session daemon.
 324         test_app_env["LTTNG_UST_REGISTER_TIMEOUT"] = "-1"
 325
 326         test_app_args = [str(binary_path)]
 327
 328         self._process = subprocess.Popen(
 329             test_app_args, env=test_app_env
 330         )  # type: subprocess.Popen
 331
 332     def wait_for_exit(self):
 333         # type: () -> None
 334         if self._process.wait() != 0:
 335             raise RuntimeError(
 336                 "Test application has exit with return code `{return_code}`".format(
 337                     return_code=self._process.returncode
 338                 )
 339             )
 340         self._has_returned = True
 341
 342     def __del__(self):
 343         if not self._has_returned:
 344             # This is potentially racy if the pid has been recycled. However,
 345             # we can't use pidfd_open since it is only available in python >= 3.9.
 346             self._process.kill()
 347             self._process.wait()
 348
 349
 350 class ProcessOutputConsumer(threading.Thread, logger._Logger):
 351     def __init__(
 352         self,
 353         process,  # type: subprocess.Popen
 354         name,  # type: str
 355         log,  # type: Callable[[str], None]
 356     ):
 357         threading.Thread.__init__(self)
 358         self._prefix = name
 359         logger._Logger.__init__(self, log)
 360         self._process = process
 361
 362     def run(self):
 363         # type: () -> None
 364         while self._process.poll() is None:
 365             assert self._process.stdout
 366             line = self._process.stdout.readline().decode("utf-8").replace("\n", "")
 367             if len(line) != 0:
 368                 self._log("{prefix}: {line}".format(prefix=self._prefix, line=line))
 369
 370
 371 # Generate a temporary environment in which to execute a test.
 372 class _Environment(logger._Logger):
 373     def __init__(
 374         self,
 375         with_sessiond,  # type: bool
 376         log=None,  # type: Optional[Callable[[str], None]]
 377     ):
 378         super().__init__(log)
 379         signal.signal(signal.SIGTERM, self._handle_termination_signal)
 380         signal.signal(signal.SIGINT, self._handle_termination_signal)
 381
 382         # Assumes the project's hierarchy to this file is:
 383         # tests/utils/python/this_file
 384         self._project_root = (
 385             pathlib.Path(__file__).absolute().parents[3]
 386         )  # type: pathlib.Path
 387         self._lttng_home = TemporaryDirectory(
 388             "lttng_test_env_home"
 389         )  # type: Optional[TemporaryDirectory]
 390
 391         self._sessiond = (
 392             self._launch_lttng_sessiond() if with_sessiond else None
 393         )  # type: Optional[subprocess.Popen[bytes]]
 394
 395     @property
 396     def lttng_home_location(self):
 397         # type: () -> pathlib.Path
 398         if self._lttng_home is None:
 399             raise RuntimeError("Attempt to access LTTng home after clean-up")
 400         return self._lttng_home.path
 401
 402     @property
 403     def lttng_client_path(self):
 404         # type: () -> pathlib.Path
 405         return self._project_root / "src" / "bin" / "lttng" / "lttng"
 406
 407     def create_temporary_directory(self, prefix=None):
 408         # type: (Optional[str]) -> pathlib.Path
 409         # Simply return a path that is contained within LTTNG_HOME; it will
 410         # be destroyed when the temporary home goes out of scope.
 411         assert self._lttng_home
 412         return pathlib.Path(
 413             tempfile.mkdtemp(
 414                 prefix="tmp" if prefix is None else prefix,
 415                 dir=str(self._lttng_home.path),
 416             )
 417         )
 418
 419     # Unpack a list of environment variables from a string
 420     # such as "HELLO=is_it ME='/you/are/looking/for'"
 421     @staticmethod
 422     def _unpack_env_vars(env_vars_string):
 423         # type: (str) -> List[Tuple[str, str]]
 424         unpacked_vars = []
 425         for var in shlex.split(env_vars_string):
 426             equal_position = var.find("=")
 427             # Must have an equal sign and not end with an equal sign
 428             if equal_position == -1 or equal_position == len(var) - 1:
 429                 raise ValueError(
 430                     "Invalid sessiond environment variable: `{}`".format(var)
 431                 )
 432
 433             var_name = var[0:equal_position]
 434             var_value = var[equal_position + 1 :]
 435             # Unquote any paths
 436             var_value = var_value.replace("'", "")
 437             var_value = var_value.replace('"', "")
 438             unpacked_vars.append((var_name, var_value))
 439
 440         return unpacked_vars
 441
 442     def _launch_lttng_sessiond(self):
 443         # type: () -> Optional[subprocess.Popen]
 444         is_64bits_host = sys.maxsize > 2**32
 445
 446         sessiond_path = (
 447             self._project_root / "src" / "bin" / "lttng-sessiond" / "lttng-sessiond"
 448         )
 449         consumerd_path_option_name = "--consumerd{bitness}-path".format(
 450             bitness="64" if is_64bits_host else "32"
 451         )
 452         consumerd_path = (
 453             self._project_root / "src" / "bin" / "lttng-consumerd" / "lttng-consumerd"
 454         )
 455
 456         no_sessiond_var = os.environ.get("TEST_NO_SESSIOND")
 457         if no_sessiond_var and no_sessiond_var == "1":
 458             # Run test without a session daemon; the user probably
 459             # intends to run one under gdb for example.
 460             return None
 461
 462         # Setup the session daemon's environment
 463         sessiond_env_vars = os.environ.get("LTTNG_SESSIOND_ENV_VARS")
 464         sessiond_env = os.environ.copy()
 465         if sessiond_env_vars:
 466             self._log("Additional lttng-sessiond environment variables:")
 467             additional_vars = self._unpack_env_vars(sessiond_env_vars)
 468             for var_name, var_value in additional_vars:
 469                 self._log("  {name}={value}".format(name=var_name, value=var_value))
 470                 sessiond_env[var_name] = var_value
 471
 472         sessiond_env["LTTNG_SESSION_CONFIG_XSD_PATH"] = str(
 473             self._project_root / "src" / "common"
 474         )
 475
 476         assert self._lttng_home is not None
 477         sessiond_env["LTTNG_HOME"] = str(self._lttng_home.path)
 478
 479         wait_queue = _SignalWaitQueue()
 480         with wait_queue.intercept_signal(signal.SIGUSR1):
 481             self._log(
 482                 "Launching session daemon with LTTNG_HOME=`{home_dir}`".format(
 483                     home_dir=str(self._lttng_home.path)
 484                 )
 485             )
 486             process = subprocess.Popen(
 487                 [
 488                     str(sessiond_path),
 489                     consumerd_path_option_name,
 490                     str(consumerd_path),
 491                     "--sig-parent",
 492                 ],
 493                 stdout=subprocess.PIPE,
 494                 stderr=subprocess.STDOUT,
 495                 env=sessiond_env,
 496             )
 497
 498             if self._logging_function:
 499                 self._sessiond_output_consumer = ProcessOutputConsumer(
 500                     process, "lttng-sessiond", self._logging_function
 501                 )  # type: Optional[ProcessOutputConsumer]
 502                 self._sessiond_output_consumer.daemon = True
 503                 self._sessiond_output_consumer.start()
 504
 505             # Wait for SIGUSR1, indicating the sessiond is ready to proceed
 506             wait_queue.wait_for_signal()
 507
 508         return process
 509
 510     def _handle_termination_signal(self, signal_number, frame):
 511         # type: (int, Optional[FrameType]) -> None
 512         self._log(
 513             "Killed by {signal_name} signal, cleaning-up".format(
 514                 signal_name=signal.strsignal(signal_number)
 515             )
 516         )
 517         self._cleanup()
 518
 519     def launch_wait_trace_test_application(
 520         self,
 521         event_count,  # type: int
 522         wait_time_between_events_us=0,
 523         wait_before_exit=False,
 524         wait_before_exit_file_path=None,
 525     ):
 526         # type: (int, int, bool, Optional[pathlib.Path]) -> _WaitTraceTestApplication
 527         """
 528         Launch an application that will wait before tracing `event_count` events.
 529         """
 530         return _WaitTraceTestApplication(
 531             self._project_root
 532             / "tests"
 533             / "utils"
 534             / "testapp"
 535             / "gen-ust-events"
 536             / "gen-ust-events",
 537             event_count,
 538             self,
 539             wait_time_between_events_us,
 540             wait_before_exit,
 541             wait_before_exit_file_path,
 542         )
 543
 544     def launch_trace_test_constructor_application(self):
 545         # type () -> TraceTestApplication
 546         """
 547         Launch an application that will trace from within constructors.
 548         """
 549         return _TraceTestApplication(
 550             self._project_root
 551             / "tests"
 552             / "utils"
 553             / "testapp"
 554             / "gen-ust-events-constructor"
 555             / "gen-ust-events-constructor",
 556             self,
 557         )
 558
 559     # Clean-up managed processes
 560     def _cleanup(self):
 561         # type: () -> None
 562         if self._sessiond and self._sessiond.poll() is None:
 563             # The session daemon is alive; kill it.
 564             self._log(
 565                 "Killing session daemon (pid = {sessiond_pid})".format(
 566                     sessiond_pid=self._sessiond.pid
 567                 )
 568             )
 569
 570             self._sessiond.terminate()
 571             self._sessiond.wait()
 572             if self._sessiond_output_consumer:
 573                 self._sessiond_output_consumer.join()
 574                 self._sessiond_output_consumer = None
 575
 576             self._log("Session daemon killed")
 577             self._sessiond = None
 578
 579         self._lttng_home = None
 580
 581     def __del__(self):
 582         self._cleanup()
 583
 584
 585 @contextlib.contextmanager
 586 def test_environment(with_sessiond, log=None):
 587     # type: (bool, Optional[Callable[[str], None]]) -> Iterator[_Environment]
 588     env = _Environment(with_sessiond, log)
 589     try:
 590         yield env
 591     finally:
 592         env._cleanup()