2 * Copyright (c) 2013 Cisco 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.bgpcep.bgp.topology.provider;
10 import static java.util.Objects.requireNonNull;
12 import com.google.common.annotations.VisibleForTesting;
13 import com.google.common.base.Preconditions;
14 import com.google.common.util.concurrent.FluentFuture;
15 import com.google.common.util.concurrent.FutureCallback;
16 import com.google.common.util.concurrent.MoreExecutors;
17 import java.util.Collection;
18 import java.util.Collections;
19 import java.util.concurrent.atomic.AtomicBoolean;
20 import org.checkerframework.checker.lock.qual.GuardedBy;
21 import org.opendaylight.bgpcep.topology.TopologyReference;
22 import org.opendaylight.mdsal.binding.api.ClusteredDataTreeChangeListener;
23 import org.opendaylight.mdsal.binding.api.DataBroker;
24 import org.opendaylight.mdsal.binding.api.DataObjectModification;
25 import org.opendaylight.mdsal.binding.api.DataTreeIdentifier;
26 import org.opendaylight.mdsal.binding.api.DataTreeModification;
27 import org.opendaylight.mdsal.binding.api.ReadWriteTransaction;
28 import org.opendaylight.mdsal.binding.api.Transaction;
29 import org.opendaylight.mdsal.binding.api.TransactionChain;
30 import org.opendaylight.mdsal.binding.api.TransactionChainListener;
31 import org.opendaylight.mdsal.binding.api.WriteTransaction;
32 import org.opendaylight.mdsal.common.api.CommitInfo;
33 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
34 import org.opendaylight.protocol.bgp.rib.RibReference;
35 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.rib.rev180329.Route;
36 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.rib.rev180329.bgp.rib.rib.LocRib;
37 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.rib.rev180329.rib.Tables;
38 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.rib.rev180329.rib.TablesKey;
39 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.types.rev180329.AddressFamily;
40 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.types.rev180329.SubsequentAddressFamily;
41 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NetworkTopology;
42 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.TopologyId;
43 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.Topology;
44 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyBuilder;
45 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyKey;
46 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.TopologyTypes;
47 import org.opendaylight.yangtools.concepts.ListenerRegistration;
48 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
49 import org.slf4j.Logger;
50 import org.slf4j.LoggerFactory;
52 public abstract class AbstractTopologyBuilder<T extends Route> implements ClusteredDataTreeChangeListener<T>,
53 TopologyReference, TransactionChainListener {
54 private static final Logger LOG = LoggerFactory.getLogger(AbstractTopologyBuilder.class);
55 // we limit the listener reset interval to be 5 min at most
56 private static final long LISTENER_RESET_LIMIT_IN_MILLSEC = 5 * 60 * 1000;
57 private static final int LISTENER_RESET_ENFORCE_COUNTER = 3;
58 private final InstanceIdentifier<Topology> topology;
59 private final RibReference locRibReference;
60 private final DataBroker dataProvider;
61 private final Class<? extends AddressFamily> afi;
62 private final Class<? extends SubsequentAddressFamily> safi;
63 private final TopologyKey topologyKey;
64 private final TopologyTypes topologyTypes;
65 private final long listenerResetLimitInMillsec;
66 private final int listenerResetEnforceCounter;
69 private ListenerRegistration<AbstractTopologyBuilder<T>> listenerRegistration = null;
71 private TransactionChain chain = null;
72 private final AtomicBoolean closed = new AtomicBoolean(false);
75 protected long listenerScheduledRestartTime = 0;
78 protected int listenerScheduledRestartEnforceCounter = 0;
80 protected AbstractTopologyBuilder(final DataBroker dataProvider, final RibReference locRibReference,
81 final TopologyId topologyId, final TopologyTypes types, final Class<? extends AddressFamily> afi,
82 final Class<? extends SubsequentAddressFamily> safi, final long listenerResetLimitInMillsec,
83 final int listenerResetEnforceCounter) {
84 this.dataProvider = dataProvider;
85 this.locRibReference = requireNonNull(locRibReference);
86 this.topologyKey = new TopologyKey(requireNonNull(topologyId));
87 this.topologyTypes = types;
90 this.listenerResetLimitInMillsec = listenerResetLimitInMillsec;
91 this.listenerResetEnforceCounter = listenerResetEnforceCounter;
92 this.topology = InstanceIdentifier.builder(NetworkTopology.class)
93 .child(Topology.class, this.topologyKey).build();
96 protected AbstractTopologyBuilder(final DataBroker dataProvider, final RibReference locRibReference,
97 final TopologyId topologyId, final TopologyTypes types, final Class<? extends AddressFamily> afi,
98 final Class<? extends SubsequentAddressFamily> safi) {
99 this(dataProvider, locRibReference, topologyId, types, afi, safi, LISTENER_RESET_LIMIT_IN_MILLSEC,
100 LISTENER_RESET_ENFORCE_COUNTER);
103 public final synchronized void start() {
104 LOG.debug("Initiating topology builder from {} at {}. AFI={}, SAFI={}", this.locRibReference, this.topology,
105 this.afi, this.safi);
106 initTransactionChain();
107 initOperationalTopology();
108 registerDataChangeListener();
112 * Register to data tree change listener.
114 private synchronized void registerDataChangeListener() {
115 Preconditions.checkState(this.listenerRegistration == null,
116 "Topology Listener on topology %s has been registered before.",
117 this.getInstanceIdentifier());
118 final InstanceIdentifier<Tables> tablesId = this.locRibReference.getInstanceIdentifier()
119 .child(LocRib.class).child(Tables.class, new TablesKey(this.afi, this.safi));
120 final DataTreeIdentifier<T> id = DataTreeIdentifier.create(LogicalDatastoreType.OPERATIONAL,
121 getRouteWildcard(tablesId));
123 this.listenerRegistration = this.dataProvider.registerDataTreeChangeListener(id, this);
124 LOG.debug("Registered listener {} on topology {}. Timestamp={}", this, this.getInstanceIdentifier(),
125 this.listenerScheduledRestartTime);
129 * Unregister to data tree change listener.
131 private synchronized void unregisterDataChangeListener() {
132 if (this.listenerRegistration != null) {
133 LOG.debug("Unregistered listener {} on topology {}", this, this.getInstanceIdentifier());
134 this.listenerRegistration.close();
135 this.listenerRegistration = null;
139 protected abstract InstanceIdentifier<T> getRouteWildcard(InstanceIdentifier<Tables> tablesId);
141 protected abstract void createObject(ReadWriteTransaction trans, InstanceIdentifier<T> id, T value);
143 protected abstract void removeObject(ReadWriteTransaction trans, InstanceIdentifier<T> id, T value);
145 protected abstract void clearTopology();
148 public final InstanceIdentifier<Topology> getInstanceIdentifier() {
149 return this.topology;
152 public final synchronized FluentFuture<? extends CommitInfo> close() {
153 if (this.closed.getAndSet(true)) {
154 LOG.trace("Transaction chain was already closed.");
155 return CommitInfo.emptyFluentFuture();
157 LOG.info("Shutting down builder for {}", getInstanceIdentifier());
158 unregisterDataChangeListener();
159 final FluentFuture<? extends CommitInfo> future = destroyOperationalTopology();
160 destroyTransactionChain();
165 @SuppressWarnings("checkstyle:IllegalCatch")
166 public synchronized void onDataTreeChanged(final Collection<DataTreeModification<T>> changes) {
167 if (this.closed.get()) {
168 LOG.trace("Transaction chain was already closed, skipping update.");
171 // check if the transaction chain needed to be restarted due to a previous error
172 if (restartTransactionChainOnDemand()) {
173 LOG.debug("The data change {} is disregarded due to restart of listener {}", changes, this);
176 final ReadWriteTransaction trans = this.chain.newReadWriteTransaction();
177 LOG.trace("Received data change {} event with transaction {}", changes, trans.getIdentifier());
178 final AtomicBoolean transactionInError = new AtomicBoolean(false);
179 for (final DataTreeModification<T> change : changes) {
181 routeChanged(change, trans);
182 } catch (final RuntimeException exc) {
183 LOG.warn("Data change {} (transaction {}) was not completely propagated to listener {}", change,
184 trans.getIdentifier(), this, exc);
185 // trans.cancel() is not supported by PingPongTransactionChain, so we just skip the problematic change
186 // trans.commit() must be called first to unlock the current transaction chain, to make the chain
187 // closable so we cannot exit the #onDataTreeChanged() yet
188 transactionInError.set(true);
192 trans.commit().addCallback(new FutureCallback<CommitInfo>() {
194 public void onSuccess(final CommitInfo result) {
195 // as we are enforcing trans.commit(), in some cases the transaction execution actually could be
196 // successfully even when an exception is captured, thus #onTransactionChainFailed() never get invoked.
197 // Though the transaction chain remains usable,
198 // the data loss will not be able to be recovered. Thus we schedule a listener restart here
199 if (transactionInError.get()) {
200 LOG.warn("Transaction {} committed successfully while exception captured. Rescheduling a restart"
201 + " of listener {}", trans
202 .getIdentifier(), AbstractTopologyBuilder.this);
203 scheduleListenerRestart();
205 LOG.trace("Transaction {} committed successfully", trans.getIdentifier());
210 public void onFailure(final Throwable throwable) {
211 // we do nothing but print out the log. Transaction chain restart will be done in
212 // #onTransactionChainFailed()
213 LOG.error("Failed to propagate change (transaction {}) by listener {}", trans.getIdentifier(),
214 AbstractTopologyBuilder.this, throwable);
216 }, MoreExecutors.directExecutor());
220 protected void routeChanged(final DataTreeModification<T> change, final ReadWriteTransaction trans) {
221 final DataObjectModification<T> root = change.getRootNode();
222 switch (root.getModificationType()) {
224 removeObject(trans, change.getRootPath().getRootIdentifier(), root.getDataBefore());
226 case SUBTREE_MODIFIED:
228 if (root.getDataBefore() != null) {
229 removeObject(trans, change.getRootPath().getRootIdentifier(), root.getDataBefore());
231 createObject(trans, change.getRootPath().getRootIdentifier(), root.getDataAfter());
234 throw new IllegalArgumentException("Unhandled modification type " + root.getModificationType());
238 private synchronized void initOperationalTopology() {
239 requireNonNull(this.chain, "A valid transaction chain must be provided.");
240 final WriteTransaction trans = this.chain.newWriteOnlyTransaction();
241 trans.put(LogicalDatastoreType.OPERATIONAL, this.topology,
242 new TopologyBuilder().withKey(this.topologyKey).setServerProvided(Boolean.TRUE)
243 .setTopologyTypes(this.topologyTypes)
244 .setLink(Collections.emptyList()).setNode(Collections.emptyList()).build(), true);
245 trans.commit().addCallback(new FutureCallback<CommitInfo>() {
247 public void onSuccess(final CommitInfo result) {
248 LOG.trace("Transaction {} committed successfully", trans.getIdentifier());
252 public void onFailure(final Throwable throwable) {
253 LOG.error("Failed to initialize topology {} (transaction {}) by listener {}",
254 AbstractTopologyBuilder.this.topology,
255 trans.getIdentifier(), AbstractTopologyBuilder.this, throwable);
257 }, MoreExecutors.directExecutor());
261 * Destroy the current operational topology data. Note a valid transaction must be provided.
263 private synchronized FluentFuture<? extends CommitInfo> destroyOperationalTopology() {
264 requireNonNull(this.chain, "A valid transaction chain must be provided.");
265 final WriteTransaction trans = this.chain.newWriteOnlyTransaction();
266 trans.delete(LogicalDatastoreType.OPERATIONAL, getInstanceIdentifier());
267 final FluentFuture<? extends CommitInfo> future = trans.commit();
268 future.addCallback(new FutureCallback<CommitInfo>() {
270 public void onSuccess(final CommitInfo result) {
271 LOG.trace("Operational topology removed {}", AbstractTopologyBuilder.this.topology);
275 public void onFailure(final Throwable throwable) {
276 LOG.error("Unable to reset operational topology {} (transaction {})",
277 AbstractTopologyBuilder.this.topology, trans.getIdentifier(), throwable);
279 }, MoreExecutors.directExecutor());
285 * Reset a transaction chain by closing the current chain and starting a new one.
287 private synchronized void initTransactionChain() {
288 LOG.debug("Initializing transaction chain for topology {}", this);
289 Preconditions.checkState(this.chain == null,
290 "Transaction chain has to be closed before being initialized");
291 this.chain = this.dataProvider.createMergingTransactionChain(this);
295 * Destroy the current transaction chain.
297 private synchronized void destroyTransactionChain() {
298 if (this.chain != null) {
299 LOG.debug("Destroy transaction chain for topology {}", this);
300 // we cannot close the transaction chain, as it will close the AbstractDOMForwardedTransactionFactory
301 // and the transaction factory cannot be reopen even if we recreate the transaction chain
302 // so we abandon the chain directly
303 // FIXME we want to close the transaction chain gracefully once the PingPongTransactionChain get improved
304 // and the above problem get resolved.
306 // this.chain.close();
307 // } catch (Exception e) {
308 // // the close() may not succeed when the transaction chain is locked
309 // LOG.error("Unable to close transaction chain {} for topology builder {}", this.chain,
310 // getInstanceIdentifier());
317 * Reset the data change listener to its initial status.
318 * By resetting the listener we will be able to recover all the data lost before
321 protected synchronized void resetListener() {
322 requireNonNull(this.listenerRegistration, "Listener on topology " + this + " hasn't been initialized.");
323 LOG.debug("Resetting data change listener for topology builder {}", getInstanceIdentifier());
324 // unregister current listener to prevent incoming data tree change first
325 unregisterDataChangeListener();
326 // create new transaction chain to reset the chain status
327 resetTransactionChain();
328 // reset the operational topology data so that we can have clean status
329 destroyOperationalTopology();
330 initOperationalTopology();
331 // re-register the data change listener to reset the operational topology
332 // we are expecting to receive all the pre-exist route change on the next onDataTreeChanged() call
333 registerDataChangeListener();
337 * Reset the transaction chain only so that the PingPong transaction chain will become usable again.
338 * However, there will be data loss if we do not apply the previous failed transaction again
341 protected synchronized void resetTransactionChain() {
342 LOG.debug("Resetting transaction chain for topology builder {}", getInstanceIdentifier());
343 destroyTransactionChain();
344 initTransactionChain();
348 * There are a few reasons we want to schedule a listener restart in a delayed manner:
349 * 1. we should avoid restarting the listener as when the topology is big, there might be huge overhead
350 * rebuilding the whole linkstate topology again and again
351 * 2. the #onTransactionChainFailed() normally get invoked after a delay. During that time gap, more
352 * data changes might still be pushed to #onDataTreeChanged(). And because #onTransactionChainFailed()
353 * is not invoked yet, listener restart/transaction chain restart is not done. Thus the new changes
354 * will still cause error and another #onTransactionChainFailed() might be invoked later. The listener
355 * will be restarted again in that case, which is unexpected. Restarting of transaction chain only introduce
356 * little overhead and it's okay to be restarted within a small time window.
357 * Note: when the listener is restarted, we can disregard all the incoming data changes before the restart is
358 * done, as after the listener unregister/reregister, the first #onDataTreeChanged() call will contain the a
359 * complete set of existing changes
361 * @return if the listener get restarted, return true; otherwise false
364 protected synchronized boolean restartTransactionChainOnDemand() {
365 if (this.listenerScheduledRestartTime > 0) {
366 // when the #this.listenerScheduledRestartTime timer timed out we can reset the listener,
367 // otherwise we should only reset the transaction chain
368 if (System.currentTimeMillis() > this.listenerScheduledRestartTime) {
369 // reset the the restart timer
370 this.listenerScheduledRestartTime = 0;
371 this.listenerScheduledRestartEnforceCounter = 0;
376 resetTransactionChain();
382 protected synchronized void scheduleListenerRestart() {
383 if (0 == this.listenerScheduledRestartTime) {
384 this.listenerScheduledRestartTime = System.currentTimeMillis() + this.listenerResetLimitInMillsec;
385 } else if (System.currentTimeMillis() > this.listenerScheduledRestartTime
386 && ++this.listenerScheduledRestartEnforceCounter < this.listenerResetEnforceCounter) {
387 // if the transaction failure happens again, we will delay the listener restart up to
388 // #LISTENER_RESET_LIMIT_IN_MILLSEC times
389 this.listenerScheduledRestartTime += this.listenerResetLimitInMillsec;
391 LOG.debug("A listener restart was scheduled at {} (current system time is {})",
392 this.listenerScheduledRestartTime, System.currentTimeMillis());
396 public final synchronized void onTransactionChainFailed(final TransactionChain transactionChain,
397 final Transaction transaction, final Throwable cause) {
398 LOG.error("Topology builder for {} failed in transaction {}.", getInstanceIdentifier(),
399 transaction != null ? transaction.getIdentifier() : null, cause);
400 scheduleListenerRestart();
401 restartTransactionChainOnDemand();
405 public final void onTransactionChainSuccessful(final TransactionChain transactionChain) {
406 LOG.info("Topology builder for {} shut down", getInstanceIdentifier());