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