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