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