Compare revisions

Raoul Hidalgo Charman · Jim MacArthur · Jim MacArthur · dececc21 · dececc21
--- a/buildstream/_artifactcache/cascache.py
+++ b/buildstream/_artifactcache/cascache.py
@@ -45,7 +45,7 @@ from .. import _yaml
 _MAX_PAYLOAD_BYTES = 1024 * 1024


-class CASRemoteSpec(namedtuple('CASRemoteSpec', 'url push server_cert client_key client_cert')):
+class CASRemoteSpec(namedtuple('CASRemoteSpec', 'url push server_cert client_key client_cert instance_name')):

    # _new_from_config_node
    #
@@ -53,7 +53,7 @@ class CASRemoteSpec(namedtuple('CASRemoteSpec', 'url push server_cert client_key
    #
    @staticmethod
    def _new_from_config_node(spec_node, basedir=None):
-        _yaml.node_validate(spec_node, ['url', 'push', 'server-cert', 'client-key', 'client-cert'])
+        _yaml.node_validate(spec_node, ['url', 'push', 'server-cert', 'client-key', 'client-cert', 'instance_name'])
        url = _yaml.node_get(spec_node, str, 'url')
        push = _yaml.node_get(spec_node, bool, 'push', default_value=False)
        if not url:
@@ -61,6 +61,8 @@ class CASRemoteSpec(namedtuple('CASRemoteSpec', 'url push server_cert client_key
            raise LoadError(LoadErrorReason.INVALID_DATA,
                            "{}: empty artifact cache URL".format(provenance))

+        instance_name = _yaml.node_get(spec_node, str, 'instance_name', default_value=None)
+
        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)
@@ -83,10 +85,10 @@ class CASRemoteSpec(namedtuple('CASRemoteSpec', 'url push server_cert client_key
            raise LoadError(LoadErrorReason.INVALID_DATA,
                            "{}: 'client-cert' was specified without 'client-key'".format(provenance))

-        return CASRemoteSpec(url, push, server_cert, client_key, client_cert)
+        return CASRemoteSpec(url, push, server_cert, client_key, client_cert, instance_name)


-CASRemoteSpec.__new__.__defaults__ = (None, None, None)
+CASRemoteSpec.__new__.__defaults__ = (None, None, None, None)


 class BlobNotFound(CASError):
@@ -248,7 +250,7 @@ class CASCache():
            remote = CASRemote(remote_spec)
            remote.init()

-            request = buildstream_pb2.StatusRequest()
+            request = buildstream_pb2.StatusRequest(instance_name=remote_spec.instance_name)
            response = remote.ref_storage.Status(request)

            if remote_spec.push and not response.allow_updates:
@@ -284,7 +286,7 @@ class CASCache():
        try:
            remote.init()

-            request = buildstream_pb2.GetReferenceRequest()
+            request = buildstream_pb2.GetReferenceRequest(instance_name=remote.spec.instance_name)
            request.key = ref
            response = remote.ref_storage.GetReference(request)

@@ -369,7 +371,7 @@ class CASCache():
                # Check whether ref is already on the server in which case
                # there is no need to push the ref
                try:
-                    request = buildstream_pb2.GetReferenceRequest()
+                    request = buildstream_pb2.GetReferenceRequest(instance_name=remote.spec.instance_name)
                    request.key = ref
                    response = remote.ref_storage.GetReference(request)

@@ -384,7 +386,7 @@ class CASCache():

                self._send_directory(remote, tree)

-                request = buildstream_pb2.UpdateReferenceRequest()
+                request = buildstream_pb2.UpdateReferenceRequest(instance_name=remote.spec.instance_name)
                request.keys.append(ref)
                request.digest.hash = tree.hash
                request.digest.size_bytes = tree.size_bytes
@@ -448,7 +450,7 @@ class CASCache():
    def verify_digest_on_remote(self, remote, digest):
        remote.init()

-        request = remote_execution_pb2.FindMissingBlobsRequest()
+        request = remote_execution_pb2.FindMissingBlobsRequest(instance_name=remote.spec.instance_name)
        request.blob_digests.extend([digest])

        response = remote.cas.FindMissingBlobs(request)
@@ -908,7 +910,10 @@ class CASCache():
            yield from self._required_blobs(dirnode.digest)

    def _fetch_blob(self, remote, digest, stream):
-        resource_name = '/'.join(['blobs', digest.hash, str(digest.size_bytes)])
+        resource_name_components = ['blobs', digest.hash, str(digest.size_bytes)]
+        if remote.spec.instance_name:
+            resource_name_components.insert(0, remote.spec.instance_name)
+        resource_name = '/'.join(resource_name_components)
        request = bytestream_pb2.ReadRequest()
        request.resource_name = resource_name
        request.read_offset = 0
@@ -1064,8 +1069,11 @@ class CASCache():
        return dirdigest

    def _send_blob(self, remote, digest, stream, u_uid=uuid.uuid4()):
-        resource_name = '/'.join(['uploads', str(u_uid), 'blobs',
-                                  digest.hash, str(digest.size_bytes)])
+        resource_name_components = ['uploads', str(u_uid), 'blobs',
+                                    digest.hash, str(digest.size_bytes)]
+        if remote.spec.instance_name:
+            resource_name_components.insert(0, remote.spec.instance_name)
+        resource_name = '/'.join(resource_name_components)

        def request_stream(resname, instream):
            offset = 0
@@ -1097,7 +1105,7 @@ class CASCache():
        missing_blobs = dict()
        # Limit size of FindMissingBlobs request
        for required_blobs_group in _grouper(required_blobs, 512):
-            request = remote_execution_pb2.FindMissingBlobsRequest()
+            request = remote_execution_pb2.FindMissingBlobsRequest(instance_name=remote.spec.instance_name)

            for required_digest in required_blobs_group:
                d = request.blob_digests.add()
@@ -1193,7 +1201,7 @@ class CASRemote():

            self.max_batch_total_size_bytes = _MAX_PAYLOAD_BYTES
            try:
-                request = remote_execution_pb2.GetCapabilitiesRequest()
+                request = remote_execution_pb2.GetCapabilitiesRequest(instance_name=self.spec.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:
@@ -1206,7 +1214,7 @@ class CASRemote():
            # Check whether the server supports BatchReadBlobs()
            self.batch_read_supported = False
            try:
-                request = remote_execution_pb2.BatchReadBlobsRequest()
+                request = remote_execution_pb2.BatchReadBlobsRequest(instance_name=self.spec.instance_name)
                response = self.cas.BatchReadBlobs(request)
                self.batch_read_supported = True
            except grpc.RpcError as e:
@@ -1216,7 +1224,7 @@ class CASRemote():
            # Check whether the server supports BatchUpdateBlobs()
            self.batch_update_supported = False
            try:
-                request = remote_execution_pb2.BatchUpdateBlobsRequest()
+                request = remote_execution_pb2.BatchUpdateBlobsRequest(instance_name=self.spec.instance_name)
                response = self.cas.BatchUpdateBlobs(request)
                self.batch_update_supported = True
            except grpc.RpcError as e:
@@ -1233,7 +1241,7 @@ class _CASBatchRead():
    def __init__(self, remote):
        self._remote = remote
        self._max_total_size_bytes = remote.max_batch_total_size_bytes
-        self._request = remote_execution_pb2.BatchReadBlobsRequest()
+        self._request = remote_execution_pb2.BatchReadBlobsRequest(instance_name=remote.spec.instance_name)
        self._size = 0
        self._sent = False

@@ -1280,7 +1288,7 @@ class _CASBatchUpdate():
    def __init__(self, remote):
        self._remote = remote
        self._max_total_size_bytes = remote.max_batch_total_size_bytes
-        self._request = remote_execution_pb2.BatchUpdateBlobsRequest()
+        self._request = remote_execution_pb2.BatchUpdateBlobsRequest(instance_name=remote.spec.instance_name)
        self._size = 0
        self._sent = False


--- a/buildstream/sandbox/_sandboxremote.py
+++ b/buildstream/sandbox/_sandboxremote.py
@@ -61,15 +61,20 @@ class SandboxRemote(Sandbox):

        self.storage_url = config.storage_service['url']
        self.exec_url = config.exec_service['url']
+
        if config.action_service:
            self.action_url = config.action_service['url']
        else:
            self.action_url = None

+        self.server_instance = config.exec_service.get('instance', None)
+        self.storage_instance = config.storage_service.get('instance', None)
+
        self.storage_remote_spec = CASRemoteSpec(self.storage_url, push=True,
                                                 server_cert=config.storage_service['server-cert'],
                                                 client_key=config.storage_service['client-key'],
-                                                 client_cert=config.storage_service['client-cert'])
+                                                 client_cert=config.storage_service['client-cert'],
+                                                 instance_name=self.storage_instance)
        self.operation_name = None

    def info(self, msg):
@@ -104,7 +109,7 @@ class SandboxRemote(Sandbox):
        remote_exec_storage_config = require_node(remote_config, 'storage-service')
        remote_exec_action_config = remote_config.get('action-cache-service')

-        _yaml.node_validate(remote_exec_service_config, ['url'])
+        _yaml.node_validate(remote_exec_service_config, ['url', 'instance'])
        _yaml.node_validate(remote_exec_storage_config, ['url'] + tls_keys)
        if remote_exec_action_config:
            _yaml.node_validate(remote_exec_action_config, ['url'])
@@ -142,7 +147,8 @@ class SandboxRemote(Sandbox):

        # Try to create a communication channel to the BuildGrid server.
        stub = remote_execution_pb2_grpc.ExecutionStub(channel)
-        request = remote_execution_pb2.ExecuteRequest(action_digest=action_digest,
+        request = remote_execution_pb2.ExecuteRequest(instance_name=self.server_instance,
+                                                      action_digest=action_digest,
                                                      skip_cache_lookup=False)

        def __run_remote_command(stub, execute_request=None, running_operation=None):
No results found