Allow push to multiple remotes, configurable on a per-remote basis

The initial multiple cache support patch implemented a rather fragile
logic where we would push to the first cache in the list that used the
ssh:// protocol, if any. If we implement push-over-https:// in future
then this will become totally unworkable.

This patch alters the logic so that each remote has a 'push' option,
and BuildStream will push to any remote that has 'push: true' set.

7 files changed, 113 insertions, 91 deletions

diff --git a/buildstream/_artifactcache/__init__.py b/buildstream/_artifactcache/__init__.py
index 7dc0ce3be..527164622 100644
--- a/buildstream/_artifactcache/__init__.py
+++ b/buildstream/_artifactcache/__init__.py
@@ -18,4 +18,5 @@
 #  Authors:
 #        Tristan Van Berkom <tristan.vanberkom@codethink.co.uk>
 
-from .artifactcache import ArtifactCache, artifact_cache_urls_from_config_node, configured_artifact_cache_urls
+from .artifactcache import ArtifactCache, ArtifactCacheSpec
+from .artifactcache import artifact_cache_specs_from_config_node, configured_remote_artifact_cache_specs
diff --git a/buildstream/_artifactcache/artifactcache.py b/buildstream/_artifactcache/artifactcache.py
index 1d30b2582..8721322c7 100644
--- a/buildstream/_artifactcache/artifactcache.py
+++ b/buildstream/_artifactcache/artifactcache.py
@@ -19,24 +19,35 @@
 #        Tristan Maat <tristan.maat@codethink.co.uk>
 
 import os
-from collections import Mapping
+from collections import Mapping, namedtuple
 
 from .._exceptions import ImplError, LoadError, LoadErrorReason
 from .. import utils
 from .. import _yaml
 
 
-def artifact_cache_url_from_spec(spec):
-    _yaml.node_validate(spec, ['url'])
-    url = _yaml.node_get(spec, str, 'url')
-    if len(url) == 0:
-        provenance = _yaml.node_get_provenance(spec)
-        raise LoadError(LoadErrorReason.INVALID_DATA,
-                        "{}: empty artifact cache URL".format(provenance))
-    return url
+# An ArtifactCacheSpec holds the user configuration for a single remote
+# artifact cache.
+#
+# Args:
+#     url (str): Location of the remote artifact cache
+#     push (bool): Whether we should attempt to push artifacts to this cache,
+#                  in addition to pulling from it.
+#
+class ArtifactCacheSpec(namedtuple('ArtifactCacheSpec', 'url push')):
+    @staticmethod
+    def new_from_config_node(spec_node):
+        _yaml.node_validate(spec_node, ['url', 'push'])
+        url = _yaml.node_get(spec_node, str, 'url')
+        push = _yaml.node_get(spec_node, bool, 'push', default_value=False)
+        if len(url) == 0:
+            provenance = _yaml.node_get_provenance(spec_node)
+            raise LoadError(LoadErrorReason.INVALID_DATA,
+                            "{}: empty artifact cache URL".format(provenance))
+        return ArtifactCacheSpec(url, push)
 
 
-# artifact_cache_urls_from_config_node()
+# artifact_cache_specs_from_config_node()
 #
 # Parses the configuration of remote artifact caches from a config block.
 #
@@ -44,29 +55,29 @@ def artifact_cache_url_from_spec(spec):
 #   config_node (dict): The config block, which may contain the 'artifacts' key
 #
 # Returns:
-#   A list of URLs pointing to remote artifact caches.
+#   A list of ArtifactCacheSpec instances.
 #
 # Raises:
 #   LoadError, if the config block contains invalid keys.
 #
-def artifact_cache_urls_from_config_node(config_node):
-    urls = []
+def artifact_cache_specs_from_config_node(config_node):
+    cache_specs = []
 
     artifacts = config_node.get('artifacts', [])
     if isinstance(artifacts, Mapping):
-        urls.append(artifact_cache_url_from_spec(artifacts))
+        cache_specs.append(ArtifactCacheSpec.new_from_config_node(artifacts))
     elif isinstance(artifacts, list):
-        for spec in artifacts:
-            urls.append(artifact_cache_url_from_spec(spec))
+        for spec_node in artifacts:
+            cache_specs.append(ArtifactCacheSpec.new_from_config_node(spec_node))
     else:
         provenance = _yaml.node_get_provenance(config_node, key='artifacts')
         raise _yaml.LoadError(_yaml.LoadErrorReason.INVALID_DATA,
                               "%s: 'artifacts' must be a single 'url:' mapping, or a list of mappings" %
                               (str(provenance)))
-    return urls
+    return cache_specs
 
 
-# configured_artifact_cache_urls():
+# configured_remote_artifact_cache_specs():
 #
 # Return the list of configured artifact remotes for a given project, in priority
 # order. This takes into account the user and project configuration.
@@ -76,14 +87,14 @@ def artifact_cache_urls_from_config_node(config_node):
 #     project (Project): The BuildStream project
 #
 # Returns:
-#   A list of URLs pointing to remote artifact caches.
+#   A list of ArtifactCacheSpec instances describing the remote artifact caches.
 #
-def configured_artifact_cache_urls(context, project):
+def configured_remote_artifact_cache_specs(context, project):
     project_overrides = context._get_overrides(project.name)
-    project_extra_urls = artifact_cache_urls_from_config_node(project_overrides)
+    project_extra_specs = artifact_cache_specs_from_config_node(project_overrides)
 
     return list(utils._deduplicate(
-        project_extra_urls + project.artifact_urls + context.artifact_urls))
+        project_extra_specs + project.artifact_cache_specs + context.artifact_cache_specs))
 
 
 # An ArtifactCache manages artifacts.
@@ -102,7 +113,7 @@ class ArtifactCache():
         self.extractdir = os.path.join(context.artifactdir, 'extract')
 
         self._local = False
-        self.urls = []
+        self.remote_specs = []
 
     # set_remotes():
     #
@@ -110,10 +121,10 @@ class ArtifactCache():
     # contact each remote cache.
     #
     # Args:
-    #     urls (list): List of artifact remote URLs, in priority order.
+    #     remote_specs (list): List of ArtifactCacheSpec instances, in priority order.
     #     on_failure (callable): Called if we fail to contact one of the caches.
-    def set_remotes(self, urls, on_failure=None):
-        self.urls = urls
+    def set_remotes(self, remote_specs, on_failure=None):
+        self.remote_specs = remote_specs
 
     # contains():
     #
@@ -169,7 +180,7 @@ class ArtifactCache():
     # Returns: True if any remote repositories are configured, False otherwise
     #
     def has_fetch_remotes(self):
-        return (len(self.urls) > 0)
+        return (len(self.remote_specs) > 0)
 
     # has_push_remotes():
     #
@@ -178,7 +189,7 @@ class ArtifactCache():
     # Returns: True if any remote repository is configured, False otherwise
     #
     def has_push_remotes(self):
-        return (len(self.urls) > 0)
+        return (any(spec for spec in self.remote_specs if spec.push) > 0)
 
     # remote_contains_key():
     #
diff --git a/buildstream/_artifactcache/ostreecache.py b/buildstream/_artifactcache/ostreecache.py
index 9987746b0..9fe72a63e 100644
--- a/buildstream/_artifactcache/ostreecache.py
+++ b/buildstream/_artifactcache/ostreecache.py
@@ -53,7 +53,7 @@ def buildref(element, key):
 # Args:
 #     context (Context): The BuildStream context
 #     project (Project): The BuildStream project
-#     enable_push (bool): Whether pushing is allowed
+#     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
@@ -69,12 +69,12 @@ class OSTreeCache(ArtifactCache):
         ostreedir = os.path.join(context.artifactdir, 'ostree')
         self.repo = _ostree.ensure(ostreedir, False)
 
-        self.push_url = None
+        self.push_urls = []
         self.pull_urls = []
         self._remote_refs = {}
 
-    def set_remotes(self, urls, on_failure=None):
-        self.urls = urls
+    def set_remotes(self, remote_specs, on_failure=None):
+        self.remote_specs = remote_specs
 
         self._initialize_remotes(on_failure)
 
@@ -82,7 +82,7 @@ class OSTreeCache(ArtifactCache):
         return (len(self.pull_urls) > 0)
 
     def has_push_remotes(self):
-        return (self.push_url is not None)
+        return (len(self.push_urls) > 0)
 
     # contains():
     #
@@ -282,44 +282,17 @@ class OSTreeCache(ArtifactCache):
     # Raises:
     #   (ArtifactError): if there was an error
     def push(self, element):
+        any_pushed = False
 
-        if self.push_url is None:
-            raise ArtifactError("The protocol in use does not support pushing.")
+        if len(self.push_urls) == 0:
+            raise ArtifactError("Push is not enabled for any of the configured remote artifact caches.")
 
         ref = buildref(element, element._get_cache_key_from_artifact())
         weak_ref = buildref(element, element._get_cache_key(strength=_KeyStrength.WEAK))
-        if self.push_url.startswith("file://"):
-            # local repository
-            push_repo = _ostree.ensure(self.push_url[7:], True)
-            _ostree.fetch(push_repo, remote=self.repo.get_path().get_uri(), ref=ref)
-            _ostree.fetch(push_repo, remote=self.repo.get_path().get_uri(), ref=weak_ref)
-
-            # Local remotes are not really a thing, just return True here
-            return True
-        else:
-            # Push over ssh
-            #
-            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)
+        for push_url in self.push_urls:
+            any_pushed |= self._push_to_remote(push_url, element, ref, weak_ref)
 
-                    # Now push the ref we want to push into our temporary archive-z2 repo
-                    _ostree.fetch(temp_repo, remote=self.repo.get_path().get_uri(), ref=ref)
-                    _ostree.fetch(temp_repo, remote=self.repo.get_path().get_uri(), ref=weak_ref)
-
-                with element.timed_activity("Sending artifact"), \
-                    element._output_file() as output_file:
-                    try:
-                        pushed = push_artifact(temp_repo.get_path().get_path(),
-                                               self.push_url,
-                                               [ref, weak_ref], output_file)
-                    except PushException as e:
-                        raise ArtifactError("Failed to push artifact {}: {}".format(ref, e)) from e
-
-                return pushed
+        return any_pushed
 
     # _initialize_remote():
     #
@@ -409,22 +382,26 @@ class OSTreeCache(ArtifactCache):
         # possible to pickle local functions such as child_action().
         #
         q = multiprocessing.Queue()
-        for url in self.urls:
-            p = multiprocessing.Process(target=child_action, args=(url, q))
+        for remote in self.remote_specs:
+            p = multiprocessing.Process(target=child_action, args=(remote.url, q))
             p.start()
             exception, push_url, pull_url, remote_refs = q.get()
             p.join()
 
             if exception and on_failure:
-                on_failure(url, exception)
+                on_failure(remote.url, exception)
             elif exception:
                 raise ArtifactError() from exception
             else:
-                # Use first pushable remote we find for pushing.
-                if push_url and not self.push_url:
-                    self.push_url = push_url
-
-                # Use all pullable remotes, in priority order
+                if remote.push:
+                    if push_url:
+                        self.push_urls.append(push_url)
+                    else:
+                        raise ArtifactError("Push enabled but not supported by repo at: {}".format(remote.url))
+
+                # The specs are deduplicated when reading the config, but since
+                # each push URL can supply an arbitrary pull URL we must dedup
+                # those again here.
                 if pull_url and pull_url not in self.pull_urls:
                     self.pull_urls.append(pull_url)
 
@@ -435,3 +412,37 @@ class OSTreeCache(ArtifactCache):
                 for ref in remote_refs:
                     if ref not in self._remote_refs:
                         self._remote_refs[ref] = remote
+
+    def _push_to_remote(self, push_url, element, ref, weak_ref):
+        if push_url.startswith("file://"):
+            # local repository
+            push_repo = _ostree.ensure(push_url[7:], True)
+            _ostree.fetch(push_repo, remote=self.repo.get_path().get_uri(), ref=ref)
+            _ostree.fetch(push_repo, remote=self.repo.get_path().get_uri(), ref=weak_ref)
+
+            # Local remotes are not really a thing, just return True here
+            return True
+        else:
+            # Push over ssh
+            #
+            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
+                    _ostree.fetch(temp_repo, remote=self.repo.get_path().get_uri(), ref=ref)
+                    _ostree.fetch(temp_repo, remote=self.repo.get_path().get_uri(), ref=weak_ref)
+
+                with element.timed_activity("Sending artifact"), \
+                    element._output_file() as output_file:
+                    try:
+                        pushed = push_artifact(temp_repo.get_path().get_path(),
+                                               push_url,
+                                               [ref, weak_ref], output_file)
+                    except PushException as e:
+                        raise ArtifactError("Failed to push artifact {}: {}".format(ref, e)) from e
+
+                return pushed
diff --git a/buildstream/_context.py b/buildstream/_context.py
index 78ec9f8b2..9b45b99da 100644
--- a/buildstream/_context.py
+++ b/buildstream/_context.py
@@ -29,7 +29,7 @@ from . import _yaml
 from ._exceptions import LoadError, LoadErrorReason, BstError
 from ._message import Message, MessageType
 from ._profile import Topics, profile_start, profile_end
-from ._artifactcache import artifact_cache_urls_from_config_node
+from ._artifactcache import artifact_cache_specs_from_config_node
 
 
 # Context()
@@ -62,8 +62,8 @@ class Context():
         # The local binary artifact cache directory
         self.artifactdir = None
 
-        # The URLs from which to push and pull prebuilt artifacts
-        self.artifact_urls = []
+        # The locations from which to push and pull prebuilt artifacts
+        self.artifact_cache_specs = []
 
         # The directory to store build logs
         self.logdir = None
@@ -162,7 +162,7 @@ class Context():
             setattr(self, dir, path)
 
         # Load artifact share configuration
-        self.artifact_urls = artifact_cache_urls_from_config_node(defaults)
+        self.artifact_cache_specs = artifact_cache_specs_from_config_node(defaults)
 
         # Load logging config
         logging = _yaml.node_get(defaults, Mapping, 'logging')
diff --git a/buildstream/_pipeline.py b/buildstream/_pipeline.py
index 01f3e41be..a4f28ce2e 100644
--- a/buildstream/_pipeline.py
+++ b/buildstream/_pipeline.py
@@ -40,7 +40,7 @@ from . import Scope
 from . import _site
 from . import utils
 from ._platform import Platform
-from ._artifactcache import configured_artifact_cache_urls
+from ._artifactcache import ArtifactCacheSpec, configured_remote_artifact_cache_specs
 
 from ._scheduler import SchedStatus, TrackQueue, FetchQueue, BuildQueue, PullQueue, PushQueue
 
@@ -155,13 +155,13 @@ class Pipeline():
 
         # Initialize remote artifact caches. We allow the commandline to override
         # the user config in some cases (for example `bst push --remote=...`).
-        artifact_urls = []
+        artifact_caches = []
         if add_remote_cache:
-            artifact_urls += [add_remote_cache]
+            artifact_caches += [ArtifactCacheSpec(add_remote_cache, push=True)]
         if use_configured_remote_caches:
-            artifact_urls += configured_artifact_cache_urls(self.context, self.project)
-        if len(artifact_urls) > 0:
-            self.initialize_remote_caches(artifact_urls)
+            artifact_caches += configured_remote_artifact_cache_specs(self.context, self.project)
+        if len(artifact_caches) > 0:
+            self.initialize_remote_caches(artifact_caches)
 
         self.resolve_cache_keys(inconsistent)
 
@@ -189,12 +189,12 @@ class Pipeline():
 
                 self.project._set_workspace(element, source, workspace)
 
-    def initialize_remote_caches(self, artifact_urls):
+    def initialize_remote_caches(self, artifact_cache_specs):
         def remote_failed(url, error):
             self.message(MessageType.WARN, "Failed to fetch remote refs from {}: {}\n".format(url, error))
 
         with self.timed_activity("Initializing remote caches", silent_nested=True):
-            self.artifacts.set_remotes(artifact_urls, on_failure=remote_failed)
+            self.artifacts.set_remotes(artifact_cache_specs, on_failure=remote_failed)
 
     def resolve_cache_keys(self, inconsistent):
         if inconsistent:
diff --git a/buildstream/_project.py b/buildstream/_project.py
index 35873fb72..3bbe61f19 100644
--- a/buildstream/_project.py
+++ b/buildstream/_project.py
@@ -28,7 +28,7 @@ from . import _yaml
 from ._profile import Topics, profile_start, profile_end
 from ._exceptions import LoadError, LoadErrorReason
 from ._options import OptionPool
-from ._artifactcache import artifact_cache_urls_from_config_node
+from ._artifactcache import artifact_cache_specs_from_config_node
 
 
 # The base BuildStream format version
@@ -173,7 +173,7 @@ class Project():
         #
 
         # Load artifacts pull/push configuration for this project
-        self.artifact_urls = artifact_cache_urls_from_config_node(config)
+        self.artifact_cache_specs = artifact_cache_specs_from_config_node(config)
 
         # Workspace configurations
         self._workspaces = self._load_workspace_config()
diff --git a/buildstream/element.py b/buildstream/element.py
index 18a23e915..1143cccf4 100644
--- a/buildstream/element.py
+++ b/buildstream/element.py
@@ -1204,8 +1204,7 @@ class Element(Plugin):
         if self._tainted():
             return True
 
-        # Do not push artifact that is already in the remote artifact repository
-        return self.__artifacts.remote_contains_key(self, self._get_cache_key_from_artifact())
+        return False
 
     # _push():
     #