blob: 018baa991b7ac35ccfbbc067efe2cae9f189a1d4 [file] [log] [blame]
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -07001#
Zsolt Haraszti3eb27a52017-01-03 21:56:48 -08002# Copyright 2017 the original author or authors.
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -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"""
18The gRPC client layer for the OpenFlow agent
19"""
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -070020from Queue import Queue, Empty
alshabib06b449c2017-01-15 17:33:16 -060021import os
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070022
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -070023from grpc import StatusCode
24from grpc._channel import _Rendezvous
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070025from structlog import get_logger
26from twisted.internet import reactor
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -070027from twisted.internet import threads
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070028from twisted.internet.defer import inlineCallbacks, returnValue, DeferredQueue
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -070029
Scott Bakerd865fa22018-11-07 11:45:28 -080030from protos.voltha_pb2 import ID, FlowTableUpdate, MeterModUpdate, \
Jonathan Hart398e4072018-05-30 16:54:00 -070031 FlowGroupTableUpdate, PacketOut
Scott Bakerd865fa22018-11-07 11:45:28 -080032from protos.voltha_pb2_grpc import VolthaLocalServiceStub
Jonathan Hart398e4072018-05-30 16:54:00 -070033from protos.logical_device_pb2 import LogicalPortId
Zsolt Haraszti7eeb2b32016-11-06 14:04:55 -080034from google.protobuf import empty_pb2
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070035
36
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -070037class GrpcClient(object):
38
Richard Jankowskidb9a86e2018-09-17 13:33:29 -040039 def __init__(self, connection_manager, channel, grpc_timeout):
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070040
Zack Williams18357ed2018-11-14 10:41:08 -070041 self.log = get_logger()
42
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070043 self.connection_manager = connection_manager
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -070044 self.channel = channel
Richard Jankowskidb9a86e2018-09-17 13:33:29 -040045 self.grpc_timeout = grpc_timeout
Zsolt Haraszti66862032016-11-28 14:28:39 -080046 self.local_stub = VolthaLocalServiceStub(channel)
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -070047
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -070048 self.stopped = False
49
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070050 self.packet_out_queue = Queue() # queue to send out PacketOut msgs
51 self.packet_in_queue = DeferredQueue() # queue to receive PacketIn
Zsolt Haraszti217a12e2016-12-19 16:37:55 -080052 self.change_event_queue = DeferredQueue() # queue change events
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -070053
Shad Ansari9109a282019-05-03 01:24:57 -070054 self.count_pkt_in = 0
55 self.count_pkt_out = 0
56
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -070057 def start(self):
Zack Williams18357ed2018-11-14 10:41:08 -070058 self.log.debug('starting', grpc_timeout=self.grpc_timeout)
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070059 self.start_packet_out_stream()
60 self.start_packet_in_stream()
Zsolt Haraszti217a12e2016-12-19 16:37:55 -080061 self.start_change_event_in_stream()
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070062 reactor.callLater(0, self.packet_in_forwarder_loop)
Zsolt Haraszti217a12e2016-12-19 16:37:55 -080063 reactor.callLater(0, self.change_event_processing_loop)
Zack Williams18357ed2018-11-14 10:41:08 -070064 self.log.info('started')
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -070065 return self
66
67 def stop(self):
Zack Williams18357ed2018-11-14 10:41:08 -070068 self.log.debug('stopping')
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -070069 self.stopped = True
Zack Williams18357ed2018-11-14 10:41:08 -070070 self.log.info('stopped')
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070071
72 def start_packet_out_stream(self):
73
74 def packet_generator():
75 while 1:
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -070076 try:
77 packet = self.packet_out_queue.get(block=True, timeout=1.0)
78 except Empty:
79 if self.stopped:
80 return
81 else:
Shad Ansari9109a282019-05-03 01:24:57 -070082 self.count_pkt_out += 1
83 self.log.debug('counters grpc_client OUT - {}'.format(self.count_pkt_out))
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -070084 yield packet
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070085
86 def stream_packets_out():
87 generator = packet_generator()
alshabib06b449c2017-01-15 17:33:16 -060088 try:
89 self.local_stub.StreamPacketsOut(generator)
90 except _Rendezvous, e:
Zack Williams18357ed2018-11-14 10:41:08 -070091 self.log.error('grpc-exception', status=e.code())
alshabib06b449c2017-01-15 17:33:16 -060092 if e.code() == StatusCode.UNAVAILABLE:
93 os.system("kill -15 {}".format(os.getpid()))
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070094
95 reactor.callInThread(stream_packets_out)
96
97 def start_packet_in_stream(self):
98
99 def receive_packet_in_stream():
Zsolt Haraszti66862032016-11-28 14:28:39 -0800100 streaming_rpc_method = self.local_stub.ReceivePacketsIn
Zsolt Haraszti9ed54292017-01-09 18:28:32 -0800101 iterator = streaming_rpc_method(empty_pb2.Empty())
alshabib06b449c2017-01-15 17:33:16 -0600102 try:
103 for packet_in in iterator:
104 reactor.callFromThread(self.packet_in_queue.put,
105 packet_in)
Zack Williams18357ed2018-11-14 10:41:08 -0700106 self.log.debug('enqued-packet-in',
alshabib06b449c2017-01-15 17:33:16 -0600107 packet_in=packet_in,
108 queue_len=len(self.packet_in_queue.pending))
Shad Ansari9109a282019-05-03 01:24:57 -0700109 self.count_pkt_in += 1
110 self.log.debug('counters grpc_client IN - {}'.format(self.count_pkt_in))
alshabib06b449c2017-01-15 17:33:16 -0600111 except _Rendezvous, e:
Zack Williams18357ed2018-11-14 10:41:08 -0700112 self.log.error('grpc-exception', status=e.code())
alshabib06b449c2017-01-15 17:33:16 -0600113 if e.code() == StatusCode.UNAVAILABLE:
114 os.system("kill -15 {}".format(os.getpid()))
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700115
116 reactor.callInThread(receive_packet_in_stream)
117
Zsolt Haraszti217a12e2016-12-19 16:37:55 -0800118 def start_change_event_in_stream(self):
119
120 def receive_change_events():
121 streaming_rpc_method = self.local_stub.ReceiveChangeEvents
Zsolt Haraszti9ed54292017-01-09 18:28:32 -0800122 iterator = streaming_rpc_method(empty_pb2.Empty())
alshabib06b449c2017-01-15 17:33:16 -0600123 try:
124 for event in iterator:
125 reactor.callFromThread(self.change_event_queue.put, event)
Zack Williams18357ed2018-11-14 10:41:08 -0700126 self.log.debug('enqued-change-event',
alshabib06b449c2017-01-15 17:33:16 -0600127 change_event=event,
128 queue_len=len(self.change_event_queue.pending))
129 except _Rendezvous, e:
Zack Williams18357ed2018-11-14 10:41:08 -0700130 self.log.error('grpc-exception', status=e.code())
alshabib06b449c2017-01-15 17:33:16 -0600131 if e.code() == StatusCode.UNAVAILABLE:
132 os.system("kill -15 {}".format(os.getpid()))
Zsolt Haraszti217a12e2016-12-19 16:37:55 -0800133
134 reactor.callInThread(receive_change_events)
135
136 @inlineCallbacks
137 def change_event_processing_loop(self):
138 while True:
139 try:
140 event = yield self.change_event_queue.get()
141 device_id = event.id
142 self.connection_manager.forward_change_event(device_id, event)
143 except Exception, e:
Zack Williams18357ed2018-11-14 10:41:08 -0700144 self.log.exception('failed-in-packet-in-handler', e=e)
Zsolt Haraszti217a12e2016-12-19 16:37:55 -0800145 if self.stopped:
146 break
147
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -0700148 @inlineCallbacks
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700149 def packet_in_forwarder_loop(self):
150 while True:
151 packet_in = yield self.packet_in_queue.get()
152 device_id = packet_in.id
153 ofp_packet_in = packet_in.packet_in
154 self.connection_manager.forward_packet_in(device_id, ofp_packet_in)
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -0700155 if self.stopped:
156 break
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700157
158 def send_packet_out(self, device_id, packet_out):
159 packet_out = PacketOut(id=device_id, packet_out=packet_out)
160 self.packet_out_queue.put(packet_out)
161
162 @inlineCallbacks
Jonathan Hart8d21c322018-04-17 07:42:02 -0700163 def get_port(self, device_id, port_id):
164 req = LogicalPortId(id=device_id, port_id=port_id)
165 res = yield threads.deferToThread(
Richard Jankowskidb9a86e2018-09-17 13:33:29 -0400166 self.local_stub.GetLogicalDevicePort, req, timeout=self.grpc_timeout)
Jonathan Hart8d21c322018-04-17 07:42:02 -0700167 returnValue(res)
168
169 @inlineCallbacks
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700170 def get_port_list(self, device_id):
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -0700171 req = ID(id=device_id)
172 res = yield threads.deferToThread(
Richard Jankowskidb9a86e2018-09-17 13:33:29 -0400173 self.local_stub.ListLogicalDevicePorts, req, timeout=self.grpc_timeout)
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -0700174 returnValue(res.items)
175
176 @inlineCallbacks
Jonathan Hart8d21c322018-04-17 07:42:02 -0700177 def enable_port(self, device_id, port_id):
178 req = LogicalPortId(
179 id=device_id,
180 port_id=port_id
181 )
182 res = yield threads.deferToThread(
Richard Jankowskidb9a86e2018-09-17 13:33:29 -0400183 self.local_stub.EnableLogicalDevicePort, req, timeout=self.grpc_timeout)
Jonathan Hart8d21c322018-04-17 07:42:02 -0700184 returnValue(res)
185
186 @inlineCallbacks
187 def disable_port(self, device_id, port_id):
188 req = LogicalPortId(
189 id=device_id,
190 port_id=port_id
191 )
192 res = yield threads.deferToThread(
Richard Jankowskidb9a86e2018-09-17 13:33:29 -0400193 self.local_stub.DisableLogicalDevicePort, req, timeout=self.grpc_timeout)
Jonathan Hart8d21c322018-04-17 07:42:02 -0700194 returnValue(res)
195
196 @inlineCallbacks
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700197 def get_device_info(self, device_id):
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -0700198 req = ID(id=device_id)
199 res = yield threads.deferToThread(
Richard Jankowskidb9a86e2018-09-17 13:33:29 -0400200 self.local_stub.GetLogicalDevice, req, timeout=self.grpc_timeout)
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -0700201 returnValue(res)
202
203 @inlineCallbacks
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700204 def update_flow_table(self, device_id, flow_mod):
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -0700205 req = FlowTableUpdate(
206 id=device_id,
207 flow_mod=flow_mod
208 )
209 res = yield threads.deferToThread(
Richard Jankowskidb9a86e2018-09-17 13:33:29 -0400210 self.local_stub.UpdateLogicalDeviceFlowTable, req, timeout=self.grpc_timeout)
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -0700211 returnValue(res)
212
213 @inlineCallbacks
Koray Kokten8592a232018-08-27 07:41:14 +0000214 def update_meter_mod_table(self, device_id, meter_mod):
215 req = MeterModUpdate(
216 id=device_id,
217 meter_mod=meter_mod
218 )
219 res = yield threads.deferToThread(
Gamze Abaka53cc0a22019-01-31 12:06:11 +0000220 self.local_stub.UpdateLogicalDeviceMeterTable, req, timeout=self.grpc_timeout)
Koray Kokten8592a232018-08-27 07:41:14 +0000221 returnValue(res)
222
223 @inlineCallbacks
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700224 def update_group_table(self, device_id, group_mod):
Zsolt Haraszti66862032016-11-28 14:28:39 -0800225 req = FlowGroupTableUpdate(
Zsolt Haraszti8a774382016-10-24 18:25:54 -0700226 id=device_id,
227 group_mod=group_mod
228 )
229 res = yield threads.deferToThread(
Richard Jankowskidb9a86e2018-09-17 13:33:29 -0400230 self.local_stub.UpdateLogicalDeviceFlowGroupTable, req, timeout=self.grpc_timeout)
Zsolt Haraszti8a774382016-10-24 18:25:54 -0700231 returnValue(res)
232
233 @inlineCallbacks
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700234 def list_flows(self, device_id):
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -0700235 req = ID(id=device_id)
236 res = yield threads.deferToThread(
Richard Jankowskidb9a86e2018-09-17 13:33:29 -0400237 self.local_stub.ListLogicalDeviceFlows, req, timeout=self.grpc_timeout)
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -0700238 returnValue(res.items)
Zsolt Haraszti8a774382016-10-24 18:25:54 -0700239
240 @inlineCallbacks
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700241 def list_groups(self, device_id):
Zsolt Haraszti8a774382016-10-24 18:25:54 -0700242 req = ID(id=device_id)
243 res = yield threads.deferToThread(
Richard Jankowskidb9a86e2018-09-17 13:33:29 -0400244 self.local_stub.ListLogicalDeviceFlowGroups, req, timeout=self.grpc_timeout)
Zsolt Haraszti8a774382016-10-24 18:25:54 -0700245 returnValue(res.items)
Nicolas Palpacuerfd7b8b12018-06-15 13:58:06 -0400246
247 @inlineCallbacks
Gamze Abaka53cc0a22019-01-31 12:06:11 +0000248 def list_meters(self, device_id):
249 req = ID(id=device_id)
250 res = yield threads.deferToThread(
251 self.local_stub.ListLogicalDeviceMeters, req, timeout=self.grpc_timeout)
252 returnValue(res.items)
253
254 @inlineCallbacks
Nicolas Palpacuerfd7b8b12018-06-15 13:58:06 -0400255 def list_ports(self, device_id):
256 req = ID(id=device_id)
257 res = yield threads.deferToThread(
Richard Jankowskidb9a86e2018-09-17 13:33:29 -0400258 self.local_stub.ListLogicalDevicePorts, req, timeout=self.grpc_timeout)
259 returnValue(res.items)
260
261 @inlineCallbacks
262 def list_logical_devices(self):
263 res = yield threads.deferToThread(
264 self.local_stub.ListLogicalDevices, empty_pb2.Empty(), timeout=self.grpc_timeout)
265 returnValue(res.items)
266
267 @inlineCallbacks
Nicolas Palpacuer75ba77f2018-08-27 17:26:57 -0400268 def list_reachable_logical_devices(self):
269 res = yield threads.deferToThread(
270 self.local_stub.ListReachableLogicalDevices, empty_pb2.Empty(),
271 timeout=self.grpc_timeout)
272 returnValue(res.items)
273
274 @inlineCallbacks
Richard Jankowskidb9a86e2018-09-17 13:33:29 -0400275 def subscribe(self, subscriber):
276 res = yield threads.deferToThread(
277 self.local_stub.Subscribe, subscriber, timeout=self.grpc_timeout)
278 returnValue(res)