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 java.io.IOException;
  11 import org.opendaylight.controller.cluster.datastore.modification.ModificationPayload;
  12 import org.opendaylight.controller.cluster.datastore.modification.MutableCompositeModification;
  13 import org.opendaylight.controller.cluster.datastore.utils.SerializationUtils;
  14 import org.opendaylight.controller.cluster.raft.RaftActorRecoveryCohort;
  15 import org.opendaylight.controller.cluster.raft.protobuff.client.messages.CompositeModificationByteStringPayload;
  16 import org.opendaylight.controller.cluster.raft.protobuff.client.messages.CompositeModificationPayload;
  17 import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
  18 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  19 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  20 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
  21 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidates;
  22 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification;
  23 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataValidationFailedException;
  24 import org.slf4j.Logger;
  25
  26 /**
  27  * Coordinates persistence recovery of journal log entries and snapshots for a shard. Each snapshot
  28  * and journal log entry batch are de-serialized and applied to their own write transaction
  29  * instance in parallel on a thread pool for faster recovery time. However the transactions are
  30  * committed to the data store in the order the corresponding snapshot or log batch are received
  31  * to preserve data store integrity.
  32  *
  33  * @author Thomas Pantelis
  34  */
  35 class ShardRecoveryCoordinator implements RaftActorRecoveryCohort {
  36     private static final YangInstanceIdentifier ROOT = YangInstanceIdentifier.builder().build();
  37     private final DataTree store;
  38     private final String shardName;
  39     private final Logger log;
  40     private DataTreeModification transaction;
  41     private int size;
  42
  43     ShardRecoveryCoordinator(ShardDataTree store, String shardName, Logger log) {
  44         this.store = store.getDataTree();
  45         this.shardName = shardName;
  46         this.log = log;
  47     }
  48
  49     @Override
  50     public void startLogRecoveryBatch(int maxBatchSize) {
  51         log.debug("{}: starting log recovery batch with max size {}", shardName, maxBatchSize);
  52         transaction = store.takeSnapshot().newModification();
  53         size = 0;
  54     }
  55
  56     @Override
  57     public void appendRecoveredLogEntry(Payload payload) {
  58         try {
  59             if (payload instanceof DataTreeCandidatePayload) {
  60                 DataTreeCandidates.applyToModification(transaction, ((DataTreeCandidatePayload)payload).getCandidate());
  61                 size++;
  62             } else if (payload instanceof ModificationPayload) {
  63                 MutableCompositeModification.fromSerializable(
  64                     ((ModificationPayload) payload).getModification()).apply(transaction);
  65                 size++;
  66             } else if (payload instanceof CompositeModificationPayload) {
  67                 MutableCompositeModification.fromSerializable(
  68                     ((CompositeModificationPayload) payload).getModification()).apply(transaction);
  69                 size++;
  70             } else if (payload instanceof CompositeModificationByteStringPayload) {
  71                 MutableCompositeModification.fromSerializable(
  72                         ((CompositeModificationByteStringPayload) payload).getModification()).apply(transaction);
  73                 size++;
  74             } else {
  75                 log.error("{}: Unknown payload {} received during recovery", shardName, payload);
  76             }
  77         } catch (IOException | ClassNotFoundException e) {
  78             log.error("{}: Error extracting ModificationPayload", shardName, e);
  79         }
  80     }
  81
  82     private void commitTransaction(DataTreeModification tx) throws DataValidationFailedException {
  83         tx.ready();
  84         store.validate(tx);
  85         store.commit(store.prepare(tx));
  86     }
  87
  88     /**
  89      * Applies the current batched log entries to the data store.
  90      */
  91     @Override
  92     public void applyCurrentLogRecoveryBatch() {
  93         log.debug("{}: Applying current log recovery batch with size {}", shardName, size);
  94         try {
  95             commitTransaction(transaction);
  96         } catch (DataValidationFailedException e) {
  97             log.error("{}: Failed to apply recovery batch", shardName, e);
  98         }
  99         transaction = null;
 100     }
 101
 102     /**
 103      * Applies a recovered snapshot to the data store.
 104      *
 105      * @param snapshotBytes the serialized snapshot
 106      */
 107     @Override
 108     public void applyRecoverySnapshot(final byte[] snapshotBytes) {
 109         log.debug("{}: Applying recovered snapshot", shardName);
 110
 111         final NormalizedNode<?, ?> node = SerializationUtils.deserializeNormalizedNode(snapshotBytes);
 112         final DataTreeModification tx = store.takeSnapshot().newModification();
 113         tx.write(ROOT, node);
 114         try {
 115             commitTransaction(tx);
 116         } catch (DataValidationFailedException e) {
 117             log.error("{}: Failed to apply recovery snapshot", shardName, e);
 118         }
 119     }
 120 }