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

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

Commits:

9cd6aa1d

by James Ennis at 2019-01-24T11:54:02Z

_stream.py: Modify behaviour of _classify_artifacts()

There is no need for this method to use a cas object.
Furthermore, we should search for artifact globs in the
project's element path, not the project's directory, as
these may differ.

ae534290

by James Ennis at 2019-01-24T11:55:14Z

_exceptions.py: Add ArtifactElementError exception

1183fdfe

by James Ennis at 2019-01-24T11:55:14Z

_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.

86c96245

by James Ennis at 2019-01-24T11:55:14Z

_project.py: Add create_artifact_element() method

c84d530f

by James Ennis at 2019-01-24T11:55:14Z

element.py: Lift ArtifactCache.get_artifact_fullname() to here

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.

67fdb4eb

by James Ennis at 2019-01-24T11:56:27Z

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

01c66e37

by James Ennis at 2019-01-24T11:56:27Z

cli.py: Remove _classify_artifacts() method

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

3dd57c1b

by James Ennis at 2019-01-24T14:09:05Z

cascache.py: Allow CASCache.list_refs() to handle globs

This commit ensures that CASCache.list_refs(), and
ArtifactCache.list_artifacts(), can both handle glob expressions.

0d456aa1

by James Ennis at 2019-01-24T14:12:42Z

_stream.py: _classify_artifacts() should be able to handle globs

_classify_artifacts() no longer filters glob expressions by first
obtaining a list of ALL refs locally cached. We know only obain refs
specified by the glob _expression_.

Furthermore, the Project.element_path is used to start searching for
globbed elements as opposed to the Project.directory.

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,11 +410,10 @@ 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()
 +    def list_artifacts(self, *, glob=None):
 +        return self.cas.list_refs(glob=glob)
      # remove():
+     #
@@ -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/_cas/cascache.py

@@ -475,14 +475,18 @@ class CASCache():
      # Returns:
      #     (list) - A list of refs in LRM order
+     #
 -    def list_refs(self):
 +    def list_refs(self, *, glob=None):
          # string of: /path/to/repo/refs/heads
          ref_heads = os.path.join(self.casdir, 'refs', 'heads')
 +        if glob is not None and '/' in glob:
 +            ref_heads_with_glob = os.path.join(ref_heads, os.path.dirname(glob))
++
          refs = []
          mtimes = []
 -        for root, _, files in os.walk(ref_heads):
 +        path = ref_heads_with_glob if glob else ref_heads
 +        for root, _, files in os.walk(path):
              for filename in files:
                  ref_path = os.path.join(root, filename)
                  refs.append(os.path.relpath(ref_path, ref_heads))

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

@@ -440,6 +440,51 @@ 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?
++
 +        # Distinguish the artifacts from the elements
 +        elements, artifacts = self._classify_artifacts(targets)
++
 +        # 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
@@ -1277,44 +1322,47 @@ class Stream():
      # _classify_artifacts()
+     #
 -    # Split up a list of tagets into element names and artifact refs
 +    # Split up a list of targets into element names and artifact refs
+     #
      # Args:
 -    #    names (list): A list of targets
 -    #    cas (CASCache): The CASCache object
 -    #    project_directory (str): Absolute path to the project
 +    #    targets (list): A list of targets
+     #
      # Returns:
      #    (list): element names present in the targets
      #    (list): artifact refs present in the targets
+     #
 -    def _classify_artifacts(names, cas, project_directory):
 +    def _classify_artifacts(self, targets):
          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)
 +        for target in targets:
 +            if target.endswith('.bst'):
 +                if any(c in "*?[" for c in target):
 +                    element_globs.append(target)
                  else:
 -                    element_targets.append(name)
 +                    element_targets.append(target)
              else:
 -                if any(c in "*?[" for c in name):
 -                    artifact_globs.append(name)
 +                if any(c in "*?[" for c in target):
 +                    artifact_globs.append(target)
                  else:
 -                    artifact_refs.append(name)
 +                    artifact_refs.append(target)
          if element_globs:
 -            for dirpath, _, filenames in os.walk(project_directory):
 +            for dirpath, _, filenames in os.walk(self._project.element_path):
                  for filename in filenames:
 -                    element_path = os.path.join(dirpath, filename).lstrip(project_directory).lstrip('/')
 +                    element_path = os.path.join(dirpath, filename).lstrip(self._project.element_path).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))
 +            for glob in artifact_globs:
 +                refs = self._artifacts.list_artifacts(glob=glob)
 +                for ref in refs:
 +                    if fnmatch(ref, glob):
 +                        artifact_refs.append(ref)
 +            if not artifact_refs:
 +                self._message(MessageType.WARN, "No artifacts found for globs: {}".format(', '.join(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] 9 commits: _stream.py: Modify behaviour of _classify_artifacts()

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

Commits:

10 changed files:

Changes: