From d644d63f392a3784586a77329817db462c228b51 Mon Sep 17 00:00:00 2001 From: Moiz Raja Date: Mon, 10 Feb 2014 19:23:50 -0800 Subject: [PATCH] Fix issue where NodeConnector ADDED events were propagated as NodeConnector CHANGED events Note that this is an interim fix which will work only in a non-clustered setup The major part of this fix is coded in InventoryAndReadAdapter. The fix is as follows when a NodeConnectorChanged event is received by the InventoryAdapter it first checks whether this node connector has ever been seen before by it. It does this by looking at a Map of Nodes to NodeConnectors that it maintains. If the Node Connector is not found in the Map then this must be a NodeConnector ADDED notification otherwise it is a NodeConnector CHANGED notification. This commit also contains some code for what would be the ideal fix - one that would work in a cluster scenario as well That code is in InventoryNotificationProvider and NodeConnectorDataChangeListener. The fix in there works like this. We set up a DataChangeListener on the MD-SAL DataProvider Service which would notify us whenever a NodeConnector is added, modified or removed. When the change event comes in depending on whether the node connector was added or modified the data change listener fires the ADDED or CHANGED event. This fix is disabled for now till another issue with MD-SAL regarding removal of nodes on mininet exit (or switch disconnect) is resolved. There are some other changes in the FlowProgrammerAdapter which basically suppress some error conditions when a flow that is being removed or modified does not exist in the cache that the FlowProgrammerAdapter maintains to translate between AD-SAL and MD-SAL flows. This fix allows pingall to work with upto 32 switches - after that other issues prevent testing with more switches. Change-Id: Ide2e6e85191856a8c6a9746b2397605905938d56 Signed-off-by: Moiz Raja --- .../compatibility/ComponentActivator.xtend | 1 + .../compatibility/FlowProgrammerAdapter.xtend | 14 +- .../InventoryAndReadAdapter.xtend | 661 ++++++++++-------- .../InventoryNotificationProvider.java | 59 ++ .../NodeConnectorDataChangeListener.java | 77 ++ .../FlowCapableInventoryProvider.xtend | 11 + 6 files changed, 523 insertions(+), 300 deletions(-) create mode 100644 opendaylight/md-sal/compatibility/sal-compatibility/src/main/java/org/opendaylight/controller/sal/compatibility/InventoryNotificationProvider.java create mode 100644 opendaylight/md-sal/compatibility/sal-compatibility/src/main/java/org/opendaylight/controller/sal/compatibility/NodeConnectorDataChangeListener.java diff --git a/opendaylight/md-sal/compatibility/sal-compatibility/src/main/java/org/opendaylight/controller/sal/compatibility/ComponentActivator.xtend b/opendaylight/md-sal/compatibility/sal-compatibility/src/main/java/org/opendaylight/controller/sal/compatibility/ComponentActivator.xtend index a6fc4b0a23..a59c2c1636 100644 --- a/opendaylight/md-sal/compatibility/sal-compatibility/src/main/java/org/opendaylight/controller/sal/compatibility/ComponentActivator.xtend +++ b/opendaylight/md-sal/compatibility/sal-compatibility/src/main/java/org/opendaylight/controller/sal/compatibility/ComponentActivator.xtend @@ -251,6 +251,7 @@ package class SalCompatibilityProvider implements BindingAwareProvider { topology.dataService = session.getSALService(DataProviderService) tpProvider.dataService = session.getSALService(DataProviderService) + inventory.start(); tpProvider.start(); diff --git a/opendaylight/md-sal/compatibility/sal-compatibility/src/main/java/org/opendaylight/controller/sal/compatibility/FlowProgrammerAdapter.xtend b/opendaylight/md-sal/compatibility/sal-compatibility/src/main/java/org/opendaylight/controller/sal/compatibility/FlowProgrammerAdapter.xtend index fac12ee10d..8a0874ee31 100644 --- a/opendaylight/md-sal/compatibility/sal-compatibility/src/main/java/org/opendaylight/controller/sal/compatibility/FlowProgrammerAdapter.xtend +++ b/opendaylight/md-sal/compatibility/sal-compatibility/src/main/java/org/opendaylight/controller/sal/compatibility/FlowProgrammerAdapter.xtend @@ -199,9 +199,11 @@ class FlowProgrammerAdapter implements IPluginInFlowProgrammerService, SalFlowLi } private def Future> internalModifyFlowAsync(Node node, Flow oldFlow, Flow newFlow, long rid) { - val flowId = getCache().remove(oldFlow); + var flowId = getCache().remove(oldFlow); if(flowId == null){ - throw new IllegalArgumentException("oldFlow is unknown"); + LOG.error("oldFlow not found in cache : " + oldFlow.hashCode); + flowId = UUID.randomUUID(); + getCache().put(oldFlow, flowId); } getCache().put(newFlow, flowId); @@ -212,7 +214,9 @@ class FlowProgrammerAdapter implements IPluginInFlowProgrammerService, SalFlowLi private def Future> internalRemoveFlowAsync(Node node, Flow adflow, long rid){ val flowId = getCache().remove(adflow); if(flowId == null){ - throw new IllegalArgumentException("adflow is unknown"); + //throw new IllegalArgumentException("adflow not found in cache : " + adflow.hashCode); + LOG.error("adflow not found in cache : " + adflow.hashCode); + return null; } val flow = adflow.toMDFlow(flowId.toString()); val modification = this._dataBrokerService.beginTransaction(); @@ -227,6 +231,10 @@ class FlowProgrammerAdapter implements IPluginInFlowProgrammerService, SalFlowLi } private def toFutureStatus(Future> future){ + if(future == null){ + return toStatus(true); + } + try { val result = future.get(); return toStatus(result); diff --git a/opendaylight/md-sal/compatibility/sal-compatibility/src/main/java/org/opendaylight/controller/sal/compatibility/InventoryAndReadAdapter.xtend b/opendaylight/md-sal/compatibility/sal-compatibility/src/main/java/org/opendaylight/controller/sal/compatibility/InventoryAndReadAdapter.xtend index 60e43247c2..0c211fd0aa 100644 --- a/opendaylight/md-sal/compatibility/sal-compatibility/src/main/java/org/opendaylight/controller/sal/compatibility/InventoryAndReadAdapter.xtend +++ b/opendaylight/md-sal/compatibility/sal-compatibility/src/main/java/org/opendaylight/controller/sal/compatibility/InventoryAndReadAdapter.xtend @@ -11,6 +11,9 @@ import java.util.ArrayList import java.util.Collections import java.util.List import java.util.Set +import java.util.ArrayList; +import java.util.concurrent.locks.ReentrantLock; +import java.util.concurrent.locks.Lock; import java.util.concurrent.CopyOnWriteArrayList; import org.opendaylight.controller.sal.binding.api.data.DataBrokerService import org.opendaylight.controller.sal.binding.api.data.DataProviderService @@ -76,17 +79,18 @@ import static extension org.opendaylight.controller.sal.compatibility.NodeMappin import org.opendaylight.controller.md.sal.binding.util.TypeSafeDataReader import java.util.concurrent.ConcurrentHashMap import java.util.Map +import java.util.HashMap class InventoryAndReadAdapter implements IPluginInReadService, - IPluginInInventoryService, - OpendaylightInventoryListener, - OpendaylightFlowStatisticsListener, - OpendaylightFlowTableStatisticsListener, - OpendaylightPortStatisticsListener { + IPluginInInventoryService, + OpendaylightInventoryListener, + OpendaylightFlowStatisticsListener, + OpendaylightFlowTableStatisticsListener, + OpendaylightPortStatisticsListener { private static val LOG = LoggerFactory.getLogger(InventoryAndReadAdapter); - private static val OPENFLOWV10_TABLE_ID = new Integer(0).shortValue; + private static val OPENFLOWV10_TABLE_ID = new Integer(0).shortValue; @Property DataBrokerService dataService; @@ -111,21 +115,34 @@ class InventoryAndReadAdapter implements IPluginInReadService, @Property List inventoryPublisher = new CopyOnWriteArrayList(); - def setInventoryPublisher(IPluginOutInventoryService listener){ + private final InventoryNotificationProvider inventoryNotificationProvider = new InventoryNotificationProvider(); + + private final Map> nodeToNodeConnectorsMap = new ConcurrentHashMap>(); + + private final Lock nodeToNodeConnectorsLock = new ReentrantLock(); + + + def start(){ + inventoryNotificationProvider.dataProviderService = dataProviderService; + inventoryNotificationProvider.inventoryPublisher = inventoryPublisher; + // inventoryNotificationProvider.start(); + } + + def setInventoryPublisher(IPluginOutInventoryService listener){ inventoryPublisher.add(listener); - } + } - def unsetInventoryPublisher(IPluginOutInventoryService listener){ + def unsetInventoryPublisher(IPluginOutInventoryService listener){ inventoryPublisher.remove(listener); - } + } def setReadPublisher(IPluginOutReadService listener) { - statisticsPublisher.add(listener); + statisticsPublisher.add(listener); } def unsetReadPublisher (IPluginOutReadService listener) { - if( listener != null) - statisticsPublisher.remove(listener); + if( listener != null) + statisticsPublisher.remove(listener); } protected def startChange() { @@ -140,33 +157,33 @@ class InventoryAndReadAdapter implements IPluginInReadService, override readAllFlow(Node node, boolean cached) { val output = new ArrayList(); - val tableRef = InstanceIdentifier.builder(Nodes) - .child(org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node, InventoryMapping.toNodeKey(node)) - .augmentation(FlowCapableNode).child(Table, new TableKey(OPENFLOWV10_TABLE_ID)).toInstance(); - - val it = this.startChange(); - - val table= it.readConfigurationData(tableRef) as Table; - - if(table != null){ - LOG.trace("Number of flows installed in table 0 of node {} : {}",node,table.flow.size); - - for(flow : table.flow){ - - val adsalFlow = ToSalConversionsUtils.toFlow(flow,node); - val statsFromDataStore = flow.getAugmentation(FlowStatisticsData); - - if(statsFromDataStore != null){ - val it = new FlowOnNode(adsalFlow); - byteCount = statsFromDataStore.flowStatistics.byteCount.value.longValue; - packetCount = statsFromDataStore.flowStatistics.packetCount.value.longValue; - durationSeconds = statsFromDataStore.flowStatistics.duration.second.value.intValue; - durationNanoseconds = statsFromDataStore.flowStatistics.duration.nanosecond.value.intValue; - - output.add(it); - } - } - } + val tableRef = InstanceIdentifier.builder(Nodes) + .child(org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node, InventoryMapping.toNodeKey(node)) + .augmentation(FlowCapableNode).child(Table, new TableKey(OPENFLOWV10_TABLE_ID)).toInstance(); + + val it = this.startChange(); + + val table= it.readConfigurationData(tableRef) as Table; + + if(table != null){ + LOG.trace("Number of flows installed in table 0 of node {} : {}",node,table.flow.size); + + for(flow : table.flow){ + + val adsalFlow = ToSalConversionsUtils.toFlow(flow,node); + val statsFromDataStore = flow.getAugmentation(FlowStatisticsData); + + if(statsFromDataStore != null){ + val it = new FlowOnNode(adsalFlow); + byteCount = statsFromDataStore.flowStatistics.byteCount.value.longValue; + packetCount = statsFromDataStore.flowStatistics.packetCount.value.longValue; + durationSeconds = statsFromDataStore.flowStatistics.duration.second.value.intValue; + durationNanoseconds = statsFromDataStore.flowStatistics.duration.nanosecond.value.intValue; + + output.add(it); + } + } + } //TODO (main): Shell we send request to the switch? It will make async request to the switch. // Once plugin receive response, it will let adaptor know through onFlowStatisticsUpdate() @@ -180,35 +197,35 @@ class InventoryAndReadAdapter implements IPluginInReadService, } override readAllNodeConnector(Node node, boolean cached) { - - val ret = new ArrayList(); - val nodeRef = InstanceIdentifier.builder(Nodes) - .child(org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node, InventoryMapping.toNodeKey(node)) - .toInstance(); - - val provider = this.startChange(); - - val dsNode= provider.readConfigurationData(nodeRef) as org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node; - - if(dsNode != null){ - - for (dsNodeConnector : dsNode.nodeConnector){ - val nodeConnectorRef = InstanceIdentifier.builder(Nodes) - .child(org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node, InventoryMapping.toNodeKey(node)) - .child(NodeConnector, dsNodeConnector.key) - .toInstance(); - - val nodeConnectorFromDS = provider.readConfigurationData(nodeConnectorRef) as NodeConnector; - - if(nodeConnectorFromDS != null){ - val nodeConnectorStatsFromDs = nodeConnectorFromDS.getAugmentation(FlowCapableNodeConnectorStatisticsData) as FlowCapableNodeConnectorStatistics; - - ret.add(toNodeConnectorStatistics(nodeConnectorStatsFromDs.flowCapableNodeConnectorStatistics,dsNode.id,dsNodeConnector.id)); - } - } - } - - //TODO: Refer TODO (main) + + val ret = new ArrayList(); + val nodeRef = InstanceIdentifier.builder(Nodes) + .child(org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node, InventoryMapping.toNodeKey(node)) + .toInstance(); + + val provider = this.startChange(); + + val dsNode= provider.readConfigurationData(nodeRef) as org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node; + + if(dsNode != null){ + + for (dsNodeConnector : dsNode.nodeConnector){ + val nodeConnectorRef = InstanceIdentifier.builder(Nodes) + .child(org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node, InventoryMapping.toNodeKey(node)) + .child(NodeConnector, dsNodeConnector.key) + .toInstance(); + + val nodeConnectorFromDS = provider.readConfigurationData(nodeConnectorRef) as NodeConnector; + + if(nodeConnectorFromDS != null){ + val nodeConnectorStatsFromDs = nodeConnectorFromDS.getAugmentation(FlowCapableNodeConnectorStatisticsData) as FlowCapableNodeConnectorStatistics; + + ret.add(toNodeConnectorStatistics(nodeConnectorStatsFromDs.flowCapableNodeConnectorStatistics,dsNode.id,dsNodeConnector.id)); + } + } + } + + //TODO: Refer TODO (main) val input = new GetAllNodeConnectorsStatisticsInputBuilder(); input.setNode(node.toNodeRef); nodeConnectorStatisticsService.getAllNodeConnectorsStatistics(input.build()); @@ -216,23 +233,23 @@ class InventoryAndReadAdapter implements IPluginInReadService, } override readAllNodeTable(Node node, boolean cached) { - val ret = new ArrayList(); - - val dsFlowCapableNode= readFlowCapableNode(node.toNodeRef) - - if(dsFlowCapableNode != null){ - - for (table : dsFlowCapableNode.table){ - - val tableStats = table.getAugmentation(FlowTableStatisticsData); - - if(tableStats != null){ - ret.add(toNodeTableStatistics(tableStats.flowTableStatistics,table.id,node)); - } - } - } - - //TODO: Refer TODO (main) + val ret = new ArrayList(); + + val dsFlowCapableNode= readFlowCapableNode(node.toNodeRef) + + if(dsFlowCapableNode != null){ + + for (table : dsFlowCapableNode.table){ + + val tableStats = table.getAugmentation(FlowTableStatisticsData); + + if(tableStats != null){ + ret.add(toNodeTableStatistics(tableStats.flowTableStatistics,table.id,node)); + } + } + } + + //TODO: Refer TODO (main) val input = new GetFlowTablesStatisticsInputBuilder(); input.setNode(node.toNodeRef); flowTableStatisticsService.getFlowTablesStatistics(input.build); @@ -241,39 +258,39 @@ class InventoryAndReadAdapter implements IPluginInReadService, override readDescription(Node node, boolean cached) { return toNodeDescription(node.toNodeRef); - } + } override readFlow(Node node, Flow targetFlow, boolean cached) { - var FlowOnNode ret= null; - - val tableRef = InstanceIdentifier.builder(Nodes) - .child(org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node, InventoryMapping.toNodeKey(node)) - .augmentation(FlowCapableNode).child(Table, new TableKey(OPENFLOWV10_TABLE_ID)).toInstance(); - - val it = this.startChange(); - - val table= it.readConfigurationData(tableRef) as Table; - - if(table != null){ - LOG.trace("Number of flows installed in table 0 of node {} : {}",node,table.flow.size); - - for(mdsalFlow : table.flow){ - if(FromSalConversionsUtils.flowEquals(mdsalFlow, MDFlowMapping.toMDSalflow(targetFlow))){ - val statsFromDataStore = mdsalFlow.getAugmentation(FlowStatisticsData); - - if(statsFromDataStore != null){ - LOG.debug("Found matching flow in the data store flow table "); - val it = new FlowOnNode(targetFlow); - byteCount = statsFromDataStore.flowStatistics.byteCount.value.longValue; - packetCount = statsFromDataStore.flowStatistics.packetCount.value.longValue; - durationSeconds = statsFromDataStore.flowStatistics.duration.second.value.intValue; - durationNanoseconds = statsFromDataStore.flowStatistics.duration.nanosecond.value.intValue; - - ret = it; - } - } - } - } + var FlowOnNode ret= null; + + val tableRef = InstanceIdentifier.builder(Nodes) + .child(org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node, InventoryMapping.toNodeKey(node)) + .augmentation(FlowCapableNode).child(Table, new TableKey(OPENFLOWV10_TABLE_ID)).toInstance(); + + val it = this.startChange(); + + val table= it.readConfigurationData(tableRef) as Table; + + if(table != null){ + LOG.trace("Number of flows installed in table 0 of node {} : {}",node,table.flow.size); + + for(mdsalFlow : table.flow){ + if(FromSalConversionsUtils.flowEquals(mdsalFlow, MDFlowMapping.toMDSalflow(targetFlow))){ + val statsFromDataStore = mdsalFlow.getAugmentation(FlowStatisticsData); + + if(statsFromDataStore != null){ + LOG.debug("Found matching flow in the data store flow table "); + val it = new FlowOnNode(targetFlow); + byteCount = statsFromDataStore.flowStatistics.byteCount.value.longValue; + packetCount = statsFromDataStore.flowStatistics.packetCount.value.longValue; + durationSeconds = statsFromDataStore.flowStatistics.duration.second.value.intValue; + durationNanoseconds = statsFromDataStore.flowStatistics.duration.nanosecond.value.intValue; + + ret = it; + } + } + } + } //TODO: Refer TODO (main) val input = new GetFlowStatisticsFromFlowTableInputBuilder; @@ -282,30 +299,30 @@ class InventoryAndReadAdapter implements IPluginInReadService, flowStatisticsService.getFlowStatisticsFromFlowTable(input.build) return ret; - + } override readNodeConnector(org.opendaylight.controller.sal.core.NodeConnector connector, boolean cached) { - var NodeConnectorStatistics nodeConnectorStatistics = null; - - val nodeConnectorRef = InstanceIdentifier.builder(Nodes) - .child(org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node, InventoryMapping.toNodeKey(connector.node)) - .child(NodeConnector, InventoryMapping.toNodeConnectorKey(connector)) - .toInstance(); - val provider = this.startChange(); - - val nodeConnectorFromDS = provider.readConfigurationData(nodeConnectorRef) as NodeConnector; - - if(nodeConnectorFromDS != null){ - val nodeConnectorStatsFromDs = nodeConnectorFromDS.getAugmentation(FlowCapableNodeConnectorStatisticsData) as FlowCapableNodeConnectorStatistics; - if(nodeConnectorStatsFromDs != null) { - nodeConnectorStatistics = toNodeConnectorStatistics(nodeConnectorStatsFromDs.flowCapableNodeConnectorStatistics, - InventoryMapping.toNodeKey(connector.node).id, - InventoryMapping.toNodeConnectorKey(connector).id); - } - } - - //TODO: Refer TODO (main) + var NodeConnectorStatistics nodeConnectorStatistics = null; + + val nodeConnectorRef = InstanceIdentifier.builder(Nodes) + .child(org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node, InventoryMapping.toNodeKey(connector.node)) + .child(NodeConnector, InventoryMapping.toNodeConnectorKey(connector)) + .toInstance(); + val provider = this.startChange(); + + val nodeConnectorFromDS = provider.readConfigurationData(nodeConnectorRef) as NodeConnector; + + if(nodeConnectorFromDS != null){ + val nodeConnectorStatsFromDs = nodeConnectorFromDS.getAugmentation(FlowCapableNodeConnectorStatisticsData) as FlowCapableNodeConnectorStatistics; + if(nodeConnectorStatsFromDs != null) { + nodeConnectorStatistics = toNodeConnectorStatistics(nodeConnectorStatsFromDs.flowCapableNodeConnectorStatistics, + InventoryMapping.toNodeKey(connector.node).id, + InventoryMapping.toNodeConnectorKey(connector).id); + } + } + + //TODO: Refer TODO (main) val input = new GetNodeConnectorStatisticsInputBuilder(); input.setNode(connector.node.toNodeRef); input.setNodeConnectorId(InventoryMapping.toNodeConnectorKey(connector).id); @@ -314,25 +331,25 @@ class InventoryAndReadAdapter implements IPluginInReadService, } override readNodeTable(NodeTable nodeTable, boolean cached) { - var NodeTableStatistics nodeStats = null - - val tableRef = InstanceIdentifier.builder(Nodes) - .child(org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node, InventoryMapping.toNodeKey(nodeTable.node)) - .augmentation(FlowCapableNode).child(Table, new TableKey(nodeTable.ID as Short)).toInstance(); - - val it = this.startChange(); - - val table= it.readConfigurationData(tableRef) as Table; - - if(table != null){ - val tableStats = table.getAugmentation(FlowTableStatisticsData); - - if(tableStats != null){ - nodeStats = toNodeTableStatistics(tableStats.flowTableStatistics,table.id,nodeTable.node); - } - } - - //TODO: Refer TODO (main) + var NodeTableStatistics nodeStats = null + + val tableRef = InstanceIdentifier.builder(Nodes) + .child(org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node, InventoryMapping.toNodeKey(nodeTable.node)) + .augmentation(FlowCapableNode).child(Table, new TableKey(nodeTable.ID as Short)).toInstance(); + + val it = this.startChange(); + + val table= it.readConfigurationData(tableRef) as Table; + + if(table != null){ + val tableStats = table.getAugmentation(FlowTableStatisticsData); + + if(tableStats != null){ + nodeStats = toNodeTableStatistics(tableStats.flowTableStatistics,table.id,nodeTable.node); + } + } + + //TODO: Refer TODO (main) val input = new GetFlowTablesStatisticsInputBuilder(); input.setNode(nodeTable.node.toNodeRef); flowTableStatisticsService.getFlowTablesStatistics(input.build); @@ -341,19 +358,22 @@ class InventoryAndReadAdapter implements IPluginInReadService, } override onNodeConnectorRemoved(NodeConnectorRemoved update) { - // NOOP + // Never received } override onNodeRemoved(NodeRemoved notification) { val properties = Collections.emptySet(); + removeNodeConnectors(notification.nodeRef.value); + publishNodeUpdate(notification.nodeRef.toADNode, UpdateType.REMOVED, properties); } override onNodeConnectorUpdated(NodeConnectorUpdated update) { var updateType = UpdateType.CHANGED; - if ( this._dataService.readOperationalData(update.nodeConnectorRef.value as InstanceIdentifier) == null ){ + if(!isKnownNodeConnector(update.nodeConnectorRef.value)){ updateType = UpdateType.ADDED; + recordNodeConnector(update.nodeConnectorRef.value); } var nodeConnector = update.nodeConnectorRef.toADNodeConnector @@ -369,16 +389,16 @@ class InventoryAndReadAdapter implements IPluginInReadService, updateType = UpdateType.ADDED; } publishNodeUpdate(notification.nodeRef.toADNode, updateType, notification.toADNodeProperties); - - //Notify the listeners of IPluginOutReadService - + + //Notify the listeners of IPluginOutReadService + for (statsPublisher : statisticsPublisher){ - val nodeRef = InstanceIdentifier.builder(Nodes).child(org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node,new NodeKey(notification.id)).toInstance; + val nodeRef = InstanceIdentifier.builder(Nodes).child(org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node,new NodeKey(notification.id)).toInstance; val description = notification.nodeRef.toNodeDescription if(description != null) { - statsPublisher.descriptionStatisticsUpdated(nodeRef.toADNode,description); - } - } + statsPublisher.descriptionStatisticsUpdated(nodeRef.toADNode,description); + } + } } override getNodeProps() { @@ -461,50 +481,50 @@ class InventoryAndReadAdapter implements IPluginInReadService, private def toNodeConnectorStatistics( org.opendaylight.yang.gen.v1.urn.opendaylight.model.statistics.types.rev130925.NodeConnectorStatistics nodeConnectorStatistics, NodeId nodeId, NodeConnectorId nodeConnectorId) { - - val it = new NodeConnectorStatistics(); - - receivePacketCount = nodeConnectorStatistics.packets.received.longValue; - transmitPacketCount = nodeConnectorStatistics.packets.transmitted.longValue; - - receiveByteCount = nodeConnectorStatistics.bytes.received.longValue; - transmitByteCount = nodeConnectorStatistics.bytes.transmitted.longValue; - - receiveDropCount = nodeConnectorStatistics.receiveDrops.longValue; - transmitDropCount = nodeConnectorStatistics.transmitDrops.longValue; - - receiveErrorCount = nodeConnectorStatistics.receiveErrors.longValue; - transmitErrorCount = nodeConnectorStatistics.transmitErrors.longValue; - - receiveFrameErrorCount = nodeConnectorStatistics.receiveFrameError.longValue; - receiveOverRunErrorCount = nodeConnectorStatistics.receiveOverRunError.longValue; - receiveCRCErrorCount = nodeConnectorStatistics.receiveCrcError.longValue; - collisionCount = nodeConnectorStatistics.collisionCount.longValue; - - val nodeConnectorRef = InstanceIdentifier.builder(Nodes) - .child(org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node,new NodeKey(nodeId)) - .child(NodeConnector,new NodeConnectorKey(nodeConnectorId)).toInstance; - - nodeConnector = NodeMapping.toADNodeConnector(new NodeConnectorRef(nodeConnectorRef)); - - return it; - } - - private def toNodeTableStatistics( - FlowTableStatistics tableStats, - Short tableId,Node node){ - var it = new NodeTableStatistics(); - - activeCount = tableStats.activeFlows.value.intValue; - lookupCount = tableStats.packetsLookedUp.value.intValue; - matchedCount = tableStats.packetsMatched.value.intValue; - name = tableId.toString; - nodeTable = new NodeTable(NodeMapping.MD_SAL_TYPE,tableId,node); - return it; - } - - private def toNodeDescription(NodeRef nodeRef){ - val capableNode = readFlowCapableNode(nodeRef); + + val it = new NodeConnectorStatistics(); + + receivePacketCount = nodeConnectorStatistics.packets.received.longValue; + transmitPacketCount = nodeConnectorStatistics.packets.transmitted.longValue; + + receiveByteCount = nodeConnectorStatistics.bytes.received.longValue; + transmitByteCount = nodeConnectorStatistics.bytes.transmitted.longValue; + + receiveDropCount = nodeConnectorStatistics.receiveDrops.longValue; + transmitDropCount = nodeConnectorStatistics.transmitDrops.longValue; + + receiveErrorCount = nodeConnectorStatistics.receiveErrors.longValue; + transmitErrorCount = nodeConnectorStatistics.transmitErrors.longValue; + + receiveFrameErrorCount = nodeConnectorStatistics.receiveFrameError.longValue; + receiveOverRunErrorCount = nodeConnectorStatistics.receiveOverRunError.longValue; + receiveCRCErrorCount = nodeConnectorStatistics.receiveCrcError.longValue; + collisionCount = nodeConnectorStatistics.collisionCount.longValue; + + val nodeConnectorRef = InstanceIdentifier.builder(Nodes) + .child(org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node,new NodeKey(nodeId)) + .child(NodeConnector,new NodeConnectorKey(nodeConnectorId)).toInstance; + + nodeConnector = NodeMapping.toADNodeConnector(new NodeConnectorRef(nodeConnectorRef)); + + return it; + } + + private def toNodeTableStatistics( + FlowTableStatistics tableStats, + Short tableId,Node node){ + var it = new NodeTableStatistics(); + + activeCount = tableStats.activeFlows.value.intValue; + lookupCount = tableStats.packetsLookedUp.value.intValue; + matchedCount = tableStats.packetsMatched.value.intValue; + name = tableId.toString; + nodeTable = new NodeTable(NodeMapping.MD_SAL_TYPE,tableId,node); + return it; + } + + private def toNodeDescription(NodeRef nodeRef){ + val capableNode = readFlowCapableNode(nodeRef); if(capableNode !=null) { val it = new NodeDescription() manufacturer = capableNode.manufacturer @@ -515,101 +535,148 @@ class InventoryAndReadAdapter implements IPluginInReadService, return it; } return null; - } + } def Edge toADEdge(Link link) { new Edge(link.source.toADNodeConnector,link.destination.toADNodeConnector) } - - /* - * OpendaylightFlowStatisticsListener interface implementation - */ - override onAggregateFlowStatisticsUpdate(AggregateFlowStatisticsUpdate notification) { + + /* + * OpendaylightFlowStatisticsListener interface implementation + */ + override onAggregateFlowStatisticsUpdate(AggregateFlowStatisticsUpdate notification) { //Ignoring this notification as there does not seem to be a way to bubble this up to AD-SAL - } - - override onFlowsStatisticsUpdate(FlowsStatisticsUpdate notification) { - - val adsalFlowsStatistics = new ArrayList(); - val nodeRef = InstanceIdentifier.builder(Nodes).child(org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node,new NodeKey(notification.id)).toInstance; - - for(flowStats : notification.flowAndStatisticsMapList){ - if(flowStats.tableId == 0) - adsalFlowsStatistics.add(toFlowOnNode(flowStats,nodeRef.toADNode)); - } - - for (statsPublisher : statisticsPublisher){ - statsPublisher.nodeFlowStatisticsUpdated(nodeRef.toADNode,adsalFlowsStatistics); - } - - } - /* - * OpendaylightFlowTableStatisticsListener interface implementation - */ - override onFlowTableStatisticsUpdate(FlowTableStatisticsUpdate notification) { - var adsalFlowTableStatistics = new ArrayList(); - - for(stats : notification.flowTableAndStatisticsMap){ - if (stats.tableId.value == 0){ - val it = new NodeTableStatistics(); - activeCount = stats.activeFlows.value.intValue; - lookupCount = stats.packetsLookedUp.value.longValue; - matchedCount = stats.packetsMatched.value.longValue; - - adsalFlowTableStatistics.add(it); - } - } - for (statsPublisher : statisticsPublisher){ - val nodeRef = InstanceIdentifier.builder(Nodes).child(org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node,new NodeKey(notification.id)).toInstance; - statsPublisher.nodeTableStatisticsUpdated(nodeRef.toADNode,adsalFlowTableStatistics); - } - } - - /* - * OpendaylightPortStatisticsUpdate interface implementation - */ - override onNodeConnectorStatisticsUpdate(NodeConnectorStatisticsUpdate notification) { - - val adsalPortStatistics = new ArrayList(); - - for(nodeConnectorStatistics : notification.nodeConnectorStatisticsAndPortNumberMap){ - adsalPortStatistics.add(toNodeConnectorStatistics(nodeConnectorStatistics,notification.id,nodeConnectorStatistics.nodeConnectorId)); - } - - for (statsPublisher : statisticsPublisher){ - val nodeRef = InstanceIdentifier.builder(Nodes).child(org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node,new NodeKey(notification.id)).toInstance; - statsPublisher.nodeConnectorStatisticsUpdated(nodeRef.toADNode,adsalPortStatistics); - } - - } - - private static def toFlowOnNode (FlowAndStatisticsMapList flowAndStatsMap,Node node){ - - val it = new FlowOnNode(ToSalConversionsUtils.toFlow(flowAndStatsMap,node)); - - byteCount = flowAndStatsMap.byteCount.value.longValue; - packetCount = flowAndStatsMap.packetCount.value.longValue; - durationSeconds = flowAndStatsMap.duration.second.value.intValue; - durationNanoseconds = flowAndStatsMap.duration.nanosecond.value.intValue; - - return it; - } - - override getConfiguredNotConnectedNodes() { + } + + override onFlowsStatisticsUpdate(FlowsStatisticsUpdate notification) { + + val adsalFlowsStatistics = new ArrayList(); + val nodeRef = InstanceIdentifier.builder(Nodes).child(org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node,new NodeKey(notification.id)).toInstance; + + for(flowStats : notification.flowAndStatisticsMapList){ + if(flowStats.tableId == 0) + adsalFlowsStatistics.add(toFlowOnNode(flowStats,nodeRef.toADNode)); + } + + for (statsPublisher : statisticsPublisher){ + statsPublisher.nodeFlowStatisticsUpdated(nodeRef.toADNode,adsalFlowsStatistics); + } + + } + /* + * OpendaylightFlowTableStatisticsListener interface implementation + */ + override onFlowTableStatisticsUpdate(FlowTableStatisticsUpdate notification) { + var adsalFlowTableStatistics = new ArrayList(); + + for(stats : notification.flowTableAndStatisticsMap){ + if (stats.tableId.value == 0){ + val it = new NodeTableStatistics(); + activeCount = stats.activeFlows.value.intValue; + lookupCount = stats.packetsLookedUp.value.longValue; + matchedCount = stats.packetsMatched.value.longValue; + + adsalFlowTableStatistics.add(it); + } + } + for (statsPublisher : statisticsPublisher){ + val nodeRef = InstanceIdentifier.builder(Nodes).child(org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node,new NodeKey(notification.id)).toInstance; + statsPublisher.nodeTableStatisticsUpdated(nodeRef.toADNode,adsalFlowTableStatistics); + } + } + + /* + * OpendaylightPortStatisticsUpdate interface implementation + */ + override onNodeConnectorStatisticsUpdate(NodeConnectorStatisticsUpdate notification) { + + val adsalPortStatistics = new ArrayList(); + + for(nodeConnectorStatistics : notification.nodeConnectorStatisticsAndPortNumberMap){ + adsalPortStatistics.add(toNodeConnectorStatistics(nodeConnectorStatistics,notification.id,nodeConnectorStatistics.nodeConnectorId)); + } + + for (statsPublisher : statisticsPublisher){ + val nodeRef = InstanceIdentifier.builder(Nodes).child(org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node,new NodeKey(notification.id)).toInstance; + statsPublisher.nodeConnectorStatisticsUpdated(nodeRef.toADNode,adsalPortStatistics); + } + + } + + private static def toFlowOnNode (FlowAndStatisticsMapList flowAndStatsMap,Node node){ + + val it = new FlowOnNode(ToSalConversionsUtils.toFlow(flowAndStatsMap,node)); + + byteCount = flowAndStatsMap.byteCount.value.longValue; + packetCount = flowAndStatsMap.packetCount.value.longValue; + durationSeconds = flowAndStatsMap.duration.second.value.intValue; + durationNanoseconds = flowAndStatsMap.duration.nanosecond.value.intValue; + + return it; + } + + override getConfiguredNotConnectedNodes() { return Collections.emptySet(); - } + } + + + private def publishNodeUpdate(Node node, UpdateType updateType, Set properties){ + for( publisher : inventoryPublisher){ + publisher.updateNode(node, updateType, properties); + } + } + + private def publishNodeConnectorUpdate(org.opendaylight.controller.sal.core.NodeConnector nodeConnector, UpdateType updateType, Set properties){ + for( publisher : inventoryPublisher){ + publisher.updateNodeConnector(nodeConnector, updateType, properties); + } + } + + private def isKnownNodeConnector(InstanceIdentifier nodeConnectorIdentifier){ + if(nodeConnectorIdentifier.path.size() < 3) { + return false; + } + val nodePath = nodeConnectorIdentifier.path.get(1); + val nodeConnectorPath = nodeConnectorIdentifier.getPath().get(2); - private def publishNodeUpdate(Node node, UpdateType updateType, Set properties){ - for( publisher : inventoryPublisher){ - publisher.updateNode(node, updateType, properties); - } - } + val nodeConnectors = nodeToNodeConnectorsMap.get(nodePath); - private def publishNodeConnectorUpdate(org.opendaylight.controller.sal.core.NodeConnector nodeConnector, UpdateType updateType, Set properties){ - for( publisher : inventoryPublisher){ - publisher.updateNodeConnector(nodeConnector, updateType, properties); - } - } + if(nodeConnectors == null){ + return false; + } + return nodeConnectors.contains(nodeConnectorPath); + } + + + private def recordNodeConnector(InstanceIdentifier nodeConnectorIdentifier){ + if(nodeConnectorIdentifier.path.size() < 3) { + return false; + } + + val nodePath = nodeConnectorIdentifier.path.get(1); + val nodeConnectorPath = nodeConnectorIdentifier.getPath().get(2); + + nodeToNodeConnectorsLock.lock(); + + try { + var nodeConnectors = nodeToNodeConnectorsMap.get(nodePath); + + if(nodeConnectors == null){ + nodeConnectors = new ArrayList(); + nodeToNodeConnectorsMap.put(nodePath, nodeConnectors); + } + + nodeConnectors.add(nodeConnectorPath); + } finally { + nodeToNodeConnectorsLock.unlock(); + } + } + + private def removeNodeConnectors(InstanceIdentifier nodeIdentifier){ + val nodePath = nodeIdentifier.path.get(1); + + nodeToNodeConnectorsMap.remove(nodePath); + } } diff --git a/opendaylight/md-sal/compatibility/sal-compatibility/src/main/java/org/opendaylight/controller/sal/compatibility/InventoryNotificationProvider.java b/opendaylight/md-sal/compatibility/sal-compatibility/src/main/java/org/opendaylight/controller/sal/compatibility/InventoryNotificationProvider.java new file mode 100644 index 0000000000..23a98ff39a --- /dev/null +++ b/opendaylight/md-sal/compatibility/sal-compatibility/src/main/java/org/opendaylight/controller/sal/compatibility/InventoryNotificationProvider.java @@ -0,0 +1,59 @@ +package org.opendaylight.controller.sal.compatibility; + +import org.opendaylight.controller.sal.binding.api.data.DataChangeListener; +import org.opendaylight.controller.sal.binding.api.data.DataProviderService; +import org.opendaylight.controller.sal.inventory.IPluginOutInventoryService; +import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.Nodes; +import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnector; +import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node; +import org.opendaylight.yangtools.concepts.ListenerRegistration; +import org.opendaylight.yangtools.yang.binding.InstanceIdentifier; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.List; + +public class InventoryNotificationProvider implements AutoCloseable{ + + private ListenerRegistration nodeConnectorDataChangeListenerRegistration; + + private NodeConnectorDataChangeListener nodeConnectorDataChangeListener; + + private DataProviderService dataProviderService; + + private List inventoryPublisher; + + private final static Logger LOG = LoggerFactory.getLogger(NodeConnectorDataChangeListener.class); + + public void start(){ + + LOG.info("InventoryNotificationProvider started"); + + if(dataProviderService != null + && inventoryPublisher!= null){ + + if(nodeConnectorDataChangeListener == null){ + InstanceIdentifier nodeConnectorPath = InstanceIdentifier.builder(Nodes.class).child(Node.class).child(NodeConnector.class).build(); + nodeConnectorDataChangeListener = new NodeConnectorDataChangeListener(); + nodeConnectorDataChangeListener.setInventoryPublisher(inventoryPublisher); + nodeConnectorDataChangeListenerRegistration = dataProviderService.registerDataChangeListener(nodeConnectorPath, nodeConnectorDataChangeListener); + } + + } + } + + @Override + public void close() throws Exception { + if(nodeConnectorDataChangeListenerRegistration != null){ + nodeConnectorDataChangeListenerRegistration.close(); + } + } + + public void setDataProviderService(DataProviderService dataProviderService) { + this.dataProviderService = dataProviderService; + } + + public void setInventoryPublisher(List inventoryPublisher) { + this.inventoryPublisher = inventoryPublisher; + } +} diff --git a/opendaylight/md-sal/compatibility/sal-compatibility/src/main/java/org/opendaylight/controller/sal/compatibility/NodeConnectorDataChangeListener.java b/opendaylight/md-sal/compatibility/sal-compatibility/src/main/java/org/opendaylight/controller/sal/compatibility/NodeConnectorDataChangeListener.java new file mode 100644 index 0000000000..eebba74244 --- /dev/null +++ b/opendaylight/md-sal/compatibility/sal-compatibility/src/main/java/org/opendaylight/controller/sal/compatibility/NodeConnectorDataChangeListener.java @@ -0,0 +1,77 @@ +package org.opendaylight.controller.sal.compatibility; + +import org.opendaylight.controller.md.sal.common.api.data.DataChangeEvent; +import org.opendaylight.controller.sal.binding.api.data.DataChangeListener; +import org.opendaylight.controller.sal.core.ConstructionException; +import org.opendaylight.controller.sal.core.NodeConnector; +import org.opendaylight.controller.sal.core.Property; +import org.opendaylight.controller.sal.core.UpdateType; +import org.opendaylight.controller.sal.inventory.IPluginOutInventoryService; +import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeConnectorRef; +import org.opendaylight.yangtools.yang.binding.DataObject; +import org.opendaylight.yangtools.yang.binding.InstanceIdentifier; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +// org.opendaylight.controller.sal.compatibility.NodeConnectorDataChangeListener +public class NodeConnectorDataChangeListener implements DataChangeListener{ + private final static Logger LOG = LoggerFactory.getLogger(NodeConnectorDataChangeListener.class); + + private List inventoryPublisher; + + public List getInventoryPublisher() { + return this.inventoryPublisher; + } + + public void setInventoryPublisher(final List inventoryPublisher) { + this.inventoryPublisher = inventoryPublisher; + } + + @Override + public void onDataChanged(DataChangeEvent, DataObject> change) { + final Map,DataObject> createdOperationalData = change.getCreatedOperationalData(); + final Map,DataObject> updatedOperationalData = change.getUpdatedOperationalData(); + + final Set,DataObject>> createdEntries = createdOperationalData.entrySet(); + final Set,DataObject>> updatedEntries = new HashSet<>(); + + updatedEntries.addAll(updatedOperationalData.entrySet()); + updatedEntries.removeAll(createdEntries); + + for(final Map.Entry,DataObject> entry : createdEntries){ + publishNodeConnectorUpdate(entry, UpdateType.ADDED); + } + + for(final Map.Entry,DataObject> entry : updatedEntries){ + publishNodeConnectorUpdate(entry, UpdateType.CHANGED); + } + } + + private void publishNodeConnectorUpdate(final Map.Entry,DataObject> entry, final UpdateType updateType) { + if (entry.getKey().getTargetType().equals(org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnector.class)) { + NodeConnectorRef nodeConnectorRef = new NodeConnectorRef(entry.getKey()); + NodeConnector nodeConnector = null; + try { + nodeConnector = NodeMapping.toADNodeConnector(nodeConnectorRef); + } catch (ConstructionException e) { + e.printStackTrace(); + } + HashSet _aDNodeConnectorProperties = NodeMapping.toADNodeConnectorProperties((org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnector) entry.getValue()); + this.publishNodeConnectorUpdate(nodeConnector, updateType, _aDNodeConnectorProperties); + } + } + + private void publishNodeConnectorUpdate(final NodeConnector nodeConnector, final UpdateType updateType, final Set properties) { + LOG.debug("Publishing NodeConnector " + updateType.toString() + " nodeConnector Id = " + nodeConnector.getNodeConnectorIdAsString()); + + List _inventoryPublisher = getInventoryPublisher(); + for (final IPluginOutInventoryService publisher : _inventoryPublisher) { + publisher.updateNodeConnector(nodeConnector, updateType, properties); + } + } +} diff --git a/opendaylight/md-sal/inventory-manager/src/main/java/org/opendaylight/controller/md/inventory/manager/FlowCapableInventoryProvider.xtend b/opendaylight/md-sal/inventory-manager/src/main/java/org/opendaylight/controller/md/inventory/manager/FlowCapableInventoryProvider.xtend index 1a66b3ba16..43f48a50e5 100644 --- a/opendaylight/md-sal/inventory-manager/src/main/java/org/opendaylight/controller/md/inventory/manager/FlowCapableInventoryProvider.xtend +++ b/opendaylight/md-sal/inventory-manager/src/main/java/org/opendaylight/controller/md/inventory/manager/FlowCapableInventoryProvider.xtend @@ -64,6 +64,8 @@ class FlowCapableInventoryProvider implements AutoCloseable { class NodeChangeCommiter implements OpendaylightInventoryListener { + static val LOG = LoggerFactory.getLogger(NodeChangeCommiter); + @Property val FlowCapableInventoryProvider manager; @@ -76,6 +78,9 @@ class NodeChangeCommiter implements OpendaylightInventoryListener { // Check path val it = manager.startChange() + + LOG.debug("removing node connector : " + ref.value.toString()); + removeOperationalData(ref.value as InstanceIdentifier); commit() } @@ -93,6 +98,8 @@ class NodeChangeCommiter implements OpendaylightInventoryListener { data.addAugmentation(FlowCapableNodeConnector, augment) } + LOG.debug("updating node connector : " + ref.value.toString()); + putOperationalData(ref.value as InstanceIdentifier, data.build()); commit() } @@ -101,6 +108,8 @@ class NodeChangeCommiter implements OpendaylightInventoryListener { val ref = node.nodeRef; val it = manager.startChange() + LOG.debug("removing node : " + ref.value.toString()); + removeOperationalData(ref.value as InstanceIdentifier); commit() } @@ -117,6 +126,8 @@ class NodeChangeCommiter implements OpendaylightInventoryListener { data.addAugmentation(FlowCapableNode, augment) } + LOG.debug("updating node : " + ref.value.toString()); + putOperationalData(ref.value as InstanceIdentifier, data.build()) commit() } -- 2.36.6