Gitiles
Code Review Sign In
gerrit.opencord.org / voltha-openolt-adapter / 6f436d0b1265819a68f393ed59de8401d60354e4 / . / python / adapters / openolt / openolt_device.py
blob: 2779dc6cb92928b83557c02f32f8feed071484f7 [file] [log] [blame]
William Kurkian6f436d02019-02-06 16:25:01 -0500[diff] [blame^]1#
2# Copyright 2018 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#
16import threading
17import binascii
18import grpc
19import socket
20import re
21import structlog
22from twisted.internet import reactor
23from scapy.layers.l2 import Ether, Dot1Q
24from transitions import Machine
25
26#from voltha.adapters.openolt.protos import openolt_pb2_grpc, openolt_pb2
27from python.protos.bbf_fiber_tcont_body_pb2 import TcontsConfigData
28from python.protos.bbf_fiber_gemport_body_pb2 import GemportsConfigData
29
30from python.adapters.extensions.alarms.onu.onu_discovery_alarm import OnuDiscoveryAlarm
31
32from python.common.utils.nethelpers import mac_str_to_tuple
33from python.protos.openflow_13_pb2 import OFPPS_LIVE, OFPPF_FIBER, \
34 OFPPS_LINK_DOWN, OFPPF_1GB_FD, \
35 OFPC_GROUP_STATS, OFPC_PORT_STATS, OFPC_TABLE_STATS, OFPC_FLOW_STATS, \
36 ofp_switch_features, ofp_port, ofp_port_stats, ofp_desc
37from python.common.utils.registry import registry
38from python.protos import openolt_pb2
39from python.protos import third_party
40from python.protos.common_pb2 import AdminStatus, OperStatus, ConnectStatus
41from python.protos.common_pb2 import LogLevel
42from python.protos.device_pb2 import Port, Device
43
44from python.protos.logical_device_pb2 import LogicalDevice, LogicalPort
45
46class OpenoltDevice(object):
47 """
48 OpenoltDevice state machine:
49
50 null ----> init ------> connected -----> up -----> down
51 ^ ^ | ^ | |
52 | | | | | |
53 | +-------------+ +---------+ |
54 | |
55 +-----------------------------------------+
56 """
57 # pylint: disable=too-many-instance-attributes
58 # pylint: disable=R0904
59 states = [
60 'state_null',
61 'state_init',
62 'state_connected',
63 'state_up',
64 'state_down']
65
66 transitions = [
67 {'trigger': 'go_state_init',
68 'source': ['state_null', 'state_connected', 'state_down'],
69 'dest': 'state_init',
70 'before': 'do_state_init',
71 'after': 'post_init'},
72 {'trigger': 'go_state_connected',
73 'source': 'state_init',
74 'dest': 'state_connected',
75 'before': 'do_state_connected'},
76 {'trigger': 'go_state_up',
77 'source': ['state_connected', 'state_down'],
78 'dest': 'state_up',
79 'before': 'do_state_up'},
80 {'trigger': 'go_state_down',
81 'source': ['state_up'],
82 'dest': 'state_down',
83 'before': 'do_state_down',
84 'after': 'post_down'}]
85
86 def __init__(self, **kwargs):
87 super(OpenoltDevice, self).__init__()
88
89 self.adapter_agent = kwargs['adapter_agent']
90 self.device_num = kwargs['device_num']
91 device = kwargs['device']
92
93 self.platform_class = kwargs['support_classes']['platform']
94 self.resource_mgr_class = kwargs['support_classes']['resource_mgr']
95 self.flow_mgr_class = kwargs['support_classes']['flow_mgr']
96 self.alarm_mgr_class = kwargs['support_classes']['alarm_mgr']
97 self.stats_mgr_class = kwargs['support_classes']['stats_mgr']
98 self.bw_mgr_class = kwargs['support_classes']['bw_mgr']
99
100 is_reconciliation = kwargs.get('reconciliation', False)
101 self.device_id = device.id
102 self.host_and_port = device.host_and_port
103 self.extra_args = device.extra_args
104 self.log = structlog.get_logger(id=self.device_id,
105 ip=self.host_and_port)
106 self.proxy = registry('core').get_proxy('/')
107
108 self.log.info('openolt-device-init')
109
110 # default device id and device serial number. If device_info provides better results, they will be updated
111 self.dpid = kwargs.get('dp_id')
112 self.serial_number = self.host_and_port # FIXME
113
114 # Device already set in the event of reconciliation
115 if not is_reconciliation:
116 self.log.info('updating-device')
117 # It is a new device
118 # Update device
119 device.root = True
120 device.connect_status = ConnectStatus.UNREACHABLE
121 device.oper_status = OperStatus.ACTIVATING
122 self.adapter_agent.update_device(device)
123
124 # If logical device does exist use it, else create one after connecting to device
125 if device.parent_id:
126 # logical device already exists
127 self.logical_device_id = device.parent_id
128 if is_reconciliation:
129 self.adapter_agent.reconcile_logical_device(
130 self.logical_device_id)
131
132 # Initialize the OLT state machine
133 self.machine = Machine(model=self, states=OpenoltDevice.states,
134 transitions=OpenoltDevice.transitions,
135 send_event=True, initial='state_null')
136 self.go_state_init()
137
138 def create_logical_device(self, device_info):
139 dpid = device_info.device_id
140 serial_number = device_info.device_serial_number
141
142 if dpid is None: dpid = self.dpid
143 if serial_number is None: serial_number = self.serial_number
144
145 if dpid == None or dpid == '':
146 uri = self.host_and_port.split(":")[0]
147 try:
148 socket.inet_pton(socket.AF_INET, uri)
149 dpid = '00:00:' + self.ip_hex(uri)
150 except socket.error:
151 # this is not an IP
152 dpid = self.stringToMacAddr(uri)
153
154 if serial_number == None or serial_number == '':
155 serial_number = self.host_and_port
156
157 self.log.info('creating-openolt-logical-device', dp_id=dpid, serial_number=serial_number)
158
159 mfr_desc = device_info.vendor
160 sw_desc = device_info.firmware_version
161 hw_desc = device_info.model
162 if device_info.hardware_version: hw_desc += '-' + device_info.hardware_version
163
164 # Create logical OF device
165 ld = LogicalDevice(
166 root_device_id=self.device_id,
167 switch_features=ofp_switch_features(
168 n_buffers=256, # TODO fake for now
169 n_tables=2, # TODO ditto
170 capabilities=( # TODO and ditto
171 OFPC_FLOW_STATS
172 | OFPC_TABLE_STATS
173 | OFPC_PORT_STATS
174 | OFPC_GROUP_STATS
175 )
176 ),
177 desc=ofp_desc(
178 serial_num=serial_number
179 )
180 )
181 ld_init = self.adapter_agent.create_logical_device(ld,
182 dpid=dpid)
183
184 self.logical_device_id = ld_init.id
185
186 device = self.adapter_agent.get_device(self.device_id)
187 device.serial_number = serial_number
188 self.adapter_agent.update_device(device)
189
190 self.dpid = dpid
191 self.serial_number = serial_number
192
193 self.log.info('created-openolt-logical-device', logical_device_id=ld_init.id)
194
195 def stringToMacAddr(self, uri):
196 regex = re.compile('[^a-zA-Z]')
197 uri = regex.sub('', uri)
198
199 l = len(uri)
200 if l > 6:
201 uri = uri[0:6]
202 else:
203 uri = uri + uri[0:6 - l]
204
205 print uri
206
207 return ":".join([hex(ord(x))[-2:] for x in uri])
208
209 def do_state_init(self, event):
210 # Initialize gRPC
211 self.channel = grpc.insecure_channel(self.host_and_port)
212 self.channel_ready_future = grpc.channel_ready_future(self.channel)
213
214 self.log.info('openolt-device-created', device_id=self.device_id)
215
216 def post_init(self, event):
217 self.log.debug('post_init')
218
219 # We have reached init state, starting the indications thread
220
221 # Catch RuntimeError exception
222 try:
223 # Start indications thread
224 self.indications_thread_handle = threading.Thread(
225 target=self.indications_thread)
226 # Old getter/setter API for daemon; use it directly as a
227 # property instead. The Jinkins error will happon on the reason of
228 # Exception in thread Thread-1 (most likely raised # during
229 # interpreter shutdown)
230 self.indications_thread_handle.setDaemon(True)
231 self.indications_thread_handle.start()
232 except Exception as e:
233 self.log.exception('post_init failed', e=e)
234
235 def do_state_connected(self, event):
236 self.log.debug("do_state_connected")
237
238 device = self.adapter_agent.get_device(self.device_id)
239
240 self.stub = openolt_pb2_grpc.OpenoltStub(self.channel)
241
242 device_info = self.stub.GetDeviceInfo(openolt_pb2.Empty())
243 self.log.info('Device connected', device_info=device_info)
244
245 self.create_logical_device(device_info)
246
247 device.serial_number = self.serial_number
248
249 self.resource_mgr = self.resource_mgr_class(self.device_id,
250 self.host_and_port,
251 self.extra_args,
252 device_info)
253 self.platform = self.platform_class(self.log, self.resource_mgr)
254 self.flow_mgr = self.flow_mgr_class(self.adapter_agent, self.log,
255 self.stub, self.device_id,
256 self.logical_device_id,
257 self.platform, self.resource_mgr)
258
259 self.alarm_mgr = self.alarm_mgr_class(self.log, self.adapter_agent,
260 self.device_id,
261 self.logical_device_id,
262 self.platform)
263 self.stats_mgr = self.stats_mgr_class(self, self.log, self.platform)
264 self.bw_mgr = self.bw_mgr_class(self.log, self.proxy)
265
266 device.vendor = device_info.vendor
267 device.model = device_info.model
268 device.hardware_version = device_info.hardware_version
269 device.firmware_version = device_info.firmware_version
270
271 # TODO: check for uptime and reboot if too long (VOL-1192)
272
273 device.connect_status = ConnectStatus.REACHABLE
274 self.adapter_agent.update_device(device)
275
276 def do_state_up(self, event):
277 self.log.debug("do_state_up")
278
279 device = self.adapter_agent.get_device(self.device_id)
280
281 # Update phys OF device
282 device.parent_id = self.logical_device_id
283 device.oper_status = OperStatus.ACTIVE
284 self.adapter_agent.update_device(device)
285
286 def do_state_down(self, event):
287 self.log.debug("do_state_down")
288 oper_state = OperStatus.UNKNOWN
289 connect_state = ConnectStatus.UNREACHABLE
290
291 # Propagating to the children
292
293 # Children ports
294 child_devices = self.adapter_agent.get_child_devices(self.device_id)
295 for onu_device in child_devices:
296 onu_adapter_agent = \
297 registry('adapter_loader').get_agent(onu_device.adapter)
298 onu_adapter_agent.update_interface(onu_device,
299 {'oper_state': 'down'})
300 self.onu_ports_down(onu_device, oper_state)
301
302 # Children devices
303 self.adapter_agent.update_child_devices_state(
304 self.device_id, oper_status=oper_state,
305 connect_status=connect_state)
306 # Device Ports
307 device_ports = self.adapter_agent.get_ports(self.device_id,
308 Port.ETHERNET_NNI)
309 logical_ports_ids = [port.label for port in device_ports]
310 device_ports += self.adapter_agent.get_ports(self.device_id,
311 Port.PON_OLT)
312
313 for port in device_ports:
314 port.oper_status = oper_state
315 self.adapter_agent.add_port(self.device_id, port)
316
317 # Device logical port
318 for logical_port_id in logical_ports_ids:
319 logical_port = self.adapter_agent.get_logical_port(
320 self.logical_device_id, logical_port_id)
321 logical_port.ofp_port.state = OFPPS_LINK_DOWN
322 self.adapter_agent.update_logical_port(self.logical_device_id,
323 logical_port)
324
325 # Device
326 device = self.adapter_agent.get_device(self.device_id)
327 device.oper_status = oper_state
328 device.connect_status = connect_state
329
330 reactor.callLater(2, self.adapter_agent.update_device, device)
331
332 # def post_up(self, event):
333 # self.log.debug('post-up')
334 # self.flow_mgr.reseed_flows()
335
336 def post_down(self, event):
337 self.log.debug('post_down')
338 self.flow_mgr.reset_flows()
339
340 def indications_thread(self):
341 self.log.debug('starting-indications-thread')
342 self.log.debug('connecting to olt', device_id=self.device_id)
343 self.channel_ready_future.result() # blocking call
344 self.log.info('connected to olt', device_id=self.device_id)
345 self.go_state_connected()
346
347 self.indications = self.stub.EnableIndication(openolt_pb2.Empty())
348
349 while True:
350 try:
351 # get the next indication from olt
352 ind = next(self.indications)
353 except Exception as e:
354 self.log.warn('gRPC connection lost', error=e)
355 reactor.callFromThread(self.go_state_down)
356 reactor.callFromThread(self.go_state_init)
357 break
358 else:
359 self.log.debug("rx indication", indication=ind)
360
361 # indication handlers run in the main event loop
362 if ind.HasField('olt_ind'):
363 reactor.callFromThread(self.olt_indication, ind.olt_ind)
364 elif ind.HasField('intf_ind'):
365 reactor.callFromThread(self.intf_indication, ind.intf_ind)
366 elif ind.HasField('intf_oper_ind'):
367 reactor.callFromThread(self.intf_oper_indication,
368 ind.intf_oper_ind)
369 elif ind.HasField('onu_disc_ind'):
370 reactor.callFromThread(self.onu_discovery_indication,
371 ind.onu_disc_ind)
372 elif ind.HasField('onu_ind'):
373 reactor.callFromThread(self.onu_indication, ind.onu_ind)
374 elif ind.HasField('omci_ind'):
375 reactor.callFromThread(self.omci_indication, ind.omci_ind)
376 elif ind.HasField('pkt_ind'):
377 reactor.callFromThread(self.packet_indication, ind.pkt_ind)
378 elif ind.HasField('port_stats'):
379 reactor.callFromThread(
380 self.stats_mgr.port_statistics_indication,
381 ind.port_stats)
382 elif ind.HasField('flow_stats'):
383 reactor.callFromThread(
384 self.stats_mgr.flow_statistics_indication,
385 ind.flow_stats)
386 elif ind.HasField('alarm_ind'):
387 reactor.callFromThread(self.alarm_mgr.process_alarms,
388 ind.alarm_ind)
389 else:
390 self.log.warn('unknown indication type')
391
392 def olt_indication(self, olt_indication):
393 if olt_indication.oper_state == "up":
394 self.go_state_up()
395 elif olt_indication.oper_state == "down":
396 self.go_state_down()
397
398 def intf_indication(self, intf_indication):
399 self.log.debug("intf indication", intf_id=intf_indication.intf_id,
400 oper_state=intf_indication.oper_state)
401
402 if intf_indication.oper_state == "up":
403 oper_status = OperStatus.ACTIVE
404 else:
405 oper_status = OperStatus.DISCOVERED
406
407 # add_port update the port if it exists
408 self.add_port(intf_indication.intf_id, Port.PON_OLT, oper_status)
409
410 def intf_oper_indication(self, intf_oper_indication):
411 self.log.debug("Received interface oper state change indication",
412 intf_id=intf_oper_indication.intf_id,
413 type=intf_oper_indication.type,
414 oper_state=intf_oper_indication.oper_state)
415
416 if intf_oper_indication.oper_state == "up":
417 oper_state = OperStatus.ACTIVE
418 else:
419 oper_state = OperStatus.DISCOVERED
420
421 if intf_oper_indication.type == "nni":
422
423 # add_(logical_)port update the port if it exists
424 port_no, label = self.add_port(intf_oper_indication.intf_id,
425 Port.ETHERNET_NNI, oper_state)
426 self.log.debug("int_oper_indication", port_no=port_no, label=label)
427 self.add_logical_port(port_no, intf_oper_indication.intf_id,
428 oper_state)
429
430 elif intf_oper_indication.type == "pon":
431 # FIXME - handle PON oper state change
432 pass
433
434 def onu_discovery_indication(self, onu_disc_indication):
435 intf_id = onu_disc_indication.intf_id
436 serial_number = onu_disc_indication.serial_number
437
438 serial_number_str = self.stringify_serial_number(serial_number)
439
440 self.log.debug("onu discovery indication", intf_id=intf_id,
441 serial_number=serial_number_str)
442
443 # Post ONU Discover alarm 20180809_0805
444 try:
445 OnuDiscoveryAlarm(self.alarm_mgr.alarms, pon_id=intf_id,
446 serial_number=serial_number_str).raise_alarm()
447 except Exception as disc_alarm_error:
448 self.log.exception("onu-discovery-alarm-error",
449 errmsg=disc_alarm_error.message)
450 # continue for now.
451
452 onu_device = self.adapter_agent.get_child_device(
453 self.device_id,
454 serial_number=serial_number_str)
455
456 if onu_device is None:
457 try:
458 onu_id = self.resource_mgr.get_onu_id(intf_id)
459 if onu_id is None:
460 raise Exception("onu-id-unavailable")
461
462 self.add_onu_device(
463 intf_id,
464 self.platform.intf_id_to_port_no(intf_id, Port.PON_OLT),
465 onu_id, serial_number)
466 self.activate_onu(intf_id, onu_id, serial_number,
467 serial_number_str)
468 except Exception as e:
469 self.log.exception('onu-activation-failed', e=e)
470
471 else:
472 if onu_device.connect_status != ConnectStatus.REACHABLE:
473 onu_device.connect_status = ConnectStatus.REACHABLE
474 self.adapter_agent.update_device(onu_device)
475
476 onu_id = onu_device.proxy_address.onu_id
477 if onu_device.oper_status == OperStatus.DISCOVERED \
478 or onu_device.oper_status == OperStatus.ACTIVATING:
479 self.log.debug("ignore onu discovery indication, \
480 the onu has been discovered and should be \
481 activating shorlty", intf_id=intf_id,
482 onu_id=onu_id, state=onu_device.oper_status)
483 elif onu_device.oper_status == OperStatus.ACTIVE:
484 self.log.warn("onu discovery indication whereas onu is \
485 supposed to be active",
486 intf_id=intf_id, onu_id=onu_id,
487 state=onu_device.oper_status)
488 elif onu_device.oper_status == OperStatus.UNKNOWN:
489 self.log.info("onu in unknown state, recovering from olt \
490 reboot probably, activate onu", intf_id=intf_id,
491 onu_id=onu_id, serial_number=serial_number_str)
492
493 onu_device.oper_status = OperStatus.DISCOVERED
494 self.adapter_agent.update_device(onu_device)
495 try:
496 self.activate_onu(intf_id, onu_id, serial_number,
497 serial_number_str)
498 except Exception as e:
499 self.log.error('onu-activation-error',
500 serial_number=serial_number_str, error=e)
501 else:
502 self.log.warn('unexpected state', onu_id=onu_id,
503 onu_device_oper_state=onu_device.oper_status)
504
505 def onu_indication(self, onu_indication):
506 self.log.debug("onu indication", intf_id=onu_indication.intf_id,
507 onu_id=onu_indication.onu_id,
508 serial_number=onu_indication.serial_number,
509 oper_state=onu_indication.oper_state,
510 admin_state=onu_indication.admin_state)
511 try:
512 serial_number_str = self.stringify_serial_number(
513 onu_indication.serial_number)
514 except Exception as e:
515 serial_number_str = None
516
517 if serial_number_str is not None:
518 onu_device = self.adapter_agent.get_child_device(
519 self.device_id,
520 serial_number=serial_number_str)
521 else:
522 onu_device = self.adapter_agent.get_child_device(
523 self.device_id,
524 parent_port_no=self.platform.intf_id_to_port_no(
525 onu_indication.intf_id, Port.PON_OLT),
526 onu_id=onu_indication.onu_id)
527
528 if onu_device is None:
529 self.log.error('onu not found', intf_id=onu_indication.intf_id,
530 onu_id=onu_indication.onu_id)
531 return
532
533 if self.platform.intf_id_from_pon_port_no(onu_device.parent_port_no) \
534 != onu_indication.intf_id:
535 self.log.warn('ONU-is-on-a-different-intf-id-now',
536 previous_intf_id=self.platform.intf_id_from_pon_port_no(
537 onu_device.parent_port_no),
538 current_intf_id=onu_indication.intf_id)
539 # FIXME - handle intf_id mismatch (ONU move?)
540
541 if onu_device.proxy_address.onu_id != onu_indication.onu_id:
542 # FIXME - handle onu id mismatch
543 self.log.warn('ONU-id-mismatch, can happen if both voltha and '
544 'the olt rebooted',
545 expected_onu_id=onu_device.proxy_address.onu_id,
546 received_onu_id=onu_indication.onu_id)
547
548 # Admin state
549 if onu_indication.admin_state == 'down':
550 if onu_indication.oper_state != 'down':
551 self.log.error('ONU-admin-state-down-and-oper-status-not-down',
552 oper_state=onu_indication.oper_state)
553 # Forcing the oper state change code to execute
554 onu_indication.oper_state = 'down'
555
556 # Port and logical port update is taken care of by oper state block
557
558 elif onu_indication.admin_state == 'up':
559 pass
560
561 else:
562 self.log.warn('Invalid-or-not-implemented-admin-state',
563 received_admin_state=onu_indication.admin_state)
564
565 self.log.debug('admin-state-dealt-with')
566
567 onu_adapter_agent = \
568 registry('adapter_loader').get_agent(onu_device.adapter)
569 if onu_adapter_agent is None:
570 self.log.error('onu_adapter_agent-could-not-be-retrieved',
571 onu_device=onu_device)
572 return
573
574 # Operating state
575 if onu_indication.oper_state == 'down':
576
577 if onu_device.connect_status != ConnectStatus.UNREACHABLE:
578 onu_device.connect_status = ConnectStatus.UNREACHABLE
579 self.adapter_agent.update_device(onu_device)
580
581 # Move to discovered state
582 self.log.debug('onu-oper-state-is-down')
583
584 if onu_device.oper_status != OperStatus.DISCOVERED:
585 onu_device.oper_status = OperStatus.DISCOVERED
586 self.adapter_agent.update_device(onu_device)
587 # Set port oper state to Discovered
588 self.onu_ports_down(onu_device, OperStatus.DISCOVERED)
589
590 onu_adapter_agent.update_interface(onu_device,
591 {'oper_state': 'down'})
592
593 elif onu_indication.oper_state == 'up':
594
595 if onu_device.connect_status != ConnectStatus.REACHABLE:
596 onu_device.connect_status = ConnectStatus.REACHABLE
597 self.adapter_agent.update_device(onu_device)
598
599 if onu_device.oper_status != OperStatus.DISCOVERED:
600 self.log.debug("ignore onu indication",
601 intf_id=onu_indication.intf_id,
602 onu_id=onu_indication.onu_id,
603 state=onu_device.oper_status,
604 msg_oper_state=onu_indication.oper_state)
605 return
606
607 # Device was in Discovered state, setting it to active
608
609 # Prepare onu configuration
610
611 onu_adapter_agent.create_interface(onu_device, onu_indication)
612
613 else:
614 self.log.warn('Not-implemented-or-invalid-value-of-oper-state',
615 oper_state=onu_indication.oper_state)
616
617 def onu_ports_down(self, onu_device, oper_state):
618 # Set port oper state to Discovered
619 # add port will update port if it exists
620 # self.adapter_agent.add_port(
621 # self.device_id,
622 # Port(
623 # port_no=uni_no,
624 # label=uni_name,
625 # type=Port.ETHERNET_UNI,
626 # admin_state=onu_device.admin_state,
627 # oper_status=oper_state))
628 # TODO this should be downning ports in onu adatper
629
630 # Disable logical port
631 onu_ports = self.proxy.get('devices/{}/ports'.format(onu_device.id))
632 for onu_port in onu_ports:
633 self.log.debug('onu-ports-down', onu_port=onu_port)
634 onu_port_id = onu_port.label
635 try:
636 onu_logical_port = self.adapter_agent.get_logical_port(
637 logical_device_id=self.logical_device_id, port_id=onu_port_id)
638 onu_logical_port.ofp_port.state = OFPPS_LINK_DOWN
639 self.adapter_agent.update_logical_port(
640 logical_device_id=self.logical_device_id,
641 port=onu_logical_port)
642 self.log.debug('cascading-oper-state-to-port-and-logical-port')
643 except KeyError as e:
644 self.log.error('matching-onu-port-label-invalid',
645 onu_id=onu_device.id, olt_id=self.device_id,
646 onu_ports=onu_ports, onu_port_id=onu_port_id,
647 error=e)
648
649 def omci_indication(self, omci_indication):
650
651 self.log.debug("omci indication", intf_id=omci_indication.intf_id,
652 onu_id=omci_indication.onu_id)
653
654 onu_device = self.adapter_agent.get_child_device(
655 self.device_id, onu_id=omci_indication.onu_id,
656 parent_port_no=self.platform.intf_id_to_port_no(
657 omci_indication.intf_id, Port.PON_OLT), )
658
659 self.adapter_agent.receive_proxied_message(onu_device.proxy_address,
660 omci_indication.pkt)
661
662 def packet_indication(self, pkt_indication):
663
664 self.log.debug("packet indication",
665 intf_type=pkt_indication.intf_type,
666 intf_id=pkt_indication.intf_id,
667 port_no=pkt_indication.port_no,
668 cookie=pkt_indication.cookie,
669 gemport_id=pkt_indication.gemport_id,
670 flow_id=pkt_indication.flow_id)
671
672 if pkt_indication.intf_type == "pon":
673 if pkt_indication.port_no:
674 logical_port_num = pkt_indication.port_no
675 else: # TODO Remove this else block after openolt device has been fully rolled out with cookie protobuf change
676 try:
677 onu_id_uni_id = self.resource_mgr.get_onu_uni_from_ponport_gemport(pkt_indication.intf_id,
678 pkt_indication.gemport_id)
679 onu_id = int(onu_id_uni_id[0])
680 uni_id = int(onu_id_uni_id[1])
681 self.log.debug("packet indication-kv", onu_id=onu_id, uni_id=uni_id)
682 if onu_id is None:
683 raise Exception("onu-id-none")
684 if uni_id is None:
685 raise Exception("uni-id-none")
686 logical_port_num = self.platform.mk_uni_port_num(pkt_indication.intf_id, onu_id, uni_id)
687 except Exception as e:
688 self.log.error("no-onu-reference-for-gem",
689 gemport_id=pkt_indication.gemport_id, e=e)
690 return
691
692
693 elif pkt_indication.intf_type == "nni":
694 logical_port_num = self.platform.intf_id_to_port_no(
695 pkt_indication.intf_id,
696 Port.ETHERNET_NNI)
697
698 pkt = Ether(pkt_indication.pkt)
699
700 self.log.debug("packet indication",
701 logical_device_id=self.logical_device_id,
702 logical_port_no=logical_port_num)
703
704 self.adapter_agent.send_packet_in(
705 logical_device_id=self.logical_device_id,
706 logical_port_no=logical_port_num,
707 packet=str(pkt))
708
709 def packet_out(self, egress_port, msg):
710 pkt = Ether(msg)
711 self.log.debug('packet out', egress_port=egress_port,
712 device_id=self.device_id,
713 logical_device_id=self.logical_device_id,
714 packet=str(pkt).encode("HEX"))
715
716 # Find port type
717 egress_port_type = self.platform.intf_id_to_port_type_name(egress_port)
718 if egress_port_type == Port.ETHERNET_UNI:
719
720 if pkt.haslayer(Dot1Q):
721 outer_shim = pkt.getlayer(Dot1Q)
722 if isinstance(outer_shim.payload, Dot1Q):
723 # If double tag, remove the outer tag
724 payload = (
725 Ether(src=pkt.src, dst=pkt.dst, type=outer_shim.type) /
726 outer_shim.payload
727 )
728 else:
729 payload = pkt
730 else:
731 payload = pkt
732
733 send_pkt = binascii.unhexlify(str(payload).encode("HEX"))
734
735 self.log.debug(
736 'sending-packet-to-ONU', egress_port=egress_port,
737 intf_id=self.platform.intf_id_from_uni_port_num(egress_port),
738 onu_id=self.platform.onu_id_from_port_num(egress_port),
739 uni_id=self.platform.uni_id_from_port_num(egress_port),
740 port_no=egress_port,
741 packet=str(payload).encode("HEX"))
742
743 onu_pkt = openolt_pb2.OnuPacket(
744 intf_id=self.platform.intf_id_from_uni_port_num(egress_port),
745 onu_id=self.platform.onu_id_from_port_num(egress_port),
746 port_no=egress_port,
747 pkt=send_pkt)
748
749 self.stub.OnuPacketOut(onu_pkt)
750
751 elif egress_port_type == Port.ETHERNET_NNI:
752 self.log.debug('sending-packet-to-uplink', egress_port=egress_port,
753 packet=str(pkt).encode("HEX"))
754
755 send_pkt = binascii.unhexlify(str(pkt).encode("HEX"))
756
757 uplink_pkt = openolt_pb2.UplinkPacket(
758 intf_id=self.platform.intf_id_from_nni_port_num(egress_port),
759 pkt=send_pkt)
760
761 self.stub.UplinkPacketOut(uplink_pkt)
762
763 else:
764 self.log.warn('Packet-out-to-this-interface-type-not-implemented',
765 egress_port=egress_port,
766 port_type=egress_port_type)
767
768 def send_proxied_message(self, proxy_address, msg):
769 onu_device = self.adapter_agent.get_child_device(
770 self.device_id, onu_id=proxy_address.onu_id,
771 parent_port_no=self.platform.intf_id_to_port_no(
772 proxy_address.channel_id, Port.PON_OLT)
773 )
774 if onu_device.connect_status != ConnectStatus.REACHABLE:
775 self.log.debug('ONU is not reachable, cannot send OMCI',
776 serial_number=onu_device.serial_number,
777 intf_id=onu_device.proxy_address.channel_id,
778 onu_id=onu_device.proxy_address.onu_id)
779 return
780 omci = openolt_pb2.OmciMsg(intf_id=proxy_address.channel_id,
781 onu_id=proxy_address.onu_id, pkt=str(msg))
782 self.stub.OmciMsgOut(omci)
783
784 def add_onu_device(self, intf_id, port_no, onu_id, serial_number):
785 self.log.info("Adding ONU", port_no=port_no, onu_id=onu_id,
786 serial_number=serial_number)
787
788 # NOTE - channel_id of onu is set to intf_id
789 proxy_address = Device.ProxyAddress(device_id=self.device_id,
790 channel_id=intf_id, onu_id=onu_id,
791 onu_session_id=onu_id)
792
793 self.log.debug("Adding ONU", proxy_address=proxy_address)
794
795 serial_number_str = self.stringify_serial_number(serial_number)
796
797 self.adapter_agent.add_onu_device(
798 parent_device_id=self.device_id, parent_port_no=port_no,
799 vendor_id=serial_number.vendor_id, proxy_address=proxy_address,
800 root=True, serial_number=serial_number_str,
801 admin_state=AdminState.ENABLED#, **{'vlan':4091} # magic still maps to brcm_openomci_onu.pon_port.BRDCM_DEFAULT_VLAN
802 )
803
804 def port_name(self, port_no, port_type, intf_id=None, serial_number=None):
805 if port_type is Port.ETHERNET_NNI:
806 return "nni-" + str(port_no)
807 elif port_type is Port.PON_OLT:
808 return "pon" + str(intf_id)
809 elif port_type is Port.ETHERNET_UNI:
810 assert False, 'local UNI management not supported'
811
812 def add_logical_port(self, port_no, intf_id, oper_state):
813 self.log.info('adding-logical-port', port_no=port_no)
814
815 label = self.port_name(port_no, Port.ETHERNET_NNI)
816
817 cap = OFPPF_1GB_FD | OFPPF_FIBER
818 curr_speed = OFPPF_1GB_FD
819 max_speed = OFPPF_1GB_FD
820
821 if oper_state == OperStatus.ACTIVE:
822 of_oper_state = OFPPS_LIVE
823 else:
824 of_oper_state = OFPPS_LINK_DOWN
825
826 ofp = ofp_port(
827 port_no=port_no,
828 hw_addr=mac_str_to_tuple(self._get_mac_form_port_no(port_no)),
829 name=label, config=0, state=of_oper_state, curr=cap,
830 advertised=cap, peer=cap, curr_speed=curr_speed,
831 max_speed=max_speed)
832
833 ofp_stats = ofp_port_stats(port_no=port_no)
834
835 logical_port = LogicalPort(
836 id=label, ofp_port=ofp, device_id=self.device_id,
837 device_port_no=port_no, root_port=True,
838 ofp_port_stats=ofp_stats)
839
840 self.adapter_agent.add_logical_port(self.logical_device_id,
841 logical_port)
842
843 def _get_mac_form_port_no(self, port_no):
844 mac = ''
845 for i in range(4):
846 mac = ':%02x' % ((port_no >> (i * 8)) & 0xff) + mac
847 return '00:00' + mac
848
849 def add_port(self, intf_id, port_type, oper_status):
850 port_no = self.platform.intf_id_to_port_no(intf_id, port_type)
851
852 label = self.port_name(port_no, port_type, intf_id)
853
854 self.log.debug('adding-port', port_no=port_no, label=label,
855 port_type=port_type)
856
857 port = Port(port_no=port_no, label=label, type=port_type,
858 admin_state=AdminState.ENABLED, oper_status=oper_status)
859
860 self.adapter_agent.add_port(self.device_id, port)
861
862 return port_no, label
863
864 def delete_logical_port(self, child_device):
865 logical_ports = self.proxy.get('/logical_devices/{}/ports'.format(
866 self.logical_device_id))
867 for logical_port in logical_ports:
868 if logical_port.device_id == child_device.id:
869 self.log.debug('delete-logical-port',
870 onu_device_id=child_device.id,
871 logical_port=logical_port)
872 self.flow_mgr.clear_flows_and_scheduler_for_logical_port(
873 child_device, logical_port)
874 self.adapter_agent.delete_logical_port(
875 self.logical_device_id, logical_port)
876 return
877
878 def delete_port(self, child_serial_number):
879 ports = self.proxy.get('/devices/{}/ports'.format(
880 self.device_id))
881 for port in ports:
882 if port.label == child_serial_number:
883 self.log.debug('delete-port',
884 onu_serial_number=child_serial_number,
885 port=port)
886 self.adapter_agent.delete_port(self.device_id, port)
887 return
888
889 def update_flow_table(self, flows):
890 self.log.debug('No updates here now, all is done in logical flows '
891 'update')
892
893 def update_logical_flows(self, flows_to_add, flows_to_remove,
894 device_rules_map):
895 if not self.is_state_up():
896 self.log.info('The OLT is not up, we cannot update flows',
897 flows_to_add=[f.id for f in flows_to_add],
898 flows_to_remove=[f.id for f in flows_to_remove])
899 return
900
901 try:
902 self.flow_mgr.update_children_flows(device_rules_map)
903 except Exception as e:
904 self.log.error('Error updating children flows', error=e)
905
906 self.log.debug('logical flows update', flows_to_add=flows_to_add,
907 flows_to_remove=flows_to_remove)
908
909 for flow in flows_to_add:
910
911 try:
912 self.flow_mgr.add_flow(flow)
913 except Exception as e:
914 self.log.error('failed to add flow', flow=flow, e=e)
915
916 for flow in flows_to_remove:
917
918 try:
919 self.flow_mgr.remove_flow(flow)
920 except Exception as e:
921 self.log.error('failed to remove flow', flow=flow, e=e)
922
923 self.flow_mgr.repush_all_different_flows()
924
925 # There has to be a better way to do this
926 def ip_hex(self, ip):
927 octets = ip.split(".")
928 hex_ip = []
929 for octet in octets:
930 octet_hex = hex(int(octet))
931 octet_hex = octet_hex.split('0x')[1]
932 octet_hex = octet_hex.rjust(2, '0')
933 hex_ip.append(octet_hex)
934 return ":".join(hex_ip)
935
936 def stringify_vendor_specific(self, vendor_specific):
937 return ''.join(str(i) for i in [
938 hex(ord(vendor_specific[0]) >> 4 & 0x0f)[2:],
939 hex(ord(vendor_specific[0]) & 0x0f)[2:],
940 hex(ord(vendor_specific[1]) >> 4 & 0x0f)[2:],
941 hex(ord(vendor_specific[1]) & 0x0f)[2:],
942 hex(ord(vendor_specific[2]) >> 4 & 0x0f)[2:],
943 hex(ord(vendor_specific[2]) & 0x0f)[2:],
944 hex(ord(vendor_specific[3]) >> 4 & 0x0f)[2:],
945 hex(ord(vendor_specific[3]) & 0x0f)[2:]])
946
947 def stringify_serial_number(self, serial_number):
948 return ''.join([serial_number.vendor_id,
949 self.stringify_vendor_specific(
950 serial_number.vendor_specific)])
951
952 def destringify_serial_number(self, serial_number_str):
953 serial_number = openolt_pb2.SerialNumber(
954 vendor_id=serial_number_str[:4].encode('utf-8'),
955 vendor_specific=binascii.unhexlify(serial_number_str[4:]))
956 return serial_number
957
958 def disable(self):
959 self.log.debug('sending-deactivate-olt-message',
960 device_id=self.device_id)
961
962 try:
963 # Send grpc call
964 self.stub.DisableOlt(openolt_pb2.Empty())
965 # The resulting indication will bring the OLT down
966 # self.go_state_down()
967 self.log.info('openolt device disabled')
968 except Exception as e:
969 self.log.error('Failure to disable openolt device', error=e)
970
971 def delete(self):
972 self.log.info('deleting-olt', device_id=self.device_id,
973 logical_device_id=self.logical_device_id)
974
975 # Clears up the data from the resource manager KV store
976 # for the device
977 del self.resource_mgr
978
979 try:
980 # Rebooting to reset the state
981 self.reboot()
982 # Removing logical device
983 ld = self.adapter_agent.get_logical_device(self.logical_device_id)
984 self.adapter_agent.delete_logical_device(ld)
985 except Exception as e:
986 self.log.error('Failure to delete openolt device', error=e)
987 raise e
988 else:
989 self.log.info('successfully-deleted-olt', device_id=self.device_id)
990
991 def reenable(self):
992 self.log.debug('reenabling-olt', device_id=self.device_id)
993
994 try:
995 self.stub.ReenableOlt(openolt_pb2.Empty())
996
997 self.log.info('enabling-all-ports', device_id=self.device_id)
998 self.adapter_agent.enable_all_ports(self.device_id)
999 except Exception as e:
1000 self.log.error('Failure to reenable openolt device', error=e)
1001 else:
1002 self.log.info('openolt device reenabled')
1003
1004 def activate_onu(self, intf_id, onu_id, serial_number,
1005 serial_number_str):
1006 pir = self.bw_mgr.pir(serial_number_str)
1007 self.log.debug("activating-onu", intf_id=intf_id, onu_id=onu_id,
1008 serial_number_str=serial_number_str,
1009 serial_number=serial_number, pir=pir)
1010 onu = openolt_pb2.Onu(intf_id=intf_id, onu_id=onu_id,
1011 serial_number=serial_number, pir=pir)
1012 self.stub.ActivateOnu(onu)
1013 self.log.info('onu-activated', serial_number=serial_number_str)
1014
1015 def delete_child_device(self, child_device):
1016 self.log.debug('sending-deactivate-onu',
1017 olt_device_id=self.device_id,
1018 onu_device=child_device,
1019 onu_serial_number=child_device.serial_number)
1020 try:
1021 self.adapter_agent.delete_child_device(self.device_id,
1022 child_device.id,
1023 child_device)
1024 except Exception as e:
1025 self.log.error('adapter_agent error', error=e)
1026 try:
1027 self.delete_logical_port(child_device)
1028 except Exception as e:
1029 self.log.error('logical_port delete error', error=e)
1030 try:
1031 self.delete_port(child_device.serial_number)
1032 except Exception as e:
1033 self.log.error('port delete error', error=e)
1034 serial_number = self.destringify_serial_number(
1035 child_device.serial_number)
1036 # TODO FIXME - For each uni.
1037 # TODO FIXME - Flows are not deleted
1038 uni_id = 0 # FIXME
1039 self.flow_mgr.delete_tech_profile_instance(
1040 child_device.proxy_address.channel_id,
1041 child_device.proxy_address.onu_id,
1042 uni_id
1043 )
1044 pon_intf_id_onu_id = (child_device.proxy_address.channel_id,
1045 child_device.proxy_address.onu_id,
1046 uni_id)
1047 # Free any PON resources that were reserved for the ONU
1048 self.resource_mgr.free_pon_resources_for_onu(pon_intf_id_onu_id)
1049
1050 onu = openolt_pb2.Onu(intf_id=child_device.proxy_address.channel_id,
1051 onu_id=child_device.proxy_address.onu_id,
1052 serial_number=serial_number)
1053 self.stub.DeleteOnu(onu)
1054
1055 def reboot(self):
1056 self.log.debug('rebooting openolt device', device_id=self.device_id)
1057 try:
1058 self.stub.Reboot(openolt_pb2.Empty())
1059 except Exception as e:
1060 self.log.error('something went wrong with the reboot', error=e)
1061 else:
1062 self.log.info('device rebooted')
1063
1064 def trigger_statistics_collection(self):
1065 try:
1066 self.stub.CollectStatistics(openolt_pb2.Empty())
1067 except Exception as e:
1068 self.log.error('Error while triggering statistics collection',
1069 error=e)
1070 else:
1071 self.log.info('statistics requested')
1072
1073 def simulate_alarm(self, alarm):
1074 self.alarm_mgr.simulate_alarm(alarm)