Merge branch 'juerg/casd' into 'master'

Do not directly communicate with CAS server

See merge request BuildStream/buildstream!1601

6 files changed, 117 insertions, 135 deletions

diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml
index 34ef47125..34067613d 100644
--- a/.gitlab-ci.yml
+++ b/.gitlab-ci.yml
@@ -1,7 +1,7 @@
 include:
   - template: Code-Quality.gitlab-ci.yml
 
-image: registry.gitlab.com/buildstream/buildstream-docker-images/testsuite-debian:9-master-80198708
+image: registry.gitlab.com/buildstream/buildstream-docker-images/testsuite-debian:9-master-83237142
 
 cache:
   key: "$CI_JOB_NAME-"
@@ -54,24 +54,24 @@ variables:
     - .coverage-reports
 
 tests-debian-9:
-  image: registry.gitlab.com/buildstream/buildstream-docker-images/testsuite-debian:9-master-80198708
+  image: registry.gitlab.com/buildstream/buildstream-docker-images/testsuite-debian:9-master-83237142
   <<: *tests
 
 tests-fedora-29:
-  image: registry.gitlab.com/buildstream/buildstream-docker-images/testsuite-fedora:29-master-80198708
+  image: registry.gitlab.com/buildstream/buildstream-docker-images/testsuite-fedora:29-master-83237142
   <<: *tests
 
 tests-fedora-30:
-  image: registry.gitlab.com/buildstream/buildstream-docker-images/testsuite-fedora:30-master-80198708
+  image: registry.gitlab.com/buildstream/buildstream-docker-images/testsuite-fedora:30-master-83237142
   <<: *tests
 
 tests-ubuntu-18.04:
-  image: registry.gitlab.com/buildstream/buildstream-docker-images/testsuite-ubuntu:18.04-master-80198708
+  image: registry.gitlab.com/buildstream/buildstream-docker-images/testsuite-ubuntu:18.04-master-83237142
   <<: *tests
 
 tests-centos-7.6:
   <<: *tests
-  image: registry.gitlab.com/buildstream/buildstream-docker-images/testsuite-centos:7.6.1810-master-80198708
+  image: registry.gitlab.com/buildstream/buildstream-docker-images/testsuite-centos:7.6.1810-master-83237142
 
 overnight-fedora-30-aarch64:
   image: registry.gitlab.com/buildstream/buildstream-docker-images/testsuite-fedora:aarch64-30-master-59168197
@@ -87,7 +87,7 @@ overnight-fedora-30-aarch64:
 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: registry.gitlab.com/buildstream/buildstream-docker-images/testsuite-fedora:29-master-80198708
+  image: registry.gitlab.com/buildstream/buildstream-docker-images/testsuite-fedora:29-master-83237142
   <<: *tests
   variables:
     BST_FORCE_SANDBOX: "chroot"
@@ -104,7 +104,7 @@ tests-unix:
     - ${TEST_COMMAND}
 
 tests-buildbox:
-  image: registry.gitlab.com/buildstream/buildstream-docker-images/testsuite-fedora:29-master-80198708
+  image: registry.gitlab.com/buildstream/buildstream-docker-images/testsuite-fedora:29-master-83237142
   <<: *tests
   variables:
     BST_FORCE_SANDBOX: "buildbox"
@@ -134,7 +134,7 @@ tests-buildbox:
 
 tests-fedora-missing-deps:
   # Ensure that tests behave nicely while missing bwrap and ostree
-  image: registry.gitlab.com/buildstream/buildstream-docker-images/testsuite-fedora:29-master-80198708
+  image: registry.gitlab.com/buildstream/buildstream-docker-images/testsuite-fedora:29-master-83237142
   <<: *tests
 
   script:
@@ -153,7 +153,7 @@ tests-fedora-update-deps:
   # Check if the tests pass after updating requirements to their latest
   # allowed version.
   allow_failure: true
-  image: registry.gitlab.com/buildstream/buildstream-docker-images/testsuite-fedora:29-master-80198708
+  image: registry.gitlab.com/buildstream/buildstream-docker-images/testsuite-fedora:29-master-83237142
   <<: *tests
 
   script:
@@ -167,7 +167,7 @@ tests-fedora-update-deps:
 
 tests-remote-execution:
   allow_failure: true
-  image: registry.gitlab.com/buildstream/buildstream-docker-images/testsuite-fedora:29-master-80198708
+  image: registry.gitlab.com/buildstream/buildstream-docker-images/testsuite-fedora:29-master-83237142
   <<: *tests
   before_script:
     - dnf install -y docker docker-compose
@@ -190,7 +190,7 @@ tests-remote-execution:
     PYTEST_ARGS: "--color=yes --remote-execution"
 
 tests-spawn-multiprocessing-start-method:
-  image: registry.gitlab.com/buildstream/buildstream-docker-images/testsuite-fedora:29-master-80198708
+  image: registry.gitlab.com/buildstream/buildstream-docker-images/testsuite-fedora:29-master-83237142
   <<: *tests
   variables:
     BST_FORCE_START_METHOD: "spawn"
@@ -230,8 +230,8 @@ tests-wsl-master:
   - PATH=/root/.local/bin:$PATH tox --version
   script:
   # Install static buildbox-casd binary
-  - wget https://buildbox-casd-binaries.nyc3.cdn.digitaloceanspaces.com/buildbox-casd-x86_64-linux-20190904-4169f21b.tar.xz
-  - tar -C /root/.local/bin -xf buildbox-casd-x86_64-linux-20190904-4169f21b.tar.xz
+  - wget https://buildbox-casd-binaries.nyc3.cdn.digitaloceanspaces.com/buildbox-casd-x86_64-linux-20190919-bc3eeb4b.tar.xz
+  - tar -C /root/.local/bin -xf buildbox-casd-x86_64-linux-20190919-bc3eeb4b.tar.xz
 
   - PATH=/root/.local/bin:$PATH ${TEST_COMMAND}
   only:
@@ -250,8 +250,8 @@ tests-wsl-non-master:
   - PATH=/root/.local/bin:$PATH tox --version
   script:
   # Install static buildbox-casd binary
-  - wget https://buildbox-casd-binaries.nyc3.cdn.digitaloceanspaces.com/buildbox-casd-x86_64-linux-20190904-4169f21b.tar.xz
-  - tar -C /root/.local/bin -xf buildbox-casd-x86_64-linux-20190904-4169f21b.tar.xz
+  - wget https://buildbox-casd-binaries.nyc3.cdn.digitaloceanspaces.com/buildbox-casd-x86_64-linux-20190919-bc3eeb4b.tar.xz
+  - tar -C /root/.local/bin -xf buildbox-casd-x86_64-linux-20190919-bc3eeb4b.tar.xz
 
   - PATH=/root/.local/bin:$PATH ${TEST_COMMAND}
   when: manual
@@ -275,7 +275,7 @@ docs:
 
 .overnight-tests: &overnight-tests-template
   stage: test
-  image: registry.gitlab.com/buildstream/buildstream-docker-images/testsuite-fedora:30-master-80198708
+  image: registry.gitlab.com/buildstream/buildstream-docker-images/testsuite-fedora:30-master-83237142
   variables:
     BST_EXT_URL: git+https://gitlab.com/BuildStream/bst-plugins-experimental.git
     BST_EXT_REF: 0.12.0-40-g7aa1423377629281decc455d1090964417c38f2e
diff --git a/src/buildstream/_artifactcache.py b/src/buildstream/_artifactcache.py
index 0e2eb1091..f8d856be7 100644
--- a/src/buildstream/_artifactcache.py
+++ b/src/buildstream/_artifactcache.py
@@ -37,6 +37,11 @@ from . import utils
 # artifact remotes.
 #
 class ArtifactRemote(BaseRemote):
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.artifact_service = None
+
     # _configure_protocols():
     #
     # Configure the protocols used by this remote as part of the
@@ -44,7 +49,19 @@ class ArtifactRemote(BaseRemote):
     # Remote.init(), and is expected to fail when called by itself.
     #
     def _configure_protocols(self):
-        # Add artifact stub
+        # Set up artifact stub
+        self.artifact_service = artifact_pb2_grpc.ArtifactServiceStub(self.channel)
+
+    # _check():
+    #
+    # Check if this remote provides everything required for the
+    # particular kind of remote. This is expected to be called as part
+    # of check(), and must be called in a non-main process.
+    #
+    # Returns:
+    #    (str|None): An error message, or None if no error message.
+    #
+    def _check(self):
         capabilities_service = buildstream_pb2_grpc.CapabilitiesStub(self.channel)
 
         # Check whether the server supports newer proto based artifact.
@@ -56,16 +73,18 @@ class ArtifactRemote(BaseRemote):
         except grpc.RpcError as e:
             # Check if this remote has the artifact service
             if e.code() == grpc.StatusCode.UNIMPLEMENTED:
-                raise ArtifactError(
-                    "Configured remote does not have the BuildStream "
-                    "capabilities service. Please check remote configuration.")
+                return ("Configured remote does not have the BuildStream "
+                        "capabilities service. Please check remote configuration.")
             # Else raise exception with details
-            raise ArtifactError(
-                "Remote initialisation failed: {}".format(e.details()))
+            return "Remote initialisation failed: {}".format(e.details())
 
         if not response.artifact_capabilities:
-            raise ArtifactError(
-                "Configured remote does not support artifact service")
+            return "Configured remote does not support artifact service"
+
+        if self.spec.push and not response.artifact_capabilities.allow_updates:
+            return 'Artifact server does not allow push'
+
+        return None
 
     # get_artifact():
     #
@@ -86,8 +105,7 @@ class ArtifactRemote(BaseRemote):
         artifact_request = artifact_pb2.GetArtifactRequest()
         artifact_request.cache_key = cache_key
 
-        artifact_service = artifact_pb2_grpc.ArtifactServiceStub(self.channel)
-        return artifact_service.GetArtifact(artifact_request)
+        return self.artifact_service.GetArtifact(artifact_request)
 
     # update_artifact():
     #
@@ -106,8 +124,7 @@ class ArtifactRemote(BaseRemote):
         update_request.cache_key = cache_key
         update_request.artifact.CopyFrom(artifact)
 
-        artifact_service = artifact_pb2_grpc.ArtifactServiceStub(self.channel)
-        artifact_service.UpdateArtifact(update_request)
+        self.artifact_service.UpdateArtifact(update_request)
 
 
 # An ArtifactCache manages artifacts.
@@ -707,8 +724,7 @@ class ArtifactCache(BaseCache):
         request = artifact_pb2.GetArtifactRequest()
         request.cache_key = ref
         try:
-            artifact_service = artifact_pb2_grpc.ArtifactServiceStub(remote.channel)
-            artifact_service.GetArtifact(request)
+            remote.artifact_service.GetArtifact(request)
         except grpc.RpcError as e:
             if e.code() != grpc.StatusCode.NOT_FOUND:
                 raise ArtifactError("Error when querying: {}".format(e.details()))
diff --git a/src/buildstream/_cas/cascache.py b/src/buildstream/_cas/cascache.py
index 1f687669b..2603b13c5 100644
--- a/src/buildstream/_cas/cascache.py
+++ b/src/buildstream/_cas/cascache.py
@@ -1,5 +1,6 @@
 #
 #  Copyright (C) 2018 Codethink Limited
+#  Copyright (C) 2018-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
@@ -88,6 +89,7 @@ class CASCache():
             self._casd_process = None
 
         self._casd_channel = None
+        self._casd_cas = None
         self._local_cas = None
         self._cache_usage_monitor = None
 
@@ -101,11 +103,12 @@ class CASCache():
 
         return state
 
-    def _get_local_cas(self):
+    def _init_casd(self):
         assert self._casd_process, "CASCache was instantiated without buildbox-casd"
 
-        if not self._local_cas:
+        if not self._casd_channel:
             self._casd_channel = grpc.insecure_channel('unix:' + self._casd_socket_path)
+            self._casd_cas = remote_execution_pb2_grpc.ContentAddressableStorageStub(self._casd_channel)
             self._local_cas = local_cas_pb2_grpc.LocalContentAddressableStorageStub(self._casd_channel)
 
             # Call GetCapabilities() to establish connection to casd
@@ -124,6 +127,22 @@ class CASCache():
 
                     raise
 
+    # _get_cas():
+    #
+    # Return ContentAddressableStorage stub for buildbox-casd channel.
+    #
+    def _get_cas(self):
+        if not self._casd_cas:
+            self._init_casd()
+        return self._casd_cas
+
+    # _get_local_cas():
+    #
+    # Return LocalCAS stub for buildbox-casd channel.
+    #
+    def _get_local_cas(self):
+        if not self._local_cas:
+            self._init_casd()
         return self._local_cas
 
     # preflight():
@@ -492,16 +511,25 @@ class CASCache():
     # Returns: List of missing Digest objects
     #
     def remote_missing_blobs(self, remote, blobs):
+        cas = self._get_cas()
+        instance_name = remote.local_cas_instance_name
+
         missing_blobs = dict()
         # Limit size of FindMissingBlobs request
         for required_blobs_group in _grouper(iter(blobs), 512):
-            request = remote_execution_pb2.FindMissingBlobsRequest(instance_name=remote.spec.instance_name)
+            request = remote_execution_pb2.FindMissingBlobsRequest(instance_name=instance_name)
 
             for required_digest in required_blobs_group:
                 d = request.blob_digests.add()
                 d.CopyFrom(required_digest)
 
-            response = remote.cas.FindMissingBlobs(request)
+            try:
+                response = cas.FindMissingBlobs(request)
+            except grpc.RpcError as e:
+                if e.code() == grpc.StatusCode.INVALID_ARGUMENT and e.details().startswith("Invalid instance name"):
+                    raise CASCacheError("Unsupported buildbox-casd version: FindMissingBlobs failed") from e
+                raise
+
             for missing_digest in response.missing_blob_digests:
                 d = remote_execution_pb2.Digest()
                 d.CopyFrom(missing_digest)
diff --git a/src/buildstream/_cas/casremote.py b/src/buildstream/_cas/casremote.py
index 43e215c63..a054b288a 100644
--- a/src/buildstream/_cas/casremote.py
+++ b/src/buildstream/_cas/casremote.py
@@ -1,9 +1,22 @@
-import grpc
+#
+#  Copyright (C) 2018-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/>.
+#
 
 from .._protos.google.rpc import code_pb2
-from .._protos.build.bazel.remote.execution.v2 import remote_execution_pb2, remote_execution_pb2_grpc
 from .._protos.build.buildgrid import local_cas_pb2
-from .._protos.buildstream.v2 import buildstream_pb2, buildstream_pb2_grpc
 
 from .._remote import BaseRemote
 from .._exceptions import CASRemoteError
@@ -33,12 +46,6 @@ class CASRemote(BaseRemote):
         super().__init__(spec, **kwargs)
 
         self.cascache = cascache
-        self.cas = None
-        self.ref_storage = None
-        self.batch_update_supported = None
-        self.batch_read_supported = None
-        self.capabilities = None
-        self.max_batch_total_size_bytes = None
         self.local_cas_instance_name = None
 
     # check_remote
@@ -48,39 +55,6 @@ class CASRemote(BaseRemote):
     # be called outside of init().
     #
     def _configure_protocols(self):
-        self.cas = remote_execution_pb2_grpc.ContentAddressableStorageStub(self.channel)
-        self.capabilities = remote_execution_pb2_grpc.CapabilitiesStub(self.channel)
-        self.ref_storage = buildstream_pb2_grpc.ReferenceStorageStub(self.channel)
-
-        # Figure out what batch sizes the server will accept, falling
-        # back to our _MAX_PAYLOAD_BYTES
-        self.max_batch_total_size_bytes = _MAX_PAYLOAD_BYTES
-        try:
-            request = remote_execution_pb2.GetCapabilitiesRequest()
-            if self.instance_name:
-                request.instance_name = self.instance_name
-            response = self.capabilities.GetCapabilities(request)
-            server_max_batch_total_size_bytes = response.cache_capabilities.max_batch_total_size_bytes
-            if 0 < server_max_batch_total_size_bytes < self.max_batch_total_size_bytes:
-                self.max_batch_total_size_bytes = server_max_batch_total_size_bytes
-        except grpc.RpcError as e:
-            # Simply use the defaults for servers that don't implement
-            # GetCapabilities()
-            if e.code() != grpc.StatusCode.UNIMPLEMENTED:
-                raise
-
-        # Check whether the server supports BatchReadBlobs()
-        self.batch_read_supported = self._check_support(
-            remote_execution_pb2.BatchReadBlobsRequest,
-            self.cas.BatchReadBlobs
-        )
-
-        # Check whether the server supports BatchUpdateBlobs()
-        self.batch_update_supported = self._check_support(
-            remote_execution_pb2.BatchUpdateBlobsRequest,
-            self.cas.BatchUpdateBlobs
-        )
-
         local_cas = self.cascache._get_local_cas()
         request = local_cas_pb2.GetInstanceNameForRemoteRequest()
         request.url = self.spec.url
@@ -95,49 +69,6 @@ class CASRemote(BaseRemote):
         response = local_cas.GetInstanceNameForRemote(request)
         self.local_cas_instance_name = response.instance_name
 
-    # _check():
-    #
-    # Check if this remote provides everything required for the
-    # particular kind of remote. This is expected to be called as part
-    # of check(), and must be called in a non-main process.
-    #
-    # Returns:
-    #    (str|None): An error message, or None if no error message.
-    #
-    def _check(self):
-        request = buildstream_pb2.StatusRequest()
-        response = self.ref_storage.Status(request)
-
-        if self.spec.push and not response.allow_updates:
-            return 'CAS server does not allow push'
-
-        return None
-
-    # _check_support():
-    #
-    # Figure out if a remote server supports a given method based on
-    # grpc.StatusCode.UNIMPLEMENTED and grpc.StatusCode.PERMISSION_DENIED.
-    #
-    # Args:
-    #    request_type (callable): The type of request to check.
-    #    invoker (callable): The remote method that will be invoked.
-    #
-    # Returns:
-    #    (bool) - Whether the request is supported.
-    #
-    def _check_support(self, request_type, invoker):
-        try:
-            request = request_type()
-            if self.instance_name:
-                request.instance_name = self.instance_name
-            invoker(request)
-            return True
-        except grpc.RpcError as e:
-            if not e.code() in (grpc.StatusCode.UNIMPLEMENTED, grpc.StatusCode.PERMISSION_DENIED):
-                raise
-
-        return False
-
     # push_message():
     #
     # Push the given protobuf message to a remote.
diff --git a/src/buildstream/_remote.py b/src/buildstream/_remote.py
index 75c626c47..0c0fed44d 100644
--- a/src/buildstream/_remote.py
+++ b/src/buildstream/_remote.py
@@ -26,7 +26,6 @@ import grpc
 from . import _signals
 from . import utils
 from ._exceptions import LoadError, LoadErrorReason, ImplError, RemoteError
-from ._protos.google.bytestream import bytestream_pb2_grpc
 from .types import FastEnum
 
 
@@ -134,7 +133,6 @@ class BaseRemote():
         self.spec = spec
         self._initialized = False
 
-        self.bytestream = None
         self.channel = None
 
         self.server_cert = None
@@ -178,9 +176,6 @@ class BaseRemote():
         else:
             raise RemoteError("Unsupported URL: {}".format(self.spec.url))
 
-        # Set up the bytestream on our channel
-        self.bytestream = bytestream_pb2_grpc.ByteStreamStub(self.channel)
-
         self._configure_protocols()
 
         self._initialized = True
diff --git a/src/buildstream/_sourcecache.py b/src/buildstream/_sourcecache.py
index 76a2e4f39..03ba9a74c 100644
--- a/src/buildstream/_sourcecache.py
+++ b/src/buildstream/_sourcecache.py
@@ -36,30 +36,42 @@ class SourceRemote(BaseRemote):
         self.source_service = None
 
     def _configure_protocols(self):
+        # set up source service
+        self.source_service = source_pb2_grpc.SourceServiceStub(self.channel)
+
+    # _check():
+    #
+    # Check if this remote provides everything required for the
+    # particular kind of remote. This is expected to be called as part
+    # of check(), and must be called in a non-main process.
+    #
+    # Returns:
+    #    (str|None): An error message, or None if no error message.
+    #
+    def _check(self):
         capabilities_service = buildstream_pb2_grpc.CapabilitiesStub(self.channel)
+
         # check that the service supports sources
         try:
             request = buildstream_pb2.GetCapabilitiesRequest()
             if self.instance_name:
                 request.instance_name = self.instance_name
-
             response = capabilities_service.GetCapabilities(request)
         except grpc.RpcError as e:
             # Check if this remote has the artifact service
             if e.code() == grpc.StatusCode.UNIMPLEMENTED:
-                raise SourceCacheError(
-                    "Configured remote does not have the BuildStream "
-                    "capabilities service. Please check remote configuration.")
+                return ("Configured remote does not have the BuildStream "
+                        "capabilities service. Please check remote configuration.")
             # Else raise exception with details
-            raise SourceCacheError(
-                "Remote initialisation failed: {}".format(e.details()))
+            return "Remote initialisation failed: {}".format(e.details())
 
         if not response.source_capabilities:
-            raise SourceCacheError(
-                "Configured remote does not support source service")
+            return "Configured remote does not support source service"
 
-        # set up source service
-        self.source_service = source_pb2_grpc.SourceServiceStub(self.channel)
+        if self.spec.push and not response.source_capabilities.allow_updates:
+            return 'Source server does not allow push'
+
+        return None
 
     # get_source():
     #