From b427c2b17e15ffb53ae5ac13837cce1dfe9b7935 Mon Sep 17 00:00:00 2001 From: Jean de Klerk Date: Wed, 11 Sep 2019 18:33:11 -0600 Subject: [PATCH] storage: add benchwrapper, support STORAGE_EMULATOR_HOST --- core/google/cloud/environment_vars.py | 3 + storage/google/cloud/storage/blob.py | 11 +- storage/google/cloud/storage/client.py | 6 + storage/test_utils/README.md | 20 +++ storage/test_utils/benchwrapper.py | 48 ++++++ storage/test_utils/storage.proto | 43 ++++++ storage/test_utils/storage_pb2.py | 195 +++++++++++++++++++++++++ storage/test_utils/storage_pb2_grpc.py | 63 ++++++++ 8 files changed, 383 insertions(+), 6 deletions(-) create mode 100644 storage/test_utils/README.md create mode 100644 storage/test_utils/benchwrapper.py create mode 100644 storage/test_utils/storage.proto create mode 100644 storage/test_utils/storage_pb2.py create mode 100644 storage/test_utils/storage_pb2_grpc.py diff --git a/core/google/cloud/environment_vars.py b/core/google/cloud/environment_vars.py index d84afca35628..110b1628bed0 100644 --- a/core/google/cloud/environment_vars.py +++ b/core/google/cloud/environment_vars.py @@ -30,6 +30,9 @@ BIGTABLE_EMULATOR = "BIGTABLE_EMULATOR_HOST" """Environment variable defining host for Bigtable emulator.""" +STORAGE_EMULATOR = "STORAGE_EMULATOR_HOST" +"""Environment variable defining host for Storage emulator.""" + DISABLE_GRPC = "GOOGLE_CLOUD_DISABLE_GRPC" """Environment variable acting as flag to disable gRPC. diff --git a/storage/google/cloud/storage/blob.py b/storage/google/cloud/storage/blob.py index 5b8f217d2ec8..cc40135b57f1 100644 --- a/storage/google/cloud/storage/blob.py +++ b/storage/google/cloud/storage/blob.py @@ -59,15 +59,14 @@ from google.cloud.storage.acl import ACL from google.cloud.storage.acl import ObjectACL +_DEFAULT_STORAGE_HOST = u"https://www.googleapis.com" +_STORAGE_HOST = os.environ.get( + "STORAGE_EMULATOR_HOST", _DEFAULT_STORAGE_HOST) _API_ACCESS_ENDPOINT = "https://storage.googleapis.com" _DEFAULT_CONTENT_TYPE = u"application/octet-stream" -_DOWNLOAD_URL_TEMPLATE = ( - u"https://www.googleapis.com/download/storage/v1{path}?alt=media" -) -_BASE_UPLOAD_TEMPLATE = ( - u"https://www.googleapis.com/upload/storage/v1{bucket_path}/o?uploadType=" -) +_DOWNLOAD_URL_TEMPLATE = _STORAGE_HOST + u"/download/storage/v1{path}?alt=media" +_BASE_UPLOAD_TEMPLATE = _STORAGE_HOST + u"/upload/storage/v1{bucket_path}/o?uploadType=" _MULTIPART_URL_TEMPLATE = _BASE_UPLOAD_TEMPLATE + u"multipart" _RESUMABLE_URL_TEMPLATE = _BASE_UPLOAD_TEMPLATE + u"resumable" # NOTE: "acl" is also writeable but we defer ACL management to diff --git a/storage/google/cloud/storage/client.py b/storage/google/cloud/storage/client.py index 0bb69252aeee..870f26b2d34c 100644 --- a/storage/google/cloud/storage/client.py +++ b/storage/google/cloud/storage/client.py @@ -14,6 +14,8 @@ """Client for interacting with the Google Cloud Storage API.""" +import os + import google.api_core.client_options from google.auth.credentials import AnonymousCredentials @@ -94,6 +96,10 @@ def __init__( ) kw_args = {"client_info": client_info} + + if os.environ.get("STORAGE_EMULATOR_HOST"): + kw_args["api_endpoint"] = os.environ.get("STORAGE_EMULATOR_HOST") + if client_options: if type(client_options) == dict: client_options = google.api_core.client_options.from_dict( diff --git a/storage/test_utils/README.md b/storage/test_utils/README.md new file mode 100644 index 000000000000..faf524fbdd76 --- /dev/null +++ b/storage/test_utils/README.md @@ -0,0 +1,20 @@ +# storage benchwrapp + +main.py is a gRPC wrapper around the storage library for benchmarking purposes. + +## Running + +``` +pip install grpcio +cd storage +pip install -e . # install google.cloud.storage locally +cd test_utils +python3 benchwrapper.py --port 8081 +``` + +## Re-generating protos + +``` +pip install grpcio-tools +python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. *.proto +``` \ No newline at end of file diff --git a/storage/test_utils/benchwrapper.py b/storage/test_utils/benchwrapper.py new file mode 100644 index 000000000000..6f4ff8529beb --- /dev/null +++ b/storage/test_utils/benchwrapper.py @@ -0,0 +1,48 @@ +import argparse +import sys +import time +import grpc +import os +from concurrent import futures +import storage_pb2_grpc +import storage_pb2 +from google.cloud import storage + +_ONE_DAY_IN_SECONDS = 60 * 60 * 24 + +parser = argparse.ArgumentParser() + +if os.environ.get('STORAGE_EMULATOR_HOST') == None: + sys.exit('This benchmarking server only works when connected to an emulator. Please set STORAGE_EMULATOR_HOST.') + +parser.add_argument('--port', help='The port to run on.') + +args = parser.parse_args() + +if args.port == None: + sys.exit('Usage: python3 main.py --port 8081') + +client = storage.Client() + +class StorageBenchWrapperServicer(storage_pb2_grpc.StorageBenchWrapperServicer): + def Write(self, request, context): + # TODO(deklerk): implement this + return storage_pb2.EmptyResponse() + + def Read(self, request, context): + bucket = client.bucket(request.bucketName) + blob = storage.Blob(request.objectName, bucket) + blob.download_as_string() + return storage_pb2.EmptyResponse() + +server = grpc.server(futures.ThreadPoolExecutor(max_workers=10)) +storage_pb2_grpc.add_StorageBenchWrapperServicer_to_server(StorageBenchWrapperServicer(), server) + +print('listening on localhost:'+args.port) +server.add_insecure_port('[::]:'+args.port) +server.start() +try: + while True: + time.sleep(_ONE_DAY_IN_SECONDS) +except KeyboardInterrupt: + server.stop(0) \ No newline at end of file diff --git a/storage/test_utils/storage.proto b/storage/test_utils/storage.proto new file mode 100644 index 000000000000..055e7e7867c0 --- /dev/null +++ b/storage/test_utils/storage.proto @@ -0,0 +1,43 @@ +// Copyright 2019 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +syntax = "proto3"; + +package storage_bench; + +message ObjectRead{ + // The bucket string identifier. + string bucketName = 1; + // The object/blob string identifier. + string objectName = 2; +} + +message ObjectWrite{ + // The bucket string identifier. + string bucketName = 1; + // The object/blob string identifiers. + string objectName = 2; + // The string containing the upload file path. + string destination = 3; +} + +message EmptyResponse{ +} + +service StorageBenchWrapper{ + // Performs an upload from a specific object. + rpc Write(ObjectWrite) returns (EmptyResponse) {} + // Read a specific object. + rpc Read(ObjectRead) returns (EmptyResponse){} +} \ No newline at end of file diff --git a/storage/test_utils/storage_pb2.py b/storage/test_utils/storage_pb2.py new file mode 100644 index 000000000000..5a5e3e229b0d --- /dev/null +++ b/storage/test_utils/storage_pb2.py @@ -0,0 +1,195 @@ +# -*- coding: utf-8 -*- +# Generated by the protocol buffer compiler. DO NOT EDIT! +# source: storage.proto + +import sys +_b=sys.version_info[0]<3 and (lambda x:x) or (lambda x:x.encode('latin1')) +from google.protobuf import descriptor as _descriptor +from google.protobuf import message as _message +from google.protobuf import reflection as _reflection +from google.protobuf import symbol_database as _symbol_database +# @@protoc_insertion_point(imports) + +_sym_db = _symbol_database.Default() + + + + +DESCRIPTOR = _descriptor.FileDescriptor( + name='storage.proto', + package='storage_bench', + syntax='proto3', + serialized_options=None, + serialized_pb=_b('\n\rstorage.proto\x12\rstorage_bench\"4\n\nObjectRead\x12\x12\n\nbucketName\x18\x01 \x01(\t\x12\x12\n\nobjectName\x18\x02 \x01(\t\"J\n\x0bObjectWrite\x12\x12\n\nbucketName\x18\x01 \x01(\t\x12\x12\n\nobjectName\x18\x02 \x01(\t\x12\x13\n\x0b\x64\x65stination\x18\x03 \x01(\t\"\x0f\n\rEmptyResponse2\x9d\x01\n\x13StorageBenchWrapper\x12\x43\n\x05Write\x12\x1a.storage_bench.ObjectWrite\x1a\x1c.storage_bench.EmptyResponse\"\x00\x12\x41\n\x04Read\x12\x19.storage_bench.ObjectRead\x1a\x1c.storage_bench.EmptyResponse\"\x00\x62\x06proto3') +) + + + + +_OBJECTREAD = _descriptor.Descriptor( + name='ObjectRead', + full_name='storage_bench.ObjectRead', + filename=None, + file=DESCRIPTOR, + containing_type=None, + fields=[ + _descriptor.FieldDescriptor( + name='bucketName', full_name='storage_bench.ObjectRead.bucketName', index=0, + number=1, type=9, cpp_type=9, label=1, + has_default_value=False, default_value=_b("").decode('utf-8'), + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + serialized_options=None, file=DESCRIPTOR), + _descriptor.FieldDescriptor( + name='objectName', full_name='storage_bench.ObjectRead.objectName', index=1, + number=2, type=9, cpp_type=9, label=1, + has_default_value=False, default_value=_b("").decode('utf-8'), + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + serialized_options=None, file=DESCRIPTOR), + ], + extensions=[ + ], + nested_types=[], + enum_types=[ + ], + serialized_options=None, + is_extendable=False, + syntax='proto3', + extension_ranges=[], + oneofs=[ + ], + serialized_start=32, + serialized_end=84, +) + + +_OBJECTWRITE = _descriptor.Descriptor( + name='ObjectWrite', + full_name='storage_bench.ObjectWrite', + filename=None, + file=DESCRIPTOR, + containing_type=None, + fields=[ + _descriptor.FieldDescriptor( + name='bucketName', full_name='storage_bench.ObjectWrite.bucketName', index=0, + number=1, type=9, cpp_type=9, label=1, + has_default_value=False, default_value=_b("").decode('utf-8'), + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + serialized_options=None, file=DESCRIPTOR), + _descriptor.FieldDescriptor( + name='objectName', full_name='storage_bench.ObjectWrite.objectName', index=1, + number=2, type=9, cpp_type=9, label=1, + has_default_value=False, default_value=_b("").decode('utf-8'), + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + serialized_options=None, file=DESCRIPTOR), + _descriptor.FieldDescriptor( + name='destination', full_name='storage_bench.ObjectWrite.destination', index=2, + number=3, type=9, cpp_type=9, label=1, + has_default_value=False, default_value=_b("").decode('utf-8'), + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + serialized_options=None, file=DESCRIPTOR), + ], + extensions=[ + ], + nested_types=[], + enum_types=[ + ], + serialized_options=None, + is_extendable=False, + syntax='proto3', + extension_ranges=[], + oneofs=[ + ], + serialized_start=86, + serialized_end=160, +) + + +_EMPTYRESPONSE = _descriptor.Descriptor( + name='EmptyResponse', + full_name='storage_bench.EmptyResponse', + filename=None, + file=DESCRIPTOR, + containing_type=None, + fields=[ + ], + extensions=[ + ], + nested_types=[], + enum_types=[ + ], + serialized_options=None, + is_extendable=False, + syntax='proto3', + extension_ranges=[], + oneofs=[ + ], + serialized_start=162, + serialized_end=177, +) + +DESCRIPTOR.message_types_by_name['ObjectRead'] = _OBJECTREAD +DESCRIPTOR.message_types_by_name['ObjectWrite'] = _OBJECTWRITE +DESCRIPTOR.message_types_by_name['EmptyResponse'] = _EMPTYRESPONSE +_sym_db.RegisterFileDescriptor(DESCRIPTOR) + +ObjectRead = _reflection.GeneratedProtocolMessageType('ObjectRead', (_message.Message,), { + 'DESCRIPTOR' : _OBJECTREAD, + '__module__' : 'storage_pb2' + # @@protoc_insertion_point(class_scope:storage_bench.ObjectRead) + }) +_sym_db.RegisterMessage(ObjectRead) + +ObjectWrite = _reflection.GeneratedProtocolMessageType('ObjectWrite', (_message.Message,), { + 'DESCRIPTOR' : _OBJECTWRITE, + '__module__' : 'storage_pb2' + # @@protoc_insertion_point(class_scope:storage_bench.ObjectWrite) + }) +_sym_db.RegisterMessage(ObjectWrite) + +EmptyResponse = _reflection.GeneratedProtocolMessageType('EmptyResponse', (_message.Message,), { + 'DESCRIPTOR' : _EMPTYRESPONSE, + '__module__' : 'storage_pb2' + # @@protoc_insertion_point(class_scope:storage_bench.EmptyResponse) + }) +_sym_db.RegisterMessage(EmptyResponse) + + + +_STORAGEBENCHWRAPPER = _descriptor.ServiceDescriptor( + name='StorageBenchWrapper', + full_name='storage_bench.StorageBenchWrapper', + file=DESCRIPTOR, + index=0, + serialized_options=None, + serialized_start=180, + serialized_end=337, + methods=[ + _descriptor.MethodDescriptor( + name='Write', + full_name='storage_bench.StorageBenchWrapper.Write', + index=0, + containing_service=None, + input_type=_OBJECTWRITE, + output_type=_EMPTYRESPONSE, + serialized_options=None, + ), + _descriptor.MethodDescriptor( + name='Read', + full_name='storage_bench.StorageBenchWrapper.Read', + index=1, + containing_service=None, + input_type=_OBJECTREAD, + output_type=_EMPTYRESPONSE, + serialized_options=None, + ), +]) +_sym_db.RegisterServiceDescriptor(_STORAGEBENCHWRAPPER) + +DESCRIPTOR.services_by_name['StorageBenchWrapper'] = _STORAGEBENCHWRAPPER + +# @@protoc_insertion_point(module_scope) diff --git a/storage/test_utils/storage_pb2_grpc.py b/storage/test_utils/storage_pb2_grpc.py new file mode 100644 index 000000000000..05c8ffb73817 --- /dev/null +++ b/storage/test_utils/storage_pb2_grpc.py @@ -0,0 +1,63 @@ +# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT! +import grpc + +import storage_pb2 as storage__pb2 + + +class StorageBenchWrapperStub(object): + # missing associated documentation comment in .proto file + pass + + def __init__(self, channel): + """Constructor. + + Args: + channel: A grpc.Channel. + """ + self.Write = channel.unary_unary( + '/storage_bench.StorageBenchWrapper/Write', + request_serializer=storage__pb2.ObjectWrite.SerializeToString, + response_deserializer=storage__pb2.EmptyResponse.FromString, + ) + self.Read = channel.unary_unary( + '/storage_bench.StorageBenchWrapper/Read', + request_serializer=storage__pb2.ObjectRead.SerializeToString, + response_deserializer=storage__pb2.EmptyResponse.FromString, + ) + + +class StorageBenchWrapperServicer(object): + # missing associated documentation comment in .proto file + pass + + def Write(self, request, context): + """Performs an upload from a specific object. + """ + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') + + def Read(self, request, context): + """Read a specific object. + """ + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') + + +def add_StorageBenchWrapperServicer_to_server(servicer, server): + rpc_method_handlers = { + 'Write': grpc.unary_unary_rpc_method_handler( + servicer.Write, + request_deserializer=storage__pb2.ObjectWrite.FromString, + response_serializer=storage__pb2.EmptyResponse.SerializeToString, + ), + 'Read': grpc.unary_unary_rpc_method_handler( + servicer.Read, + request_deserializer=storage__pb2.ObjectRead.FromString, + response_serializer=storage__pb2.EmptyResponse.SerializeToString, + ), + } + generic_handler = grpc.method_handlers_generic_handler( + 'storage_bench.StorageBenchWrapper', rpc_method_handlers) + server.add_generic_rpc_handlers((generic_handler,))