Revert "[Fuchsia] Remove system logging logic from RunPackage()"

This reverts commit d6fe2c47.

Reason for revert: This causes us to lose backtraces from crashes under
the runner scripts. Although the output can be re-enabled via the
--system-log-file option, that bypasses symbolization so is also not
useful.

Original change's description:
> [Fuchsia] Remove system logging logic from RunPackage()
> 
> Previously runner scripts implemented system logs collection in two
> enabled by default, but could be turned off with --exclude-system-logs.
> (2) was disabled by default, but can be turned on with
> --system-log-file. (2) is enabled on the bots now. This CL removes (1)
> in order to remove noice from runner script output and to simplify code.
> When necessary --system-log-file=- flag can be used to get the previous
> behavior.
> 
> places: (1) in RunPackage, and (2) in Target implementations. (1) was
> Change-Id: I787ff4d8ac54f5c0cd13d0e9d8a71aad05cdd45c
> Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/1674668
> Commit-Queue: Sergey Ulanov <sergeyu@chromium.org>
> Commit-Queue: Kevin Marshall <kmarshall@chromium.org>
> Reviewed-by: Kevin Marshall <kmarshall@chromium.org>
> Cr-Commit-Position: refs/heads/master@{#672180}

TBR=kmarshall@chromium.org,sergeyu@chromium.org
Bug: 984993

# Not skipping CQ checks because original CL landed > 1 day ago.

Change-Id: Ib84af3efcb5c5859938328977a2aecb4a5428422
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/1707051Reviewed-by: Wez <wez@chromium.org>
Commit-Queue: Wez <wez@chromium.org>
Cr-Commit-Position: refs/heads/master@{#678524}

build/fuchsia/common_args.py

@@ -55,6 +55,10 @@ def AddCommonArgs(arg_parser):
   common_args.add_argument('--system-log-file',
                            help='File to write system logs to. Specify - to '
                                 'log to stdout.')
+  common_args.add_argument('--exclude-system-logs',
+                           action='store_false',
+                           dest='include_system_logs',
+                           help='Do not show system log data.')
   common_args.add_argument('--verbose', '-v', default=False,
                            action='store_true',
                            help='Enable debug-level logging.')
@@ -100,8 +104,7 @@ def GetDeploymentTargetForArgs(args):
 
   if not args.device:
     # KVM is required on x64 test bots.
-    require_kvm = hasattr(args, "test_launcher_bot_mode") and \
-        args.test_launcher_bot_mode and args.target_cpu == "x64"
+    require_kvm = args.test_launcher_bot_mode and args.target_cpu == "x64"
 
     return QemuTarget(output_dir=args.output_directory,
                       target_cpu=args.target_cpu,

build/fuchsia/run_package.py

@@ -28,6 +28,87 @@ FAR = os.path.join(common.SDK_ROOT, 'tools', 'far')
 _JOIN_TIMEOUT_SECS = 5
 
 
+def _AttachKernelLogReader(target):
+  """Attaches a kernel log reader as a long-running SSH task."""
+
+  logging.info('Attaching kernel logger.')
+  return target.RunCommandPiped(['dlog', '-f'], stdin=open(os.devnull, 'r'),
+                                stdout=subprocess.PIPE)
+
+
+class MergedInputStream(object):
+  """Merges a number of input streams into a UNIX pipe on a dedicated thread.
+  Terminates when the file descriptor of the primary stream (the first in
+  the sequence) is closed."""
+
+  def __init__(self, streams):
+    assert len(streams) > 0
+    self._streams = streams
+    self._output_stream = None
+    self._thread = None
+
+  def Start(self):
+    """Returns a pipe to the merged output stream."""
+
+    read_pipe, write_pipe = os.pipe()
+
+    # Disable buffering for the stream to make sure there is no delay in logs.
+    self._output_stream = os.fdopen(write_pipe, 'w', 0)
+    self._thread = threading.Thread(target=self._Run)
+    self._thread.start();
+
+    return os.fdopen(read_pipe, 'r')
+
+  def _Run(self):
+    streams_by_fd = {}
+    primary_fd = self._streams[0].fileno()
+    for s in self._streams:
+      streams_by_fd[s.fileno()] = s
+
+    # Set when the primary FD is closed. Input from other FDs will continue to
+    # be processed until select() runs dry.
+    flush = False
+
+    # The lifetime of the MergedInputStream is bound to the lifetime of
+    # |primary_fd|.
+    while primary_fd:
+      # When not flushing: block until data is read or an exception occurs.
+      rlist, _, xlist = select.select(streams_by_fd, [], streams_by_fd)
+
+      if len(rlist) == 0 and flush:
+        break
+
+      for fileno in xlist:
+        del streams_by_fd[fileno]
+        if fileno == primary_fd:
+          primary_fd = None
+
+      for fileno in rlist:
+        line = streams_by_fd[fileno].readline()
+        if line:
+          self._output_stream.write(line + '\n')
+        else:
+          del streams_by_fd[fileno]
+          if fileno == primary_fd:
+            primary_fd = None
+
+    # Flush the streams by executing nonblocking reads from the input file
+    # descriptors until no more data is available,  or all the streams are
+    # closed.
+    while streams_by_fd:
+      rlist, _, _ = select.select(streams_by_fd, [], [], 0)
+
+      if not rlist:
+        break
+
+      for fileno in rlist:
+        line = streams_by_fd[fileno].readline()
+        if line:
+          self._output_stream.write(line + '\n')
+        else:
+          del streams_by_fd[fileno]
+
+
 def _GetComponentUri(package_name):
   return 'fuchsia-pkg://fuchsia.com/%s#meta/%s.cmx' % (package_name,
                                                        package_name)
@@ -53,10 +134,23 @@ class RunPackageArgs:
   def FromCommonArgs(args):
     run_package_args = RunPackageArgs()
     run_package_args.install_only = args.install_only
+    run_package_args.system_logging = args.include_system_logs
     run_package_args.target_staging_path = args.target_staging_path
     return run_package_args
 
 
+def _DrainStreamToStdout(stream, quit_event):
+  """Outputs the contents of |stream| until |quit_event| is set."""
+
+  while not quit_event.is_set():
+    rlist, _, _ = select.select([ stream ], [], [], 0.1)
+    if rlist:
+      line = rlist[0].readline()
+      if not line:
+        return
+      print(line.rstrip())
+
+
 def RunPackage(output_dir, target, package_path, package_name,
                package_deps, package_args, args):
   """Installs the Fuchsia package at |package_path| on the target,
@@ -70,36 +164,67 @@ def RunPackage(output_dir, target, package_path, package_name,
   args: Structure of arguments to configure how the package will be run.
 
   Returns the exit code of the remote package process."""
-  target.InstallPackage(package_path, package_name, package_deps)
-
-  if args.install_only:
-    logging.info('Installation complete.')
-    return
-
-  logging.info('Running application.')
-  command = ['run', _GetComponentUri(package_name)] + package_args
-  process = target.RunCommandPiped(command,
-                                   stdin=open(os.devnull, 'r'),
-                                   stdout=subprocess.PIPE,
-                                   stderr=subprocess.STDOUT)
-
-  # Run the log data through the symbolizer process.
-  build_ids_paths = map(
-      lambda package_path: os.path.join(
-          os.path.dirname(package_path), 'ids.txt'),
-      [package_path] + package_deps)
-  output_stream = SymbolizerFilter(process.stdout, build_ids_paths)
-
-  for next_line in output_stream:
-    print(next_line.rstrip())
-
-  process.wait()
-  if process.returncode == 0:
-    logging.info('Process exited normally with status code 0.')
-  else:
-    # The test runner returns an error status code if *any* tests fail,
-    # so we should proceed anyway.
-    logging.warning('Process exited with status code %d.' %
-                    process.returncode)
+
+  system_logger = (
+      _AttachKernelLogReader(target) if args.system_logging else None)
+  try:
+    if system_logger:
+      # Spin up a thread to asynchronously dump the system log to stdout
+      # for easier diagnoses of early, pre-execution failures.
+      log_output_quit_event = multiprocessing.Event()
+      log_output_thread = threading.Thread(
+          target=lambda: _DrainStreamToStdout(system_logger.stdout,
+                                              log_output_quit_event))
+      log_output_thread.daemon = True
+      log_output_thread.start()
+
+    target.InstallPackage(package_path, package_name, package_deps)
+
+    if system_logger:
+      log_output_quit_event.set()
+      log_output_thread.join(timeout=_JOIN_TIMEOUT_SECS)
+
+    if args.install_only:
+      logging.info('Installation complete.')
+      return
+
+    logging.info('Running application.')
+    command = ['run', _GetComponentUri(package_name)] + package_args
+    process = target.RunCommandPiped(command,
+                                     stdin=open(os.devnull, 'r'),
+                                     stdout=subprocess.PIPE,
+                                     stderr=subprocess.STDOUT)
+
+    if system_logger:
+      output_stream = MergedInputStream([process.stdout,
+                                         system_logger.stdout]).Start()
+    else:
+      output_stream = process.stdout
+
+    # Run the log data through the symbolizer process.
+    build_ids_paths = map(
+        lambda package_path: os.path.join(
+            os.path.dirname(package_path), 'ids.txt'),
+        [package_path] + package_deps)
+    output_stream = SymbolizerFilter(output_stream, build_ids_paths)
+
+    for next_line in output_stream:
+      print(next_line.rstrip())
+
+    process.wait()
+    if process.returncode == 0:
+      logging.info('Process exited normally with status code 0.')
+    else:
+      # The test runner returns an error status code if *any* tests fail,
+      # so we should proceed anyway.
+      logging.warning('Process exited with status code %d.' %
+                      process.returncode)
+
+  finally:
+    if system_logger:
+      logging.info('Terminating kernel log reader.')
+      log_output_quit_event.set()
+      log_output_thread.join()
+      system_logger.kill()
 
   return process.returncode