southbound/southbound-impl/src/main/java/org/opendaylight/ovsdb/southbound/reconciliation/ReconciliationManager.java

   1 /*
   2  * Copyright © 2016, 2017 Brocade Communications 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.ovsdb.southbound.reconciliation;
   9
  10 import static java.util.Objects.requireNonNull;
  11
  12 import com.google.common.cache.CacheBuilder;
  13 import com.google.common.cache.CacheLoader;
  14 import com.google.common.cache.LoadingCache;
  15 import com.google.common.util.concurrent.ThreadFactoryBuilder;
  16 import com.google.common.util.concurrent.UncheckedExecutionException;
  17 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
  18 import java.util.Collection;
  19 import java.util.HashMap;
  20 import java.util.List;
  21 import java.util.Map;
  22 import java.util.NoSuchElementException;
  23 import java.util.concurrent.ExecutionException;
  24 import java.util.concurrent.ExecutorService;
  25 import java.util.concurrent.Executors;
  26 import java.util.concurrent.ScheduledExecutorService;
  27 import java.util.concurrent.ThreadFactory;
  28 import java.util.concurrent.TimeUnit;
  29 import org.eclipse.jdt.annotation.NonNull;
  30 import org.opendaylight.mdsal.binding.api.ClusteredDataTreeChangeListener;
  31 import org.opendaylight.mdsal.binding.api.DataBroker;
  32 import org.opendaylight.mdsal.binding.api.DataTreeIdentifier;
  33 import org.opendaylight.mdsal.binding.api.DataTreeModification;
  34 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  35 import org.opendaylight.ovsdb.southbound.InstanceIdentifierCodec;
  36 import org.opendaylight.ovsdb.southbound.OvsdbConnectionInstance;
  37 import org.opendaylight.ovsdb.southbound.OvsdbConnectionManager;
  38 import org.opendaylight.ovsdb.southbound.SouthboundMapper;
  39 import org.opendaylight.ovsdb.southbound.ovsdb.transact.TransactUtils;
  40 import org.opendaylight.ovsdb.southbound.reconciliation.configuration.TerminationPointConfigReconciliationTask;
  41 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.ovsdb.rev150105.OvsdbBridgeAugmentation;
  42 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.ovsdb.rev150105.OvsdbTerminationPointAugmentation;
  43 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
  44 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.NodeKey;
  45 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  46 import org.opendaylight.yangtools.util.concurrent.SpecialExecutors;
  47 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  48 import org.slf4j.Logger;
  49 import org.slf4j.LoggerFactory;
  50
  51 /**
  52  * This class provides the implementation of ovsdb southbound plugins
  53  * configuration reconciliation engine. This engine provide interfaces
  54  * to enqueue (one time retry)/ enqueueForRetry(periodic retry)/ dequeue
  55  * (remove from retry queue) reconciliation task. Reconciliation task can
  56  * be a connection reconciliation or configuration reconciliation of any
  57  * ovsdb managed resource like bridge, termination point etc. This engine
  58  * execute all the reconciliation task through a fixed size thread pool.
  59  * If submitted task need to be retry after a periodic interval they are
  60  * submitted to a single thread executor to periodically wake up and check
  61  * if task is ready for execution.
  62  * Ideally, addition of any type of reconciliation task should not require
  63  * any change in this reconciliation manager execution engine.
  64  * <p>
  65  * 3-Node Cluster:
  66  * Reconciliation manager is agnostic of whether it's running in single
  67  * node cluster or 3-node cluster. It's a responsibility of the task
  68  * submitter to make sure that it submit the task for reconciliation only
  69  * if it's an owner of that device EXCEPT controller initiated Connection.
  70  * Reconciliation of controller initiated connection should be done by all
  71  * the 3-nodes in the cluster, because connection to individual controller
  72  * can be interrupted for various reason.
  73  * </p>
  74  */
  75 public class ReconciliationManager implements AutoCloseable {
  76     private static final Logger LOG = LoggerFactory.getLogger(ReconciliationManager.class);
  77
  78     private static final int NO_OF_RECONCILER = 10;
  79     private static final int RECON_TASK_QUEUE_SIZE = 5000;
  80     private static final long BRIDGE_CACHE_TIMEOUT_IN_SECONDS = 30;
  81
  82     private final DataBroker db;
  83     private final InstanceIdentifierCodec instanceIdentifierCodec;
  84     private final ExecutorService reconcilers;
  85     private final ScheduledExecutorService taskTriager;
  86
  87     // Timeout cache contains the list of bridges to be reconciled for termination points
  88     private LoadingCache<NodeKey, NodeConnectionMetadata> bridgeNodeCache = null;
  89
  90     // Listens for new bridge creations in the operational DS
  91     private ListenerRegistration<BridgeCreatedDataTreeChangeListener> bridgeCreatedDataTreeChangeRegistration = null;
  92
  93     private final ReconciliationTaskManager reconTaskManager = new ReconciliationTaskManager();
  94
  95     public ReconciliationManager(final DataBroker db, final InstanceIdentifierCodec instanceIdentifierCodec) {
  96         this.db = db;
  97         this.instanceIdentifierCodec = instanceIdentifierCodec;
  98         reconcilers = SpecialExecutors.newBoundedCachedThreadPool(NO_OF_RECONCILER, RECON_TASK_QUEUE_SIZE,
  99                 "ovsdb-reconciler", getClass());
 100
 101         ThreadFactory threadFact = new ThreadFactoryBuilder()
 102                 .setNameFormat("ovsdb-recon-task-triager-%d").build();
 103         taskTriager = Executors.newSingleThreadScheduledExecutor(threadFact);
 104
 105         bridgeNodeCache = buildBridgeNodeCache();
 106     }
 107
 108     public boolean isEnqueued(final ReconciliationTask task) {
 109         return reconTaskManager.isTaskQueued(task);
 110     }
 111
 112     public void enqueue(final ReconciliationTask task) {
 113         LOG.trace("Reconciliation task submitted for execution {}",task);
 114         reconTaskManager.cacheTask(task, reconcilers.submit(task));
 115     }
 116
 117     public void enqueueForRetry(final ReconciliationTask task) {
 118         LOG.trace("Reconciliation task re-queued for re-execution {}",task);
 119         reconTaskManager.cacheTask(task, taskTriager.schedule(
 120                 task::checkReadinessAndProcess, task.retryDelayInMills(), TimeUnit.MILLISECONDS
 121             )
 122         );
 123     }
 124
 125     public void dequeue(final ReconciliationTask task) {
 126         reconTaskManager.cancelTask(task);
 127     }
 128
 129     public DataBroker getDb() {
 130         return db;
 131     }
 132
 133     @Override
 134     public void close() throws Exception {
 135         if (this.reconcilers != null) {
 136             this.reconcilers.shutdownNow();
 137         }
 138
 139         if (this.taskTriager != null) {
 140             this.taskTriager.shutdownNow();
 141         }
 142     }
 143
 144     /**
 145      * This method reconciles Termination Point configurations for the given list of bridge nodes.
 146      *
 147      * @param connectionManager OvsdbConnectionManager object
 148      * @param connectionInstance OvsdbConnectionInstance object
 149      * @param bridgeNodes list of bridge nodes be reconciled for termination points
 150      */
 151     public void reconcileTerminationPoints(final OvsdbConnectionManager connectionManager,
 152                                            final OvsdbConnectionInstance connectionInstance,
 153                                            final List<Node> bridgeNodes) {
 154         LOG.debug("Reconcile Termination Point Configuration for Bridges {}", bridgeNodes);
 155         requireNonNull(bridgeNodes, "Bridge Node list must not be null");
 156         if (!bridgeNodes.isEmpty()) {
 157             for (Node node : bridgeNodes) {
 158                 bridgeNodeCache.put(node.key(),
 159                         new NodeConnectionMetadata(node, connectionManager, connectionInstance));
 160             }
 161             registerBridgeCreatedDataTreeChangeListener();
 162         }
 163     }
 164
 165     public void cancelTerminationPointReconciliation() {
 166         cleanupBridgeCreatedDataTreeChangeRegistration();
 167         for (NodeConnectionMetadata nodeConnectionMetadata : bridgeNodeCache.asMap().values()) {
 168             if (nodeConnectionMetadata.getNodeIid() != null) {
 169                 dequeue(new TerminationPointConfigReconciliationTask(
 170                         this,
 171                         nodeConnectionMetadata.getConnectionManager(),
 172                         nodeConnectionMetadata.getNode(),
 173                         nodeConnectionMetadata.getNodeIid(),
 174                         nodeConnectionMetadata.getConnectionInstance(),
 175                         nodeConnectionMetadata.getOperTerminationPoints(),
 176                         instanceIdentifierCodec
 177                 ));
 178             }
 179         }
 180         bridgeNodeCache.invalidateAll();
 181     }
 182
 183     private synchronized void registerBridgeCreatedDataTreeChangeListener() {
 184         if (bridgeCreatedDataTreeChangeRegistration == null) {
 185             BridgeCreatedDataTreeChangeListener bridgeCreatedDataTreeChangeListener =
 186                     new BridgeCreatedDataTreeChangeListener();
 187             InstanceIdentifier<Node> path = SouthboundMapper.createTopologyInstanceIdentifier()
 188                     .child(Node.class);
 189             DataTreeIdentifier<Node> dataTreeIdentifier =
 190                     DataTreeIdentifier.create(LogicalDatastoreType.OPERATIONAL, path);
 191
 192             bridgeCreatedDataTreeChangeRegistration = db.registerDataTreeChangeListener(dataTreeIdentifier,
 193                     bridgeCreatedDataTreeChangeListener);
 194         }
 195     }
 196
 197     private static LoadingCache<NodeKey, NodeConnectionMetadata> buildBridgeNodeCache() {
 198         return CacheBuilder.newBuilder()
 199                 .expireAfterWrite(BRIDGE_CACHE_TIMEOUT_IN_SECONDS, TimeUnit.SECONDS)
 200                 .build(new CacheLoader<NodeKey, NodeConnectionMetadata>() {
 201                     @Override
 202                     public NodeConnectionMetadata load(NodeKey nodeKey) throws Exception {
 203                         // the termination points are explicitly added to the cache, retrieving bridges that are not in
 204                         // the cache results in NoSuchElementException
 205                         throw new NoSuchElementException();
 206                     }
 207                 });
 208     }
 209
 210     /**
 211      * This class listens for bridge creations in the operational data store.
 212      * If the newly created bridge is in the 'bridgeNodeCache', termination point reconciliation for the bridge
 213      * is triggered and the bridge entry is removed from the cache.
 214      * Once cache is empty, either being removed explicitly or expired, the the listener de-registered.
 215      */
 216     class BridgeCreatedDataTreeChangeListener implements ClusteredDataTreeChangeListener<Node> {
 217         @Override
 218         public void onDataTreeChanged(@NonNull Collection<DataTreeModification<Node>> changes) {
 219             bridgeNodeCache.cleanUp();
 220             if (!bridgeNodeCache.asMap().isEmpty()) {
 221                 Map<InstanceIdentifier<OvsdbBridgeAugmentation>, OvsdbBridgeAugmentation> nodes =
 222                         TransactUtils.extractCreated(changes, OvsdbBridgeAugmentation.class);
 223                 Map<InstanceIdentifier<OvsdbTerminationPointAugmentation>, OvsdbTerminationPointAugmentation>
 224                             terminationPointsAug =
 225                         TransactUtils.extractCreated(changes, OvsdbTerminationPointAugmentation.class);
 226                 for (Map.Entry<InstanceIdentifier<OvsdbBridgeAugmentation>, OvsdbBridgeAugmentation> entry :
 227                         nodes.entrySet()) {
 228                     InstanceIdentifier<?> bridgeIid = entry.getKey();
 229                     NodeKey nodeKey = bridgeIid.firstKeyOf(Node.class);
 230                     try {
 231                         NodeConnectionMetadata bridgeNodeMetaData = bridgeNodeCache.get(nodeKey);
 232                         bridgeNodeMetaData.setNodeIid(bridgeIid);
 233                         bridgeNodeMetaData.setOperTerminationPoints(
 234                                 filterTerminationPointsForBridge(nodeKey, terminationPointsAug));
 235                         TerminationPointConfigReconciliationTask tpReconciliationTask =
 236                                 new TerminationPointConfigReconciliationTask(ReconciliationManager.this,
 237                                         bridgeNodeMetaData.getConnectionManager(),
 238                                         bridgeNodeMetaData.getNode(),
 239                                         bridgeIid,
 240                                         bridgeNodeMetaData.getConnectionInstance(),
 241                                         bridgeNodeMetaData.getOperTerminationPoints(),
 242                                         instanceIdentifierCodec);
 243                         enqueue(tpReconciliationTask);
 244                         bridgeNodeCache.invalidate(nodeKey);
 245                     } catch (UncheckedExecutionException ex) {
 246                         // Ignore NoSuchElementException which indicates bridge node is not in the list of
 247                         // pending reconciliation
 248                         if (!(ex.getCause() instanceof NoSuchElementException)) {
 249                             LOG.error("Error getting Termination Point node from LoadingCache", ex);
 250                         }
 251
 252                     } catch (ExecutionException ex) {
 253                         LOG.error("Error getting Termination Point node from LoadingCache", ex);
 254                     }
 255                     if (bridgeNodeCache.asMap().isEmpty()) {
 256                         LOG.debug("De-registering for bridge creation event");
 257                         cleanupBridgeCreatedDataTreeChangeRegistration();
 258                     }
 259                 }
 260             } else {
 261                 LOG.debug("Cache expired - De-registering for bridge creation event");
 262                 cleanupBridgeCreatedDataTreeChangeRegistration();
 263             }
 264         }
 265     }
 266
 267     @SuppressFBWarnings(value = "UPM_UNCALLED_PRIVATE_METHOD",
 268             justification = "https://github.com/spotbugs/spotbugs/issues/811")
 269     private static Map<InstanceIdentifier<OvsdbTerminationPointAugmentation>, OvsdbTerminationPointAugmentation>
 270         filterTerminationPointsForBridge(NodeKey nodeKey,
 271             Map<InstanceIdentifier<OvsdbTerminationPointAugmentation>, OvsdbTerminationPointAugmentation>
 272             terminationPoints) {
 273
 274         Map<InstanceIdentifier<OvsdbTerminationPointAugmentation>, OvsdbTerminationPointAugmentation>
 275                 filteredTerminationPoints = new HashMap<>();
 276         for (Map.Entry<InstanceIdentifier<OvsdbTerminationPointAugmentation>, OvsdbTerminationPointAugmentation> entry :
 277                 terminationPoints.entrySet()) {
 278             InstanceIdentifier<?> bridgeIid = entry.getKey();
 279             NodeKey terminationPointNodeKey = bridgeIid.firstKeyOf(Node.class);
 280             if (terminationPointNodeKey.getNodeId().equals(nodeKey.getNodeId())) {
 281                 LOG.trace("TP Match found: {} {} ", terminationPointNodeKey.getNodeId(), nodeKey.getNodeId());
 282                 filteredTerminationPoints.put(entry.getKey(), entry.getValue());
 283             } else {
 284                 LOG.trace("TP No Match found : {} {} ", terminationPointNodeKey.getNodeId(), nodeKey.getNodeId());
 285             }
 286         }
 287         return filteredTerminationPoints;
 288
 289     }
 290
 291     private void cleanupBridgeCreatedDataTreeChangeRegistration() {
 292         if (bridgeCreatedDataTreeChangeRegistration != null) {
 293             bridgeCreatedDataTreeChangeRegistration.close();
 294             bridgeCreatedDataTreeChangeRegistration = null;
 295         }
 296     }
 297
 298     private static class NodeConnectionMetadata {
 299         private final Node node;
 300         private InstanceIdentifier<?> nodeIid;
 301         private final OvsdbConnectionManager connectionManager;
 302         private final OvsdbConnectionInstance connectionInstance;
 303         private Map<InstanceIdentifier<OvsdbTerminationPointAugmentation>, OvsdbTerminationPointAugmentation>
 304             operTerminationPoints;
 305
 306         public Map<InstanceIdentifier<OvsdbTerminationPointAugmentation>, OvsdbTerminationPointAugmentation>
 307             getOperTerminationPoints() {
 308             return operTerminationPoints;
 309         }
 310
 311         public void setOperTerminationPoints(
 312             Map<InstanceIdentifier<OvsdbTerminationPointAugmentation>, OvsdbTerminationPointAugmentation>
 313                 operTerminationPoints) {
 314             this.operTerminationPoints = operTerminationPoints;
 315         }
 316
 317         NodeConnectionMetadata(Node node,
 318                                OvsdbConnectionManager connectionManager,
 319                                OvsdbConnectionInstance connectionInstance) {
 320             this.node = node;
 321             this.connectionManager = connectionManager;
 322             this.connectionInstance = connectionInstance;
 323         }
 324
 325         public Node getNode() {
 326             return node;
 327         }
 328
 329         public OvsdbConnectionManager getConnectionManager() {
 330             return connectionManager;
 331         }
 332
 333         public OvsdbConnectionInstance getConnectionInstance() {
 334             return connectionInstance;
 335         }
 336
 337         public void setNodeIid(InstanceIdentifier<?> nodeIid) {
 338             this.nodeIid = nodeIid;
 339         }
 340
 341         public InstanceIdentifier<?> getNodeIid() {
 342             return nodeIid;
 343         }
 344     }
 345 }