#4871 support recurring transaction recovery and extend recovery statistics with failure count

Signed-off-by: pm-osc <[email protected]>

Author: pm-osc

docs/operations/recovery.md

@@ -45,12 +45,58 @@ identify partially failed transaction and repair any inconsistencies
 caused. It is suggested to run the transaction repair process on a
 separate machine connected to the cluster to isolate failures. Configure
 a separately controlled process to run the following where the start
-time specifies the time since epoch where the recovery process should
-start reading from the write-ahead log.
+time (Java Instant that specifies the time since epoch) where the recovery
+process should start reading from the write-ahead log.
 ```groovy
-recovery = JanusGraphFactory.startTransactionRecovery(graph, startTime, TimeUnit.MILLISECONDS);
+recovery = JanusGraphFactory.startTransactionRecovery(graph, startTime);
 ```
 
+Once the recovery process is started, the process never ends and stops
+only if:
+
+1. it is manually stopped by calling `recovery.shutdown()`
+2. the process encounters errors and fails due to exception
+3. the graph gets closed
+
+While the recovery process runs, `recovery.getStatistics()` call provides
+information about the progress of recovery process by returning three numbers:
+
+1. the first number shows how many secondary persistence transaction succeeded
+2. the second number shows how many secondary persistence transaction failed
+and attempted to be recovered
+3. the third number shows how many failed secondary persistence transaction
+could not be recovered
+
+Depending on the used `startTime` value and configured `log.tx.read-interval`
+configuration option, the recovery process might need to run for hours in
+order to process all relevant entries from the write-ahead log.
+`startTime` defines the point in time from which the write-ahead
+log should be read. The log is read in every `log.tx.read-interval`
+millisecond and an approximately 100 seconds long chunk is processed in
+one iteration.
+
+For example, if `startTime` is configured to look for log entries from
+the last twenty hours (72 000 seconds) and `log.tx.read-interval` is set to
+5 000 ms (5 seconds), it might take approximately at least one hour
+(72 000 / 100 * 5 = 3 600 seconds = 1 hour) while all log entries are processed.
+Note: in case there are many failed secondary persistence transactions, the recovery
+process might take much longer as fixing those transactions takes time.
+
+When a write-ahead log entry is found that should be repaired,
+the following INFO level log message appears in the JanusGraph's
+logging system where the transaction ID appears between the
+squared brackets:
+
+```
+Attempting to repair partially failed transaction [...]
+```
+
+Even if the recovery process is stopped by calling `recovery.shutdown()`,
+when it started again for the same graph `Provided read marker is not compatible
+with existing read marker for previously registered readers` error is shown.
+To avoid the error, the graph needs to be closed by `graph.close()` before the
+process is started again.
+
 Enabling the transaction write-ahead log causes an additional write
 operation for mutating transactions which increases the latency. Also
 note, that additional space is required to store the log. The
@@ -59,6 +105,27 @@ which means that log entries expire after that time to keep the storage
 overhead small. Refer to [Configuration Reference](../configs/configuration-reference.md) for a complete list of all
 log related configuration options to fine tune logging behavior.
 
+### Recurring Transaction Recovery
+
+In case of daily data ingestion, transaction recovery needs to run recurring to ensure that
+both primary and secondary persistence (e.g. indexing data by the mixed index backend)
+of the data succeeds each day.
+
+Since `JanusGraphFactory.startTransactionRecovery()` is not meant to be executed on
+recurring way, JanusGraph provides a dedicated way to run transaction recovery multiple
+times on the same graph:
+
+```groovy
+recovery = JanusGraphFactory.startRecurringTransactionRecovery(graph, startTime);
+```
+
+Similarly to the normal transaction recovery process, the recurring transaction recovery
+process has the same `graph` and `startTime` parameters and provides the same `getStatistics()`
+and `shutdown()` methods.
+
+Once the process is stopped, `startRecurringTransactionRecovery()` can be used
+to start the process again from the same or from another start time.
+
 ## JanusGraph Instance Failure
 
 JanusGraph is robust against individual instance failure in that other

janusgraph-backend-testutils/src/main/java/org/janusgraph/graphdb/JanusGraphIndexTest.java

@@ -131,6 +131,11 @@ public synchronized boolean unregisterReader(MessageReader reader) {
             return readers.remove(reader);
         }
 
+        @Override
+        public synchronized boolean unregisterReaderAndStopReadingProcess(MessageReader reader) {
+            return readers.remove(reader);
+        }
+
         @Override
         public String getName() {
             return name;

janusgraph-backend-testutils/src/main/java/org/janusgraph/graphdb/TestMockLog.java

@@ -303,6 +303,18 @@ public static TransactionRecovery startTransactionRecovery(JanusGraph graph, Ins
         return new StandardTransactionLogProcessor((StandardJanusGraph)graph, start);
     }
 
+    /**
+     * Returns a {@link TransactionRecovery} process for recovering partially failed transactions. The recovery process
+     * will start processing the write-ahead transaction log at the specified transaction time.
+     * Once stopped, this method can be used to start the recovery process again from a specified transaction time.
+     * @param graph
+     * @param start
+     * @return
+     */
+    public static TransactionRecovery startRecurringTransactionRecovery(JanusGraph graph, Instant start) {
+        return new StandardTransactionLogProcessor((StandardJanusGraph) graph, start, true);
+    }
+
     //###################################
     //          HELPER METHODS
     //###################################

janusgraph-core/src/main/java/org/janusgraph/core/JanusGraphFactory.java

@@ -82,6 +82,17 @@ public interface Log {
      */
     boolean unregisterReader(MessageReader reader);
 
+    /**
+     * Removes the given reader from the list of registered readers, terminates the reader thread pool and returns
+     * whether this reader was registered in the first place.
+     * Note, that removing the reader by this method also stops the reading process.
+     *
+     * @param reader
+     * @return true if this MessageReader was registered before and was successfully
+     *         unregistered, else false
+     */
+    boolean unregisterReaderAndStopReadingProcess(MessageReader reader);
+
     /**
      * Returns the name of this log
      * @return

janusgraph-core/src/main/java/org/janusgraph/diskstorage/log/Log.java

@@ -26,12 +26,17 @@ public class ReadMarker {
 
     private final String identifier;
     private Instant startTime;
+    private boolean isRecurring;
 
+    private ReadMarker(String identifier, Instant startTime) {
+        this(identifier, startTime, false);
+    }
 
 
-    private ReadMarker(String identifier, Instant startTime) {
+    private ReadMarker(String identifier, Instant startTime, boolean isRecurring) {
         this.identifier = identifier;
         this.startTime = startTime;
+        this.isRecurring = isRecurring;
     }
 
     /**
@@ -55,6 +60,10 @@ public boolean hasStartTime() {
         return startTime!=null;
     }
 
+    public boolean isRecurring() {
+        return isRecurring;
+    }
+
     /**
      * Returns the start time of this marker if such has been defined or the current time if not
      * @return
@@ -72,6 +81,12 @@ public synchronized Instant getStartTime(TimestampProvider times) {
      * @return
      */
     public boolean isCompatible(ReadMarker newMarker) {
+        if (isRecurring() && newMarker.isRecurring()) {
+            // recurring read marker is not expected to have identifier
+            // See this comment for more information:
+            // https://github.com/JanusGraph/janusgraph/pull/4872#discussion_r2490578432
+            return !hasIdentifier() && !newMarker.hasIdentifier();
+        }
         if (newMarker.hasIdentifier()) {
             return hasIdentifier() && identifier.equals(newMarker.identifier);
         }
@@ -96,6 +111,17 @@ public static ReadMarker fromTime(Instant timestamp) {
         return new ReadMarker(null, timestamp);
     }
 
+    /**
+     * Starts reading the log from the given timestamp onward. The specified timestamp is included.
+     * The read marker is set to allow recurring mode which means that reading the log is supported
+     * several times from the same or from different timestamp.
+     * @param timestamp
+     * @return
+     */
+    public static ReadMarker fromTimeRecurring(Instant timestamp) {
+        return new ReadMarker(null, timestamp, true);
+    }
+
     /**
      * Starts reading the log from the last recorded point in the log for the given id.
      * If the log has a record of such an id, it will use it as the starting point.

janusgraph-core/src/main/java/org/janusgraph/diskstorage/log/ReadMarker.java

@@ -669,7 +669,12 @@ public synchronized void registerReaders(ReadMarker readMarker, Iterable<Message
         Preconditions.checkArgument(readMarker!=null,"Read marker cannot be null");
         Preconditions.checkArgument(this.readMarker==null || this.readMarker.isCompatible(readMarker),
                 "Provided read marker is not compatible with existing read marker for previously registered readers");
-        if (this.readMarker==null) this.readMarker=readMarker;
+        if (this.readMarker==null || this.readMarker.isRecurring())
+            // for recurring read marker the marker needs to be updated
+            // so also the start time of the marker will be updated and the log
+            // is read from the right start time
+            this.readMarker = readMarker;
+
         boolean firstRegistration = this.readers.isEmpty();
         for (MessageReader reader : readers) {
             Preconditions.checkNotNull(reader);
@@ -707,6 +712,30 @@ public synchronized boolean unregisterReader(MessageReader reader) {
         return this.readers.remove(reader);
     }
 
+    @Override
+    public synchronized boolean unregisterReaderAndStopReadingProcess(MessageReader reader) {
+        ResourceUnavailableException.verifyOpen(isOpen, "Log", name);
+
+        if (readExecutor!=null) {
+            readExecutor.shutdown();
+            try {
+                readExecutor.awaitTermination(1,TimeUnit.SECONDS);
+            } catch (InterruptedException e) {
+                log.error("Could not terminate reader thread pool for KCVSLog {} due to interruption", name, e);
+            }
+            if (!readExecutor.isTerminated()) {
+                readExecutor.shutdownNow();
+                log.error("Reader thread pool for KCVSLog {} did not shut down in time - could not clean up or set read markers", name);
+            } else {
+                for (MessagePuller puller : msgPullers) {
+                    puller.close();
+                }
+            }
+        }
+
+        return this.readers.remove(reader);
+    }
+
     private class MessageReaderStateUpdater implements Runnable {
         @Override
         public void run() {

janusgraph-core/src/main/java/org/janusgraph/diskstorage/log/kcvs/KCVSLog.java

@@ -93,13 +93,29 @@ public class StandardTransactionLogProcessor implements TransactionRecovery {
     private final BackgroundCleaner cleaner;
     private final boolean verboseLogging;
 
+    // flag to support starting the log processor multiple times
+    // for a given graph
+    private final boolean recurringMode;
+
     private final AtomicLong successTxCounter = new AtomicLong(0);
     private final AtomicLong failureTxCounter = new AtomicLong(0);
 
+    // counter for storing how many transactions from failureTxCounter
+    // could not be repaired due to any exception
+    private final AtomicLong failureTxRepairExceptionCounter = new AtomicLong(0);
+
+    private final Log txLog;
+    private final TxLogMessageReader reader;
+
     private final Cache<StandardTransactionId,TxEntry> txCache;
 
     public StandardTransactionLogProcessor(StandardJanusGraph graph,
                                            Instant startTime) {
+        this(graph, startTime, false);
+    }
+
+    public StandardTransactionLogProcessor(StandardJanusGraph graph,
+                                           Instant startTime, boolean recurringMode) {
         Preconditions.checkArgument(graph != null && graph.isOpen());
         Preconditions.checkArgument(startTime!=null);
         Preconditions.checkArgument(graph.getConfiguration().hasLogTransactions(),
@@ -109,9 +125,10 @@ public StandardTransactionLogProcessor(StandardJanusGraph graph,
         Preconditions.checkArgument(maxTxLength != null && !maxTxLength.isZero(),
                 "Max transaction time cannot be 0");
         this.graph = graph;
+        this.recurringMode = recurringMode;
         this.serializer = graph.getDataSerializer();
         this.times = graph.getConfiguration().getTimestampProvider();
-        final Log txLog = graph.getBackend().getSystemTxLog();
+        this.txLog = graph.getBackend().getSystemTxLog();
         this.persistenceTime = graph.getConfiguration().getMaxWriteTime();
         this.verboseLogging = graph.getConfiguration().getConfiguration()
                 .get(GraphDatabaseConfiguration.VERBOSE_TX_RECOVERY);
@@ -123,26 +140,54 @@ public StandardTransactionLogProcessor(StandardJanusGraph graph,
                         "Unexpected removal cause [%s] for transaction [%s]", cause, key);
                     if (entry.status == LogTxStatus.SECONDARY_FAILURE || entry.status == LogTxStatus.PRIMARY_SUCCESS) {
                         failureTxCounter.incrementAndGet();
-                        fixSecondaryFailure(key, entry);
+                        try {
+                            fixSecondaryFailure(key, entry);
+                        } catch (Exception e) {
+                            failureTxRepairExceptionCounter.incrementAndGet();
+                            // pass exception up - here the exception is caught only for
+                            // incrementing the failureTxRepairExceptionCounter counter
+                            throw e;
+                        }
                     } else {
                         successTxCounter.incrementAndGet();
                     }
                 })
                 .build();
 
-        ReadMarker start = ReadMarker.fromTime(startTime);
-        txLog.registerReader(start,new TxLogMessageReader());
+        ReadMarker start;
+
+        if (this.recurringMode) {
+            // create the read marker with recurring mode so it can be registered multiple times
+            start = ReadMarker.fromTimeRecurring(startTime);
+        } else {
+            start = ReadMarker.fromTime(startTime);
+        }
+
+        this.reader = new TxLogMessageReader();
+        txLog.registerReader(start,this.reader);
 
         cleaner = new BackgroundCleaner();
         cleaner.start();
     }
 
     public long[] getStatistics() {
-        return new long[]{successTxCounter.get(),failureTxCounter.get()};
+        return new long[]{successTxCounter.get(),failureTxCounter.get(),failureTxRepairExceptionCounter.get()};
     }
 
     public synchronized void shutdown() throws JanusGraphException {
         cleaner.close(CLEAN_SLEEP_TIME);
+
+        // in recurring mode, the reader needs to be unregistered and the reading process
+        // needs to be stopped so resources are not occupied while the log processor is not
+        // actively running
+        if (this.recurringMode) {
+            try {
+                Thread.sleep(CLEAN_SLEEP_TIME.toMillis());
+                txLog.unregisterReaderAndStopReadingProcess(this.reader);
+            } catch (Exception e) {
+                logger.error("Interrupted while waiting for background cleaner to stop. Reader is not unregistered.", e);
+            }
+        }
     }
 
     private void logRecoveryMsg(String message, Object... args) {