opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/SnapshotManagerTest.java

   1 package org.opendaylight.controller.cluster.raft;
   2
   3 import static junit.framework.TestCase.assertTrue;
   4 import static org.junit.Assert.assertEquals;
   5 import static org.mockito.Matchers.any;
   6 import static org.mockito.Matchers.anyLong;
   7 import static org.mockito.Mockito.doReturn;
   8 import static org.mockito.Mockito.mock;
   9 import static org.mockito.Mockito.never;
  10 import static org.mockito.Mockito.times;
  11 import static org.mockito.Mockito.verify;
  12 import akka.actor.ActorRef;
  13 import akka.japi.Procedure;
  14 import akka.persistence.SnapshotSelectionCriteria;
  15 import akka.testkit.TestActorRef;
  16 import java.util.Arrays;
  17 import java.util.HashMap;
  18 import java.util.List;
  19 import org.junit.After;
  20 import org.junit.Before;
  21 import org.junit.Test;
  22 import org.mockito.ArgumentCaptor;
  23 import org.mockito.Mock;
  24 import org.mockito.Mockito;
  25 import org.mockito.MockitoAnnotations;
  26 import org.opendaylight.controller.cluster.DataPersistenceProvider;
  27 import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshot;
  28 import org.opendaylight.controller.cluster.raft.base.messages.SendInstallSnapshot;
  29 import org.opendaylight.controller.cluster.raft.behaviors.RaftActorBehavior;
  30 import org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor;
  31 import org.slf4j.LoggerFactory;
  32
  33 public class SnapshotManagerTest extends AbstractActorTest {
  34
  35     @Mock
  36     private RaftActorContext mockRaftActorContext;
  37
  38     @Mock
  39     private ConfigParams mockConfigParams;
  40
  41     @Mock
  42     private ReplicatedLog mockReplicatedLog;
  43
  44     @Mock
  45     private DataPersistenceProvider mockDataPersistenceProvider;
  46
  47     @Mock
  48     private RaftActorBehavior mockRaftActorBehavior;
  49
  50     @Mock
  51     private Procedure<Void> mockProcedure;
  52
  53     private SnapshotManager snapshotManager;
  54
  55     private TestActorFactory factory;
  56
  57     private TestActorRef<MessageCollectorActor> actorRef;
  58
  59     @Before
  60     public void setUp(){
  61         MockitoAnnotations.initMocks(this);
  62
  63         doReturn(new HashMap<>()).when(mockRaftActorContext).getPeerAddresses();
  64         doReturn(mockConfigParams).when(mockRaftActorContext).getConfigParams();
  65         doReturn(10L).when(mockConfigParams).getSnapshotBatchCount();
  66         doReturn(mockReplicatedLog).when(mockRaftActorContext).getReplicatedLog();
  67         doReturn("123").when(mockRaftActorContext).getId();
  68         doReturn("123").when(mockRaftActorBehavior).getLeaderId();
  69
  70         snapshotManager = new SnapshotManager(mockRaftActorContext, LoggerFactory.getLogger(this.getClass()));
  71         factory = new TestActorFactory(getSystem());
  72
  73         actorRef = factory.createTestActor(MessageCollectorActor.props(), factory.generateActorId("test-"));
  74         doReturn(actorRef).when(mockRaftActorContext).getActor();
  75
  76     }
  77
  78     @After
  79     public void tearDown(){
  80         factory.close();
  81     }
  82
  83     @Test
  84     public void testConstruction(){
  85         assertEquals(false, snapshotManager.isCapturing());
  86     }
  87
  88     @Test
  89     public void testCaptureToInstall(){
  90
  91         // Force capturing toInstall = true
  92         snapshotManager.captureToInstall(new MockRaftActorContext.MockReplicatedLogEntry(1, 0,
  93                 new MockRaftActorContext.MockPayload()), 0);
  94
  95         assertEquals(true, snapshotManager.isCapturing());
  96
  97         CaptureSnapshot captureSnapshot = MessageCollectorActor.expectFirstMatching(actorRef, CaptureSnapshot.class);
  98
  99         // LastIndex and LastTerm are picked up from the lastLogEntry
 100         assertEquals(0L, captureSnapshot.getLastIndex());
 101         assertEquals(1L, captureSnapshot.getLastTerm());
 102
 103         // Since the actor does not have any followers (no peer addresses) lastApplied will be from lastLogEntry
 104         assertEquals(0L, captureSnapshot.getLastAppliedIndex());
 105         assertEquals(1L, captureSnapshot.getLastAppliedTerm());
 106
 107         //
 108         assertEquals(-1L, captureSnapshot.getReplicatedToAllIndex());
 109         assertEquals(-1L, captureSnapshot.getReplicatedToAllTerm());
 110         actorRef.underlyingActor().clear();
 111     }
 112
 113     @Test
 114     public void testCapture(){
 115         snapshotManager.capture(new MockRaftActorContext.MockReplicatedLogEntry(1,9,
 116                 new MockRaftActorContext.MockPayload()), 9);
 117
 118         assertEquals(true, snapshotManager.isCapturing());
 119
 120         CaptureSnapshot captureSnapshot = MessageCollectorActor.expectFirstMatching(actorRef, CaptureSnapshot.class);
 121         // LastIndex and LastTerm are picked up from the lastLogEntry
 122         assertEquals(9L, captureSnapshot.getLastIndex());
 123         assertEquals(1L, captureSnapshot.getLastTerm());
 124
 125         // Since the actor does not have any followers (no peer addresses) lastApplied will be from lastLogEntry
 126         assertEquals(9L, captureSnapshot.getLastAppliedIndex());
 127         assertEquals(1L, captureSnapshot.getLastAppliedTerm());
 128
 129         //
 130         assertEquals(-1L, captureSnapshot.getReplicatedToAllIndex());
 131         assertEquals(-1L, captureSnapshot.getReplicatedToAllTerm());
 132
 133         actorRef.underlyingActor().clear();
 134
 135     }
 136
 137     @Test
 138     public void testIllegalCapture() throws Exception {
 139         snapshotManager.capture(new MockRaftActorContext.MockReplicatedLogEntry(1,9,
 140                 new MockRaftActorContext.MockPayload()), 9);
 141
 142         List<CaptureSnapshot> allMatching = MessageCollectorActor.getAllMatching(actorRef, CaptureSnapshot.class);
 143
 144         assertEquals(1, allMatching.size());
 145
 146         // This will not cause snapshot capture to start again
 147         snapshotManager.capture(new MockRaftActorContext.MockReplicatedLogEntry(1,9,
 148                 new MockRaftActorContext.MockPayload()), 9);
 149
 150         allMatching = MessageCollectorActor.getAllMatching(actorRef, CaptureSnapshot.class);
 151
 152         assertEquals(1, allMatching.size());
 153     }
 154
 155     @Test
 156     public void testPersistWhenReplicatedToAllIndexMinusOne(){
 157         doReturn("123").when(mockRaftActorContext).getId();
 158         doReturn(45L).when(mockReplicatedLog).getSnapshotIndex();
 159         doReturn(6L).when(mockReplicatedLog).getSnapshotTerm();
 160
 161         // when replicatedToAllIndex = -1
 162         snapshotManager.capture(new MockRaftActorContext.MockReplicatedLogEntry(6,9,
 163                 new MockRaftActorContext.MockPayload()), -1);
 164
 165         snapshotManager.create(mockProcedure);
 166
 167         byte[] bytes = new byte[] {1,2,3,4,5,6,7,8,9,10};
 168         snapshotManager.persist(mockDataPersistenceProvider, bytes, mockRaftActorBehavior);
 169
 170         ArgumentCaptor<Snapshot> snapshotArgumentCaptor = ArgumentCaptor.forClass(Snapshot.class);
 171         verify(mockDataPersistenceProvider).saveSnapshot(snapshotArgumentCaptor.capture());
 172
 173         Snapshot snapshot = snapshotArgumentCaptor.getValue();
 174
 175         assertEquals(6, snapshot.getLastAppliedTerm());
 176         assertEquals(9, snapshot.getLastAppliedIndex());
 177         assertEquals(9, snapshot.getLastIndex());
 178         assertEquals(6, snapshot.getLastTerm());
 179         assertEquals(10, snapshot.getState().length);
 180         assertTrue(Arrays.equals(bytes, snapshot.getState()));
 181         assertEquals(0, snapshot.getUnAppliedEntries().size());
 182
 183         verify(mockReplicatedLog).snapshotPreCommit(45L, 6L);
 184     }
 185
 186
 187     @Test
 188     public void testCreate() throws Exception {
 189         // when replicatedToAllIndex = -1
 190         snapshotManager.capture(new MockRaftActorContext.MockReplicatedLogEntry(6,9,
 191                 new MockRaftActorContext.MockPayload()), -1);
 192
 193         snapshotManager.create(mockProcedure);
 194
 195         verify(mockProcedure).apply(null);
 196     }
 197
 198     @Test
 199     public void testCallingCreateMultipleTimesCausesNoHarm() throws Exception {
 200         // when replicatedToAllIndex = -1
 201         snapshotManager.capture(new MockRaftActorContext.MockReplicatedLogEntry(6,9,
 202                 new MockRaftActorContext.MockPayload()), -1);
 203
 204         snapshotManager.create(mockProcedure);
 205
 206         snapshotManager.create(mockProcedure);
 207
 208         verify(mockProcedure, times(1)).apply(null);
 209     }
 210
 211     @Test
 212     public void testCallingCreateBeforeCapture() throws Exception {
 213         snapshotManager.create(mockProcedure);
 214
 215         verify(mockProcedure, times(0)).apply(null);
 216     }
 217
 218     @Test
 219     public void testCallingCreateAfterPersist() throws Exception {
 220         // when replicatedToAllIndex = -1
 221         snapshotManager.capture(new MockRaftActorContext.MockReplicatedLogEntry(6,9,
 222                 new MockRaftActorContext.MockPayload()), -1);
 223
 224         snapshotManager.create(mockProcedure);
 225
 226         verify(mockProcedure, times(1)).apply(null);
 227
 228         snapshotManager.persist(mockDataPersistenceProvider, new byte[]{}, mockRaftActorBehavior);
 229
 230         Mockito.reset(mockProcedure);
 231
 232         snapshotManager.create(mockProcedure);
 233
 234         verify(mockProcedure, never()).apply(null);
 235     }
 236
 237     @Test
 238     public void testPersistWhenReplicatedToAllIndexNotMinus(){
 239         doReturn(45L).when(mockReplicatedLog).getSnapshotIndex();
 240         doReturn(6L).when(mockReplicatedLog).getSnapshotTerm();
 241         ReplicatedLogEntry replicatedLogEntry = mock(ReplicatedLogEntry.class);
 242         doReturn(replicatedLogEntry).when(mockReplicatedLog).get(9);
 243         doReturn(6L).when(replicatedLogEntry).getTerm();
 244         doReturn(9L).when(replicatedLogEntry).getIndex();
 245
 246         // when replicatedToAllIndex != -1
 247         snapshotManager.capture(new MockRaftActorContext.MockReplicatedLogEntry(6,9,
 248                 new MockRaftActorContext.MockPayload()), 9);
 249
 250         snapshotManager.create(mockProcedure);
 251
 252         snapshotManager.persist(mockDataPersistenceProvider, new byte[]{}, mockRaftActorBehavior);
 253
 254         verify(mockDataPersistenceProvider).saveSnapshot(any(Snapshot.class));
 255
 256         verify(mockReplicatedLog).snapshotPreCommit(9L, 6L);
 257
 258         verify(mockRaftActorBehavior).setReplicatedToAllIndex(9);
 259     }
 260
 261
 262     @Test
 263     public void testPersistWhenReplicatedLogDataSizeGreaterThanThreshold(){
 264         doReturn(Integer.MAX_VALUE).when(mockReplicatedLog).dataSize();
 265
 266         // when replicatedToAllIndex = -1
 267         snapshotManager.capture(new MockRaftActorContext.MockReplicatedLogEntry(6,9,
 268                 new MockRaftActorContext.MockPayload()), -1);
 269
 270         snapshotManager.create(mockProcedure);
 271
 272         snapshotManager.persist(mockDataPersistenceProvider, new byte[]{}, mockRaftActorBehavior);
 273
 274         verify(mockDataPersistenceProvider).saveSnapshot(any(Snapshot.class));
 275
 276         verify(mockReplicatedLog).snapshotPreCommit(9L, 6L);
 277
 278         verify(mockRaftActorBehavior).setReplicatedToAllIndex(-1);
 279     }
 280
 281     @Test
 282     public void testPersistSendInstallSnapshot(){
 283         doReturn(Integer.MAX_VALUE).when(mockReplicatedLog).dataSize();
 284
 285         // when replicatedToAllIndex = -1
 286         snapshotManager.captureToInstall(new MockRaftActorContext.MockReplicatedLogEntry(6, 9,
 287                 new MockRaftActorContext.MockPayload()), -1);
 288
 289         snapshotManager.create(mockProcedure);
 290
 291         byte[] bytes = new byte[] {1,2,3,4,5,6,7,8,9,10};
 292
 293         snapshotManager.persist(mockDataPersistenceProvider, bytes, mockRaftActorBehavior);
 294
 295         verify(mockDataPersistenceProvider).saveSnapshot(any(Snapshot.class));
 296
 297         verify(mockReplicatedLog).snapshotPreCommit(9L, 6L);
 298
 299         ArgumentCaptor<SendInstallSnapshot> sendInstallSnapshotArgumentCaptor
 300                 = ArgumentCaptor.forClass(SendInstallSnapshot.class);
 301
 302         verify(mockRaftActorBehavior).handleMessage(any(ActorRef.class), sendInstallSnapshotArgumentCaptor.capture());
 303
 304         SendInstallSnapshot sendInstallSnapshot = sendInstallSnapshotArgumentCaptor.getValue();
 305
 306         assertTrue(Arrays.equals(bytes, sendInstallSnapshot.getSnapshot().toByteArray()));
 307     }
 308
 309     @Test
 310     public void testCallingPersistWithoutCaptureWillDoNothing(){
 311         snapshotManager.persist(mockDataPersistenceProvider, new byte[]{}, mockRaftActorBehavior);
 312
 313         verify(mockDataPersistenceProvider, never()).saveSnapshot(any(Snapshot.class));
 314
 315         verify(mockReplicatedLog, never()).snapshotPreCommit(9L, 6L);
 316
 317         verify(mockRaftActorBehavior, never()).handleMessage(any(ActorRef.class), any(SendInstallSnapshot.class));
 318     }
 319     @Test
 320     public void testCallingPersistTwiceWillDoNoHarm(){
 321         doReturn(Integer.MAX_VALUE).when(mockReplicatedLog).dataSize();
 322
 323         // when replicatedToAllIndex = -1
 324         snapshotManager.captureToInstall(new MockRaftActorContext.MockReplicatedLogEntry(6, 9,
 325                 new MockRaftActorContext.MockPayload()), -1);
 326
 327         snapshotManager.create(mockProcedure);
 328
 329         snapshotManager.persist(mockDataPersistenceProvider, new byte[]{}, mockRaftActorBehavior);
 330
 331         snapshotManager.persist(mockDataPersistenceProvider, new byte[]{}, mockRaftActorBehavior);
 332
 333         verify(mockDataPersistenceProvider).saveSnapshot(any(Snapshot.class));
 334
 335         verify(mockReplicatedLog).snapshotPreCommit(9L, 6L);
 336
 337         verify(mockRaftActorBehavior).handleMessage(any(ActorRef.class), any(SendInstallSnapshot.class));
 338     }
 339
 340     @Test
 341     public void testCommit(){
 342         // when replicatedToAllIndex = -1
 343         snapshotManager.captureToInstall(new MockRaftActorContext.MockReplicatedLogEntry(6, 9,
 344                 new MockRaftActorContext.MockPayload()), -1);
 345
 346         snapshotManager.create(mockProcedure);
 347
 348         snapshotManager.persist(mockDataPersistenceProvider, new byte[]{}, mockRaftActorBehavior);
 349
 350         snapshotManager.commit(mockDataPersistenceProvider, 100L);
 351
 352         verify(mockReplicatedLog).snapshotCommit();
 353
 354         verify(mockDataPersistenceProvider).deleteMessages(100L);
 355
 356         ArgumentCaptor<SnapshotSelectionCriteria> criteriaCaptor = ArgumentCaptor.forClass(SnapshotSelectionCriteria.class);
 357
 358         verify(mockDataPersistenceProvider).deleteSnapshots(criteriaCaptor.capture());
 359
 360         assertEquals(90, criteriaCaptor.getValue().maxSequenceNr()); // sequenceNumber = 100
 361                                                                      // config snapShotBatchCount = 10
 362                                                                      // therefore maxSequenceNumber = 90
 363     }
 364
 365     @Test
 366     public void testCommitBeforePersist(){
 367         // when replicatedToAllIndex = -1
 368         snapshotManager.captureToInstall(new MockRaftActorContext.MockReplicatedLogEntry(6, 9,
 369                 new MockRaftActorContext.MockPayload()), -1);
 370
 371         snapshotManager.commit(mockDataPersistenceProvider, 100L);
 372
 373         verify(mockReplicatedLog, never()).snapshotCommit();
 374
 375         verify(mockDataPersistenceProvider, never()).deleteMessages(100L);
 376
 377         verify(mockDataPersistenceProvider, never()).deleteSnapshots(any(SnapshotSelectionCriteria.class));
 378
 379     }
 380
 381     @Test
 382     public void testCommitBeforeCapture(){
 383         snapshotManager.commit(mockDataPersistenceProvider, 100L);
 384
 385         verify(mockReplicatedLog, never()).snapshotCommit();
 386
 387         verify(mockDataPersistenceProvider, never()).deleteMessages(anyLong());
 388
 389         verify(mockDataPersistenceProvider, never()).deleteSnapshots(any(SnapshotSelectionCriteria.class));
 390
 391     }
 392
 393     @Test
 394     public void testCallingCommitMultipleTimesCausesNoHarm(){
 395         // when replicatedToAllIndex = -1
 396         snapshotManager.captureToInstall(new MockRaftActorContext.MockReplicatedLogEntry(6, 9,
 397                 new MockRaftActorContext.MockPayload()), -1);
 398
 399         snapshotManager.create(mockProcedure);
 400
 401         snapshotManager.persist(mockDataPersistenceProvider, new byte[]{}, mockRaftActorBehavior);
 402
 403         snapshotManager.commit(mockDataPersistenceProvider, 100L);
 404
 405         snapshotManager.commit(mockDataPersistenceProvider, 100L);
 406
 407         verify(mockReplicatedLog, times(1)).snapshotCommit();
 408
 409         verify(mockDataPersistenceProvider, times(1)).deleteMessages(100L);
 410
 411         verify(mockDataPersistenceProvider, times(1)).deleteSnapshots(any(SnapshotSelectionCriteria.class));
 412     }
 413
 414     @Test
 415     public void testRollback(){
 416         // when replicatedToAllIndex = -1
 417         snapshotManager.captureToInstall(new MockRaftActorContext.MockReplicatedLogEntry(6, 9,
 418                 new MockRaftActorContext.MockPayload()), -1);
 419
 420         snapshotManager.create(mockProcedure);
 421
 422         snapshotManager.persist(mockDataPersistenceProvider, new byte[]{}, mockRaftActorBehavior);
 423
 424         snapshotManager.rollback();
 425
 426         verify(mockReplicatedLog).snapshotRollback();
 427     }
 428
 429
 430     @Test
 431     public void testRollbackBeforePersist(){
 432         // when replicatedToAllIndex = -1
 433         snapshotManager.captureToInstall(new MockRaftActorContext.MockReplicatedLogEntry(6, 9,
 434                 new MockRaftActorContext.MockPayload()), -1);
 435
 436         snapshotManager.rollback();
 437
 438         verify(mockReplicatedLog, never()).snapshotRollback();
 439     }
 440
 441     @Test
 442     public void testRollbackBeforeCapture(){
 443         snapshotManager.rollback();
 444
 445         verify(mockReplicatedLog, never()).snapshotRollback();
 446     }
 447
 448     @Test
 449     public void testCallingRollbackMultipleTimesCausesNoHarm(){
 450         // when replicatedToAllIndex = -1
 451         snapshotManager.captureToInstall(new MockRaftActorContext.MockReplicatedLogEntry(6, 9,
 452                 new MockRaftActorContext.MockPayload()), -1);
 453
 454         snapshotManager.create(mockProcedure);
 455
 456         snapshotManager.persist(mockDataPersistenceProvider, new byte[]{}, mockRaftActorBehavior);
 457
 458         snapshotManager.rollback();
 459
 460         snapshotManager.rollback();
 461
 462         verify(mockReplicatedLog, times(1)).snapshotRollback();
 463     }
 464
 465     @Test
 466     public void testTrimLog(){
 467         ElectionTerm mockElectionTerm = mock(ElectionTerm.class);
 468         ReplicatedLogEntry replicatedLogEntry = mock(ReplicatedLogEntry.class);
 469         doReturn(20L).when(mockRaftActorContext).getLastApplied();
 470         doReturn(true).when(mockReplicatedLog).isPresent(10);
 471         doReturn(mockElectionTerm).when(mockRaftActorContext).getTermInformation();
 472         doReturn(5L).when(mockElectionTerm).getCurrentTerm();
 473         doReturn(replicatedLogEntry).when((mockReplicatedLog)).get(10);
 474         doReturn(5L).when(replicatedLogEntry).getTerm();
 475
 476         snapshotManager.trimLog(10);
 477
 478         verify(mockReplicatedLog).snapshotPreCommit(10, 5);
 479         verify(mockReplicatedLog).snapshotCommit();
 480     }
 481
 482     @Test
 483     public void testTrimLogAfterCapture(){
 484         snapshotManager.capture(new MockRaftActorContext.MockReplicatedLogEntry(1,9,
 485                 new MockRaftActorContext.MockPayload()), 9);
 486
 487         assertEquals(true, snapshotManager.isCapturing());
 488
 489         ElectionTerm mockElectionTerm = mock(ElectionTerm.class);
 490         ReplicatedLogEntry replicatedLogEntry = mock(ReplicatedLogEntry.class);
 491         doReturn(20L).when(mockRaftActorContext).getLastApplied();
 492         doReturn(true).when(mockReplicatedLog).isPresent(10);
 493         doReturn(mockElectionTerm).when(mockRaftActorContext).getTermInformation();
 494         doReturn(5L).when(mockElectionTerm).getCurrentTerm();
 495         doReturn(replicatedLogEntry).when((mockReplicatedLog)).get(10);
 496         doReturn(5L).when(replicatedLogEntry).getTerm();
 497
 498         snapshotManager.trimLog(10);
 499
 500         verify(mockReplicatedLog, never()).snapshotPreCommit(anyLong(), anyLong());
 501         verify(mockReplicatedLog, never()).snapshotCommit();
 502
 503     }
 504
 505     @Test
 506     public void testTrimLogAfterCaptureToInstall(){
 507         snapshotManager.captureToInstall(new MockRaftActorContext.MockReplicatedLogEntry(1,9,
 508                 new MockRaftActorContext.MockPayload()), 9);
 509
 510         assertEquals(true, snapshotManager.isCapturing());
 511
 512         ElectionTerm mockElectionTerm = mock(ElectionTerm.class);
 513         ReplicatedLogEntry replicatedLogEntry = mock(ReplicatedLogEntry.class);
 514         doReturn(20L).when(mockRaftActorContext).getLastApplied();
 515         doReturn(true).when(mockReplicatedLog).isPresent(10);
 516         doReturn(mockElectionTerm).when(mockRaftActorContext).getTermInformation();
 517         doReturn(5L).when(mockElectionTerm).getCurrentTerm();
 518         doReturn(replicatedLogEntry).when((mockReplicatedLog)).get(10);
 519         doReturn(5L).when(replicatedLogEntry).getTerm();
 520
 521         snapshotManager.trimLog(10);
 522
 523         verify(mockReplicatedLog, never()).snapshotPreCommit(10, 5);
 524         verify(mockReplicatedLog, never()).snapshotCommit();
 525
 526     }
 527
 528 }