blob: 7d3839ddff753a7ca5026e82f9a454aca9b6f146 [file] [log] [blame]
Scott Bakerbef5fd92019-02-21 10:24:02 -08001#
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"""
18gRPC client meant to connect to a gRPC server endpoint, and query the
19end-point's schema by calling SchemaService.Schema(Empty) and all of its
20semantics are derived from the recovered schema.
21"""
22
Zack Williams5c2ea232019-01-30 15:23:01 -070023from __future__ import absolute_import
Scott Bakerbef5fd92019-02-21 10:24:02 -080024import os
25import sys
26import time
27from random import randint
28from zlib import decompress
29
30import functools
31import grpc
32from consul import Consul
33from grpc._channel import _Rendezvous
34from structlog import get_logger
35from twisted.internet import reactor
36from twisted.internet.defer import inlineCallbacks, returnValue
Zack Williams5c2ea232019-01-30 15:23:01 -070037from twisted.internet.error import ConnectError
Scott Bakerbef5fd92019-02-21 10:24:02 -080038
Zack Williams5c2ea232019-01-30 15:23:01 -070039from .protos.schema_pb2_grpc import SchemaServiceStub
Scott Bakerbef5fd92019-02-21 10:24:02 -080040from google.protobuf.empty_pb2 import Empty
41
Zack Williams5c2ea232019-01-30 15:23:01 -070042from .asleep import asleep
Scott Bakerbef5fd92019-02-21 10:24:02 -080043
44log = get_logger()
45
46
47class 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 Williams5c2ea232019-01-30 15:23:01 -0700147 except _Rendezvous as e:
Scott Bakerbef5fd92019-02-21 10:24:02 -0800148 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 Williams5c2ea232019-01-30 15:23:01 -0700154 except Exception:
Scott Bakerbef5fd92019-02-21 10:24:02 -0800155 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 Williams5c2ea232019-01-30 15:23:01 -0700281 _ = __import__(modname) # noqa: F841
Scott Bakerbef5fd92019-02-21 10:24:02 -0800282
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 Williams5c2ea232019-01-30 15:23:01 -0700295 raise ConnectError()
Scott Bakerbef5fd92019-02-21 10:24:02 -0800296
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 Williams5c2ea232019-01-30 15:23:01 -0700302 except grpc._channel._Rendezvous as e:
Scott Bakerbef5fd92019-02-21 10:24:02 -0800303 code = e.code()
304 if code == grpc.StatusCode.UNAVAILABLE:
Zack Williams5c2ea232019-01-30 15:23:01 -0700305 e = ConnectError()
Scott Bakerbef5fd92019-02-21 10:24:02 -0800306
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