GNOME.org
 

[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 
         """

    • —
    View it on GitLab.
    You're receiving this email because of your account on gitlab.com. If you'd like to receive fewer emails, you can adjust your notification settings.

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