Bug 8497 - Provide config knob to disable the Forwarding Rule Manager reconciliation
[openflowplugin.git] / applications / forwardingrules-manager / src / main / java / org / opendaylight / openflowplugin / applications / frm / impl / FlowNodeReconciliationImpl.java
index 909f4c1423b3113cea57c6c820cccc44aff45c71..c530713c59c3ca7ddfbafcfb19acd8502930c708 100644 (file)
@@ -1,5 +1,5 @@
 /**
- * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
+ * 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,
@@ -8,28 +8,35 @@
 
 package org.opendaylight.openflowplugin.applications.frm.impl;
 
-import java.math.BigInteger;
-import java.util.*;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.atomic.AtomicInteger;
+
 import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
-import com.google.common.collect.Iterables;
 import com.google.common.collect.Lists;
 import com.google.common.util.concurrent.CheckedFuture;
 import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
-
-import java.util.concurrent.Callable;
-
-import org.opendaylight.controller.md.sal.binding.api.*;
+import com.google.common.util.concurrent.JdkFutureAdapters;
+import com.google.common.util.concurrent.ListenableFuture;
+import java.math.BigInteger;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.ListIterator;
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+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.openflowplugin.applications.frm.FlowNodeReconciliation;
 import org.opendaylight.openflowplugin.applications.frm.ForwardingRulesManager;
-import org.opendaylight.openflowplugin.common.wait.SimpleTaskRetryLooper;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.action.types.rev131112.action.action.GroupActionCase;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.action.types.rev131112.action.action.OutputActionCase;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.action.types.rev131112.action.list.Action;
@@ -48,174 +55,65 @@ import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.ta
 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.StaleFlow;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.StaleFlowKey;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.GroupId;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.group.Buckets;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.group.buckets.Bucket;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.groups.Group;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.groups.GroupBuilder;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.groups.GroupKey;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.groups.StaleGroup;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.groups.StaleGroupKey;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.Nodes;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.types.rev130918.MeterId;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.table.types.rev131026.table.features.TableFeatures;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.table.types.rev131026.table.features.TableFeaturesKey;
-import org.opendaylight.yangtools.concepts.ListenerRegistration;
 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import javax.annotation.Nonnull;
-
-
 /**
- * forwardingrules-manager
- * org.opendaylight.openflowplugin.applications.frm
- *
- * FlowNode Reconciliation Listener
- * Reconciliation for a new FlowNode
+ * Default implementation of {@link ForwardingRulesManager}
  *
  * @author <a href="mailto:vdemcak@cisco.com">Vaclav Demcak</a>
- *
- * Created: Jun 13, 2014
  */
 public class FlowNodeReconciliationImpl implements FlowNodeReconciliation {
+     private static final Logger LOG = LoggerFactory.getLogger(FlowNodeReconciliationImpl.class);
 
-    private static final Logger LOG = LoggerFactory.getLogger(FlowNodeReconciliationImpl.class);
+    //The number of nanoseconds to wait for a single group to be added.
+    private static final long  ADD_GROUP_TIMEOUT = TimeUnit.SECONDS.toNanos(3);
 
-    private final DataBroker dataBroker;
+     //The maximum number of nanoseconds to wait for completion of add-group RPCs.
+    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 final DataBroker dataBroker;
     private final ForwardingRulesManager provider;
-    public static final String SEPARATOR = ":";
-
-    private ListenerRegistration<FlowNodeReconciliationImpl> listenerRegistration;
-
-    private final int THREAD_POOL_SIZE = 4;
-    ExecutorService executor = Executors.newFixedThreadPool(THREAD_POOL_SIZE);
-
-    private static final InstanceIdentifier<FlowCapableNode> II_TO_FLOW_CAPABLE_NODE
-            = InstanceIdentifier.builder(Nodes.class)
-            .child(Node.class)
-            .augmentation(FlowCapableNode.class)
-            .build();
+    private final ExecutorService executor = Executors.newFixedThreadPool(THREAD_POOL_SIZE);
 
     public FlowNodeReconciliationImpl (final ForwardingRulesManager manager, final DataBroker db) {
         this.provider = Preconditions.checkNotNull(manager, "ForwardingRulesManager can not be null!");
         dataBroker = Preconditions.checkNotNull(db, "DataBroker can not be null!");
-        /* Build Path */
-        final InstanceIdentifier<FlowCapableNode> flowNodeWildCardIdentifier = InstanceIdentifier.create(Nodes.class)
-                .child(Node.class).augmentation(FlowCapableNode.class);
-
-        final DataTreeIdentifier<FlowCapableNode> treeId =
-                new DataTreeIdentifier<>(LogicalDatastoreType.OPERATIONAL, flowNodeWildCardIdentifier);
-
-        try {
-        SimpleTaskRetryLooper looper = new SimpleTaskRetryLooper(ForwardingRulesManagerImpl.STARTUP_LOOP_TICK,
-                ForwardingRulesManagerImpl.STARTUP_LOOP_MAX_RETRIES);
-
-            listenerRegistration = looper.loopUntilNoException(new Callable<ListenerRegistration<FlowNodeReconciliationImpl>>() {
-                @Override
-                public ListenerRegistration<FlowNodeReconciliationImpl> call() throws Exception {
-                    return dataBroker.registerDataTreeChangeListener(treeId, FlowNodeReconciliationImpl.this);
-                }
-            });
-        } catch (Exception e) {
-            LOG.warn("data listener registration failed: {}", e.getMessage());
-            LOG.debug("data listener registration failed.. ", e);
-            throw new IllegalStateException("FlowNodeReconciliation startup fail! System needs restart.", e);
-        }
     }
 
     @Override
     public void close() {
-        if (listenerRegistration != null) {
-            try {
-                listenerRegistration.close();
-            } catch (Exception e) {
-                LOG.warn("Error by stop FRM FlowNodeReconilListener: {}", e.getMessage());
-                LOG.debug("Error by stop FRM FlowNodeReconilListener..", e);
-            }
-            listenerRegistration = null;
+        if (executor != null) {
+            executor.shutdownNow();
         }
     }
 
     @Override
-    public void onDataTreeChanged(@Nonnull Collection<DataTreeModification<FlowCapableNode>> changes) {
-        Preconditions.checkNotNull(changes, "Changes may not be null!");
-
-        for (DataTreeModification<FlowCapableNode> change : changes) {
-            final InstanceIdentifier<FlowCapableNode> key = change.getRootPath().getRootIdentifier();
-            final DataObjectModification<FlowCapableNode> mod = change.getRootNode();
-            final InstanceIdentifier<FlowCapableNode> nodeIdent =
-                    key.firstIdentifierOf(FlowCapableNode.class);
-
-            switch (mod.getModificationType()) {
-                case DELETE:
-                    if (mod.getDataAfter() == null) {
-                        remove(key, mod.getDataBefore(), nodeIdent);
-                    }
-                    break;
-                case SUBTREE_MODIFIED:
-                    //NO-OP since we donot need to reconciliate on Node-updated
-                    break;
-                case WRITE:
-                    if (mod.getDataBefore() == null) {
-                        add(key, mod.getDataAfter(), nodeIdent);
-                    }
-                    break;
-                default:
-                    throw new IllegalArgumentException("Unhandled modification type " + mod.getModificationType());
-            }
-        }
-    }
-
-
-
-    public void remove(InstanceIdentifier<FlowCapableNode> identifier, FlowCapableNode del,
-                       InstanceIdentifier<FlowCapableNode> nodeIdent) {
-        if(compareInstanceIdentifierTail(identifier,II_TO_FLOW_CAPABLE_NODE)){
-            if (LOG.isDebugEnabled()) {
-                LOG.debug("Node removed: {}",nodeIdent.firstKeyOf(Node.class).getId().getValue());
-            }
-
-            if ( ! nodeIdent.isWildcarded()) {
-                flowNodeDisconnected(nodeIdent);
-            }
-
+    public void reconcileConfiguration(InstanceIdentifier<FlowCapableNode> connectedNode) {
+        if (provider.isReconciliationDisabled()) {
+            LOG.debug("Reconciliation is disabled by user. Skipping reconciliation of node : {}", connectedNode
+                    .firstKeyOf(Node.class));
+            return;
         }
-    }
-
-    public void add(InstanceIdentifier<FlowCapableNode> identifier, FlowCapableNode add,
-                    InstanceIdentifier<FlowCapableNode> nodeIdent) {
-        if(compareInstanceIdentifierTail(identifier,II_TO_FLOW_CAPABLE_NODE)){
-            if (LOG.isDebugEnabled()) {
-                LOG.debug("Node added: {}",nodeIdent.firstKeyOf(Node.class).getId().getValue());
-            }
-
-            if ( ! nodeIdent.isWildcarded()) {
-                flowNodeConnected(nodeIdent);
-            }
-        }
-    }
-
-    @Override
-    public void flowNodeDisconnected(InstanceIdentifier<FlowCapableNode> disconnectedNode) {
-        provider.unregistrateNode(disconnectedNode);
-    }
-
-    @Override
-    public void flowNodeConnected(InstanceIdentifier<FlowCapableNode> connectedNode) {
-        flowNodeConnected(connectedNode, false);
-    }
-
-    private void flowNodeConnected(InstanceIdentifier<FlowCapableNode> connectedNode, boolean force) {
-        if (force || !provider.isNodeActive(connectedNode)) {
-            provider.registrateNewNode(connectedNode);
-
-            if(!provider.isNodeOwner(connectedNode)) { return; }
-
-            if (provider.getConfiguration().isStaleMarkingEnabled()) {
+        if (provider.isNodeOwner(connectedNode)) {
+            LOG.info("Triggering reconciliation for device {}", connectedNode.firstKeyOf(Node.class));
+            if (provider.isStaleMarkingEnabled()) {
                 LOG.info("Stale-Marking is ENABLED and proceeding with deletion of stale-marked entities on switch {}",
                         connectedNode.toString());
                 reconciliationPreProcess(connectedNode);
@@ -242,9 +140,8 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation {
             ReadOnlyTransaction trans = provider.getReadTranaction();
             Optional<FlowCapableNode> flowNode = Optional.absent();
 
-            AtomicInteger counter = new AtomicInteger();
             //initialize the counter
-            counter.set(0);
+            int counter = 0;
             try {
                 flowNode = trans.read(LogicalDatastoreType.CONFIGURATION, nodeIdentity).get();
             } catch (Exception e) {
@@ -268,12 +165,12 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation {
                         ? flowNode.get().getGroup() : Collections.<Group>emptyList();
                 List<Group> toBeInstalledGroups = new ArrayList<>();
                 toBeInstalledGroups.addAll(groups);
-                List<Long> alreadyInstalledGroupids = new ArrayList<>();
                 //new list for suspected groups pointing to ports .. when the ports come up late
                 List<Group> suspectedGroups = new ArrayList<>();
+                Map<Long, ListenableFuture<?>> groupFutures = new HashMap<>();
 
                 while ((!(toBeInstalledGroups.isEmpty()) || !(suspectedGroups.isEmpty())) &&
-                        (counter.get() <= provider.getConfiguration().getReconciliationRetryCount())) { //also check if the counter has not crossed the threshold
+                        (counter <= provider.getReconciliationRetryCount())) { //also check if the counter has not crossed the threshold
 
                     if (toBeInstalledGroups.isEmpty() && !suspectedGroups.isEmpty()) {
                         LOG.error("These Groups are pointing to node-connectors that are not up yet {}", suspectedGroups.toString());
@@ -285,8 +182,18 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation {
                     while (iterator.hasNext()) {
                         Group group = iterator.next();
                         boolean okToInstall = true;
-                        for (Bucket bucket : group.getBuckets().getBucket()) {
-                            for (Action action : bucket.getAction()) {
+                        Buckets buckets = group.getBuckets();
+                        List<Bucket> bucketList = (buckets == null)
+                            ? null : buckets.getBucket();
+                        if (bucketList == null) {
+                            bucketList = Collections.<Bucket>emptyList();
+                        }
+                        for (Bucket bucket : bucketList) {
+                            List<Action> actions = bucket.getAction();
+                            if (actions == null) {
+                                actions = Collections.<Action>emptyList();
+                            }
+                            for (Action action : actions) {
                                 //chained-port
                                 if (action.getAction().getImplementedInterface().getName()
                                         .equals("org.opendaylight.yang.gen.v1.urn.opendaylight.action.types.rev131112.action.action.OutputActionCase")) {
@@ -316,30 +223,30 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation {
                                 else if (action.getAction().getImplementedInterface().getName()
                                         .equals("org.opendaylight.yang.gen.v1.urn.opendaylight.action.types.rev131112.action.action.GroupActionCase")) {
                                     Long groupId = ((GroupActionCase) (action.getAction())).getGroupAction().getGroupId();
-                                    if (!alreadyInstalledGroupids.contains(groupId)) {
+                                    ListenableFuture<?> future =
+                                        groupFutures.get(groupId);
+                                    if (future == null) {
                                         okToInstall = false;
                                         break;
                                     }
+
+                                    // Need to ensure that the group specified
+                                    // by group-action is already installed.
+                                    awaitGroup(sNode, future);
                                 }
                             }
                             if (!okToInstall) {
                                 //increment retry counter value
-                                counter.incrementAndGet();
+                                counter++;
                                 break;
                             }
-
-
                         }
 
-
                         if (okToInstall) {
-                            final KeyedInstanceIdentifier<Group, GroupKey> groupIdent =
-                                    nodeIdentity.child(Group.class, group.getKey());
-                            provider.getGroupCommiter().add(groupIdent, group, nodeIdentity);
-                            alreadyInstalledGroupids.add(group.getGroupId().getValue());
+                            addGroup(groupFutures, group);
                             iterator.remove();
                             // resetting the counter to zero
-                            counter.set(0);
+                            counter = 0;
                         }
                     }
                 }
@@ -348,10 +255,8 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation {
                 if (!toBeInstalledGroups.isEmpty()) {
                     for (Group group : toBeInstalledGroups) {
                         LOG.error("Installing the group {} finally although the port is not up after checking for {} times "
-                                , group.getGroupId().toString(), provider.getConfiguration().getReconciliationRetryCount());
-                        final KeyedInstanceIdentifier<Group, GroupKey> groupIdent =
-                                nodeIdentity.child(Group.class, group.getKey());
-                        provider.getGroupCommiter().add(groupIdent, group, nodeIdentity);
+                                , group.getGroupId().toString(), provider.getReconciliationRetryCount());
+                        addGroup(groupFutures, group);
                     }
                 }
             /* Meters */
@@ -362,6 +267,11 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation {
                             nodeIdentity.child(Meter.class, meter.getKey());
                     provider.getMeterCommiter().add(meterIdent, meter, nodeIdentity);
                 }
+
+                // Need to wait for all groups to be installed before adding
+                // flows.
+                awaitGroups(sNode, groupFutures.values());
+
             /* Flows */
                 List<Table> tables = flowNode.get().getTable() != null
                         ? flowNode.get().getTable() : Collections.<Table>emptyList();
@@ -379,8 +289,85 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation {
         /* clean transaction */
             trans.close();
         }
+
+        /**
+         * Invoke add-group RPC, and put listenable future associated with the
+         * RPC into the given map.
+         *
+         * @param map        The map to store listenable futures associated with
+         *                   add-group RPC.
+         * @param group      The group to add.
+         */
+        private void addGroup(Map<Long, ListenableFuture<?>> map, Group group) {
+            KeyedInstanceIdentifier<Group, GroupKey> groupIdent =
+                nodeIdentity.child(Group.class, group.getKey());
+            final Long groupId = group.getGroupId().getValue();
+            ListenableFuture<?> future = JdkFutureAdapters.listenInPoolThread(
+                provider.getGroupCommiter().add(
+                    groupIdent, group, nodeIdentity));
+
+            Futures.addCallback(future, new FutureCallback<Object>() {
+                @Override
+                public void onSuccess(Object result) {
+                    if (LOG.isTraceEnabled()) {
+                        LOG.trace("add-group RPC completed: node={}, id={}",
+                                  nodeIdentity.firstKeyOf(Node.class).getId().
+                                  getValue(), groupId);
+                    }
+                }
+
+                @Override
+                public void onFailure(Throwable cause) {
+                    String msg = "add-group RPC failed: node=" +
+                        nodeIdentity.firstKeyOf(Node.class).getId().getValue() +
+                        ", id=" + groupId;
+                    LOG.error(msg, cause);
+                }
+            });
+
+            map.put(groupId, future);
+        }
+
+        /**
+         * Wait for completion of add-group RPC.
+         *
+         * @param nodeId  The identifier for the target node.
+         * @param future  Future associated with add-group RPC that installs
+         *                the target group.
+         */
+        private void awaitGroup(String nodeId, ListenableFuture<?> future) {
+            awaitGroups(nodeId, Collections.singleton(future));
+        }
+
+        /**
+         * Wait for completion of add-group RPCs.
+         *
+         * @param nodeId   The identifier for the target node.
+         * @param futures  A collection of futures associated with add-group
+         *                 RPCs.
+         */
+        private void awaitGroups(String nodeId,
+                                 Collection<ListenableFuture<?>> futures) {
+            if (!futures.isEmpty()) {
+                long timeout = Math.min(
+                    ADD_GROUP_TIMEOUT * futures.size(), MAX_ADD_GROUP_TIMEOUT);
+                try {
+                    Futures.successfulAsList(futures).
+                        get(timeout, TimeUnit.NANOSECONDS);
+                    LOG.trace("awaitGroups() completed: node={}", nodeId);
+                } catch (TimeoutException e) {
+                    LOG.warn("add-group RPCs did not complete: node={}",
+                             nodeId);
+                } catch (Exception e) {
+                    LOG.error("Unhandled exception while waiting for group installation on node {}",
+                              nodeId, e);
+                }
+            }
+        }
     }
+
     private BigInteger getDpnIdFromNodeName(String nodeName) {
+
         String dpId = nodeName.substring(nodeName.lastIndexOf(SEPARATOR) + 1);
         return new BigInteger(dpId);
     }
@@ -487,7 +474,6 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation {
 
     private void deleteDSStaleFlows(List<InstanceIdentifier<StaleFlow>> flowsForBulkDelete){
         ImmutableList.Builder<InstanceIdentifier<StaleFlow>> builder = ImmutableList.builder();
-        ImmutableList<InstanceIdentifier<StaleFlow>> bulkDelFlows = builder.addAll(flowsForBulkDelete.iterator()).build();
 
         WriteTransaction writeTransaction = dataBroker.newWriteOnlyTransaction();
 
@@ -501,7 +487,6 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation {
 
     private void deleteDSStaleGroups(List<InstanceIdentifier<StaleGroup>> groupsForBulkDelete){
         ImmutableList.Builder<InstanceIdentifier<StaleGroup>> builder = ImmutableList.builder();
-        ImmutableList<InstanceIdentifier<StaleGroup>> bulkDelGroups = builder.addAll(groupsForBulkDelete.iterator()).build();
 
         WriteTransaction writeTransaction = dataBroker.newWriteOnlyTransaction();
 
@@ -516,7 +501,6 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation {
 
     private void deleteDSStaleMeters(List<InstanceIdentifier<StaleMeter>> metersForBulkDelete){
         ImmutableList.Builder<InstanceIdentifier<StaleMeter>> builder = ImmutableList.builder();
-        ImmutableList<InstanceIdentifier<StaleMeter>> bulkDelGroups = builder.addAll(metersForBulkDelete.iterator()).build();
 
         WriteTransaction writeTransaction = dataBroker.newWriteOnlyTransaction();
 
@@ -553,7 +537,8 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation {
     private void handleStaleEntityDeletionResultFuture(CheckedFuture<Void, TransactionCommitFailedException> submitFuture) {
         Futures.addCallback(submitFuture, new FutureCallback<Void>() {
             @Override
-            public void onSuccess(Void result) { LOG.debug("Stale entity removal success");
+            public void onSuccess(Void result) {
+                LOG.debug("Stale entity removal success");
             }
 
             @Override
@@ -561,13 +546,6 @@ public class FlowNodeReconciliationImpl implements FlowNodeReconciliation {
                 LOG.error("Stale entity removal failed {}", t);
             }
         });
-
-    }
-
-
-    private boolean compareInstanceIdentifierTail(InstanceIdentifier<?> identifier1,
-                                                  InstanceIdentifier<?> identifier2) {
-        return Iterables.getLast(identifier1.getPathArguments()).equals(Iterables.getLast(identifier2.getPathArguments()));
     }
 }