Improve TopologyStatsProvider transaction chain failure

Create new chain when onTransactionChainFailed
for TopologyStatsProvider.

Change-Id: Ib3ff3b7b1c9381efa8f35cbb5ed88195320f06c9
Signed-off-by: Claudio D. Gasparini <claudio.gasparini@pantheon.tech>

diff --git a/pcep/topology/topology-stats/src/main/java/org/opendaylight/bgpcep/pcep/topology/stats/provider/TopologyStatsProviderImpl.java b/pcep/topology/topology-stats/src/main/java/org/opendaylight/bgpcep/pcep/topology/stats/provider/TopologyStatsProviderImpl.java
index 05791767492c09d47edc98c9e04b47f060cfff4b..e7f27a02f8db2fa5f2c1d9b5f29d354b43efd127 100644 (file)
--- a/pcep/topology/topology-stats/src/main/java/org/opendaylight/bgpcep/pcep/topology/stats/provider/TopologyStatsProviderImpl.java
+++ b/pcep/topology/topology-stats/src/main/java/org/opendaylight/bgpcep/pcep/topology/stats/provider/TopologyStatsProviderImpl.java
@@ -11,6 +11,9 @@ package org.opendaylight.bgpcep.pcep.topology.stats.provider;
 import static java.util.Objects.requireNonNull;
 import static java.util.concurrent.TimeUnit.SECONDS;
 
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.MoreExecutors;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.TimerTask;
@@ -18,6 +21,7 @@ import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
 import javax.annotation.Nonnull;
 import javax.annotation.concurrent.GuardedBy;
 import org.opendaylight.bgpcep.pcep.topology.spi.stats.TopologySessionStatsRegistry;
@@ -50,6 +54,7 @@ public final class TopologyStatsProviderImpl implements TransactionChainListener
     private BindingTransactionChain transactionChain;
     private ScheduledFuture<?> scheduleTask;
     private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
+    private final AtomicBoolean closed = new AtomicBoolean(false);
 
     public TopologyStatsProviderImpl(@Nonnull final DataBroker dataBroker, final int timeout) {
         this.dataBroker = requireNonNull(dataBroker);
@@ -69,42 +74,66 @@ public final class TopologyStatsProviderImpl implements TransactionChainListener
         this.scheduleTask = this.scheduler.scheduleAtFixedRate(task, 0, this.timeout, SECONDS);
     }
 
+    @SuppressWarnings("checkstyle:IllegalCatch")
     private synchronized void updatePcepStats() {
         final WriteTransaction tx = TopologyStatsProviderImpl.this.transactionChain.newWriteOnlyTransaction();
 
-        for (final Map.Entry<KeyedInstanceIdentifier<Node, NodeKey>, PcepSessionState> entry
-                : this.statsMap.entrySet()) {
-            final PcepTopologyNodeStatsAug nodeStatsAug = new PcepTopologyNodeStatsAugBuilder()
-                    .setPcepSessionState(new PcepSessionStateBuilder(entry.getValue()).build()).build();
-            final InstanceIdentifier<PcepTopologyNodeStatsAug> statId =
-                    entry.getKey().augmentation(PcepTopologyNodeStatsAug.class);
-            tx.put(LogicalDatastoreType.OPERATIONAL, statId, nodeStatsAug);
+        try {
+            for (final Map.Entry<KeyedInstanceIdentifier<Node, NodeKey>, PcepSessionState> entry
+                    : this.statsMap.entrySet()) {
+                final PcepTopologyNodeStatsAug nodeStatsAug = new PcepTopologyNodeStatsAugBuilder()
+                        .setPcepSessionState(new PcepSessionStateBuilder(entry.getValue()).build()).build();
+                final InstanceIdentifier<PcepTopologyNodeStatsAug> statId =
+                        entry.getKey().augmentation(PcepTopologyNodeStatsAug.class);
+                tx.put(LogicalDatastoreType.OPERATIONAL, statId, nodeStatsAug);
+            }
+            Futures.addCallback(tx.submit(), new FutureCallback<Void>() {
+                @Override
+                public void onSuccess(Void result) {
+                    LOG.debug("Successfully committed Topology stats update");
+                }
+
+                @Override
+                public void onFailure(Throwable ex) {
+                    LOG.error("Failed to commit Topology stats update", ex);
+                }
+            }, MoreExecutors.directExecutor());
+        } catch (final Exception e) {
+            LOG.warn("Failed to prepare Tx for BGP stats update", e);
+            tx.cancel();
         }
-        tx.submit();
     }
 
     @Override
     public synchronized void close() throws Exception {
-        LOG.info("Closing TopologyStatsProvider service.", this);
-        this.scheduleTask.cancel(true);
-        final WriteTransaction wTx = this.transactionChain.newWriteOnlyTransaction();
-        for (final KeyedInstanceIdentifier<Node, NodeKey> statId : this.statsMap.keySet()) {
-            wTx.delete(LogicalDatastoreType.OPERATIONAL, statId);
+        if (closed.compareAndSet(false, true)) {
+            LOG.info("Closing TopologyStatsProvider service.", this);
+            this.scheduleTask.cancel(true);
+            final WriteTransaction wTx = this.transactionChain.newWriteOnlyTransaction();
+            for (final KeyedInstanceIdentifier<Node, NodeKey> statId : this.statsMap.keySet()) {
+                wTx.delete(LogicalDatastoreType.OPERATIONAL, statId);
+            }
+            wTx.submit().get();
+            this.statsMap.clear();
+            this.transactionChain.close();
+            this.scheduler.shutdown();
         }
-        wTx.submit().get();
-        this.statsMap.clear();
-        this.transactionChain.close();
-        this.scheduler.shutdown();
     }
 
     @Override
-    public void onTransactionChainFailed(final TransactionChain<?, ?> chain, final AsyncTransaction<?, ?> transaction,
-            final Throwable cause) {
-        LOG.error("Transaction chain failed {}.", transaction != null ? transaction.getIdentifier() : null, cause);
+    public synchronized void onTransactionChainFailed(final TransactionChain<?, ?> chain,
+            final AsyncTransaction<?, ?> transaction, final Throwable cause) {
+        LOG.error("Transaction chain {} failed for tx {}",
+                chain, transaction != null ? transaction.getIdentifier() : null, cause);
+
+        if (!closed.get()) {
+            transactionChain.close();
+            transactionChain = dataBroker.createTransactionChain(this);
+        }
     }
 
     @Override
-    public void onTransactionChainSuccessful(final TransactionChain<?, ?> chain) {
+    public synchronized void onTransactionChainSuccessful(final TransactionChain<?, ?> chain) {
         LOG.debug("Transaction chain {} successful.", chain);
     }