.coveragerc

@@ -6,6 +6,8 @@ include =
 omit =
   # Omit profiling helper module
   */buildstream/_profile.py
+  # Omit generated code
+  */buildstream/_protos/*
   */.eggs/*
 
 [report]

.gitignore

@@ -15,6 +15,7 @@ tmp
 .coverage
 .coverage.*
 .cache
+.pytest_cache/
 *.bst/
 
 # Pycache, in case buildstream is ran directly from within the source

.gitlab-ci.yml

-image: buildstream/testsuite-debian:8-master-57-be5a863
+image: buildstream/testsuite-debian:9-master-102-9067e269
 
 cache:
   key: "$CI_JOB_NAME-"
@@ -88,22 +88,19 @@ source_dist:
     paths:
     - coverage-linux/
 
-tests-debian-8:
-  <<: *linux-tests
-
 tests-debian-9:
-  image: buildstream/buildstream-debian:master-81-caa5241
+  image: buildstream/testsuite-debian:9-master-102-9067e269
   <<: *linux-tests
 
 tests-fedora-27:
-  image: buildstream/buildstream-fedora:master-56-5d7ee17
+  image: buildstream/testsuite-fedora:27-master-102-9067e269
   <<: *linux-tests
 
 
 tests-unix:
   # Use fedora here, to a) run a test on fedora and b) ensure that we
   # can get rid of ostree - this is not possible with debian-8
-  image: buildstream/buildstream-fedora:master-56-5d7ee17
+  image: buildstream/testsuite-fedora:27-master-102-9067e269
   stage: test
   variables:
     BST_FORCE_BACKEND: "unix"
@@ -141,12 +138,13 @@ tests-unix:
 docs:
   stage: test
   script:
-  - pip3 install sphinx==1.7.1
+  - export BST_SOURCE_CACHE="$(pwd)/cache/integration-cache/sources"
+  - pip3 install sphinx
   - pip3 install sphinx-click
   - pip3 install sphinx_rtd_theme
   - cd dist && ./unpack.sh && cd buildstream
   - pip3 install .
-  - make -C doc
+  - make BST_FORCE_SESSION_REBUILD=1 -C doc
   - cd ../..
   - mv dist/buildstream/doc/build/html public
   artifacts:
@@ -222,7 +220,6 @@ coverage:
     - coverage combine --rcfile=../.coveragerc -a coverage.*
     - coverage report --rcfile=../.coveragerc -m
   dependencies:
-  - tests-debian-8
   - tests-debian-9
   - tests-fedora-27
   - tests-unix

.pylintrc

@@ -11,7 +11,7 @@ ignore=CVS,tests,doc
 
 # Add files or directories matching the regex patterns to the blacklist. The
 # regex matches against base names, not paths.
-ignore-patterns=
+ignore-patterns=.*_pb2.py,.*_pb2_grpc.py
 
 # Python code to execute, usually for sys.path manipulation such as
 # pygtk.require().
@@ -190,7 +190,7 @@ ignored-classes=optparse.Values,thread._local,_thread._local,contextlib.closing,
 # (useful for modules/projects where namespaces are manipulated during runtime
 # and thus existing member attributes cannot be deduced by static analysis. It
 # supports qualified module names, as well as Unix pattern matching.
-ignored-modules=pkg_resources,gi.repository
+ignored-modules=pkg_resources,gi.repository,grpc,buildstream._protos.*
 
 # Show a hint with possible names when a member name was not found. The aspect
 # of finding the hint is based on edit distance.

HACKING.rst

-Hacking on BuildStream
-======================
+Contributing
+============
 Some tips and guidelines for developers hacking on BuildStream
 
 
@@ -23,26 +23,31 @@ a reasonable timeframe for identifying these.
 
 Patch submissions
 -----------------
-Branches must be submitted as merge requests in gitlab and should usually
-be associated to an issue report on gitlab.
+Branches must be submitted as merge requests in gitlab. If the branch
+fixes an issue or is related to any issues, these issues must be mentioned
+in the merge request or preferably the commit messages themselves.
 
-Commits in the branch which address specific issues must specify the
-issue number in the commit message.
+Branch names for merge requests should be prefixed with the submitter's
+name or nickname, e.g. ``username/implement-flying-ponies``.
 
-Merge requests that are not yet ready for review must be prefixed with the
-``WIP:`` identifier. A merge request is not ready for review until the
-submitter expects that the patch is ready to actually land.
+You may open merge requests for the branches you create before you
+are ready to have them reviewed upstream, as long as your merge request
+is not yet ready for review then it must be prefixed with the ``WIP:``
+identifier.
 
 Submitted branches must not contain a history of the work done in the
 feature branch. Please use git's interactive rebase feature in order to
 compose a clean patch series suitable for submission.
 
-We prefer that test case and documentation changes be submitted
-in separate commits from the code changes which they test.
+We prefer that documentation changes be submitted in separate commits from
+the code changes which they document, and new test cases are also preferred
+in separate commits.
 
-Ideally every commit in the history of master passes its test cases. This
-makes bisections more easy to perform, but is not always practical with
-more complex branches.
+If a commit in your branch modifies behavior such that a test must also
+be changed to match the new behavior, then the tests should be updated
+with the same commit. Ideally every commit in the history of master passes
+its test cases, this makes bisections more easy to perform, but is not
+always practical with more complex branches.
 
 
 Commit messages
@@ -54,9 +59,6 @@ the change.
 The summary line must start with what changed, followed by a colon and
 a very brief description of the change.
 
-If there is an associated issue, it **must** be mentioned somewhere
-in the commit message.
-
 **Example**::
 
   element.py: Added the frobnicator so that foos are properly frobbed.
@@ -65,8 +67,6 @@ in the commit message.
   the element. Elements that are not properly frobnicated raise
   an error to inform the user of invalid frobnication rules.
 
-  This fixes issue #123
-
 
 Coding style
 ------------
@@ -257,7 +257,7 @@ using pip or some other mechanism::
   pip3 install --user arpy
 
 Furthermore, the documentation build requires that BuildStream itself
-be installed.
+be installed, as it will be used in the process of generating its docs.
 
 To build the documentation, just run the following::
 
@@ -267,6 +267,23 @@ This will give you a ``doc/build/html`` directory with the html docs which
 you can view in your browser locally to test.
 
 
+Regenerating session html
+'''''''''''''''''''''''''
+The documentation build will build the session files if they are missing,
+or if explicitly asked to rebuild. We revision the generated session html files
+in order to reduce the burden on documentation contributors.
+
+To explicitly rebuild the session snapshot html files, it is recommended that you
+first set the ``BST_SOURCE_CACHE`` environment variable to your source cache, this
+will make the docs build reuse already downloaded sources::
+
+  export BST_SOURCE_CACHE=~/.cache/buildstream/sources
+
+To force rebuild session html while building the doc, simply build the docs like this::
+
+  make BST_FORCE_SESSION_REBUILD=1 -C doc
+
+
 Man pages
 ~~~~~~~~~
 Unfortunately it is quite difficult to integrate the man pages build
@@ -277,7 +294,7 @@ committed with that.
 To do this, first ensure you have ``click_man`` installed, possibly
 with::
 
-  pip install --user click_man
+  pip3 install --user click_man
 
 Then, in the toplevel directory of buildstream, run the following::
 
@@ -314,6 +331,106 @@ format::
   """
 
 
+Documentation Examples
+~~~~~~~~~~~~~~~~~~~~~~
+The examples section of the documentation contains a series of standalone
+examples, here are the criteria for an example addition.
+
+* The example has a ``${name}``
+
+* The example has a project users can copy and use
+
+  * This project is added in the directory ``doc/examples/${name}``
+
+* The example has a documentation component
+
+  * This is added at ``doc/source/examples/${name}.rst``
+  * A reference to ``examples/${name}`` is added to the toctree in ``doc/source/examples.rst``
+  * This documentation discusses the project elements declared in the project and may
+    provide some BuildStream command examples
+  * This documentation links out to the reference manual at every opportunity
+
+* The example has a CI test component
+
+  * This is an integration test added at ``tests/examples/${name}``
+  * This test runs BuildStream in the ways described in the example
+    and assert that we get the results which we advertize to users in
+    the said examples.
+
+
+Adding BuildStream command output
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+As a part of building the docs, BuildStream will run itself and extract
+some html for the colorized output which is produced.
+
+If you want to run BuildStream to produce some nice html for your
+documentation, then you can do so by adding new ``.run`` files to the
+``doc/sessions/`` directory.
+
+Any files added as ``doc/sessions/${example}.run`` will result in generated
+file at ``doc/source/sessions/${example}.html``, and these files can be
+included in the reStructuredText documentation at any time with::
+
+  .. raw:: html
+     :file: sessions/${example}.html
+
+The ``.run`` file format is just another YAML dictionary which consists of a
+``commands`` list, instructing the program what to do command by command.
+
+Each *command* is a dictionary, the members of which are listed here:
+
+* ``directory``: The input file relative project directory
+
+* ``output``: The input file relative output html file to generate (optional)
+
+* ``fake-output``: Don't really run the command, just pretend to and pretend
+  this was the output, an empty string will enable this too.
+
+* ``command``: The command to run, without the leading ``bst``
+
+When adding a new ``.run`` file, one should normally also commit the new
+resulting generated ``.html`` file(s) into the ``doc/source/sessions-stored/``
+directory at the same time, this ensures that other developers do not need to
+regenerate them locally in order to build the docs.
+
+**Example**:
+
+.. code:: yaml
+
+   commands:
+
+   # Make it fetch first
+   - directory: ../examples/foo
+     command: fetch hello.bst
+
+   # Capture a build output
+   - directory: ../examples/foo
+     output: ../source/sessions/foo-build.html
+     command: build hello.bst
+
+
+Protocol Buffers
+----------------
+BuildStream uses protobuf and gRPC for serialization and communication with
+artifact cache servers.  This requires ``.proto`` files and Python code
+generated from the ``.proto`` files using protoc.  All these files live in the
+``buildstream/_protos`` directory.  The generated files are included in the
+git repository to avoid depending on grpcio-tools for user installations.
+
+
+Regenerating code
+~~~~~~~~~~~~~~~~~
+When ``.proto`` files are modified, the corresponding Python code needs to
+be regenerated.  As a prerequisite for code generation you need to install
+``grpcio-tools`` using pip or some other mechanism::
+
+  pip3 install --user grpcio-tools
+
+To actually regenerate the code::
+
+  ./setup.py build_grpc
+
+
 Testing BuildStream
 -------------------
 BuildStream uses pytest for regression tests and testing out
@@ -333,7 +450,7 @@ To run the tests, just type::
 At the toplevel.
 
 When debugging a test, it can be desirable to see the stdout
-and stderr generated by a test, to do this use the --addopts
+and stderr generated by a test, to do this use the ``--addopts``
 function to feed arguments to pytest as such::
 
   ./setup.py test --addopts -s
@@ -413,7 +530,7 @@ tool.
 
 Python provides `cProfile <https://docs.python.org/3/library/profile.html>`_
 which gives you a list of all functions called during execution and how much
-time was spent in each function. Here is an example of running `bst --help`
+time was spent in each function. Here is an example of running ``bst --help``
 under cProfile:
 
     python3 -m cProfile -o bst.cprofile -- $(which bst) --help

MANIFEST.in

 # Basic toplevel package includes
-include buildstream.doap
+include BuildStream.doap
 include COPYING
 include HACKING.rst
 include MAINTAINERS
@@ -18,3 +18,6 @@ recursive-include tests *.bst
 recursive-include tests *.conf
 recursive-include tests *.sh
 recursive-include tests *.expected
+
+# Protocol Buffers
+recursive-include buildstream/_protos *.proto

NEWS

+=================
+buildstream 1.3.1
+=================
+
+  o Add a `--tar` option to `bst checkout` which allows a tarball to be
+    created from the artifact contents.
+
+  o Fetching and tracking will consult mirrors defined in project config,
+    and the preferred mirror to fetch from can be defined in the command
+    line or user config.
+
 =================
 buildstream 1.1.4
 =================
@@ -7,6 +18,18 @@ buildstream 1.1.4
     `bst track` on a filter element will track the sources on the
     element that it depends on (if it has sources).
 
+  o Added new simple `make` element
+
+  o Switch to Remote Execution CAS-based artifact cache on all platforms.
+
+    Artifact servers need to be migrated.
+
+  o BuildStream now requires python version >= 3.5
+
+  o BuildStream will now automatically clean up old artifacts when it
+    runs out of space. The exact behavior is configurable in the user's
+    buildstream.conf.
+
 
 =================
 buildstream 1.1.3

README.rst

-About BuildStream
------------------
+About
+-----
 .. image:: https://gitlab.com/BuildStream/buildstream/badges/master/pipeline.svg
    :target: https://gitlab.com/BuildStream/buildstream/commits/master
 
@@ -25,7 +25,7 @@ BuildStream offers the following advantages:
 
 * **Declarative build instructions/definitions**
 
-  BuildStream provides a a flexible and extensible framework for the modelling
+  BuildStream provides a flexible and extensible framework for the modelling
   of software build pipelines in a declarative YAML format, which allows you to
   manipulate filesystem data in a controlled, reproducible sandboxed environment.
 
@@ -61,25 +61,29 @@ How does BuildStream work?
 ==========================
 BuildStream operates on a set of YAML files (.bst files), as follows:
 
-* loads the YAML files which describe the target(s) and all dependencies
-* evaluates the version information and build instructions to calculate a build
+* Loads the YAML files which describe the target(s) and all dependencies.
+* Evaluates the version information and build instructions to calculate a build
   graph for the target(s) and all dependencies and unique cache-keys for each
-  element
-* retrieves elements from cache if they are already built, or builds them in a
-  sandboxed environment using the instructions declared in the .bst files
-* transforms/configures and/or deploys the resulting target(s) based on the
+  element.
+* Retrieves previously built elements (artifacts) from a local/remote cache, or
+  builds the elements in a sandboxed environment using the instructions declared
+  in the .bst files.
+* Transforms/configures and/or deploys the resulting target(s) based on the
   instructions declared in the .bst files.
 
 
 How can I get started?
 ======================
-The easiest way to get started is to explore some existing .bst files, for example:
+To start using BuildStream, first,
+`install <https://buildstream.gitlab.io/buildstream/main_install.html>`_
+BuildStream onto your machine and then follow our
+`tutorial <https://buildstream.gitlab.io/buildstream/using_tutorial.html>`_.
+
+We also recommend exploring some existing BuildStream projects:
 
 * https://gitlab.gnome.org/GNOME/gnome-build-meta/
 * https://gitlab.com/freedesktop-sdk/freedesktop-sdk
 * https://gitlab.com/baserock/definitions
-* https://gitlab.com/BuildStream/buildstream-examples/tree/master/build-x86image
-* https://gitlab.com/BuildStream/buildstream-examples/tree/master/netsurf-flatpak
 
 If you have any questions please ask on our `#buildstream <irc://irc.gnome.org/buildstream>`_ channel in `irc.gnome.org <irc://irc.gnome.org>`_
 

buildstream/__init__.py

-#!/usr/bin/env python3
 #
 #  Copyright (C) 2016 Codethink Limited
 #
@@ -30,7 +29,7 @@ if "_BST_COMPLETION" not in os.environ:
     from .utils import UtilError, ProgramNotFoundError
     from .sandbox import Sandbox, SandboxFlags
     from .plugin import Plugin
-    from .source import Source, SourceError, Consistency
+    from .source import Source, SourceError, Consistency, SourceFetcher
     from .element import Element, ElementError, Scope
     from .buildelement import BuildElement
     from .scriptelement import ScriptElement

buildstream/_artifactcache/__init__.py

-#!/usr/bin/env python3
 #
 #  Copyright (C) 2017-2018 Codethink Limited
 #

buildstream/_artifactcache/artifactcache.py

-#!/usr/bin/env python3
 #
 #  Copyright (C) 2017-2018 Codethink Limited
 #
@@ -22,7 +21,8 @@ import os
 import string
 from collections import Mapping, namedtuple
 
-from .._exceptions import ImplError, LoadError, LoadErrorReason
+from ..element import _KeyStrength
+from .._exceptions import ArtifactError, ImplError, LoadError, LoadErrorReason
 from .._message import Message, MessageType
 from .. import utils
 from .. import _yaml
@@ -36,22 +36,38 @@ from .. import _yaml
 #     push (bool): Whether we should attempt to push artifacts to this cache,
 #                  in addition to pulling from it.
 #
-class ArtifactCacheSpec(namedtuple('ArtifactCacheSpec', 'url push')):
+class ArtifactCacheSpec(namedtuple('ArtifactCacheSpec', 'url push server_cert client_key client_cert')):
 
     # _new_from_config_node
     #
     # Creates an ArtifactCacheSpec() from a YAML loaded node
     #
     @staticmethod
-    def _new_from_config_node(spec_node):
-        _yaml.node_validate(spec_node, ['url', 'push'])
+    def _new_from_config_node(spec_node, basedir=None):
+        _yaml.node_validate(spec_node, ['url', 'push', 'server-cert', 'client-key', 'client-cert'])
         url = _yaml.node_get(spec_node, str, 'url')
         push = _yaml.node_get(spec_node, bool, 'push', default_value=False)
         if not url:
             provenance = _yaml.node_get_provenance(spec_node)
             raise LoadError(LoadErrorReason.INVALID_DATA,
                             "{}: empty artifact cache URL".format(provenance))
-        return ArtifactCacheSpec(url, push)
+
+        server_cert = _yaml.node_get(spec_node, str, 'server-cert', default_value=None)
+        if server_cert and basedir:
+            server_cert = os.path.join(basedir, server_cert)
+
+        client_key = _yaml.node_get(spec_node, str, 'client-key', default_value=None)
+        if client_key and basedir:
+            client_key = os.path.join(basedir, client_key)
+
+        client_cert = _yaml.node_get(spec_node, str, 'client-cert', default_value=None)
+        if client_cert and basedir:
+            client_cert = os.path.join(basedir, client_cert)
+
+        return ArtifactCacheSpec(url, push, server_cert, client_key, client_cert)
+
+
+ArtifactCacheSpec.__new__.__defaults__ = (None, None, None)
 
 
 # An ArtifactCache manages artifacts.
@@ -62,11 +78,16 @@ class ArtifactCacheSpec(namedtuple('ArtifactCacheSpec', 'url push')):
 class ArtifactCache():
     def __init__(self, context):
         self.context = context
+        self.required_artifacts = set()
         self.extractdir = os.path.join(context.artifactdir, 'extract')
+        self.max_size = context.cache_quota
+        self.estimated_size = None
+
         self.global_remote_specs = []
         self.project_remote_specs = {}
 
         self._local = False
+        self.cache_size = None
 
         os.makedirs(context.artifactdir, exist_ok=True)
 
@@ -139,6 +160,7 @@ class ArtifactCache():
     #
     # Args:
     #   config_node (dict): The config block, which may contain the 'artifacts' key
+    #   basedir (str): The base directory for relative paths
     #
     # Returns:
     #   A list of ArtifactCacheSpec instances.
@@ -147,15 +169,15 @@ class ArtifactCache():
     #   LoadError, if the config block contains invalid keys.
     #
     @staticmethod
-    def specs_from_config_node(config_node):
+    def specs_from_config_node(config_node, basedir=None):
         cache_specs = []
 
         artifacts = config_node.get('artifacts', [])
         if isinstance(artifacts, Mapping):
-            cache_specs.append(ArtifactCacheSpec._new_from_config_node(artifacts))
+            cache_specs.append(ArtifactCacheSpec._new_from_config_node(artifacts, basedir))
         elif isinstance(artifacts, list):
             for spec_node in artifacts:
-                cache_specs.append(ArtifactCacheSpec._new_from_config_node(spec_node))
+                cache_specs.append(ArtifactCacheSpec._new_from_config_node(spec_node, basedir))
         else:
             provenance = _yaml.node_get_provenance(config_node, key='artifacts')
             raise _yaml.LoadError(_yaml.LoadErrorReason.INVALID_DATA,
@@ -163,10 +185,119 @@ class ArtifactCache():
                                   (str(provenance)))
         return cache_specs
 
+    # append_required_artifacts():
+    #
+    # Append to the list of elements whose artifacts are required for
+    # the current run. Artifacts whose elements are in this list will
+    # be locked by the artifact cache and not touched for the duration
+    # of the current pipeline.
+    #
+    # Args:
+    #     elements (iterable): A set of elements to mark as required
+    #
+    def append_required_artifacts(self, elements):
+        # We lock both strong and weak keys - deleting one but not the
+        # other won't save space in most cases anyway, but would be a
+        # user inconvenience.
+
+        for element in elements:
+            strong_key = element._get_cache_key(strength=_KeyStrength.STRONG)
+            weak_key = element._get_cache_key(strength=_KeyStrength.WEAK)
+
+            for key in (strong_key, weak_key):
+                if key and key not in self.required_artifacts:
+                    self.required_artifacts.add(key)
+
+                    # We also update the usage times of any artifacts
+                    # we will be using, which helps preventing a
+                    # buildstream process that runs in parallel with
+                    # this one from removing artifacts in-use.
+                    try:
+                        self.update_atime(key)
+                    except ArtifactError:
+                        pass
+
+    # clean():
+    #
+    # Clean the artifact cache as much as possible.
+    #
+    def clean(self):
+        artifacts = self.list_artifacts()
+
+        while self.calculate_cache_size() >= self.context.cache_quota - self.context.cache_lower_threshold:
+            try:
+                to_remove = artifacts.pop(0)
+            except IndexError:
+                # If too many artifacts are required, and we therefore
+                # can't remove them, we have to abort the build.
+                #
+                # FIXME: Asking the user what to do may be neater
+                default_conf = os.path.join(os.environ['XDG_CONFIG_HOME'],
+                                            'buildstream.conf')
+                detail = ("There is not enough space to build the given element.\n"
+                          "Please increase the cache-quota in {}."
+                          .format(self.context.config_origin or default_conf))
+
+                if self.calculate_cache_size() > self.context.cache_quota:
+                    raise ArtifactError("Cache too full. Aborting.",
+                                        detail=detail,
+                                        reason="cache-too-full")
+                else:
+                    break
+
+            key = to_remove.rpartition('/')[2]
+            if key not in self.required_artifacts:
+                size = self.remove(to_remove)
+                if size:
+                    self.cache_size -= size
+
+        # This should be O(1) if implemented correctly
+        return self.calculate_cache_size()
+
+    # get_approximate_cache_size()
+    #
+    # A cheap method that aims to serve as an upper limit on the
+    # artifact cache size.
+    #
+    # The cache size reported by this function will normally be larger
+    # than the real cache size, since it is calculated using the
+    # pre-commit artifact size, but for very small artifacts in
+    # certain caches additional overhead could cause this to be
+    # smaller than, but close to, the actual size.
+    #
+    # Nonetheless, in practice this should be safe to use as an upper
+    # limit on the cache size.
+    #
+    # If the cache has built-in constant-time size reporting, please
+    # feel free to override this method with a more accurate
+    # implementation.
+    #
+    # Returns:
+    #     (int) An approximation of the artifact cache size.
+    #
+    def get_approximate_cache_size(self):
+        # If we don't currently have an estimate, figure out the real
+        # cache size.
+        if self.estimated_size is None:
+            self.estimated_size = self.calculate_cache_size()
+
+        return self.estimated_size
+
     ################################################
     # Abstract methods for subclasses to implement #
     ################################################
 
+    # update_atime()
+    #
+    # Update the atime of an artifact.
+    #
+    # Args:
+    #     key (str): The key of the artifact.
+    #
+    def update_atime(self, key):
+        raise ImplError("Cache '{kind}' does not implement contains()"
+                        .format(kind=type(self).__name__))
+
     # initialize_remotes():
     #
     # This will contact each remote cache.
@@ -192,6 +323,32 @@ class ArtifactCache():
         raise ImplError("Cache '{kind}' does not implement contains()"
                         .format(kind=type(self).__name__))
 
+    # list_artifacts():
+    #
+    # 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
+    #
+    def list_artifacts(self):
+        raise ImplError("Cache '{kind}' does not implement list_artifacts()"
+                        .format(kind=type(self).__name__))
+
+    # remove():
+    #
+    # Removes the artifact for the specified ref from the local
+    # artifact cache.
+    #
+    # Args:
+    #     ref (artifact_name): The name of the artifact to remove (as
+    #                          generated by
+    #                          `ArtifactCache.get_artifact_fullname`)
+    #
+    def remove(self, artifact_name):
+        raise ImplError("Cache '{kind}' does not implement remove()"
+                        .format(kind=type(self).__name__))
+
     # extract():
     #
     # Extract cached artifact for the specified Element if it hasn't
@@ -226,15 +383,6 @@ class ArtifactCache():
         raise ImplError("Cache '{kind}' does not implement commit()"
                         .format(kind=type(self).__name__))
 
-    # can_diff():
-    #
-    # Whether this cache implementation can diff (unfortunately
-    # there's no way to tell if an implementation is going to throw
-    # ImplError without abc).
-    #
-    def can_diff(self):
-        return False
-
     # diff():
     #
     # Return a list of files that have been added or modified between
@@ -321,6 +469,20 @@ class ArtifactCache():
         raise ImplError("Cache '{kind}' does not implement link_key()"
                         .format(kind=type(self).__name__))
 
+    # calculate_cache_size()
+    #
+    # Return the real artifact cache size.
+    #
+    # Implementations should also use this to update estimated_size.
+    #
+    # Returns:
+    #
+    # (int) The size of the artifact cache.
+    #
+    def calculate_cache_size(self):
+        raise ImplError("Cache '{kind}' does not implement calculate_cache_size()"
+                        .format(kind=type(self).__name__))
+
     ################################################
     #               Local Private Methods          #
     ################################################
@@ -357,11 +519,35 @@ class ArtifactCache():
     #
     def _initialize_remotes(self):
         def remote_failed(url, error):
-            self._message(MessageType.WARN, "Failed to fetch remote refs from {}: {}".format(url, error))
+            self._message(MessageType.WARN, "Failed to initialize remote {}: {}".format(url, error))
 
         with self.context.timed_activity("Initializing remote caches", silent_nested=True):
             self.initialize_remotes(on_failure=remote_failed)
 
+    # _add_artifact_size()
+    #
+    # Since we cannot keep track of the cache size between threads,
+    # this method will be called by the main process every time a
+    # process that added something to the cache finishes.
+    #
+    # This will then add the reported size to
+    # ArtifactCache.estimated_size.
+    #
+    def _add_artifact_size(self, artifact_size):
+        if not self.estimated_size:
+            self.estimated_size = self.calculate_cache_size()
+
+        self.estimated_size += artifact_size
+
+    # _set_cache_size()
+    #
+    # Similarly to the above method, when we calculate the actual size
+    # in a child thread, we can't update it. We instead pass the value
+    # back to the main thread and update it there.
+    #
+    def _set_cache_size(self, cache_size):
+        self.estimated_size = cache_size
+
 
 # _configured_remote_artifact_cache_specs():
 #

buildstream/_artifactcache/casserver.py

+#
+#  Copyright (C) 2018 Codethink Limited
+#
+#  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:
+#        Jürg Billeter <juerg.billeter@codethink.co.uk>
+
+from concurrent import futures
+import logging
+import os
+import signal
+import sys
+import tempfile
+
+import click
+import grpc
+
+from .._protos.build.bazel.remote.execution.v2 import remote_execution_pb2, remote_execution_pb2_grpc
+from .._protos.google.bytestream import bytestream_pb2, bytestream_pb2_grpc
+from .._protos.buildstream.v2 import buildstream_pb2, buildstream_pb2_grpc
+
+from .._exceptions import ArtifactError
+from .._context import Context
+
+from .cascache import CASCache
+
+
+# Trying to push an artifact that is too large
+class ArtifactTooLargeException(Exception):
+    pass
+
+
+# create_server():
+#
+# Create gRPC CAS artifact server as specified in the Remote Execution API.
+#
+# Args:
+#     repo (str): Path to CAS repository
+#     enable_push (bool): Whether to allow blob uploads and artifact updates
+#
+def create_server(repo, *, enable_push):
+    context = Context()
+    context.artifactdir = os.path.abspath(repo)
+
+    artifactcache = CASCache(context)
+
+    # Use max_workers default from Python 3.5+
+    max_workers = (os.cpu_count() or 1) * 5
+    server = grpc.server(futures.ThreadPoolExecutor(max_workers))
+
+    bytestream_pb2_grpc.add_ByteStreamServicer_to_server(
+        _ByteStreamServicer(artifactcache, enable_push=enable_push), server)
+
+    remote_execution_pb2_grpc.add_ContentAddressableStorageServicer_to_server(
+        _ContentAddressableStorageServicer(artifactcache), server)
+
+    buildstream_pb2_grpc.add_ReferenceStorageServicer_to_server(
+        _ReferenceStorageServicer(artifactcache, enable_push=enable_push), server)
+
+    return server
+
+
+@click.command(short_help="CAS Artifact Server")
+@click.option('--port', '-p', type=click.INT, required=True, help="Port number")
+@click.option('--server-key', help="Private server key for TLS (PEM-encoded)")
+@click.option('--server-cert', help="Public server certificate for TLS (PEM-encoded)")
+@click.option('--client-certs', help="Public client certificates for TLS (PEM-encoded)")
+@click.option('--enable-push', default=False, is_flag=True,
+              help="Allow clients to upload blobs and update artifact cache")
+@click.argument('repo')
+def server_main(repo, port, server_key, server_cert, client_certs, enable_push):
+    server = create_server(repo, enable_push=enable_push)
+
+    use_tls = bool(server_key)
+
+    if bool(server_cert) != use_tls:
+        click.echo("ERROR: --server-key and --server-cert are both required for TLS", err=True)
+        sys.exit(-1)
+
+    if client_certs and not use_tls:
+        click.echo("ERROR: --client-certs can only be used with --server-key", err=True)
+        sys.exit(-1)
+
+    if use_tls:
+        # Read public/private key pair
+        with open(server_key, 'rb') as f:
+            server_key_bytes = f.read()
+        with open(server_cert, 'rb') as f:
+            server_cert_bytes = f.read()
+
+        if client_certs:
+            with open(client_certs, 'rb') as f:
+                client_certs_bytes = f.read()
+        else:
+            client_certs_bytes = None
+
+        credentials = grpc.ssl_server_credentials([(server_key_bytes, server_cert_bytes)],
+                                                  root_certificates=client_certs_bytes,
+                                                  require_client_auth=bool(client_certs))
+        server.add_secure_port('[::]:{}'.format(port), credentials)
+    else:
+        server.add_insecure_port('[::]:{}'.format(port))
+
+    # Run artifact server
+    server.start()
+    try:
+        while True:
+            signal.pause()
+    except KeyboardInterrupt:
+        server.stop(0)
+
+
+class _ByteStreamServicer(bytestream_pb2_grpc.ByteStreamServicer):
+    def __init__(self, cas, *, enable_push):
+        super().__init__()
+        self.cas = cas
+        self.enable_push = enable_push
+
+    def Read(self, request, context):
+        resource_name = request.resource_name
+        client_digest = _digest_from_resource_name(resource_name)
+        assert request.read_offset <= client_digest.size_bytes
+
+        try:
+            with open(self.cas.objpath(client_digest), 'rb') as f:
+                assert os.fstat(f.fileno()).st_size == client_digest.size_bytes
+                if request.read_offset > 0:
+                    f.seek(request.read_offset)
+
+                remaining = client_digest.size_bytes - request.read_offset
+                while remaining > 0:
+                    chunk_size = min(remaining, 64 * 1024)
+                    remaining -= chunk_size
+
+                    response = bytestream_pb2.ReadResponse()
+                    # max. 64 kB chunks
+                    response.data = f.read(chunk_size)
+                    yield response
+        except FileNotFoundError:
+            context.set_code(grpc.StatusCode.NOT_FOUND)
+
+    def Write(self, request_iterator, context):
+        response = bytestream_pb2.WriteResponse()
+
+        if not self.enable_push:
+            context.set_code(grpc.StatusCode.PERMISSION_DENIED)
+            return response
+
+        offset = 0
+        finished = False
+        resource_name = None
+        with tempfile.NamedTemporaryFile(dir=os.path.join(self.cas.casdir, 'tmp')) as out:
+            for request in request_iterator:
+                assert not finished
+                assert request.write_offset == offset
+                if resource_name is None:
+                    # First request
+                    resource_name = request.resource_name
+                    client_digest = _digest_from_resource_name(resource_name)
+                    try:
+                        _clean_up_cache(self.cas, client_digest.size_bytes)
+                    except ArtifactTooLargeException as e:
+                        context.set_code(grpc.StatusCode.RESOURCE_EXHAUSTED)
+                        context.set_details(str(e))
+                        return response
+                elif request.resource_name:
+                    # If it is set on subsequent calls, it **must** match the value of the first request.
+                    assert request.resource_name == resource_name
+                out.write(request.data)
+                offset += len(request.data)
+                if request.finish_write:
+                    assert client_digest.size_bytes == offset
+                    out.flush()
+                    digest = self.cas.add_object(path=out.name)
+                    assert digest.hash == client_digest.hash
+                    finished = True
+
+        assert finished
+
+        response.committed_size = offset
+        return response
+
+
+class _ContentAddressableStorageServicer(remote_execution_pb2_grpc.ContentAddressableStorageServicer):
+    def __init__(self, cas):
+        super().__init__()
+        self.cas = cas
+
+    def FindMissingBlobs(self, request, context):
+        response = remote_execution_pb2.FindMissingBlobsResponse()
+        for digest in request.blob_digests:
+            if not _has_object(self.cas, digest):
+                d = response.missing_blob_digests.add()
+                d.hash = digest.hash
+                d.size_bytes = digest.size_bytes
+        return response
+
+
+class _ReferenceStorageServicer(buildstream_pb2_grpc.ReferenceStorageServicer):
+    def __init__(self, cas, *, enable_push):
+        super().__init__()
+        self.cas = cas
+        self.enable_push = enable_push
+
+    def GetReference(self, request, context):
+        response = buildstream_pb2.GetReferenceResponse()
+
+        try:
+            tree = self.cas.resolve_ref(request.key, update_mtime=True)
+
+            response.digest.hash = tree.hash
+            response.digest.size_bytes = tree.size_bytes
+        except ArtifactError:
+            context.set_code(grpc.StatusCode.NOT_FOUND)
+
+        return response
+
+    def UpdateReference(self, request, context):
+        response = buildstream_pb2.UpdateReferenceResponse()
+
+        if not self.enable_push:
+            context.set_code(grpc.StatusCode.PERMISSION_DENIED)
+            return response
+
+        for key in request.keys:
+            self.cas.set_ref(key, request.digest)
+
+        return response
+
+    def Status(self, request, context):
+        response = buildstream_pb2.StatusResponse()
+
+        response.allow_updates = self.enable_push
+
+        return response
+
+
+def _digest_from_resource_name(resource_name):
+    parts = resource_name.split('/')
+    assert len(parts) == 2
+    digest = remote_execution_pb2.Digest()
+    digest.hash = parts[0]
+    digest.size_bytes = int(parts[1])
+    return digest
+
+
+def _has_object(cas, digest):
+    objpath = cas.objpath(digest)
+    return os.path.exists(objpath)
+
+
+# _clean_up_cache()
+#
+# Keep removing Least Recently Pushed (LRP) artifacts in a cache until there
+# is enough space for the incoming artifact
+#
+# Args:
+#   cas: CASCache object
+#   object_size: The size of the object being received in bytes
+#
+# Returns:
+#   int: The total bytes removed on the filesystem
+#
+def _clean_up_cache(cas, object_size):
+    # Determine the available disk space, in bytes, of the file system
+    # which mounts the repo
+    stats = os.statvfs(cas.casdir)
+    buffer_ = int(2e9)                # Add a 2 GB buffer
+    free_disk_space = (stats.f_bfree * stats.f_bsize) - buffer_
+    total_disk_space = (stats.f_blocks * stats.f_bsize) - buffer_
+
+    if object_size > total_disk_space:
+        raise ArtifactTooLargeException("Artifact of size: {} is too large for "
+                                        "the filesystem which mounts the remote "
+                                        "cache".format(object_size))
+
+    if object_size <= free_disk_space:
+        # No need to clean up
+        return 0
+
+    # obtain a list of LRP artifacts
+    LRP_artifacts = cas.list_artifacts()
+
+    removed_size = 0  # in bytes
+    while object_size - removed_size > free_disk_space:
+        try:
+            to_remove = LRP_artifacts.pop(0)  # The first element in the list is the LRP artifact
+        except IndexError:
+            # This exception is caught if there are no more artifacts in the list
+            # LRP_artifacts. This means the the artifact is too large for the filesystem
+            # so we abort the process
+            raise ArtifactTooLargeException("Artifact of size {} is too large for "
+                                            "the filesystem which mounts the remote "
+                                            "cache".format(object_size))
+
+        removed_size += cas.remove(to_remove, defer_prune=False)
+
+    if removed_size > 0:
+        logging.info("Successfully removed {} bytes from the cache".format(removed_size))
+    else:
+        logging.info("No artifacts were removed from the cache.")
+
+    return removed_size

buildstream/_artifactcache/ostreecache.py

-#!/usr/bin/env python3
-#
-#  Copyright (C) 2017-2018 Codethink Limited
-#
-#  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:
-#        Jürg Billeter <juerg.billeter@codethink.co.uk>
-
-import multiprocessing
-import os
-import signal
-import tempfile
-
-from .. import _ostree, _signals, utils
-from .._exceptions import ArtifactError
-from .._ostree import OSTreeError
-
-from . import ArtifactCache
-from .pushreceive import initialize_push_connection
-from .pushreceive import push as push_artifact
-from .pushreceive import PushException
-
-
-# An OSTreeCache manages artifacts in an OSTree repository
-#
-# Args:
-#     context (Context): The BuildStream context
-#     project (Project): The BuildStream project
-#     enable_push (bool): Whether pushing is allowed by the platform
-#
-# Pushing is explicitly disabled by the platform in some cases,
-# like when we are falling back to functioning without using
-# user namespaces.
-#
-class OSTreeCache(ArtifactCache):
-
-    def __init__(self, context, *, enable_push):
-        super().__init__(context)
-
-        self.enable_push = enable_push
-
-        ostreedir = os.path.join(context.artifactdir, 'ostree')
-        self.repo = _ostree.ensure(ostreedir, False)
-
-        # Per-project list of OSTreeRemote instances.
-        self._remotes = {}
-
-        self._has_fetch_remotes = False
-        self._has_push_remotes = False
-
-    ################################################
-    #     Implementation of abstract methods       #
-    ################################################
-    def has_fetch_remotes(self, *, element=None):
-        if not self._has_fetch_remotes:
-            # No project has push remotes
-            return False
-        elif element is None:
-            # At least one (sub)project has fetch remotes
-            return True
-        else:
-            # Check whether the specified element's project has fetch remotes
-            remotes_for_project = self._remotes[element._get_project()]
-            return bool(remotes_for_project)
-
-    def has_push_remotes(self, *, element=None):
-        if not self._has_push_remotes:
-            # No project has push remotes
-            return False
-        elif element is None:
-            # At least one (sub)project has push remotes
-            return True
-        else:
-            # Check whether the specified element's project has push remotes
-            remotes_for_project = self._remotes[element._get_project()]
-            return any(remote.spec.push for remote in remotes_for_project)
-
-    def contains(self, element, key):
-        ref = self.get_artifact_fullname(element, key)
-        return _ostree.exists(self.repo, ref)
-
-    def extract(self, element, key):
-        ref = self.get_artifact_fullname(element, key)
-
-        # resolve ref to checksum
-        rev = _ostree.checksum(self.repo, ref)
-
-        # Extracting a nonexistent artifact is a bug
-        assert rev, "Artifact missing for {}".format(ref)
-
-        dest = os.path.join(self.extractdir, element._get_project().name, element.normal_name, rev)
-        if os.path.isdir(dest):
-            # artifact has already been extracted
-            return dest
-
-        os.makedirs(self.extractdir, exist_ok=True)
-        with tempfile.TemporaryDirectory(prefix='tmp', dir=self.extractdir) as tmpdir:
-
-            checkoutdir = os.path.join(tmpdir, ref)
-
-            _ostree.checkout(self.repo, checkoutdir, rev, user=True)
-
-            os.makedirs(os.path.dirname(dest), exist_ok=True)
-            try:
-                os.rename(checkoutdir, dest)
-            except OSError as e:
-                # With rename, it's possible to get either ENOTEMPTY or EEXIST
-                # in the case that the destination path is a not empty directory.
-                #
-                # If rename fails with these errors, another process beat
-                # us to it so just ignore.
-                if e.errno not in [os.errno.ENOTEMPTY, os.errno.EEXIST]:
-                    raise ArtifactError("Failed to extract artifact for ref '{}': {}"
-                                        .format(ref, e)) from e
-
-        return dest
-
-    def commit(self, element, content, keys):
-        refs = [self.get_artifact_fullname(element, key) for key in keys]
-
-        try:
-            _ostree.commit(self.repo, content, refs)
-        except OSTreeError as e:
-            raise ArtifactError("Failed to commit artifact: {}".format(e)) from e
-
-    def can_diff(self):
-        return True
-
-    def diff(self, element, key_a, key_b, *, subdir=None):
-        _, a, _ = self.repo.read_commit(self.get_artifact_fullname(element, key_a))
-        _, b, _ = self.repo.read_commit(self.get_artifact_fullname(element, key_b))
-
-        if subdir:
-            a = a.get_child(subdir)
-            b = b.get_child(subdir)
-
-            subpath = a.get_path()
-        else:
-            subpath = '/'
-
-        modified, removed, added = _ostree.diff_dirs(a, b)
-
-        modified = [os.path.relpath(item.target.get_path(), subpath) for item in modified]
-        removed = [os.path.relpath(item.get_path(), subpath) for item in removed]
-        added = [os.path.relpath(item.get_path(), subpath) for item in added]
-
-        return modified, removed, added
-
-    def pull(self, element, key, *, progress=None):
-        project = element._get_project()
-
-        ref = self.get_artifact_fullname(element, key)
-
-        for remote in self._remotes[project]:
-            try:
-                # fetch the artifact from highest priority remote using the specified cache key
-                remote_name = self._ensure_remote(self.repo, remote.pull_url)
-                _ostree.fetch(self.repo, remote=remote_name, ref=ref, progress=progress)
-                return True
-            except OSTreeError:
-                # Try next remote
-                continue
-
-        return False
-
-    def link_key(self, element, oldkey, newkey):
-        oldref = self.get_artifact_fullname(element, oldkey)
-        newref = self.get_artifact_fullname(element, newkey)
-
-        # resolve ref to checksum
-        rev = _ostree.checksum(self.repo, oldref)
-
-        # create additional ref for the same checksum
-        _ostree.set_ref(self.repo, newref, rev)
-
-    def push(self, element, keys):
-        any_pushed = False
-
-        project = element._get_project()
-
-        push_remotes = [r for r in self._remotes[project] if r.spec.push]
-
-        if not push_remotes:
-            raise ArtifactError("Push is not enabled for any of the configured remote artifact caches.")
-
-        refs = [self.get_artifact_fullname(element, key) for key in keys]
-
-        for remote in push_remotes:
-            any_pushed |= self._push_to_remote(remote, element, refs)
-
-        return any_pushed
-
-    def initialize_remotes(self, *, on_failure=None):
-        remote_specs = self.global_remote_specs.copy()
-
-        for project in self.project_remote_specs:
-            remote_specs.extend(self.project_remote_specs[project])
-
-        remote_specs = list(utils._deduplicate(remote_specs))
-
-        remote_results = {}
-
-        # Callback to initialize one remote in a 'multiprocessing' subprocess.
-        #
-        # We cannot do this in the main process because of the way the tasks
-        # run by the main scheduler calls into libostree using
-        # fork()-without-exec() subprocesses. OSTree fetch operations in
-        # subprocesses hang if fetch operations were previously done in the
-        # main process.
-        #
-        def child_action(url, q):
-            try:
-                push_url, pull_url = self._initialize_remote(url)
-                q.put((None, push_url, pull_url))
-            except Exception as e:               # pylint: disable=broad-except
-                # Whatever happens, we need to return it to the calling process
-                #
-                q.put((str(e), None, None, None))
-
-        # Kick off all the initialization jobs one by one.
-        #
-        # Note that we cannot use multiprocessing.Pool here because it's not
-        # possible to pickle local functions such as child_action().
-        #
-        q = multiprocessing.Queue()
-        for remote_spec in remote_specs:
-            p = multiprocessing.Process(target=child_action, args=(remote_spec.url, q))
-
-            try:
-
-                # Keep SIGINT blocked in the child process
-                with _signals.blocked([signal.SIGINT], ignore=False):
-                    p.start()
-
-                error, push_url, pull_url = q.get()
-                p.join()
-            except KeyboardInterrupt:
-                utils._kill_process_tree(p.pid)
-                raise
-
-            if error and on_failure:
-                on_failure(remote_spec.url, error)
-            elif error:
-                raise ArtifactError(error)
-            else:
-                if remote_spec.push and push_url:
-                    self._has_push_remotes = True
-                if pull_url:
-                    self._has_fetch_remotes = True
-
-                remote_results[remote_spec.url] = (push_url, pull_url)
-
-        # Prepare push_urls and pull_urls for each project
-        for project in self.context.get_projects():
-            remote_specs = self.global_remote_specs
-            if project in self.project_remote_specs:
-                remote_specs = list(utils._deduplicate(remote_specs + self.project_remote_specs[project]))
-
-            remotes = []
-
-            for remote_spec in remote_specs:
-                # Errors are already handled in the loop above,
-                # skip unreachable remotes here.
-                if remote_spec.url not in remote_results:
-                    continue
-
-                push_url, pull_url = remote_results[remote_spec.url]
-
-                if remote_spec.push and not push_url:
-                    raise ArtifactError("Push enabled but not supported by repo at: {}".format(remote_spec.url))
-
-                remote = _OSTreeRemote(remote_spec, pull_url, push_url)
-                remotes.append(remote)
-
-            self._remotes[project] = remotes
-
-    ################################################
-    #             Local Private Methods            #
-    ################################################
-
-    # _initialize_remote():
-    #
-    # Do protocol-specific initialization necessary to use a given OSTree
-    # remote.
-    #
-    # The SSH protocol that we use only supports pushing so initializing these
-    # involves contacting the remote to find out the corresponding pull URL.
-    #
-    # Args:
-    #     url (str): URL of the remote
-    #
-    # Returns:
-    #     (str, str): the pull URL and push URL for the remote
-    #
-    # Raises:
-    #     ArtifactError: if there was an error
-    def _initialize_remote(self, url):
-        if url.startswith('ssh://'):
-            try:
-                push_url = url
-                pull_url = initialize_push_connection(url)
-            except PushException as e:
-                raise ArtifactError(e) from e
-        elif url.startswith('/'):
-            push_url = pull_url = 'file://' + url
-        elif url.startswith('file://'):
-            push_url = pull_url = url
-        elif url.startswith('http://') or url.startswith('https://'):
-            push_url = None
-            pull_url = url
-        else:
-            raise ArtifactError("Unsupported URL: {}".format(url))
-
-        return push_url, pull_url
-
-    # _ensure_remote():
-    #
-    # Ensure that our OSTree repo has a remote configured for the given URL.
-    # Note that SSH access to remotes is not handled by libostree itself.
-    #
-    # Args:
-    #     repo (OSTree.Repo): an OSTree repository
-    #     pull_url (str): the URL where libostree can pull from the remote
-    #
-    # Returns:
-    #     (str): the name of the remote, which can be passed to various other
-    #            operations implemented by the _ostree module.
-    #
-    # Raises:
-    #     OSTreeError: if there was a problem reported by libostree
-    def _ensure_remote(self, repo, pull_url):
-        remote_name = utils.url_directory_name(pull_url)
-        _ostree.configure_remote(repo, remote_name, pull_url)
-        return remote_name
-
-    def _push_to_remote(self, remote, element, refs):
-        with utils._tempdir(dir=self.context.artifactdir, prefix='push-repo-') as temp_repo_dir:
-
-            with element.timed_activity("Preparing compressed archive"):
-                # First create a temporary archive-z2 repository, we can
-                # only use ostree-push with archive-z2 local repo.
-                temp_repo = _ostree.ensure(temp_repo_dir, True)
-
-                # Now push the ref we want to push into our temporary archive-z2 repo
-                for ref in refs:
-                    _ostree.fetch(temp_repo, remote=self.repo.get_path().get_uri(), ref=ref)
-
-            with element.timed_activity("Sending artifact"), \
-                element._output_file() as output_file:
-                try:
-                    pushed = push_artifact(temp_repo.get_path().get_path(),
-                                           remote.push_url,
-                                           refs, output_file)
-                except PushException as e:
-                    raise ArtifactError("Failed to push artifact {}: {}".format(refs, e)) from e
-
-            return pushed
-
-
-# Represents a single remote OSTree cache.
-#
-class _OSTreeRemote():
-    def __init__(self, spec, pull_url, push_url):
-        self.spec = spec
-        self.pull_url = pull_url
-        self.push_url = push_url

buildstream/_artifactcache/tarcache.py

-#!/usr/bin/env python3
-#
-#  Copyright (C) 2017 Codethink Limited
-#
-#  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:
-#        Tristan Maat <tristan.maat@codethink.co.uk>
-
-import os
-import shutil
-import tarfile
-import subprocess
-
-from .. import utils, ProgramNotFoundError
-from .._exceptions import ArtifactError
-
-from . import ArtifactCache
-
-
-class TarCache(ArtifactCache):
-
-    def __init__(self, context):
-
-        super().__init__(context)
-
-        self.tardir = os.path.join(context.artifactdir, 'tar')
-        os.makedirs(self.tardir, exist_ok=True)
-
-    ################################################
-    #     Implementation of abstract methods       #
-    ################################################
-    def contains(self, element, key):
-        path = os.path.join(self.tardir, _tarpath(element, key))
-        return os.path.isfile(path)
-
-    def commit(self, element, content, keys):
-        os.makedirs(os.path.join(self.tardir, element._get_project().name, element.normal_name), exist_ok=True)
-
-        with utils._tempdir() as temp:
-            for key in keys:
-                ref = _tarpath(element, key)
-
-                refdir = os.path.join(temp, key)
-                shutil.copytree(content, refdir, symlinks=True)
-
-                _Tar.archive(os.path.join(self.tardir, ref), key, temp)
-
-    def extract(self, element, key):
-
-        fullname = self.get_artifact_fullname(element, key)
-        path = _tarpath(element, key)
-
-        # Extracting a nonexistent artifact is a bug
-        assert os.path.isfile(os.path.join(self.tardir, path)), "Artifact missing for {}".format(fullname)
-
-        # If the destination already exists, the artifact has been extracted
-        dest = os.path.join(self.extractdir, fullname)
-        if os.path.isdir(dest):
-            return dest
-
-        os.makedirs(self.extractdir, exist_ok=True)
-
-        with utils._tempdir(dir=self.extractdir) as tmpdir:
-            _Tar.extract(os.path.join(self.tardir, path), tmpdir)
-
-            os.makedirs(os.path.join(self.extractdir, element._get_project().name, element.normal_name),
-                        exist_ok=True)
-            try:
-                os.rename(os.path.join(tmpdir, key), dest)
-            except OSError as e:
-                # With rename, it's possible to get either ENOTEMPTY or EEXIST
-                # in the case that the destination path is a not empty directory.
-                #
-                # If rename fails with these errors, another process beat
-                # us to it so just ignore.
-                if e.errno not in [os.errno.ENOTEMPTY, os.errno.EEXIST]:
-                    raise ArtifactError("Failed to extract artifact '{}': {}"
-                                        .format(fullname, e)) from e
-
-        return dest
-
-
-# _tarpath()
-#
-# Generate a relative tarball path for a given element and it's cache key
-#
-# Args:
-#    element (Element): The Element object
-#    key (str): The element's cache key
-#
-# Returns:
-#    (str): The relative path to use for storing tarballs
-#
-def _tarpath(element, key):
-    project = element._get_project()
-    return os.path.join(project.name, element.normal_name, key + '.tar.bz2')
-
-
-# A helper class that contains tar archive/extract functions
-class _Tar():
-
-    # archive()
-    #
-    # Attempt to archive the given tarfile with the `tar` command,
-    # falling back to python's `tarfile` if this fails.
-    #
-    # Args:
-    #     location (str): The path to the tar to create
-    #     content (str): The path to the content to archvive
-    #     cwd (str): The cwd
-    #
-    # This is done since AIX tar does not support 2G+ files.
-    #
-    @classmethod
-    def archive(cls, location, content, cwd=os.getcwd()):
-
-        try:
-            cls._archive_with_tar(location, content, cwd)
-            return
-        except tarfile.TarError:
-            pass
-        except ProgramNotFoundError:
-            pass
-
-        # If the former did not complete successfully, we try with
-        # python's tar implementation (since it's hard to detect
-        # specific issues with specific tar implementations - a
-        # fallback).
-
-        try:
-            cls._archive_with_python(location, content, cwd)
-        except tarfile.TarError as e:
-            raise ArtifactError("Failed to archive {}: {}"
-                                .format(location, e)) from e
-
-    # extract()
-    #
-    # Attempt to extract the given tarfile with the `tar` command,
-    # falling back to python's `tarfile` if this fails.
-    #
-    # Args:
-    #     location (str): The path to the tar to extract
-    #     dest (str): The destination path to extract to
-    #
-    # This is done since python tarfile extraction is horrendously
-    # slow (2 hrs+ for base images).
-    #
-    @classmethod
-    def extract(cls, location, dest):
-
-        try:
-            cls._extract_with_tar(location, dest)
-            return
-        except tarfile.TarError:
-            pass
-        except ProgramNotFoundError:
-            pass
-
-        try:
-            cls._extract_with_python(location, dest)
-        except tarfile.TarError as e:
-            raise ArtifactError("Failed to extract {}: {}"
-                                .format(location, e)) from e
-
-    # _get_host_tar()
-    #
-    # Get the host tar command.
-    #
-    # Raises:
-    #     ProgramNotFoundError: If the tar executable cannot be
-    #                           located
-    #
-    @classmethod
-    def _get_host_tar(cls):
-        tar_cmd = None
-
-        for potential_tar_cmd in ['gtar', 'tar']:
-            try:
-                tar_cmd = utils.get_host_tool(potential_tar_cmd)
-                break
-            except ProgramNotFoundError:
-                continue
-
-        # If we still couldn't find tar, raise the ProgramNotfounderror
-        if tar_cmd is None:
-            raise ProgramNotFoundError("Did not find tar in PATH: {}"
-                                       .format(os.environ.get('PATH')))
-
-        return tar_cmd
-
-    # _archive_with_tar()
-    #
-    # Archive with an implementation of the `tar` command
-    #
-    # Args:
-    #     location (str): The path to the tar to create
-    #     content (str): The path to the content to archvive
-    #     cwd (str): The cwd
-    #
-    # Raises:
-    #     TarError: If an error occurs during extraction
-    #     ProgramNotFoundError: If the tar executable cannot be
-    #                           located
-    #
-    @classmethod
-    def _archive_with_tar(cls, location, content, cwd):
-        tar_cmd = cls._get_host_tar()
-
-        process = subprocess.Popen(
-            [tar_cmd, 'jcaf', location, content],
-            cwd=cwd,
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE
-        )
-
-        _, err = process.communicate()
-        if process.poll() != 0:
-            # Clean up in case the command failed in a broken state
-            try:
-                os.remove(location)
-            except FileNotFoundError:
-                pass
-
-            raise tarfile.TarError("Failed to archive '{}': {}"
-                                   .format(content, err.decode('utf8')))
-
-    # _archive_with_python()
-    #
-    # Archive with the python `tarfile` module
-    #
-    # Args:
-    #     location (str): The path to the tar to create
-    #     content (str): The path to the content to archvive
-    #     cwd (str): The cwd
-    #
-    # Raises:
-    #     TarError: If an error occurs during extraction
-    #
-    @classmethod
-    def _archive_with_python(cls, location, content, cwd):
-        with tarfile.open(location, mode='w:bz2') as tar:
-            tar.add(os.path.join(cwd, content), arcname=content)
-
-    # _extract_with_tar()
-    #
-    # Extract with an implementation of the `tar` command
-    #
-    # Args:
-    #     location (str): The path to the tar to extract
-    #     dest (str): The destination path to extract to
-    #
-    # Raises:
-    #     TarError: If an error occurs during extraction
-    #
-    @classmethod
-    def _extract_with_tar(cls, location, dest):
-        tar_cmd = cls._get_host_tar()
-
-        # Some tar implementations do not support '-C'
-        process = subprocess.Popen(
-            [tar_cmd, 'jxf', location],
-            cwd=dest,
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE
-        )
-
-        _, err = process.communicate()
-        if process.poll() != 0:
-            raise tarfile.TarError("Failed to extract '{}': {}"
-                                   .format(location, err.decode('utf8')))
-
-    # _extract_with_python()
-    #
-    # Extract with the python `tarfile` module
-    #
-    # Args:
-    #     location (str): The path to the tar to extract
-    #     dest (str): The destination path to extract to
-    #
-    # Raises:
-    #     TarError: If an error occurs during extraction
-    #
-    @classmethod
-    def _extract_with_python(cls, location, dest):
-        with tarfile.open(location) as tar:
-            tar.extractall(path=dest)

buildstream/_cachekey.py

-#!/usr/bin/env python3
 #
 #  Copyright (C) 2018 Codethink Limited
 #

buildstream/_context.py

-#!/usr/bin/env python3
 #
 #  Copyright (C) 2016-2018 Codethink Limited
 #
@@ -22,6 +21,7 @@ import os
 import datetime
 from collections import deque, Mapping
 from contextlib import contextmanager
+from . import utils
 from . import _cachekey
 from . import _signals
 from . import _site
@@ -30,6 +30,8 @@ from ._exceptions import LoadError, LoadErrorReason, BstError
 from ._message import Message, MessageType
 from ._profile import Topics, profile_start, profile_end
 from ._artifactcache import ArtifactCache
+from ._workspaces import Workspaces
+from .plugin import _plugin_lookup
 
 
 # Context()
@@ -62,6 +64,12 @@ class Context():
         # The locations from which to push and pull prebuilt artifacts
         self.artifact_cache_specs = []
 
+        # The artifact cache quota
+        self.cache_quota = None
+
+        # The lower threshold to which we aim to reduce the cache size
+        self.cache_lower_threshold = None
+
         # The directory to store build logs
         self.logdir = None
 
@@ -113,6 +121,9 @@ class Context():
         self._message_depth = deque()
         self._projects = []
         self._project_overrides = {}
+        self._workspaces = None
+        self._log_handle = None
+        self._log_filename = None
 
     # load()
     #
@@ -152,6 +163,7 @@ class Context():
         _yaml.node_validate(defaults, [
             'sourcedir', 'builddir', 'artifactdir', 'logdir',
             'scheduler', 'artifacts', 'logging', 'projects',
+            'cache'
         ])
 
         for directory in ['sourcedir', 'builddir', 'artifactdir', 'logdir']:
@@ -161,8 +173,82 @@ class Context():
             path = _yaml.node_get(defaults, str, directory)
             path = os.path.expanduser(path)
             path = os.path.expandvars(path)
+            path = os.path.normpath(path)
             setattr(self, directory, path)
 
+        # Load quota configuration
+        # We need to find the first existing directory in the path of
+        # our artifactdir - the artifactdir may not have been created
+        # yet.
+        cache = _yaml.node_get(defaults, Mapping, 'cache')
+        _yaml.node_validate(cache, ['quota'])
+
+        artifactdir_volume = self.artifactdir
+        while not os.path.exists(artifactdir_volume):
+            artifactdir_volume = os.path.dirname(artifactdir_volume)
+
+        # We read and parse the cache quota as specified by the user
+        cache_quota = _yaml.node_get(cache, str, 'quota', default_value='infinity')
+        try:
+            cache_quota = utils._parse_size(cache_quota, artifactdir_volume)
+        except utils.UtilError as e:
+            raise LoadError(LoadErrorReason.INVALID_DATA,
+                            "{}\nPlease specify the value in bytes or as a % of full disk space.\n"
+                            "\nValid values are, for example: 800M 10G 1T 50%\n"
+                            .format(str(e))) from e
+
+        # Headroom intended to give BuildStream a bit of leeway.
+        # This acts as the minimum size of cache_quota and also
+        # is taken from the user requested cache_quota.
+        #
+        if 'BST_TEST_SUITE' in os.environ:
+            headroom = 0
+        else:
+            headroom = 2e9
+
+        stat = os.statvfs(artifactdir_volume)
+        available_space = (stat.f_bsize * stat.f_bavail)
+
+        # Again, the artifact directory may not yet have been created yet
+        #
+        if not os.path.exists(self.artifactdir):
+            cache_size = 0
+        else:
+            cache_size = utils._get_dir_size(self.artifactdir)
+
+        # Ensure system has enough storage for the cache_quota
+        #
+        # If cache_quota is none, set it to the maximum it could possibly be.
+        #
+        # Also check that cache_quota is atleast as large as our headroom.
+        #
+        if cache_quota is None:  # Infinity, set to max system storage
+            cache_quota = cache_size + available_space
+        if cache_quota < headroom:  # Check minimum
+            raise LoadError(LoadErrorReason.INVALID_DATA,
+                            "Invalid cache quota ({}): ".format(utils._pretty_size(cache_quota)) +
+                            "BuildStream requires a minimum cache quota of 2G.")
+        elif cache_quota > cache_size + available_space:  # Check maximum
+            raise LoadError(LoadErrorReason.INVALID_DATA,
+                            ("Your system does not have enough available " +
+                             "space to support the cache quota specified.\n" +
+                             "You currently have:\n" +
+                             "- {used} of cache in use at {local_cache_path}\n" +
+                             "- {available} of available system storage").format(
+                                 used=utils._pretty_size(cache_size),
+                                 local_cache_path=self.artifactdir,
+                                 available=utils._pretty_size(available_space)))
+
+        # Place a slight headroom (2e9 (2GB) on the cache_quota) into
+        # cache_quota to try and avoid exceptions.
+        #
+        # Of course, we might still end up running out during a build
+        # if we end up writing more than 2G, but hey, this stuff is
+        # already really fuzzy.
+        #
+        self.cache_quota = cache_quota - headroom
+        self.cache_lower_threshold = self.cache_quota / 2
+
         # Load artifact share configuration
         self.artifact_cache_specs = ArtifactCache.specs_from_config_node(defaults)
 
@@ -199,7 +285,7 @@ class Context():
         # Shallow validation of overrides, parts of buildstream which rely
         # on the overrides are expected to validate elsewhere.
         for _, overrides in _yaml.node_items(self._project_overrides):
-            _yaml.node_validate(overrides, ['artifacts', 'options', 'strict'])
+            _yaml.node_validate(overrides, ['artifacts', 'options', 'strict', 'default-mirror'])
 
         profile_end(Topics.LOAD_CONTEXT, 'load')
 
@@ -218,6 +304,8 @@ class Context():
     #    project (Project): The project to add
     #
     def add_project(self, project):
+        if not self._projects:
+            self._workspaces = Workspaces(project)
         self._projects.append(project)
 
     # get_projects():
@@ -241,6 +329,9 @@ class Context():
     def get_toplevel_project(self):
         return self._projects[0]
 
+    def get_workspaces(self):
+        return self._workspaces
+
     # get_overrides():
     #
     # Fetch the override dictionary for the active project. This returns
@@ -323,8 +414,11 @@ class Context():
         if message.depth is None:
             message.depth = len(list(self._message_depth))
 
+        # If we are recording messages, dump a copy into the open log file.
+        self._record_message(message)
+
         # Send it off to the log handler (can be the frontend,
-        # or it can be the child task which will log and propagate
+        # or it can be the child task which will propagate
         # to the frontend)
         assert self._message_handler
 
@@ -394,6 +488,137 @@ class Context():
             self._pop_message_depth()
             self.message(message)
 
+    # recorded_messages()
+    #
+    # Records all messages in a log file while the context manager
+    # is active.
+    #
+    # In addition to automatically writing all messages to the
+    # specified logging file, an open file handle for process stdout
+    # and stderr will be available via the Context.get_log_handle() API,
+    # and the full logfile path will be available via the
+    # Context.get_log_filename() API.
+    #
+    # Args:
+    #     filename (str): A logging directory relative filename,
+    #                     the pid and .log extension will be automatically
+    #                     appended
+    #
+    # Yields:
+    #     (str): The fully qualified log filename
+    #
+    @contextmanager
+    def recorded_messages(self, filename):
+
+        # We dont allow recursing in this context manager, and
+        # we also do not allow it in the main process.
+        assert self._log_handle is None
+        assert self._log_filename is None
+        assert not utils._is_main_process()
+
+        # Create the fully qualified logfile in the log directory,
+        # appending the pid and .log extension at the end.
+        self._log_filename = os.path.join(self.logdir,
+                                          '{}.{}.log'.format(filename, os.getpid()))
+
+        # Ensure the directory exists first
+        directory = os.path.dirname(self._log_filename)
+        os.makedirs(directory, exist_ok=True)
+
+        with open(self._log_filename, 'a') as logfile:
+
+            # Write one last line to the log and flush it to disk
+            def flush_log():
+
+                # If the process currently had something happening in the I/O stack
+                # then trying to reenter the I/O stack will fire a runtime error.
+                #
+                # So just try to flush as well as we can at SIGTERM time
+                try:
+                    logfile.write('\n\nForcefully terminated\n')
+                    logfile.flush()
+                except RuntimeError:
+                    os.fsync(logfile.fileno())
+
+            self._log_handle = logfile
+            with _signals.terminator(flush_log):
+                yield self._log_filename
+
+            self._log_handle = None
+            self._log_filename = None
+
+    # get_log_handle()
+    #
+    # Fetches the active log handle, this will return the active
+    # log file handle when the Context.recorded_messages() context
+    # manager is active
+    #
+    # Returns:
+    #     (file): The active logging file handle, or None
+    #
+    def get_log_handle(self):
+        return self._log_handle
+
+    # get_log_filename()
+    #
+    # Fetches the active log filename, this will return the active
+    # log filename when the Context.recorded_messages() context
+    # manager is active
+    #
+    # Returns:
+    #     (str): The active logging filename, or None
+    #
+    def get_log_filename(self):
+        return self._log_filename
+
+    # _record_message()
+    #
+    # Records the message if recording is enabled
+    #
+    # Args:
+    #    message (Message): The message to record
+    #
+    def _record_message(self, message):
+
+        if self._log_handle is None:
+            return
+
+        INDENT = "    "
+        EMPTYTIME = "--:--:--"
+        template = "[{timecode: <8}] {type: <7}"
+
+        # If this message is associated with a plugin, print what
+        # we know about the plugin.
+        plugin_name = ""
+        if message.unique_id:
+            template += " {plugin}"
+            plugin = _plugin_lookup(message.unique_id)
+            plugin_name = plugin.name
+
+        template += ": {message}"
+
+        detail = ''
+        if message.detail is not None:
+            template += "\n\n{detail}"
+            detail = message.detail.rstrip('\n')
+            detail = INDENT + INDENT.join(detail.splitlines(True))
+
+        timecode = EMPTYTIME
+        if message.message_type in (MessageType.SUCCESS, MessageType.FAIL):
+            hours, remainder = divmod(int(message.elapsed.total_seconds()), 60**2)
+            minutes, seconds = divmod(remainder, 60)
+            timecode = "{0:02d}:{1:02d}:{2:02d}".format(hours, minutes, seconds)
+
+        text = template.format(timecode=timecode,
+                               plugin=plugin_name,
+                               type=message.message_type.upper(),
+                               message=message.message,
+                               detail=detail)
+
+        # Write to the open log file
+        self._log_handle.write('{}\n'.format(text))
+        self._log_handle.flush()
+
     # _push_message_depth() / _pop_message_depth()
     #
     # For status messages, send the depth of timed

buildstream/_elementfactory.py

-#!/usr/bin/env python3
 #
 #  Copyright (C) 2016 Codethink Limited
 #

buildstream/_exceptions.py

-#!/usr/bin/env python3
 #
 #  Copyright (C) 2016 Codethink Limited
 #
@@ -100,7 +99,7 @@ class ErrorDomain(Enum):
 #
 class BstError(Exception):
 
-    def __init__(self, message, *, detail=None, domain=None, reason=None):
+    def __init__(self, message, *, detail=None, domain=None, reason=None, temporary=False):
         global _last_exception
 
         super().__init__(message)
@@ -115,6 +114,11 @@ class BstError(Exception):
         #
         self.sandbox = None
 
+        # When this exception occurred during the handling of a job, indicate
+        # whether or not there is any point retrying the job.
+        #
+        self.temporary = temporary
+
         # Error domain and reason
         #
         self.domain = domain
@@ -132,8 +136,8 @@ class BstError(Exception):
 # or by the base :class:`.Plugin` element itself.
 #
 class PluginError(BstError):
-    def __init__(self, message, reason=None):
-        super().__init__(message, domain=ErrorDomain.PLUGIN, reason=reason)
+    def __init__(self, message, reason=None, temporary=False):
+        super().__init__(message, domain=ErrorDomain.PLUGIN, reason=reason, temporary=False)
 
 
 # LoadErrorReason
@@ -198,6 +202,9 @@ class LoadErrorReason(Enum):
     # A project.conf file was missing
     MISSING_PROJECT_CONF = 17
 
+    # Try to load a directory not a yaml file
+    LOADING_DIRECTORY = 18
+
 
 # LoadError
 #
@@ -247,8 +254,8 @@ class SandboxError(BstError):
 # Raised when errors are encountered in the artifact caches
 #
 class ArtifactError(BstError):
-    def __init__(self, message, reason=None):
-        super().__init__(message, domain=ErrorDomain.ARTIFACT, reason=reason)
+    def __init__(self, message, *, detail=None, reason=None, temporary=False):
+        super().__init__(message, detail=detail, domain=ErrorDomain.ARTIFACT, reason=reason, temporary=True)
 
 
 # PipelineError