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 static java.util.Objects.requireNonNull;
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;
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;
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.
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.
75 public class ReconciliationManager implements AutoCloseable {
76 private static final Logger LOG = LoggerFactory.getLogger(ReconciliationManager.class);
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;
82 private final DataBroker db;
83 private final InstanceIdentifierCodec instanceIdentifierCodec;
84 private final ExecutorService reconcilers;
85 private final ScheduledExecutorService taskTriager;
87 // Timeout cache contains the list of bridges to be reconciled for termination points
88 private LoadingCache<NodeKey, NodeConnectionMetadata> bridgeNodeCache = null;
90 // Listens for new bridge creations in the operational DS
91 private ListenerRegistration<BridgeCreatedDataTreeChangeListener> bridgeCreatedDataTreeChangeRegistration = null;
93 private final ReconciliationTaskManager reconTaskManager = new ReconciliationTaskManager();
95 public ReconciliationManager(final DataBroker db, final InstanceIdentifierCodec instanceIdentifierCodec) {
97 this.instanceIdentifierCodec = instanceIdentifierCodec;
98 reconcilers = SpecialExecutors.newBoundedCachedThreadPool(NO_OF_RECONCILER, RECON_TASK_QUEUE_SIZE,
99 "ovsdb-reconciler", getClass());
101 ThreadFactory threadFact = new ThreadFactoryBuilder()
102 .setNameFormat("ovsdb-recon-task-triager-%d").build();
103 taskTriager = Executors.newSingleThreadScheduledExecutor(threadFact);
105 bridgeNodeCache = buildBridgeNodeCache();
108 public boolean isEnqueued(final ReconciliationTask task) {
109 return reconTaskManager.isTaskQueued(task);
112 public void enqueue(final ReconciliationTask task) {
113 LOG.trace("Reconciliation task submitted for execution {}",task);
114 reconTaskManager.cacheTask(task, reconcilers.submit(task));
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
125 public void dequeue(final ReconciliationTask task) {
126 reconTaskManager.cancelTask(task);
129 public DataBroker getDb() {
134 public void close() throws Exception {
135 if (this.reconcilers != null) {
136 this.reconcilers.shutdownNow();
139 if (this.taskTriager != null) {
140 this.taskTriager.shutdownNow();
145 * This method reconciles Termination Point configurations for the given list of bridge nodes.
147 * @param connectionManager OvsdbConnectionManager object
148 * @param connectionInstance OvsdbConnectionInstance object
149 * @param bridgeNodes list of bridge nodes be reconciled for termination points
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));
161 registerBridgeCreatedDataTreeChangeListener();
165 public void cancelTerminationPointReconciliation() {
166 cleanupBridgeCreatedDataTreeChangeRegistration();
167 for (NodeConnectionMetadata nodeConnectionMetadata : bridgeNodeCache.asMap().values()) {
168 if (nodeConnectionMetadata.getNodeIid() != null) {
169 dequeue(new TerminationPointConfigReconciliationTask(
171 nodeConnectionMetadata.getConnectionManager(),
172 nodeConnectionMetadata.getNode(),
173 nodeConnectionMetadata.getNodeIid(),
174 nodeConnectionMetadata.getConnectionInstance(),
175 nodeConnectionMetadata.getOperTerminationPoints(),
176 instanceIdentifierCodec
180 bridgeNodeCache.invalidateAll();
183 private synchronized void registerBridgeCreatedDataTreeChangeListener() {
184 if (bridgeCreatedDataTreeChangeRegistration == null) {
185 BridgeCreatedDataTreeChangeListener bridgeCreatedDataTreeChangeListener =
186 new BridgeCreatedDataTreeChangeListener();
187 InstanceIdentifier<Node> path = SouthboundMapper.createTopologyInstanceIdentifier()
189 DataTreeIdentifier<Node> dataTreeIdentifier =
190 DataTreeIdentifier.create(LogicalDatastoreType.OPERATIONAL, path);
192 bridgeCreatedDataTreeChangeRegistration = db.registerDataTreeChangeListener(dataTreeIdentifier,
193 bridgeCreatedDataTreeChangeListener);
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>() {
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();
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.
216 class BridgeCreatedDataTreeChangeListener implements ClusteredDataTreeChangeListener<Node> {
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 :
228 InstanceIdentifier<?> bridgeIid = entry.getKey();
229 NodeKey nodeKey = bridgeIid.firstKeyOf(Node.class);
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(),
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);
252 } catch (ExecutionException ex) {
253 LOG.error("Error getting Termination Point node from LoadingCache", ex);
255 if (bridgeNodeCache.asMap().isEmpty()) {
256 LOG.debug("De-registering for bridge creation event");
257 cleanupBridgeCreatedDataTreeChangeRegistration();
261 LOG.debug("Cache expired - De-registering for bridge creation event");
262 cleanupBridgeCreatedDataTreeChangeRegistration();
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>
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());
284 LOG.trace("TP No Match found : {} {} ", terminationPointNodeKey.getNodeId(), nodeKey.getNodeId());
287 return filteredTerminationPoints;
291 private void cleanupBridgeCreatedDataTreeChangeRegistration() {
292 if (bridgeCreatedDataTreeChangeRegistration != null) {
293 bridgeCreatedDataTreeChangeRegistration.close();
294 bridgeCreatedDataTreeChangeRegistration = null;
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;
306 public Map<InstanceIdentifier<OvsdbTerminationPointAugmentation>, OvsdbTerminationPointAugmentation>
307 getOperTerminationPoints() {
308 return operTerminationPoints;
311 public void setOperTerminationPoints(
312 Map<InstanceIdentifier<OvsdbTerminationPointAugmentation>, OvsdbTerminationPointAugmentation>
313 operTerminationPoints) {
314 this.operTerminationPoints = operTerminationPoints;
317 NodeConnectionMetadata(Node node,
318 OvsdbConnectionManager connectionManager,
319 OvsdbConnectionInstance connectionInstance) {
321 this.connectionManager = connectionManager;
322 this.connectionInstance = connectionInstance;
325 public Node getNode() {
329 public OvsdbConnectionManager getConnectionManager() {
330 return connectionManager;
333 public OvsdbConnectionInstance getConnectionInstance() {
334 return connectionInstance;
337 public void setNodeIid(InstanceIdentifier<?> nodeIid) {
338 this.nodeIid = nodeIid;
341 public InstanceIdentifier<?> getNodeIid() {