378bc717f4da3b22f2516a2c303c43abc9144fab
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / test / java / org / opendaylight / controller / cluster / datastore / AbstractShardTest.java
1 /*
2  * Copyright (c) 2015 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 static org.junit.Assert.assertEquals;
11 import static org.junit.Assert.assertNotNull;
12 import static org.junit.Assert.assertTrue;
13 import static org.junit.Assert.fail;
14 import static org.mockito.Mockito.doAnswer;
15 import static org.mockito.Mockito.mock;
16 import akka.actor.ActorRef;
17 import akka.actor.PoisonPill;
18 import akka.actor.Props;
19 import akka.japi.Creator;
20 import akka.testkit.TestActorRef;
21 import com.google.common.base.Function;
22 import com.google.common.base.Optional;
23 import com.google.common.util.concurrent.CheckedFuture;
24 import com.google.common.util.concurrent.ListenableFuture;
25 import com.google.common.util.concurrent.Uninterruptibles;
26 import java.util.Collections;
27 import java.util.Set;
28 import java.util.concurrent.CountDownLatch;
29 import java.util.concurrent.ExecutionException;
30 import java.util.concurrent.TimeUnit;
31 import java.util.concurrent.atomic.AtomicInteger;
32 import org.junit.After;
33 import org.junit.Assert;
34 import org.junit.Before;
35 import org.mockito.invocation.InvocationOnMock;
36 import org.mockito.stubbing.Answer;
37 import org.opendaylight.controller.cluster.datastore.DatastoreContext.Builder;
38 import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
39 import org.opendaylight.controller.cluster.datastore.modification.MutableCompositeModification;
40 import org.opendaylight.controller.cluster.datastore.modification.WriteModification;
41 import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal;
42 import org.opendaylight.controller.cluster.raft.utils.InMemorySnapshotStore;
43 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
44 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
45 import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStore;
46 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction;
47 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
48 import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
49 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
50 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgument;
51 import org.opendaylight.yangtools.yang.data.api.schema.DataContainerChild;
52 import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode;
53 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
54 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
55
56 /**
57  * Abstract base for shard unit tests.
58  *
59  * @author Thomas Pantelis
60  */
61 public abstract class AbstractShardTest extends AbstractActorTest{
62     protected static final SchemaContext SCHEMA_CONTEXT = TestModel.createTestContext();
63
64     private static final AtomicInteger NEXT_SHARD_NUM = new AtomicInteger();
65
66     protected final ShardIdentifier shardID = ShardIdentifier.builder().memberName("member-1")
67             .shardName("inventory").type("config" + NEXT_SHARD_NUM.getAndIncrement()).build();
68
69     protected final Builder dataStoreContextBuilder = DatastoreContext.newBuilder().
70             shardJournalRecoveryLogBatchSize(3).shardSnapshotBatchCount(5000).
71             shardHeartbeatIntervalInMillis(100);
72
73     @Before
74     public void setUp() {
75         InMemorySnapshotStore.clear();
76         InMemoryJournal.clear();
77     }
78
79     @After
80     public void tearDown() {
81         InMemorySnapshotStore.clear();
82         InMemoryJournal.clear();
83     }
84
85     protected DatastoreContext newDatastoreContext() {
86         return dataStoreContextBuilder.build();
87     }
88
89     protected Props newShardProps() {
90         return Shard.props(shardID, Collections.<String,String>emptyMap(),
91                 newDatastoreContext(), SCHEMA_CONTEXT);
92     }
93
94     protected void testRecovery(Set<Integer> listEntryKeys) throws Exception {
95         // Create the actor and wait for recovery complete.
96
97         int nListEntries = listEntryKeys.size();
98
99         final CountDownLatch recoveryComplete = new CountDownLatch(1);
100
101         @SuppressWarnings("serial")
102         Creator<Shard> creator = new Creator<Shard>() {
103             @Override
104             public Shard create() throws Exception {
105                 return new Shard(shardID, Collections.<String,String>emptyMap(),
106                         newDatastoreContext(), SCHEMA_CONTEXT) {
107                     @Override
108                     protected void onRecoveryComplete() {
109                         try {
110                             super.onRecoveryComplete();
111                         } finally {
112                             recoveryComplete.countDown();
113                         }
114                     }
115                 };
116             }
117         };
118
119         TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
120                 Props.create(new DelegatingShardCreator(creator)), "testRecovery");
121
122         assertEquals("Recovery complete", true, recoveryComplete.await(5, TimeUnit.SECONDS));
123
124         // Verify data in the data store.
125
126         NormalizedNode<?, ?> outerList = readStore(shard, TestModel.OUTER_LIST_PATH);
127         assertNotNull(TestModel.OUTER_LIST_QNAME.getLocalName() + " not found", outerList);
128         assertTrue(TestModel.OUTER_LIST_QNAME.getLocalName() + " value is not Iterable",
129                 outerList.getValue() instanceof Iterable);
130         for(Object entry: (Iterable<?>) outerList.getValue()) {
131             assertTrue(TestModel.OUTER_LIST_QNAME.getLocalName() + " entry is not MapEntryNode",
132                     entry instanceof MapEntryNode);
133             MapEntryNode mapEntry = (MapEntryNode)entry;
134             Optional<DataContainerChild<? extends PathArgument, ?>> idLeaf =
135                     mapEntry.getChild(new YangInstanceIdentifier.NodeIdentifier(TestModel.ID_QNAME));
136             assertTrue("Missing leaf " + TestModel.ID_QNAME.getLocalName(), idLeaf.isPresent());
137             Object value = idLeaf.get().getValue();
138             assertTrue("Unexpected value for leaf "+ TestModel.ID_QNAME.getLocalName() + ": " + value,
139                     listEntryKeys.remove(value));
140         }
141
142         if(!listEntryKeys.isEmpty()) {
143             fail("Missing " + TestModel.OUTER_LIST_QNAME.getLocalName() + " entries with keys: " +
144                     listEntryKeys);
145         }
146
147         assertEquals("Last log index", nListEntries,
148                 shard.underlyingActor().getShardMBean().getLastLogIndex());
149         assertEquals("Commit index", nListEntries,
150                 shard.underlyingActor().getShardMBean().getCommitIndex());
151         assertEquals("Last applied", nListEntries,
152                 shard.underlyingActor().getShardMBean().getLastApplied());
153
154         shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
155     }
156
157     protected void verifyLastApplied(TestActorRef<Shard> shard, long expectedValue) {
158         long lastApplied = -1;
159         for(int i = 0; i < 20 * 5; i++) {
160             lastApplied = shard.underlyingActor().getShardMBean().getLastApplied();
161             if(lastApplied == expectedValue) {
162                 return;
163             }
164             Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS);
165         }
166
167         Assert.fail(String.format("Expected last applied: %d, Actual: %d", expectedValue, lastApplied));
168     }
169
170     protected NormalizedNode<?, ?> readStore(final InMemoryDOMDataStore store) throws ReadFailedException {
171         DOMStoreReadTransaction transaction = store.newReadOnlyTransaction();
172         CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> read =
173             transaction.read(YangInstanceIdentifier.builder().build());
174
175         Optional<NormalizedNode<?, ?>> optional = read.checkedGet();
176
177         NormalizedNode<?, ?> normalizedNode = optional.get();
178
179         transaction.close();
180
181         return normalizedNode;
182     }
183
184     protected DOMStoreThreePhaseCommitCohort setupMockWriteTransaction(final String cohortName,
185             final InMemoryDOMDataStore dataStore, final YangInstanceIdentifier path, final NormalizedNode<?, ?> data,
186             final MutableCompositeModification modification) {
187         return setupMockWriteTransaction(cohortName, dataStore, path, data, modification, null);
188     }
189
190     protected DOMStoreThreePhaseCommitCohort setupMockWriteTransaction(final String cohortName,
191             final InMemoryDOMDataStore dataStore, final YangInstanceIdentifier path, final NormalizedNode<?, ?> data,
192             final MutableCompositeModification modification,
193             final Function<DOMStoreThreePhaseCommitCohort,ListenableFuture<Void>> preCommit) {
194
195         DOMStoreWriteTransaction tx = dataStore.newWriteOnlyTransaction();
196         tx.write(path, data);
197         DOMStoreThreePhaseCommitCohort cohort = createDelegatingMockCohort(cohortName, tx.ready(), preCommit);
198
199         modification.addModification(new WriteModification(path, data));
200
201         return cohort;
202     }
203
204     protected DOMStoreThreePhaseCommitCohort createDelegatingMockCohort(final String cohortName,
205             final DOMStoreThreePhaseCommitCohort actual) {
206         return createDelegatingMockCohort(cohortName, actual, null);
207     }
208
209     protected DOMStoreThreePhaseCommitCohort createDelegatingMockCohort(final String cohortName,
210             final DOMStoreThreePhaseCommitCohort actual,
211             final Function<DOMStoreThreePhaseCommitCohort,ListenableFuture<Void>> preCommit) {
212         DOMStoreThreePhaseCommitCohort cohort = mock(DOMStoreThreePhaseCommitCohort.class, cohortName);
213
214         doAnswer(new Answer<ListenableFuture<Boolean>>() {
215             @Override
216             public ListenableFuture<Boolean> answer(final InvocationOnMock invocation) {
217                 return actual.canCommit();
218             }
219         }).when(cohort).canCommit();
220
221         doAnswer(new Answer<ListenableFuture<Void>>() {
222             @Override
223             public ListenableFuture<Void> answer(final InvocationOnMock invocation) throws Throwable {
224                 return actual.preCommit();
225             }
226         }).when(cohort).preCommit();
227
228         doAnswer(new Answer<ListenableFuture<Void>>() {
229             @Override
230             public ListenableFuture<Void> answer(final InvocationOnMock invocation) throws Throwable {
231                 return actual.commit();
232             }
233         }).when(cohort).commit();
234
235         doAnswer(new Answer<ListenableFuture<Void>>() {
236             @Override
237             public ListenableFuture<Void> answer(final InvocationOnMock invocation) throws Throwable {
238                 return actual.abort();
239             }
240         }).when(cohort).abort();
241
242         return cohort;
243     }
244
245     public static NormalizedNode<?,?> readStore(final TestActorRef<Shard> shard, final YangInstanceIdentifier id)
246             throws ExecutionException, InterruptedException {
247         return readStore(shard.underlyingActor().getDataStore(), id);
248     }
249
250     public static NormalizedNode<?,?> readStore(final InMemoryDOMDataStore store, final YangInstanceIdentifier id)
251             throws ExecutionException, InterruptedException {
252         DOMStoreReadTransaction transaction = store.newReadOnlyTransaction();
253
254         CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> future =
255             transaction.read(id);
256
257         Optional<NormalizedNode<?, ?>> optional = future.get();
258         NormalizedNode<?, ?> node = optional.isPresent()? optional.get() : null;
259
260         transaction.close();
261
262         return node;
263     }
264
265     public static void writeToStore(final TestActorRef<Shard> shard, final YangInstanceIdentifier id,
266             final NormalizedNode<?,?> node) throws ExecutionException, InterruptedException {
267         writeToStore(shard.underlyingActor().getDataStore(), id, node);
268     }
269
270     public static void writeToStore(final InMemoryDOMDataStore store, final YangInstanceIdentifier id,
271             final NormalizedNode<?,?> node) throws ExecutionException, InterruptedException {
272         DOMStoreWriteTransaction transaction = store.newWriteOnlyTransaction();
273
274         transaction.write(id, node);
275
276         DOMStoreThreePhaseCommitCohort commitCohort = transaction.ready();
277         commitCohort.preCommit().get();
278         commitCohort.commit().get();
279     }
280
281     @SuppressWarnings("serial")
282     public static final class DelegatingShardCreator implements Creator<Shard> {
283         private final Creator<Shard> delegate;
284
285         DelegatingShardCreator(final Creator<Shard> delegate) {
286             this.delegate = delegate;
287         }
288
289         @Override
290         public Shard create() throws Exception {
291             return delegate.create();
292         }
293     }
294 }

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.