blob: 9dee5db650ebd0934fd0ca31a28e87319b63c0ca [file] [log] [blame]
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -07001#
Zsolt Harasztiaccad4a2017-01-03 21:56:48 -08002# Copyright 2017 the original author or authors.
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -07003#
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"""
Zsolt Harasztie7b60762016-10-05 17:49:27 -070018gRPC client meant to connect to a gRPC server endpoint, and query the
Zsolt Haraszticba96de2016-11-06 14:04:55 -080019end-point's schema by calling SchemaService.Schema(Empty) and all of its
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -070020semantics are derived from the recovered schema.
21"""
Zsolt Harasztie7b60762016-10-05 17:49:27 -070022
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -070023import os
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -070024import sys
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -070025from random import randint
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -070026from zlib import decompress
27
Zsolt Haraszti3d55ffc2016-10-03 22:26:41 -070028import grpc
29from consul import Consul
Zsolt Harasztie7b60762016-10-05 17:49:27 -070030from grpc._channel import _Rendezvous
Zsolt Haraszti3d55ffc2016-10-03 22:26:41 -070031from structlog import get_logger
Zsolt Harasztie7b60762016-10-05 17:49:27 -070032from twisted.internet import reactor
Zsolt Haraszti2aac6232016-11-23 11:18:23 -080033from twisted.internet.defer import inlineCallbacks, returnValue
Zsolt Harasztie7b60762016-10-05 17:49:27 -070034from werkzeug.exceptions import ServiceUnavailable
Zsolt Haraszti3d55ffc2016-10-03 22:26:41 -070035
Zsolt Haraszti21980762016-11-08 10:57:19 -080036from chameleon.protos import third_party
Zsolt Haraszticba96de2016-11-06 14:04:55 -080037from chameleon.protos.schema_pb2 import SchemaServiceStub
38from google.protobuf.empty_pb2 import Empty
39
alshabib6ca05622017-01-31 14:08:36 -080040from chameleon.utils.asleep import asleep
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -070041
42log = get_logger()
43
44
45class GrpcClient(object):
Zsolt Harasztie7b60762016-10-05 17:49:27 -070046 """
47 Connect to a gRPC server, fetch its schema, and process the downloaded
48 schema files to drive the customization of the north-bound interface(s)
49 of Chameleon.
50 """
51 RETRY_BACKOFF = [0.05, 0.1, 0.2, 0.5, 1, 2, 5]
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -070052
Zsolt Harasztidca6fa12016-11-03 16:56:17 -070053 def __init__(self, consul_endpoint, work_dir, endpoint='localhost:50055',
Scott Baker74548db2017-02-15 14:16:13 -080054 reconnect_callback=None, credentials=None):
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -070055 self.consul_endpoint = consul_endpoint
56 self.endpoint = endpoint
Zsolt Haraszti3d55ffc2016-10-03 22:26:41 -070057 self.work_dir = work_dir
Zsolt Harasztidca6fa12016-11-03 16:56:17 -070058 self.reconnect_callback = reconnect_callback
Scott Baker74548db2017-02-15 14:16:13 -080059 self.credentials = credentials
Zsolt Harasztidca6fa12016-11-03 16:56:17 -070060
Zsolt Haraszti3d55ffc2016-10-03 22:26:41 -070061 self.plugin_dir = os.path.abspath(os.path.join(
62 os.path.dirname(__file__), '../protoc_plugins'))
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -070063
64 self.channel = None
65 self.schema = None
Zsolt Harasztie7b60762016-10-05 17:49:27 -070066 self.retries = 0
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -070067 self.shutting_down = False
Zsolt Haraszti3cf36342016-10-05 20:40:19 -070068 self.connected = False
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -070069
Zsolt Harasztidca6fa12016-11-03 16:56:17 -070070 def start(self):
71 log.debug('starting')
Zsolt Haraszti3cf36342016-10-05 20:40:19 -070072 if not self.connected:
73 reactor.callLater(0, self.connect)
Zsolt Harasztidca6fa12016-11-03 16:56:17 -070074 log.info('started')
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -070075 return self
76
Zsolt Harasztidca6fa12016-11-03 16:56:17 -070077 def stop(self):
78 log.debug('stopping')
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -070079 if self.shutting_down:
80 return
81 self.shutting_down = True
Zsolt Harasztidca6fa12016-11-03 16:56:17 -070082 log.info('stopped')
83
84 def set_reconnect_callback(self, reconnect_callback):
85 self.reconnect_callback = reconnect_callback
86 return self
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -070087
Zsolt Harasztie7b60762016-10-05 17:49:27 -070088 @inlineCallbacks
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -070089 def connect(self):
Zsolt Harasztie7b60762016-10-05 17:49:27 -070090 """
91 (Re-)Connect to end-point
92 """
93
Zsolt Haraszti3cf36342016-10-05 20:40:19 -070094 if self.shutting_down or self.connected:
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -070095 return
96
97 try:
98 if self.endpoint.startswith('@'):
Zsolt Haraszti267a9062016-12-26 23:11:15 -080099 _endpoint = yield self._get_endpoint_from_consul(
100 self.endpoint[1:])
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -0700101 else:
102 _endpoint = self.endpoint
103
Scott Baker74548db2017-02-15 14:16:13 -0800104 if self.credentials:
105 log.info('securely connecting', endpoint=_endpoint)
106 self.channel = grpc.secure_channel(_endpoint, self.credentials)
107 else:
108 log.info('insecurely connecting', endpoint=_endpoint)
109 self.channel = grpc.insecure_channel(_endpoint)
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -0700110
Zsolt Haraszti21980762016-11-08 10:57:19 -0800111 swagger_from = self._retrieve_schema()
112 self._compile_proto_files(swagger_from)
Zsolt Harasztie7b60762016-10-05 17:49:27 -0700113 self._clear_backoff()
114
Zsolt Haraszti3cf36342016-10-05 20:40:19 -0700115 self.connected = True
Zsolt Harasztidca6fa12016-11-03 16:56:17 -0700116 if self.reconnect_callback is not None:
117 reactor.callLater(0, self.reconnect_callback)
Zsolt Harasztie7b60762016-10-05 17:49:27 -0700118
119 return
120
121 except _Rendezvous, e:
122 if e.code() == grpc.StatusCode.UNAVAILABLE:
123 log.info('grpc-endpoint-not-available')
124 else:
125 log.exception(e)
126 yield self._backoff('not-available')
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -0700127
128 except Exception, e:
Zsolt Harasztie7b60762016-10-05 17:49:27 -0700129 if not self.shutting_down:
130 log.exception('cannot-connect', endpoint=_endpoint)
131 yield self._backoff('unknown-error')
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -0700132
Zsolt Harasztie7b60762016-10-05 17:49:27 -0700133 reactor.callLater(0, self.connect)
134
135 def _backoff(self, msg):
136 wait_time = self.RETRY_BACKOFF[min(self.retries,
137 len(self.RETRY_BACKOFF) - 1)]
138 self.retries += 1
139 log.error(msg, retry_in=wait_time)
140 return asleep(wait_time)
141
142 def _clear_backoff(self):
143 if self.retries:
144 log.info('reconnected', after_retries=self.retries)
145 self.retries = 0
146
Zsolt Haraszti267a9062016-12-26 23:11:15 -0800147 @inlineCallbacks
Zsolt Harasztie7b60762016-10-05 17:49:27 -0700148 def _get_endpoint_from_consul(self, service_name):
149 """
150 Look up an appropriate grpc endpoint (host, port) from
151 consul, under the service name specified by service-name
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -0700152 """
153 host = self.consul_endpoint.split(':')[0].strip()
154 port = int(self.consul_endpoint.split(':')[1].strip())
155
Zsolt Haraszti267a9062016-12-26 23:11:15 -0800156 while True:
157 log.debug('consul-lookup', host=host, port=port)
158 consul = Consul(host=host, port=port)
159 _, services = consul.catalog.service(service_name)
160 log.debug('consul-response', services=services)
161 if services:
162 break
163 log.warning('no-service', consul_host=host, consul_port=port,
164 service_name=service_name)
165 yield asleep(1.0)
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -0700166
alshabibc8224902017-01-26 11:59:52 -0800167 # pick local addresses when resolving a service via consul
168 # see CORD-815 (https://jira.opencord.org/browse/CORD-815)
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -0700169
170 service = services[randint(0, len(services) - 1)]
171 endpoint = '{}:{}'.format(service['ServiceAddress'],
172 service['ServicePort'])
Zsolt Haraszti267a9062016-12-26 23:11:15 -0800173 returnValue(endpoint)
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -0700174
175 def _retrieve_schema(self):
Zsolt Harasztie7b60762016-10-05 17:49:27 -0700176 """
177 Retrieve schema from gRPC end-point, and save all *.proto files in
178 the work directory.
179 """
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -0700180 assert isinstance(self.channel, grpc.Channel)
181 stub = SchemaServiceStub(self.channel)
Zsolt Harasztie7b60762016-10-05 17:49:27 -0700182 # try:
Zsolt Haraszticba96de2016-11-06 14:04:55 -0800183 schemas = stub.GetSchema(Empty())
Zsolt Harasztie7b60762016-10-05 17:49:27 -0700184 # except _Rendezvous, e:
185 # if e.code == grpc.StatusCode.UNAVAILABLE:
186 #
187 # else:
188 # raise e
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -0700189
190 os.system('mkdir -p %s' % self.work_dir)
191 os.system('rm -fr /tmp/%s/*' %
192 self.work_dir.replace('/tmp/', '')) # safer
193
Zsolt Harasztidca6fa12016-11-03 16:56:17 -0700194 for proto_file in schemas.protos:
195 proto_fname = proto_file.file_name
196 proto_content = proto_file.proto
197 log.debug('saving-proto', fname=proto_fname, dir=self.work_dir,
198 length=len(proto_content))
199 with open(os.path.join(self.work_dir, proto_fname), 'w') as f:
200 f.write(proto_content)
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -0700201
Zsolt Harasztidca6fa12016-11-03 16:56:17 -0700202 desc_content = decompress(proto_file.descriptor)
203 desc_fname = proto_fname.replace('.proto', '.desc')
204 log.debug('saving-descriptor', fname=desc_fname, dir=self.work_dir,
205 length=len(desc_content))
206 with open(os.path.join(self.work_dir, desc_fname), 'wb') as f:
207 f.write(desc_content)
Zsolt Haraszti21980762016-11-08 10:57:19 -0800208 return schemas.swagger_from
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -0700209
Zsolt Haraszti21980762016-11-08 10:57:19 -0800210 def _compile_proto_files(self, swagger_from):
Zsolt Harasztie7b60762016-10-05 17:49:27 -0700211 """
212 For each *.proto file in the work directory, compile the proto
213 file into the respective *_pb2.py file as well as generate the
214 web server gateway python file *_gw.py.
215 :return: None
216 """
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -0700217 google_api_dir = os.path.abspath(os.path.join(
Zsolt Haraszti46c72002016-10-10 09:55:30 -0700218 os.path.dirname(__file__), '../protos/third_party'
219 ))
220
221 chameleon_base_dir = os.path.abspath(os.path.join(
222 os.path.dirname(__file__), '../..'
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -0700223 ))
224
225 for fname in [f for f in os.listdir(self.work_dir)
226 if f.endswith('.proto')]:
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -0700227
Zsolt Haraszti21980762016-11-08 10:57:19 -0800228 need_swagger = fname == swagger_from
229 log.debug('compiling', file=fname, need_swagger=need_swagger)
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -0700230 cmd = (
231 'cd %s && '
Zsolt Haraszti05b837a2016-10-05 00:18:57 -0700232 'env PATH=%s PYTHONPATH=%s '
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -0700233 'python -m grpc.tools.protoc '
234 '-I. '
235 '-I%s '
236 '--python_out=. '
237 '--grpc_python_out=. '
Zsolt Haraszti3d55ffc2016-10-03 22:26:41 -0700238 '--plugin=protoc-gen-gw=%s/gw_gen.py '
239 '--gw_out=. '
Zsolt Haraszti46c72002016-10-10 09:55:30 -0700240 '--plugin=protoc-gen-swagger=%s/swagger_gen.py '
Zsolt Haraszti21980762016-11-08 10:57:19 -0800241 '%s'
Zsolt Haraszti3d55ffc2016-10-03 22:26:41 -0700242 '%s' % (
243 self.work_dir,
244 ':'.join([os.environ['PATH'], self.plugin_dir]),
Zsolt Haraszti46c72002016-10-10 09:55:30 -0700245 ':'.join([google_api_dir, chameleon_base_dir]),
Zsolt Haraszti3d55ffc2016-10-03 22:26:41 -0700246 google_api_dir,
Zsolt Haraszti46c72002016-10-10 09:55:30 -0700247 self.plugin_dir,
Zsolt Haraszti3d55ffc2016-10-03 22:26:41 -0700248 self.plugin_dir,
Zsolt Haraszti21980762016-11-08 10:57:19 -0800249 '--swagger_out=. ' if need_swagger else '',
Zsolt Haraszti3d55ffc2016-10-03 22:26:41 -0700250 fname)
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -0700251 )
Zsolt Harasztie7b60762016-10-05 17:49:27 -0700252 log.debug('executing', cmd=cmd, file=fname)
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -0700253 os.system(cmd)
Zsolt Harasztie7b60762016-10-05 17:49:27 -0700254 log.info('compiled', file=fname)
Zsolt Harasztia9a12dc2016-09-27 13:48:35 -0700255
256 # test-load each _pb2 file to see all is right
257 if self.work_dir not in sys.path:
258 sys.path.insert(0, self.work_dir)
259
260 for fname in [f for f in os.listdir(self.work_dir)
261 if f.endswith('_pb2.py')]:
262 modname = fname[:-len('.py')]
263 log.debug('test-import', modname=modname)
264 _ = __import__(modname)
Zsolt Haraszti3d55ffc2016-10-03 22:26:41 -0700265
Zsolt Haraszti2aac6232016-11-23 11:18:23 -0800266 @inlineCallbacks
Zsolt Harasztic8cfdf32016-11-28 14:28:39 -0800267 def invoke(self, stub, method_name, request, metadata, retry=1):
Zsolt Harasztie7b60762016-10-05 17:49:27 -0700268 """
269 Invoke a gRPC call to the remote server and return the response.
270 :param stub: Reference to the *_pb2 service stub
271 :param method_name: The method name inside the service stub
272 :param request: The request protobuf message
Zsolt Harasztic8cfdf32016-11-28 14:28:39 -0800273 :param metadata: [(str, str), (str, str), ...]
274 :return: The response protobuf message and returned trailing metadata
Zsolt Harasztie7b60762016-10-05 17:49:27 -0700275 """
276
Zsolt Haraszti3cf36342016-10-05 20:40:19 -0700277 if not self.connected:
Zsolt Harasztie7b60762016-10-05 17:49:27 -0700278 raise ServiceUnavailable()
279
280 try:
Zsolt Harasztic8cfdf32016-11-28 14:28:39 -0800281 method = getattr(stub(self.channel), method_name)
282 response, rendezvous = method.with_call(request, metadata=metadata)
283 returnValue((response, rendezvous.trailing_metadata()))
Zsolt Harasztie7b60762016-10-05 17:49:27 -0700284
285 except grpc._channel._Rendezvous, e:
Zsolt Harasztic8cfdf32016-11-28 14:28:39 -0800286 code = e.code()
287 if code == grpc.StatusCode.UNAVAILABLE:
Zsolt Harasztie7b60762016-10-05 17:49:27 -0700288 e = ServiceUnavailable()
Zsolt Haraszti2aac6232016-11-23 11:18:23 -0800289
290 if self.connected:
291 self.connected = False
292 yield self.connect()
293 if retry > 0:
294 response = yield self.invoke(stub, method_name,
Zsolt Harasztic8cfdf32016-11-28 14:28:39 -0800295 request, metadata,
Zsolt Haraszti2aac6232016-11-23 11:18:23 -0800296 retry=retry - 1)
297 returnValue(response)
298
Zsolt Harasztic8cfdf32016-11-28 14:28:39 -0800299 elif code in (
300 grpc.StatusCode.NOT_FOUND,
301 grpc.StatusCode.INVALID_ARGUMENT,
Matteo Scandolo92f7f1f2017-04-28 15:56:29 -0700302 grpc.StatusCode.ALREADY_EXISTS,
303 grpc.StatusCode.UNAUTHENTICATED,
304 grpc.StatusCode.PERMISSION_DENIED):
Zsolt Harasztic8cfdf32016-11-28 14:28:39 -0800305
306 pass # don't log error, these occur naturally
307
Zsolt Harasztie7b60762016-10-05 17:49:27 -0700308 else:
309 log.exception(e)
310
Zsolt Harasztie7b60762016-10-05 17:49:27 -0700311 raise e