[Notes] [Git][BuildGrid/buildgrid][finn/separate-services] Adding remote storage



Title: GitLab

finn pushed to branch finn/separate-services at BuildGrid / buildgrid

Commits:

2 changed files:

Changes:

  • buildgrid/server/cas/storage/remote.py
    1
    +# Copyright (C) 2018 Bloomberg LP
    
    2
    +#
    
    3
    +# Licensed under the Apache License, Version 2.0 (the "License");
    
    4
    +# you may not use this file except in compliance with the License.
    
    5
    +# You may obtain a copy of the License at
    
    6
    +#
    
    7
    +#  <http://www.apache.org/licenses/LICENSE-2.0>
    
    8
    +#
    
    9
    +# Unless required by applicable law or agreed to in writing, software
    
    10
    +# distributed under the License is distributed on an "AS IS" BASIS,
    
    11
    +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    
    12
    +# See the License for the specific language governing permissions and
    
    13
    +# limitations under the License.
    
    14
    +
    
    15
    +
    
    16
    +"""
    
    17
    +RemoteStorage
    
    18
    +==================
    
    19
    +
    
    20
    +Forwwards storage requests to a remote storage.
    
    21
    +"""
    
    22
    +
    
    23
    +import io
    
    24
    +import logging
    
    25
    +
    
    26
    +import grpc
    
    27
    +
    
    28
    +from buildgrid.utils import create_digest, gen_fetch_blob, gen_write_request_blob
    
    29
    +from buildgrid._protos.google.bytestream import bytestream_pb2_grpc
    
    30
    +from buildgrid._protos.build.bazel.remote.execution.v2 import remote_execution_pb2, remote_execution_pb2_grpc
    
    31
    +
    
    32
    +from .storage_abc import StorageABC
    
    33
    +
    
    34
    +
    
    35
    +class RemoteStorage(StorageABC):
    
    36
    +
    
    37
    +    def __init__(self, channel, instance_name):
    
    38
    +        self.logger = logging.getLogger(__name__)
    
    39
    +        self._instance_name = instance_name
    
    40
    +        self._stub_bs = bytestream_pb2_grpc.ByteStreamStub(channel)
    
    41
    +        self._stub_cas = remote_execution_pb2_grpc.ContentAddressableStorageStub(channel)
    
    42
    +
    
    43
    +    def has_blob(self, digest):
    
    44
    +        try:
    
    45
    +            self.get_blob(digest)
    
    46
    +            return True
    
    47
    +
    
    48
    +        except grpc.RpcError as e:
    
    49
    +            if e.code() == grpc.StatusCode.NOT_FOUND:
    
    50
    +                return False
    
    51
    +            else:
    
    52
    +                raise
    
    53
    +
    
    54
    +    def get_blob(self, digest):
    
    55
    +        fetched_data = None
    
    56
    +        for data in gen_fetch_blob(self._stub_bs, digest, self._instance_name):
    
    57
    +            fetched_data += data
    
    58
    +
    
    59
    +        assert digest.size_bytes == len(fetched_data)
    
    60
    +
    
    61
    +        return io.BytesIO(fetched_data)
    
    62
    +
    
    63
    +    def begin_write(self, digest):
    
    64
    +        return gen_write_request_blob(digest, self._instance_name)
    
    65
    +
    
    66
    +    def commit_write(self, digest, write_session):
    
    67
    +        for request in write_session:
    
    68
    +            self._stub_bs.Write(request)
    
    69
    +
    
    70
    +    def missing_blobs(self, blobs):
    
    71
    +        digests = []
    
    72
    +        for blob in blobs:
    
    73
    +            digests.append(create_digest(blob))
    
    74
    +
    
    75
    +        request = remote_execution_pb2.FindMissingBlobsRequest(instance_name=self._instance_name,
    
    76
    +                                                               blob_digests=digests)
    
    77
    +        response = self._stub_cas.FindMissingBlobs(request)
    
    78
    +
    
    79
    +        return response.missing_blob_digests
    
    80
    +
    
    81
    +    def bulk_update_blobs(self, blobs):
    
    82
    +        requests = []
    
    83
    +        for blob in blobs:
    
    84
    +            request = remote_execution_pb2.BatchUpdateBlobsRequest.Request(digest=create_digest(blob),
    
    85
    +                                                                           data=blob)
    
    86
    +            requests.append(request)
    
    87
    +
    
    88
    +        response = self._stub_cas.BatchUpdateBlobs(instance_name=self._instance_name,
    
    89
    +                                                   requests=requests)
    
    90
    +        result = [x.status for x in response.responses]
    
    91
    +        return result

  • buildgrid/utils.py
    ... ... @@ -14,6 +14,7 @@
    14 14
     
    
    15 15
     
    
    16 16
     import os
    
    17
    +import uuid
    
    17 18
     
    
    18 19
     from buildgrid.settings import HASH
    
    19 20
     from buildgrid._protos.build.bazel.remote.execution.v2 import remote_execution_pb2
    
    ... ... @@ -31,6 +32,33 @@ def gen_fetch_blob(stub, digest, instance_name=""):
    31 32
             yield response.data
    
    32 33
     
    
    33 34
     
    
    35
    +def gen_write_request_blob(digest, instance_name=""):
    
    36
    +    """ Generates a write request
    
    37
    +    """
    
    38
    +    resource_name = os.path.join(instance_name, 'uploads', uuid.uuid4(),
    
    39
    +                                 'blobs', digest.hash, str(digest.size_bytes))
    
    40
    +
    
    41
    +    offset = 0
    
    42
    +    finished = False
    
    43
    +    remaining = digest.size_bytes
    
    44
    +    digest_bytes = str.encode(digest.SerializeToString())
    
    45
    +
    
    46
    +    while not finished:
    
    47
    +        chunk_size = min(remaining, 64 * 1024)
    
    48
    +        remaining -= chunk_size
    
    49
    +        finished = remaining <= 0
    
    50
    +
    
    51
    +        request = bytestream_pb2.WriteRequest()
    
    52
    +        request.resource_name = resource_name
    
    53
    +        request.write_offset = offset
    
    54
    +        request.data = digest_bytes[offset: chunk_size]
    
    55
    +        request.finish_write = finished
    
    56
    +
    
    57
    +        yield request
    
    58
    +
    
    59
    +        offset += chunk_size
    
    60
    +
    
    61
    +
    
    34 62
     def write_fetch_directory(directory, stub, digest, instance_name=""):
    
    35 63
         """ Given a directory digest, fetches files and writes them to a directory
    
    36 64
         """
    



  • [Date Prev][Date Next]   [Thread Prev][Thread Next]   [Thread Index] [Date Index] [Author Index]