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