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

©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.