_sourcecache.py: Introduce SourceCache class

Introduces source cache which allows sources to be staged into the local
CAS for use in builds. Registers methods with CASQuota in order to allow
for the clean method to remove unrequired sources as well.

Part of #440

2 files changed, 154 insertions, 0 deletions

diff --git a/buildstream/_exceptions.py b/buildstream/_exceptions.py
index 8728f6e69..f2d34bcba 100644
--- a/buildstream/_exceptions.py
+++ b/buildstream/_exceptions.py
@@ -272,6 +272,15 @@ class SandboxError(BstError):
         super().__init__(message, detail=detail, domain=ErrorDomain.SANDBOX, reason=reason)
 
 
+# SourceCacheError
+#
+# Raised when errors are encountered in the source caches
+#
+class SourceCacheError(BstError):
+    def __init__(self, message, detail=None, reason=None):
+        super().__init__(message, detail=detail, domain=ErrorDomain.SANDBOX, reason=reason)
+
+
 # ArtifactError
 #
 # Raised when errors are encountered in the artifact caches
diff --git a/buildstream/_sourcecache.py b/buildstream/_sourcecache.py
new file mode 100644
index 000000000..b21edaa81
--- /dev/null
+++ b/buildstream/_sourcecache.py
@@ -0,0 +1,145 @@
+#
+#  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:
+#        Raoul Hidalgo Charman <raoul.hidalgocharman@codethink.co.uk>
+#
+from ._cas import CASRemoteSpec
+from .storage._casbaseddirectory import CasBasedDirectory
+from ._basecache import BaseCache
+from ._exceptions import CASCacheError, SourceCacheError
+from . import utils
+
+
+# Holds configuration for a remote used for the source cache.
+#
+# Args:
+#     url (str): Location of the remote source cache
+#     push (bool): Whether we should attempt to push sources to this cache,
+#                  in addition to pulling from it.
+#     instance-name (str): Name if any, of instance of server
+#
+class SourceCacheSpec(CASRemoteSpec):
+    pass
+
+
+# Class that keeps config of remotes and deals with caching of sources.
+#
+# Args:
+#    context (Context): The Buildstream context
+#
+class SourceCache(BaseCache):
+
+    spec_class = SourceCacheSpec
+    spec_name = "source_cache_specs"
+    spec_error = SourceCacheError
+    config_node_name = "source-caches"
+
+    def __init__(self, context):
+        super().__init__(context)
+
+        self._required_sources = set()
+
+        self.casquota.add_ref_callbacks(self.required_sources())
+        self.casquota.add_remove_callbacks((lambda x: x.startswith('@sources/'), self.cas.remove))
+
+    # mark_required_sources()
+    #
+    # Mark sources that are required by the current run.
+    #
+    # Sources that are in this list will not be removed during the current
+    # pipeline.
+    #
+    # Args:
+    #     sources (iterable): An iterable over sources that are required
+    #
+    def mark_required_sources(self, sources):
+        sources = list(sources)  # in case it's a generator
+
+        self._required_sources.update(sources)
+
+        # update mtimes just in case
+        for source in sources:
+            ref = source._get_source_name()
+            try:
+                self.cas.update_mtime(ref)
+            except CASCacheError:
+                pass
+
+    # required_sources()
+    #
+    # Yields the keys of all sources marked as required
+    #
+    # Returns:
+    #     iterable (str): iterable over the source keys
+    #
+    def required_sources(self):
+        for source in self._required_sources:
+            yield source._key
+
+    # contains()
+    #
+    # Given a source, gets the ref name and checks whether the local CAS
+    # contains it.
+    #
+    # Args:
+    #    source (Source): Source to check
+    #
+    # Returns:
+    #    (bool): whether the CAS contains this source or not
+    #
+    def contains(self, source):
+        ref = source._get_source_name()
+        return self.cas.contains(ref)
+
+    # commit()
+    #
+    # Given a source along with previous sources, it stages and commits these
+    # to the local CAS. This is done due to some types of sources being
+    # dependent on previous sources, such as the patch source.
+    #
+    # Args:
+    #    source: last source
+    #    previous_sources: rest of the sources.
+    def commit(self, source, previous_sources):
+        ref = source._get_source_name()
+
+        # Use tmpdir for now
+        with utils._tempdir(dir=self.context.tmpdir, prefix='staging-temp') as tmpdir:
+            for previous_source in previous_sources:
+                previous_source._stage(tmpdir)
+            source._stage(tmpdir)
+
+            self.cas.commit([ref], tmpdir)
+
+    # export()
+    #
+    # Exports a source in the CAS to a virtual directory
+    #
+    # Args:
+    #    source (Source): source we want to export
+    #
+    # Returns:
+    #    CASBasedDirectory
+    def export(self, source):
+        ref = source._get_source_name()
+
+        try:
+            digest = self.cas.resolve_ref(ref)
+        except CASCacheError as e:
+            raise SourceCacheError("Error exporting source: {}".format(e))
+
+        return CasBasedDirectory(self.cas, digest=digest)