[ 3050 ] Fixing cut and paste error in voltha.proto.
This commit is an amendment to the previous commit with the
following changes:
1) Simplify the gRPC API to replace the activate_olt and re_enable apis
with only one enable api. Note the adapter interface remains
unchanged to keep the flexibility of operations between different
device adapters.
2) Small changes following the initial code review
This commit consits of the following updates:
1) Support for the following config changes:
1a) Reboot of an OLT/ONU
1b) Deletion of an OLT/ONU
1c) Disabling of an OLT/ONU
1d) Re-enabling of an OLT/ONU
2) Corresponding APIs are added to the voltha.proto file
3) The adapter interface has been augmented with the above
APIs
4) The ponsim_olt and ponsim_onu adapters have been updated to
implement the above APIs
TODOs:
1) Existing flows on the ponsim devices have not been updated
to reflect the above changes.
2) ponsim needs to be augmented to support the above APIs
3) integration tests
The above will be addressed in a separate commit
Change-Id: Ia7af7d773517df269cdc2b0c629d5ef8f1fb6e3a
diff --git a/cli/main.py b/cli/main.py
index f5ba38b..3d9de58 100755
--- a/cli/main.py
+++ b/cli/main.py
@@ -278,30 +278,74 @@
self.poutput('success (device id = {})'.format(device.id))
self.default_device_id = device.id
- def do_activate_olt(self, line):
+ def do_enable(self, line):
"""
- Activate an OLT. If the <id> is not provided, it will be on the last
- pre-provisioned OLT.
+ Enable a device. If the <id> is not provided, it will be on the last
+ pre-provisioned device.
"""
device_id = line or self.default_device_id
self.poutput('activating {}'.format(device_id))
stub = voltha_pb2.VolthaLocalServiceStub(self.get_channel())
- stub.ActivateDevice(voltha_pb2.ID(id=device_id))
+ stub.EnableDevice(voltha_pb2.ID(id=device_id))
- # try to acquire logical device id
while True:
device = stub.GetDevice(voltha_pb2.ID(id=device_id))
+ # If this is an OLT then acquire logical device id
if device.oper_status == voltha_pb2.OperStatus.ACTIVE:
- assert device.parent_id
- self.default_logical_device_id = device.parent_id
+ if device.type.endswith('_olt'):
+ assert device.parent_id
+ self.default_logical_device_id = device.parent_id
+ self.poutput('success (logical device id = {})'.format(
+ self.default_logical_device_id))
+ else:
+ self.poutput('success (device id = {})'.format(device.id))
break
- self.poutput('waiting for device to be activated...')
+ self.poutput('waiting for device to be enabled...')
sleep(.5)
- self.poutput('success (logical device id = {})'.format(
- self.default_logical_device_id))
complete_activate_olt = complete_device
+ def do_reboot(self, line):
+ """
+ Rebooting a device. ID of the device needs to be provided
+ """
+ device_id = line or self.default_device_id
+ self.poutput('rebooting {}'.format(device_id))
+ stub = voltha_pb2.VolthaLocalServiceStub(self.get_channel())
+ stub.RebootDevice(voltha_pb2.ID(id=device_id))
+ self.poutput('rebooted {}'.format(device_id))
+
+ def do_delete(self, line):
+ """
+ Deleting a device. ID of the device needs to be provided
+ """
+ device_id = line or self.default_device_id
+ self.poutput('deleting {}'.format(device_id))
+ stub = voltha_pb2.VolthaLocalServiceStub(self.get_channel())
+ stub.DeleteDevice(voltha_pb2.ID(id=device_id))
+ self.poutput('deleted {}'.format(device_id))
+
+ def do_disable(self, line):
+ """
+ Disable a device. ID of the device needs to be provided
+ """
+ device_id = line
+ self.poutput('disabling {}'.format(device_id))
+ stub = voltha_pb2.VolthaLocalServiceStub(self.get_channel())
+ stub.DisableDevice(voltha_pb2.ID(id=device_id))
+
+ # Do device query and verify that the device admin status is
+ # DISABLED and Operational Status is unknown
+ device = stub.GetDevice(voltha_pb2.ID(id=device_id))
+ if device.oper_status == voltha_pb2.OperStatus.UNKNOWN and \
+ device.admin_state == voltha_pb2.AdminState.DISABLED:
+ self.poutput('disabled successfully {}'.format(device_id))
+ else:
+ self.poutput('disabling failed {}. Admin State:{} '
+ 'Operation State: {}'.format(device_id,
+ device.admin_state,
+ device.oper_status))
+
def do_test(self, line):
"""Enter test mode, which makes a bunch on new commands available"""
sub = TestCli(self.history, self.get_channel, self.voltha_grpc,
diff --git a/voltha/adapters/broadcom_onu/broadcom_onu.py b/voltha/adapters/broadcom_onu/broadcom_onu.py
index 212c0cf..1854d32 100644
--- a/voltha/adapters/broadcom_onu/broadcom_onu.py
+++ b/voltha/adapters/broadcom_onu/broadcom_onu.py
@@ -97,7 +97,19 @@
def abandon_device(self, device):
raise NotImplementedError()
- def deactivate_device(self, device):
+ def disable_device(self, device):
+ raise NotImplementedError()
+
+ def reenable_device(self, device):
+ raise NotImplementedError()
+
+ def reboot_device(self, device):
+ raise NotImplementedError()
+
+ def delete_device(self, device):
+ raise NotImplementedError()
+
+ def get_device_details(self, device):
raise NotImplementedError()
def update_flows_bulk(self, device, flows, groups):
diff --git a/voltha/adapters/dpoe_onu/dpoe_onu.py b/voltha/adapters/dpoe_onu/dpoe_onu.py
index e1877e9..89973aa 100644
--- a/voltha/adapters/dpoe_onu/dpoe_onu.py
+++ b/voltha/adapters/dpoe_onu/dpoe_onu.py
@@ -197,7 +197,19 @@
def abandon_device(self, device):
raise NotImplementedError(0
)
- def deactivate_device(self, device):
+ def disable_device(self, device):
+ raise NotImplementedError()
+
+ def reenable_device(self, device):
+ raise NotImplementedError()
+
+ def reboot_device(self, device):
+ raise NotImplementedError()
+
+ def delete_device(self, device):
+ raise NotImplementedError()
+
+ def get_device_details(self, device):
raise NotImplementedError()
def update_flows_bulk(self, device, flows, groups):
diff --git a/voltha/adapters/interface.py b/voltha/adapters/interface.py
index 4111735..c7016d1 100644
--- a/voltha/adapters/interface.py
+++ b/voltha/adapters/interface.py
@@ -89,11 +89,45 @@
:return: (Deferred) Shall be fired to acknowledge abandonment.
"""
- def deactivate_device(device):
+ def disable_device(device):
"""
- Called if the device is to be deactivate based on a NBI call.
+ This is called when a previously enabled device needs to be disabled
+ based on a NBI call.
:param device: A Voltha.Device object.
- :return: (Deferred) Shall be fired to acknowledge deactivation.
+ :return: (Deferred) Shall be fired to acknowledge disabling the device.
+ """
+
+ def reenable_device(device):
+ """
+ This is called when a previously disabled device needs to be enabled
+ based on a NBI call.
+ :param device: A Voltha.Device object.
+ :return: (Deferred) Shall be fired to acknowledge re-enabling the
+ device.
+ """
+
+ def reboot_device(device):
+ """
+ This is called to reboot a device based on a NBI call. The admin
+ state of the device will not change after the reboot
+ :param device: A Voltha.Device object.
+ :return: (Deferred) Shall be fired to acknowledge the reboot.
+ """
+
+ def delete_device(device):
+ """
+ This is called to delete a device from the PON based on a NBI call.
+ If the device is an OLT then the whole PON will be deleted.
+ :param device: A Voltha.Device object.
+ :return: (Deferred) Shall be fired to acknowledge the deletion.
+ """
+
+ def get_device_details(device):
+ """
+ This is called to get additional device details based on a NBI call.
+ :param device: A Voltha.Device object.
+ :return: (Deferred) Shall be fired to acknowledge the retrieval of
+ additional details.
"""
def update_flows_bulk(device, flows, groups):
@@ -151,8 +185,8 @@
:return: None
"""
- # TODO work in progress
- # ...
+ # TODO work in progress
+ # ...
class IAdapterAgent(Interface):
diff --git a/voltha/adapters/maple_olt/maple_olt.py b/voltha/adapters/maple_olt/maple_olt.py
index e80cc79..6d5183c 100644
--- a/voltha/adapters/maple_olt/maple_olt.py
+++ b/voltha/adapters/maple_olt/maple_olt.py
@@ -190,7 +190,19 @@
def abandon_device(self, device):
raise NotImplementedError()
- def deactivate_device(self, device):
+ def disable_device(self, device):
+ raise NotImplementedError()
+
+ def reenable_device(self, device):
+ raise NotImplementedError()
+
+ def reboot_device(self, device):
+ raise NotImplementedError()
+
+ def delete_device(self, device):
+ raise NotImplementedError()
+
+ def get_device_details(self, device):
raise NotImplementedError()
def update_flows_bulk(self, device, flows, groups):
diff --git a/voltha/adapters/microsemi_olt/microsemi_olt.py b/voltha/adapters/microsemi_olt/microsemi_olt.py
index 86026f4..152f8e3 100644
--- a/voltha/adapters/microsemi_olt/microsemi_olt.py
+++ b/voltha/adapters/microsemi_olt/microsemi_olt.py
@@ -108,8 +108,20 @@
def abandon_device(self, device):
self._abandon(device.mac_address)
- def deactivate_device(self, device):
- self._abandon(device.mac_address)
+ def disable_device(self, device):
+ raise NotImplementedError()
+
+ def reenable_device(self, device):
+ raise NotImplementedError()
+
+ def reboot_device(self, device):
+ raise NotImplementedError()
+
+ def delete_device(self, device):
+ raise NotImplementedError()
+
+ def get_device_details(self, device):
+ raise NotImplementedError()
def update_flows_bulk(self, device, flows, groups):
log.debug('bulk-flow-update', device_id=device.id,
diff --git a/voltha/adapters/ponsim_olt/ponsim_olt.py b/voltha/adapters/ponsim_olt/ponsim_olt.py
index dcc55a7..08da855 100644
--- a/voltha/adapters/ponsim_olt/ponsim_olt.py
+++ b/voltha/adapters/ponsim_olt/ponsim_olt.py
@@ -23,9 +23,11 @@
import structlog
from scapy.layers.l2 import Ether, Dot1Q
from twisted.internet import reactor
+from twisted.internet.defer import inlineCallbacks
from zope.interface import implementer
from common.frameio.frameio import BpfProgramFilter, hexify
+from common.utils.asleep import asleep
from voltha.adapters.interface import IAdapterInterface
from voltha.core.logical_device_agent import mac_str_to_tuple
from voltha.protos import third_party
@@ -85,6 +87,11 @@
log.info('started')
def stop(self):
+ """
+ This method is called when this device instance is no longer
+ required, which means there is a request to remove this device.
+ :return:
+ """
log.debug('stopping')
log.info('stopped')
@@ -108,7 +115,27 @@
def abandon_device(self, device):
raise NotImplementedError()
- def deactivate_device(self, device):
+ def disable_device(self, device):
+ log.info('disable-device', device_id=device.id)
+ reactor.callLater(0, self.devices_handlers[device.id].disable)
+ return device
+
+ def reenable_device(self, device):
+ log.info('reenable-device', device_id=device.id)
+ reactor.callLater(0, self.devices_handlers[device.id].reenable)
+ return device
+
+ def reboot_device(self, device):
+ log.info('reboot-device', device_id=device.id)
+ reactor.callLater(0, self.devices_handlers[device.id].reboot)
+ return device
+
+ def delete_device(self, device):
+ log.info('delete-device', device_id=device.id)
+ reactor.callLater(0, self.devices_handlers[device.id].delete)
+ return device
+
+ def get_device_details(self, device):
raise NotImplementedError()
def update_flows_bulk(self, device, flows, groups):
@@ -154,6 +181,7 @@
self.channel = None
self.io_port = None
self.logical_device_id = None
+ self.nni_port = None
self.interface = registry('main').get_args().interface
def __del__(self):
@@ -193,6 +221,7 @@
admin_state=AdminState.ENABLED,
oper_status=OperStatus.ACTIVE
)
+ self.nni_port = nni_port
self.adapter_agent.add_port(device.id, nni_port)
self.adapter_agent.add_port(device.id, Port(
port_no=1,
@@ -292,6 +321,8 @@
self.adapter_agent.send_packet_in(
packet=str(popped_frame), **kw)
+
+
def update_flow_table(self, flows):
stub = ponsim_pb2.PonSimStub(self.get_channel())
self.log.info('pushing-olt-flow-table')
@@ -320,3 +351,151 @@
pkt.payload
)
self.io_port.send(str(out_pkt))
+
+ @inlineCallbacks
+ def reboot(self):
+ self.log.info('rebooting', device_id=self.device_id)
+
+ # Update the operational status to ACTIVATING and connect status to
+ # UNREACHABLE
+ device = self.adapter_agent.get_device(self.device_id)
+ previous_oper_status = device.oper_status
+ previous_conn_status = device.connect_status
+ device.oper_status = OperStatus.ACTIVATING
+ device.connect_status = ConnectStatus.UNREACHABLE
+ self.adapter_agent.update_device(device)
+
+ # Sleep 10 secs, simulating a reboot
+ #TODO: send alert and clear alert after the reboot
+ yield asleep(10)
+
+ # Change the operational status back to its previous state. With a
+ # real OLT the operational state should be the state the device is
+ # after a reboot.
+ # Get the latest device reference
+ device = self.adapter_agent.get_device(self.device_id)
+ device.oper_status = previous_oper_status
+ device.connect_status = previous_conn_status
+ self.adapter_agent.update_device(device)
+ self.log.info('rebooted', device_id=self.device_id)
+
+ def disable(self):
+ self.log.info('disabling', device_id=self.device_id)
+
+ # Get the latest device reference
+ device = self.adapter_agent.get_device(self.device_id)
+
+ # Update the operational status to UNKNOWN
+ device.oper_status = OperStatus.UNKNOWN
+ device.connect_status = ConnectStatus.UNREACHABLE
+ self.adapter_agent.update_device(device)
+
+ # Remove the logical device
+ logical_device = self.adapter_agent.get_logical_device(
+ self.logical_device_id)
+ self.adapter_agent.delete_logical_device(logical_device)
+
+ # Disable all child devices first
+ self.adapter_agent.disable_all_child_devices(self.device_id)
+
+ # # Remove all child devices
+ # self.adapter_agent.remove_all_child_devices(self.device_id)
+
+ # Remove the peer references from this device
+ self.adapter_agent.delete_all_peer_references(self.device_id)
+
+ # close the frameio port
+ registry('frameio').close_port(self.io_port)
+
+ # TODO:
+ # 1) Remove all flows from the device
+ # 2) Remove the device from ponsim
+
+ self.log.info('disabled', device_id=device.id)
+
+
+ def reenable(self):
+ self.log.info('re-enabling', device_id=self.device_id)
+
+ # Get the latest device reference
+ device = self.adapter_agent.get_device(self.device_id)
+
+ # Update the connect status to REACHABLE
+ device.connect_status = ConnectStatus.REACHABLE
+ self.adapter_agent.update_device(device)
+
+ ld = LogicalDevice(
+ # not setting id and datapth_id will let the adapter agent pick id
+ desc=ofp_desc(
+ mfr_desc='cord porject',
+ hw_desc='simulated pon',
+ sw_desc='simulated pon',
+ serial_num=uuid4().hex,
+ dp_desc='n/a'
+ ),
+ switch_features=ofp_switch_features(
+ n_buffers=256, # TODO fake for now
+ n_tables=2, # TODO ditto
+ capabilities=( # TODO and ditto
+ OFPC_FLOW_STATS
+ | OFPC_TABLE_STATS
+ | OFPC_PORT_STATS
+ | OFPC_GROUP_STATS
+ )
+ ),
+ root_device_id=device.id
+ )
+ ld_initialized = self.adapter_agent.create_logical_device(ld)
+ cap = OFPPF_1GB_FD | OFPPF_FIBER
+ self.adapter_agent.add_logical_port(ld_initialized.id, LogicalPort(
+ id='nni',
+ ofp_port=ofp_port(
+ # port_no=info.nni_port,
+ # hw_addr=mac_str_to_tuple('00:00:00:00:00:%02x' % info.nni_port),
+ port_no=self.nni_port.port_no,
+ hw_addr=mac_str_to_tuple(
+ '00:00:00:00:00:%02x' % self.nni_port.port_no),
+ name='nni',
+ config=0,
+ state=OFPPS_LIVE,
+ curr=cap,
+ advertised=cap,
+ peer=cap,
+ curr_speed=OFPPF_1GB_FD,
+ max_speed=OFPPF_1GB_FD
+ ),
+ device_id=device.id,
+ device_port_no=self.nni_port.port_no,
+ root_port=True
+ ))
+
+ device = self.adapter_agent.get_device(device.id)
+ device.parent_id = ld_initialized.id
+ device.oper_status = OperStatus.ACTIVE
+ self.adapter_agent.update_device(device)
+ self.logical_device_id = ld_initialized.id
+
+ # Reenable all child devices
+ self.adapter_agent.reenable_all_child_devices(device.id)
+
+
+ # finally, open the frameio port to receive in-band packet_in messages
+ self.log.info('registering-frameio')
+ self.io_port = registry('frameio').open_port(
+ self.interface, self.rcv_io, is_inband_frame)
+ self.log.info('registered-frameio')
+
+ self.log.info('re-enabled', device_id=device.id)
+
+
+ def delete(self):
+ self.log.info('deleting', device_id=self.device_id)
+
+ # Remove all child devices
+ self.adapter_agent.delete_all_child_devices(self.device_id)
+
+ # TODO:
+ # 1) Remove all flows from the device
+ # 2) Remove the device from ponsim
+
+ self.log.info('deleted', device_id=self.device_id)
diff --git a/voltha/adapters/ponsim_onu/ponsim_onu.py b/voltha/adapters/ponsim_onu/ponsim_onu.py
index 8740b76..36d0013 100644
--- a/voltha/adapters/ponsim_onu/ponsim_onu.py
+++ b/voltha/adapters/ponsim_onu/ponsim_onu.py
@@ -22,6 +22,7 @@
from twisted.internet import reactor
from twisted.internet.defer import DeferredQueue, inlineCallbacks
from zope.interface import implementer
+from common.utils.asleep import asleep
from voltha.adapters.interface import IAdapterInterface
from voltha.core.logical_device_agent import mac_str_to_tuple
@@ -33,7 +34,8 @@
from voltha.protos.device_pb2 import DeviceType, DeviceTypes, Port
from voltha.protos.health_pb2 import HealthStatus
from voltha.protos.logical_device_pb2 import LogicalPort
-from voltha.protos.openflow_13_pb2 import OFPPS_LIVE, OFPPF_FIBER, OFPPF_1GB_FD
+from voltha.protos.openflow_13_pb2 import OFPPS_LIVE, OFPPF_FIBER, \
+ OFPPF_1GB_FD, OFPPC_NO_RECV
from voltha.protos.openflow_13_pb2 import ofp_port
from voltha.protos.ponsim_pb2 import FlowTable
@@ -93,7 +95,27 @@
def abandon_device(self, device):
raise NotImplementedError()
- def deactivate_device(self, device):
+ def disable_device(self, device):
+ log.info('disable-device', device_id=device.id)
+ reactor.callLater(0, self.devices_handlers[device.id].disable)
+ return device
+
+ def reenable_device(self, device):
+ log.info('reenable-device', device_id=device.id)
+ reactor.callLater(0, self.devices_handlers[device.id].reenable)
+ return device
+
+ def reboot_device(self, device):
+ log.info('rebooting', device_id=device.id)
+ reactor.callLater(0, self.devices_handlers[device.id].reboot)
+ return device
+
+ def delete_device(self, device):
+ log.info('delete-device', device_id=device.id)
+ reactor.callLater(0, self.devices_handlers[device.id].delete)
+ return device
+
+ def get_device_details(self, device):
raise NotImplementedError()
def update_flows_bulk(self, device, flows, groups):
@@ -129,6 +151,10 @@
self.log = structlog.get_logger(device_id=device_id)
self.incoming_messages = DeferredQueue()
self.proxy_address = None
+ # reference of uni_port is required when re-enabling the device if
+ # it was disabled previously
+ self.uni_port = None
+ self.pon_port = None
def receive_message(self, msg):
self.incoming_messages.put(msg)
@@ -153,15 +179,14 @@
self.adapter_agent.update_device(device)
# register physical ports
- uni_port = Port(
+ self.uni_port = Port(
port_no=2,
label='UNI facing Ethernet port',
type=Port.ETHERNET_UNI,
admin_state=AdminState.ENABLED,
oper_status=OperStatus.ACTIVE
)
- self.adapter_agent.add_port(device.id, uni_port)
- self.adapter_agent.add_port(device.id, Port(
+ self.pon_port = Port(
port_no=1,
label='PON port',
type=Port.PON_ONU,
@@ -173,7 +198,9 @@
port_no=device.parent_port_no
)
]
- ))
+ )
+ self.adapter_agent.add_port(device.id, self.uni_port)
+ self.adapter_agent.add_port(device.id, self.pon_port)
# add uni port to logical device
parent_device = self.adapter_agent.get_device(device.parent_id)
@@ -196,7 +223,7 @@
max_speed=OFPPF_1GB_FD
),
device_id=device.id,
- device_port_no=uni_port.port_no
+ device_port_no=self.uni_port.port_no
))
device = self.adapter_agent.get_device(device.id)
@@ -219,3 +246,143 @@
self.adapter_agent.send_proxied_message(self.proxy_address, msg)
yield self.incoming_messages.get()
+
+
+ @inlineCallbacks
+ def reboot(self):
+ self.log.info('rebooting', device_id=self.device_id)
+
+ # Update the operational status to ACTIVATING and connect status to
+ # UNREACHABLE
+ device = self.adapter_agent.get_device(self.device_id)
+ previous_oper_status = device.oper_status
+ previous_conn_status = device.connect_status
+ device.oper_status = OperStatus.ACTIVATING
+ device.connect_status = ConnectStatus.UNREACHABLE
+ self.adapter_agent.update_device(device)
+
+ # Sleep 10 secs, simulating a reboot
+ #TODO: send alert and clear alert after the reboot
+ yield asleep(10)
+
+ # Change the operational status back to its previous state. With a
+ # real OLT the operational state should be the state the device is
+ # after a reboot.
+ # Get the latest device reference
+ device = self.adapter_agent.get_device(self.device_id)
+ device.oper_status = previous_oper_status
+ device.connect_status = previous_conn_status
+ self.adapter_agent.update_device(device)
+ self.log.info('rebooted', device_id=self.device_id)
+
+
+ def disable(self):
+ self.log.info('disabling', device_id=self.device_id)
+
+ # Get the latest device reference
+ device = self.adapter_agent.get_device(self.device_id)
+
+ # Disable all ports on that device
+ self.adapter_agent.disable_all_ports(self.device_id)
+
+ # Update the device operational status to UNKNOWN
+ device.oper_status = OperStatus.UNKNOWN
+ device.connect_status = ConnectStatus.UNREACHABLE
+ self.adapter_agent.update_device(device)
+
+ # Remove the uni logical port from the OLT, if still present
+ parent_device = self.adapter_agent.get_device(device.parent_id)
+ assert parent_device
+ logical_device_id = parent_device.parent_id
+ assert logical_device_id
+ port_no = device.proxy_address.channel_id
+ port_id = 'uni-{}'.format(port_no)
+ try:
+ port = self.adapter_agent.get_logical_port(logical_device_id, port_id)
+ self.adapter_agent.delete_logical_port(logical_device_id, port)
+ except KeyError:
+ self.log.info('logical-port-not-found', device_id=self.device_id,
+ portid=port_id)
+
+ # Remove pon port from parent
+ self.adapter_agent.delete_port_reference_from_parent(self.device_id,
+ self.pon_port)
+
+ # Just updating the port status may be an option as well
+ # port.ofp_port.config = OFPPC_NO_RECV
+ # yield self.adapter_agent.update_logical_port(logical_device_id,
+ # port)
+ # Unregister for proxied message
+ self.adapter_agent.unregister_for_proxied_messages(device.proxy_address)
+
+ # TODO:
+ # 1) Remove all flows from the device
+ # 2) Remove the device from ponsim
+
+ self.log.info('disabled', device_id=device.id)
+
+
+ def reenable(self):
+ self.log.info('re-enabling', device_id=self.device_id)
+
+ # Get the latest device reference
+ device = self.adapter_agent.get_device(self.device_id)
+
+ # First we verify that we got parent reference and proxy info
+ assert self.uni_port
+ assert device.parent_id
+ assert device.proxy_address.device_id
+ assert device.proxy_address.channel_id
+
+ # Re-register for proxied messages right away
+ self.proxy_address = device.proxy_address
+ self.adapter_agent.register_for_proxied_messages(device.proxy_address)
+
+ # Re-enable the ports on that device
+ self.adapter_agent.reenable_all_ports(self.device_id)
+
+ # Update the connect status to REACHABLE
+ device.connect_status = ConnectStatus.REACHABLE
+ self.adapter_agent.update_device(device)
+
+ # re-add uni port to logical device
+ parent_device = self.adapter_agent.get_device(device.parent_id)
+ logical_device_id = parent_device.parent_id
+ assert logical_device_id
+ port_no = device.proxy_address.channel_id
+ cap = OFPPF_1GB_FD | OFPPF_FIBER
+ self.adapter_agent.add_logical_port(logical_device_id, LogicalPort(
+ id='uni-{}'.format(port_no),
+ ofp_port=ofp_port(
+ port_no=port_no,
+ hw_addr=mac_str_to_tuple('00:00:00:00:00:%02x' % port_no),
+ name='uni-{}'.format(port_no),
+ config=0,
+ state=OFPPS_LIVE,
+ curr=cap,
+ advertised=cap,
+ peer=cap,
+ curr_speed=OFPPF_1GB_FD,
+ max_speed=OFPPF_1GB_FD
+ ),
+ device_id=device.id,
+ device_port_no=self.uni_port.port_no
+ ))
+
+ device = self.adapter_agent.get_device(device.id)
+ device.oper_status = OperStatus.ACTIVE
+ self.adapter_agent.update_device(device)
+
+ self.log.info('re-enabled', device_id=device.id)
+
+
+ def delete(self):
+ self.log.info('deleting', device_id=self.device_id)
+
+ # A delete request may be received when an OLT is dsiabled
+
+ # TODO:
+ # 1) Remove all flows from the device
+ # 2) Remove the device from ponsim
+
+ self.log.info('deleted', device_id=self.device_id)
diff --git a/voltha/adapters/simulated_olt/simulated_olt.py b/voltha/adapters/simulated_olt/simulated_olt.py
index 1116ac1..b5a6c3f 100644
--- a/voltha/adapters/simulated_olt/simulated_olt.py
+++ b/voltha/adapters/simulated_olt/simulated_olt.py
@@ -152,7 +152,19 @@
def abandon_device(self, device):
raise NotImplementedError()
- def deactivate_device(self, device):
+ def disable_device(self, device):
+ raise NotImplementedError()
+
+ def reenable_device(self, device):
+ raise NotImplementedError()
+
+ def reboot_device(self, device):
+ raise NotImplementedError()
+
+ def delete_device(self, device):
+ raise NotImplementedError()
+
+ def get_device_details(self, device):
raise NotImplementedError()
def _tmp_populate_stuff(self):
diff --git a/voltha/adapters/simulated_onu/simulated_onu.py b/voltha/adapters/simulated_onu/simulated_onu.py
index 2b0ea9f..9b45af5 100644
--- a/voltha/adapters/simulated_onu/simulated_onu.py
+++ b/voltha/adapters/simulated_onu/simulated_onu.py
@@ -93,7 +93,19 @@
def abandon_device(self, device):
raise NotImplementedError()
- def deactivate_device(self, device):
+ def disable_device(self, device):
+ raise NotImplementedError()
+
+ def reenable_device(self, device):
+ raise NotImplementedError()
+
+ def reboot_device(self, device):
+ raise NotImplementedError()
+
+ def delete_device(self, device):
+ raise NotImplementedError()
+
+ def get_device_details(self, device):
raise NotImplementedError()
@inlineCallbacks
diff --git a/voltha/adapters/tibit_olt/tibit_olt.py b/voltha/adapters/tibit_olt/tibit_olt.py
index 4551b35..3bf9bfa 100644
--- a/voltha/adapters/tibit_olt/tibit_olt.py
+++ b/voltha/adapters/tibit_olt/tibit_olt.py
@@ -463,7 +463,19 @@
def abandon_device(self, device):
raise NotImplementedError(0
)
- def deactivate_device(self, device):
+ def disable_device(self, device):
+ raise NotImplementedError()
+
+ def reenable_device(self, device):
+ raise NotImplementedError()
+
+ def reboot_device(self, device):
+ raise NotImplementedError()
+
+ def delete_device(self, device):
+ raise NotImplementedError()
+
+ def get_device_details(self, device):
raise NotImplementedError()
def update_flows_bulk(self, device, flows, groups):
diff --git a/voltha/adapters/tibit_onu/tibit_onu.py b/voltha/adapters/tibit_onu/tibit_onu.py
index 5c11245..33b7751 100644
--- a/voltha/adapters/tibit_onu/tibit_onu.py
+++ b/voltha/adapters/tibit_onu/tibit_onu.py
@@ -202,7 +202,19 @@
def abandon_device(self, device):
raise NotImplementedError(0
)
- def deactivate_device(self, device):
+ def disable_device(self, device):
+ raise NotImplementedError()
+
+ def reenable_device(self, device):
+ raise NotImplementedError()
+
+ def reboot_device(self, device):
+ raise NotImplementedError()
+
+ def delete_device(self, device):
+ raise NotImplementedError()
+
+ def get_device_details(self, device):
raise NotImplementedError()
def update_flows_bulk(self, device, flows, groups):
diff --git a/voltha/core/adapter_agent.py b/voltha/core/adapter_agent.py
index e7a26f8..4859d27 100644
--- a/voltha/core/adapter_agent.py
+++ b/voltha/core/adapter_agent.py
@@ -35,7 +35,7 @@
from voltha.protos.events_pb2 import KpiEvent, AlarmEvent, AlarmEventType, \
AlarmEventSeverity, AlarmEventState, AlarmEventCategory
from voltha.protos.voltha_pb2 import DeviceGroup, LogicalDevice, \
- LogicalPort, AdminState
+ LogicalPort, AdminState, OperStatus
from voltha.registry import registry
from voltha.core.flow_decomposer import OUTPUT
@@ -62,6 +62,7 @@
self._rx_event_subscriptions = {}
self._tx_event_subscriptions = {}
self.event_bus = EventBusClient()
+ self.packet_out_subscription = None
self.log = structlog.get_logger(adapter_name=adapter_name)
@inlineCallbacks
@@ -130,6 +131,22 @@
root_proxy.add(container_path, data)
return full_path
+ def _remove_node(self, container_path, key):
+ """
+ Remove a node from the data model
+ :param container_path: path to node
+ :param key: node
+ :return: None
+ """
+ full_path = container_path + '/' + str(key)
+ root_proxy = self.core.get_proxy('/')
+ try:
+ root_proxy.get(full_path)
+ root_proxy.remove(full_path)
+ except KeyError:
+ # Node does not exist
+ pass
+
# ~~~~~~~~~~~~~~~~~~~~~ Core-Facing Service ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
def adopt_device(self, device):
@@ -138,8 +155,20 @@
def abandon_device(self, device):
return self.adapter.abandon_device(device)
- def deactivate_device(self, device):
- return self.adapter.deactivate_device(device)
+ def disable_device(self, device):
+ return self.adapter.disable_device(device)
+
+ def reenable_device(self, device):
+ return self.adapter.reenable_device(device)
+
+ def reboot_device(self, device):
+ return self.adapter.reboot_device(device)
+
+ def delete_device(self, device):
+ return self.adapter.delete_device(device)
+
+ def get_device_details(self, device):
+ return self.adapter.get_device_details(device)
def update_flows_bulk(self, device, flows, groups):
return self.adapter.update_flows_bulk(device, flows, groups)
@@ -174,10 +203,6 @@
device_agent = self.core.get_device_agent(device.id)
device_agent.update_device(device)
- def remove_device(self, device_id):
- device_agent = self.core.get_device_agent(device_id)
- device_agent.remove_device(device_id)
-
def add_port(self, device_id, port):
assert isinstance(port, Port)
@@ -196,6 +221,71 @@
self._make_up_to_date('/devices/{}/ports'.format(device_id),
port.port_no, port)
+ def disable_all_ports(self, device_id):
+ """
+ Disable all ports on that device, i.e. change the admin status to
+ disable and operational status to UNKNOWN
+ :param device_id: device id
+ :return: None
+ """
+
+ # get all device ports
+ ports = self.root_proxy.get('/devices/{}/ports'.format(device_id))
+ for port in ports:
+ port.admin_state = AdminState.DISABLED
+ port.oper_status = OperStatus.UNKNOWN
+ self._make_up_to_date('/devices/{}/ports'.format(device_id),
+ port.port_no, port)
+
+ def reenable_all_ports(self, device_id):
+ """
+ Re-enable all ports on that device, i.e. change the admin status to
+ enabled and operational status to ACTIVE
+ :param device_id: device id
+ :return: None
+ """
+
+ # get all device ports
+ ports = self.root_proxy.get('/devices/{}/ports'.format(device_id))
+ for port in ports:
+ port.admin_state = AdminState.ENABLED
+ port.oper_status = OperStatus.ACTIVE
+ self._make_up_to_date('/devices/{}/ports'.format(device_id),
+ port.port_no, port)
+
+ def delete_all_peer_references(self, device_id):
+ """
+ Remove all peer port references for that device
+ :param device_id: device_id of device
+ :return: None
+ """
+ ports = self.root_proxy.get('/devices/{}/ports'.format(device_id))
+ for port in ports:
+ port_path = '/devices/{}/ports/{}'.format(device_id, port.port_no)
+ for peer in port.peers:
+ port.peers.remove(peer)
+ self.root_proxy.update(port_path, port)
+
+ def delete_port_reference_from_parent(self, device_id, port):
+ """
+ Delete the port reference from the parent device
+ :param device_id: id of device containing the port
+ :param port: port to remove
+ :return: None
+ """
+ assert isinstance(port, Port)
+ self.log.info('delete-port-reference', device_id=device_id, port=port)
+
+ # for referential integrity, remove references
+ me_as_peer = Port.PeerPort(device_id=device_id, port_no=port.port_no)
+ for peer in port.peers:
+ peer_port_path = '/devices/{}/ports/{}'.format(
+ peer.device_id, peer.port_no)
+ peer_port = self.root_proxy.get(peer_port_path)
+ if me_as_peer in peer_port.peers:
+ peer_port.peers.remove(me_as_peer)
+ self.root_proxy.update(peer_port_path, peer_port)
+
def _find_first_available_id(self):
logical_devices = self.root_proxy.get('/logical_devices')
existing_ids = set(ld.id for ld in logical_devices)
@@ -210,6 +300,10 @@
return self.root_proxy.get('/logical_devices/{}'.format(
logical_device_id))
+ def get_logical_port(self, logical_device_id, port_id):
+ return self.root_proxy.get('/logical_devices/{}/ports/{}'.format(
+ logical_device_id, port_id))
+
def create_logical_device(self, logical_device):
assert isinstance(logical_device, LogicalDevice)
@@ -221,13 +315,34 @@
self._make_up_to_date('/logical_devices',
logical_device.id, logical_device)
- self.event_bus.subscribe(
+ # Keep a reference to the packet out subscription as it will be
+ # referred during removal
+ self.packet_out_subscription = self.event_bus.subscribe(
topic='packet-out:{}'.format(logical_device.id),
callback=lambda _, p: self.receive_packet_out(logical_device.id, p)
)
return logical_device
+
+ def delete_logical_device(self, logical_device):
+ """
+ This will remove the logical device as well as all logical ports
+ associated with it
+ :param logical_device: The logical device to remove
+ :return: None
+ """
+ assert isinstance(logical_device, LogicalDevice)
+
+ # Remove packet out subscription
+ self.event_bus.unsubscribe(self.packet_out_subscription)
+
+ # Remove node from the data model - this will trigger the logical
+ # device 'remove callbacks' as well as logical ports 'remove
+ # callbacks' if present
+ self._remove_node('/logical_devices', logical_device.id)
+
+
def receive_packet_out(self, logical_device_id, ofp_packet_out):
def get_port_out(opo):
@@ -245,6 +360,21 @@
'/logical_devices/{}/ports'.format(logical_device_id),
port.id, port)
+ def delete_logical_port(self, logical_device_id, port):
+ assert isinstance(port, LogicalPort)
+ self._remove_node('/logical_devices/{}/ports'.format(
+ logical_device_id), port.id)
+
+ def update_logical_port(self, logical_device_id, port):
+ assert isinstance(port, LogicalPort)
+ self.log.debug('update-logical-port',
+ logical_device_id=logical_device_id,
+ port=port)
+
+ self._make_up_to_date(
+ '/logical_devices/{}/ports'.format(logical_device_id),
+ port.id, port)
+
def child_device_detected(self,
parent_device_id,
parent_port_no,
@@ -269,6 +399,49 @@
self._tx_event_subscriptions[topic] = self.event_bus.subscribe(
topic, lambda t, m: self._send_proxied_message(proxy_address, m))
+ def remove_all_logical_ports(self, logical_device_id):
+ """ Remove all logical ports from a given logical device"""
+ ports = self.root_proxy.get('/logical_devices/{}/ports')
+ for port in ports:
+ self._remove_node('/logical_devices/{}/ports', port.id)
+
+ def delete_all_child_devices(self, parent_device_id):
+ """ Remove all ONUs from a given OLT """
+ devices = self.root_proxy.get('/devices')
+ children_ids = set(d.id for d in devices if d.parent_id == parent_device_id)
+ self.log.debug('devices-to-delete',
+ parent_id=parent_device_id,
+ children_ids=children_ids)
+ for child_id in children_ids:
+ self._remove_node('/devices', child_id)
+
+ def reenable_all_child_devices(self, parent_device_id):
+ """ Re-enable all ONUs from a given OLT """
+ devices = self.root_proxy.get('/devices')
+ children_ids = set(d.id for d in devices if d.parent_id == parent_device_id)
+ self.log.debug('devices-to-reenable',
+ parent_id=parent_device_id,
+ children_ids=children_ids)
+ for child_id in children_ids:
+ device = self.get_device(child_id)
+ device.admin_state = AdminState.ENABLED
+ self._make_up_to_date(
+ '/devices', device.id, device)
+
+ def disable_all_child_devices(self, parent_device_id):
+ """ Disable all ONUs from a given OLT """
+ devices = self.root_proxy.get('/devices')
+ children_ids = set(d.id for d in devices if d.parent_id == parent_device_id)
+ self.log.debug('devices-to-disable',
+ parent_id=parent_device_id,
+ children_ids=children_ids)
+ for child_id in children_ids:
+ # Change the admin state pf the device to DISABLE
+ device = self.get_device(child_id)
+ device.admin_state = AdminState.DISABLED
+ self._make_up_to_date(
+ '/devices', device.id, device)
+
def _gen_rx_proxy_address_topic(self, proxy_address):
"""Generate unique topic name specific to this proxy address for rx"""
topic = 'rx:' + MessageToJson(proxy_address)
@@ -285,6 +458,11 @@
topic,
lambda t, m: self._receive_proxied_message(proxy_address, m))
+ def unregister_for_proxied_messages(self, proxy_address):
+ topic = self._gen_rx_proxy_address_topic(proxy_address)
+ self.event_bus.unsubscribe(self._rx_event_subscriptions[topic])
+ del self._rx_event_subscriptions[topic]
+
def _receive_proxied_message(self, proxy_address, msg):
self.adapter.receive_proxied_message(proxy_address, msg)
diff --git a/voltha/core/core.py b/voltha/core/core.py
index 3b28cbd..c11136b 100644
--- a/voltha/core/core.py
+++ b/voltha/core/core.py
@@ -96,7 +96,7 @@
pass # ignore others
def _post_remove_callback(self, data, *args, **kw):
- log.debug('added', data=data, args=args, kw=kw)
+ log.debug('removed', data=data, args=args, kw=kw)
if isinstance(data, Device):
self._handle_remove_device(data)
elif isinstance(data, LogicalDevice):
@@ -118,7 +118,7 @@
@inlineCallbacks
def _handle_remove_device(self, device):
if device.id in self.device_agents:
- yield self.device_agents[device.id].stop()
+ yield self.device_agents[device.id].stop(device)
del self.device_agents[device.id]
def get_device_agent(self, device_id):
diff --git a/voltha/core/device_agent.py b/voltha/core/device_agent.py
index 55feb14..2cb56ec 100644
--- a/voltha/core/device_agent.py
+++ b/voltha/core/device_agent.py
@@ -70,14 +70,31 @@
self.log.info('started')
returnValue(self)
- def stop(self):
- self.log.debug('stopping')
+ @inlineCallbacks
+ def stop(self, device):
+ self.log.debug('stopping', device=device)
+
+ # First, propagate this request to the device agents
+ yield self._delete_device(device)
+
self.proxy.unregister_callback(
CallbackType.PRE_UPDATE, self._validate_update)
self.proxy.unregister_callback(
CallbackType.POST_UPDATE, self._process_update)
self.log.info('stopped')
+ @inlineCallbacks
+ def reboot_device(self, device, dry_run=False):
+ self.log.info('reboot-device', device=device, dry_run=dry_run)
+ if not dry_run:
+ yield self.adapter_agent.reboot_device(device)
+
+ @inlineCallbacks
+ def get_device_details(self, device, dry_run=False):
+ self.log.info('get-device-details', device=device, dry_run=dry_run)
+ if not dry_run:
+ yield self.adapter_agent.get_device_details(device)
+
def _set_adapter_agent(self):
adapter_name = self._tmp_initial_data.adapter
if adapter_name == '':
@@ -144,9 +161,6 @@
self.last_data = device # so that we don't propagate back
self.proxy.update('/', device)
- def remove_device(self, device_id):
- raise NotImplementedError()
-
def _propagate_change(self, device, dry_run=False):
self.log.info('propagate-change', device=device, dry_run=dry_run)
if device != self.last_data:
@@ -158,13 +172,23 @@
self.log.info('abandon-device', device=device, dry_run=dry_run)
raise NotImplementedError()
+ @inlineCallbacks
def _disable_device(self, device, dry_run=False):
self.log.info('disable-device', device=device, dry_run=dry_run)
- raise NotImplementedError()
+ if not dry_run:
+ yield self.adapter_agent.disable_device(device)
+ @inlineCallbacks
def _reenable_device(self, device, dry_run=False):
self.log.info('reenable-device', device=device, dry_run=dry_run)
- raise NotImplementedError()
+ if not dry_run:
+ yield self.adapter_agent.reenable_device(device)
+
+ @inlineCallbacks
+ def _delete_device(self, device, dry_run=False):
+ self.log.info('delete-device', device=device, dry_run=dry_run)
+ if not dry_run:
+ yield self.adapter_agent.delete_device(device)
admin_state_fsm = {
@@ -204,7 +228,7 @@
# add ability to notify called when an flow update completes
# see https://jira.opencord.org/browse/CORD-839
- elif self.accepts_add_remove_flow_updates:
+ elif self.device_type.accepts_add_remove_flow_updates:
raise NotImplementedError()
else:
@@ -228,7 +252,7 @@
# add ability to notify called when an group update completes
# see https://jira.opencord.org/browse/CORD-839
- elif self.accepts_add_remove_flow_updates:
+ elif self.device_type.accepts_add_remove_flow_updates:
raise NotImplementedError()
else:
diff --git a/voltha/core/global_handler.py b/voltha/core/global_handler.py
index 326f407..7d52d29 100644
--- a/voltha/core/global_handler.py
+++ b/voltha/core/global_handler.py
@@ -264,13 +264,87 @@
context)
@twisted_async
- def ActivateDevice(self, request, context):
+ def EnableDevice(self, request, context):
log.info('grpc-request', request=request)
- # TODO dispatching to local instead of passing it to leader
+
+ try:
+ instance_id = self.dispatcher.instance_id_by_device_id(
+ request.id
+ )
+ except KeyError:
+ context.set_details(
+ 'Device \'{}\' not found'.format(request.id))
+ context.set_code(StatusCode.NOT_FOUND)
+ return Device()
+
return self.dispatcher.dispatch(
- self.instance_id,
+ instance_id,
VolthaLocalServiceStub,
- 'ActivateDevice',
+ 'EnableDevice',
+ request,
+ context)
+
+
+ @twisted_async
+ def DisableDevice(self, request, context):
+ log.info('grpc-request', request=request)
+
+ try:
+ instance_id = self.dispatcher.instance_id_by_device_id(
+ request.id
+ )
+ except KeyError:
+ context.set_details(
+ 'Device \'{}\' not found'.format(request.id))
+ context.set_code(StatusCode.NOT_FOUND)
+ return Device()
+
+ return self.dispatcher.dispatch(
+ instance_id,
+ VolthaLocalServiceStub,
+ 'DisableDevice',
+ request,
+ context)
+
+ @twisted_async
+ def RebootDevice(self, request, context):
+ log.info('grpc-request', request=request)
+
+ try:
+ instance_id = self.dispatcher.instance_id_by_device_id(
+ request.id
+ )
+ except KeyError:
+ context.set_details(
+ 'Device \'{}\' not found'.format(request.id))
+ context.set_code(StatusCode.NOT_FOUND)
+ return Device()
+
+ return self.dispatcher.dispatch(
+ instance_id,
+ VolthaLocalServiceStub,
+ 'RebootDevice',
+ request,
+ context)
+
+ @twisted_async
+ def DeleteDevice(self, request, context):
+ log.info('grpc-request', request=request)
+
+ try:
+ instance_id = self.dispatcher.instance_id_by_device_id(
+ request.id
+ )
+ except KeyError:
+ context.set_details(
+ 'Device \'{}\' not found'.format(request.id))
+ context.set_code(StatusCode.NOT_FOUND)
+ return Device()
+
+ return self.dispatcher.dispatch(
+ instance_id,
+ VolthaLocalServiceStub,
+ 'DeleteDevice',
request,
context)
diff --git a/voltha/core/local_handler.py b/voltha/core/local_handler.py
index c0309eb..e7a4912 100644
--- a/voltha/core/local_handler.py
+++ b/voltha/core/local_handler.py
@@ -35,7 +35,6 @@
class LocalHandler(VolthaLocalServiceServicer):
-
def __init__(self, core, **init_kw):
self.core = core
self.init_kw = init_kw
@@ -48,10 +47,12 @@
if 'root' in config_backend:
# This is going to block the entire reactor until loading is completed
log.info('loading config from persisted backend')
- self.root = ConfigRoot.load(VolthaInstance, kv_store=config_backend)
+ self.root = ConfigRoot.load(VolthaInstance,
+ kv_store=config_backend)
else:
log.info('initializing new config')
- self.root = ConfigRoot(VolthaInstance(**self.init_kw), kv_store=config_backend)
+ self.root = ConfigRoot(VolthaInstance(**self.init_kw),
+ kv_store=config_backend)
else:
self.root = ConfigRoot(VolthaInstance(**self.init_kw))
@@ -126,7 +127,8 @@
return LogicalPorts()
try:
- items = self.root.get('/logical_devices/{}/ports'.format(request.id))
+ items = self.root.get(
+ '/logical_devices/{}/ports'.format(request.id))
return LogicalPorts(items=items)
except KeyError:
context.set_details(
@@ -145,7 +147,8 @@
return Flows()
try:
- flows = self.root.get('/logical_devices/{}/flows'.format(request.id))
+ flows = self.root.get(
+ '/logical_devices/{}/flows'.format(request.id))
return flows
except KeyError:
context.set_details(
@@ -153,7 +156,6 @@
context.set_code(StatusCode.NOT_FOUND)
return Flows()
-
@twisted_async
def UpdateLogicalDeviceFlowTable(self, request, context):
log.info('grpc-request', request=request)
@@ -277,7 +279,7 @@
return request
@twisted_async
- def ActivateDevice(self, request, context):
+ def EnableDevice(self, request, context):
log.info('grpc-request', request=request)
if '/' in request.id:
@@ -289,9 +291,109 @@
try:
path = '/devices/{}'.format(request.id)
device = self.root.get(path)
+ assert device.admin_state in (AdminState.PREPROVISIONED,
+ AdminState.DISABLED), \
+ 'Device to enable cannot be ' \
+ 'in admin state \'{}\''.format(device.admin_state)
device.admin_state = AdminState.ENABLED
self.root.update(path, device, strict=True)
+ except AssertionError, e:
+ context.set_details(e.msg)
+ context.set_code(StatusCode.INVALID_ARGUMENT)
+ return Device()
+
+ except KeyError:
+ context.set_details(
+ 'Device \'{}\' not found'.format(request.id))
+ context.set_code(StatusCode.NOT_FOUND)
+
+ return Empty()
+
+ @twisted_async
+ def DisableDevice(self, request, context):
+ log.info('grpc-request', request=request)
+
+ if '/' in request.id:
+ context.set_details(
+ 'Malformed device id \'{}\''.format(request.id))
+ context.set_code(StatusCode.INVALID_ARGUMENT)
+ return Device()
+
+ try:
+ path = '/devices/{}'.format(request.id)
+ device = self.root.get(path)
+ assert device.admin_state == AdminState.ENABLED, \
+ 'Device to disable cannot be ' \
+ 'in admin state \'{}\''.format(device.admin_state)
+ device.admin_state = AdminState.DISABLED
+ self.root.update(path, device, strict=True)
+
+ except AssertionError, e:
+ context.set_details(e.msg)
+ context.set_code(StatusCode.INVALID_ARGUMENT)
+ return Device()
+
+ except KeyError:
+ context.set_details(
+ 'Device \'{}\' not found'.format(request.id))
+ context.set_code(StatusCode.NOT_FOUND)
+
+ return Empty()
+
+ @twisted_async
+ def RebootDevice(self, request, context):
+ log.info('grpc-request', request=request)
+
+ if '/' in request.id:
+ context.set_details(
+ 'Malformed device id \'{}\''.format(request.id))
+ context.set_code(StatusCode.INVALID_ARGUMENT)
+ return Device()
+
+ try:
+ path = '/devices/{}'.format(request.id)
+ device = self.root.get(path)
+
+ agent = self.core.get_device_agent(device.id)
+ agent.reboot_device(device)
+
+ except AssertionError, e:
+ context.set_details(e.msg)
+ context.set_code(StatusCode.INVALID_ARGUMENT)
+ return Device()
+
+ except KeyError:
+ context.set_details(
+ 'Device \'{}\' not found'.format(request.id))
+ context.set_code(StatusCode.NOT_FOUND)
+
+ return Empty()
+
+ @twisted_async
+ def DeleteDevice(self, request, context):
+ log.info('grpc-request', request=request)
+
+ if '/' in request.id:
+ context.set_details(
+ 'Malformed device id \'{}\''.format(request.id))
+ context.set_code(StatusCode.INVALID_ARGUMENT)
+ return Device()
+
+ try:
+ path = '/devices/{}'.format(request.id)
+ device = self.root.get(path)
+ assert device.admin_state == AdminState.DISABLED, \
+ 'Device to delete cannot be ' \
+ 'in admin state \'{}\''.format(device.admin_state)
+
+ self.root.remove(path)
+
+ except AssertionError, e:
+ context.set_details(e.msg)
+ context.set_code(StatusCode.INVALID_ARGUMENT)
+ return Device()
+
except KeyError:
context.set_details(
'Device \'{}\' not found'.format(request.id))
@@ -318,7 +420,6 @@
context.set_code(StatusCode.NOT_FOUND)
return Ports()
-
@twisted_async
def ListDeviceFlows(self, request, context):
log.info('grpc-request', request=request)
@@ -338,7 +439,6 @@
context.set_code(StatusCode.NOT_FOUND)
return Flows()
-
@twisted_async
def ListDeviceFlowGroups(self, request, context):
log.info('grpc-request', request=request)
@@ -350,7 +450,8 @@
return FlowGroups()
try:
- groups = self.root.get('/devices/{}/flow_groups'.format(request.id))
+ groups = self.root.get(
+ '/devices/{}/flow_groups'.format(request.id))
return groups
except KeyError:
context.set_details(
diff --git a/voltha/core/logical_device_agent.py b/voltha/core/logical_device_agent.py
index e34f89b..1ecd032 100644
--- a/voltha/core/logical_device_agent.py
+++ b/voltha/core/logical_device_agent.py
@@ -28,7 +28,7 @@
from voltha.core.flow_decomposer import FlowDecomposer, \
flow_stats_entry_from_flow_mod_message, group_entry_from_group_mod, \
mk_flow_stat, in_port, vlan_vid, vlan_pcp, pop_vlan, output, set_field, \
- push_vlan
+ push_vlan, mk_simple_flow_mod
from voltha.protos import third_party
from voltha.protos import openflow_13_pb2 as ofp
from voltha.protos.device_pb2 import Port
@@ -53,6 +53,8 @@
'/logical_devices/{}/flows'.format(logical_device.id))
self.groups_proxy = core.get_proxy(
'/logical_devices/{}/flow_groups'.format(logical_device.id))
+ # self.port_proxy = core.get_proxy(
+ # '/logical_devices/{}/ports'.format(logical_device.id))
self.self_proxy = core.get_proxy(
'/logical_devices/{}'.format(logical_device.id))
@@ -60,6 +62,8 @@
CallbackType.POST_UPDATE, self._flow_table_updated)
self.groups_proxy.register_callback(
CallbackType.POST_UPDATE, self._group_table_updated)
+ # self.port_proxy.register_callback(
+ # CallbackType.POST_UPDATE, self._port_changed)
self.self_proxy.register_callback(
CallbackType.POST_ADD, self._port_added)
self.self_proxy.register_callback(
@@ -89,6 +93,9 @@
CallbackType.POST_ADD, self._port_list_updated)
self.self_proxy.unregister_callback(
CallbackType.POST_REMOVE, self._port_list_updated)
+
+ # Remove subscription to the event bus
+ self.event_bus.unsubscribe(self.packet_in_subscription)
self.log.info('stopped')
def announce_flows_deleted(self, flows):
@@ -108,6 +115,15 @@
def signal_group_mod_error(self, code, group_mod):
pass # TODO
+ def delete_all_flows(self):
+ self.update_flow_table(mk_simple_flow_mod(
+ command=ofp.OFPFC_DELETE,
+ out_port=ofp.OFPP_ANY,
+ out_group=ofp.OFPG_ANY,
+ match_fields=[],
+ actions=[]
+ ))
+
def update_flow_table(self, flow_mod):
command = flow_mod.command
@@ -520,6 +536,7 @@
# ~~~~~~~~~~~~~~~~~~~ APIs NEEDED BY FLOW DECOMPOSER ~~~~~~~~~~~~~~~~~~~~~~
def _port_added(self, port):
+ self.log.debug('port-added', port=port)
assert isinstance(port, LogicalPort)
self._port_list_updated(port)
self.local_handler.send_port_change_event(
@@ -531,6 +548,7 @@
)
def _port_removed(self, port):
+ self.log.debug('port-removed', port=port)
assert isinstance(port, LogicalPort)
self._port_list_updated(port)
self.local_handler.send_port_change_event(
diff --git a/voltha/protos/voltha.proto b/voltha/protos/voltha.proto
index 6cb8262..db98c2a 100644
--- a/voltha/protos/voltha.proto
+++ b/voltha/protos/voltha.proto
@@ -9,6 +9,7 @@
package voltha;
import "google/protobuf/empty.proto";
+import "google/protobuf/any.proto";
import "google/api/annotations.proto";
import "yang_options.proto";
@@ -201,10 +202,33 @@
};
}
- // Activate a pre-provisioned device
- rpc ActivateDevice(ID) returns(google.protobuf.Empty) {
+ // Enable a device. If the device was in pre-provisioned state then it
+ // will tansition to ENABLED state. If it was is DISABLED state then it
+ // will tansition to ENABLED state as well.
+ rpc EnableDevice(ID) returns(google.protobuf.Empty) {
option (google.api.http) = {
- post: "/api/v1/devices/{id}/activate"
+ post: "/api/v1/devices/{id}/enable"
+ };
+ }
+
+ // Disable a device
+ rpc DisableDevice(ID) returns(google.protobuf.Empty) {
+ option (google.api.http) = {
+ post: "/api/v1/devices/{id}/disable"
+ };
+ }
+
+ // Reboot a device
+ rpc RebootDevice(ID) returns(google.protobuf.Empty) {
+ option (google.api.http) = {
+ post: "/api/v1/devices/{id}/reboot"
+ };
+ }
+
+ // Delete a device
+ rpc DeleteDevice(ID) returns(google.protobuf.Empty) {
+ option (google.api.http) = {
+ delete: "/api/v1/devices/{id}/delete"
};
}
@@ -379,10 +403,33 @@
};
}
- // Activate a pre-provisioned device
- rpc ActivateDevice(ID) returns(google.protobuf.Empty) {
+ // Enable a device. If the device was in pre-provisioned state then it
+ // will tansition to ENABLED state. If it was is DISABLED state then it
+ // will tansition to ENABLED state as well.
+ rpc EnableDevice(ID) returns(google.protobuf.Empty) {
option (google.api.http) = {
- post: "/api/v1/local/devices/{id}/activate"
+ post: "/api/v1/local/devices/{id}/enable"
+ };
+ }
+
+ // Disable a device
+ rpc DisableDevice(ID) returns(google.protobuf.Empty) {
+ option (google.api.http) = {
+ post: "/api/v1/local/devices/{id}/disable"
+ };
+ }
+
+ // Reboot a device
+ rpc RebootDevice(ID) returns(google.protobuf.Empty) {
+ option (google.api.http) = {
+ post: "/api/v1/local/devices/{id}/reboot"
+ };
+ }
+
+ // Delete a device
+ rpc DeleteDevice(ID) returns(google.protobuf.Empty) {
+ option (google.api.http) = {
+ post: "/api/v1/local/devices/{id}/delete"
};
}