Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 1 | # |
Zsolt Haraszti | 3eb27a5 | 2017-01-03 21:56:48 -0800 | [diff] [blame] | 2 | # Copyright 2017 the original author or authors. |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 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 | # |
alshabib | 06b449c | 2017-01-15 17:33:16 -0600 | [diff] [blame] | 16 | import os |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 17 | |
| 18 | import sys |
| 19 | |
| 20 | from twisted.internet import reactor |
| 21 | from twisted.internet.defer import Deferred, inlineCallbacks, returnValue |
| 22 | |
| 23 | from common.utils.asleep import asleep |
| 24 | from common.utils.consulhelpers import get_endpoint_from_consul |
| 25 | from structlog import get_logger |
| 26 | import grpc |
Rouzbahan Rashidi-Tabrizi | 34b7dd7 | 2017-03-13 17:45:20 -0400 | [diff] [blame] | 27 | from grpc import StatusCode |
| 28 | from grpc._channel import _Rendezvous |
Zsolt Haraszti | 1edb828 | 2016-11-08 10:57:19 -0800 | [diff] [blame] | 29 | from ofagent.protos import third_party |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 30 | from protos import voltha_pb2 |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 31 | from protos.voltha_pb2 import OfAgentSubscriber |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 32 | from grpc_client import GrpcClient |
| 33 | |
| 34 | from agent import Agent |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 35 | from common.utils.dockerhelpers import get_my_containers_name |
Zsolt Haraszti | 7eeb2b3 | 2016-11-06 14:04:55 -0800 | [diff] [blame] | 36 | |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 37 | |
Zsolt Haraszti | 7eeb2b3 | 2016-11-06 14:04:55 -0800 | [diff] [blame] | 38 | # _ = third_party |
Zsolt Haraszti | cd22adc | 2016-10-25 00:13:06 -0700 | [diff] [blame] | 39 | |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 40 | class ConnectionManager(object): |
Richard Jankowski | db9a86e | 2018-09-17 13:33:29 -0400 | [diff] [blame] | 41 | def __init__(self, consul_endpoint, vcore_endpoint, vcore_grpc_timeout, |
| 42 | controller_endpoints, instance_id, |
Girish | f6eeaea | 2017-11-13 10:53:57 +0530 | [diff] [blame] | 43 | enable_tls=False, key_file=None, cert_file=None, |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 44 | vcore_retry_interval=0.5, devices_refresh_interval=5, |
| 45 | subscription_refresh_interval=5): |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 46 | |
Zack Williams | 0ad0a39 | 2018-11-14 10:41:08 -0700 | [diff] [blame] | 47 | self.log = get_logger() |
| 48 | self.log.info('init-connection-manager') |
| 49 | self.log.info('list-of-controllers', |
| 50 | controller_endpoints=controller_endpoints) |
| 51 | |
sgovinda | cc73678 | 2017-05-02 20:06:37 +0530 | [diff] [blame] | 52 | self.controller_endpoints = controller_endpoints |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 53 | self.consul_endpoint = consul_endpoint |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 54 | self.vcore_endpoint = vcore_endpoint |
Richard Jankowski | db9a86e | 2018-09-17 13:33:29 -0400 | [diff] [blame] | 55 | self.grpc_timeout = vcore_grpc_timeout |
Richard Jankowski | c9d8920 | 2018-01-25 10:25:10 -0500 | [diff] [blame] | 56 | self.instance_id = instance_id |
Girish | f6eeaea | 2017-11-13 10:53:57 +0530 | [diff] [blame] | 57 | self.enable_tls = enable_tls |
| 58 | self.key_file = key_file |
| 59 | self.cert_file = cert_file |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 60 | |
| 61 | self.channel = None |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 62 | self.grpc_client = None # single, shared gRPC client to vcore |
Zsolt Haraszti | cd22adc | 2016-10-25 00:13:06 -0700 | [diff] [blame] | 63 | |
sgovinda | cc73678 | 2017-05-02 20:06:37 +0530 | [diff] [blame] | 64 | self.agent_map = {} # (datapath_id, controller_endpoint) -> Agent() |
Zsolt Haraszti | cd22adc | 2016-10-25 00:13:06 -0700 | [diff] [blame] | 65 | self.device_id_to_datapath_id_map = {} |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 66 | |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 67 | self.vcore_retry_interval = vcore_retry_interval |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 68 | self.devices_refresh_interval = devices_refresh_interval |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 69 | self.subscription_refresh_interval = subscription_refresh_interval |
| 70 | self.subscription = None |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 71 | |
| 72 | self.running = False |
| 73 | |
Zsolt Haraszti | 2bdb6b3 | 2016-11-03 16:56:17 -0700 | [diff] [blame] | 74 | def start(self): |
Zsolt Haraszti | cd22adc | 2016-10-25 00:13:06 -0700 | [diff] [blame] | 75 | |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 76 | if self.running: |
| 77 | return |
| 78 | |
Zack Williams | 0ad0a39 | 2018-11-14 10:41:08 -0700 | [diff] [blame] | 79 | self.log.debug('starting') |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 80 | |
| 81 | self.running = True |
| 82 | |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 83 | # Start monitoring the vcore grpc channel |
| 84 | reactor.callInThread(self.monitor_vcore_grpc_channel) |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 85 | |
Zsolt Haraszti | cd22adc | 2016-10-25 00:13:06 -0700 | [diff] [blame] | 86 | # Start monitoring logical devices and manage agents accordingly |
| 87 | reactor.callLater(0, self.monitor_logical_devices) |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 88 | |
Zack Williams | 0ad0a39 | 2018-11-14 10:41:08 -0700 | [diff] [blame] | 89 | self.log.info('started') |
Zsolt Haraszti | 2bdb6b3 | 2016-11-03 16:56:17 -0700 | [diff] [blame] | 90 | |
Zsolt Haraszti | cd22adc | 2016-10-25 00:13:06 -0700 | [diff] [blame] | 91 | return self |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 92 | |
Zsolt Haraszti | 2bdb6b3 | 2016-11-03 16:56:17 -0700 | [diff] [blame] | 93 | def stop(self): |
Zack Williams | 0ad0a39 | 2018-11-14 10:41:08 -0700 | [diff] [blame] | 94 | self.log.debug('stopping') |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 95 | # clean up all controller connections |
Zsolt Haraszti | 2bdb6b3 | 2016-11-03 16:56:17 -0700 | [diff] [blame] | 96 | for agent in self.agent_map.itervalues(): |
| 97 | agent.stop() |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 98 | self.running = False |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 99 | |
| 100 | self._reset_grpc_attributes() |
| 101 | |
Zack Williams | 0ad0a39 | 2018-11-14 10:41:08 -0700 | [diff] [blame] | 102 | self.log.info('stopped') |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 103 | |
| 104 | def resolve_endpoint(self, endpoint): |
| 105 | ip_port_endpoint = endpoint |
| 106 | if endpoint.startswith('@'): |
| 107 | try: |
| 108 | ip_port_endpoint = get_endpoint_from_consul( |
| 109 | self.consul_endpoint, endpoint[1:]) |
Zack Williams | 0ad0a39 | 2018-11-14 10:41:08 -0700 | [diff] [blame] | 110 | self.log.info( |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 111 | '{}-service-endpoint-found'.format(endpoint), address=ip_port_endpoint) |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 112 | except Exception as e: |
Zack Williams | 0ad0a39 | 2018-11-14 10:41:08 -0700 | [diff] [blame] | 113 | self.log.error('{}-service-endpoint-not-found'.format(endpoint), exception=repr(e)) |
| 114 | self.log.error('committing-suicide') |
alshabib | 06b449c | 2017-01-15 17:33:16 -0600 | [diff] [blame] | 115 | # Committing suicide in order to let docker restart ofagent |
| 116 | os.system("kill -15 {}".format(os.getpid())) |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 117 | if ip_port_endpoint: |
| 118 | host, port = ip_port_endpoint.split(':', 2) |
| 119 | return host, int(port) |
| 120 | |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 121 | def _reset_grpc_attributes(self): |
Zack Williams | 0ad0a39 | 2018-11-14 10:41:08 -0700 | [diff] [blame] | 122 | self.log.debug('start-reset-grpc-attributes') |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 123 | |
| 124 | if self.grpc_client is not None: |
| 125 | self.grpc_client.stop() |
| 126 | |
| 127 | if self.channel is not None: |
| 128 | del self.channel |
| 129 | |
| 130 | self.is_alive = False |
| 131 | self.channel = None |
| 132 | self.subscription = None |
| 133 | self.grpc_client = None |
| 134 | |
Zack Williams | 0ad0a39 | 2018-11-14 10:41:08 -0700 | [diff] [blame] | 135 | self.log.debug('stop-reset-grpc-attributes') |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 136 | |
| 137 | def _assign_grpc_attributes(self): |
Zack Williams | 0ad0a39 | 2018-11-14 10:41:08 -0700 | [diff] [blame] | 138 | self.log.debug('start-assign-grpc-attributes') |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 139 | |
| 140 | host, port = self.resolve_endpoint(self.vcore_endpoint) |
Zack Williams | 0ad0a39 | 2018-11-14 10:41:08 -0700 | [diff] [blame] | 141 | self.log.info('revolved-vcore-endpoint', endpoint=self.vcore_endpoint, host=host, port=port) |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 142 | |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 143 | assert host is not None |
| 144 | assert port is not None |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 145 | |
| 146 | # Establish a connection to the vcore GRPC server |
| 147 | self.channel = grpc.insecure_channel('{}:{}'.format(host, port)) |
| 148 | self.is_alive = True |
| 149 | |
Zack Williams | 0ad0a39 | 2018-11-14 10:41:08 -0700 | [diff] [blame] | 150 | self.log.debug('stop-assign-grpc-attributes') |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 151 | |
| 152 | @inlineCallbacks |
| 153 | def monitor_vcore_grpc_channel(self): |
Zack Williams | 0ad0a39 | 2018-11-14 10:41:08 -0700 | [diff] [blame] | 154 | self.log.debug('start-monitor-vcore-grpc-channel') |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 155 | |
| 156 | while self.running: |
| 157 | try: |
| 158 | # If a subscription is not yet assigned then establish new GRPC connection |
| 159 | # ... otherwise keep using existing connection details |
| 160 | if self.subscription is None: |
| 161 | self._assign_grpc_attributes() |
| 162 | |
| 163 | # Send subscription request to register the current ofagent instance |
Richard Jankowski | c9d8920 | 2018-01-25 10:25:10 -0500 | [diff] [blame] | 164 | container_name = self.instance_id |
Richard Jankowski | db9a86e | 2018-09-17 13:33:29 -0400 | [diff] [blame] | 165 | if self.grpc_client is None: |
| 166 | self.grpc_client = GrpcClient(self, self.channel, self.grpc_timeout) |
| 167 | subscription = yield self.grpc_client.subscribe( |
| 168 | OfAgentSubscriber(ofagent_id=container_name)) |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 169 | |
| 170 | # If the subscriber id matches the current instance |
| 171 | # ... then the subscription has succeeded |
| 172 | if subscription is not None and subscription.ofagent_id == container_name: |
| 173 | if self.subscription is None: |
| 174 | # Keep details on the current GRPC session and subscription |
Zack Williams | 0ad0a39 | 2018-11-14 10:41:08 -0700 | [diff] [blame] | 175 | self.log.debug('subscription-with-vcore-successful', subscription=subscription) |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 176 | self.subscription = subscription |
Richard Jankowski | db9a86e | 2018-09-17 13:33:29 -0400 | [diff] [blame] | 177 | self.grpc_client.start() |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 178 | |
| 179 | # Sleep a bit in between each subscribe |
| 180 | yield asleep(self.subscription_refresh_interval) |
| 181 | |
| 182 | # Move on to next subscribe request |
| 183 | continue |
| 184 | |
| 185 | # The subscription did not succeed, reset and move on |
| 186 | else: |
Zack Williams | 0ad0a39 | 2018-11-14 10:41:08 -0700 | [diff] [blame] | 187 | self.log.info('subscription-with-vcore-unavailable', subscription=subscription) |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 188 | |
| 189 | except _Rendezvous, e: |
Zack Williams | 0ad0a39 | 2018-11-14 10:41:08 -0700 | [diff] [blame] | 190 | self.log.error('subscription-with-vcore-terminated',exception=e, status=e.code()) |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 191 | |
| 192 | except Exception as e: |
Zack Williams | 0ad0a39 | 2018-11-14 10:41:08 -0700 | [diff] [blame] | 193 | self.log.exception('unexpected-subscription-termination-with-vcore', e=e) |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 194 | |
| 195 | # Reset grpc details |
| 196 | # The vcore instance is either not available for subscription |
| 197 | # or a failure occurred with the existing communication. |
| 198 | self._reset_grpc_attributes() |
| 199 | |
| 200 | # Sleep for a short period and retry |
| 201 | yield asleep(self.vcore_retry_interval) |
| 202 | |
Zack Williams | 0ad0a39 | 2018-11-14 10:41:08 -0700 | [diff] [blame] | 203 | self.log.debug('stop-monitor-vcore-grpc-channel') |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 204 | |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 205 | @inlineCallbacks |
Nicolas Palpacuer | 75ba77f | 2018-08-27 17:26:57 -0400 | [diff] [blame] | 206 | def get_list_of_reachable_logical_devices_from_voltha(self): |
Zsolt Haraszti | cd22adc | 2016-10-25 00:13:06 -0700 | [diff] [blame] | 207 | |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 208 | while self.running: |
Zack Williams | 0ad0a39 | 2018-11-14 10:41:08 -0700 | [diff] [blame] | 209 | self.log.debug('retrieve-logical-device-list') |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 210 | try: |
Nicolas Palpacuer | 75ba77f | 2018-08-27 17:26:57 -0400 | [diff] [blame] | 211 | devices = yield \ |
| 212 | self.grpc_client.list_reachable_logical_devices() |
| 213 | |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 214 | for device in devices: |
Zack Williams | 0ad0a39 | 2018-11-14 10:41:08 -0700 | [diff] [blame] | 215 | self.log.debug("reachable-logical-device-entry", id=device.id, |
| 216 | datapath_id=device.datapath_id) |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 217 | |
Zsolt Haraszti | cd22adc | 2016-10-25 00:13:06 -0700 | [diff] [blame] | 218 | returnValue(devices) |
| 219 | |
Rouzbahan Rashidi-Tabrizi | 34b7dd7 | 2017-03-13 17:45:20 -0400 | [diff] [blame] | 220 | except _Rendezvous, e: |
Richard Jankowski | db9a86e | 2018-09-17 13:33:29 -0400 | [diff] [blame] | 221 | status = e.code() |
Zack Williams | 0ad0a39 | 2018-11-14 10:41:08 -0700 | [diff] [blame] | 222 | self.log.error('vcore-communication-failure', exception=e, status=status) |
Richard Jankowski | db9a86e | 2018-09-17 13:33:29 -0400 | [diff] [blame] | 223 | if status == StatusCode.UNAVAILABLE or status == StatusCode.DEADLINE_EXCEEDED: |
Rouzbahan Rashidi-Tabrizi | 34b7dd7 | 2017-03-13 17:45:20 -0400 | [diff] [blame] | 224 | os.system("kill -15 {}".format(os.getpid())) |
| 225 | |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 226 | except Exception as e: |
Zack Williams | 0ad0a39 | 2018-11-14 10:41:08 -0700 | [diff] [blame] | 227 | self.log.exception('logical-devices-retrieval-failure', exception=e) |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 228 | |
Zack Williams | 0ad0a39 | 2018-11-14 10:41:08 -0700 | [diff] [blame] | 229 | self.log.info('reconnect', after_delay=self.vcore_retry_interval) |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 230 | yield asleep(self.vcore_retry_interval) |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 231 | |
Zsolt Haraszti | cd22adc | 2016-10-25 00:13:06 -0700 | [diff] [blame] | 232 | def refresh_agent_connections(self, devices): |
| 233 | """ |
| 234 | Based on the new device list, update the following state in the class: |
| 235 | * agent_map |
| 236 | * datapath_map |
| 237 | * device_id_map |
| 238 | :param devices: full device list freshly received from Voltha |
| 239 | :return: None |
| 240 | """ |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 241 | |
Zsolt Haraszti | cd22adc | 2016-10-25 00:13:06 -0700 | [diff] [blame] | 242 | # Use datapath ids for deciding what's new and what's obsolete |
| 243 | desired_datapath_ids = set(d.datapath_id for d in devices) |
sgovinda | cc73678 | 2017-05-02 20:06:37 +0530 | [diff] [blame] | 244 | current_datapath_ids = set(datapath_ids[0] for datapath_ids in self.agent_map.iterkeys()) |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 245 | |
Zsolt Haraszti | cd22adc | 2016-10-25 00:13:06 -0700 | [diff] [blame] | 246 | # if identical, nothing to do |
| 247 | if desired_datapath_ids == current_datapath_ids: |
| 248 | return |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 249 | |
Zsolt Haraszti | cd22adc | 2016-10-25 00:13:06 -0700 | [diff] [blame] | 250 | # ... otherwise calculate differences |
| 251 | to_add = desired_datapath_ids.difference(current_datapath_ids) |
| 252 | to_del = current_datapath_ids.difference(desired_datapath_ids) |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 253 | |
Zsolt Haraszti | cd22adc | 2016-10-25 00:13:06 -0700 | [diff] [blame] | 254 | # remove what we don't need |
| 255 | for datapath_id in to_del: |
| 256 | self.delete_agent(datapath_id) |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 257 | |
Zsolt Haraszti | cd22adc | 2016-10-25 00:13:06 -0700 | [diff] [blame] | 258 | # start new agents as needed |
| 259 | for device in devices: |
| 260 | if device.datapath_id in to_add: |
| 261 | self.create_agent(device) |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 262 | |
Zack Williams | 0ad0a39 | 2018-11-14 10:41:08 -0700 | [diff] [blame] | 263 | self.log.debug('updated-agent-list', count=len(self.agent_map)) |
| 264 | self.log.debug('updated-device-id-to-datapath-id-map', |
Zsolt Haraszti | cd22adc | 2016-10-25 00:13:06 -0700 | [diff] [blame] | 265 | map=str(self.device_id_to_datapath_id_map)) |
| 266 | |
| 267 | def create_agent(self, device): |
| 268 | datapath_id = device.datapath_id |
| 269 | device_id = device.id |
sgovinda | cc73678 | 2017-05-02 20:06:37 +0530 | [diff] [blame] | 270 | for controller_endpoint in self.controller_endpoints: |
| 271 | agent = Agent(controller_endpoint, datapath_id, |
Girish | f6eeaea | 2017-11-13 10:53:57 +0530 | [diff] [blame] | 272 | device_id, self.grpc_client, self.enable_tls, |
| 273 | self.key_file, self.cert_file) |
sgovinda | cc73678 | 2017-05-02 20:06:37 +0530 | [diff] [blame] | 274 | agent.start() |
| 275 | self.agent_map[(datapath_id,controller_endpoint)] = agent |
| 276 | self.device_id_to_datapath_id_map[device_id] = datapath_id |
Zsolt Haraszti | cd22adc | 2016-10-25 00:13:06 -0700 | [diff] [blame] | 277 | |
| 278 | def delete_agent(self, datapath_id): |
sgovinda | cc73678 | 2017-05-02 20:06:37 +0530 | [diff] [blame] | 279 | for controller_endpoint in self.controller_endpoints: |
| 280 | agent = self.agent_map[(datapath_id,controller_endpoint)] |
| 281 | device_id = agent.get_device_id() |
| 282 | agent.stop() |
| 283 | del self.agent_map[(datapath_id,controller_endpoint)] |
| 284 | del self.device_id_to_datapath_id_map[device_id] |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 285 | |
| 286 | @inlineCallbacks |
Zsolt Haraszti | cd22adc | 2016-10-25 00:13:06 -0700 | [diff] [blame] | 287 | def monitor_logical_devices(self): |
Zack Williams | 0ad0a39 | 2018-11-14 10:41:08 -0700 | [diff] [blame] | 288 | self.log.debug('start-monitor-logical-devices') |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 289 | |
| 290 | while self.running: |
Zack Williams | 0ad0a39 | 2018-11-14 10:41:08 -0700 | [diff] [blame] | 291 | self.log.debug('monitoring-logical-devices') |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 292 | |
alshabib | c3fb494 | 2017-01-26 15:34:24 -0800 | [diff] [blame] | 293 | # should change to a gRPC streaming call |
| 294 | # see https://jira.opencord.org/browse/CORD-821 |
Zsolt Haraszti | cd22adc | 2016-10-25 00:13:06 -0700 | [diff] [blame] | 295 | |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 296 | try: |
Richard Jankowski | db9a86e | 2018-09-17 13:33:29 -0400 | [diff] [blame] | 297 | if self.channel is not None and self.grpc_client is not None and \ |
| 298 | self.subscription is not None: |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 299 | # get current list from Voltha |
Nicolas Palpacuer | 75ba77f | 2018-08-27 17:26:57 -0400 | [diff] [blame] | 300 | reachable_devices = yield \ |
| 301 | self.get_list_of_reachable_logical_devices_from_voltha() |
Zsolt Haraszti | cd22adc | 2016-10-25 00:13:06 -0700 | [diff] [blame] | 302 | |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 303 | # update agent list and mapping tables as needed |
Nicolas Palpacuer | 75ba77f | 2018-08-27 17:26:57 -0400 | [diff] [blame] | 304 | self.refresh_agent_connections(reachable_devices) |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 305 | else: |
Zack Williams | 0ad0a39 | 2018-11-14 10:41:08 -0700 | [diff] [blame] | 306 | self.log.warning('vcore-communication-unavailable') |
Zsolt Haraszti | cd22adc | 2016-10-25 00:13:06 -0700 | [diff] [blame] | 307 | |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 308 | # wait before next poll |
| 309 | yield asleep(self.devices_refresh_interval) |
| 310 | |
| 311 | except _Rendezvous, e: |
Zack Williams | 0ad0a39 | 2018-11-14 10:41:08 -0700 | [diff] [blame] | 312 | self.log.error('vcore-communication-failure', exception=repr(e), status=e.code()) |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 313 | |
| 314 | except Exception as e: |
Zack Williams | 0ad0a39 | 2018-11-14 10:41:08 -0700 | [diff] [blame] | 315 | self.log.exception('unexpected-vcore-communication-failure', exception=repr(e)) |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 316 | |
Zack Williams | 0ad0a39 | 2018-11-14 10:41:08 -0700 | [diff] [blame] | 317 | self.log.debug('stop-monitor-logical-devices') |
Khen Nursimulu | 68b9be3 | 2016-10-25 11:57:04 -0400 | [diff] [blame] | 318 | |
Zsolt Haraszti | cd22adc | 2016-10-25 00:13:06 -0700 | [diff] [blame] | 319 | def forward_packet_in(self, device_id, ofp_packet_in): |
| 320 | datapath_id = self.device_id_to_datapath_id_map.get(device_id, None) |
| 321 | if datapath_id: |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 322 | for controller_endpoint in self.controller_endpoints: |
| 323 | agent = self.agent_map[(datapath_id, controller_endpoint)] |
| 324 | agent.forward_packet_in(ofp_packet_in) |
Zsolt Haraszti | 217a12e | 2016-12-19 16:37:55 -0800 | [diff] [blame] | 325 | |
| 326 | def forward_change_event(self, device_id, event): |
| 327 | datapath_id = self.device_id_to_datapath_id_map.get(device_id, None) |
| 328 | if datapath_id: |
Stephane Barbarie | 2940dac | 2017-08-18 14:15:17 -0400 | [diff] [blame] | 329 | for controller_endpoint in self.controller_endpoints: |
| 330 | agent = self.agent_map[(datapath_id, controller_endpoint)] |
| 331 | agent.forward_change_event(event) |