import org.opendaylight.controller.cluster.datastore.persisted.CreateLocalHistoryPayload;
import org.opendaylight.controller.cluster.datastore.persisted.DataTreeCandidateInputOutput.DataTreeCandidateWithVersion;
import org.opendaylight.controller.cluster.datastore.persisted.MetadataShardDataTreeSnapshot;
+import org.opendaylight.controller.cluster.datastore.persisted.PayloadVersion;
import org.opendaylight.controller.cluster.datastore.persisted.PurgeLocalHistoryPayload;
import org.opendaylight.controller.cluster.datastore.persisted.PurgeTransactionPayload;
import org.opendaylight.controller.cluster.datastore.persisted.ShardDataTreeSnapshot;
import org.opendaylight.controller.cluster.datastore.persisted.ShardDataTreeSnapshotMetadata;
+import org.opendaylight.controller.cluster.datastore.persisted.ShardSnapshotState;
import org.opendaylight.controller.cluster.datastore.utils.DataTreeModificationOutput;
import org.opendaylight.controller.cluster.datastore.utils.PruningDataTreeModification;
import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeTip;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataValidationFailedException;
import org.opendaylight.yangtools.yang.data.api.schema.tree.TreeType;
+import org.opendaylight.yangtools.yang.data.codec.binfmt.NormalizedNodeStreamVersion;
import org.opendaylight.yangtools.yang.data.impl.schema.tree.InMemoryDataTreeFactory;
import org.opendaylight.yangtools.yang.data.util.DataSchemaContextTree;
import org.opendaylight.yangtools.yang.model.api.SchemaContext;
* @throws DataValidationFailedException when the snapshot fails to apply
*/
void applySnapshot(final @NonNull ShardDataTreeSnapshot snapshot) throws DataValidationFailedException {
+ // TODO: we should be taking ShardSnapshotState here and performing forward-compatibility translation
applySnapshot(snapshot, UnaryOperator.identity());
}
- private PruningDataTreeModification wrapWithPruning(final DataTreeModification delegate) {
- return new PruningDataTreeModification(delegate, dataTree,
- // TODO: we should be able to reuse the pruner, provided we are not reentrant
- ReusableNormalizedNodePruner.forDataSchemaContext(dataSchemaContext));
- }
-
/**
* Apply a snapshot coming from recovery. This method does not assume the SchemaContexts match and performs data
* pruning in an attempt to adjust the state to our current SchemaContext.
* @param snapshot Snapshot that needs to be applied
* @throws DataValidationFailedException when the snapshot fails to apply
*/
- void applyRecoverySnapshot(final @NonNull ShardDataTreeSnapshot snapshot) throws DataValidationFailedException {
- applySnapshot(snapshot, this::wrapWithPruning);
+ void applyRecoverySnapshot(final @NonNull ShardSnapshotState snapshot) throws DataValidationFailedException {
+ // TODO: we should be able to reuse the pruner, provided we are not reentrant
+ ReusableNormalizedNodePruner pruner = ReusableNormalizedNodePruner.forDataSchemaContext(dataSchemaContext);
+ if (snapshot.needsMigration()) {
+ pruner = pruner.withUintAdaption();
+ }
+
+ // For lambda below
+ final ReusableNormalizedNodePruner finalPruner = pruner;
+ applySnapshot(snapshot.getSnapshot(),
+ delegate -> new PruningDataTreeModification(delegate, dataTree, finalPruner));
}
@SuppressWarnings("checkstyle:IllegalCatch")
private void applyRecoveryCandidate(final CommitTransactionPayload payload) throws IOException {
final Entry<TransactionIdentifier, DataTreeCandidateWithVersion> entry = payload.getCandidate();
final DataTreeModification unwrapped = dataTree.takeSnapshot().newModification();
- // FIXME: CONTROLLER-1923: examine version first
- final PruningDataTreeModification mod = wrapWithPruning(unwrapped);
+
+ // TODO: we should be able to reuse the pruner, provided we are not reentrant
+ ReusableNormalizedNodePruner pruner = ReusableNormalizedNodePruner.forDataSchemaContext(dataSchemaContext);
+ if (NormalizedNodeStreamVersion.MAGNESIUM.compareTo(entry.getValue().getVersion()) > 0) {
+ pruner = pruner.withUintAdaption();
+ }
+
+ final PruningDataTreeModification mod = new PruningDataTreeModification(unwrapped, dataTree, pruner);
DataTreeCandidates.applyToModification(mod, entry.getValue().getCandidate());
mod.ready();
final TransactionIdentifier txId = cohort.getIdentifier();
final Payload payload;
try {
- payload = CommitTransactionPayload.create(txId, candidate,
+ payload = CommitTransactionPayload.create(txId, candidate, PayloadVersion.current(),
shard.getDatastoreContext().getInitialPayloadSerializedBufferCapacity());
} catch (IOException e) {
LOG.error("{}: Failed to encode transaction {} candidate {}", logContext, txId, candidate, e);
public void applyRecoverySnapshot(final Snapshot.State snapshotState) {
if (!(snapshotState instanceof ShardSnapshotState)) {
log.debug("{}: applyRecoverySnapshot ignoring snapshot: {}", shardName, snapshotState);
+ return;
}
log.debug("{}: Applying recovered snapshot", shardName);
-
- ShardDataTreeSnapshot shardSnapshot = ((ShardSnapshotState)snapshotState).getSnapshot();
+ final ShardSnapshotState shardSnapshotState = (ShardSnapshotState)snapshotState;
try {
- store.applyRecoverySnapshot(shardSnapshot);
+ store.applyRecoverySnapshot(shardSnapshotState);
} catch (Exception e) {
+ final ShardDataTreeSnapshot shardSnapshot = shardSnapshotState.getSnapshot();
final File f = writeRoot("snapshot", shardSnapshot.getRootNode().orElse(null));
throw new IllegalStateException(String.format(
"%s: Failed to apply recovery snapshot %s. Node data was written to file %s",
import java.io.StreamCorruptedException;
import java.util.AbstractMap.SimpleImmutableEntry;
import java.util.Map.Entry;
+import org.eclipse.jdt.annotation.NonNull;
import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
import org.opendaylight.controller.cluster.datastore.persisted.DataTreeCandidateInputOutput.DataTreeCandidateWithVersion;
import org.opendaylight.controller.cluster.raft.protobuff.client.messages.IdentifiablePayload;
}
- public static CommitTransactionPayload create(final TransactionIdentifier transactionId,
- final DataTreeCandidate candidate, final int initialSerializedBufferCapacity) throws IOException {
-
+ public static @NonNull CommitTransactionPayload create(final TransactionIdentifier transactionId,
+ final DataTreeCandidate candidate, final PayloadVersion version, final int initialSerializedBufferCapacity)
+ throws IOException {
final ChunkedOutputStream cos = new ChunkedOutputStream(initialSerializedBufferCapacity);
try (DataOutputStream dos = new DataOutputStream(cos)) {
transactionId.writeTo(dos);
- DataTreeCandidateInputOutput.writeDataTreeCandidate(dos, candidate);
+ DataTreeCandidateInputOutput.writeDataTreeCandidate(dos, version, candidate);
}
final Variant<byte[], ChunkedByteArray> source = cos.toVariant();
}
@VisibleForTesting
- public static CommitTransactionPayload create(final TransactionIdentifier transactionId,
+ public static @NonNull CommitTransactionPayload create(final TransactionIdentifier transactionId,
+ final DataTreeCandidate candidate, final PayloadVersion version) throws IOException {
+ return create(transactionId, candidate, version, 512);
+ }
+
+ @VisibleForTesting
+ public static @NonNull CommitTransactionPayload create(final TransactionIdentifier transactionId,
final DataTreeCandidate candidate) throws IOException {
- return create(transactionId, candidate, 512);
+ return create(transactionId, candidate, PayloadVersion.current());
}
- public Entry<TransactionIdentifier, DataTreeCandidateWithVersion> getCandidate() throws IOException {
+ public @NonNull Entry<TransactionIdentifier, DataTreeCandidateWithVersion> getCandidate() throws IOException {
Entry<TransactionIdentifier, DataTreeCandidateWithVersion> localCandidate = candidate;
if (localCandidate == null) {
synchronized (this) {
return localCandidate;
}
- public final Entry<TransactionIdentifier, DataTreeCandidateWithVersion> getCandidate(
+ public final @NonNull Entry<TransactionIdentifier, DataTreeCandidateWithVersion> getCandidate(
final ReusableStreamReceiver receiver) throws IOException {
final DataInput in = newDataInput();
return new SimpleImmutableEntry<>(TransactionIdentifier.readFrom(in),
import static java.util.Objects.requireNonNull;
import com.google.common.annotations.Beta;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableList;
import java.io.DataInput;
import java.io.DataOutput;
}
}
- public static void writeDataTreeCandidate(final DataOutput out, final DataTreeCandidate candidate)
- throws IOException {
- try (NormalizedNodeDataOutput writer = PayloadVersion.current().getStreamVersion().newDataOutput(out)) {
+ @VisibleForTesting
+ public static void writeDataTreeCandidate(final DataOutput out, final PayloadVersion version,
+ final DataTreeCandidate candidate) throws IOException {
+ try (NormalizedNodeDataOutput writer = version.getStreamVersion().newDataOutput(out)) {
writer.writeYangInstanceIdentifier(candidate.getRootPath());
final DataTreeCandidateNode node = candidate.getRootNode();
}
}
+ public static void writeDataTreeCandidate(final DataOutput out, final DataTreeCandidate candidate)
+ throws IOException {
+ writeDataTreeCandidate(out, PayloadVersion.current(), candidate);
+ }
+
private static void throwUnhandledNodeType(final DataTreeCandidateNode node) {
throw new IllegalArgumentException("Unhandled node type " + node.getModificationType());
}
import static java.util.Objects.requireNonNull;
+import com.google.common.annotations.VisibleForTesting;
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
import java.io.Externalizable;
import java.io.IOException;
private final @NonNull ShardDataTreeSnapshot snapshot;
private final boolean migrated;
- ShardSnapshotState(final @NonNull ShardDataTreeSnapshot snapshot, final boolean migrated) {
+ @VisibleForTesting
+ public ShardSnapshotState(final @NonNull ShardDataTreeSnapshot snapshot, final boolean migrated) {
this.snapshot = requireNonNull(snapshot);
this.migrated = migrated;
}
import com.google.common.base.Ticker;
import com.google.common.primitives.UnsignedLong;
import com.google.common.util.concurrent.FutureCallback;
+import java.io.IOException;
+import java.math.BigInteger;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import org.mockito.Mockito;
import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats;
import org.opendaylight.controller.cluster.datastore.persisted.CommitTransactionPayload;
+import org.opendaylight.controller.cluster.datastore.persisted.MetadataShardDataTreeSnapshot;
+import org.opendaylight.controller.cluster.datastore.persisted.PayloadVersion;
+import org.opendaylight.controller.cluster.datastore.persisted.ShardSnapshotState;
import org.opendaylight.controller.md.cluster.datastore.model.CarsModel;
import org.opendaylight.controller.md.cluster.datastore.model.PeopleModel;
import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
import org.opendaylight.mdsal.dom.api.DOMDataTreeChangeListener;
import org.opendaylight.yangtools.yang.common.Uint64;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifierWithPredicates;
+import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode;
import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidates;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeSnapshot;
+import org.opendaylight.yangtools.yang.data.api.schema.tree.DataValidationFailedException;
import org.opendaylight.yangtools.yang.data.api.schema.tree.ModificationType;
import org.opendaylight.yangtools.yang.data.api.schema.tree.TreeType;
+import org.opendaylight.yangtools.yang.data.impl.schema.Builders;
+import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
import org.opendaylight.yangtools.yang.model.api.SchemaContext;
public class ShardDataTreeTest extends AbstractTest {
assertEquals("People node", peopleNode, optional.get());
}
+ @Test
+ public void testUintCommitPayload() throws IOException {
+ shardDataTree.applyRecoveryPayload(CommitTransactionPayload.create(nextTransactionId(),
+ DataTreeCandidates.fromNormalizedNode(YangInstanceIdentifier.empty(), bigIntegerRoot()),
+ PayloadVersion.SODIUM_SR1));
+
+ assertCarsUint64();
+ }
+
+ @Test
+ public void testUintSnapshot() throws IOException, DataValidationFailedException {
+ shardDataTree.applyRecoverySnapshot(new ShardSnapshotState(new MetadataShardDataTreeSnapshot(bigIntegerRoot()),
+ true));
+
+ assertCarsUint64();
+ }
+
+ private void assertCarsUint64() {
+ final DataTreeSnapshot snapshot = shardDataTree.newReadOnlyTransaction(nextTransactionId()).getSnapshot();
+ final NormalizedNode<?, ?> cars = snapshot.readNode(CarsModel.CAR_LIST_PATH).get();
+
+ assertEquals(Builders.mapBuilder()
+ .withNodeIdentifier(new NodeIdentifier(CarsModel.CAR_QNAME))
+ .withChild(Builders.mapEntryBuilder()
+ .withNodeIdentifier(NodeIdentifierWithPredicates.of(CarsModel.CAR_QNAME,
+ CarsModel.CAR_NAME_QNAME, "foo"))
+ .withChild(ImmutableNodes.leafNode(CarsModel.CAR_NAME_QNAME, "foo"))
+ // Note: Uint64
+ .withChild(ImmutableNodes.leafNode(CarsModel.CAR_PRICE_QNAME, Uint64.ONE))
+ .build())
+ .build(), cars);
+ }
+
+ private static ContainerNode bigIntegerRoot() {
+ return Builders.containerBuilder()
+ .withNodeIdentifier(new NodeIdentifier(SchemaContext.NAME))
+ .withChild(Builders.containerBuilder()
+ .withNodeIdentifier(new NodeIdentifier(CarsModel.CARS_QNAME))
+ .withChild(Builders.mapBuilder()
+ .withNodeIdentifier(new NodeIdentifier(CarsModel.CAR_QNAME))
+ .withChild(Builders.mapEntryBuilder()
+ .withNodeIdentifier(NodeIdentifierWithPredicates.of(CarsModel.CAR_QNAME,
+ CarsModel.CAR_NAME_QNAME, "foo"))
+ .withChild(ImmutableNodes.leafNode(CarsModel.CAR_NAME_QNAME, "foo"))
+ // Note: old BigInteger
+ .withChild(ImmutableNodes.leafNode(CarsModel.CAR_PRICE_QNAME, BigInteger.ONE))
+ .build())
+ .build())
+ .build())
+ .build();
+ }
+
private ShardDataTreeCohort newShardDataTreeCohort(final DataTreeOperation operation) {
final ReadWriteShardDataTreeTransaction transaction =
shardDataTree.newReadWriteTransaction(nextTransactionId());