# Copyright Spack Project Developers. See COPYRIGHT file for details.
#
# SPDX-License-Identifier: (Apache-2.0 OR MIT)
"""Tools to produce reports of spec installations or tests"""
import collections
import gzip
import os
import time
import traceback
from typing import Optional
import spack.error
reporter = None
report_file = None
Property = collections.namedtuple("Property", ["name", "value"])
[docs]
class Record(dict):
"""Data class that provides attr-style access to a dictionary
Attributes beginning with ``_`` are reserved for the Record class itself."""
def __getattr__(self, name):
# only called if no attribute exists
if name in self:
return self[name]
raise AttributeError(f"Record for {self.name} has no attribute {name}")
def __setattr__(self, name, value):
if name.startswith("_"):
super().__setattr__(name, value)
else:
self[name] = value
[docs]
class RequestRecord(Record):
"""Data class for recording outcomes for an entire DAG
Each BuildRequest in the installer and each root spec in a TestSuite generates a
RequestRecord. The ``packages`` list of the RequestRecord is a list of SpecRecord
objects recording individual data for each node in the Spec represented by the
RequestRecord.
These data classes are collated by the reporters in lib/spack/spack/reporters
"""
def __init__(self, spec):
super().__init__()
self._spec = spec
self.name = spec.name
self.nerrors = None
self.nfailures = None
self.npackages = None
self.time = None
self.timestamp = time.strftime("%a, %d %b %Y %H:%M:%S", time.gmtime())
self.properties = [
Property("architecture", spec.architecture)
# Property("compiler", spec.compiler),
]
self.packages = []
[docs]
def skip_installed(self):
"""Insert records for all nodes in the DAG that are no-ops for this request"""
for dep in filter(lambda x: x.installed or x.external, self._spec.traverse()):
record = InstallRecord(dep)
record.skip(msg="Spec external or already installed")
self.packages.append(record)
[docs]
def append_record(self, record):
self.packages.append(record)
[docs]
def summarize(self):
"""Construct request-level summaries of the individual records"""
self.npackages = len(self.packages)
self.nfailures = len([r for r in self.packages if r.result == "failure"])
self.nerrors = len([r for r in self.packages if r.result == "error"])
self.time = sum(float(r.elapsed_time or 0.0) for r in self.packages)
[docs]
class SpecRecord(Record):
"""Individual record for a single spec within a request"""
def __init__(self, spec):
super().__init__()
self._spec = spec
self._package = spec.package
self._start_time = None
self.name = spec.name
self.id = spec.dag_hash()
self.elapsed_time = None
[docs]
def start(self):
self._start_time = time.time()
[docs]
def skip(self, msg):
self.result = "skipped"
self.elapsed_time = 0.0
self.message = msg
[docs]
def fetch_log(self, log_path: Optional[str] = None) -> str:
"""Fetch the log for this spec record. Subclasses should override."""
return ""
[docs]
def fail(self, exc, log_path: Optional[str] = None):
"""Record failure based on exception type
Errors wrapped by spack.error.InstallError are "failures"
Other exceptions are "errors".
"""
if isinstance(exc, spack.error.InstallError):
self.result = "failure"
self.message = exc.message or "Installation failure"
self.exception = exc.traceback
else:
self.result = "error"
self.message = str(exc) or "Unknown error"
self.exception = traceback.format_exc()
self.stdout = self.fetch_log(log_path) + self.message
assert self._start_time, "Start time is None"
self.elapsed_time = time.time() - self._start_time
[docs]
def succeed(self, log_path: Optional[str] = None):
"""Record success for this spec"""
self.result = "success"
self.stdout = self.fetch_log(log_path)
assert self._start_time, "Start time is None"
self.elapsed_time = time.time() - self._start_time
[docs]
class InstallRecord(SpecRecord):
"""Record class with specialization for install logs."""
def __init__(self, spec):
super().__init__(spec)
self.installed_from_binary_cache = None
[docs]
def fetch_log(self, log_path: Optional[str] = None) -> str:
"""Install log comes from log_path if provided, install prefix, or stage dir."""
try:
if log_path and os.path.exists(log_path):
stream = open(log_path, encoding="utf-8", errors="replace")
elif os.path.exists(self._package.install_log_path):
stream = gzip.open(
self._package.install_log_path, "rt", encoding="utf-8", errors="replace"
)
else:
stream = open(self._package.log_path, encoding="utf-8", errors="replace")
with stream as f:
return f.read()
except OSError:
return f"Cannot open log for {self._spec.cshort_spec}"
[docs]
def succeed(self, log_path: Optional[str] = None):
super().succeed(log_path)
self.installed_from_binary_cache = self._package.installed_from_binary_cache
[docs]
class NullInstallRecord(InstallRecord):
"""No-op drop-in for InstallRecord when no reporter is configured.
Avoids reading log files from disk on every completed build."""
[docs]
def start(self) -> None:
pass
[docs]
def succeed(self, log_path: Optional[str] = None) -> None:
pass
[docs]
def fail(self, exc, log_path: Optional[str] = None) -> None:
pass
[docs]
def skip(self, msg: str = "") -> None:
pass
[docs]
class NullRequestRecord(RequestRecord):
"""No-op drop-in for RequestRecord when no reporter is configured.
Avoids traversing the DAG and accumulating data that will not be reported."""
def __init__(self) -> None:
dict.__init__(self)
[docs]
def skip_installed(self) -> None:
pass
[docs]
def append_record(self, record) -> None:
pass
[docs]
def summarize(self) -> None:
pass
[docs]
class TestRecord(SpecRecord):
"""Record class with specialization for test logs."""
def __init__(self, spec, directory):
super().__init__(spec)
self.directory = directory
[docs]
def fetch_log(self, log_path: Optional[str] = None) -> str:
"""Get output from test log"""
log_file = os.path.join(self.directory, self._package.test_suite.test_log_name(self._spec))
try:
with open(log_file, "r", encoding="utf-8", errors="replace") as stream:
return "".join(stream.readlines())
except Exception:
return f"Cannot open log for {self._spec.cshort_spec}"
[docs]
def succeed(self, externals):
"""Test reports skip externals by default."""
if self._spec.external and not externals:
return self.skip(msg="Skipping test of external package")
super().succeed()