applications/forwardingrules-manager/src/main/java/org/opendaylight/openflowplugin/applications/frm/impl/FlowNodeReconciliationImpl.java

   1 /**
   2  * Copyright (c) 2014, 2017 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8
   9 package org.opendaylight.openflowplugin.applications.frm.impl;
  10
  11 import com.google.common.base.Optional;
  12 import com.google.common.base.Preconditions;
  13 import com.google.common.collect.ImmutableList;
  14 import com.google.common.collect.Lists;
  15 import com.google.common.util.concurrent.CheckedFuture;
  16 import com.google.common.util.concurrent.FutureCallback;
  17 import com.google.common.util.concurrent.Futures;
  18 import com.google.common.util.concurrent.JdkFutureAdapters;
  19 import com.google.common.util.concurrent.ListenableFuture;
  20 import java.math.BigInteger;
  21 import java.util.ArrayList;
  22 import java.util.Collection;
  23 import java.util.Collections;
  24 import java.util.HashMap;
  25 import java.util.List;
  26 import java.util.ListIterator;
  27 import java.util.Map;
  28 import java.util.concurrent.Callable;
  29 import java.util.concurrent.ExecutionException;
  30 import java.util.concurrent.ExecutorService;
  31 import java.util.concurrent.Executors;
  32 import java.util.concurrent.Future;
  33 import java.util.concurrent.TimeUnit;
  34 import java.util.concurrent.TimeoutException;
  35 import java.util.concurrent.atomic.AtomicLong;
  36 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  37 import org.opendaylight.controller.md.sal.binding.api.ReadOnlyTransaction;
  38 import org.opendaylight.controller.md.sal.binding.api.WriteTransaction;
  39 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  40 import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
  41 import org.opendaylight.openflowplugin.api.OFConstants;
  42 import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo;
  43 import org.opendaylight.openflowplugin.applications.frm.FlowNodeReconciliation;
  44 import org.opendaylight.openflowplugin.applications.frm.ForwardingRulesManager;
  45 import org.opendaylight.yang.gen.v1.urn.opendaylight.action.types.rev131112.action.action.GroupActionCase;
  46 import org.opendaylight.yang.gen.v1.urn.opendaylight.action.types.rev131112.action.action.OutputActionCase;
  47 import org.opendaylight.yang.gen.v1.urn.opendaylight.action.types.rev131112.action.list.Action;
  48 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
  49 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowId;
  50 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.meters.Meter;
  51 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.meters.MeterBuilder;
  52 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.meters.MeterKey;
  53 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.meters.StaleMeter;
  54 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.meters.StaleMeterKey;
  55 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
  56 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.TableKey;
  57 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow;
  58 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowBuilder;
  59 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowKey;
  60 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.StaleFlow;
  61 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.StaleFlowKey;
  62 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.GroupId;
  63 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.GroupTypes;
  64 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.group.Buckets;
  65 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.group.buckets.Bucket;
  66 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.groups.Group;
  67 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.groups.GroupBuilder;
  68 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.groups.GroupKey;
  69 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.groups.StaleGroup;
  70 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.groups.StaleGroupKey;
  71 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
  72 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
  73 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
  74 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.types.rev130918.MeterId;
  75 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.AddBundleMessagesInput;
  76 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.AddBundleMessagesInputBuilder;
  77 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.ControlBundleInput;
  78 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.ControlBundleInputBuilder;
  79 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.SalBundleService;
  80 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.add.bundle.messages.input.Messages;
  81 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.add.bundle.messages.input.MessagesBuilder;
  82 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.add.bundle.messages.input.messages.Message;
  83 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.add.bundle.messages.input.messages.MessageBuilder;
  84 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.bundle.inner.message.grouping.bundle.inner.message.BundleAddFlowCaseBuilder;
  85 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.bundle.inner.message.grouping.bundle.inner.message.BundleAddGroupCaseBuilder;
  86 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.bundle.inner.message.grouping.bundle.inner.message.BundleRemoveFlowCaseBuilder;
  87 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.bundle.service.rev170124.bundle.inner.message.grouping.bundle.inner.message.BundleRemoveGroupCaseBuilder;
  88 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;
  89 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;
  90 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;
  91 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;
  92 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.rev170124.BundleControlType;
  93 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.rev170124.BundleFlags;
  94 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.onf.rev170124.BundleId;
  95 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflowplugin.rf.state.rev170713.ResultState;
  96 import org.opendaylight.yang.gen.v1.urn.opendaylight.table.types.rev131026.table.features.TableFeatures;
  97 import org.opendaylight.yang.gen.v1.urn.opendaylight.table.types.rev131026.table.features.TableFeaturesKey;
  98 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  99 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
 100 import org.opendaylight.yangtools.yang.common.RpcResult;
 101 import org.slf4j.Logger;
 102 import org.slf4j.LoggerFactory;
 103
 104 /**
 105  * Default implementation of {@link ForwardingRulesManager}
 106  *
 107  * @author <a href="mailto:vdemcak@cisco.com">Vaclav Demcak</a>
 108  */
 109 public class FlowNodeReconciliationImpl implements FlowNodeReconciliation {
 110      private static final Logger LOG = LoggerFactory.getLogger(FlowNodeReconciliationImpl.class);
 111
 112     //The number of nanoseconds to wait for a single group to be added.
 113     private static final long  ADD_GROUP_TIMEOUT = TimeUnit.SECONDS.toNanos(3);
 114
 115      //The maximum number of nanoseconds to wait for completion of add-group RPCs.
 116     private static final long  MAX_ADD_GROUP_TIMEOUT = TimeUnit.SECONDS.toNanos(20);
 117     private static final String SEPARATOR = ":";
 118     private static final int THREAD_POOL_SIZE = 4;
 119
 120     private final DataBroker dataBroker;
 121     private final ForwardingRulesManager provider;
 122     private final String serviceName;
 123     final private int priority;
 124     final private ResultState resultState;
 125     private Map<DeviceInfo, ListenableFuture<Boolean>> futureMap = new HashMap<>();
 126
 127     private final ExecutorService executor = Executors.newFixedThreadPool(THREAD_POOL_SIZE);
 128
 129     private final SalBundleService salBundleService;
 130
 131     private static final AtomicLong BUNDLE_ID = new AtomicLong();
 132     private static final BundleFlags BUNDLE_FLAGS = new BundleFlags(true, true);
 133
 134     public FlowNodeReconciliationImpl (final ForwardingRulesManager manager, final DataBroker db,
 135             final String serviceName, final int priority, final ResultState resultState) {
 136         this.provider = Preconditions.checkNotNull(manager, "ForwardingRulesManager can not be null!");
 137         dataBroker = Preconditions.checkNotNull(db, "DataBroker can not be null!");
 138         this.serviceName = serviceName;
 139         this.priority = priority;
 140         this.resultState = resultState;
 141         salBundleService = Preconditions.checkNotNull(manager.getSalBundleService(),"salBundleService can not be null!");
 142     }
 143
 144     @Override
 145     public void close() {
 146         if (executor != null) {
 147             executor.shutdownNow();
 148         }
 149     }
 150
 151     private ListenableFuture<Boolean> reconcileConfiguration(InstanceIdentifier<FlowCapableNode> connectedNode) {
 152         LOG.info("Triggering reconciliation for device {}", connectedNode.firstKeyOf(Node.class));
 153         if (provider.isStaleMarkingEnabled()) {
 154             LOG.info("Stale-Marking is ENABLED and proceeding with deletion of "
 155                     + "stale-marked entities on switch {}",
 156                     connectedNode.toString());
 157             reconciliationPreProcess(connectedNode);
 158         }
 159         LOG.debug("Bundle based reconciliation status : {}", provider.isBundleBasedReconciliationEnabled()?"Enable":"Disable");
 160         if (provider.isBundleBasedReconciliationEnabled()) {
 161             BundleBasedReconciliationTask bundleBasedReconTask = new BundleBasedReconciliationTask(connectedNode);
 162             return JdkFutureAdapters.listenInPoolThread(executor.submit(bundleBasedReconTask));
 163         } else {
 164             ReconciliationTask reconciliationTask = new ReconciliationTask(connectedNode);
 165             return JdkFutureAdapters.listenInPoolThread(executor.submit(reconciliationTask));
 166         }
 167     }
 168
 169     private class BundleBasedReconciliationTask implements Callable<Boolean> {
 170         final InstanceIdentifier<FlowCapableNode> nodeIdentity;
 171
 172         public BundleBasedReconciliationTask(final InstanceIdentifier<FlowCapableNode> nodeIdent) {
 173             nodeIdentity = nodeIdent;
 174         }
 175
 176         @Override
 177         public Boolean call() {
 178             String sNode = nodeIdentity.firstKeyOf(Node.class, NodeKey.class).getId().getValue();
 179             Optional<FlowCapableNode> flowNode = Optional.absent();
 180             BundleId bundleIdValue = new BundleId(BUNDLE_ID.getAndIncrement());
 181             BigInteger nDpId = getDpnIdFromNodeName(sNode);
 182             LOG.debug("Triggering bundle based reconciliation for device :{}", nDpId);
 183             ReadOnlyTransaction trans = provider.getReadTranaction();
 184             try {
 185                 flowNode = trans.read(LogicalDatastoreType.CONFIGURATION, nodeIdentity).get();
 186             } catch (Exception e) {
 187                 LOG.error("Error occurred while reading the configuration data store for node {}", nodeIdentity, e);
 188             }
 189
 190             if (flowNode.isPresent()) {
 191                 LOG.debug("FlowNode present for Datapath ID {}", nDpId);
 192                 final NodeRef nodeRef = new NodeRef(nodeIdentity.firstIdentifierOf(Node.class));
 193
 194                 final ControlBundleInput openBundleInput = new ControlBundleInputBuilder()
 195                         .setNode(nodeRef)
 196                         .setBundleId(bundleIdValue)
 197                         .setFlags(BUNDLE_FLAGS)
 198                         .setType(BundleControlType.ONFBCTOPENREQUEST)
 199                         .build();
 200
 201                 final ControlBundleInput commitBundleInput = new ControlBundleInputBuilder()
 202                         .setNode(nodeRef)
 203                         .setBundleId(bundleIdValue)
 204                         .setFlags(BUNDLE_FLAGS)
 205                         .setType(BundleControlType.ONFBCTCOMMITREQUEST)
 206                         .build();
 207
 208                 final AddBundleMessagesInput addBundleMessagesInput = new AddBundleMessagesInputBuilder()
 209                         .setNode(nodeRef)
 210                         .setBundleId(bundleIdValue)
 211                         .setFlags(BUNDLE_FLAGS)
 212                         .setMessages(createMessages(nodeRef, flowNode))
 213                         .build();
 214
 215                 Future<RpcResult<Void>> openBundle = salBundleService.controlBundle(openBundleInput);
 216
 217                 ListenableFuture<RpcResult<Void>> addBundleMessagesFuture =
 218                         Futures.transformAsync(JdkFutureAdapters.listenInPoolThread(openBundle), rpcResult -> {
 219                             if (rpcResult.isSuccessful()) {
 220                                 return JdkFutureAdapters.listenInPoolThread(
 221                                         salBundleService.addBundleMessages(addBundleMessagesInput));
 222                             }
 223                             return Futures.immediateFuture(null);
 224                         });
 225
 226                 ListenableFuture<RpcResult<Void>> commitBundleFuture =
 227                         Futures.transformAsync(addBundleMessagesFuture, rpcResult -> {
 228                             if (rpcResult.isSuccessful()) {
 229                                 return JdkFutureAdapters.listenInPoolThread(
 230                                         salBundleService.controlBundle(commitBundleInput));
 231                             }
 232                             return Futures.immediateFuture(null);
 233                         });
 234
 235                 /* Bundles not supported for meters*/
 236                 List<Meter> meters = flowNode.get().getMeter() != null
 237                         ? flowNode.get().getMeter() : Collections.emptyList();
 238                         ListenableFuture<RpcResult<Void>> meterFuture =
 239                                 Futures.transformAsync(commitBundleFuture, rpcResult -> {
 240                                     if (rpcResult.isSuccessful()) {
 241                                         for (Meter meter : meters) {
 242                                             final KeyedInstanceIdentifier<Meter, MeterKey> meterIdent =
 243                                                     nodeIdentity.child(Meter.class, meter.getKey());
 244                                             provider.getMeterCommiter().add(meterIdent, meter, nodeIdentity);
 245                                         }
 246                                     }
 247                                     return Futures.immediateFuture(null);
 248                                 });
 249
 250                         trans.close();
 251                         try {
 252                             if(commitBundleFuture.get().isSuccessful()) {
 253                                 LOG.debug("Completing bundle based reconciliation for device ID:{}", nDpId);
 254                                 return true;
 255                             } else {
 256                                 return false;
 257                             }
 258                         } catch (InterruptedException | ExecutionException e) {
 259                             LOG.error("Error while doing bundle based reconciliation for device ID:{}", nodeIdentity);
 260                             return false;
 261                         }
 262             }
 263             LOG.error("FlowNode not present for Datapath ID {}", nDpId);
 264             return false;
 265         }
 266     }
 267
 268     @Override
 269     public ListenableFuture<Boolean> startReconciliation(DeviceInfo node) {
 270         InstanceIdentifier<FlowCapableNode> connectedNode = node.getNodeInstanceIdentifier()
 271                 .augmentation(FlowCapableNode.class);
 272         return futureMap.computeIfAbsent(node, future -> reconcileConfiguration(connectedNode));
 273     }
 274
 275     @Override
 276     public ListenableFuture<Boolean> endReconciliation(DeviceInfo node) {
 277         futureMap.computeIfPresent(node, (key, future) -> future).cancel(true);
 278         futureMap.remove(node);
 279         return Futures.immediateFuture(true);
 280     }
 281
 282     @Override
 283     public int getPriority() {
 284         return priority;
 285     }
 286
 287     @Override
 288     public String getName() {
 289         return serviceName;
 290     }
 291
 292     @Override
 293     public ResultState getResultState() {
 294         return resultState;
 295     }
 296
 297     private class ReconciliationTask implements Callable<Boolean> {
 298
 299         InstanceIdentifier<FlowCapableNode> nodeIdentity;
 300
 301         public ReconciliationTask(final InstanceIdentifier<FlowCapableNode> nodeIdent) {
 302            nodeIdentity = nodeIdent;
 303         }
 304
 305         public Boolean call() {
 306             String sNode = nodeIdentity.firstKeyOf(Node.class, NodeKey.class).getId().getValue();
 307             BigInteger nDpId = getDpnIdFromNodeName(sNode);
 308
 309             ReadOnlyTransaction trans = provider.getReadTranaction();
 310             Optional<FlowCapableNode> flowNode = Optional.absent();
 311             //initialize the counter
 312             int counter = 0;
 313             try {
 314                 flowNode = trans.read(LogicalDatastoreType.CONFIGURATION, nodeIdentity).get();
 315             } catch (Exception e) {
 316                 LOG.warn("Fail with read Config/DS for Node {} !", nodeIdentity, e);
 317                 return false;
 318             }
 319
 320             if (flowNode.isPresent()) {
 321             /* Tables - have to be pushed before groups */
 322                 // CHECK if while pusing the update, updateTableInput can be null to emulate a table add
 323                 List<TableFeatures> tableList = flowNode.get().getTableFeatures() != null
 324                         ? flowNode.get().getTableFeatures() : Collections.<TableFeatures>emptyList();
 325                 for (TableFeatures tableFeaturesItem : tableList) {
 326                     TableFeaturesKey tableKey = tableFeaturesItem.getKey();
 327                     KeyedInstanceIdentifier<TableFeatures, TableFeaturesKey> tableFeaturesII
 328                             = nodeIdentity.child(TableFeatures.class, new TableFeaturesKey(tableKey.getTableId()));
 329                     provider.getTableFeaturesCommiter().update(tableFeaturesII, tableFeaturesItem, null, nodeIdentity);
 330                 }
 331
 332             /* Groups - have to be first */
 333                 List<Group> groups = flowNode.get().getGroup() != null
 334                         ? flowNode.get().getGroup() : Collections.<Group>emptyList();
 335                 List<Group> toBeInstalledGroups = new ArrayList<>();
 336                 toBeInstalledGroups.addAll(groups);
 337                 //new list for suspected groups pointing to ports .. when the ports come up late
 338                 List<Group> suspectedGroups = new ArrayList<>();
 339                 Map<Long, ListenableFuture<?>> groupFutures = new HashMap<>();
 340
 341                 while ((!(toBeInstalledGroups.isEmpty()) || !(suspectedGroups.isEmpty())) &&
 342                         (counter <= provider.getReconciliationRetryCount())) { //also check if the counter has not crossed the threshold
 343
 344                     if (toBeInstalledGroups.isEmpty() && !suspectedGroups.isEmpty()) {
 345                         LOG.debug("These Groups are pointing to node-connectors that are not up yet {}", suspectedGroups.toString());
 346                         toBeInstalledGroups.addAll(suspectedGroups);
 347                         break;
 348                     }
 349
 350                     ListIterator<Group> iterator = toBeInstalledGroups.listIterator();
 351                     while (iterator.hasNext()) {
 352                         Group group = iterator.next();
 353                         boolean okToInstall = true;
 354                         Buckets buckets = group.getBuckets();
 355                         List<Bucket> bucketList = (buckets == null)
 356                             ? null : buckets.getBucket();
 357                         if (bucketList == null) {
 358                             bucketList = Collections.<Bucket>emptyList();
 359                         }
 360                         for (Bucket bucket : bucketList) {
 361                             List<Action> actions = bucket.getAction();
 362                             if (actions == null) {
 363                                 actions = Collections.<Action>emptyList();
 364                             }
 365                             for (Action action : actions) {
 366                                 //chained-port
 367                                 if (action.getAction().getImplementedInterface().getName()
 368                                         .equals("org.opendaylight.yang.gen.v1.urn.opendaylight.action.types.rev131112.action.action.OutputActionCase")) {
 369                                     String nodeConnectorUri = ((OutputActionCase) (action.getAction()))
 370                                             .getOutputAction().getOutputNodeConnector().getValue();
 371
 372                                     LOG.debug("Installing the group for node connector {}", nodeConnectorUri);
 373
 374                                     //check if the nodeconnector is there in the multimap
 375                                     boolean isPresent = provider.getFlowNodeConnectorInventoryTranslatorImpl()
 376                                             .isNodeConnectorUpdated(nDpId, nodeConnectorUri);
 377                                     //if yes set okToInstall = true
 378
 379                                     if (isPresent) {
 380                                         break;
 381                                     }//else put it in a different list and still set okToInstall = true
 382                                     else {
 383                                         suspectedGroups.add(group);
 384                                         LOG.debug("Not yet received the node-connector updated for {} " +
 385                                                 "for the group with id {}", nodeConnectorUri, group.getGroupId().toString());
 386                                         break;
 387                                     }
 388
 389
 390                                 }
 391                                 //chained groups
 392                                 else if (action.getAction().getImplementedInterface().getName()
 393                                         .equals("org.opendaylight.yang.gen.v1.urn.opendaylight.action.types.rev131112.action.action.GroupActionCase")) {
 394                                     Long groupId = ((GroupActionCase) (action.getAction())).getGroupAction().getGroupId();
 395                                     ListenableFuture<?> future =
 396                                         groupFutures.get(groupId);
 397                                     if (future == null) {
 398                                         okToInstall = false;
 399                                         break;
 400                                     }
 401
 402                                     // Need to ensure that the group specified
 403                                     // by group-action is already installed.
 404                                     awaitGroup(sNode, future);
 405                                 }
 406                             }
 407                             if (!okToInstall) {
 408                                 //increment retry counter value
 409                                 counter++;
 410                                 break;
 411                             }
 412                         }
 413
 414                         if (okToInstall) {
 415                             addGroup(groupFutures, group);
 416                             iterator.remove();
 417                             // resetting the counter to zero
 418                             counter = 0;
 419                         }
 420                     }
 421                 }
 422
 423             /* installation of suspected groups*/
 424                 if (!toBeInstalledGroups.isEmpty()) {
 425                     for (Group group : toBeInstalledGroups) {
 426                         LOG.debug("Installing the group {} finally although the port is not up after checking for {} times "
 427                                 , group.getGroupId().toString(), provider.getReconciliationRetryCount());
 428                         addGroup(groupFutures, group);
 429                     }
 430                 }
 431             /* Meters */
 432                 List<Meter> meters = flowNode.get().getMeter() != null
 433                         ? flowNode.get().getMeter() : Collections.<Meter>emptyList();
 434                 for (Meter meter : meters) {
 435                     final KeyedInstanceIdentifier<Meter, MeterKey> meterIdent =
 436                             nodeIdentity.child(Meter.class, meter.getKey());
 437                     provider.getMeterCommiter().add(meterIdent, meter, nodeIdentity);
 438                 }
 439
 440                 // Need to wait for all groups to be installed before adding
 441                 // flows.
 442                 awaitGroups(sNode, groupFutures.values());
 443
 444             /* Flows */
 445                 List<Table> tables = flowNode.get().getTable() != null
 446                         ? flowNode.get().getTable() : Collections.<Table>emptyList();
 447                 for (Table table : tables) {
 448                     final KeyedInstanceIdentifier<Table, TableKey> tableIdent =
 449                             nodeIdentity.child(Table.class, table.getKey());
 450                     List<Flow> flows = table.getFlow() != null ? table.getFlow() : Collections.<Flow>emptyList();
 451                     for (Flow flow : flows) {
 452                         final KeyedInstanceIdentifier<Flow, FlowKey> flowIdent =
 453                                 tableIdent.child(Flow.class, flow.getKey());
 454                         provider.getFlowCommiter().add(flowIdent, flow, nodeIdentity);
 455                     }
 456                 }
 457             }
 458         /* clean transaction */
 459             trans.close();
 460             return true;
 461         }
 462
 463         /**
 464          * Invoke add-group RPC, and put listenable future associated with the
 465          * RPC into the given map.
 466          *
 467          * @param map        The map to store listenable futures associated with
 468          *                   add-group RPC.
 469          * @param group      The group to add.
 470          */
 471         private void addGroup(Map<Long, ListenableFuture<?>> map, Group group) {
 472             KeyedInstanceIdentifier<Group, GroupKey> groupIdent =
 473                 nodeIdentity.child(Group.class, group.getKey());
 474             final Long groupId = group.getGroupId().getValue();
 475             ListenableFuture<?> future = JdkFutureAdapters.listenInPoolThread(
 476                 provider.getGroupCommiter().add(
 477                     groupIdent, group, nodeIdentity));
 478
 479             Futures.addCallback(future, new FutureCallback<Object>() {
 480                 @Override
 481                 public void onSuccess(Object result) {
 482                     if (LOG.isTraceEnabled()) {
 483                         LOG.trace("add-group RPC completed: node={}, id={}",
 484                                   nodeIdentity.firstKeyOf(Node.class).getId().
 485                                   getValue(), groupId);
 486                     }
 487                 }
 488
 489                 @Override
 490                 public void onFailure(Throwable cause) {
 491                     String msg = "add-group RPC failed: node=" +
 492                         nodeIdentity.firstKeyOf(Node.class).getId().getValue() +
 493                         ", id=" + groupId;
 494                     LOG.debug(msg, cause);
 495                 }
 496             });
 497
 498             map.put(groupId, future);
 499         }
 500
 501         /**
 502          * Wait for completion of add-group RPC.
 503          *
 504          * @param nodeId  The identifier for the target node.
 505          * @param future  Future associated with add-group RPC that installs
 506          *                the target group.
 507          */
 508         private void awaitGroup(String nodeId, ListenableFuture<?> future) {
 509             awaitGroups(nodeId, Collections.singleton(future));
 510         }
 511
 512         /**
 513          * Wait for completion of add-group RPCs.
 514          *
 515          * @param nodeId   The identifier for the target node.
 516          * @param futures  A collection of futures associated with add-group
 517          *                 RPCs.
 518          */
 519         private void awaitGroups(String nodeId,
 520                                  Collection<ListenableFuture<?>> futures) {
 521             if (!futures.isEmpty()) {
 522                 long timeout = Math.min(
 523                     ADD_GROUP_TIMEOUT * futures.size(), MAX_ADD_GROUP_TIMEOUT);
 524                 try {
 525                     Futures.successfulAsList(futures).
 526                         get(timeout, TimeUnit.NANOSECONDS);
 527                     LOG.trace("awaitGroups() completed: node={}", nodeId);
 528                 } catch (TimeoutException e) {
 529                     LOG.debug("add-group RPCs did not complete: node={}",
 530                              nodeId);
 531                 } catch (Exception e) {
 532                     LOG.debug("Unhandled exception while waiting for group installation on node {}",
 533                               nodeId, e);
 534                 }
 535             }
 536         }
 537     }
 538
 539     private BigInteger getDpnIdFromNodeName(String nodeName) {
 540
 541         String dpId = nodeName.substring(nodeName.lastIndexOf(SEPARATOR) + 1);
 542         return new BigInteger(dpId);
 543     }
 544
 545     private void reconciliationPreProcess(final InstanceIdentifier<FlowCapableNode> nodeIdent) {
 546
 547         List<InstanceIdentifier<StaleFlow>> staleFlowsToBeBulkDeleted = Lists.newArrayList();
 548         List<InstanceIdentifier<StaleGroup>> staleGroupsToBeBulkDeleted = Lists.newArrayList();
 549         List<InstanceIdentifier<StaleMeter>> staleMetersToBeBulkDeleted = Lists.newArrayList();
 550
 551
 552         ReadOnlyTransaction trans = provider.getReadTranaction();
 553         Optional<FlowCapableNode> flowNode = Optional.absent();
 554
 555         try {
 556             flowNode = trans.read(LogicalDatastoreType.CONFIGURATION, nodeIdent).get();
 557         }
 558         catch (Exception e) {
 559             LOG.warn("Reconciliation Pre-Processing Fail with read Config/DS for Node {} !", nodeIdent, e);
 560         }
 561
 562         if (flowNode.isPresent()) {
 563
 564             LOG.debug("Proceeding with deletion of stale-marked Flows on switch {} using Openflow interface",
 565                     nodeIdent.toString());
 566             /* Stale-Flows - Stale-marked Flows have to be removed first for safety */
 567             List<Table> tables = flowNode.get().getTable() != null
 568                     ? flowNode.get().getTable() : Collections.<Table> emptyList();
 569             for (Table table : tables) {
 570                 final KeyedInstanceIdentifier<Table, TableKey> tableIdent =
 571                         nodeIdent.child(Table.class, table.getKey());
 572                 List<StaleFlow> staleFlows = table.getStaleFlow() != null ? table.getStaleFlow() : Collections.<StaleFlow> emptyList();
 573                 for (StaleFlow staleFlow : staleFlows) {
 574
 575                     FlowBuilder flowBuilder = new FlowBuilder(staleFlow);
 576                     Flow toBeDeletedFlow = flowBuilder.setId(staleFlow.getId()).build();
 577
 578                     final KeyedInstanceIdentifier<Flow, FlowKey> flowIdent =
 579                             tableIdent.child(Flow.class, toBeDeletedFlow.getKey());
 580
 581
 582                     this.provider.getFlowCommiter().remove(flowIdent, toBeDeletedFlow, nodeIdent);
 583
 584                     staleFlowsToBeBulkDeleted.add(getStaleFlowInstanceIdentifier(staleFlow, nodeIdent));
 585                 }
 586             }
 587
 588
 589             LOG.debug("Proceeding with deletion of stale-marked Groups for switch {} using Openflow interface",
 590                     nodeIdent.toString());
 591
 592             // TODO: Should we collate the futures of RPC-calls to be sure that groups are Flows are fully deleted
 593             // before attempting to delete groups - just in case there are references
 594
 595             /* Stale-marked Groups - Can be deleted after flows */
 596             List<StaleGroup> staleGroups = flowNode.get().getStaleGroup() != null
 597                     ? flowNode.get().getStaleGroup() : Collections.<StaleGroup> emptyList();
 598             for (StaleGroup staleGroup : staleGroups) {
 599
 600                 GroupBuilder groupBuilder = new GroupBuilder(staleGroup);
 601                 Group toBeDeletedGroup = groupBuilder.setGroupId(staleGroup.getGroupId()).build();
 602
 603                 final KeyedInstanceIdentifier<Group, GroupKey> groupIdent =
 604                         nodeIdent.child(Group.class, toBeDeletedGroup.getKey());
 605
 606                 this.provider.getGroupCommiter().remove(groupIdent, toBeDeletedGroup, nodeIdent);
 607
 608                 staleGroupsToBeBulkDeleted.add(getStaleGroupInstanceIdentifier(staleGroup, nodeIdent));
 609             }
 610
 611             LOG.debug("Proceeding with deletion of stale-marked Meters for switch {} using Openflow interface",
 612                     nodeIdent.toString());
 613             /* Stale-marked Meters - can be deleted anytime - so least priority */
 614             List<StaleMeter> staleMeters = flowNode.get().getStaleMeter() != null
 615                     ? flowNode.get().getStaleMeter() : Collections.<StaleMeter> emptyList();
 616
 617             for (StaleMeter staleMeter : staleMeters) {
 618
 619                 MeterBuilder meterBuilder = new MeterBuilder(staleMeter);
 620                 Meter toBeDeletedMeter = meterBuilder.setMeterId(staleMeter.getMeterId()).build();
 621
 622                 final KeyedInstanceIdentifier<Meter, MeterKey> meterIdent =
 623                         nodeIdent.child(Meter.class, toBeDeletedMeter.getKey());
 624
 625
 626                 this.provider.getMeterCommiter().remove(meterIdent, toBeDeletedMeter, nodeIdent);
 627
 628                 staleMetersToBeBulkDeleted.add(getStaleMeterInstanceIdentifier(staleMeter, nodeIdent));
 629             }
 630
 631         }
 632         /* clean transaction */
 633         trans.close();
 634
 635         LOG.debug("Deleting all stale-marked flows/groups/meters of for switch {} in Configuration DS",
 636                 nodeIdent.toString());
 637                 // Now, do the bulk deletions
 638                 deleteDSStaleFlows(staleFlowsToBeBulkDeleted);
 639         deleteDSStaleGroups(staleGroupsToBeBulkDeleted);
 640         deleteDSStaleMeters(staleMetersToBeBulkDeleted);
 641
 642     }
 643
 644
 645     private void deleteDSStaleFlows(List<InstanceIdentifier<StaleFlow>> flowsForBulkDelete){
 646         ImmutableList.Builder<InstanceIdentifier<StaleFlow>> builder = ImmutableList.builder();
 647
 648         WriteTransaction writeTransaction = dataBroker.newWriteOnlyTransaction();
 649
 650         for (InstanceIdentifier<StaleFlow> staleFlowIId : flowsForBulkDelete){
 651             writeTransaction.delete(LogicalDatastoreType.CONFIGURATION, staleFlowIId);
 652         }
 653
 654         CheckedFuture<Void, TransactionCommitFailedException> submitFuture = writeTransaction.submit();
 655         handleStaleEntityDeletionResultFuture(submitFuture);
 656     }
 657
 658     private void deleteDSStaleGroups(List<InstanceIdentifier<StaleGroup>> groupsForBulkDelete){
 659         ImmutableList.Builder<InstanceIdentifier<StaleGroup>> builder = ImmutableList.builder();
 660
 661         WriteTransaction writeTransaction = dataBroker.newWriteOnlyTransaction();
 662
 663         for (InstanceIdentifier<StaleGroup> staleGroupIId : groupsForBulkDelete){
 664             writeTransaction.delete(LogicalDatastoreType.CONFIGURATION, staleGroupIId);
 665         }
 666
 667         CheckedFuture<Void, TransactionCommitFailedException> submitFuture = writeTransaction.submit();
 668         handleStaleEntityDeletionResultFuture(submitFuture);
 669
 670     }
 671
 672     private void deleteDSStaleMeters(List<InstanceIdentifier<StaleMeter>> metersForBulkDelete){
 673         ImmutableList.Builder<InstanceIdentifier<StaleMeter>> builder = ImmutableList.builder();
 674
 675         WriteTransaction writeTransaction = dataBroker.newWriteOnlyTransaction();
 676
 677         for (InstanceIdentifier<StaleMeter> staleMeterIId : metersForBulkDelete){
 678             writeTransaction.delete(LogicalDatastoreType.CONFIGURATION, staleMeterIId);
 679         }
 680
 681         CheckedFuture<Void, TransactionCommitFailedException> submitFuture = writeTransaction.submit();
 682         handleStaleEntityDeletionResultFuture(submitFuture);
 683
 684
 685     }
 686
 687
 688     private InstanceIdentifier<org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.StaleFlow> getStaleFlowInstanceIdentifier(StaleFlow staleFlow, InstanceIdentifier<FlowCapableNode> nodeIdent) {
 689         return nodeIdent
 690                 .child(Table.class, new TableKey(staleFlow.getTableId()))
 691                 .child(org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.StaleFlow.class,
 692                         new StaleFlowKey(new FlowId(staleFlow.getId())));
 693     }
 694
 695     private InstanceIdentifier<org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.groups.StaleGroup> getStaleGroupInstanceIdentifier(StaleGroup staleGroup, InstanceIdentifier<FlowCapableNode> nodeIdent) {
 696         return nodeIdent
 697                 .child(StaleGroup.class, new StaleGroupKey(new GroupId(staleGroup.getGroupId())));
 698     }
 699
 700
 701     private InstanceIdentifier<org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.meters.StaleMeter> getStaleMeterInstanceIdentifier(StaleMeter staleMeter, InstanceIdentifier<FlowCapableNode> nodeIdent) {
 702         return nodeIdent
 703                 .child(StaleMeter.class, new StaleMeterKey(new MeterId(staleMeter.getMeterId())));
 704     }
 705
 706
 707     private void handleStaleEntityDeletionResultFuture(CheckedFuture<Void, TransactionCommitFailedException> submitFuture) {
 708         Futures.addCallback(submitFuture, new FutureCallback<Void>() {
 709             @Override
 710             public void onSuccess(Void result) {
 711                 LOG.debug("Stale entity removal success");
 712             }
 713
 714             @Override
 715             public void onFailure(Throwable t) {
 716                 LOG.debug("Stale entity removal failed {}", t);
 717             }
 718         });
 719     }
 720
 721     private Flow getDeleteAllFlow(){
 722         final FlowBuilder flowBuilder = new FlowBuilder();
 723         flowBuilder.setTableId(OFConstants.OFPTT_ALL);
 724         return flowBuilder.build();
 725     }
 726
 727     private Group getDeleteAllGroup(){
 728         final GroupBuilder groupBuilder = new GroupBuilder();
 729         groupBuilder.setGroupType(GroupTypes.GroupAll);
 730         groupBuilder.setGroupId(new GroupId(OFConstants.OFPG_ALL));
 731         return groupBuilder.build();
 732     }
 733
 734     private Messages createMessages(final NodeRef nodeRef , final Optional<FlowCapableNode> flowNode) {
 735         final List<Message> messages  = new ArrayList<>();
 736         messages.add(new MessageBuilder().setNode(nodeRef).setBundleInnerMessage(
 737                 new BundleRemoveFlowCaseBuilder()
 738                         .setRemoveFlowCaseData(new RemoveFlowCaseDataBuilder(getDeleteAllFlow()).build()).build()).build());
 739
 740         messages.add(new MessageBuilder().setNode(nodeRef).setBundleInnerMessage(
 741                 new BundleRemoveGroupCaseBuilder()
 742                         .setRemoveGroupCaseData(new RemoveGroupCaseDataBuilder(getDeleteAllGroup()).build()).build()).build());
 743
 744         if(flowNode.get().getGroup()!= null) {
 745             for (Group gr : flowNode.get().getGroup()) {
 746                 messages.add(new MessageBuilder().setNode(nodeRef).setBundleInnerMessage(
 747                         new BundleAddGroupCaseBuilder()
 748                                 .setAddGroupCaseData(new AddGroupCaseDataBuilder(gr).build()).build()).build());
 749             }
 750         }
 751
 752         if(flowNode.get().getTable()!= null) {
 753             for (Table table : flowNode.get().getTable()) {
 754                 for (Flow flow : table.getFlow()) {
 755                     messages.add(new MessageBuilder().setNode(nodeRef).setBundleInnerMessage(
 756                             new BundleAddFlowCaseBuilder()
 757                                     .setAddFlowCaseData(new AddFlowCaseDataBuilder(flow).build()).build()).build());
 758                 }
 759             }
 760         }
 761
 762         LOG.debug("The size of the flows and group messages created in createMessage() {}", messages.size());
 763         return new MessagesBuilder().setMessage(messages).build();
 764     }
 765 }
 766