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