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=b3ef2f2f0bf1b125676548537f864e5b1f8171f3;hb=13e1d5e6c0237b9378d60526dd8c1d79db6d2b49;hp=03625601c2ab97b7424436656bb820142828dece;hpb=0217e646030922e7ce48c560218e71d2191c587b;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 03625601c2..b3ef2f2f0b 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,21 +1,21 @@ -/** +/* * 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.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 edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import java.math.BigInteger; import java.util.ArrayList; import java.util.Collection; @@ -24,6 +24,7 @@ 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.ExecutionException; import java.util.concurrent.ExecutorService; @@ -31,10 +32,10 @@ import java.util.concurrent.Executors; 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.device.DeviceInfo; import org.opendaylight.openflowplugin.applications.frm.FlowNodeReconciliation; @@ -65,7 +66,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 +80,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 +93,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 +105,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); @@ -176,11 +174,11 @@ 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); LOG.info("Triggering bundle based reconciliation for device : {}", dpnId); - try (ReadOnlyTransaction trans = provider.getReadTransaction()) { + 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); @@ -188,6 +186,7 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { if (flowNode.isPresent()) { 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) @@ -202,10 +201,11 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { .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); @@ -217,23 +217,33 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { MoreExecutors.directExecutor()); /* Push groups and flows via bundle add messages */ - ListenableFuture> addBundleMessagesFuture + ListenableFuture> deleteAllFlowGroupsFuture = Futures.transformAsync(openBundle, rpcResult -> { if (rpcResult.isSuccessful()) { - return salBundleService.addBundleMessages(addBundleMessagesInput); + return salBundleService.addBundleMessages(deleteAllFlowGroupsInput); } return Futures.immediateFuture(null); }, MoreExecutors.directExecutor()); - /* 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()); + /* 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); + }, MoreExecutors.directExecutor()); + /* Bundles not supported for meters */ List meters = flowNode.get().getMeter() != null ? flowNode.get().getMeter() : Collections.emptyList(); @@ -248,16 +258,19 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { } return Futures.immediateFuture(null); }, MoreExecutors.directExecutor()); - try { - if (commitBundleFuture.get().isSuccessful()) { + RpcResult bundleFuture = commitBundleFuture.get(); + if (bundleFuture != null && bundleFuture.isSuccessful()) { LOG.debug("Completing bundle based reconciliation for device ID:{}", dpnId); + OF_EVENT_LOG.debug("Bundle Reconciliation Finish, Node: {}", dpnId); return true; } else { + 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); + LOG.error("Error while doing bundle based reconciliation for device ID:{}", dpnId); return false; } } @@ -309,11 +322,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); Optional flowNode; // initialize the counter int counter = 0; - try (ReadOnlyTransaction trans = provider.getReadTransaction()) { + 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); @@ -342,7 +356,7 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { // 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 @@ -371,7 +385,7 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { } 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,11 +409,12 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { nodeConnectorUri, group.getGroupId().toString()); 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; @@ -451,16 +466,19 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { /* Flows */ List tables = flowNode.get().getTable() != null ? flowNode.get().getTable() : 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(); + flowCount += flows.size(); for (Flow flow : flows) { final KeyedInstanceIdentifier flowIdent = tableIdent.child(Flow.class, flow.key()); provider.getFlowCommiter().add(flowIdent, flow, nodeIdentity); } } + OF_EVENT_LOG.debug("Reconciliation Finish, Node: {}, flow count: {}", dpnId, flowCount); } return true; } @@ -474,9 +492,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)); @@ -491,9 +509,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()); @@ -534,6 +551,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); @@ -545,9 +564,9 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { List> staleGroupsToBeBulkDeleted = Lists.newArrayList(); List> staleMetersToBeBulkDeleted = Lists.newArrayList(); - Optional flowNode = Optional.absent(); + Optional flowNode = Optional.empty(); - try (ReadOnlyTransaction trans = provider.getReadTransaction()) { + 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); @@ -638,7 +657,7 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { writeTransaction.delete(LogicalDatastoreType.CONFIGURATION, staleFlowIId); } - ListenableFuture submitFuture = writeTransaction.submit(); + FluentFuture submitFuture = writeTransaction.commit(); handleStaleEntityDeletionResultFuture(submitFuture); } @@ -649,7 +668,7 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { writeTransaction.delete(LogicalDatastoreType.CONFIGURATION, staleGroupIId); } - ListenableFuture submitFuture = writeTransaction.submit(); + FluentFuture submitFuture = writeTransaction.commit(); handleStaleEntityDeletionResultFuture(submitFuture); } @@ -660,7 +679,7 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation { writeTransaction.delete(LogicalDatastoreType.CONFIGURATION, staleMeterIId); } - ListenableFuture submitFuture = writeTransaction.submit(); + FluentFuture submitFuture = writeTransaction.commit(); handleStaleEntityDeletionResultFuture(submitFuture); } @@ -684,16 +703,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()) { + final KeyedInstanceIdentifier groupIdent = nodeIdentity.child(Group.class, group.key()); + futureList.add(provider.getBundleGroupListener().add(groupIdent, group, nodeIdentity, bundleIdValue)); + } + + for (Table table : flowNode.nonnullTable()) { + final KeyedInstanceIdentifier tableIdent = nodeIdentity.child(Table.class, table.key()); + for (Flow flow : table.nonnullFlow()) { + 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()); } @@ -711,7 +752,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() @@ -722,29 +765,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(); } }