[Notes] [Git][BuildStream/buildstream][jennis/refactor_artifact

James Ennis pushed to branch jennis/refactor_artifact_log at BuildStream / buildstream

Commits:

cf5b906f

by James Ennis at 2019-01-23T12:52:54Z

_stream.py: Add the _classify_artifacts() helper

7b3c1c05

by James Ennis at 2019-01-23T12:52:54Z

_stream.py: No need to use a CASObject in _classify_artifacts()

0eb067f7

by James Ennis at 2019-01-23T12:52:54Z

_exceptions.py: Add ArtifactElementError exception

b0eda0a0

by James Ennis at 2019-01-23T14:44:42Z

_artifactelement.py: New module containing ArtifactElement class

The ArtifactElement class is to be used when we want to pass a
dummy artifact to our scheduler, typically this will be when we
are directly working with artifact refs.

c7cd8c7e

by James Ennis at 2019-01-23T14:44:42Z

_project.py: Add create_artifact_element() method

4d7c408c

by James Ennis at 2019-01-23T14:44:42Z

element.py: Add method get_artifact_name()

This commit removes the method ArtifactCache.get_artifact_fullname()
and replaces it with Element.get_artifact_name()

Given a key, we are now able to construct the full name of any of an
element's artifacts.

2d439bd8

by James Ennis at 2019-01-23T14:44:42Z

cli.py: Move artifact ref handling logic to stream

The loading of elements and the handling of artifacts does not belong
in this module. Such logic should be invoked using the Stream API

42f3f456

by James Ennis at 2019-01-23T14:44:42Z

cli.py: Remove _classify_artifacts() method

This method has been moved to Stream, where is it used there exclusively.

Changes:

buildstream/_artifactcache.py

@@ -19,7 +19,6 @@
  import multiprocessing
  import os
 -import string
  from collections.abc import Mapping
  from .types import _KeyStrength
@@ -77,37 +76,6 @@ class ArtifactCache():
          self._calculate_cache_quota()
 -    # get_artifact_fullname()
 -    #
 -    # Generate a full name for an artifact, including the
 -    # project namespace, element name and cache key.
 -    #
 -    # This can also be used as a relative path safely, and
 -    # will normalize parts of the element name such that only
 -    # digits, letters and some select characters are allowed.
 -    #
 -    # Args:
 -    #    element (Element): The Element object
 -    #    key (str): The element's cache key
 -    #
 -    # Returns:
 -    #    (str): The relative path for the artifact
 -    #
 -    def get_artifact_fullname(self, element, key):
 -        project = element._get_project()
+-
 -        # Normalize ostree ref unsupported chars
 -        valid_chars = string.digits + string.ascii_letters + '-._'
 -        element_name = ''.join([
 -            x if x in valid_chars else '_'
 -            for x in element.normal_name
 -        ])
+-
 -        assert key is not None
+-
 -        # assume project and element names are not allowed to contain slashes
 -        return '{0}/{1}/{2}'.format(project.name, element_name, key)
+-
      # setup_remotes():
+     #
      # Sets up which remotes to use
@@ -206,7 +174,7 @@ class ArtifactCache():
              for key in (strong_key, weak_key):
                  if key:
                      try:
 -                        ref = self.get_artifact_fullname(element, key)
 +                        ref = element.get_artifact_name(key)
                          self.cas.update_mtime(ref)
                      except CASError:
@@ -417,7 +385,7 @@ class ArtifactCache():
      # Returns: True if the artifact is in the cache, False otherwise
+     #
      def contains(self, element, key):
 -        ref = self.get_artifact_fullname(element, key)
 +        ref = element.get_artifact_name(key)
          return self.cas.contains(ref)
@@ -434,7 +402,7 @@ class ArtifactCache():
      # Returns: True if the subdir exists & is populated in the cache, False otherwise
+     #
      def contains_subdir_artifact(self, element, key, subdir):
 -        ref = self.get_artifact_fullname(element, key)
 +        ref = element.get_artifact_name(key)
          return self.cas.contains_subdir_artifact(ref, subdir)
      # list_artifacts():
@@ -442,8 +410,7 @@ class ArtifactCache():
      # List artifacts in this cache in LRU order.
+     #
      # Returns:
 -    #     ([str]) - A list of artifact names as generated by
 -    #               `ArtifactCache.get_artifact_fullname` in LRU order
 +    #     ([str]) - A list of artifact names as generated in LRU order
+     #
      def list_artifacts(self):
          return self.cas.list_refs()
@@ -455,8 +422,7 @@ class ArtifactCache():
+     #
      # Args:
      #     ref (artifact_name): The name of the artifact to remove (as
 -    #                          generated by
 -    #                          `ArtifactCache.get_artifact_fullname`)
 +    #                          generated by `Element.get_artifact_name`)
+     #
      # Returns:
      #    (int|None) The amount of space pruned from the repository in
@@ -503,7 +469,7 @@ class ArtifactCache():
      # Returns: path to extracted artifact
+     #
      def extract(self, element, key, subdir=None):
 -        ref = self.get_artifact_fullname(element, key)
 +        ref = element.get_artifact_name(key)
          path = os.path.join(self.extractdir, element._get_project().name, element.normal_name)
@@ -519,7 +485,7 @@ class ArtifactCache():
      #     keys (list): The cache keys to use
+     #
      def commit(self, element, content, keys):
 -        refs = [self.get_artifact_fullname(element, key) for key in keys]
 +        refs = [element.get_artifact_name(key) for key in keys]
          self.cas.commit(refs, content)
@@ -535,8 +501,8 @@ class ArtifactCache():
      #     subdir (str): A subdirectory to limit the comparison to
+     #
      def diff(self, element, key_a, key_b, *, subdir=None):
 -        ref_a = self.get_artifact_fullname(element, key_a)
 -        ref_b = self.get_artifact_fullname(element, key_b)
 +        ref_a = element.get_artifact_name(key_a)
 +        ref_b = element.get_artifact_name(key_b)
          return self.cas.diff(ref_a, ref_b, subdir=subdir)
@@ -597,7 +563,7 @@ class ArtifactCache():
      #   (ArtifactError): if there was an error
+     #
      def push(self, element, keys):
 -        refs = [self.get_artifact_fullname(element, key) for key in list(keys)]
 +        refs = [element.get_artifact_name(key) for key in list(keys)]
          project = element._get_project()
@@ -635,7 +601,7 @@ class ArtifactCache():
      #   (bool): True if pull was successful, False if artifact was not available
+     #
      def pull(self, element, key, *, progress=None, subdir=None, excluded_subdirs=None):
 -        ref = self.get_artifact_fullname(element, key)
 +        ref = element.get_artifact_name(key)
          project = element._get_project()
@@ -747,8 +713,8 @@ class ArtifactCache():
      #     newkey (str): A new cache key for the artifact
+     #
      def link_key(self, element, oldkey, newkey):
 -        oldref = self.get_artifact_fullname(element, oldkey)
 -        newref = self.get_artifact_fullname(element, newkey)
 +        oldref = element.get_artifact_name(oldkey)
 +        newref = element.get_artifact_name(newkey)
          self.cas.link_ref(oldref, newref)

buildstream/_artifactelement.py

 +#
 +#  Copyright (C) 2019 Bloomberg Finance LP
 +#
 +#  This program is free software; you can redistribute it and/or
 +#  modify it under the terms of the GNU Lesser General Public
 +#  License as published by the Free Software Foundation; either
 +#  version 2 of the License, or (at your option) any later version.
 +#
 +#  This library is distributed in the hope that it will be useful,
 +#  but WITHOUT ANY WARRANTY; without even the implied warranty of
 +#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.	 See the GNU
 +#  Lesser General Public License for more details.
 +#
 +#  You should have received a copy of the GNU Lesser General Public
 +#  License along with this library. If not, see <http://www.gnu.org/licenses/>.
 +#
 +#  Authors:
 +#        James Ennis <james ennis codethink co uk>
 +from ._exceptions import ArtifactElementError
++
++
 +# ArtifactElement()
 +#
 +# Object to be used for directly processing an artifact
 +#
 +# Args:
 +#    context (Context): The Context object
 +#    ref (str): The artifact ref
 +#
 +class ArtifactElement():
 +    def __init__(self, context, ref):
 +        try:
 +            project_name, element, key = ref.split('/', 2)
 +        except ValueError:
 +            raise ArtifactElementError("Artifact: {} is not of the expected format".format(ref))
++
 +        self.ref = ref
 +        self.project_name = project_name
 +        self.element = element
 +        self.key = key
++
 +        self._context = context
 +        self._artifacts = context.artifactcache
 +        self._cas = context.artifactcache.cas  # is this bad...?
++
 +    def get_artifact_name(self):
 +        return self.ref
++
 +    def _cached(self):
 +        return self._cas.contains(self.ref)

buildstream/_exceptions.py

@@ -344,3 +344,12 @@ class AppError(BstError):
+ #
  class SkipJob(Exception):
      pass
++
++
 +# ArtifactElementError
 +#
 +# Raised when errors are encountered by artifact elements
 +#
 +class ArtifactElementError(BstError):
 +    def __init__(self, message, *, detail=None, reason=None, temporary=False):
 +        super().__init__(message, detail=detail, domain=ErrorDomain.ELEMENT, reason=reason, temporary=True)

buildstream/_frontend/cli.py

  import os
  import sys
  from contextlib import ExitStack
 -from fnmatch import fnmatch
  from functools import partial
  from tempfile import TemporaryDirectory
@@ -859,38 +858,6 @@ def workspace_list(app):
  #############################################################
  #                     Artifact Commands                     #
  #############################################################
 -def _classify_artifacts(names, cas, project_directory):
 -    element_targets = []
 -    artifact_refs = []
 -    element_globs = []
 -    artifact_globs = []
+-
 -    for name in names:
 -        if name.endswith('.bst'):
 -            if any(c in "*?[" for c in name):
 -                element_globs.append(name)
 -            else:
 -                element_targets.append(name)
 -        else:
 -            if any(c in "*?[" for c in name):
 -                artifact_globs.append(name)
 -            else:
 -                artifact_refs.append(name)
+-
 -    if element_globs:
 -        for dirpath, _, filenames in os.walk(project_directory):
 -            for filename in filenames:
 -                element_path = os.path.join(dirpath, filename).lstrip(project_directory).lstrip('/')
 -                if any(fnmatch(element_path, glob) for glob in element_globs):
 -                    element_targets.append(element_path)
+-
 -    if artifact_globs:
 -        artifact_refs.extend(ref for ref in cas.list_refs()
 -                             if any(fnmatch(ref, glob) for glob in artifact_globs))
+-
 -    return element_targets, artifact_refs
+-
+-
  @cli.group(short_help="Manipulate cached artifacts")
  def artifact():
      """Manipulate cached artifacts"""
@@ -1045,53 +1012,31 @@ def artifact_push(app, elements, deps, remote):
  @click.pass_obj
  def artifact_log(app, artifacts):
      """Show logs of all artifacts"""
 -    from .._exceptions import CASError
 -    from .._message import MessageType
 -    from .._pipeline import PipelineSelection
 -    from ..storage._casbaseddirectory import CasBasedDirectory
+-
 -    with ExitStack() as stack:
 -        stack.enter_context(app.initialized())
 -        cache = app.context.artifactcache
 +    # Guess the element if we're in a workspace
 +    if not artifacts:
 +        guessed_target = app.context.guess_element()
 +        if guessed_target:
 +            artifacts = [guessed_target]
 -        elements, artifacts = _classify_artifacts(artifacts, cache.cas,
 -                                                  app.project.directory)
+-
 -        vdirs = []
 -        extractdirs = []
 -        if artifacts:
 -            for ref in artifacts:
 -                try:
 -                    cache_id = cache.cas.resolve_ref(ref, update_mtime=True)
 -                    vdir = CasBasedDirectory(cache.cas, cache_id)
 -                    vdirs.append(vdir)
 -                except CASError as e:
 -                    app._message(MessageType.WARN, "Artifact {} is not cached".format(ref), detail=str(e))
 -                    continue
 -        if elements:
 -            elements = app.stream.load_selection(elements, selection=PipelineSelection.NONE)
 -            for element in elements:
 -                if not element._cached():
 -                    app._message(MessageType.WARN, "Element {} is not cached".format(element))
 -                    continue
 -                ref = cache.get_artifact_fullname(element, element._get_cache_key())
 -                cache_id = cache.cas.resolve_ref(ref, update_mtime=True)
 -                vdir = CasBasedDirectory(cache.cas, cache_id)
 -                vdirs.append(vdir)
+-
 -        for vdir in vdirs:
 -            # NOTE: If reading the logs feels unresponsive, here would be a good place to provide progress information.
 -            logsdir = vdir.descend(["logs"])
 -            td = stack.enter_context(TemporaryDirectory())
 -            logsdir.export_files(td, can_link=True)
 -            extractdirs.append(td)
+-
 -        for extractdir in extractdirs:
 -            for log in (os.path.join(extractdir, log) for log in os.listdir(extractdir)):
 -                # NOTE: Should click gain the ability to pass files to the pager this can be optimised.
 -                with open(log) as f:
 -                    data = f.read()
 -                    click.echo_via_pager(data)
 +    with app.initialized():
 +        vdirs = app.stream.artifact_log(artifacts)
++
 +        with ExitStack() as stack:
 +            extractdirs = []
 +            for vdir in vdirs:
 +                # NOTE: If reading the logs feels unresponsive, here would be a good place
 +                # to provide progress information.
 +                logsdir = vdir.descend(["logs"])
 +                td = stack.enter_context(TemporaryDirectory())
 +                logsdir.export_files(td, can_link=True)
 +                extractdirs.append(td)
++
 +            for extractdir in extractdirs:
 +                for log in (os.path.join(extractdir, log) for log in os.listdir(extractdir)):
 +                    # NOTE: Should click gain the ability to pass files to the pager this can be optimised.
 +                    with open(log) as f:
 +                        data = f.read()
 +                        click.echo_via_pager(data)
  ##################################################################

buildstream/_project.py

@@ -26,6 +26,7 @@ from . import utils
  from . import _cachekey
  from . import _site
  from . import _yaml
 +from ._artifactelement import ArtifactElement
  from ._profile import Topics, profile_start, profile_end
  from ._exceptions import LoadError, LoadErrorReason
  from ._options import OptionPool
@@ -252,6 +253,19 @@ class Project():
          else:
              return self.config.element_factory.create(self._context, self, meta)
 +    # create_artifact_element()
 +    #
 +    # Instantiate and return an ArtifactElement
 +    #
 +    # Args:
 +    #    ref (str): A string of the artifact ref
 +    #
 +    # Returns:
 +    #    (ArtifactElement): A newly created ArtifactElement object of the appropriate kind
 +    #
 +    def create_artifact_element(self, ref):
 +        return ArtifactElement(self._context, ref)
++
      # create_source()
+     #
      # Instantiate and return a Source

buildstream/_stream.py

@@ -27,6 +27,7 @@ import shutil
  import tarfile
  import tempfile
  from contextlib import contextmanager, suppress
 +from fnmatch import fnmatch
  from ._exceptions import StreamError, ImplError, BstError, set_last_task_error
  from ._message import Message, MessageType
@@ -439,6 +440,53 @@ class Stream():
              raise StreamError("Error while staging dependencies into a sandbox"
                                ": '{}'".format(e), detail=e.detail, reason=e.reason) from e
 +    # artifact_log()
 +    #
 +    # Show the full log of an artifact
 +    #
 +    # Args:
 +    #    targets (str): Targets to view the logs of
 +    #
 +    # Returns:
 +    #    vdirs (list): A list of CasBasedDirectory objects
 +    #
 +    def artifact_log(self, targets):
 +        from .storage._casbaseddirectory import CasBasedDirectory  # is this bad?
++
 +        cas = self._artifacts.cas  # Is this bad?
 +        cached_refs = cas.list_refs()
 +        project_dir = self._project.directory
++
 +        # Distinguish the artifacts from the elements
 +        elements, artifacts = self._classify_artifacts(targets, cached_refs, project_dir)
++
 +        # Obtain Element objects
 +        if elements:
 +            elements = self.load_selection(elements, selection=PipelineSelection.NONE)
++
 +        # Obtain ArtifactElement objects
 +        artifact_elements = []
 +        if artifacts:
 +            for ref in artifacts:
 +                artifact_element = self._project.create_artifact_element(ref)
 +                artifact_elements.append(artifact_element)
++
 +        # Concatenate the lists
 +        objects = elements + artifact_elements
++
 +        vdirs = []
 +        for obj in objects:
 +            ref = obj.get_artifact_name()
 +            if not obj._cached():
 +                self._message(MessageType.WARN, "{} is not cached".format(ref))
 +                continue
++
 +            cache_id = cas.resolve_ref(ref, update_mtime=True)
 +            vdir = CasBasedDirectory(cas, cache_id)
 +            vdirs.append(vdir)
++
 +        return vdirs
++
      # source_checkout()
+     #
      # Checkout sources of the target element to the specified location
@@ -1273,3 +1321,47 @@ class Stream():
                  required_list.append(element)
          return required_list
++
 +    # _classify_artifacts()
 +    #
 +    # Split up a list of tagets into element names and artifact refs
 +    #
 +    # Args:
 +    #    names (list): A list of targets
 +    #    cached (list): A list of locally cached refs
 +    #    project_directory (str): Absolute path to the project
 +    #
 +    # Returns:
 +    #    (list): element names present in the targets
 +    #    (list): artifact refs present in the targets
 +    #
 +    def _classify_artifacts(self, names, cached, project_directory):
 +        element_targets = []
 +        artifact_refs = []
 +        element_globs = []
 +        artifact_globs = []
++
 +        for name in names:
 +            if name.endswith('.bst'):
 +                if any(c in "*?[" for c in name):
 +                    element_globs.append(name)
 +                else:
 +                    element_targets.append(name)
 +            else:
 +                if any(c in "*?[" for c in name):
 +                    artifact_globs.append(name)
 +                else:
 +                    artifact_refs.append(name)
++
 +        if element_globs:
 +            for dirpath, _, filenames in os.walk(project_directory):
 +                for filename in filenames:
 +                    element_path = os.path.join(dirpath, filename).lstrip(project_directory).lstrip('/')
 +                    if any(fnmatch(element_path, glob) for glob in element_globs):
 +                        element_targets.append(element_path)
++
 +        if artifact_globs:
 +            artifact_refs.extend(ref for ref in cached
 +                                 if any(fnmatch(ref, glob) for glob in artifact_globs))
++
 +        return element_targets, artifact_refs

buildstream/element.py

@@ -82,6 +82,7 @@ import contextlib
  from contextlib import contextmanager
  import tempfile
  import shutil
 +import string
  from . import _yaml
  from ._variables import Variables
@@ -577,6 +578,37 @@ class Element(Plugin):
          self.__assert_cached()
          return self.__compute_splits(include, exclude, orphans)
 +    def get_artifact_name(self, key=None):
 +        """Compute and return this element's full artifact name
++
 +        Generate a full name for an artifact, including the project
 +        namespace, element name and cache key.
++
 +        This can also be used as a relative path safely, and
 +        will normalize parts of the element name such that only
 +        digits, letters and some select characters are allowed.
++
 +        Args:
 +           key (str): The element's cache key. Defaults to None
++
 +        Returns:
 +           (str): The relative path for the artifact
 +        """
 +        project = self._get_project()
 +        if key is None:
 +            key = self._get_cache_key()
++
 +        assert key is not None
++
 +        valid_chars = string.digits + string.ascii_letters + '-._'
 +        element_name = ''.join([
 +            x if x in valid_chars else '_'
 +            for x in self.normal_name
 +        ])
++
 +        # assume project and element names are not allowed to contain slashes
 +        return '{0}/{1}/{2}'.format(project.name, element_name, key)
++
      def stage_artifact(self, sandbox, *, path=None, include=None, exclude=None, orphans=True, update_mtimes=None):
          """Stage this element's output artifact in the sandbox

tests/artifactcache/pull.py

@@ -210,7 +210,7 @@ def test_pull_tree(cli, tmpdir, datafiles):
          assert artifactcache.contains(element, element_key)
          # Retrieve the Directory object from the cached artifact
 -        artifact_ref = artifactcache.get_artifact_fullname(element, element_key)
 +        artifact_ref = element.get_artifact_name(element_key)
          artifact_digest = cas.resolve_ref(artifact_ref)
          queue = multiprocessing.Queue()

tests/artifactcache/push.py

@@ -190,7 +190,7 @@ def test_push_directory(cli, tmpdir, datafiles):
          assert artifactcache.has_push_remotes(element=element)
          # Recreate the CasBasedDirectory object from the cached artifact
 -        artifact_ref = artifactcache.get_artifact_fullname(element, element_key)
 +        artifact_ref = element.get_artifact_name(element_key)
          artifact_digest = cas.resolve_ref(artifact_ref)
          queue = multiprocessing.Queue()

[Notes] [Git][BuildStream/buildstream][jennis/refactor_artifact_log] 8 commits: _stream.py: Add the _classify_artifacts() helper

James Ennis pushed to branch jennis/refactor_artifact_log at BuildStream / buildstream

Commits:

9 changed files:

Changes: