2 * Copyright © 2016, 2017 Brocade Communications Systems, Inc. and others. All rights reserved.
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
8 package org.opendaylight.ovsdb.southbound.reconciliation;
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;
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.controller.md.sal.binding.api.ClusteredDataTreeChangeListener;
30 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
31 import org.opendaylight.controller.md.sal.binding.api.DataTreeIdentifier;
32 import org.opendaylight.controller.md.sal.binding.api.DataTreeModification;
33 import org.opendaylight.controller.md.sal.common.api.data.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;
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.
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.
74 public class ReconciliationManager implements AutoCloseable {
75 private static final Logger LOG = LoggerFactory.getLogger(ReconciliationManager.class);
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;
81 private final DataBroker db;
82 private final InstanceIdentifierCodec instanceIdentifierCodec;
83 private final ExecutorService reconcilers;
84 private final ScheduledExecutorService taskTriager;
86 // Timeout cache contains the list of bridges to be reconciled for termination points
87 private LoadingCache<NodeKey, NodeConnectionMetadata> bridgeNodeCache = null;
89 // Listens for new bridge creations in the operational DS
90 private ListenerRegistration<BridgeCreatedDataTreeChangeListener> bridgeCreatedDataTreeChangeRegistration = null;
92 private final ReconciliationTaskManager reconTaskManager = new ReconciliationTaskManager();
94 public ReconciliationManager(final DataBroker db, final InstanceIdentifierCodec instanceIdentifierCodec) {
96 this.instanceIdentifierCodec = instanceIdentifierCodec;
97 reconcilers = SpecialExecutors.newBoundedCachedThreadPool(NO_OF_RECONCILER, RECON_TASK_QUEUE_SIZE,
98 "ovsdb-reconciler", getClass());
100 ThreadFactory threadFact = new ThreadFactoryBuilder()
101 .setNameFormat("ovsdb-recon-task-triager-%d").build();
102 taskTriager = Executors.newSingleThreadScheduledExecutor(threadFact);
104 bridgeNodeCache = buildBridgeNodeCache();
107 public boolean isEnqueued(final ReconciliationTask task) {
108 return reconTaskManager.isTaskQueued(task);
111 public void enqueue(final ReconciliationTask task) {
112 LOG.trace("Reconciliation task submitted for execution {}",task);
113 reconTaskManager.cacheTask(task, reconcilers.submit(task));
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
124 public void dequeue(final ReconciliationTask task) {
125 reconTaskManager.cancelTask(task);
128 public DataBroker getDb() {
133 public void close() throws Exception {
134 if (this.reconcilers != null) {
135 this.reconcilers.shutdownNow();
138 if (this.taskTriager != null) {
139 this.taskTriager.shutdownNow();
144 * This method reconciles Termination Point configurations for the given list of bridge nodes.
146 * @param connectionManager OvsdbConnectionManager object
147 * @param connectionInstance OvsdbConnectionInstance object
148 * @param bridgeNodes list of bridge nodes be reconciled for termination points
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));
160 registerBridgeCreatedDataTreeChangeListener();
164 public void cancelTerminationPointReconciliation() {
165 cleanupBridgeCreatedDataTreeChangeRegistration();
166 for (NodeConnectionMetadata nodeConnectionMetadata : bridgeNodeCache.asMap().values()) {
167 if (nodeConnectionMetadata.getNodeIid() != null) {
168 dequeue(new TerminationPointConfigReconciliationTask(
170 nodeConnectionMetadata.getConnectionManager(),
171 nodeConnectionMetadata.getNode(),
172 nodeConnectionMetadata.getNodeIid(),
173 nodeConnectionMetadata.getConnectionInstance(),
174 nodeConnectionMetadata.getOperTerminationPoints(),
175 instanceIdentifierCodec
179 bridgeNodeCache.invalidateAll();
182 private synchronized void registerBridgeCreatedDataTreeChangeListener() {
183 if (bridgeCreatedDataTreeChangeRegistration == null) {
184 BridgeCreatedDataTreeChangeListener bridgeCreatedDataTreeChangeListener =
185 new BridgeCreatedDataTreeChangeListener();
186 InstanceIdentifier<Node> path = SouthboundMapper.createTopologyInstanceIdentifier()
188 DataTreeIdentifier<Node> dataTreeIdentifier =
189 new DataTreeIdentifier<>(LogicalDatastoreType.OPERATIONAL, path);
191 bridgeCreatedDataTreeChangeRegistration = db.registerDataTreeChangeListener(dataTreeIdentifier,
192 bridgeCreatedDataTreeChangeListener);
196 private LoadingCache<NodeKey, NodeConnectionMetadata> buildBridgeNodeCache() {
197 return CacheBuilder.newBuilder()
198 .expireAfterWrite(BRIDGE_CACHE_TIMEOUT_IN_SECONDS, TimeUnit.SECONDS)
199 .build(new CacheLoader<NodeKey, NodeConnectionMetadata>() {
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();
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.
215 class BridgeCreatedDataTreeChangeListener implements ClusteredDataTreeChangeListener<Node> {
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 :
227 InstanceIdentifier<?> bridgeIid = entry.getKey();
228 NodeKey nodeKey = bridgeIid.firstKeyOf(Node.class);
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(),
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);
251 } catch (ExecutionException ex) {
252 LOG.error("Error getting Termination Point node from LoadingCache", ex);
254 if (bridgeNodeCache.asMap().isEmpty()) {
255 LOG.debug("De-registering for bridge creation event");
256 cleanupBridgeCreatedDataTreeChangeRegistration();
260 LOG.debug("Cache expired - De-registering for bridge creation event");
261 cleanupBridgeCreatedDataTreeChangeRegistration();
266 @SuppressFBWarnings(value = "UPM_UNCALLED_PRIVATE_METHOD",
267 justification = "https://github.com/spotbugs/spotbugs/issues/811")
268 private Map<InstanceIdentifier<OvsdbTerminationPointAugmentation>, OvsdbTerminationPointAugmentation>
269 filterTerminationPointsForBridge(NodeKey nodeKey,
270 Map<InstanceIdentifier<OvsdbTerminationPointAugmentation>, OvsdbTerminationPointAugmentation>
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());
283 LOG.trace("TP No Match found : {} {} ", terminationPointNodeKey.getNodeId(), nodeKey.getNodeId());
286 return filteredTerminationPoints;
290 private void cleanupBridgeCreatedDataTreeChangeRegistration() {
291 if (bridgeCreatedDataTreeChangeRegistration != null) {
292 bridgeCreatedDataTreeChangeRegistration.close();
293 bridgeCreatedDataTreeChangeRegistration = null;
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;
305 public Map<InstanceIdentifier<OvsdbTerminationPointAugmentation>, OvsdbTerminationPointAugmentation>
306 getOperTerminationPoints() {
307 return operTerminationPoints;
310 public void setOperTerminationPoints(
311 Map<InstanceIdentifier<OvsdbTerminationPointAugmentation>, OvsdbTerminationPointAugmentation>
312 operTerminationPoints) {
313 this.operTerminationPoints = operTerminationPoints;
316 NodeConnectionMetadata(Node node,
317 OvsdbConnectionManager connectionManager,
318 OvsdbConnectionInstance connectionInstance) {
320 this.connectionManager = connectionManager;
321 this.connectionInstance = connectionInstance;
324 public Node getNode() {
328 public OvsdbConnectionManager getConnectionManager() {
329 return connectionManager;
332 public OvsdbConnectionInstance getConnectionInstance() {
333 return connectionInstance;
336 public void setNodeIid(InstanceIdentifier<?> nodeIid) {
337 this.nodeIid = nodeIid;
340 public InstanceIdentifier<?> getNodeIid() {