X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-akka-raft%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fraft%2FSnapshotManager.java;h=8037fb8d73ce88e6309f6a0f56b6d4fb787e8761;hp=02269a289bcbef6723d3fb78e34591a1e34328f5;hb=ff29db5dc6012f77bbe53f57ddce929b0de093b3;hpb=609076e66a0298d0d3c912ade66e813a464c4c8a

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 02269a289b..8037fb8d73 100644
--- 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
@@ -55,7 +55,7 @@ public class SnapshotManager implements SnapshotState {
     private CaptureSnapshot captureSnapshot;
     private long lastSequenceNumber = -1;
 
-    private Consumer<Optional<OutputStream>> createSnapshotProcedure;
+    private Consumer<Optional<OutputStream>> createSnapshotProcedure = null;
 
     private ApplySnapshot applySnapshot;
     private RaftActorSnapshotCohort snapshotCohort = NoopRaftActorSnapshotCohort.INSTANCE;
@@ -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());
@@ -170,18 +176,23 @@ public class SnapshotManager implements SnapshotState {
 
         List<ReplicatedLogEntry> unAppliedEntries = context.getReplicatedLog().getFrom(lastAppliedIndex + 1);
 
-        long lastLogEntryIndex = lastAppliedIndex;
-        long lastLogEntryTerm = lastAppliedTerm;
-        if (lastLogEntry != null) {
+        final long lastLogEntryIndex;
+        final long lastLogEntryTerm;
+        if (lastLogEntry == null) {
+            // When we don't have journal present, for example two captureSnapshots executed right after another with no
+            // new journal we still want to preserve the index and term in the snapshot.
+            lastAppliedIndex = lastLogEntryIndex = context.getReplicatedLog().getSnapshotIndex();
+            lastAppliedTerm = lastLogEntryTerm = context.getReplicatedLog().getSnapshotTerm();
+
+            log.debug("{}: Capturing Snapshot : lastLogEntry is null. Using snapshot values lastAppliedIndex {} and "
+                    + "lastAppliedTerm {} instead.", persistenceId(), lastAppliedIndex, lastAppliedTerm);
+        } else {
             lastLogEntryIndex = lastLogEntry.getIndex();
             lastLogEntryTerm = lastLogEntry.getTerm();
-        } else {
-            log.debug("{}: Capturing Snapshot : lastLogEntry is null. Using lastAppliedIndex {} and "
-                    + "lastAppliedTerm {} instead.", persistenceId(), lastAppliedIndex, lastAppliedTerm);
         }
 
         return new CaptureSnapshot(lastLogEntryIndex, lastLogEntryTerm, lastAppliedIndex, lastAppliedTerm,
-                newReplicatedToAllIndex, newReplicatedToAllTerm, unAppliedEntries);
+                newReplicatedToAllIndex, newReplicatedToAllTerm, unAppliedEntries, mandatoryTrim);
     }
 
     private class AbstractSnapshotState implements SnapshotState {
@@ -204,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);
@@ -248,7 +265,7 @@ public class SnapshotManager implements SnapshotState {
                 //use the term of the temp-min, since we check for isPresent, entry will not be null
                 ReplicatedLogEntry entry = context.getReplicatedLog().get(tempMin);
                 context.getReplicatedLog().snapshotPreCommit(tempMin, entry.getTerm());
-                context.getReplicatedLog().snapshotCommit();
+                context.getReplicatedLog().snapshotCommit(false);
                 return tempMin;
             }
 
@@ -274,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) {
@@ -305,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
@@ -357,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());
                     }
                 }