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             SnapshotManager.this.currentState = CREATING;
 205
 206             try {
 207                 createSnapshotProcedure.apply(null);
 208             } catch (Exception e) {
 209                 SnapshotManager.this.currentState = IDLE;
 210                 LOG.error("Error creating snapshot", e);
 211                 return false;
 212             }
 213
 214             return true;
 215         }
 216
 217         @Override
 218         public boolean capture(ReplicatedLogEntry lastLogEntry, long replicatedToAllIndex) {
 219             return capture(lastLogEntry, replicatedToAllIndex, null);
 220         }
 221
 222         @Override
 223         public boolean captureToInstall(ReplicatedLogEntry lastLogEntry, long replicatedToAllIndex, String targetFollower) {
 224             return capture(lastLogEntry, replicatedToAllIndex, targetFollower);
 225         }
 226
 227         @Override
 228         public String toString() {
 229             return "Idle";
 230         }
 231
 232         @Override
 233         public long trimLog(long desiredTrimIndex, RaftActorBehavior currentBehavior) {
 234             return doTrimLog(desiredTrimIndex, currentBehavior);
 235         }
 236     }
 237
 238     private class Creating extends AbstractSnapshotState {
 239
 240         @Override
 241         public boolean isCapturing() {
 242             return true;
 243         }
 244
 245         @Override
 246         public void persist(byte[] snapshotBytes, RaftActorBehavior currentBehavior, long totalMemory) {
 247             // create a snapshot object from the state provided and save it
 248             // when snapshot is saved async, SaveSnapshotSuccess is raised.
 249
 250             Snapshot sn = Snapshot.create(snapshotBytes,
 251                     captureSnapshot.getUnAppliedEntries(),
 252                     captureSnapshot.getLastIndex(), captureSnapshot.getLastTerm(),
 253                     captureSnapshot.getLastAppliedIndex(), captureSnapshot.getLastAppliedTerm());
 254
 255             context.getPersistenceProvider().saveSnapshot(sn);
 256
 257             LOG.info("{}: Persisting of snapshot done:{}", persistenceId(), sn.getLogMessage());
 258
 259             long dataThreshold = totalMemory *
 260                     context.getConfigParams().getSnapshotDataThresholdPercentage() / 100;
 261             if (context.getReplicatedLog().dataSize() > dataThreshold) {
 262
 263                 if(LOG.isDebugEnabled()) {
 264                     LOG.debug("{}: dataSize {} exceeds dataThreshold {} - doing snapshotPreCommit with index {}",
 265                             persistenceId(), context.getReplicatedLog().dataSize(), dataThreshold,
 266                             captureSnapshot.getLastAppliedIndex());
 267                 }
 268
 269                 // if memory is less, clear the log based on lastApplied.
 270                 // this could/should only happen if one of the followers is down
 271                 // as normally we keep removing from the log when its replicated to all.
 272                 context.getReplicatedLog().snapshotPreCommit(captureSnapshot.getLastAppliedIndex(),
 273                         captureSnapshot.getLastAppliedTerm());
 274
 275                 // Don't reset replicatedToAllIndex to -1 as this may prevent us from trimming the log after an
 276                 // install snapshot to a follower.
 277                 if(captureSnapshot.getReplicatedToAllIndex() >= 0) {
 278                     currentBehavior.setReplicatedToAllIndex(captureSnapshot.getReplicatedToAllIndex());
 279                 }
 280
 281             } else if(captureSnapshot.getReplicatedToAllIndex() != -1){
 282                 // clear the log based on replicatedToAllIndex
 283                 context.getReplicatedLog().snapshotPreCommit(captureSnapshot.getReplicatedToAllIndex(),
 284                         captureSnapshot.getReplicatedToAllTerm());
 285
 286                 currentBehavior.setReplicatedToAllIndex(captureSnapshot.getReplicatedToAllIndex());
 287             } else {
 288                 // The replicatedToAllIndex was not found in the log
 289                 // This means that replicatedToAllIndex never moved beyond -1 or that it is already in the snapshot.
 290                 // In this scenario we may need to save the snapshot to the akka persistence
 291                 // snapshot for recovery but we do not need to do the replicated log trimming.
 292                 context.getReplicatedLog().snapshotPreCommit(context.getReplicatedLog().getSnapshotIndex(),
 293                         context.getReplicatedLog().getSnapshotTerm());
 294             }
 295
 296             LOG.info("{}: Removed in-memory snapshotted entries, adjusted snaphsotIndex:{} " +
 297                             "and term:{}", persistenceId(), captureSnapshot.getLastAppliedIndex(),
 298                     captureSnapshot.getLastAppliedTerm());
 299
 300             if (context.getId().equals(currentBehavior.getLeaderId())
 301                     && captureSnapshot.isInstallSnapshotInitiated()) {
 302                 // this would be call straight to the leader and won't initiate in serialization
 303                 currentBehavior.handleMessage(context.getActor(), new SendInstallSnapshot(
 304                         ByteString.copyFrom(snapshotBytes)));
 305             }
 306
 307             captureSnapshot = null;
 308             SnapshotManager.this.currentState = PERSISTING;
 309         }
 310
 311         @Override
 312         public String toString() {
 313             return "Creating";
 314         }
 315
 316     }
 317
 318     private class Persisting extends AbstractSnapshotState {
 319
 320         @Override
 321         public void commit(long sequenceNumber) {
 322             context.getReplicatedLog().snapshotCommit();
 323             context.getPersistenceProvider().deleteSnapshots(new SnapshotSelectionCriteria(
 324                     sequenceNumber - context.getConfigParams().getSnapshotBatchCount(), 43200000));
 325
 326             context.getPersistenceProvider().deleteMessages(lastSequenceNumber);
 327
 328             lastSequenceNumber = -1;
 329             SnapshotManager.this.currentState = IDLE;
 330         }
 331
 332         @Override
 333         public void rollback() {
 334             context.getReplicatedLog().snapshotRollback();
 335
 336             LOG.info("{}: Replicated Log rolled back. Snapshot will be attempted in the next cycle." +
 337                             "snapshotIndex:{}, snapshotTerm:{}, log-size:{}", persistenceId(),
 338                     context.getReplicatedLog().getSnapshotIndex(),
 339                     context.getReplicatedLog().getSnapshotTerm(),
 340                     context.getReplicatedLog().size());
 341
 342             SnapshotManager.this.currentState = IDLE;
 343         }
 344
 345         @Override
 346         public String toString() {
 347             return "Persisting";
 348         }
 349
 350     }
 351
 352     private static interface TermInformationReader {
 353         long getIndex();
 354         long getTerm();
 355     }
 356
 357     static class LastAppliedTermInformationReader implements TermInformationReader{
 358         private long index;
 359         private long term;
 360
 361         public LastAppliedTermInformationReader init(ReplicatedLog log, long originalIndex,
 362                                          ReplicatedLogEntry lastLogEntry, boolean hasFollowers){
 363             ReplicatedLogEntry entry = log.get(originalIndex);
 364             this.index = -1L;
 365             this.term = -1L;
 366             if (!hasFollowers) {
 367                 if(lastLogEntry != null) {
 368                     index = lastLogEntry.getIndex();
 369                     term = lastLogEntry.getTerm();
 370                 }
 371             } else if (entry != null) {
 372                 index = entry.getIndex();
 373                 term = entry.getTerm();
 374             } else if(log.getSnapshotIndex() > -1){
 375                 index = log.getSnapshotIndex();
 376                 term = log.getSnapshotTerm();
 377             }
 378             return this;
 379         }
 380
 381         @Override
 382         public long getIndex(){
 383             return this.index;
 384         }
 385
 386         @Override
 387         public long getTerm(){
 388             return this.term;
 389         }
 390     }
 391
 392     private static class ReplicatedToAllTermInformationReader implements TermInformationReader{
 393         private long index;
 394         private long term;
 395
 396         ReplicatedToAllTermInformationReader init(ReplicatedLog log, long originalIndex){
 397             ReplicatedLogEntry entry = log.get(originalIndex);
 398             this.index = -1L;
 399             this.term = -1L;
 400
 401             if (entry != null) {
 402                 index = entry.getIndex();
 403                 term = entry.getTerm();
 404             }
 405
 406             return this;
 407         }
 408
 409         @Override
 410         public long getIndex(){
 411             return this.index;
 412         }
 413
 414         @Override
 415         public long getTerm(){
 416             return this.term;
 417         }
 418     }
 419 }