opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardRecoveryCoordinator.java

   1 /*
   2  * Copyright (c) 2014 Brocade Communications 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 package org.opendaylight.controller.cluster.datastore;
   9
  10 import com.google.common.base.Preconditions;
  11 import java.io.File;
  12 import org.opendaylight.controller.cluster.datastore.persisted.ShardDataTreeSnapshot;
  13 import org.opendaylight.controller.cluster.datastore.persisted.ShardSnapshotState;
  14 import org.opendaylight.controller.cluster.datastore.utils.NormalizedNodeXMLOutput;
  15 import org.opendaylight.controller.cluster.raft.RaftActorRecoveryCohort;
  16 import org.opendaylight.controller.cluster.raft.persisted.Snapshot;
  17 import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
  18 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  19 import org.slf4j.Logger;
  20
  21 /**
  22  * Coordinates persistence recovery of journal log entries and snapshots for a shard. Each snapshot
  23  * and journal log entry batch are de-serialized and applied to their own write transaction
  24  * instance in parallel on a thread pool for faster recovery time. However the transactions are
  25  * committed to the data store in the order the corresponding snapshot or log batch are received
  26  * to preserve data store integrity.
  27  *
  28  * @author Thomas Pantelis
  29  */
  30 abstract class ShardRecoveryCoordinator implements RaftActorRecoveryCohort {
  31     private static final class Simple extends ShardRecoveryCoordinator {
  32         Simple(final ShardDataTree store, final String shardName, final Logger log) {
  33             super(store, shardName, log);
  34         }
  35
  36         @Override
  37         public Snapshot getRestoreFromSnapshot() {
  38             return null;
  39         }
  40     }
  41
  42     private static final class WithSnapshot extends ShardRecoveryCoordinator {
  43         private final Snapshot restoreFromSnapshot;
  44
  45         WithSnapshot(final ShardDataTree store, final String shardName, final Logger log, final Snapshot snapshot) {
  46             super(store, shardName, log);
  47             this.restoreFromSnapshot = Preconditions.checkNotNull(snapshot);
  48         }
  49
  50         @Override
  51         public Snapshot getRestoreFromSnapshot() {
  52             return restoreFromSnapshot;
  53         }
  54     }
  55
  56     private final ShardDataTree store;
  57     private final String shardName;
  58     private final Logger log;
  59
  60     private boolean open;
  61
  62     ShardRecoveryCoordinator(final ShardDataTree store, final String shardName, final Logger log) {
  63         this.store = Preconditions.checkNotNull(store);
  64         this.shardName = Preconditions.checkNotNull(shardName);
  65         this.log = Preconditions.checkNotNull(log);
  66     }
  67
  68     static ShardRecoveryCoordinator create(final ShardDataTree store, final String shardName, final Logger log) {
  69         return new Simple(store, shardName, log);
  70     }
  71
  72     static ShardRecoveryCoordinator forSnapshot(final ShardDataTree store, final String shardName, final Logger log,
  73             final Snapshot snapshot) {
  74         return new WithSnapshot(store, shardName, log, snapshot);
  75     }
  76
  77     @Override
  78     public void startLogRecoveryBatch(final int maxBatchSize) {
  79         log.debug("{}: starting log recovery batch with max size {}", shardName, maxBatchSize);
  80         open = true;
  81     }
  82
  83     @Override
  84     @SuppressWarnings("checkstyle:IllegalCatch")
  85     public void appendRecoveredLogEntry(final Payload payload) {
  86         Preconditions.checkState(open, "call startLogRecovery before calling appendRecoveredLogEntry");
  87
  88         try {
  89             store.applyRecoveryPayload(payload);
  90         } catch (Exception e) {
  91             log.error("{}: failed to apply payload {}", shardName, payload, e);
  92             throw new IllegalStateException(String.format("%s: Failed to apply recovery payload %s",
  93                 shardName, payload), e);
  94         }
  95     }
  96
  97     /**
  98      * Applies the current batched log entries to the data store.
  99      */
 100     @Override
 101     public void applyCurrentLogRecoveryBatch() {
 102         Preconditions.checkState(open, "call startLogRecovery before calling applyCurrentLogRecoveryBatch");
 103         open = false;
 104     }
 105
 106     private File writeRoot(final String kind, final NormalizedNode<?, ?> node) {
 107         final File file = new File(System.getProperty("karaf.data", "."),
 108             "failed-recovery-" + kind + "-" + shardName + ".xml");
 109         NormalizedNodeXMLOutput.toFile(file, node);
 110         return file;
 111     }
 112
 113     /**
 114      * Applies a recovered snapshot to the data store.
 115      *
 116      * @param snapshotState the serialized snapshot
 117      */
 118     @Override
 119     @SuppressWarnings("checkstyle:IllegalCatch")
 120     public void applyRecoverySnapshot(final Snapshot.State snapshotState) {
 121         if (!(snapshotState instanceof ShardSnapshotState)) {
 122             log.debug("{}: applyRecoverySnapshot ignoring snapshot: {}", shardName, snapshotState);
 123         }
 124
 125         log.debug("{}: Applying recovered snapshot", shardName);
 126
 127         ShardDataTreeSnapshot shardSnapshot = ((ShardSnapshotState)snapshotState).getSnapshot();
 128         try {
 129             store.applyRecoverySnapshot(shardSnapshot);
 130         } catch (Exception e) {
 131             final File f = writeRoot("snapshot", shardSnapshot.getRootNode().orElse(null));
 132             throw new IllegalStateException(String.format(
 133                     "%s: Failed to apply recovery snapshot %s. Node data was written to file %s",
 134                     shardName, shardSnapshot, f), e);
 135         }
 136     }
 137 }