Add direct in-memory journal threshold

[controller.git] / opendaylight / md-sal / sal-akka-raft / src / main / java / org / opendaylight / controller / cluster / raft / SnapshotManager.java

diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/SnapshotManager.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/SnapshotManager.java
index 7d9a1bbf5bdbbfba4fd8df57bf17a17f8e8b125f..791a791027ab2a2c2e7dc8786d167514c754dae6 100644 (file)
--- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/SnapshotManager.java
+++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/SnapshotManager.java
@@ -91,6 +91,11 @@ public class SnapshotManager implements SnapshotState {
         return currentState.capture(lastLogEntry, replicatedToAllIndex);
     }
 
+    @Override
+    public boolean captureWithForcedTrim(final ReplicatedLogEntry lastLogEntry, final long replicatedToAllIndex) {
+        return currentState.captureWithForcedTrim(lastLogEntry, replicatedToAllIndex);
+    }
+
     @Override
     public void apply(final ApplySnapshot snapshot) {
         currentState.apply(snapshot);
@@ -154,7 +159,8 @@ public class SnapshotManager implements SnapshotState {
      * @param replicatedToAllIndex the index of the last entry replicated to all followers.
      * @return a new CaptureSnapshot instance.
      */
-    public CaptureSnapshot newCaptureSnapshot(final ReplicatedLogEntry lastLogEntry, final long replicatedToAllIndex) {
+    public CaptureSnapshot newCaptureSnapshot(final ReplicatedLogEntry lastLogEntry, final long replicatedToAllIndex,
+                                              final boolean mandatoryTrim) {
         TermInformationReader lastAppliedTermInfoReader =
                 lastAppliedTermInformationReader.init(context.getReplicatedLog(), context.getLastApplied(),
                         lastLogEntry, hasFollowers());
@@ -186,7 +192,7 @@ public class SnapshotManager implements SnapshotState {
         }
 
         return new CaptureSnapshot(lastLogEntryIndex, lastLogEntryTerm, lastAppliedIndex, lastAppliedTerm,
-                newReplicatedToAllIndex, newReplicatedToAllTerm, unAppliedEntries);
+                newReplicatedToAllIndex, newReplicatedToAllTerm, unAppliedEntries, mandatoryTrim);
     }
 
     private class AbstractSnapshotState implements SnapshotState {
@@ -209,6 +215,12 @@ public class SnapshotManager implements SnapshotState {
             return false;
         }
 
+        @Override
+        public boolean captureWithForcedTrim(final ReplicatedLogEntry lastLogEntry, final long replicatedToAllIndex) {
+            log.debug("captureWithForcedTrim should not be called in state {}", this);
+            return false;
+        }
+
         @Override
         public void apply(final ApplySnapshot snapshot) {
             log.debug("apply should not be called in state {}", this);
@@ -279,8 +291,8 @@ public class SnapshotManager implements SnapshotState {
 
         @SuppressWarnings("checkstyle:IllegalCatch")
         private boolean capture(final ReplicatedLogEntry lastLogEntry, final long replicatedToAllIndex,
-                final String targetFollower) {
-            captureSnapshot = newCaptureSnapshot(lastLogEntry, replicatedToAllIndex);
+                final String targetFollower, final boolean mandatoryTrim) {
+            captureSnapshot = newCaptureSnapshot(lastLogEntry, replicatedToAllIndex, mandatoryTrim);
 
             OutputStream installSnapshotStream = null;
             if (targetFollower != null) {
@@ -310,13 +322,18 @@ public class SnapshotManager implements SnapshotState {
 
         @Override
         public boolean capture(final ReplicatedLogEntry lastLogEntry, final long replicatedToAllIndex) {
-            return capture(lastLogEntry, replicatedToAllIndex, null);
+            return capture(lastLogEntry, replicatedToAllIndex, null, false);
         }
 
         @Override
         public boolean captureToInstall(final ReplicatedLogEntry lastLogEntry, final long replicatedToAllIndex,
                 final String targetFollower) {
-            return capture(lastLogEntry, replicatedToAllIndex, targetFollower);
+            return capture(lastLogEntry, replicatedToAllIndex, targetFollower, false);
+        }
+
+        @Override
+        public boolean captureWithForcedTrim(final ReplicatedLogEntry lastLogEntry, final long replicatedToAllIndex) {
+            return capture(lastLogEntry, replicatedToAllIndex, null, true);
         }
 
         @Override
@@ -362,24 +379,29 @@ public class SnapshotManager implements SnapshotState {
 
             log.info("{}: Persisting of snapshot done: {}", persistenceId(), snapshot);
 
-            long dataThreshold = totalMemory * context.getConfigParams().getSnapshotDataThresholdPercentage() / 100;
-            boolean dataSizeThresholdExceeded = context.getReplicatedLog().dataSize() > dataThreshold;
+            final ConfigParams config = context.getConfigParams();
+            final long absoluteThreshold = config.getSnapshotDataThreshold();
+            final long dataThreshold = absoluteThreshold != 0 ? absoluteThreshold * ConfigParams.MEGABYTE
+                    : totalMemory * config.getSnapshotDataThresholdPercentage() / 100;
 
-            boolean logSizeExceededSnapshotBatchCount =
-                    context.getReplicatedLog().size() >= context.getConfigParams().getSnapshotBatchCount();
+            final boolean dataSizeThresholdExceeded = context.getReplicatedLog().dataSize() > dataThreshold;
+            final boolean logSizeExceededSnapshotBatchCount =
+                    context.getReplicatedLog().size() >= config.getSnapshotBatchCount();
 
             final RaftActorBehavior currentBehavior = context.getCurrentBehavior();
-            if (dataSizeThresholdExceeded || logSizeExceededSnapshotBatchCount) {
+            if (dataSizeThresholdExceeded || logSizeExceededSnapshotBatchCount || captureSnapshot.isMandatoryTrim()) {
                 if (log.isDebugEnabled()) {
                     if (dataSizeThresholdExceeded) {
                         log.debug("{}: log data size {} exceeds the memory threshold {} - doing snapshotPreCommit "
                                 + "with index {}", context.getId(), context.getReplicatedLog().dataSize(),
                                 dataThreshold, captureSnapshot.getLastAppliedIndex());
-                    } else {
+                    } else if (logSizeExceededSnapshotBatchCount) {
                         log.debug("{}: log size {} exceeds the snapshot batch count {} - doing snapshotPreCommit with "
                                 + "index {}", context.getId(), context.getReplicatedLog().size(),
-                                context.getConfigParams().getSnapshotBatchCount(),
-                                captureSnapshot.getLastAppliedIndex());
+                                config.getSnapshotBatchCount(), captureSnapshot.getLastAppliedIndex());
+                    } else {
+                        log.debug("{}: user triggered or root overwrite snapshot encountered, trimming log up to "
+                                + "last applied index {}", context.getId(), captureSnapshot.getLastAppliedIndex());
                     }
                 }