Scott Baker | bef5fd9 | 2019-02-21 10:24:02 -0800 | [diff] [blame] | 1 | # |
| 2 | # Copyright 2017 the original author or authors. |
| 3 | # |
| 4 | # Licensed under the Apache License, Version 2.0 (the "License"); |
| 5 | # you may not use this file except in compliance with the License. |
| 6 | # You may obtain a copy of the License at |
| 7 | # |
| 8 | # http://www.apache.org/licenses/LICENSE-2.0 |
| 9 | # |
| 10 | # Unless required by applicable law or agreed to in writing, software |
| 11 | # distributed under the License is distributed on an "AS IS" BASIS, |
| 12 | # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 13 | # See the License for the specific language governing permissions and |
| 14 | # limitations under the License. |
| 15 | # |
| 16 | |
| 17 | """ |
| 18 | gRPC client meant to connect to a gRPC server endpoint, and query the |
| 19 | end-point's schema by calling SchemaService.Schema(Empty) and all of its |
| 20 | semantics are derived from the recovered schema. |
| 21 | """ |
| 22 | |
Zack Williams | 5c2ea23 | 2019-01-30 15:23:01 -0700 | [diff] [blame] | 23 | from __future__ import absolute_import |
Scott Baker | bef5fd9 | 2019-02-21 10:24:02 -0800 | [diff] [blame] | 24 | import os |
| 25 | import sys |
| 26 | import time |
| 27 | from random import randint |
| 28 | from zlib import decompress |
| 29 | |
| 30 | import functools |
| 31 | import grpc |
| 32 | from consul import Consul |
| 33 | from grpc._channel import _Rendezvous |
| 34 | from structlog import get_logger |
| 35 | from twisted.internet import reactor |
| 36 | from twisted.internet.defer import inlineCallbacks, returnValue |
Zack Williams | 5c2ea23 | 2019-01-30 15:23:01 -0700 | [diff] [blame] | 37 | from twisted.internet.error import ConnectError |
Scott Baker | bef5fd9 | 2019-02-21 10:24:02 -0800 | [diff] [blame] | 38 | |
Zack Williams | 5c2ea23 | 2019-01-30 15:23:01 -0700 | [diff] [blame] | 39 | from .protos.schema_pb2_grpc import SchemaServiceStub |
Scott Baker | bef5fd9 | 2019-02-21 10:24:02 -0800 | [diff] [blame] | 40 | from google.protobuf.empty_pb2 import Empty |
| 41 | |
Zack Williams | 5c2ea23 | 2019-01-30 15:23:01 -0700 | [diff] [blame] | 42 | from .asleep import asleep |
Scott Baker | bef5fd9 | 2019-02-21 10:24:02 -0800 | [diff] [blame] | 43 | |
| 44 | log = get_logger() |
| 45 | |
| 46 | |
| 47 | class GrpcClient(object): |
| 48 | """ |
| 49 | Connect to a gRPC server, fetch its schema, and process the downloaded |
| 50 | schema files to drive the customization of the north-bound interface(s) |
| 51 | of Chameleon. |
| 52 | """ |
| 53 | RETRY_BACKOFF = [0.05, 0.1, 0.2, 0.5, 1, 2, 5] |
| 54 | |
| 55 | def __init__(self, consul_endpoint, work_dir, endpoint='localhost:50055', |
| 56 | reconnect_callback=None, credentials=None, restart_on_disconnect=False): |
| 57 | self.consul_endpoint = consul_endpoint |
| 58 | self.endpoint = endpoint |
| 59 | self.work_dir = work_dir |
| 60 | self.reconnect_callback = reconnect_callback |
| 61 | self.credentials = credentials |
| 62 | self.restart_on_disconnect = restart_on_disconnect |
| 63 | |
| 64 | self.plugin_dir = os.path.abspath(os.path.join( |
| 65 | os.path.dirname(__file__), 'protoc_plugins')) |
| 66 | |
| 67 | self.channel = None |
| 68 | self.schema = None |
| 69 | self.retries = 0 |
| 70 | self.shutting_down = False |
| 71 | self.connected = False |
| 72 | self.was_connected = False |
| 73 | |
| 74 | def start(self): |
| 75 | log.debug('starting') |
| 76 | if not self.connected: |
| 77 | reactor.callLater(0, self.connect) |
| 78 | log.info('started') |
| 79 | return self |
| 80 | |
| 81 | def stop(self): |
| 82 | log.debug('stopping') |
| 83 | if self.shutting_down: |
| 84 | return |
| 85 | self.shutting_down = True |
| 86 | log.info('stopped') |
| 87 | |
| 88 | def set_reconnect_callback(self, reconnect_callback): |
| 89 | self.reconnect_callback = reconnect_callback |
| 90 | return self |
| 91 | |
| 92 | def connectivity_callback(self, client, connectivity): |
| 93 | if (self.was_connected) and (connectivity in [connectivity.TRANSIENT_FAILURE, connectivity.SHUTDOWN]): |
| 94 | log.info("connectivity lost -- restarting") |
| 95 | os.execv(sys.executable, ['python'] + sys.argv) |
| 96 | |
| 97 | if (connectivity == connectivity.READY): |
| 98 | self.was_connected = True |
| 99 | |
| 100 | # Sometimes gRPC transitions from READY to IDLE, skipping TRANSIENT_FAILURE even though a socket is |
| 101 | # disconnected. So on idle, force a connectivity check. |
| 102 | if (connectivity == connectivity.IDLE) and (self.was_connected): |
| 103 | connectivity = client.channel._channel.check_connectivity_state(True) |
| 104 | # The result will probably show IDLE, but passing in True has the side effect of reconnecting if the |
| 105 | # connection has been lost, which will trigger the TRANSIENT_FALURE we were looking for. |
| 106 | |
| 107 | @inlineCallbacks |
| 108 | def connect(self): |
| 109 | """ |
| 110 | (Re-)Connect to end-point |
| 111 | """ |
| 112 | |
| 113 | if self.shutting_down or self.connected: |
| 114 | return |
| 115 | |
| 116 | try: |
| 117 | if self.endpoint.startswith('@'): |
| 118 | _endpoint = yield self._get_endpoint_from_consul( |
| 119 | self.endpoint[1:]) |
| 120 | else: |
| 121 | _endpoint = self.endpoint |
| 122 | |
| 123 | if self.credentials: |
| 124 | log.info('securely connecting', endpoint=_endpoint) |
| 125 | self.channel = grpc.secure_channel(_endpoint, self.credentials) |
| 126 | else: |
| 127 | log.info('insecurely connecting', endpoint=_endpoint) |
| 128 | self.channel = grpc.insecure_channel(_endpoint) |
| 129 | |
| 130 | if self.restart_on_disconnect: |
| 131 | connectivity_callback = functools.partial(self.connectivity_callback, self) |
| 132 | self.channel.subscribe(connectivity_callback) |
| 133 | |
| 134 | # Delay between initiating connection and executing first gRPC. See CORD-3012. |
| 135 | time.sleep(0.5) |
| 136 | |
| 137 | swagger_from = self._retrieve_schema() |
| 138 | self._compile_proto_files(swagger_from) |
| 139 | self._clear_backoff() |
| 140 | |
| 141 | self.connected = True |
| 142 | if self.reconnect_callback is not None: |
| 143 | reactor.callLater(0, self.reconnect_callback) |
| 144 | |
| 145 | return |
| 146 | |
Zack Williams | 5c2ea23 | 2019-01-30 15:23:01 -0700 | [diff] [blame] | 147 | except _Rendezvous as e: |
Scott Baker | bef5fd9 | 2019-02-21 10:24:02 -0800 | [diff] [blame] | 148 | if e.code() == grpc.StatusCode.UNAVAILABLE: |
| 149 | log.info('grpc-endpoint-not-available') |
| 150 | else: |
| 151 | log.exception('rendezvous error', e=e) |
| 152 | yield self._backoff('not-available') |
| 153 | |
Zack Williams | 5c2ea23 | 2019-01-30 15:23:01 -0700 | [diff] [blame] | 154 | except Exception: |
Scott Baker | bef5fd9 | 2019-02-21 10:24:02 -0800 | [diff] [blame] | 155 | if not self.shutting_down: |
| 156 | log.exception('cannot-connect', endpoint=_endpoint) |
| 157 | yield self._backoff('unknown-error') |
| 158 | |
| 159 | reactor.callLater(0, self.connect) |
| 160 | |
| 161 | def _backoff(self, msg): |
| 162 | wait_time = self.RETRY_BACKOFF[min(self.retries, |
| 163 | len(self.RETRY_BACKOFF) - 1)] |
| 164 | self.retries += 1 |
| 165 | log.error(msg, retry_in=wait_time) |
| 166 | return asleep(wait_time) |
| 167 | |
| 168 | def _clear_backoff(self): |
| 169 | if self.retries: |
| 170 | log.info('reconnected', after_retries=self.retries) |
| 171 | self.retries = 0 |
| 172 | |
| 173 | @inlineCallbacks |
| 174 | def _get_endpoint_from_consul(self, service_name): |
| 175 | """ |
| 176 | Look up an appropriate grpc endpoint (host, port) from |
| 177 | consul, under the service name specified by service-name |
| 178 | """ |
| 179 | host = self.consul_endpoint.split(':')[0].strip() |
| 180 | port = int(self.consul_endpoint.split(':')[1].strip()) |
| 181 | |
| 182 | while True: |
| 183 | log.debug('consul-lookup', host=host, port=port) |
| 184 | consul = Consul(host=host, port=port) |
| 185 | _, services = consul.catalog.service(service_name) |
| 186 | log.debug('consul-response', services=services) |
| 187 | if services: |
| 188 | break |
| 189 | log.warning('no-service', consul_host=host, consul_port=port, |
| 190 | service_name=service_name) |
| 191 | yield asleep(1.0) |
| 192 | |
| 193 | # pick local addresses when resolving a service via consul |
| 194 | # see CORD-815 (https://jira.opencord.org/browse/CORD-815) |
| 195 | |
| 196 | service = services[randint(0, len(services) - 1)] |
| 197 | endpoint = '{}:{}'.format(service['ServiceAddress'], |
| 198 | service['ServicePort']) |
| 199 | returnValue(endpoint) |
| 200 | |
| 201 | def _retrieve_schema(self): |
| 202 | """ |
| 203 | Retrieve schema from gRPC end-point, and save all *.proto files in |
| 204 | the work directory. |
| 205 | """ |
| 206 | assert isinstance(self.channel, grpc.Channel) |
| 207 | stub = SchemaServiceStub(self.channel) |
| 208 | # try: |
| 209 | schemas = stub.GetSchema(Empty(), timeout=120) |
| 210 | # except _Rendezvous, e: |
| 211 | # if e.code == grpc.StatusCode.UNAVAILABLE: |
| 212 | # |
| 213 | # else: |
| 214 | # raise e |
| 215 | |
| 216 | os.system('mkdir -p %s' % self.work_dir) |
| 217 | os.system('rm -fr /tmp/%s/*' % |
| 218 | self.work_dir.replace('/tmp/', '')) # safer |
| 219 | |
| 220 | for proto_file in schemas.protos: |
| 221 | proto_fname = proto_file.file_name |
| 222 | proto_content = proto_file.proto |
| 223 | log.debug('saving-proto', fname=proto_fname, dir=self.work_dir, |
| 224 | length=len(proto_content)) |
| 225 | with open(os.path.join(self.work_dir, proto_fname), 'w') as f: |
| 226 | f.write(proto_content) |
| 227 | |
| 228 | desc_content = decompress(proto_file.descriptor) |
| 229 | desc_fname = proto_fname.replace('.proto', '.desc') |
| 230 | log.debug('saving-descriptor', fname=desc_fname, dir=self.work_dir, |
| 231 | length=len(desc_content)) |
| 232 | with open(os.path.join(self.work_dir, desc_fname), 'wb') as f: |
| 233 | f.write(desc_content) |
| 234 | return schemas.swagger_from |
| 235 | |
| 236 | def _compile_proto_files(self, swagger_from): |
| 237 | """ |
| 238 | For each *.proto file in the work directory, compile the proto |
| 239 | file into the respective *_pb2.py file as well as generate the |
| 240 | web server gateway python file *_gw.py. |
| 241 | :return: None |
| 242 | """ |
| 243 | |
| 244 | chameleon_base_dir = os.path.abspath(os.path.join( |
| 245 | os.path.dirname(__file__), '.' |
| 246 | )) |
| 247 | |
| 248 | for fname in [f for f in os.listdir(self.work_dir) |
| 249 | if f.endswith('.proto')]: |
| 250 | |
| 251 | need_swagger = fname == swagger_from |
| 252 | log.debug('compiling', file=fname, need_swagger=need_swagger) |
| 253 | cmd = ( |
| 254 | 'cd %s && ' |
| 255 | 'env PATH=%s PYTHONPATH=%s ' |
| 256 | 'python -m grpc.tools.protoc ' |
| 257 | '-I. ' |
| 258 | '--python_out=. ' |
| 259 | '--grpc_python_out=. ' |
| 260 | '--plugin=protoc-gen-gw=%s/gw_gen.py ' |
| 261 | '--gw_out=. ' |
| 262 | '%s' % ( |
| 263 | self.work_dir, |
| 264 | ':'.join([os.environ['PATH'], self.plugin_dir]), |
| 265 | chameleon_base_dir, |
| 266 | self.plugin_dir, |
| 267 | fname) |
| 268 | ) |
| 269 | log.debug('executing', cmd=cmd, file=fname) |
| 270 | os.system(cmd) |
| 271 | log.info('compiled', file=fname) |
| 272 | |
| 273 | # test-load each _pb2 file to see all is right |
| 274 | if self.work_dir not in sys.path: |
| 275 | sys.path.insert(0, self.work_dir) |
| 276 | |
| 277 | for fname in [f for f in os.listdir(self.work_dir) |
| 278 | if f.endswith('_pb2.py')]: |
| 279 | modname = fname[:-len('.py')] |
| 280 | log.debug('test-import', modname=modname) |
Zack Williams | 5c2ea23 | 2019-01-30 15:23:01 -0700 | [diff] [blame] | 281 | _ = __import__(modname) # noqa: F841 |
Scott Baker | bef5fd9 | 2019-02-21 10:24:02 -0800 | [diff] [blame] | 282 | |
| 283 | @inlineCallbacks |
| 284 | def invoke(self, stub, method_name, request, metadata, retry=1): |
| 285 | """ |
| 286 | Invoke a gRPC call to the remote server and return the response. |
| 287 | :param stub: Reference to the *_pb2 service stub |
| 288 | :param method_name: The method name inside the service stub |
| 289 | :param request: The request protobuf message |
| 290 | :param metadata: [(str, str), (str, str), ...] |
| 291 | :return: The response protobuf message and returned trailing metadata |
| 292 | """ |
| 293 | |
| 294 | if not self.connected: |
Zack Williams | 5c2ea23 | 2019-01-30 15:23:01 -0700 | [diff] [blame] | 295 | raise ConnectError() |
Scott Baker | bef5fd9 | 2019-02-21 10:24:02 -0800 | [diff] [blame] | 296 | |
| 297 | try: |
| 298 | method = getattr(stub(self.channel), method_name) |
| 299 | response, rendezvous = method.with_call(request, metadata=metadata) |
| 300 | returnValue((response, rendezvous.trailing_metadata())) |
| 301 | |
Zack Williams | 5c2ea23 | 2019-01-30 15:23:01 -0700 | [diff] [blame] | 302 | except grpc._channel._Rendezvous as e: |
Scott Baker | bef5fd9 | 2019-02-21 10:24:02 -0800 | [diff] [blame] | 303 | code = e.code() |
| 304 | if code == grpc.StatusCode.UNAVAILABLE: |
Zack Williams | 5c2ea23 | 2019-01-30 15:23:01 -0700 | [diff] [blame] | 305 | e = ConnectError() |
Scott Baker | bef5fd9 | 2019-02-21 10:24:02 -0800 | [diff] [blame] | 306 | |
| 307 | if self.connected: |
| 308 | self.connected = False |
| 309 | yield self.connect() |
| 310 | if retry > 0: |
| 311 | response = yield self.invoke(stub, method_name, |
| 312 | request, metadata, |
| 313 | retry=retry - 1) |
| 314 | returnValue(response) |
| 315 | |
| 316 | elif code in ( |
| 317 | grpc.StatusCode.NOT_FOUND, |
| 318 | grpc.StatusCode.INVALID_ARGUMENT, |
| 319 | grpc.StatusCode.ALREADY_EXISTS, |
| 320 | grpc.StatusCode.UNAUTHENTICATED, |
| 321 | grpc.StatusCode.PERMISSION_DENIED): |
| 322 | |
| 323 | pass # don't log error, these occur naturally |
| 324 | |
| 325 | else: |
| 326 | log.exception(e) |
| 327 | |
| 328 | raise e |