xref: /aosp_15_r20/external/angle/build/fuchsia/test/log_manager.py (revision 8975f5c5ed3d1c378011245431ada316dfb6f244) 
1# Copyright 2022 The Chromium Authors
2# Use of this source code is governed by a BSD-style license that can be
3# found in the LICENSE file.
4"""Reads log data from a device."""
5
6import os
7import subprocess
8import sys
9
10from contextlib import AbstractContextManager
11from typing import Iterable, Optional, TextIO
12
13from common import run_continuous_ffx_command
14from ffx_integration import run_symbolizer
15
16
17class LogManager(AbstractContextManager):
18    """Handles opening and closing file streams for logging purposes."""
19
20    def __init__(self, logs_dir: Optional[str]) -> None:
21        self._logs_dir = logs_dir
22
23        # A dictionary with the log file path as the key and a file stream as
24        # value.
25        self._log_files = {}
26        self._log_procs = []
27        self._scoped_ffx_log = None
28
29    def __enter__(self):
30        return self
31
32    def is_logging_enabled(self) -> bool:
33        """Check whether logging is turned on."""
34
35        return self._logs_dir is not None
36
37    def add_log_process(self, process: subprocess.Popen) -> None:
38        """Register a logging process to LogManager to be killed at LogManager
39        teardown."""
40
41        self._log_procs.append(process)
42
43    def open_log_file(self, log_file_name: str) -> TextIO:
44        """Open a file stream with log_file_name in the logs directory."""
45
46        assert self._logs_dir, 'Logging directory is not specified.'
47        log_file_path = os.path.join(self._logs_dir, log_file_name)
48        log_file = open(log_file_path, 'w', buffering=1)
49        self._log_files[log_file_path] = log_file
50        return log_file
51
52    def __exit__(self, exc_type, exc_value, traceback):
53        """Stop all active logging instances."""
54        for proc in self._log_procs:
55            proc.kill()
56        for log in self._log_files.values():
57            log.close()
58        return False
59
60
61def start_system_log(log_manager: LogManager,
62                     log_to_stdout: bool,
63                     pkg_paths: Optional[Iterable[str]] = None,
64                     log_args: Optional[Iterable[str]] = None,
65                     target_id: Optional[str] = None) -> None:
66    """
67    Start system logging.
68
69    Args:
70        log_manager: A LogManager class that manages the log file and process.
71        log_to_stdout: If set to True, print logs directly to stdout.
72        pkg_paths: Path to the packages
73        log_args: Arguments forwarded to `ffx log` command.
74        target_id: Specify a target to use.
75    """
76
77    if not log_manager.is_logging_enabled() and not log_to_stdout:
78        return
79    symbol_paths = None
80    if pkg_paths:
81        symbol_paths = []
82
83        # Locate debug symbols for each package.
84        for pkg_path in pkg_paths:
85            assert os.path.isfile(pkg_path), '%s does not exist' % pkg_path
86            symbol_paths.append(
87                os.path.join(os.path.dirname(pkg_path), 'ids.txt'))
88
89    if log_to_stdout:
90        system_log = sys.stdout
91    else:
92        system_log = log_manager.open_log_file('system_log')
93    log_cmd = ['log', '--symbolize', 'off', '--no-color']
94    if log_args:
95        log_cmd.extend(log_args)
96    if symbol_paths:
97        log_proc = run_continuous_ffx_command(log_cmd,
98                                              target_id,
99                                              stdout=subprocess.PIPE)
100        log_manager.add_log_process(log_proc)
101        log_manager.add_log_process(
102            run_symbolizer(symbol_paths, log_proc.stdout, system_log))
103    else:
104        log_manager.add_log_process(
105            run_continuous_ffx_command(log_cmd, target_id, stdout=system_log))
106