Allow splitting artifact caches

This is now split into storage/index remotes, where the former is
expected to be a CASRemote and the latter a BuildStream-specific
remote with the extensions required to store BuildStream artifact
protos.

7 files changed, 272 insertions, 46 deletions

diff --git a/tests/artifactcache/config.py b/tests/artifactcache/config.py
index 2f235f38c..8b01a9ebe 100644
--- a/tests/artifactcache/config.py
+++ b/tests/artifactcache/config.py
@@ -6,7 +6,7 @@ import os
 
 import pytest
 
-from buildstream._remote import RemoteSpec
+from buildstream._remote import RemoteSpec, RemoteType
 from buildstream._artifactcache import ArtifactCache
 from buildstream._project import Project
 from buildstream.utils import _deduplicate
@@ -24,12 +24,28 @@ cache2 = RemoteSpec(url='https://example.com/cache2', push=False)
 cache3 = RemoteSpec(url='https://example.com/cache3', push=False)
 cache4 = RemoteSpec(url='https://example.com/cache4', push=False)
 cache5 = RemoteSpec(url='https://example.com/cache5', push=False)
-cache6 = RemoteSpec(url='https://example.com/cache6', push=True)
+cache6 = RemoteSpec(url='https://example.com/cache6',
+                    push=True,
+                    type=RemoteType.ALL)
+cache7 = RemoteSpec(url='https://index.example.com/cache1',
+                    push=True,
+                    type=RemoteType.INDEX)
+cache8 = RemoteSpec(url='https://storage.example.com/cache1',
+                    push=True,
+                    type=RemoteType.STORAGE)
 
 
 # Generate cache configuration fragments for the user config and project config files.
 #
-def configure_remote_caches(override_caches, project_caches=None, user_caches=None):
+def configure_remote_caches(override_caches,
+                            project_caches=None,
+                            user_caches=None):
+    type_strings = {
+        RemoteType.INDEX: 'index',
+        RemoteType.STORAGE: 'storage',
+        RemoteType.ALL: 'all'
+    }
+
     if project_caches is None:
         project_caches = []
 
@@ -41,10 +57,15 @@ def configure_remote_caches(override_caches, project_caches=None, user_caches=No
         user_config['artifacts'] = {
             'url': user_caches[0].url,
             'push': user_caches[0].push,
+            'type': type_strings[user_caches[0].type]
         }
     elif len(user_caches) > 1:
         user_config['artifacts'] = [
-            {'url': cache.url, 'push': cache.push} for cache in user_caches
+            {
+                'url': cache.url,
+                'push': cache.push,
+                'type': type_strings[cache.type]
+            } for cache in user_caches
         ]
 
     if len(override_caches) == 1:
@@ -53,6 +74,7 @@ def configure_remote_caches(override_caches, project_caches=None, user_caches=No
                 'artifacts': {
                     'url': override_caches[0].url,
                     'push': override_caches[0].push,
+                    'type': type_strings[override_caches[0].type]
                 }
             }
         }
@@ -60,7 +82,11 @@ def configure_remote_caches(override_caches, project_caches=None, user_caches=No
         user_config['projects'] = {
             'test': {
                 'artifacts': [
-                    {'url': cache.url, 'push': cache.push} for cache in override_caches
+                    {
+                        'url': cache.url,
+                        'push': cache.push,
+                        'type': type_strings[cache.type]
+                    } for cache in override_caches
                 ]
             }
         }
@@ -72,12 +98,17 @@ def configure_remote_caches(override_caches, project_caches=None, user_caches=No
                 'artifacts': {
                     'url': project_caches[0].url,
                     'push': project_caches[0].push,
+                    'type': type_strings[project_caches[0].type],
                 }
             })
         elif len(project_caches) > 1:
             project_config.update({
                 'artifacts': [
-                    {'url': cache.url, 'push': cache.push} for cache in project_caches
+                    {
+                        'url': cache.url,
+                        'push': cache.push,
+                        'type': type_strings[cache.type]
+                    } for cache in project_caches
                 ]
             })
 
@@ -96,6 +127,7 @@ def configure_remote_caches(override_caches, project_caches=None, user_caches=No
         pytest.param([cache1], [cache2], [cache3], id='project-override-in-user-config'),
         pytest.param([cache1, cache2], [cache3, cache4], [cache5, cache6], id='list-order'),
         pytest.param([cache1, cache2, cache1], [cache2], [cache2, cache1], id='duplicates'),
+        pytest.param([cache7, cache8], [], [cache1], id='split-caches')
     ])
 def test_artifact_cache_precedence(tmpdir, override_caches, project_caches, user_caches):
     # Produce a fake user and project config with the cache configuration.
@@ -149,3 +181,36 @@ def test_missing_certs(cli, datafiles, config_key, config_value):
     # This does not happen for a simple `bst show`.
     result = cli.run(project=project, args=['artifact', 'pull', 'element.bst'])
     result.assert_main_error(ErrorDomain.LOAD, LoadErrorReason.INVALID_DATA)
+
+
+# Assert that BuildStream complains when someone attempts to define
+# only one type of storage.
+@pytest.mark.datafiles(DATA_DIR)
+@pytest.mark.parametrize(
+    'override_caches, project_caches, user_caches',
+    [
+        # The leftmost cache is the highest priority one in all cases here.
+        pytest.param([], [], [cache7], id='index-user'),
+        pytest.param([], [], [cache8], id='storage-user'),
+        pytest.param([], [cache7], [], id='index-project'),
+        pytest.param([], [cache8], [], id='storage-project'),
+        pytest.param([cache7], [], [], id='index-override'),
+        pytest.param([cache8], [], [], id='storage-override'),
+    ])
+def test_only_one(cli, datafiles, override_caches, project_caches, user_caches):
+    project = os.path.join(datafiles.dirname, datafiles.basename, 'only-one')
+
+    # Produce a fake user and project config with the cache configuration.
+    user_config, project_config = configure_remote_caches(override_caches, project_caches, user_caches)
+    project_config['name'] = 'test'
+
+    cli.configure(user_config)
+
+    project_config_file = os.path.join(project, 'project.conf')
+    _yaml.roundtrip_dump(project_config, file=project_config_file)
+
+    # Use `pull` here to ensure we try to initialize the remotes, triggering the error
+    #
+    # This does not happen for a simple `bst show`.
+    result = cli.run(project=project, args=['artifact', 'pull', 'element.bst'])
+    result.assert_main_error(ErrorDomain.STREAM, None)
diff --git a/tests/artifactcache/only-one/element.bst b/tests/artifactcache/only-one/element.bst
new file mode 100644
index 000000000..3c29b4ea1
--- /dev/null
+++ b/tests/artifactcache/only-one/element.bst
@@ -0,0 +1 @@
+kind: autotools
diff --git a/tests/artifactcache/push.py b/tests/artifactcache/push.py
index 20d9ccfec..364ac39f0 100644
--- a/tests/artifactcache/push.py
+++ b/tests/artifactcache/push.py
@@ -10,7 +10,7 @@ from buildstream._project import Project
 from buildstream._protos.build.bazel.remote.execution.v2 import remote_execution_pb2
 from buildstream.testing import cli  # pylint: disable=unused-import
 
-from tests.testutils import create_artifact_share, dummy_context
+from tests.testutils import create_artifact_share, create_split_share, dummy_context
 
 
 # Project directory
@@ -20,6 +20,39 @@ DATA_DIR = os.path.join(
 )
 
 
+# Push the given element and return its artifact key for assertions.
+def _push(cli, cache_dir, project_dir, config_file, target):
+    with dummy_context(config=config_file) as context:
+        # Load the project manually
+        project = Project(project_dir, context)
+        project.ensure_fully_loaded()
+
+        # Assert that the element's artifact is cached
+        element = project.load_elements(['target.bst'])[0]
+        element_key = cli.get_element_key(project_dir, 'target.bst')
+        assert cli.artifact.is_cached(cache_dir, element, element_key)
+
+        # Create a local artifact cache handle
+        artifactcache = context.artifactcache
+
+        # Ensure the element's artifact memeber is initialised
+        # This is duplicated from Pipeline.resolve_elements()
+        # as this test does not use the cli frontend.
+        for e in element.dependencies(Scope.ALL):
+            # Determine initial element state.
+            e._update_state()
+
+        # Manually setup the CAS remotes
+        artifactcache.setup_remotes(use_config=True)
+        artifactcache.initialize_remotes()
+
+        assert artifactcache.has_push_remotes(plugin=element), \
+            "No remote configured for element target.bst"
+        assert element._push(), "Push operation failed"
+
+    return element_key
+
+
 @pytest.mark.in_subprocess
 @pytest.mark.datafiles(DATA_DIR)
 def test_push(cli, tmpdir, datafiles):
@@ -50,36 +83,52 @@ def test_push(cli, tmpdir, datafiles):
 
         # Write down the user configuration file
         _yaml.roundtrip_dump(user_config, file=user_config_file)
+        element_key = _push(cli, rootcache_dir, project_dir, user_config_file, 'target.bst')
+        assert share.has_artifact(cli.get_artifact_name(project_dir, 'test', 'target.bst', cache_key=element_key))
 
-        with dummy_context(config=user_config_file) as context:
-            # Load the project manually
-            project = Project(project_dir, context)
-            project.ensure_fully_loaded()
 
-            # Assert that the element's artifact is cached
-            element = project.load_elements(['target.bst'])[0]
-            element_key = cli.get_element_key(project_dir, 'target.bst')
-            assert cli.artifact.is_cached(rootcache_dir, element, element_key)
+@pytest.mark.in_subprocess
+@pytest.mark.datafiles(DATA_DIR)
+def test_push_split(cli, tmpdir, datafiles):
+    project_dir = str(datafiles)
 
-            # Create a local artifact cache handle
-            artifactcache = context.artifactcache
+    # First build the project without the artifact cache configured
+    result = cli.run(project=project_dir, args=['build', 'target.bst'])
+    result.assert_success()
 
-            # Ensure the element's artifact memeber is initialised
-            # This is duplicated from Pipeline.resolve_elements()
-            # as this test does not use the cli frontend.
-            for e in element.dependencies(Scope.ALL):
-                # Determine initial element state.
-                e._update_state()
+    # Assert that we are now cached locally
+    assert cli.get_element_state(project_dir, 'target.bst') == 'cached'
 
-            # Manually setup the CAS remotes
-            artifactcache.setup_remotes(use_config=True)
-            artifactcache.initialize_remotes()
+    indexshare = os.path.join(str(tmpdir), 'indexshare')
+    storageshare = os.path.join(str(tmpdir), 'storageshare')
 
-            assert artifactcache.has_push_remotes(plugin=element), \
-                "No remote configured for element target.bst"
-            assert element._push(), "Push operation failed"
+    # Set up an artifact cache.
+    with create_split_share(indexshare, storageshare) as (index, storage):
+        rootcache_dir = os.path.join(str(tmpdir), 'cache')
+        user_config = {
+            'scheduler': {
+                'pushers': 1
+            },
+            'artifacts': [{
+                'url': index.repo,
+                'push': True,
+                'type': 'index'
+            }, {
+                'url': storage.repo,
+                'push': True,
+                'type': 'storage'
+            }],
+            'cachedir': rootcache_dir
+        }
+        config_path = str(tmpdir.join('buildstream.conf'))
+        _yaml.roundtrip_dump(user_config, file=config_path)
 
-        assert share.has_artifact(cli.get_artifact_name(project_dir, 'test', 'target.bst', cache_key=element_key))
+        element_key = _push(cli, rootcache_dir, project_dir, config_path, 'target.bst')
+        proto = index.get_artifact_proto(cli.get_artifact_name(project_dir,
+                                                               'test',
+                                                               'target.bst',
+                                                               cache_key=element_key))
+        assert storage.get_cas_files(proto) is not None
 
 
 @pytest.mark.in_subprocess
@@ -88,7 +137,8 @@ def test_push_message(tmpdir, datafiles):
     project_dir = str(datafiles)
 
     # Set up an artifact cache.
-    with create_artifact_share(os.path.join(str(tmpdir), 'artifactshare')) as share:
+    artifactshare = os.path.join(str(tmpdir), 'artifactshare')
+    with create_artifact_share(artifactshare) as share:
         # Configure artifact share
         rootcache_dir = os.path.join(str(tmpdir), 'cache')
         user_config_file = str(tmpdir.join('buildstream.conf'))
diff --git a/tests/frontend/push.py b/tests/frontend/push.py
index 4f0fa3c19..57e670fe6 100644
--- a/tests/frontend/push.py
+++ b/tests/frontend/push.py
@@ -464,7 +464,16 @@ def test_artifact_too_large(cli, datafiles, tmpdir):
         # Create and try to push a 6MB element.
         create_element_size('large_element.bst', project, element_path, [], int(6e6))
         result = cli.run(project=project, args=['build', 'large_element.bst'])
-        result.assert_success()
+        # This should fail; the server will refuse to store the CAS
+        # blobs for the artifact, and then fail to find the files for
+        # the uploaded artifact proto.
+        #
+        # FIXME: This should be extremely uncommon in practice, since
+        # the artifact needs to be at least half the cache size for
+        # this to happen. Nonetheless, a nicer error message would be
+        # nice (perhaps we should just disallow uploading artifacts
+        # that large).
+        result.assert_main_error(ErrorDomain.STREAM, None)
 
         # Ensure that the small artifact is still in the share
         states = cli.get_element_states(project, ['small_element.bst', 'large_element.bst'])
diff --git a/tests/sourcecache/push.py b/tests/sourcecache/push.py
index ad9653f9d..1be2d40cd 100644
--- a/tests/sourcecache/push.py
+++ b/tests/sourcecache/push.py
@@ -21,6 +21,8 @@
 # pylint: disable=redefined-outer-name
 import os
 import shutil
+from contextlib import contextmanager, ExitStack
+
 import pytest
 
 from buildstream._exceptions import ErrorDomain
@@ -38,6 +40,82 @@ def message_handler(message, is_silenced):
     pass
 
 
+# Args:
+#    tmpdir: A temporary directory to use as root.
+#    directories: Directory names to use as cache directories.
+#
+@contextmanager
+def _configure_caches(tmpdir, *directories):
+    with ExitStack() as stack:
+        def create_share(directory):
+            return create_artifact_share(os.path.join(str(tmpdir), directory))
+
+        yield (stack.enter_context(create_share(remote)) for remote in directories)
+
+
+@pytest.mark.datafiles(DATA_DIR)
+def test_source_push_split(cli, tmpdir, datafiles):
+    cache_dir = os.path.join(str(tmpdir), 'cache')
+    project_dir = str(datafiles)
+
+    with _configure_caches(tmpdir, 'indexshare', 'storageshare') as (index, storage):
+        user_config_file = str(tmpdir.join('buildstream.conf'))
+        user_config = {
+            'scheduler': {
+                'pushers': 1
+            },
+            'source-caches': [{
+                'url': index.repo,
+                'push': True,
+                'type': 'index'
+            }, {
+                'url': storage.repo,
+                'push': True,
+                'type': 'storage'
+            }],
+            'cachedir': cache_dir
+        }
+        _yaml.roundtrip_dump(user_config, file=user_config_file)
+        cli.configure(user_config)
+
+        repo = create_repo('git', str(tmpdir))
+        ref = repo.create(os.path.join(project_dir, 'files'))
+        element_path = os.path.join(project_dir, 'elements')
+        element_name = 'push.bst'
+        element = {
+            'kind': 'import',
+            'sources': [repo.source_config(ref=ref)]
+        }
+        _yaml.roundtrip_dump(element, os.path.join(element_path, element_name))
+
+        # get the source object
+        with dummy_context(config=user_config_file) as context:
+            project = Project(project_dir, context)
+            project.ensure_fully_loaded()
+
+            element = project.load_elements(['push.bst'])[0]
+            assert not element._source_cached()
+            source = list(element.sources())[0]
+
+            # check we don't have it in the current cache
+            cas = context.get_cascache()
+            assert not cas.contains(source._get_source_name())
+
+            # build the element, this should fetch and then push the source to the
+            # remote
+            res = cli.run(project=project_dir, args=['build', 'push.bst'])
+            res.assert_success()
+            assert "Pushed source" in res.stderr
+
+            # check that we've got the remote locally now
+            sourcecache = context.sourcecache
+            assert sourcecache.contains(source)
+
+            # check that the remote CAS now has it
+            digest = sourcecache.export(source)._get_digest()
+            assert storage.has_object(digest)
+
+
 @pytest.mark.datafiles(DATA_DIR)
 def test_source_push(cli, tmpdir, datafiles):
     cache_dir = os.path.join(str(tmpdir), 'cache')
diff --git a/tests/testutils/__init__.py b/tests/testutils/__init__.py
index 25fa6d763..9642ddf47 100644
--- a/tests/testutils/__init__.py
+++ b/tests/testutils/__init__.py
@@ -23,7 +23,7 @@
 #           William Salmon <will.salmon@codethink.co.uk>
 #
 
-from .artifactshare import create_artifact_share, assert_shared, assert_not_shared
+from .artifactshare import create_artifact_share, create_split_share, assert_shared, assert_not_shared
 from .context import dummy_context
 from .element_generators import create_element_size, update_element_size
 from .junction import generate_junction
diff --git a/tests/testutils/artifactshare.py b/tests/testutils/artifactshare.py
index 7d5faeb66..e20f6e694 100644
--- a/tests/testutils/artifactshare.py
+++ b/tests/testutils/artifactshare.py
@@ -25,7 +25,7 @@ from buildstream._protos.buildstream.v2 import artifact_pb2
 #
 class ArtifactShare():
 
-    def __init__(self, directory, *, quota=None, casd=False):
+    def __init__(self, directory, *, quota=None, casd=False, index_only=False):
 
         # The working directory for the artifact share (in case it
         # needs to do something outside of its backend's storage folder).
@@ -45,6 +45,7 @@ class ArtifactShare():
         self.cas = CASCache(self.repodir, casd=casd)
 
         self.quota = quota
+        self.index_only = index_only
 
         q = Queue()
 
@@ -72,7 +73,8 @@ class ArtifactShare():
         try:
             with create_server(self.repodir,
                                quota=self.quota,
-                               enable_push=True) as server:
+                               enable_push=True,
+                               index_only=self.index_only) as server:
                 port = server.add_insecure_port('localhost:0')
 
                 server.start()
@@ -104,17 +106,7 @@ class ArtifactShare():
 
         return os.path.exists(object_path)
 
-    # has_artifact():
-    #
-    # Checks whether the artifact is present in the share
-    #
-    # Args:
-    #    artifact_name (str): The composed complete artifact name
-    #
-    # Returns:
-    #    (str): artifact digest if the artifact exists in the share, otherwise None.
-    def has_artifact(self, artifact_name):
-
+    def get_artifact_proto(self, artifact_name):
         artifact_proto = artifact_pb2.Artifact()
         artifact_path = os.path.join(self.artifactdir, artifact_name)
 
@@ -124,6 +116,10 @@ class ArtifactShare():
         except FileNotFoundError:
             return None
 
+        return artifact_proto
+
+    def get_cas_files(self, artifact_proto):
+
         reachable = set()
 
         def reachable_dir(digest):
@@ -153,6 +149,21 @@ class ArtifactShare():
         except FileNotFoundError:
             return None
 
+    # has_artifact():
+    #
+    # Checks whether the artifact is present in the share
+    #
+    # Args:
+    #    artifact_name (str): The composed complete artifact name
+    #
+    # Returns:
+    #    (str): artifact digest if the artifact exists in the share, otherwise None.
+    def has_artifact(self, artifact_name):
+        artifact_proto = self.get_artifact_proto(artifact_name)
+        if not artifact_proto:
+            return None
+        return self.get_cas_files(artifact_proto)
+
     # close():
     #
     # Remove the artifact share.
@@ -179,6 +190,18 @@ def create_artifact_share(directory, *, quota=None, casd=False):
         share.close()
 
 
+@contextmanager
+def create_split_share(directory1, directory2, *, quota=None, casd=False):
+    index = ArtifactShare(directory1, quota=quota, casd=casd, index_only=True)
+    storage = ArtifactShare(directory2, quota=quota, casd=casd)
+
+    try:
+        yield index, storage
+    finally:
+        index.close()
+        storage.close()
+
+
 statvfs_result = namedtuple('statvfs_result', 'f_blocks f_bfree f_bsize f_bavail')