/** * Copyright (c) 2014 Cisco Systems, Inc. and others. All rights reserved. * * This program and the accompanying materials are made available under the * terms of the Eclipse Public License v1.0 which accompanies this distribution, * and is available at http://www.eclipse.org/legal/epl-v10.html */ package org.opendaylight.openflowplugin.applications.frm.impl; import java.util.*; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.atomic.AtomicInteger; import com.google.common.base.Optional; import com.google.common.base.Preconditions; import com.google.common.collect.ImmutableList; import com.google.common.collect.Iterables; import com.google.common.collect.Lists; import com.google.common.util.concurrent.CheckedFuture; import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; import java.util.concurrent.Callable; import org.opendaylight.controller.md.sal.binding.api.*; import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType; import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException; import org.opendaylight.openflowplugin.applications.frm.FlowNodeReconciliation; import org.opendaylight.openflowplugin.applications.frm.ForwardingRulesManager; import org.opendaylight.openflowplugin.common.wait.SimpleTaskRetryLooper; import org.opendaylight.yang.gen.v1.urn.opendaylight.action.types.rev131112.action.action.GroupActionCase; import org.opendaylight.yang.gen.v1.urn.opendaylight.action.types.rev131112.action.action.OutputActionCase; import org.opendaylight.yang.gen.v1.urn.opendaylight.action.types.rev131112.action.list.Action; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowId; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.meters.Meter; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.meters.MeterBuilder; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.meters.MeterKey; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.meters.StaleMeter; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.meters.StaleMeterKey; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.TableKey; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowBuilder; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowKey; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.StaleFlow; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.StaleFlowKey; import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.GroupId; import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.group.buckets.Bucket; import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.groups.Group; import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.groups.GroupBuilder; import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.groups.GroupKey; import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.groups.StaleGroup; import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.groups.StaleGroupKey; import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.Nodes; import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node; import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey; import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.types.rev130918.MeterId; import org.opendaylight.yang.gen.v1.urn.opendaylight.table.types.rev131026.table.features.TableFeatures; import org.opendaylight.yang.gen.v1.urn.opendaylight.table.types.rev131026.table.features.TableFeaturesKey; import org.opendaylight.yangtools.concepts.ListenerRegistration; import org.opendaylight.yangtools.yang.binding.InstanceIdentifier; import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import javax.annotation.Nonnull; /** * forwardingrules-manager * org.opendaylight.openflowplugin.applications.frm * * FlowNode Reconciliation Listener * Reconciliation for a new FlowNode * * @author Vaclav Demcak * * Created: Jun 13, 2014 */ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { private static final Logger LOG = LoggerFactory.getLogger(FlowNodeReconciliationImpl.class); private final DataBroker dataBroker; private final ForwardingRulesManager provider; public static final String SEPARATOR = ":"; private ListenerRegistration listenerRegistration; private final int THREAD_POOL_SIZE = 4; ExecutorService executor = Executors.newFixedThreadPool(THREAD_POOL_SIZE); private static final InstanceIdentifier II_TO_FLOW_CAPABLE_NODE = InstanceIdentifier.builder(Nodes.class) .child(Node.class) .augmentation(FlowCapableNode.class) .build(); public FlowNodeReconciliationImpl (final ForwardingRulesManager manager, final DataBroker db) { this.provider = Preconditions.checkNotNull(manager, "ForwardingRulesManager can not be null!"); dataBroker = Preconditions.checkNotNull(db, "DataBroker can not be null!"); /* Build Path */ final InstanceIdentifier flowNodeWildCardIdentifier = InstanceIdentifier.create(Nodes.class) .child(Node.class).augmentation(FlowCapableNode.class); final DataTreeIdentifier treeId = new DataTreeIdentifier<>(LogicalDatastoreType.OPERATIONAL, flowNodeWildCardIdentifier); try { SimpleTaskRetryLooper looper = new SimpleTaskRetryLooper(ForwardingRulesManagerImpl.STARTUP_LOOP_TICK, ForwardingRulesManagerImpl.STARTUP_LOOP_MAX_RETRIES); listenerRegistration = looper.loopUntilNoException(new Callable>() { @Override public ListenerRegistration call() throws Exception { return dataBroker.registerDataTreeChangeListener(treeId, FlowNodeReconciliationImpl.this); } }); } catch (Exception e) { LOG.warn("data listener registration failed: {}", e.getMessage()); LOG.debug("data listener registration failed.. ", e); throw new IllegalStateException("FlowNodeReconciliation startup fail! System needs restart.", e); } } @Override public void close() { if (listenerRegistration != null) { try { listenerRegistration.close(); } catch (Exception e) { LOG.warn("Error by stop FRM FlowNodeReconilListener: {}", e.getMessage()); LOG.debug("Error by stop FRM FlowNodeReconilListener..", e); } listenerRegistration = null; } } @Override public void onDataTreeChanged(@Nonnull Collection> changes) { Preconditions.checkNotNull(changes, "Changes may not be null!"); for (DataTreeModification change : changes) { final InstanceIdentifier key = change.getRootPath().getRootIdentifier(); final DataObjectModification mod = change.getRootNode(); final InstanceIdentifier nodeIdent = key.firstIdentifierOf(FlowCapableNode.class); switch (mod.getModificationType()) { case DELETE: if (mod.getDataAfter() == null) { remove(key, mod.getDataBefore(), nodeIdent); } break; case SUBTREE_MODIFIED: //NO-OP since we donot need to reconciliate on Node-updated break; case WRITE: if (mod.getDataBefore() == null) { add(key, mod.getDataAfter(), nodeIdent); } break; default: throw new IllegalArgumentException("Unhandled modification type " + mod.getModificationType()); } } } public void remove(InstanceIdentifier identifier, FlowCapableNode del, InstanceIdentifier nodeIdent) { if(compareInstanceIdentifierTail(identifier,II_TO_FLOW_CAPABLE_NODE)){ if (LOG.isDebugEnabled()) { LOG.debug("Node removed: {}",nodeIdent.firstKeyOf(Node.class).getId().getValue()); } if ( ! nodeIdent.isWildcarded()) { flowNodeDisconnected(nodeIdent); } } } public void add(InstanceIdentifier identifier, FlowCapableNode add, InstanceIdentifier nodeIdent) { if(compareInstanceIdentifierTail(identifier,II_TO_FLOW_CAPABLE_NODE)){ if (LOG.isDebugEnabled()) { LOG.debug("Node added: {}",nodeIdent.firstKeyOf(Node.class).getId().getValue()); } if ( ! nodeIdent.isWildcarded()) { flowNodeConnected(nodeIdent); } } } @Override public void flowNodeDisconnected(InstanceIdentifier disconnectedNode) { provider.unregistrateNode(disconnectedNode); } @Override public void flowNodeConnected(InstanceIdentifier connectedNode) { flowNodeConnected(connectedNode, false); } private void flowNodeConnected(InstanceIdentifier connectedNode, boolean force) { if (force || !provider.isNodeActive(connectedNode)) { provider.registrateNewNode(connectedNode); if(!provider.isNodeOwner(connectedNode)) { return; } if (provider.getConfiguration().isStaleMarkingEnabled()) { LOG.info("Stale-Marking is ENABLED and proceeding with deletion of stale-marked entities on switch {}", connectedNode.toString()); reconciliationPreProcess(connectedNode); } ReconciliationTask reconciliationTask = new ReconciliationTask(connectedNode); executor.execute(reconciliationTask); } } private class ReconciliationTask implements Runnable { InstanceIdentifier nodeIdentity; public ReconciliationTask(final InstanceIdentifier nodeIdent) { nodeIdentity = nodeIdent; } @Override public void run() { String sNode = nodeIdentity.firstKeyOf(Node.class, NodeKey.class).getId().getValue(); long nDpId = getDpnIdFromNodeName(sNode); ReadOnlyTransaction trans = provider.getReadTranaction(); Optional flowNode = Optional.absent(); AtomicInteger counter = new AtomicInteger(); //initialize the counter counter.set(0); try { flowNode = trans.read(LogicalDatastoreType.CONFIGURATION, nodeIdentity).get(); } catch (Exception e) { LOG.error("Fail with read Config/DS for Node {} !", nodeIdentity, e); } if (flowNode.isPresent()) { /* Tables - have to be pushed before groups */ // CHECK if while pusing the update, updateTableInput can be null to emulate a table add List tableList = flowNode.get().getTableFeatures() != null ? flowNode.get().getTableFeatures() : Collections.emptyList(); for (TableFeatures tableFeaturesItem : tableList) { TableFeaturesKey tableKey = tableFeaturesItem.getKey(); KeyedInstanceIdentifier tableFeaturesII = nodeIdentity.child(TableFeatures.class, new TableFeaturesKey(tableKey.getTableId())); provider.getTableFeaturesCommiter().update(tableFeaturesII, tableFeaturesItem, null, nodeIdentity); } /* Groups - have to be first */ List groups = flowNode.get().getGroup() != null ? flowNode.get().getGroup() : Collections.emptyList(); List toBeInstalledGroups = new ArrayList<>(); toBeInstalledGroups.addAll(groups); List alreadyInstalledGroupids = new ArrayList<>(); //new list for suspected groups pointing to ports .. when the ports come up late List suspectedGroups = new ArrayList<>(); while ((!(toBeInstalledGroups.isEmpty()) || !(suspectedGroups.isEmpty())) && (counter.get() <= provider.getConfiguration().getReconciliationRetryCount())) { //also check if the counter has not crossed the threshold if (toBeInstalledGroups.isEmpty() && !suspectedGroups.isEmpty()) { LOG.error("These Groups are pointing to node-connectors that are not up yet {}", suspectedGroups.toString()); toBeInstalledGroups.addAll(suspectedGroups); break; } ListIterator iterator = toBeInstalledGroups.listIterator(); while (iterator.hasNext()) { Group group = iterator.next(); boolean okToInstall = true; for (Bucket bucket : group.getBuckets().getBucket()) { for (Action action : bucket.getAction()) { //chained-port if (action.getAction().getImplementedInterface().getName() .equals("org.opendaylight.yang.gen.v1.urn.opendaylight.action.types.rev131112.action.action.OutputActionCase")) { String nodeConnectorUri = ((OutputActionCase) (action.getAction())) .getOutputAction().getOutputNodeConnector().getValue(); LOG.warn("Installing the group for node connector {}", nodeConnectorUri); //check if the nodeconnector is there in the multimap boolean isPresent = provider.getFlowNodeConnectorInventoryTranslatorImpl() .isNodeConnectorUpdated(nDpId, nodeConnectorUri); //if yes set okToInstall = true if (isPresent) { break; }//else put it in a different list and still set okToInstall = true else { suspectedGroups.add(group); LOG.error("Not yet received the node-connector updated for {} " + "for the group with id {}", nodeConnectorUri, group.getGroupId().toString()); break; } } //chained groups else if (action.getAction().getImplementedInterface().getName() .equals("org.opendaylight.yang.gen.v1.urn.opendaylight.action.types.rev131112.action.action.GroupActionCase")) { Long groupId = ((GroupActionCase) (action.getAction())).getGroupAction().getGroupId(); if (!alreadyInstalledGroupids.contains(groupId)) { okToInstall = false; break; } } } if (!okToInstall) { //increment retry counter value counter.incrementAndGet(); break; } } if (okToInstall) { final KeyedInstanceIdentifier groupIdent = nodeIdentity.child(Group.class, group.getKey()); provider.getGroupCommiter().add(groupIdent, group, nodeIdentity); alreadyInstalledGroupids.add(group.getGroupId().getValue()); iterator.remove(); // resetting the counter to zero counter.set(0); } } } /* installation of suspected groups*/ if (!toBeInstalledGroups.isEmpty()) { for (Group group : toBeInstalledGroups) { LOG.error("Installing the group {} finally although the port is not up after checking for {} times " , group.getGroupId().toString(), provider.getConfiguration().getReconciliationRetryCount()); final KeyedInstanceIdentifier groupIdent = nodeIdentity.child(Group.class, group.getKey()); provider.getGroupCommiter().add(groupIdent, group, nodeIdentity); } } /* Meters */ List meters = flowNode.get().getMeter() != null ? flowNode.get().getMeter() : Collections.emptyList(); for (Meter meter : meters) { final KeyedInstanceIdentifier meterIdent = nodeIdentity.child(Meter.class, meter.getKey()); provider.getMeterCommiter().add(meterIdent, meter, nodeIdentity); } /* Flows */ List tables = flowNode.get().getTable() != null ? flowNode.get().getTable() : Collections.
emptyList(); for (Table table : tables) { final KeyedInstanceIdentifier tableIdent = nodeIdentity.child(Table.class, table.getKey()); List flows = table.getFlow() != null ? table.getFlow() : Collections.emptyList(); for (Flow flow : flows) { final KeyedInstanceIdentifier flowIdent = tableIdent.child(Flow.class, flow.getKey()); provider.getFlowCommiter().add(flowIdent, flow, nodeIdentity); } } } /* clean transaction */ trans.close(); } } private long getDpnIdFromNodeName(String nodeName) { String dpId = nodeName.substring(nodeName.lastIndexOf(SEPARATOR) + 1); return Long.parseLong(dpId); } private void reconciliationPreProcess(final InstanceIdentifier nodeIdent) { List> staleFlowsToBeBulkDeleted = Lists.newArrayList(); List> staleGroupsToBeBulkDeleted = Lists.newArrayList(); List> staleMetersToBeBulkDeleted = Lists.newArrayList(); ReadOnlyTransaction trans = provider.getReadTranaction(); Optional flowNode = Optional.absent(); try { flowNode = trans.read(LogicalDatastoreType.CONFIGURATION, nodeIdent).get(); } catch (Exception e) { LOG.error("Reconciliation Pre-Processing Fail with read Config/DS for Node {} !", nodeIdent, e); } if (flowNode.isPresent()) { LOG.debug("Proceeding with deletion of stale-marked Flows on switch {} using Openflow interface", nodeIdent.toString()); /* Stale-Flows - Stale-marked Flows have to be removed first for safety */ List
tables = flowNode.get().getTable() != null ? flowNode.get().getTable() : Collections.
emptyList(); for (Table table : tables) { final KeyedInstanceIdentifier tableIdent = nodeIdent.child(Table.class, table.getKey()); List staleFlows = table.getStaleFlow() != null ? table.getStaleFlow() : Collections. emptyList(); for (StaleFlow staleFlow : staleFlows) { FlowBuilder flowBuilder = new FlowBuilder(staleFlow); Flow toBeDeletedFlow = flowBuilder.setId(staleFlow.getId()).build(); final KeyedInstanceIdentifier flowIdent = tableIdent.child(Flow.class, toBeDeletedFlow.getKey()); this.provider.getFlowCommiter().remove(flowIdent, toBeDeletedFlow, nodeIdent); staleFlowsToBeBulkDeleted.add(getStaleFlowInstanceIdentifier(staleFlow, nodeIdent)); } } LOG.debug("Proceeding with deletion of stale-marked Groups for switch {} using Openflow interface", nodeIdent.toString()); // TODO: Should we collate the futures of RPC-calls to be sure that groups are Flows are fully deleted // before attempting to delete groups - just in case there are references /* Stale-marked Groups - Can be deleted after flows */ List staleGroups = flowNode.get().getStaleGroup() != null ? flowNode.get().getStaleGroup() : Collections. emptyList(); for (StaleGroup staleGroup : staleGroups) { GroupBuilder groupBuilder = new GroupBuilder(staleGroup); Group toBeDeletedGroup = groupBuilder.setGroupId(staleGroup.getGroupId()).build(); final KeyedInstanceIdentifier groupIdent = nodeIdent.child(Group.class, toBeDeletedGroup.getKey()); this.provider.getGroupCommiter().remove(groupIdent, toBeDeletedGroup, nodeIdent); staleGroupsToBeBulkDeleted.add(getStaleGroupInstanceIdentifier(staleGroup, nodeIdent)); } LOG.debug("Proceeding with deletion of stale-marked Meters for switch {} using Openflow interface", nodeIdent.toString()); /* Stale-marked Meters - can be deleted anytime - so least priority */ List staleMeters = flowNode.get().getStaleMeter() != null ? flowNode.get().getStaleMeter() : Collections. emptyList(); for (StaleMeter staleMeter : staleMeters) { MeterBuilder meterBuilder = new MeterBuilder(staleMeter); Meter toBeDeletedMeter = meterBuilder.setMeterId(staleMeter.getMeterId()).build(); final KeyedInstanceIdentifier meterIdent = nodeIdent.child(Meter.class, toBeDeletedMeter.getKey()); this.provider.getMeterCommiter().remove(meterIdent, toBeDeletedMeter, nodeIdent); staleMetersToBeBulkDeleted.add(getStaleMeterInstanceIdentifier(staleMeter, nodeIdent)); } } /* clean transaction */ trans.close(); LOG.debug("Deleting all stale-marked flows/groups/meters of for switch {} in Configuration DS", nodeIdent.toString()); // Now, do the bulk deletions deleteDSStaleFlows(staleFlowsToBeBulkDeleted); deleteDSStaleGroups(staleGroupsToBeBulkDeleted); deleteDSStaleMeters(staleMetersToBeBulkDeleted); } private void deleteDSStaleFlows(List> flowsForBulkDelete){ ImmutableList.Builder> builder = ImmutableList.builder(); ImmutableList> bulkDelFlows = builder.addAll(flowsForBulkDelete.iterator()).build(); WriteTransaction writeTransaction = dataBroker.newWriteOnlyTransaction(); for (InstanceIdentifier staleFlowIId : flowsForBulkDelete){ writeTransaction.delete(LogicalDatastoreType.CONFIGURATION, staleFlowIId); } CheckedFuture submitFuture = writeTransaction.submit(); handleStaleEntityDeletionResultFuture(submitFuture); } private void deleteDSStaleGroups(List> groupsForBulkDelete){ ImmutableList.Builder> builder = ImmutableList.builder(); ImmutableList> bulkDelGroups = builder.addAll(groupsForBulkDelete.iterator()).build(); WriteTransaction writeTransaction = dataBroker.newWriteOnlyTransaction(); for (InstanceIdentifier staleGroupIId : groupsForBulkDelete){ writeTransaction.delete(LogicalDatastoreType.CONFIGURATION, staleGroupIId); } CheckedFuture submitFuture = writeTransaction.submit(); handleStaleEntityDeletionResultFuture(submitFuture); } private void deleteDSStaleMeters(List> metersForBulkDelete){ ImmutableList.Builder> builder = ImmutableList.builder(); ImmutableList> bulkDelGroups = builder.addAll(metersForBulkDelete.iterator()).build(); WriteTransaction writeTransaction = dataBroker.newWriteOnlyTransaction(); for (InstanceIdentifier staleMeterIId : metersForBulkDelete){ writeTransaction.delete(LogicalDatastoreType.CONFIGURATION, staleMeterIId); } CheckedFuture submitFuture = writeTransaction.submit(); handleStaleEntityDeletionResultFuture(submitFuture); } private InstanceIdentifier getStaleFlowInstanceIdentifier(StaleFlow staleFlow, InstanceIdentifier nodeIdent) { return nodeIdent .child(Table.class, new TableKey(staleFlow.getTableId())) .child(org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.StaleFlow.class, new StaleFlowKey(new FlowId(staleFlow.getId()))); } private InstanceIdentifier getStaleGroupInstanceIdentifier(StaleGroup staleGroup, InstanceIdentifier nodeIdent) { return nodeIdent .child(StaleGroup.class, new StaleGroupKey(new GroupId(staleGroup.getGroupId()))); } private InstanceIdentifier getStaleMeterInstanceIdentifier(StaleMeter staleMeter, InstanceIdentifier nodeIdent) { return nodeIdent .child(StaleMeter.class, new StaleMeterKey(new MeterId(staleMeter.getMeterId()))); } private void handleStaleEntityDeletionResultFuture(CheckedFuture submitFuture) { Futures.addCallback(submitFuture, new FutureCallback() { @Override public void onSuccess(Void result) { LOG.debug("Stale entity removal success"); } @Override public void onFailure(Throwable t) { LOG.error("Stale entity removal failed {}", t); } }); } private boolean compareInstanceIdentifierTail(InstanceIdentifier identifier1, InstanceIdentifier identifier2) { return Iterables.getLast(identifier1.getPathArguments()).equals(Iterables.getLast(identifier2.getPathArguments())); } }