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