Source code for spack.cmd.logs

# Copyright Spack Project Developers. See COPYRIGHT file for details.
#
# SPDX-License-Identifier: (Apache-2.0 OR MIT)

import argparse
import errno
import gzip
import io
import os
import shutil
import sys

import spack.cmd
import spack.spec
import spack.util.compression as compression
from spack.cmd.common import arguments
from spack.main import SpackCommandError

description = "print out logs for packages"
section = "query"
level = "long"




def setup_parser(subparser: argparse.ArgumentParser) -> None:
    arguments.add_common_arguments(subparser, ["spec"])



def _dump_byte_stream_to_stdout(instream: io.BufferedIOBase) -> None:
    # Reopen stdout in binary mode so we don't have to worry about encoding
    outstream = os.fdopen(sys.stdout.fileno(), "wb", closefd=False)
    shutil.copyfileobj(instream, outstream)


def _logs(cmdline_spec: spack.spec.Spec, concrete_spec: spack.spec.Spec):
    if concrete_spec.installed:
        log_path = concrete_spec.package.install_log_path
    elif os.path.exists(concrete_spec.package.stage.path):
        # TODO: `spack logs` can currently not show the logs while a package is being built, as the
        # combined log file is only written after the build is finished.
        log_path = concrete_spec.package.log_path
    else:
        raise SpackCommandError(f"{cmdline_spec} is not installed or staged")

    try:
        stream = open(log_path, "rb")
    except OSError as e:
        if e.errno == errno.ENOENT:
            raise SpackCommandError(f"No logs are available for {cmdline_spec}") from e
        raise SpackCommandError(f"Error reading logs for {cmdline_spec}: {e}") from e

    with stream as f:
        ext = compression.extension_from_magic_numbers_by_stream(f, decompress=False)
        if ext and ext != "gz":
            raise SpackCommandError(f"Unsupported storage format for {log_path}: {ext}")

        # If the log file is gzip compressed, wrap it with a decompressor
        _dump_byte_stream_to_stdout(gzip.GzipFile(fileobj=f) if ext == "gz" else f)




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)