"spack logs": print log files for packages (either partially built or installed) (#42202)

lib/spack/spack/cmd/logs.py

@@ -0,0 +1,78 @@
+# Copyright 2013-2024 Lawrence Livermore National Security, LLC and other
+# Spack Project Developers. See the top-level COPYRIGHT file for details.
+#
+# SPDX-License-Identifier: (Apache-2.0 OR MIT)
+
+import errno
+import gzip
+import os
+import shutil
+import sys
+
+import spack.cmd
+import spack.util.compression as compression
+from spack.cmd.common import arguments
+from spack.main import SpackCommandError
+
+description = "print out logs for packages"
+section = "basic"
+level = "long"
+
+
+def setup_parser(subparser):
+    arguments.add_common_arguments(subparser, ["spec"])
+
+
+def _dump_byte_stream_to_stdout(instream):
+    outstream = os.fdopen(sys.stdout.fileno(), "wb", closefd=False)
+
+    shutil.copyfileobj(instream, outstream)
+
+
+def dump_build_log(package):
+    with open(package.log_path, "rb") as f:
+        _dump_byte_stream_to_stdout(f)
+
+
+def _logs(cmdline_spec, concrete_spec):
+    if concrete_spec.installed:
+        log_path = concrete_spec.package.install_log_path
+    elif os.path.exists(concrete_spec.package.stage.path):
+        dump_build_log(concrete_spec.package)
+        return
+    else:
+        raise SpackCommandError(f"{cmdline_spec} is not installed or staged")
+
+    try:
+        compression_ext = compression.extension_from_file(log_path)
+        with open(log_path, "rb") as fstream:
+            if compression_ext == "gz":
+                # If the log file is compressed, wrap it with a decompressor
+                fstream = gzip.open(log_path, "rb")
+            elif compression_ext:
+                raise SpackCommandError(
+                    f"Unsupported storage format for {log_path}: {compression_ext}"
+                )
+
+            _dump_byte_stream_to_stdout(fstream)
+    except OSError as e:
+        if e.errno == errno.ENOENT:
+            raise SpackCommandError(f"No logs are available for {cmdline_spec}") from e
+        elif e.errno == errno.EPERM:
+            raise SpackCommandError(f"Permission error accessing {log_path}") from e
+        else:
+            raise
+
+
+def logs(parser, args):
+    specs = spack.cmd.parse_specs(args.spec)
+
+    if not specs:
+        raise SpackCommandError("You must supply a spec.")
+
+    if len(specs) != 1:
+        raise SpackCommandError("Too many specs. Supply only one.")
+
+    concrete_spec = spack.cmd.matching_spec_from_env(specs[0])
+
+    _logs(specs[0], concrete_spec)

lib/spack/spack/test/cmd/logs.py

@@ -0,0 +1,119 @@
+# Copyright 2013-2024 Lawrence Livermore National Security, LLC and other
+# Spack Project Developers. See the top-level COPYRIGHT file for details.
+#
+# SPDX-License-Identifier: (Apache-2.0 OR MIT)
+
+import gzip
+import os
+import sys
+import tempfile
+from contextlib import contextmanager
+from io import BytesIO, TextIOWrapper
+
+import pytest
+
+import spack
+from spack.main import SpackCommand
+
+logs = SpackCommand("logs")
+install = SpackCommand("install")
+
+
+@contextmanager
+def stdout_as_buffered_text_stream():
+    """Attempt to simulate "typical" interface for stdout when user is
+    running Spack/Python from terminal. "spack log" should not be run
+    for all possible cases of what stdout might look like, in
+    particular some programmatic redirections of stdout like StringIO
+    are not meant to be supported by this command; more-generally,
+    mechanisms that depend on decoding binary output prior to write
+    are not supported for "spack log".
+    """
+    original_stdout = sys.stdout
+
+    with tempfile.TemporaryFile(mode="w+b") as tf:
+        sys.stdout = TextIOWrapper(tf)
+        try:
+            yield tf
+        finally:
+            sys.stdout = original_stdout
+
+
+def _rewind_collect_and_decode(rw_stream):
+    rw_stream.seek(0)
+    return rw_stream.read().decode("utf-8")
+
+
+@pytest.fixture
+def disable_capture(capfd):
+    with capfd.disabled():
+        yield
+
+
+def test_logs_cmd_errors(install_mockery, mock_fetch, mock_archive, mock_packages):
+    spec = spack.spec.Spec("libelf").concretized()
+    assert not spec.installed
+
+    with pytest.raises(spack.main.SpackCommandError, match="is not installed or staged"):
+        logs("libelf")
+
+    with pytest.raises(spack.main.SpackCommandError, match="Too many specs"):
+        logs("libelf mpi")
+
+    install("libelf")
+    os.remove(spec.package.install_log_path)
+    with pytest.raises(spack.main.SpackCommandError, match="No logs are available"):
+        logs("libelf")
+
+
+def _write_string_to_path(string, path):
+    """Write a string to a file, preserving newline format in the string."""
+    with open(path, "wb") as f:
+        f.write(string.encode("utf-8"))
+
+
+def test_dump_logs(install_mockery, mock_fetch, mock_archive, mock_packages, disable_capture):
+    """Test that ``spack log`` can find (and print) the logs for partial
+    builds and completed installs.
+
+    Also make sure that for compressed logs, that we automatically
+    decompress them.
+    """
+    cmdline_spec = spack.spec.Spec("libelf")
+    concrete_spec = cmdline_spec.concretized()
+
+    # Sanity check, make sure this test is checking what we want: to
+    # start with
+    assert not concrete_spec.installed
+
+    stage_log_content = "test_log stage output\nanother line"
+    installed_log_content = "test_log install output\nhere to test multiple lines"
+
+    with concrete_spec.package.stage:
+        _write_string_to_path(stage_log_content, concrete_spec.package.log_path)
+        with stdout_as_buffered_text_stream() as redirected_stdout:
+            spack.cmd.logs._logs(cmdline_spec, concrete_spec)
+            assert _rewind_collect_and_decode(redirected_stdout) == stage_log_content
+
+    install("libelf")
+
+    # Sanity check: make sure a path is recorded, regardless of whether
+    # it exists (if it does exist, we will overwrite it with content
+    # in this test)
+    assert concrete_spec.package.install_log_path
+
+    with gzip.open(concrete_spec.package.install_log_path, "wb") as compressed_file:
+        bstream = BytesIO(installed_log_content.encode("utf-8"))
+        compressed_file.writelines(bstream)
+
+    with stdout_as_buffered_text_stream() as redirected_stdout:
+        spack.cmd.logs._logs(cmdline_spec, concrete_spec)
+        assert _rewind_collect_and_decode(redirected_stdout) == installed_log_content
+
+    with concrete_spec.package.stage:
+        _write_string_to_path(stage_log_content, concrete_spec.package.log_path)
+        # We re-create the stage, but "spack log" should ignore that
+        # if the package is installed
+        with stdout_as_buffered_text_stream() as redirected_stdout:
+            spack.cmd.logs._logs(cmdline_spec, concrete_spec)
+            assert _rewind_collect_and_decode(redirected_stdout) == installed_log_content