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