X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=applications%2Fforwardingrules-manager%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fopenflowplugin%2Fapplications%2Ffrm%2Fimpl%2FFlowNodeReconciliationImpl.java;h=4e932496f1d434bea5fd2b9f16a75ab1d278e0b5;hb=0267562f3c2be39373f289008a367776dd325ff8;hp=282024869a547e7b9c154642e0ea17073933eef8;hpb=c00b850d37900bdc8f2901b8356cfa3ffa653def;p=openflowplugin.git diff --git a/applications/forwardingrules-manager/src/main/java/org/opendaylight/openflowplugin/applications/frm/impl/FlowNodeReconciliationImpl.java b/applications/forwardingrules-manager/src/main/java/org/opendaylight/openflowplugin/applications/frm/impl/FlowNodeReconciliationImpl.java index 282024869a..4e932496f1 100644 --- a/applications/forwardingrules-manager/src/main/java/org/opendaylight/openflowplugin/applications/frm/impl/FlowNodeReconciliationImpl.java +++ b/applications/forwardingrules-manager/src/main/java/org/opendaylight/openflowplugin/applications/frm/impl/FlowNodeReconciliationImpl.java @@ -1,23 +1,26 @@ -/** +/* * Copyright (c) 2014, 2017 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 com.google.common.base.Optional; -import com.google.common.base.Preconditions; -import com.google.common.collect.Lists; -import com.google.common.util.concurrent.CheckedFuture; +import static java.util.Objects.requireNonNull; +import static org.opendaylight.openflowplugin.api.openflow.ReconciliationState.ReconciliationStatus.COMPLETED; +import static org.opendaylight.openflowplugin.api.openflow.ReconciliationState.ReconciliationStatus.FAILED; +import static org.opendaylight.openflowplugin.api.openflow.ReconciliationState.ReconciliationStatus.STARTED; + +import com.google.common.util.concurrent.FluentFuture; import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.JdkFutureAdapters; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.MoreExecutors; +import com.google.common.util.concurrent.ThreadFactoryBuilder; import java.math.BigInteger; +import java.time.LocalDateTime; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; @@ -25,23 +28,27 @@ import java.util.HashMap; import java.util.List; import java.util.ListIterator; import java.util.Map; +import java.util.Optional; import java.util.concurrent.Callable; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; -import java.util.concurrent.Future; +import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicLong; -import org.opendaylight.controller.md.sal.binding.api.DataBroker; -import org.opendaylight.controller.md.sal.binding.api.ReadOnlyTransaction; -import org.opendaylight.controller.md.sal.binding.api.WriteTransaction; -import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType; -import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException; +import org.opendaylight.mdsal.binding.api.DataBroker; +import org.opendaylight.mdsal.binding.api.ReadTransaction; +import org.opendaylight.mdsal.binding.api.WriteTransaction; +import org.opendaylight.mdsal.common.api.LogicalDatastoreType; import org.opendaylight.openflowplugin.api.OFConstants; +import org.opendaylight.openflowplugin.api.openflow.FlowGroupCacheManager; +import org.opendaylight.openflowplugin.api.openflow.ReconciliationState; import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo; import org.opendaylight.openflowplugin.applications.frm.FlowNodeReconciliation; import org.opendaylight.openflowplugin.applications.frm.ForwardingRulesManager; +import org.opendaylight.openflowplugin.applications.frm.util.FrmUtil; 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; @@ -70,23 +77,20 @@ import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.group import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.groups.StaleGroupKey; import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef; 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.openflowplugin.extension.onf.bundle.service.rev170124.AddBundleMessagesInput; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.AddBundleMessagesInputBuilder; +import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.AddBundleMessagesOutput; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.ControlBundleInput; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.ControlBundleInputBuilder; +import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.ControlBundleOutput; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.SalBundleService; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.add.bundle.messages.input.Messages; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.add.bundle.messages.input.MessagesBuilder; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.add.bundle.messages.input.messages.Message; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.add.bundle.messages.input.messages.MessageBuilder; -import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.bundle.inner.message.grouping.bundle.inner.message.BundleAddFlowCaseBuilder; -import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.bundle.inner.message.grouping.bundle.inner.message.BundleAddGroupCaseBuilder; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.bundle.inner.message.grouping.bundle.inner.message.BundleRemoveFlowCaseBuilder; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.bundle.inner.message.grouping.bundle.inner.message.BundleRemoveGroupCaseBuilder; -import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.bundle.inner.message.grouping.bundle.inner.message.bundle.add.flow._case.AddFlowCaseDataBuilder; -import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.bundle.inner.message.grouping.bundle.inner.message.bundle.add.group._case.AddGroupCaseDataBuilder; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.bundle.inner.message.grouping.bundle.inner.message.bundle.remove.flow._case.RemoveFlowCaseDataBuilder; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.bundle.inner.message.grouping.bundle.inner.message.bundle.remove.group._case.RemoveGroupCaseDataBuilder; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.rev170124.BundleControlType; @@ -98,6 +102,7 @@ import org.opendaylight.yang.gen.v1.urn.opendaylight.table.types.rev131026.table import org.opendaylight.yangtools.yang.binding.InstanceIdentifier; import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier; import org.opendaylight.yangtools.yang.common.RpcResult; +import org.opendaylight.yangtools.yang.common.Uint32; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -109,6 +114,7 @@ import org.slf4j.LoggerFactory; public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { private static final Logger LOG = LoggerFactory.getLogger(FlowNodeReconciliationImpl.class); + private static final Logger OF_EVENT_LOG = LoggerFactory.getLogger("OfEventLog"); // The number of nanoseconds to wait for a single group to be added. private static final long ADD_GROUP_TIMEOUT = TimeUnit.SECONDS.toNanos(3); @@ -117,13 +123,17 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { private static final long MAX_ADD_GROUP_TIMEOUT = TimeUnit.SECONDS.toNanos(20); private static final String SEPARATOR = ":"; private static final int THREAD_POOL_SIZE = 4; - + private static final ThreadFactory THREAD_FACTORY = new ThreadFactoryBuilder() + .setNameFormat("BundleResync-%d") + .setDaemon(false) + .setUncaughtExceptionHandler((thread, ex) -> LOG.error("Uncaught exception {}", thread, ex)) + .build(); private final DataBroker dataBroker; private final ForwardingRulesManager provider; private final String serviceName; private final int priority; private final ResultState resultState; - private final Map> futureMap = new HashMap<>(); + private final Map> futureMap = new ConcurrentHashMap<>(); private final ExecutorService executor = Executors.newFixedThreadPool(THREAD_POOL_SIZE); @@ -131,16 +141,18 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { private static final AtomicLong BUNDLE_ID = new AtomicLong(); private static final BundleFlags BUNDLE_FLAGS = new BundleFlags(true, true); + private final Map reconciliationStates; public FlowNodeReconciliationImpl(final ForwardingRulesManager manager, final DataBroker db, - final String serviceName, final int priority, final ResultState resultState) { - this.provider = Preconditions.checkNotNull(manager, "ForwardingRulesManager can not be null!"); - dataBroker = Preconditions.checkNotNull(db, "DataBroker can not be null!"); + final String serviceName, final int priority, final ResultState resultState, + final FlowGroupCacheManager flowGroupCacheManager) { + provider = requireNonNull(manager, "ForwardingRulesManager can not be null!"); + dataBroker = requireNonNull(db, "DataBroker can not be null!"); this.serviceName = serviceName; this.priority = priority; this.resultState = resultState; - salBundleService = Preconditions.checkNotNull(manager.getSalBundleService(), - "salBundleService can not be null!"); + salBundleService = requireNonNull(manager.getSalBundleService(), "salBundleService can not be null!"); + reconciliationStates = flowGroupCacheManager.getReconciliationStates(); } @Override @@ -150,15 +162,17 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { } } - private ListenableFuture reconcileConfiguration(InstanceIdentifier connectedNode) { + @Override + public ListenableFuture reconcileConfiguration(final InstanceIdentifier connectedNode) { LOG.info("Triggering reconciliation for device {}", connectedNode.firstKeyOf(Node.class)); + // Clearing the group registry cache for the connected node if exists + String nodeId = FrmUtil.getNodeIdValueFromNodeIdentifier(connectedNode); + provider.getDevicesGroupRegistry().clearNodeGroups(nodeId); if (provider.isStaleMarkingEnabled()) { LOG.info("Stale-Marking is ENABLED and proceeding with deletion of " + "stale-marked entities on switch {}", - connectedNode.toString()); + connectedNode); reconciliationPreProcess(connectedNode); } - LOG.debug("Bundle based reconciliation status : {}", - provider.isBundleBasedReconciliationEnabled() ? "Enable" : "Disable"); if (provider.isBundleBasedReconciliationEnabled()) { BundleBasedReconciliationTask bundleBasedReconTask = new BundleBasedReconciliationTask(connectedNode); return JdkFutureAdapters.listenInPoolThread(executor.submit(bundleBasedReconTask)); @@ -168,6 +182,13 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { } } + @Override + public void flowNodeDisconnected(final InstanceIdentifier disconnectedNode) { + String node = disconnectedNode.firstKeyOf(Node.class).getId().getValue(); + BigInteger dpnId = getDpnIdFromNodeName(node); + reconciliationStates.remove(dpnId.toString()); + } + private class BundleBasedReconciliationTask implements Callable { final InstanceIdentifier nodeIdentity; @@ -177,79 +198,118 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { @Override public Boolean call() { - String node = nodeIdentity.firstKeyOf(Node.class, NodeKey.class).getId().getValue(); - Optional flowNode = Optional.absent(); - BundleId bundleIdValue = new BundleId(BUNDLE_ID.getAndIncrement()); + String node = nodeIdentity.firstKeyOf(Node.class).getId().getValue(); + Optional flowNode = Optional.empty(); + BundleId bundleIdValue = new BundleId(Uint32.valueOf(BUNDLE_ID.getAndIncrement())); BigInteger dpnId = getDpnIdFromNodeName(node); - LOG.debug("Triggering bundle based reconciliation for device :{}", dpnId); - ReadOnlyTransaction trans = provider.getReadTranaction(); - try { + ExecutorService service = Executors.newSingleThreadExecutor(THREAD_FACTORY); + LOG.info("Triggering bundle based reconciliation for device : {}", dpnId); + try (ReadTransaction trans = provider.getReadTransaction()) { flowNode = trans.read(LogicalDatastoreType.CONFIGURATION, nodeIdentity).get(); } catch (ExecutionException | InterruptedException e) { LOG.error("Error occurred while reading the configuration data store for node {}", nodeIdentity, e); } if (flowNode.isPresent()) { + ReconciliationState reconciliationState = new ReconciliationState( + STARTED, LocalDateTime.now()); + //put the dpn info into the map + reconciliationStates.put(dpnId.toString(), reconciliationState); LOG.debug("FlowNode present for Datapath ID {}", dpnId); + OF_EVENT_LOG.debug("Bundle Reconciliation Start, Node: {}", dpnId); final NodeRef nodeRef = new NodeRef(nodeIdentity.firstIdentifierOf(Node.class)); + final ControlBundleInput closeBundleInput = new ControlBundleInputBuilder().setNode(nodeRef) + .setBundleId(bundleIdValue).setFlags(BUNDLE_FLAGS) + .setType(BundleControlType.ONFBCTCLOSEREQUEST).build(); + final ControlBundleInput openBundleInput = new ControlBundleInputBuilder().setNode(nodeRef) - .setBundleId(bundleIdValue).setFlags(BUNDLE_FLAGS).setType(BundleControlType.ONFBCTOPENREQUEST) - .build(); + .setBundleId(bundleIdValue).setFlags(BUNDLE_FLAGS) + .setType(BundleControlType.ONFBCTOPENREQUEST).build(); final ControlBundleInput commitBundleInput = new ControlBundleInputBuilder().setNode(nodeRef) .setBundleId(bundleIdValue).setFlags(BUNDLE_FLAGS) .setType(BundleControlType.ONFBCTCOMMITREQUEST).build(); - final AddBundleMessagesInput addBundleMessagesInput = new AddBundleMessagesInputBuilder() + final AddBundleMessagesInput deleteAllFlowGroupsInput = new AddBundleMessagesInputBuilder() .setNode(nodeRef).setBundleId(bundleIdValue).setFlags(BUNDLE_FLAGS) - .setMessages(createMessages(nodeRef, flowNode)).build(); + .setMessages(createMessages(nodeRef)).build(); - Future> openBundle = salBundleService.controlBundle(openBundleInput); + LOG.debug("Closing openflow bundle for device {}", dpnId); + /* Close previously opened bundle on the openflow switch if any */ + ListenableFuture> closeBundle + = salBundleService.controlBundle(closeBundleInput); - ListenableFuture> addBundleMessagesFuture = Futures - .transformAsync(JdkFutureAdapters.listenInPoolThread(openBundle), rpcResult -> { + /* Open a new bundle on the switch */ + ListenableFuture> openBundle + = Futures.transformAsync(closeBundle, rpcResult -> { if (rpcResult.isSuccessful()) { - return JdkFutureAdapters - .listenInPoolThread(salBundleService.addBundleMessages(addBundleMessagesInput)); + LOG.debug("Existing bundle is successfully closed for device {}", dpnId); + } + return salBundleService.controlBundle(openBundleInput); + }, service); + + /* Push groups and flows via bundle add messages */ + ListenableFuture> deleteAllFlowGroupsFuture + = Futures.transformAsync(openBundle, rpcResult -> { + if (rpcResult.isSuccessful()) { + LOG.debug("Open bundle is successful for device {}", dpnId); + return salBundleService.addBundleMessages(deleteAllFlowGroupsInput); } return Futures.immediateFuture(null); - }, MoreExecutors.directExecutor()); - ListenableFuture> commitBundleFuture = Futures.transformAsync(addBundleMessagesFuture, - rpcResult -> { + }, service); + + /* Push flows and groups via bundle add messages */ + Optional finalFlowNode = flowNode; + ListenableFuture>> addbundlesFuture = + Futures.transformAsync(deleteAllFlowGroupsFuture, rpcResult -> { if (rpcResult.isSuccessful()) { - return JdkFutureAdapters - .listenInPoolThread(salBundleService.controlBundle(commitBundleInput)); + LOG.debug("Adding delete all flow/group message is successful for device {}", dpnId); + return Futures.allAsList(addBundleMessages(finalFlowNode.orElseThrow(), bundleIdValue, + nodeIdentity)); } return Futures.immediateFuture(null); - }, MoreExecutors.directExecutor()); + }, service); + + /* Commit the bundle on the openflow switch */ + ListenableFuture> commitBundleFuture = + Futures.transformAsync(addbundlesFuture, rpcResult -> { + LOG.debug("Adding bundle messages completed for device {}", dpnId); + return salBundleService.controlBundle(commitBundleInput); + }, service); /* Bundles not supported for meters */ - List meters = flowNode.get().getMeter() != null ? flowNode.get().getMeter() - : Collections.emptyList(); + Collection meters = finalFlowNode.orElseThrow().nonnullMeter().values(); Futures.transformAsync(commitBundleFuture, rpcResult -> { if (rpcResult.isSuccessful()) { for (Meter meter : meters) { final KeyedInstanceIdentifier meterIdent = nodeIdentity - .child(Meter.class, meter.getKey()); + .child(Meter.class, meter.key()); provider.getMeterCommiter().add(meterIdent, meter, nodeIdentity); } } return Futures.immediateFuture(null); - }, MoreExecutors.directExecutor()); - - trans.close(); + }, service); try { - if (commitBundleFuture.get().isSuccessful()) { + RpcResult bundleFuture = commitBundleFuture.get(); + if (bundleFuture != null && bundleFuture.isSuccessful()) { + reconciliationState.setState(COMPLETED, LocalDateTime.now()); LOG.debug("Completing bundle based reconciliation for device ID:{}", dpnId); + OF_EVENT_LOG.debug("Bundle Reconciliation Finish, Node: {}", dpnId); return true; } else { + reconciliationState.setState(FAILED, LocalDateTime.now()); + LOG.error("commit bundle failed for device {} with error {}", dpnId, + commitBundleFuture.get().getErrors()); return false; } } catch (InterruptedException | ExecutionException e) { - LOG.error("Error while doing bundle based reconciliation for device ID:{}", nodeIdentity); + reconciliationState.setState(FAILED, LocalDateTime.now()); + LOG.error("Error while doing bundle based reconciliation for device ID:{}", dpnId, e); return false; + } finally { + service.shutdown(); } } LOG.error("FlowNode not present for Datapath ID {}", dpnId); @@ -258,16 +318,21 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { } @Override - public ListenableFuture startReconciliation(DeviceInfo node) { + public ListenableFuture startReconciliation(final DeviceInfo node) { InstanceIdentifier connectedNode = node.getNodeInstanceIdentifier() .augmentation(FlowCapableNode.class); + // Clearing the group registry cache for the connected node if exists + provider.getDevicesGroupRegistry().clearNodeGroups(node.toString()); return futureMap.computeIfAbsent(node, future -> reconcileConfiguration(connectedNode)); } @Override - public ListenableFuture endReconciliation(DeviceInfo node) { - futureMap.computeIfPresent(node, (key, future) -> future).cancel(true); - futureMap.remove(node); + public ListenableFuture endReconciliation(final DeviceInfo node) { + ListenableFuture listenableFuture = futureMap.computeIfPresent(node, (key, future) -> future); + if (listenableFuture != null) { + listenableFuture.cancel(true); + futureMap.remove(node); + } return Futures.immediateFuture(true); } @@ -296,43 +361,43 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { @Override public Boolean call() { - String node = nodeIdentity.firstKeyOf(Node.class, NodeKey.class).getId().getValue(); + String node = nodeIdentity.firstKeyOf(Node.class).getId().getValue(); BigInteger dpnId = getDpnIdFromNodeName(node); + OF_EVENT_LOG.debug("Reconciliation Start, Node: {}", dpnId); - ReadOnlyTransaction trans = provider.getReadTranaction(); - Optional flowNode = Optional.absent(); + Optional optFlowNode; // initialize the counter int counter = 0; - try { - flowNode = trans.read(LogicalDatastoreType.CONFIGURATION, nodeIdentity).get(); + try (ReadTransaction trans = provider.getReadTransaction()) { + optFlowNode = trans.read(LogicalDatastoreType.CONFIGURATION, nodeIdentity).get(); } catch (ExecutionException | InterruptedException e) { LOG.warn("Fail with read Config/DS for Node {} !", nodeIdentity, e); return false; } - if (flowNode.isPresent()) { + if (optFlowNode.isPresent()) { /* Tables - have to be pushed before groups */ // CHECK if while pushing the update, updateTableInput can be null to emulate a // table add - List tableList = flowNode.get().getTableFeatures() != null - ? flowNode.get().getTableFeatures() - : Collections.emptyList(); + ReconciliationState reconciliationState = new ReconciliationState(STARTED, LocalDateTime.now()); + //put the dpn info into the map + reconciliationStates.put(dpnId.toString(), reconciliationState); + LOG.debug("Triggering reconciliation for node {} with state: {}", dpnId, STARTED); + FlowCapableNode flowNode = optFlowNode.orElseThrow(); + Collection tableList = flowNode.nonnullTableFeatures().values(); for (TableFeatures tableFeaturesItem : tableList) { - TableFeaturesKey tableKey = tableFeaturesItem.getKey(); + TableFeaturesKey tableKey = tableFeaturesItem.key(); 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 toBeInstalledGroups = new ArrayList<>(flowNode.nonnullGroup().values()); // new list for suspected groups pointing to ports .. when the ports come up // late List suspectedGroups = new ArrayList<>(); - Map> groupFutures = new HashMap<>(); + Map> groupFutures = new HashMap<>(); while ((!toBeInstalledGroups.isEmpty() || !suspectedGroups.isEmpty()) && counter <= provider.getReconciliationRetryCount()) { // also check if the counter has not @@ -340,7 +405,7 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { if (toBeInstalledGroups.isEmpty() && !suspectedGroups.isEmpty()) { LOG.debug("These Groups are pointing to node-connectors that are not up yet {}", - suspectedGroups.toString()); + suspectedGroups); toBeInstalledGroups.addAll(suspectedGroups); break; } @@ -350,18 +415,18 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { Group group = iterator.next(); boolean okToInstall = true; Buckets buckets = group.getBuckets(); - List bucketList = buckets == null ? null : buckets.getBucket(); + Collection bucketList = buckets == null ? null : buckets.nonnullBucket().values(); if (bucketList == null) { bucketList = Collections.emptyList(); } for (Bucket bucket : bucketList) { - List actions = bucket.getAction(); + Collection actions = bucket.nonnullAction().values(); if (actions == null) { actions = Collections.emptyList(); } for (Action action : actions) { // chained-port - if (action.getAction().getImplementedInterface().getName() + if (action.getAction().implementedInterface().getName() .equals("org.opendaylight.yang.gen.v1.urn.opendaylight" + ".action.types.rev131112.action.action.OutputActionCase")) { String nodeConnectorUri = ((OutputActionCase) action.getAction()).getOutputAction() @@ -382,14 +447,15 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { LOG.debug( "Not yet received the node-connector updated for {} " + "for the group with id {}", - nodeConnectorUri, group.getGroupId().toString()); + nodeConnectorUri, group.getGroupId()); break; } - } else if (action.getAction().getImplementedInterface().getName() + } else if (action.getAction().implementedInterface().getName() .equals("org.opendaylight.yang.gen.v1.urn.opendaylight" + ".action.types.rev131112.action.action.GroupActionCase")) { // chained groups - Long groupId = ((GroupActionCase) action.getAction()).getGroupAction().getGroupId(); + Uint32 groupId = ((GroupActionCase) action.getAction()).getGroupAction() + .getGroupId(); ListenableFuture future = groupFutures.get(groupId); if (future == null) { okToInstall = false; @@ -419,18 +485,16 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { if (!toBeInstalledGroups.isEmpty()) { for (Group group : toBeInstalledGroups) { LOG.debug( - "Installing the group {} finally although " - + "the port is not up after checking for {} times ", - group.getGroupId().toString(), provider.getReconciliationRetryCount()); + "Installing the group {} finally although the port is not up after checking for {} times ", + group.getGroupId(), provider.getReconciliationRetryCount()); addGroup(groupFutures, group); } } /* Meters */ - List meters = flowNode.get().getMeter() != null ? flowNode.get().getMeter() - : Collections.emptyList(); + Collection meters = flowNode.nonnullMeter().values(); for (Meter meter : meters) { final KeyedInstanceIdentifier meterIdent = nodeIdentity.child(Meter.class, - meter.getKey()); + meter.key()); provider.getMeterCommiter().add(meterIdent, meter, nodeIdentity); } @@ -439,21 +503,21 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { awaitGroups(node, groupFutures.values()); /* Flows */ - List tables = flowNode.get().getTable() != null ? flowNode.get().getTable() - : Collections.
emptyList(); - for (Table table : tables) { + int flowCount = 0; + for (Table table : flowNode.nonnullTable().values()) { final KeyedInstanceIdentifier tableIdent = nodeIdentity.child(Table.class, - table.getKey()); - List flows = table.getFlow() != null ? table.getFlow() : Collections.emptyList(); + table.key()); + Collection flows = table.nonnullFlow().values(); + flowCount += flows.size(); for (Flow flow : flows) { final KeyedInstanceIdentifier flowIdent = tableIdent.child(Flow.class, - flow.getKey()); + flow.key()); provider.getFlowCommiter().add(flowIdent, flow, nodeIdentity); } } + reconciliationState.setState(COMPLETED, LocalDateTime.now()); + OF_EVENT_LOG.debug("Reconciliation Finish, Node: {}, flow count: {}", dpnId, flowCount); } - /* clean transaction */ - trans.close(); return true; } @@ -466,15 +530,15 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { * @param group * The group to add. */ - private void addGroup(Map> map, Group group) { - KeyedInstanceIdentifier groupIdent = nodeIdentity.child(Group.class, group.getKey()); - final Long groupId = group.getGroupId().getValue(); + private void addGroup(final Map> map, final Group group) { + KeyedInstanceIdentifier groupIdent = nodeIdentity.child(Group.class, group.key()); + final Uint32 groupId = group.getGroupId().getValue(); ListenableFuture future = JdkFutureAdapters .listenInPoolThread(provider.getGroupCommiter().add(groupIdent, group, nodeIdentity)); Futures.addCallback(future, new FutureCallback() { @Override - public void onSuccess(Object result) { + public void onSuccess(final Object result) { if (LOG.isTraceEnabled()) { LOG.trace("add-group RPC completed: node={}, id={}", nodeIdentity.firstKeyOf(Node.class).getId().getValue(), groupId); @@ -482,10 +546,9 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { } @Override - public void onFailure(Throwable cause) { - String msg = "add-group RPC failed: node=" + nodeIdentity.firstKeyOf(Node.class).getId().getValue() - + ", id=" + groupId; - LOG.debug(msg, cause); + public void onFailure(final Throwable cause) { + LOG.debug("add-group RPC failed: node={}, id={}", + nodeIdentity.firstKeyOf(Node.class).getId().getValue(), groupId, cause); } }, MoreExecutors.directExecutor()); @@ -501,7 +564,7 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { * Future associated with add-group RPC that installs the target * group. */ - private void awaitGroup(String nodeId, ListenableFuture future) { + private void awaitGroup(final String nodeId, final ListenableFuture future) { awaitGroups(nodeId, Collections.singleton(future)); } @@ -513,7 +576,7 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { * @param futures * A collection of futures associated with add-group RPCs. */ - private void awaitGroups(String nodeId, Collection> futures) { + private void awaitGroups(final String nodeId, final Collection> futures) { if (!futures.isEmpty()) { long timeout = Math.min(ADD_GROUP_TIMEOUT * futures.size(), MAX_ADD_GROUP_TIMEOUT); try { @@ -526,188 +589,193 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { } } - private BigInteger getDpnIdFromNodeName(String nodeName) { - + private static BigInteger getDpnIdFromNodeName(final String nodeName) { String dpId = nodeName.substring(nodeName.lastIndexOf(SEPARATOR) + 1); return new BigInteger(dpId); } private void reconciliationPreProcess(final InstanceIdentifier nodeIdent) { - List> staleFlowsToBeBulkDeleted = Lists.newArrayList(); - List> staleGroupsToBeBulkDeleted = Lists.newArrayList(); - List> staleMetersToBeBulkDeleted = Lists.newArrayList(); + List> staleFlowsToBeBulkDeleted = new ArrayList<>(); + List> staleGroupsToBeBulkDeleted = new ArrayList<>(); + List> staleMetersToBeBulkDeleted = new ArrayList<>(); - ReadOnlyTransaction trans = provider.getReadTranaction(); - Optional flowNode = Optional.absent(); + Optional optFlowNode = Optional.empty(); - try { - flowNode = trans.read(LogicalDatastoreType.CONFIGURATION, nodeIdent).get(); + try (ReadTransaction trans = provider.getReadTransaction()) { + optFlowNode = trans.read(LogicalDatastoreType.CONFIGURATION, nodeIdent).get(); } catch (ExecutionException | InterruptedException e) { LOG.warn("Reconciliation Pre-Processing Fail with read Config/DS for Node {} !", nodeIdent, e); } - if (flowNode.isPresent()) { - + if (optFlowNode.isPresent()) { + final FlowCapableNode flowNode = optFlowNode.orElseThrow(); LOG.debug("Proceeding with deletion of stale-marked Flows on switch {} using Openflow interface", - nodeIdent.toString()); + nodeIdent); /* Stale-Flows - Stale-marked Flows have to be removed first for safety */ - List
tables = flowNode.get().getTable() != null ? flowNode.get().getTable() - : Collections.
emptyList(); + Collection
tables = flowNode.nonnullTable().values(); for (Table table : tables) { final KeyedInstanceIdentifier tableIdent = nodeIdent.child(Table.class, - table.getKey()); - List staleFlows = table.getStaleFlow() != null ? table.getStaleFlow() - : Collections.emptyList(); + table.key()); + Collection staleFlows = table.nonnullStaleFlow().values(); 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()); + toBeDeletedFlow.key()); - this.provider.getFlowCommiter().remove(flowIdent, toBeDeletedFlow, nodeIdent); + 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()); + nodeIdent); // 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) { - + for (StaleGroup staleGroup : flowNode.nonnullStaleGroup().values()) { GroupBuilder groupBuilder = new GroupBuilder(staleGroup); Group toBeDeletedGroup = groupBuilder.setGroupId(staleGroup.getGroupId()).build(); final KeyedInstanceIdentifier groupIdent = nodeIdent.child(Group.class, - toBeDeletedGroup.getKey()); + toBeDeletedGroup.key()); - this.provider.getGroupCommiter().remove(groupIdent, toBeDeletedGroup, nodeIdent); + 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()); + nodeIdent); /* 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) { + for (StaleMeter staleMeter : flowNode.nonnullStaleMeter().values()) { MeterBuilder meterBuilder = new MeterBuilder(staleMeter); Meter toBeDeletedMeter = meterBuilder.setMeterId(staleMeter.getMeterId()).build(); final KeyedInstanceIdentifier meterIdent = nodeIdent.child(Meter.class, - toBeDeletedMeter.getKey()); + toBeDeletedMeter.key()); - this.provider.getMeterCommiter().remove(meterIdent, toBeDeletedMeter, nodeIdent); + 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()); + nodeIdent); // Now, do the bulk deletions deleteDSStaleFlows(staleFlowsToBeBulkDeleted); deleteDSStaleGroups(staleGroupsToBeBulkDeleted); deleteDSStaleMeters(staleMetersToBeBulkDeleted); } - private void deleteDSStaleFlows(List> flowsForBulkDelete) { + private void deleteDSStaleFlows(final List> flowsForBulkDelete) { WriteTransaction writeTransaction = dataBroker.newWriteOnlyTransaction(); for (InstanceIdentifier staleFlowIId : flowsForBulkDelete) { writeTransaction.delete(LogicalDatastoreType.CONFIGURATION, staleFlowIId); } - CheckedFuture submitFuture = writeTransaction.submit(); + FluentFuture submitFuture = writeTransaction.commit(); handleStaleEntityDeletionResultFuture(submitFuture); } - private void deleteDSStaleGroups(List> groupsForBulkDelete) { + private void deleteDSStaleGroups(final List> groupsForBulkDelete) { WriteTransaction writeTransaction = dataBroker.newWriteOnlyTransaction(); for (InstanceIdentifier staleGroupIId : groupsForBulkDelete) { writeTransaction.delete(LogicalDatastoreType.CONFIGURATION, staleGroupIId); } - CheckedFuture submitFuture = writeTransaction.submit(); + FluentFuture submitFuture = writeTransaction.commit(); handleStaleEntityDeletionResultFuture(submitFuture); } - private void deleteDSStaleMeters(List> metersForBulkDelete) { + private void deleteDSStaleMeters(final List> metersForBulkDelete) { WriteTransaction writeTransaction = dataBroker.newWriteOnlyTransaction(); for (InstanceIdentifier staleMeterIId : metersForBulkDelete) { writeTransaction.delete(LogicalDatastoreType.CONFIGURATION, staleMeterIId); } - CheckedFuture submitFuture = writeTransaction.submit(); + FluentFuture submitFuture = writeTransaction.commit(); handleStaleEntityDeletionResultFuture(submitFuture); } - private InstanceIdentifier getStaleFlowInstanceIdentifier( - StaleFlow staleFlow, InstanceIdentifier nodeIdent) { + final StaleFlow staleFlow, final 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) { + final StaleGroup staleGroup, final InstanceIdentifier nodeIdent) { return nodeIdent.child(StaleGroup.class, new StaleGroupKey(new GroupId(staleGroup.getGroupId()))); } - private InstanceIdentifier getStaleMeterInstanceIdentifier( - StaleMeter staleMeter, InstanceIdentifier nodeIdent) { + final StaleMeter staleMeter, final InstanceIdentifier nodeIdent) { return nodeIdent.child(StaleMeter.class, new StaleMeterKey(new MeterId(staleMeter.getMeterId()))); } - private void handleStaleEntityDeletionResultFuture( - CheckedFuture submitFuture) { - Futures.addCallback(submitFuture, new FutureCallback() { + private List>> addBundleMessages(final FlowCapableNode flowNode, + final BundleId bundleIdValue, + final InstanceIdentifier nodeIdentity) { + List>> futureList = new ArrayList<>(); + for (Group group : flowNode.nonnullGroup().values()) { + final KeyedInstanceIdentifier groupIdent = nodeIdentity.child(Group.class, group.key()); + futureList.add(provider.getBundleGroupListener().add(groupIdent, group, nodeIdentity, bundleIdValue)); + } + + for (Table table : flowNode.nonnullTable().values()) { + final KeyedInstanceIdentifier tableIdent = nodeIdentity.child(Table.class, table.key()); + for (Flow flow : table.nonnullFlow().values()) { + final KeyedInstanceIdentifier flowIdent = tableIdent.child(Flow.class, flow.key()); + futureList.add(provider.getBundleFlowListener().add(flowIdent, flow, nodeIdentity, bundleIdValue)); + } + } + OF_EVENT_LOG.debug("Flow/Group count is {}", futureList.size()); + return futureList; + } + + private static void handleStaleEntityDeletionResultFuture(final FluentFuture submitFuture) { + submitFuture.addCallback(new FutureCallback() { @Override - public void onSuccess(Void result) { + public void onSuccess(final Object result) { LOG.debug("Stale entity removal success"); } @Override - public void onFailure(Throwable throwable) { - LOG.debug("Stale entity removal failed {}", throwable); + public void onFailure(final Throwable throwable) { + LOG.debug("Stale entity removal failed", throwable); } }, MoreExecutors.directExecutor()); } - private Flow getDeleteAllFlow() { - final FlowBuilder flowBuilder = new FlowBuilder(); - flowBuilder.setTableId(OFConstants.OFPTT_ALL); - return flowBuilder.build(); + private static Flow getDeleteAllFlow() { + return new FlowBuilder().setTableId(OFConstants.OFPTT_ALL).build(); } - private Group getDeleteAllGroup() { - final GroupBuilder groupBuilder = new GroupBuilder(); - groupBuilder.setGroupType(GroupTypes.GroupAll); - groupBuilder.setGroupId(new GroupId(OFConstants.OFPG_ALL)); - return groupBuilder.build(); + private static Group getDeleteAllGroup() { + return new GroupBuilder() + .setGroupType(GroupTypes.GroupAll) + .setGroupId(new GroupId(OFConstants.OFPG_ALL)) + .build(); } - private Messages createMessages(final NodeRef nodeRef, final Optional flowNode) { + private static Messages createMessages(final NodeRef nodeRef) { final List messages = new ArrayList<>(); messages.add(new MessageBuilder().setNode(nodeRef) .setBundleInnerMessage(new BundleRemoveFlowCaseBuilder() @@ -718,27 +786,6 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { .setBundleInnerMessage(new BundleRemoveGroupCaseBuilder() .setRemoveGroupCaseData(new RemoveGroupCaseDataBuilder(getDeleteAllGroup()).build()).build()) .build()); - - if (flowNode.get().getGroup() != null) { - for (Group gr : flowNode.get().getGroup()) { - messages.add(new MessageBuilder().setNode(nodeRef).setBundleInnerMessage(new BundleAddGroupCaseBuilder() - .setAddGroupCaseData(new AddGroupCaseDataBuilder(gr).build()).build()).build()); - } - } - - if (flowNode.get().getTable() != null) { - for (Table table : flowNode.get().getTable()) { - for (Flow flow : table.getFlow()) { - messages.add( - new MessageBuilder().setNode(nodeRef) - .setBundleInnerMessage(new BundleAddFlowCaseBuilder() - .setAddFlowCaseData(new AddFlowCaseDataBuilder(flow).build()).build()) - .build()); - } - } - } - - LOG.debug("The size of the flows and group messages created in createMessage() {}", messages.size()); return new MessagesBuilder().setMessage(messages).build(); } }