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