# 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 glob
import os
import posixpath
import re
import shutil
import sys
from contextlib import contextmanager
from pathlib import Path
import llnl.util.filesystem as fs
import llnl.util.tty as tty
import spack.config
import spack.hash_types as ht
import spack.spec
import spack.util.spack_json as sjson
from spack.error import SpackError
# Note: Posixpath is used here as opposed to
# os.path.join due to spack.spec.Spec.format
# requiring forward slash path seperators at this stage
default_projections = {
"all": posixpath.join(
"{architecture}", "{compiler.name}-{compiler.version}", "{name}-{version}-{hash}"
)
}
def _check_concrete(spec):
"""If the spec is not concrete, raise a ValueError"""
if not spec.concrete:
raise ValueError("Specs passed to a DirectoryLayout must be concrete!")
[docs]
class DirectoryLayout:
"""A directory layout is used to associate unique paths with specs.
Different installations are going to want different layouts for their
install, and they can use this to customize the nesting structure of
spack installs. The default layout is:
* <install root>/
* <platform-os-target>/
* <compiler>-<compiler version>/
* <name>-<version>-<hash>
The hash here is a SHA-1 hash for the full DAG plus the build
spec.
The installation directory projections can be modified with the
projections argument.
"""
def __init__(self, root, **kwargs):
self.root = root
self.check_upstream = True
projections = kwargs.get("projections") or default_projections
self.projections = dict(
(key, projection.lower()) for key, projection in projections.items()
)
# apply hash length as appropriate
self.hash_length = kwargs.get("hash_length", None)
if self.hash_length is not None:
for when_spec, projection in self.projections.items():
if "{hash}" not in projection:
if "{hash" in projection:
raise InvalidDirectoryLayoutParametersError(
"Conflicting options for installation layout hash" " length"
)
else:
raise InvalidDirectoryLayoutParametersError(
"Cannot specify hash length when the hash is not"
" part of all install_tree projections"
)
self.projections[when_spec] = projection.replace(
"{hash}", "{hash:%d}" % self.hash_length
)
# If any of these paths change, downstream databases may not be able to
# locate files in older upstream databases
self.metadata_dir = ".spack"
self.deprecated_dir = "deprecated"
self.spec_file_name = "spec.json"
# Use for checking yaml and deprecated types
self._spec_file_name_yaml = "spec.yaml"
self.extension_file_name = "extensions.yaml"
self.packages_dir = "repos" # archive of package.py files
self.manifest_file_name = "install_manifest.json"
@property
def hidden_file_regexes(self):
return ("^{0}$".format(re.escape(self.metadata_dir)),)
[docs]
def relative_path_for_spec(self, spec):
_check_concrete(spec)
projection = spack.projections.get_projection(self.projections, spec)
path = spec.format_path(projection)
return str(Path(path))
[docs]
def write_spec(self, spec, path):
"""Write a spec out to a file."""
_check_concrete(spec)
with open(path, "w") as f:
# The hash of the projection is the DAG hash which contains
# the full provenance, so it's availabe if we want it later
spec.to_json(f, hash=ht.dag_hash)
[docs]
def write_host_environment(self, spec):
"""The host environment is a json file with os, kernel, and spack
versioning. We use it in the case that an analysis later needs to
easily access this information.
"""
env_file = self.env_metadata_path(spec)
environ = spack.spec.get_host_environment_metadata()
with open(env_file, "w") as fd:
sjson.dump(environ, fd)
[docs]
def read_spec(self, path):
"""Read the contents of a file and parse them as a spec"""
try:
with open(path) as f:
extension = os.path.splitext(path)[-1].lower()
if extension == ".json":
spec = spack.spec.Spec.from_json(f)
elif extension == ".yaml":
# Too late for conversion; spec_file_path() already called.
spec = spack.spec.Spec.from_yaml(f)
else:
raise SpecReadError(
"Did not recognize spec file extension:" " {0}".format(extension)
)
except Exception as e:
if spack.config.get("config:debug"):
raise
raise SpecReadError("Unable to read file: %s" % path, "Cause: " + str(e))
# Specs read from actual installations are always concrete
spec._mark_concrete()
return spec
[docs]
def spec_file_path(self, spec):
"""Gets full path to spec file"""
_check_concrete(spec)
# Attempts to convert to JSON if possible.
# Otherwise just returns the YAML.
yaml_path = os.path.join(self.metadata_path(spec), self._spec_file_name_yaml)
json_path = os.path.join(self.metadata_path(spec), self.spec_file_name)
if os.path.exists(yaml_path) and fs.can_write_to_dir(yaml_path):
self.write_spec(spec, json_path)
try:
os.remove(yaml_path)
except OSError as err:
tty.debug("Could not remove deprecated {0}".format(yaml_path))
tty.debug(err)
elif os.path.exists(yaml_path):
return yaml_path
return json_path
[docs]
def deprecated_file_path(self, deprecated_spec, deprecator_spec=None):
"""Gets full path to spec file for deprecated spec
If the deprecator_spec is provided, use that. Otherwise, assume
deprecated_spec is already deprecated and its prefix links to the
prefix of its deprecator."""
_check_concrete(deprecated_spec)
if deprecator_spec:
_check_concrete(deprecator_spec)
# If deprecator spec is None, assume deprecated_spec already deprecated
# and use its link to find the file.
base_dir = (
self.path_for_spec(deprecator_spec)
if deprecator_spec
else os.readlink(deprecated_spec.prefix)
)
yaml_path = os.path.join(
base_dir,
self.metadata_dir,
self.deprecated_dir,
deprecated_spec.dag_hash() + "_" + self._spec_file_name_yaml,
)
json_path = os.path.join(
base_dir,
self.metadata_dir,
self.deprecated_dir,
deprecated_spec.dag_hash() + "_" + self.spec_file_name,
)
if os.path.exists(yaml_path) and fs.can_write_to_dir(yaml_path):
self.write_spec(deprecated_spec, json_path)
try:
os.remove(yaml_path)
except (IOError, OSError) as err:
tty.debug("Could not remove deprecated {0}".format(yaml_path))
tty.debug(err)
elif os.path.exists(yaml_path):
return yaml_path
return json_path
[docs]
@contextmanager
def disable_upstream_check(self):
self.check_upstream = False
yield
self.check_upstream = True
[docs]
def build_packages_path(self, spec):
return os.path.join(self.metadata_path(spec), self.packages_dir)
[docs]
def create_install_directory(self, spec):
_check_concrete(spec)
# Create install directory with properly configured permissions
# Cannot import at top of file
from spack.package_prefs import get_package_dir_permissions, get_package_group
# Each package folder can have its own specific permissions, while
# intermediate folders (arch/compiler) are set with access permissions
# equivalent to the root permissions of the layout.
group = get_package_group(spec)
perms = get_package_dir_permissions(spec)
fs.mkdirp(spec.prefix, mode=perms, group=group, default_perms="parents")
fs.mkdirp(self.metadata_path(spec), mode=perms, group=group) # in prefix
self.write_spec(spec, self.spec_file_path(spec))
[docs]
def ensure_installed(self, spec):
"""
Throws InconsistentInstallDirectoryError if:
1. spec prefix does not exist
2. spec prefix does not contain a spec file, or
3. We read a spec with the wrong DAG hash out of an existing install directory.
"""
_check_concrete(spec)
path = self.path_for_spec(spec)
spec_file_path = self.spec_file_path(spec)
if not os.path.isdir(path):
raise InconsistentInstallDirectoryError(
"Install prefix {0} does not exist.".format(path)
)
if not os.path.isfile(spec_file_path):
raise InconsistentInstallDirectoryError(
"Install prefix exists but contains no spec.json:", " " + path
)
installed_spec = self.read_spec(spec_file_path)
if installed_spec.dag_hash() != spec.dag_hash():
raise InconsistentInstallDirectoryError(
"Spec file in %s does not match hash!" % spec_file_path
)
[docs]
def all_specs(self):
if not os.path.isdir(self.root):
return []
specs = []
for _, path_scheme in self.projections.items():
path_elems = ["*"] * len(path_scheme.split(posixpath.sep))
# NOTE: Does not validate filename extension; should happen later
path_elems += [self.metadata_dir, "spec.json"]
pattern = os.path.join(self.root, *path_elems)
spec_files = glob.glob(pattern)
if not spec_files: # we're probably looking at legacy yaml...
path_elems += [self.metadata_dir, "spec.yaml"]
pattern = os.path.join(self.root, *path_elems)
spec_files = glob.glob(pattern)
specs.extend([self.read_spec(s) for s in spec_files])
return specs
[docs]
def all_deprecated_specs(self):
if not os.path.isdir(self.root):
return []
deprecated_specs = set()
for _, path_scheme in self.projections.items():
path_elems = ["*"] * len(path_scheme.split(posixpath.sep))
# NOTE: Does not validate filename extension; should happen later
path_elems += [
self.metadata_dir,
self.deprecated_dir,
"*_spec.*",
] # + self.spec_file_name]
pattern = os.path.join(self.root, *path_elems)
spec_files = glob.glob(pattern)
get_depr_spec_file = lambda x: os.path.join(
os.path.dirname(os.path.dirname(x)), self.spec_file_name
)
deprecated_specs |= set(
(self.read_spec(s), self.read_spec(get_depr_spec_file(s))) for s in spec_files
)
return deprecated_specs
[docs]
def specs_by_hash(self):
by_hash = {}
for spec in self.all_specs():
by_hash[spec.dag_hash()] = spec
return by_hash
[docs]
def path_for_spec(self, spec):
"""Return absolute path from the root to a directory for the spec."""
_check_concrete(spec)
if spec.external:
return spec.external_path
if self.check_upstream:
upstream, record = spack.store.STORE.db.query_by_spec_hash(spec.dag_hash())
if upstream:
raise SpackError(
"Internal error: attempted to call path_for_spec on"
" upstream-installed package."
)
path = self.relative_path_for_spec(spec)
assert not path.startswith(self.root)
return os.path.join(self.root, path)
[docs]
def remove_install_directory(self, spec, deprecated=False):
"""Removes a prefix and any empty parent directories from the root.
Raised RemoveFailedError if something goes wrong.
"""
path = self.path_for_spec(spec)
assert path.startswith(self.root)
# Windows readonly files cannot be removed by Python
# directly, change permissions before attempting to remove
if sys.platform == "win32":
kwargs = {
"ignore_errors": False,
"onerror": fs.readonly_file_handler(ignore_errors=False),
}
else:
kwargs = {} # the default value for ignore_errors is false
if deprecated:
if os.path.exists(path):
try:
metapath = self.deprecated_file_path(spec)
os.unlink(path)
os.remove(metapath)
except OSError as e:
raise RemoveFailedError(spec, path, e) from e
elif os.path.exists(path):
try:
shutil.rmtree(path, **kwargs)
except OSError as e:
raise RemoveFailedError(spec, path, e) from e
path = os.path.dirname(path)
while path != self.root:
if os.path.isdir(path):
try:
os.rmdir(path)
except OSError as e:
if e.errno == errno.ENOENT:
# already deleted, continue with parent
pass
elif e.errno == errno.ENOTEMPTY:
# directory wasn't empty, done
return
else:
raise e
path = os.path.dirname(path)
[docs]
class DirectoryLayoutError(SpackError):
"""Superclass for directory layout errors."""
def __init__(self, message, long_msg=None):
super().__init__(message, long_msg)
[docs]
class RemoveFailedError(DirectoryLayoutError):
"""Raised when a DirectoryLayout cannot remove an install prefix."""
def __init__(self, installed_spec, prefix, error):
super().__init__(
"Could not remove prefix %s for %s : %s" % (prefix, installed_spec.short_spec, error)
)
self.cause = error
[docs]
class InconsistentInstallDirectoryError(DirectoryLayoutError):
"""Raised when a package seems to be installed to the wrong place."""
def __init__(self, message, long_msg=None):
super().__init__(message, long_msg)
[docs]
class SpecReadError(DirectoryLayoutError):
"""Raised when directory layout can't read a spec."""
[docs]
class InvalidDirectoryLayoutParametersError(DirectoryLayoutError):
"""Raised when a invalid directory layout parameters are supplied"""
def __init__(self, message, long_msg=None):
super().__init__(message, long_msg)
[docs]
class InvalidExtensionSpecError(DirectoryLayoutError):
"""Raised when an extension file has a bad spec in it."""
[docs]
class ExtensionAlreadyInstalledError(DirectoryLayoutError):
"""Raised when an extension is added to a package that already has it."""
def __init__(self, spec, ext_spec):
super().__init__("%s is already installed in %s" % (ext_spec.short_spec, spec.short_spec))
[docs]
class ExtensionConflictError(DirectoryLayoutError):
"""Raised when an extension is added to a package that already has it."""
def __init__(self, spec, ext_spec, conflict):
super().__init__(
"%s cannot be installed in %s because it conflicts with %s"
% (ext_spec.short_spec, spec.short_spec, conflict.short_spec)
)