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=9859e9cce40b263e68e66c978f67a074652b60f8;hb=05f8db12159673d0e0a95642fe86e62c14b7dc7b;hp=c4e9ad8b726980fa60d30512b4e312df3f88f1e1;hpb=b4f4b4b702e2ccd8a7c62fd2a5c184c5b1cbe665;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 c4e9ad8b72..9859e9cce4 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,22 +1,28 @@ -/** +/* * 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 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.base.Preconditions; import com.google.common.collect.Lists; +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 edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import java.math.BigInteger; +import java.time.LocalDateTime; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; @@ -24,21 +30,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.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.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; @@ -65,7 +77,6 @@ import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.group 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.NodeId; 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.meter.types.rev130918.MeterId; @@ -80,12 +91,8 @@ import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.on 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; @@ -97,6 +104,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; @@ -108,6 +116,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); @@ -116,13 +125,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); @@ -130,9 +143,11 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { private static final AtomicLong BUNDLE_ID = new AtomicLong(); private static final BundleFlags BUNDLE_FLAGS = new BundleFlags(true, true); + private Map reconciliationStates; public FlowNodeReconciliationImpl(final ForwardingRulesManager manager, final DataBroker db, - final String serviceName, final int priority, final ResultState resultState) { + final String serviceName, final int priority, final ResultState resultState, + final FlowGroupCacheManager flowGroupCacheManager) { this.provider = Preconditions.checkNotNull(manager, "ForwardingRulesManager can not be null!"); dataBroker = Preconditions.checkNotNull(db, "DataBroker can not be null!"); this.serviceName = serviceName; @@ -140,6 +155,7 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { this.resultState = resultState; salBundleService = Preconditions.checkNotNull(manager.getSalBundleService(), "salBundleService can not be null!"); + reconciliationStates = flowGroupCacheManager.getReconciliationStates(); } @Override @@ -152,9 +168,12 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { @Override public ListenableFuture reconcileConfiguration(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); } if (provider.isBundleBasedReconciliationEnabled()) { @@ -166,6 +185,13 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { } } + @Override + public void flowNodeDisconnected(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; @@ -176,19 +202,24 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { @Override public Boolean call() { String node = nodeIdentity.firstKeyOf(Node.class).getId().getValue(); - Optional flowNode = Optional.absent(); + Optional flowNode = Optional.empty(); BundleId bundleIdValue = new BundleId(BUNDLE_ID.getAndIncrement()); BigInteger dpnId = getDpnIdFromNodeName(node); + ExecutorService service = Executors.newSingleThreadExecutor(THREAD_FACTORY); LOG.info("Triggering bundle based reconciliation for device : {}", dpnId); - ReadOnlyTransaction trans = provider.getReadTranaction(); - try { + 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) @@ -196,17 +227,18 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { .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(); + LOG.debug("Closing openflow bundle for device {}", dpnId); /* Close previously opened bundle on the openflow switch if any */ ListenableFuture> closeBundle = salBundleService.controlBundle(closeBundleInput); @@ -215,29 +247,38 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { ListenableFuture> openBundle = Futures.transformAsync(closeBundle, rpcResult -> salBundleService.controlBundle(openBundleInput), - MoreExecutors.directExecutor()); + service); - /* Push groups and flows via bundle add messages */ - ListenableFuture> addBundleMessagesFuture + /* Push groups and flows via bundle add messages */ + ListenableFuture> deleteAllFlowGroupsFuture = Futures.transformAsync(openBundle, rpcResult -> { if (rpcResult.isSuccessful()) { - return salBundleService.addBundleMessages(addBundleMessagesInput); + return salBundleService.addBundleMessages(deleteAllFlowGroupsInput); } return Futures.immediateFuture(null); - }, MoreExecutors.directExecutor()); + }, service); - /* Commit the bundle on the openflow switch */ - ListenableFuture> commitBundleFuture - = Futures.transformAsync(addBundleMessagesFuture, rpcResult -> { + /* Push flows and groups via bundle add messages */ + Optional finalFlowNode = flowNode; + ListenableFuture>> addbundlesFuture + = Futures.transformAsync(deleteAllFlowGroupsFuture, rpcResult -> { if (rpcResult.isSuccessful()) { - return salBundleService.controlBundle(commitBundleInput); + LOG.debug("Adding delete all flow/group message is successful for device {}", dpnId); + return Futures.allAsList(addBundleMessages(finalFlowNode.get(), 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 = flowNode.get().nonnullMeter().values(); Futures.transformAsync(commitBundleFuture, rpcResult -> { if (rpcResult.isSuccessful()) { @@ -248,19 +289,26 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { } } 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("commit bundle failed for device {} with error ", dpnId, e); return false; + } finally { + service.shutdown(); } } LOG.error("FlowNode not present for Datapath ID {}", dpnId); @@ -273,14 +321,17 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { InstanceIdentifier connectedNode = node.getNodeInstanceIdentifier() .augmentation(FlowCapableNode.class); // Clearing the group registry cache for the connected node if exists - provider.getDevicesGroupRegistry().clearNodeGroups(node.getNodeId()); + 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); + ListenableFuture listenableFuture = futureMap.computeIfPresent(node, (key, future) -> future); + if (listenableFuture != null) { + listenableFuture.cancel(true); + futureMap.remove(node); + } return Futures.immediateFuture(true); } @@ -311,12 +362,12 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { public Boolean call() { 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; // initialize the counter int counter = 0; - try { + try (ReadTransaction trans = provider.getReadTransaction()) { flowNode = trans.read(LogicalDatastoreType.CONFIGURATION, nodeIdentity).get(); } catch (ExecutionException | InterruptedException e) { LOG.warn("Fail with read Config/DS for Node {} !", nodeIdentity, e); @@ -327,9 +378,12 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { /* 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); + Collection tableList = flowNode.get().nonnullTableFeatures().values(); for (TableFeatures tableFeaturesItem : tableList) { TableFeaturesKey tableKey = tableFeaturesItem.key(); KeyedInstanceIdentifier tableFeaturesII = nodeIdentity @@ -338,14 +392,13 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { } /* Groups - have to be first */ - List groups = flowNode.get().getGroup() != null ? flowNode.get().getGroup() - : Collections.emptyList(); + Collection groups = flowNode.get().nonnullGroup().values(); List toBeInstalledGroups = new ArrayList<>(); toBeInstalledGroups.addAll(groups); // 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 @@ -353,7 +406,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; } @@ -363,18 +416,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() @@ -395,14 +448,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; @@ -434,13 +488,12 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { LOG.debug( "Installing the group {} finally although " + "the port is not up after checking for {} times ", - group.getGroupId().toString(), provider.getReconciliationRetryCount()); + group.getGroupId(), provider.getReconciliationRetryCount()); addGroup(groupFutures, group); } } /* Meters */ - List meters = flowNode.get().getMeter() != null ? flowNode.get().getMeter() - : Collections.emptyList(); + Collection meters = flowNode.get().nonnullMeter().values(); for (Meter meter : meters) { final KeyedInstanceIdentifier meterIdent = nodeIdentity.child(Meter.class, meter.key()); @@ -452,21 +505,23 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { awaitGroups(node, groupFutures.values()); /* Flows */ - List tables = flowNode.get().getTable() != null ? flowNode.get().getTable() + Collection
tables = flowNode.get().getTable() != null ? flowNode.get().nonnullTable().values() : Collections.
emptyList(); + int flowCount = 0; for (Table table : tables) { final KeyedInstanceIdentifier tableIdent = nodeIdentity.child(Table.class, table.key()); - List flows = table.getFlow() != null ? table.getFlow() : Collections.emptyList(); + Collection flows = table.nonnullFlow().values(); + flowCount += flows.size(); for (Flow flow : flows) { final KeyedInstanceIdentifier flowIdent = tableIdent.child(Flow.class, 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; } @@ -479,9 +534,9 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { * @param group * The group to add. */ - private void addGroup(Map> map, Group group) { + private void addGroup(Map> map, Group group) { KeyedInstanceIdentifier groupIdent = nodeIdentity.child(Group.class, group.key()); - final Long groupId = group.getGroupId().getValue(); + final Uint32 groupId = group.getGroupId().getValue(); ListenableFuture future = JdkFutureAdapters .listenInPoolThread(provider.getGroupCommiter().add(groupIdent, group, nodeIdentity)); @@ -496,9 +551,8 @@ 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); + LOG.debug("add-group RPC failed: node={}, id={}", + nodeIdentity.firstKeyOf(Node.class).getId().getValue(), groupId, cause); } }, MoreExecutors.directExecutor()); @@ -539,6 +593,8 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { } } + @SuppressFBWarnings(value = "UPM_UNCALLED_PRIVATE_METHOD", + justification = "https://github.com/spotbugs/spotbugs/issues/811") private BigInteger getDpnIdFromNodeName(String nodeName) { String dpId = nodeName.substring(nodeName.lastIndexOf(SEPARATOR) + 1); @@ -550,10 +606,9 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { List> staleGroupsToBeBulkDeleted = Lists.newArrayList(); List> staleMetersToBeBulkDeleted = Lists.newArrayList(); - ReadOnlyTransaction trans = provider.getReadTranaction(); - Optional flowNode = Optional.absent(); + Optional flowNode = Optional.empty(); - try { + try (ReadTransaction trans = provider.getReadTransaction()) { flowNode = trans.read(LogicalDatastoreType.CONFIGURATION, nodeIdent).get(); } catch (ExecutionException | InterruptedException e) { LOG.warn("Reconciliation Pre-Processing Fail with read Config/DS for Node {} !", nodeIdent, e); @@ -562,15 +617,13 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { if (flowNode.isPresent()) { 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.get().nonnullTable().values(); for (Table table : tables) { final KeyedInstanceIdentifier tableIdent = nodeIdent.child(Table.class, table.key()); - List staleFlows = table.getStaleFlow() != null ? table.getStaleFlow() - : Collections.emptyList(); + Collection staleFlows = table.nonnullStaleFlow().values(); for (StaleFlow staleFlow : staleFlows) { FlowBuilder flowBuilder = new FlowBuilder(staleFlow); @@ -586,15 +639,14 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { } 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(); + Collection staleGroups = flowNode.get().nonnullStaleGroup().values(); for (StaleGroup staleGroup : staleGroups) { GroupBuilder groupBuilder = new GroupBuilder(staleGroup); @@ -609,10 +661,9 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { } 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(); + Collection staleMeters = flowNode.get().getStaleMeter().values(); for (StaleMeter staleMeter : staleMeters) { @@ -628,11 +679,9 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { } } - /* 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); @@ -646,7 +695,7 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { writeTransaction.delete(LogicalDatastoreType.CONFIGURATION, staleFlowIId); } - ListenableFuture submitFuture = writeTransaction.submit(); + FluentFuture submitFuture = writeTransaction.commit(); handleStaleEntityDeletionResultFuture(submitFuture); } @@ -657,7 +706,7 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { writeTransaction.delete(LogicalDatastoreType.CONFIGURATION, staleGroupIId); } - ListenableFuture submitFuture = writeTransaction.submit(); + FluentFuture submitFuture = writeTransaction.commit(); handleStaleEntityDeletionResultFuture(submitFuture); } @@ -668,7 +717,7 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { writeTransaction.delete(LogicalDatastoreType.CONFIGURATION, staleMeterIId); } - ListenableFuture submitFuture = writeTransaction.submit(); + FluentFuture submitFuture = writeTransaction.commit(); handleStaleEntityDeletionResultFuture(submitFuture); } @@ -692,16 +741,38 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { return nodeIdent.child(StaleMeter.class, new StaleMeterKey(new MeterId(staleMeter.getMeterId()))); } - private void handleStaleEntityDeletionResultFuture(ListenableFuture submitFuture) { - Futures.addCallback(submitFuture, new FutureCallback() { + @SuppressFBWarnings(value = "UPM_UNCALLED_PRIVATE_METHOD", + justification = "https://github.com/spotbugs/spotbugs/issues/811") + 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 void handleStaleEntityDeletionResultFuture(FluentFuture submitFuture) { + submitFuture.addCallback(new FutureCallback() { @Override - public void onSuccess(Void result) { + public void onSuccess(Object result) { LOG.debug("Stale entity removal success"); } @Override public void onFailure(Throwable throwable) { - LOG.debug("Stale entity removal failed {}", throwable); + LOG.debug("Stale entity removal failed", throwable); } }, MoreExecutors.directExecutor()); } @@ -719,7 +790,9 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { return groupBuilder.build(); } - private Messages createMessages(final NodeRef nodeRef, final Optional flowNode) { + @SuppressFBWarnings(value = "UPM_UNCALLED_PRIVATE_METHOD", + justification = "https://github.com/spotbugs/spotbugs/issues/811") + private Messages createMessages(final NodeRef nodeRef) { final List messages = new ArrayList<>(); messages.add(new MessageBuilder().setNode(nodeRef) .setBundleInnerMessage(new BundleRemoveFlowCaseBuilder() @@ -730,29 +803,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()) { - NodeId nodeId = nodeRef.getValue().firstKeyOf(Node.class).getId(); - provider.getDevicesGroupRegistry().storeGroup(nodeId,gr.getGroupId().getValue()); - 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(); } }