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

   1 /*
   2  * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   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
   7  */
   8
   9 package org.opendaylight.controller.cluster.raft;
  10
  11 import akka.japi.Procedure;
  12 import akka.persistence.SnapshotSelectionCriteria;
  13 import com.google.common.annotations.VisibleForTesting;
  14 import com.google.protobuf.ByteString;
  15 import java.util.List;
  16 import org.opendaylight.controller.cluster.DataPersistenceProvider;
  17 import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshot;
  18 import org.opendaylight.controller.cluster.raft.base.messages.SendInstallSnapshot;
  19 import org.opendaylight.controller.cluster.raft.behaviors.RaftActorBehavior;
  20 import org.slf4j.Logger;
  21
  22 public class SnapshotManager implements SnapshotState {
  23
  24     private final SnapshotState IDLE = new Idle();
  25     private final SnapshotState PERSISTING = new Persisting();
  26     private final SnapshotState CREATING = new Creating();
  27
  28     private final Logger LOG;
  29     private final RaftActorContext context;
  30     private final LastAppliedTermInformationReader lastAppliedTermInformationReader =
  31             new LastAppliedTermInformationReader();
  32     private final ReplicatedToAllTermInformationReader replicatedToAllTermInformationReader =
  33             new ReplicatedToAllTermInformationReader();
  34
  35
  36     private SnapshotState currentState = IDLE;
  37     private CaptureSnapshot captureSnapshot;
  38     private long lastSequenceNumber = -1;
  39
  40     private Procedure<Void> createSnapshotProcedure;
  41
  42     public SnapshotManager(RaftActorContext context, Logger logger) {
  43         this.context = context;
  44         this.LOG = logger;
  45     }
  46
  47     @Override
  48     public boolean isCapturing() {
  49         return currentState.isCapturing();
  50     }
  51
  52     @Override
  53     public boolean captureToInstall(ReplicatedLogEntry lastLogEntry, long replicatedToAllIndex, String targetFollower) {
  54         return currentState.captureToInstall(lastLogEntry, replicatedToAllIndex, targetFollower);
  55     }
  56
  57     @Override
  58     public boolean capture(ReplicatedLogEntry lastLogEntry, long replicatedToAllIndex) {
  59         return currentState.capture(lastLogEntry, replicatedToAllIndex);
  60     }
  61
  62     @Override
  63     public void persist(DataPersistenceProvider persistenceProvider, byte[] snapshotBytes,
  64                         RaftActorBehavior currentBehavior, long totalMemory) {
  65         currentState.persist(persistenceProvider, snapshotBytes, currentBehavior, totalMemory);
  66     }
  67
  68     @Override
  69     public void commit(DataPersistenceProvider persistenceProvider, long sequenceNumber) {
  70         currentState.commit(persistenceProvider, sequenceNumber);
  71     }
  72
  73     @Override
  74     public void rollback() {
  75         currentState.rollback();
  76     }
  77
  78     @Override
  79     public long trimLog(long desiredTrimIndex, RaftActorBehavior currentBehavior) {
  80         return currentState.trimLog(desiredTrimIndex, currentBehavior);
  81     }
  82
  83     public void setCreateSnapshotCallable(Procedure<Void> createSnapshotProcedure) {
  84         this.createSnapshotProcedure = createSnapshotProcedure;
  85     }
  86
  87     @VisibleForTesting
  88     public CaptureSnapshot getCaptureSnapshot() {
  89         return captureSnapshot;
  90     }
  91
  92     private boolean hasFollowers(){
  93         return context.getPeerAddresses().keySet().size() > 0;
  94     }
  95
  96     private String persistenceId(){
  97         return context.getId();
  98     }
  99
 100     private class AbstractSnapshotState implements SnapshotState {
 101
 102         @Override
 103         public boolean isCapturing() {
 104             return false;
 105         }
 106
 107         @Override
 108         public boolean capture(ReplicatedLogEntry lastLogEntry, long replicatedToAllIndex) {
 109             LOG.debug("capture should not be called in state {}", this);
 110             return false;
 111         }
 112
 113         @Override
 114         public boolean captureToInstall(ReplicatedLogEntry lastLogEntry, long replicatedToAllIndex, String targetFollower) {
 115             LOG.debug("captureToInstall should not be called in state {}", this);
 116             return false;
 117         }
 118
 119         @Override
 120         public void persist(DataPersistenceProvider persistenceProvider, byte[] snapshotBytes,
 121                             RaftActorBehavior currentBehavior, long totalMemory) {
 122             LOG.debug("persist should not be called in state {}", this);
 123         }
 124
 125         @Override
 126         public void commit(DataPersistenceProvider persistenceProvider, long sequenceNumber) {
 127             LOG.debug("commit should not be called in state {}", this);
 128         }
 129
 130         @Override
 131         public void rollback() {
 132             LOG.debug("rollback should not be called in state {}", this);
 133         }
 134
 135         @Override
 136         public long trimLog(long desiredTrimIndex, RaftActorBehavior currentBehavior) {
 137             LOG.debug("trimLog should not be called in state {}", this);
 138             return -1;
 139         }
 140
 141         protected long doTrimLog(long desiredTrimIndex, RaftActorBehavior currentBehavior){
 142             //  we would want to keep the lastApplied as its used while capturing snapshots
 143             long lastApplied = context.getLastApplied();
 144             long tempMin = Math.min(desiredTrimIndex, (lastApplied > -1 ? lastApplied - 1 : -1));
 145
 146             if(LOG.isTraceEnabled()) {
 147                 LOG.trace("{}: performSnapshotWithoutCapture: desiredTrimIndex: {}, lastApplied: {}, tempMin: {}",
 148                         persistenceId(), desiredTrimIndex, lastApplied, tempMin);
 149             }
 150
 151             if (tempMin > -1 && context.getReplicatedLog().isPresent(tempMin)) {
 152                 LOG.debug("{}: fakeSnapshot purging log to {} for term {}", persistenceId(), tempMin,
 153                         context.getTermInformation().getCurrentTerm());
 154
 155                 //use the term of the temp-min, since we check for isPresent, entry will not be null
 156                 ReplicatedLogEntry entry = context.getReplicatedLog().get(tempMin);
 157                 context.getReplicatedLog().snapshotPreCommit(tempMin, entry.getTerm());
 158                 context.getReplicatedLog().snapshotCommit();
 159                 return tempMin;
 160             } else if(tempMin > currentBehavior.getReplicatedToAllIndex()) {
 161                 // It's possible a follower was lagging and an install snapshot advanced its match index past
 162                 // the current replicatedToAllIndex. Since the follower is now caught up we should advance the
 163                 // replicatedToAllIndex (to tempMin). The fact that tempMin wasn't found in the log is likely
 164                 // due to a previous snapshot triggered by the memory threshold exceeded, in that case we
 165                 // trim the log to the last applied index even if previous entries weren't replicated to all followers.
 166                 currentBehavior.setReplicatedToAllIndex(tempMin);
 167             }
 168             return -1;
 169         }
 170     }
 171
 172     private class Idle extends AbstractSnapshotState {
 173
 174         private boolean capture(ReplicatedLogEntry lastLogEntry, long replicatedToAllIndex, String targetFollower) {
 175             TermInformationReader lastAppliedTermInfoReader =
 176                     lastAppliedTermInformationReader.init(context.getReplicatedLog(), context.getLastApplied(),
 177                             lastLogEntry, hasFollowers());
 178
 179             long lastAppliedIndex = lastAppliedTermInfoReader.getIndex();
 180             long lastAppliedTerm = lastAppliedTermInfoReader.getTerm();
 181
 182             TermInformationReader replicatedToAllTermInfoReader =
 183                     replicatedToAllTermInformationReader.init(context.getReplicatedLog(), replicatedToAllIndex);
 184
 185             long newReplicatedToAllIndex = replicatedToAllTermInfoReader.getIndex();
 186             long newReplicatedToAllTerm = replicatedToAllTermInfoReader.getTerm();
 187
 188             // send a CaptureSnapshot to self to make the expensive operation async.
 189
 190             List<ReplicatedLogEntry> unAppliedEntries = context.getReplicatedLog().getFrom(lastAppliedIndex + 1);
 191
 192             captureSnapshot = new CaptureSnapshot(lastLogEntry.getIndex(),
 193                     lastLogEntry.getTerm(), lastAppliedIndex, lastAppliedTerm,
 194                     newReplicatedToAllIndex, newReplicatedToAllTerm, unAppliedEntries, targetFollower != null);
 195
 196             if(captureSnapshot.isInstallSnapshotInitiated()) {
 197                 LOG.info("{}: Initiating snapshot capture {} to install on {}",
 198                         persistenceId(), captureSnapshot, targetFollower);
 199             } else {
 200                 LOG.info("{}: Initiating snapshot capture {}", persistenceId(), captureSnapshot);
 201             }
 202
 203             lastSequenceNumber = context.getPersistenceProvider().getLastSequenceNumber();
 204
 205             LOG.debug("lastSequenceNumber prior to capture: {}", lastSequenceNumber);
 206
 207             try {
 208                 createSnapshotProcedure.apply(null);
 209             } catch (Exception e) {
 210                 LOG.error("Error creating snapshot", e);
 211                 return false;
 212             }
 213
 214             SnapshotManager.this.currentState = CREATING;
 215             return true;
 216         }
 217
 218         @Override
 219         public boolean capture(ReplicatedLogEntry lastLogEntry, long replicatedToAllIndex) {
 220             return capture(lastLogEntry, replicatedToAllIndex, null);
 221         }
 222
 223         @Override
 224         public boolean captureToInstall(ReplicatedLogEntry lastLogEntry, long replicatedToAllIndex, String targetFollower) {
 225             return capture(lastLogEntry, replicatedToAllIndex, targetFollower);
 226         }
 227
 228         @Override
 229         public String toString() {
 230             return "Idle";
 231         }
 232
 233         @Override
 234         public long trimLog(long desiredTrimIndex, RaftActorBehavior currentBehavior) {
 235             return doTrimLog(desiredTrimIndex, currentBehavior);
 236         }
 237     }
 238
 239     private class Creating extends AbstractSnapshotState {
 240
 241         @Override
 242         public boolean isCapturing() {
 243             return true;
 244         }
 245
 246         @Override
 247         public void persist(DataPersistenceProvider persistenceProvider, byte[] snapshotBytes,
 248                             RaftActorBehavior currentBehavior, long totalMemory) {
 249             // create a snapshot object from the state provided and save it
 250             // when snapshot is saved async, SaveSnapshotSuccess is raised.
 251
 252             Snapshot sn = Snapshot.create(snapshotBytes,
 253                     captureSnapshot.getUnAppliedEntries(),
 254                     captureSnapshot.getLastIndex(), captureSnapshot.getLastTerm(),
 255                     captureSnapshot.getLastAppliedIndex(), captureSnapshot.getLastAppliedTerm());
 256
 257             persistenceProvider.saveSnapshot(sn);
 258
 259             LOG.info("{}: Persisting of snapshot done:{}", persistenceId(), sn.getLogMessage());
 260
 261             long dataThreshold = totalMemory *
 262                     context.getConfigParams().getSnapshotDataThresholdPercentage() / 100;
 263             if (context.getReplicatedLog().dataSize() > dataThreshold) {
 264
 265                 if(LOG.isDebugEnabled()) {
 266                     LOG.debug("{}: dataSize {} exceeds dataThreshold {} - doing snapshotPreCommit with index {}",
 267                             persistenceId(), context.getReplicatedLog().dataSize(), dataThreshold,
 268                             captureSnapshot.getLastAppliedIndex());
 269                 }
 270
 271                 // if memory is less, clear the log based on lastApplied.
 272                 // this could/should only happen if one of the followers is down
 273                 // as normally we keep removing from the log when its replicated to all.
 274                 context.getReplicatedLog().snapshotPreCommit(captureSnapshot.getLastAppliedIndex(),
 275                         captureSnapshot.getLastAppliedTerm());
 276
 277                 // Don't reset replicatedToAllIndex to -1 as this may prevent us from trimming the log after an
 278                 // install snapshot to a follower.
 279                 if(captureSnapshot.getReplicatedToAllIndex() >= 0) {
 280                     currentBehavior.setReplicatedToAllIndex(captureSnapshot.getReplicatedToAllIndex());
 281                 }
 282
 283             } else if(captureSnapshot.getReplicatedToAllIndex() != -1){
 284                 // clear the log based on replicatedToAllIndex
 285                 context.getReplicatedLog().snapshotPreCommit(captureSnapshot.getReplicatedToAllIndex(),
 286                         captureSnapshot.getReplicatedToAllTerm());
 287
 288                 currentBehavior.setReplicatedToAllIndex(captureSnapshot.getReplicatedToAllIndex());
 289             } else {
 290                 // The replicatedToAllIndex was not found in the log
 291                 // This means that replicatedToAllIndex never moved beyond -1 or that it is already in the snapshot.
 292                 // In this scenario we may need to save the snapshot to the akka persistence
 293                 // snapshot for recovery but we do not need to do the replicated log trimming.
 294                 context.getReplicatedLog().snapshotPreCommit(context.getReplicatedLog().getSnapshotIndex(),
 295                         context.getReplicatedLog().getSnapshotTerm());
 296             }
 297
 298             LOG.info("{}: Removed in-memory snapshotted entries, adjusted snaphsotIndex:{} " +
 299                             "and term:{}", persistenceId(), captureSnapshot.getLastAppliedIndex(),
 300                     captureSnapshot.getLastAppliedTerm());
 301
 302             if (context.getId().equals(currentBehavior.getLeaderId())
 303                     && captureSnapshot.isInstallSnapshotInitiated()) {
 304                 // this would be call straight to the leader and won't initiate in serialization
 305                 currentBehavior.handleMessage(context.getActor(), new SendInstallSnapshot(
 306                         ByteString.copyFrom(snapshotBytes)));
 307             }
 308
 309             captureSnapshot = null;
 310             SnapshotManager.this.currentState = PERSISTING;
 311         }
 312
 313         @Override
 314         public String toString() {
 315             return "Creating";
 316         }
 317
 318     }
 319
 320     private class Persisting extends AbstractSnapshotState {
 321
 322         @Override
 323         public void commit(DataPersistenceProvider persistenceProvider, long sequenceNumber) {
 324             context.getReplicatedLog().snapshotCommit();
 325             persistenceProvider.deleteSnapshots(new SnapshotSelectionCriteria(
 326                     sequenceNumber - context.getConfigParams().getSnapshotBatchCount(), 43200000));
 327
 328             persistenceProvider.deleteMessages(lastSequenceNumber);
 329
 330             lastSequenceNumber = -1;
 331             SnapshotManager.this.currentState = IDLE;
 332         }
 333
 334         @Override
 335         public void rollback() {
 336             context.getReplicatedLog().snapshotRollback();
 337
 338             LOG.info("{}: Replicated Log rolled back. Snapshot will be attempted in the next cycle." +
 339                             "snapshotIndex:{}, snapshotTerm:{}, log-size:{}", persistenceId(),
 340                     context.getReplicatedLog().getSnapshotIndex(),
 341                     context.getReplicatedLog().getSnapshotTerm(),
 342                     context.getReplicatedLog().size());
 343
 344             SnapshotManager.this.currentState = IDLE;
 345         }
 346
 347         @Override
 348         public String toString() {
 349             return "Persisting";
 350         }
 351
 352     }
 353
 354     private static interface TermInformationReader {
 355         long getIndex();
 356         long getTerm();
 357     }
 358
 359     static class LastAppliedTermInformationReader implements TermInformationReader{
 360         private long index;
 361         private long term;
 362
 363         public LastAppliedTermInformationReader init(ReplicatedLog log, long originalIndex,
 364                                          ReplicatedLogEntry lastLogEntry, boolean hasFollowers){
 365             ReplicatedLogEntry entry = log.get(originalIndex);
 366             this.index = -1L;
 367             this.term = -1L;
 368             if (!hasFollowers) {
 369                 if(lastLogEntry != null) {
 370                     index = lastLogEntry.getIndex();
 371                     term = lastLogEntry.getTerm();
 372                 }
 373             } else if (entry != null) {
 374                 index = entry.getIndex();
 375                 term = entry.getTerm();
 376             } else if(log.getSnapshotIndex() > -1){
 377                 index = log.getSnapshotIndex();
 378                 term = log.getSnapshotTerm();
 379             }
 380             return this;
 381         }
 382
 383         @Override
 384         public long getIndex(){
 385             return this.index;
 386         }
 387
 388         @Override
 389         public long getTerm(){
 390             return this.term;
 391         }
 392     }
 393
 394     private static class ReplicatedToAllTermInformationReader implements TermInformationReader{
 395         private long index;
 396         private long term;
 397
 398         ReplicatedToAllTermInformationReader init(ReplicatedLog log, long originalIndex){
 399             ReplicatedLogEntry entry = log.get(originalIndex);
 400             this.index = -1L;
 401             this.term = -1L;
 402
 403             if (entry != null) {
 404                 index = entry.getIndex();
 405                 term = entry.getTerm();
 406             }
 407
 408             return this;
 409         }
 410
 411         @Override
 412         public long getIndex(){
 413             return this.index;
 414         }
 415
 416         @Override
 417         public long getTerm(){
 418             return this.term;
 419         }
 420     }
 421 }