blob: 9ccf115277afd43846a829ce85e20d457b161b1b [file] [log] [blame]
Khen Nursimulu68b9be32016-10-25 11:57:04 -04001#
Zsolt Haraszti3eb27a52017-01-03 21:56:48 -08002# Copyright 2017 the original author or authors.
Khen Nursimulu68b9be32016-10-25 11:57:04 -04003#
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#
alshabib06b449c2017-01-15 17:33:16 -060016import os
Khen Nursimulu68b9be32016-10-25 11:57:04 -040017
18import sys
19
20from twisted.internet import reactor
21from twisted.internet.defer import Deferred, inlineCallbacks, returnValue
22
23from common.utils.asleep import asleep
24from common.utils.consulhelpers import get_endpoint_from_consul
25from structlog import get_logger
26import grpc
Zsolt Haraszti1edb8282016-11-08 10:57:19 -080027from ofagent.protos import third_party
Khen Nursimulu68b9be32016-10-25 11:57:04 -040028from protos import voltha_pb2
29from grpc_client import GrpcClient
30
31from agent import Agent
Zsolt Haraszti7eeb2b32016-11-06 14:04:55 -080032from google.protobuf.empty_pb2 import Empty
33
Khen Nursimulu68b9be32016-10-25 11:57:04 -040034
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070035log = get_logger()
Zsolt Haraszti7eeb2b32016-11-06 14:04:55 -080036# _ = third_party
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070037
Khen Nursimulu68b9be32016-10-25 11:57:04 -040038class ConnectionManager(object):
39
Khen Nursimulu68b9be32016-10-25 11:57:04 -040040 def __init__(self, consul_endpoint, voltha_endpoint, controller_endpoint,
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070041 voltha_retry_interval=0.5, devices_refresh_interval=5):
Khen Nursimulu68b9be32016-10-25 11:57:04 -040042
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070043 log.info('init-connection-manager')
Khen Nursimulu68b9be32016-10-25 11:57:04 -040044 self.controller_endpoint = controller_endpoint
45 self.consul_endpoint = consul_endpoint
46 self.voltha_endpoint = voltha_endpoint
47
48 self.channel = None
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070049 self.grpc_client = None # single, shared gRPC client to Voltha
50
51 self.agent_map = {} # datapath_id -> Agent()
52 self.device_id_to_datapath_id_map = {}
Khen Nursimulu68b9be32016-10-25 11:57:04 -040053
54 self.voltha_retry_interval = voltha_retry_interval
55 self.devices_refresh_interval = devices_refresh_interval
56
57 self.running = False
58
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -070059 def start(self):
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070060
Khen Nursimulu68b9be32016-10-25 11:57:04 -040061 if self.running:
62 return
63
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -070064 log.debug('starting')
Khen Nursimulu68b9be32016-10-25 11:57:04 -040065
66 self.running = True
67
68 # Get voltha grpc endpoint
69 self.channel = self.get_grpc_channel_with_voltha()
70
Khen Nursimulu68b9be32016-10-25 11:57:04 -040071 # Create shared gRPC API object
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -070072 self.grpc_client = GrpcClient(self, self.channel).start()
Khen Nursimulu68b9be32016-10-25 11:57:04 -040073
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070074 # Start monitoring logical devices and manage agents accordingly
75 reactor.callLater(0, self.monitor_logical_devices)
Khen Nursimulu68b9be32016-10-25 11:57:04 -040076
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -070077 log.info('started')
78
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070079 return self
Khen Nursimulu68b9be32016-10-25 11:57:04 -040080
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -070081 def stop(self):
82 log.debug('stopping')
Khen Nursimulu68b9be32016-10-25 11:57:04 -040083 # clean up all controller connections
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -070084 for agent in self.agent_map.itervalues():
85 agent.stop()
Khen Nursimulu68b9be32016-10-25 11:57:04 -040086 self.running = False
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -070087 self.grpc_client.stop()
88 del self.channel
89 log.info('stopped')
Khen Nursimulu68b9be32016-10-25 11:57:04 -040090
91 def resolve_endpoint(self, endpoint):
92 ip_port_endpoint = endpoint
93 if endpoint.startswith('@'):
94 try:
95 ip_port_endpoint = get_endpoint_from_consul(
96 self.consul_endpoint, endpoint[1:])
Zsolt Haraszticd22adc2016-10-25 00:13:06 -070097 log.info(
Khen Nursimulu68b9be32016-10-25 11:57:04 -040098 'Found endpoint {} service at {}'.format(endpoint,
99 ip_port_endpoint))
100 except Exception as e:
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700101 log.error('Failure to locate {} service from '
Khen Nursimulu68b9be32016-10-25 11:57:04 -0400102 'consul {}:'.format(endpoint, repr(e)))
alshabib06b449c2017-01-15 17:33:16 -0600103 log.error('Committing suicide...')
104 # Committing suicide in order to let docker restart ofagent
105 os.system("kill -15 {}".format(os.getpid()))
Khen Nursimulu68b9be32016-10-25 11:57:04 -0400106 if ip_port_endpoint:
107 host, port = ip_port_endpoint.split(':', 2)
108 return host, int(port)
109
110 def get_grpc_channel_with_voltha(self):
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700111 log.info('Resolving voltha endpoint {} from consul'.format(
Khen Nursimulu68b9be32016-10-25 11:57:04 -0400112 self.voltha_endpoint))
113 host, port = self.resolve_endpoint(self.voltha_endpoint)
114 assert host is not None
115 assert port is not None
116 # Create grpc channel to Voltha
117 channel = grpc.insecure_channel('{}:{}'.format(host, port))
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700118 log.info('Acquired a grpc channel to voltha')
Khen Nursimulu68b9be32016-10-25 11:57:04 -0400119 return channel
120
Khen Nursimulu68b9be32016-10-25 11:57:04 -0400121 @inlineCallbacks
122 def get_list_of_logical_devices_from_voltha(self):
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700123
Khen Nursimulu68b9be32016-10-25 11:57:04 -0400124 while True:
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700125 log.info('Retrieve devices from voltha')
Khen Nursimulu68b9be32016-10-25 11:57:04 -0400126 try:
Zsolt Haraszti66862032016-11-28 14:28:39 -0800127 stub = voltha_pb2.VolthaLocalServiceStub(self.channel)
Zsolt Haraszti7eeb2b32016-11-06 14:04:55 -0800128 devices = stub.ListLogicalDevices(Empty()).items
Khen Nursimulu68b9be32016-10-25 11:57:04 -0400129 for device in devices:
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700130 log.info("Devices {} -> {}".format(device.id,
Zsolt Haraszti3300f742017-01-09 01:14:20 -0800131 device.datapath_id))
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700132 returnValue(devices)
133
Khen Nursimulu68b9be32016-10-25 11:57:04 -0400134 except Exception as e:
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700135 log.error('Failure to retrieve devices from '
Khen Nursimulu68b9be32016-10-25 11:57:04 -0400136 'voltha: {}'.format(repr(e)))
137
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700138 log.info('reconnect', after_delay=self.voltha_retry_interval)
Khen Nursimulu68b9be32016-10-25 11:57:04 -0400139 yield asleep(self.voltha_retry_interval)
140
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700141 def refresh_agent_connections(self, devices):
142 """
143 Based on the new device list, update the following state in the class:
144 * agent_map
145 * datapath_map
146 * device_id_map
147 :param devices: full device list freshly received from Voltha
148 :return: None
149 """
Khen Nursimulu68b9be32016-10-25 11:57:04 -0400150
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700151 # Use datapath ids for deciding what's new and what's obsolete
152 desired_datapath_ids = set(d.datapath_id for d in devices)
153 current_datapath_ids = set(self.agent_map.iterkeys())
Khen Nursimulu68b9be32016-10-25 11:57:04 -0400154
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700155 # if identical, nothing to do
156 if desired_datapath_ids == current_datapath_ids:
157 return
Khen Nursimulu68b9be32016-10-25 11:57:04 -0400158
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700159 # ... otherwise calculate differences
160 to_add = desired_datapath_ids.difference(current_datapath_ids)
161 to_del = current_datapath_ids.difference(desired_datapath_ids)
Khen Nursimulu68b9be32016-10-25 11:57:04 -0400162
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700163 # remove what we don't need
164 for datapath_id in to_del:
165 self.delete_agent(datapath_id)
Khen Nursimulu68b9be32016-10-25 11:57:04 -0400166
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700167 # start new agents as needed
168 for device in devices:
169 if device.datapath_id in to_add:
170 self.create_agent(device)
Khen Nursimulu68b9be32016-10-25 11:57:04 -0400171
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700172 log.debug('updated-agent-list', count=len(self.agent_map))
173 log.debug('updated-device-id-to-datapath-id-map',
174 map=str(self.device_id_to_datapath_id_map))
175
176 def create_agent(self, device):
177 datapath_id = device.datapath_id
178 device_id = device.id
179 agent = Agent(self.controller_endpoint, datapath_id,
180 device_id, self.grpc_client)
Zsolt Haraszti2bdb6b32016-11-03 16:56:17 -0700181 agent.start()
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700182 self.agent_map[datapath_id] = agent
183 self.device_id_to_datapath_id_map[device_id] = datapath_id
184
185 def delete_agent(self, datapath_id):
186 agent = self.agent_map[datapath_id]
187 device_id = agent.get_device_id()
188 agent.stop()
189 del self.agent_map[datapath_id]
190 del self.device_id_to_datapath_id_map[device_id]
Khen Nursimulu68b9be32016-10-25 11:57:04 -0400191
192 @inlineCallbacks
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700193 def monitor_logical_devices(self):
Khen Nursimulu68b9be32016-10-25 11:57:04 -0400194 while True:
alshabibc3fb4942017-01-26 15:34:24 -0800195 # should change to a gRPC streaming call
196 # see https://jira.opencord.org/browse/CORD-821
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700197
198 # get current list from Voltha
199 devices = yield self.get_list_of_logical_devices_from_voltha()
200
201 # update agent list and mapping tables as needed
202 self.refresh_agent_connections(devices)
203
204 # wait before next poll
Khen Nursimulu68b9be32016-10-25 11:57:04 -0400205 yield asleep(self.devices_refresh_interval)
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700206 log.info('Monitor connections')
Khen Nursimulu68b9be32016-10-25 11:57:04 -0400207
Zsolt Haraszticd22adc2016-10-25 00:13:06 -0700208 def forward_packet_in(self, device_id, ofp_packet_in):
209 datapath_id = self.device_id_to_datapath_id_map.get(device_id, None)
210 if datapath_id:
211 agent = self.agent_map[datapath_id]
212 agent.forward_packet_in(ofp_packet_in)
Zsolt Haraszti217a12e2016-12-19 16:37:55 -0800213
214 def forward_change_event(self, device_id, event):
215 datapath_id = self.device_id_to_datapath_id_map.get(device_id, None)
216 if datapath_id:
217 agent = self.agent_map[datapath_id]
218 agent.forward_change_event(event)