blob: 13b49b9cb86b729cbd6898156a4edc377a1f1915 [file] [log] [blame]
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -07001#
2# Copyright 2016 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"""
18The gRPC client layer for the OpenFlow agent
19"""
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -070020from Queue import Queue, Empty
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070021
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -070022from grpc import StatusCode
23from grpc._channel import _Rendezvous
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070024from structlog import get_logger
25from twisted.internet import reactor
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -070026from twisted.internet import threads
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070027from twisted.internet.defer import inlineCallbacks, returnValue, DeferredQueue
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -070028
Zsolt Haraszti66862032016-11-28 14:28:39 -080029from protos.voltha_pb2 import ID, VolthaLocalServiceStub, FlowTableUpdate, \
30 FlowGroupTableUpdate, PacketOut
Zsolt Haraszti7eeb2b32016-11-06 14:04:55 -080031from google.protobuf import empty_pb2
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070032
33
34log = get_logger()
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -070035
36
37class GrpcClient(object):
38
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070039 def __init__(self, connection_manager, channel):
40
41 self.connection_manager = connection_manager
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -070042 self.channel = channel
Zsolt Haraszti66862032016-11-28 14:28:39 -080043 self.local_stub = VolthaLocalServiceStub(channel)
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -070044
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -070045 self.stopped = False
46
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070047 self.packet_out_queue = Queue() # queue to send out PacketOut msgs
48 self.packet_in_queue = DeferredQueue() # queue to receive PacketIn
Zsolt Haraszti217a12e2016-12-19 16:37:55 -080049 self.change_event_queue = DeferredQueue() # queue change events
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -070050
51 def start(self):
52 log.debug('starting')
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070053 self.start_packet_out_stream()
54 self.start_packet_in_stream()
Zsolt Haraszti217a12e2016-12-19 16:37:55 -080055 self.start_change_event_in_stream()
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070056 reactor.callLater(0, self.packet_in_forwarder_loop)
Zsolt Haraszti217a12e2016-12-19 16:37:55 -080057 reactor.callLater(0, self.change_event_processing_loop)
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -070058 log.info('started')
59 return self
60
61 def stop(self):
62 log.debug('stopping')
63 self.stopped = True
64 log.info('stopped')
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070065
66 def start_packet_out_stream(self):
67
68 def packet_generator():
69 while 1:
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -070070 try:
71 packet = self.packet_out_queue.get(block=True, timeout=1.0)
72 except Empty:
73 if self.stopped:
74 return
75 else:
76 yield packet
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070077
78 def stream_packets_out():
79 generator = packet_generator()
Zsolt Haraszti66862032016-11-28 14:28:39 -080080 self.local_stub.StreamPacketsOut(generator)
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070081
82 reactor.callInThread(stream_packets_out)
83
84 def start_packet_in_stream(self):
85
86 def receive_packet_in_stream():
Zsolt Haraszti66862032016-11-28 14:28:39 -080087 streaming_rpc_method = self.local_stub.ReceivePacketsIn
Zsolt Haraszti7eeb2b32016-11-06 14:04:55 -080088 iterator = streaming_rpc_method(empty_pb2.Empty())
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -070089 for packet_in in iterator:
90 reactor.callFromThread(self.packet_in_queue.put,
91 packet_in)
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070092 log.debug('enqued-packet-in',
93 packet_in=packet_in,
94 queue_len=len(self.packet_in_queue.pending))
95
96 reactor.callInThread(receive_packet_in_stream)
97
Zsolt Haraszti217a12e2016-12-19 16:37:55 -080098 def start_change_event_in_stream(self):
99
100 def receive_change_events():
101 streaming_rpc_method = self.local_stub.ReceiveChangeEvents
102 iterator = streaming_rpc_method(empty_pb2.Empty())
103 for event in iterator:
104 reactor.callFromThread(self.change_event_queue.put, event)
105 log.debug('enqued-change-event',
106 change_event=event,
107 queue_len=len(self.change_event_queue.pending))
108
109 reactor.callInThread(receive_change_events)
110
111 @inlineCallbacks
112 def change_event_processing_loop(self):
113 while True:
114 try:
115 event = yield self.change_event_queue.get()
116 device_id = event.id
117 self.connection_manager.forward_change_event(device_id, event)
118 except Exception, e:
119 log.exception('failed-in-packet-in-handler', e=e)
120 if self.stopped:
121 break
122
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -0700123 @inlineCallbacks
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700124 def packet_in_forwarder_loop(self):
125 while True:
126 packet_in = yield self.packet_in_queue.get()
127 device_id = packet_in.id
128 ofp_packet_in = packet_in.packet_in
129 self.connection_manager.forward_packet_in(device_id, ofp_packet_in)
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -0700130 if self.stopped:
131 break
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700132
133 def send_packet_out(self, device_id, packet_out):
134 packet_out = PacketOut(id=device_id, packet_out=packet_out)
135 self.packet_out_queue.put(packet_out)
136
137 @inlineCallbacks
138 def get_port_list(self, device_id):
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -0700139 req = ID(id=device_id)
140 res = yield threads.deferToThread(
Zsolt Haraszti66862032016-11-28 14:28:39 -0800141 self.local_stub.ListLogicalDevicePorts, req)
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -0700142 returnValue(res.items)
143
144 @inlineCallbacks
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700145 def get_device_info(self, device_id):
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -0700146 req = ID(id=device_id)
147 res = yield threads.deferToThread(
Zsolt Haraszti66862032016-11-28 14:28:39 -0800148 self.local_stub.GetLogicalDevice, req)
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -0700149 returnValue(res)
150
151 @inlineCallbacks
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700152 def update_flow_table(self, device_id, flow_mod):
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -0700153 req = FlowTableUpdate(
154 id=device_id,
155 flow_mod=flow_mod
156 )
157 res = yield threads.deferToThread(
Zsolt Haraszti66862032016-11-28 14:28:39 -0800158 self.local_stub.UpdateLogicalDeviceFlowTable, req)
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -0700159 returnValue(res)
160
161 @inlineCallbacks
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700162 def update_group_table(self, device_id, group_mod):
Zsolt Haraszti66862032016-11-28 14:28:39 -0800163 req = FlowGroupTableUpdate(
Zsolt Haraszti8a774382016-10-24 18:25:54 -0700164 id=device_id,
165 group_mod=group_mod
166 )
167 res = yield threads.deferToThread(
Zsolt Haraszti66862032016-11-28 14:28:39 -0800168 self.local_stub.UpdateLogicalDeviceFlowGroupTable, req)
Zsolt Haraszti8a774382016-10-24 18:25:54 -0700169 returnValue(res)
170
171 @inlineCallbacks
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700172 def list_flows(self, device_id):
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -0700173 req = ID(id=device_id)
174 res = yield threads.deferToThread(
Zsolt Haraszti66862032016-11-28 14:28:39 -0800175 self.local_stub.ListLogicalDeviceFlows, req)
Zsolt Haraszti023ea7c2016-10-16 19:30:34 -0700176 returnValue(res.items)
Zsolt Haraszti8a774382016-10-24 18:25:54 -0700177
178 @inlineCallbacks
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700179 def list_groups(self, device_id):
Zsolt Haraszti8a774382016-10-24 18:25:54 -0700180 req = ID(id=device_id)
181 res = yield threads.deferToThread(
Zsolt Haraszti66862032016-11-28 14:28:39 -0800182 self.local_stub.ListLogicalDeviceFlowGroups, req)
Zsolt Haraszti8a774382016-10-24 18:25:54 -0700183 returnValue(res.items)