opendaylight/md-sal/sal-clustering-commons/src/test/java/org/opendaylight/controller/cluster/persistence/LocalSnapshotStoreTest.java

   1 /*
   2  * Copyright (c) 2017 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.persistence;
   9
  10 import static java.lang.Boolean.FALSE;
  11 import static java.lang.Boolean.TRUE;
  12 import static org.junit.Assert.assertEquals;
  13 import static org.opendaylight.controller.cluster.persistence.LocalSnapshotStoreSpecTest.SNAPSHOT_DIR;
  14 import static org.opendaylight.controller.cluster.persistence.LocalSnapshotStoreSpecTest.cleanSnapshotDir;
  15 import static org.opendaylight.controller.cluster.persistence.LocalSnapshotStoreSpecTest.createSnapshotDir;
  16
  17 import akka.actor.ActorRef;
  18 import akka.actor.ActorSystem;
  19 import akka.actor.ExtendedActorSystem;
  20 import akka.persistence.Persistence;
  21 import akka.persistence.SelectedSnapshot;
  22 import akka.persistence.SnapshotMetadata;
  23 import akka.persistence.SnapshotProtocol;
  24 import akka.persistence.SnapshotProtocol.LoadSnapshot;
  25 import akka.persistence.SnapshotProtocol.LoadSnapshotFailed;
  26 import akka.persistence.SnapshotProtocol.LoadSnapshotResult;
  27 import akka.persistence.SnapshotSelectionCriteria;
  28 import akka.persistence.serialization.Snapshot;
  29 import akka.persistence.serialization.SnapshotSerializer;
  30 import akka.testkit.JavaTestKit;
  31 import com.typesafe.config.ConfigFactory;
  32 import java.io.File;
  33 import java.io.FileOutputStream;
  34 import java.io.IOException;
  35 import java.io.UnsupportedEncodingException;
  36 import java.net.URLEncoder;
  37 import java.nio.charset.StandardCharsets;
  38 import org.apache.commons.io.FileUtils;
  39 import org.apache.commons.lang.SerializationUtils;
  40 import org.junit.After;
  41 import org.junit.AfterClass;
  42 import org.junit.Before;
  43 import org.junit.BeforeClass;
  44 import org.junit.Test;
  45 import org.mockito.MockitoAnnotations;
  46 import scala.Option;
  47
  48 /**
  49  * Unit tests for LocalSnapshotStore. These are in addition to LocalSnapshotStoreSpecTest to cover a few cases
  50  * that SnapshotStoreSpec doesn't.
  51  *
  52  * @author Thomas Pantelis
  53  */
  54 public class LocalSnapshotStoreTest {
  55     private static final String PERSISTENCE_ID = "member-1-shard-default-config";
  56     private static final String PREFIX_BASED_SHARD_PERSISTENCE_ID = "member-1-shard-id-ints!-config";
  57
  58     private static ActorSystem system;
  59     private static ActorRef snapshotStore;
  60
  61     @BeforeClass
  62     public static void staticSetup() {
  63         createSnapshotDir();
  64
  65         system = ActorSystem.create("test", ConfigFactory.load("LocalSnapshotStoreTest.conf"));
  66         snapshotStore = system.registerExtension(Persistence.lookup()).snapshotStoreFor(null);
  67     }
  68
  69     @AfterClass
  70     public static void staticCleanup() {
  71         FileUtils.deleteQuietly(SNAPSHOT_DIR);
  72         JavaTestKit.shutdownActorSystem(system);
  73     }
  74
  75     @Before
  76     public void setup() {
  77         MockitoAnnotations.initMocks(this);
  78         cleanSnapshotDir();
  79     }
  80
  81     @After
  82     public void cleanup() {
  83         cleanSnapshotDir();
  84     }
  85
  86     @Test
  87     public void testDoLoadAsync() throws IOException {
  88         createSnapshotFile(PERSISTENCE_ID, "one", 0, 1000);
  89         createSnapshotFile(PERSISTENCE_ID, "two", 1, 2000);
  90         createSnapshotFile(PERSISTENCE_ID, "three", 1, 3000);
  91
  92         createSnapshotFile(PREFIX_BASED_SHARD_PERSISTENCE_ID, "foo", 0, 1000);
  93         createSnapshotFile(PREFIX_BASED_SHARD_PERSISTENCE_ID, "bar", 1, 2000);
  94         createSnapshotFile(PREFIX_BASED_SHARD_PERSISTENCE_ID, "foobar", 1, 3000);
  95
  96         createSnapshotFile("member-1-shard-default-oper", "foo", 0, 1000);
  97         createSnapshotFile("member-1-shard-toaster-oper", "foo", 0, 1000);
  98         new File(SNAPSHOT_DIR, "other").createNewFile();
  99         new File(SNAPSHOT_DIR, "other-1485349217290").createNewFile();
 100
 101         SnapshotMetadata metadata3 = new SnapshotMetadata(PERSISTENCE_ID, 1, 3000);
 102
 103         JavaTestKit probe = new JavaTestKit(system);
 104         snapshotStore.tell(new LoadSnapshot(PERSISTENCE_ID,
 105                 SnapshotSelectionCriteria.latest(), Long.MAX_VALUE), probe.getRef());
 106         LoadSnapshotResult result = probe.expectMsgClass(LoadSnapshotResult.class);
 107         Option<SelectedSnapshot> possibleSnapshot = result.snapshot();
 108
 109         assertEquals("SelectedSnapshot present", TRUE, possibleSnapshot.nonEmpty());
 110         assertEquals("SelectedSnapshot metadata", metadata3, possibleSnapshot.get().metadata());
 111         assertEquals("SelectedSnapshot snapshot", "three", possibleSnapshot.get().snapshot());
 112
 113         snapshotStore.tell(new LoadSnapshot(PREFIX_BASED_SHARD_PERSISTENCE_ID,
 114                 SnapshotSelectionCriteria.latest(), Long.MAX_VALUE), probe.getRef());
 115         result = probe.expectMsgClass(LoadSnapshotResult.class);
 116         possibleSnapshot = result.snapshot();
 117
 118         SnapshotMetadata prefixBasedShardMetada3 = new SnapshotMetadata(PREFIX_BASED_SHARD_PERSISTENCE_ID, 1, 3000);
 119
 120         assertEquals("SelectedSnapshot present", TRUE, possibleSnapshot.nonEmpty());
 121         assertEquals("SelectedSnapshot metadata", prefixBasedShardMetada3, possibleSnapshot.get().metadata());
 122         assertEquals("SelectedSnapshot snapshot", "foobar", possibleSnapshot.get().snapshot());
 123     }
 124
 125     @Test
 126     public void testDoLoadAsyncWithNoSnapshots() throws IOException {
 127         JavaTestKit probe = new JavaTestKit(system);
 128         snapshotStore.tell(new LoadSnapshot(PERSISTENCE_ID,
 129                 SnapshotSelectionCriteria.latest(), Long.MAX_VALUE), probe.getRef());
 130         LoadSnapshotResult result = probe.expectMsgClass(LoadSnapshotResult.class);
 131         Option<SelectedSnapshot> possibleSnapshot = result.snapshot();
 132
 133         assertEquals("SelectedSnapshot present", FALSE, possibleSnapshot.nonEmpty());
 134     }
 135
 136     @Test
 137     public void testDoLoadAsyncWithRetry() throws IOException  {
 138         createSnapshotFile(PERSISTENCE_ID, "one", 0, 1000);
 139         createSnapshotFile(PERSISTENCE_ID, null, 1, 2000);
 140
 141         SnapshotMetadata metadata = new SnapshotMetadata(PERSISTENCE_ID, 0, 1000);
 142
 143         JavaTestKit probe = new JavaTestKit(system);
 144         snapshotStore.tell(new LoadSnapshot(PERSISTENCE_ID,
 145                 SnapshotSelectionCriteria.latest(), Long.MAX_VALUE), probe.getRef());
 146         LoadSnapshotResult result = probe.expectMsgClass(LoadSnapshotResult.class);
 147         Option<SelectedSnapshot> possibleSnapshot = result.snapshot();
 148
 149         assertEquals("SelectedSnapshot present", TRUE, possibleSnapshot.nonEmpty());
 150         assertEquals("SelectedSnapshot metadata", metadata, possibleSnapshot.get().metadata());
 151         assertEquals("SelectedSnapshot snapshot", "one", possibleSnapshot.get().snapshot());
 152     }
 153
 154     @SuppressWarnings("checkstyle:illegalThrows")
 155     @Test(expected = IOException.class)
 156     public void testDoLoadAsyncWithFailure() throws Throwable {
 157         createSnapshotFile(PERSISTENCE_ID, null, 1, 2000);
 158
 159         JavaTestKit probe = new JavaTestKit(system);
 160         snapshotStore.tell(new SnapshotProtocol.LoadSnapshot(PERSISTENCE_ID,
 161                 SnapshotSelectionCriteria.latest(), Long.MAX_VALUE), probe.getRef());
 162         LoadSnapshotFailed failed = probe.expectMsgClass(LoadSnapshotFailed.class);
 163         throw failed.cause();
 164     }
 165
 166     @Test
 167     public void testDoLoadAsyncWithAkkaSerializedSnapshot() throws IOException {
 168         SnapshotSerializer snapshotSerializer = new SnapshotSerializer((ExtendedActorSystem) system);
 169
 170         String name = toSnapshotName(PERSISTENCE_ID, 1, 1000);
 171         try (FileOutputStream fos = new FileOutputStream(new File(SNAPSHOT_DIR, name))) {
 172             fos.write(snapshotSerializer.toBinary(new Snapshot("one")));
 173         }
 174
 175         SnapshotMetadata metadata = new SnapshotMetadata(PERSISTENCE_ID, 1, 1000);
 176
 177         JavaTestKit probe = new JavaTestKit(system);
 178         snapshotStore.tell(new LoadSnapshot(PERSISTENCE_ID,
 179                 SnapshotSelectionCriteria.latest(), Long.MAX_VALUE), probe.getRef());
 180         LoadSnapshotResult result = probe.expectMsgClass(LoadSnapshotResult.class);
 181         Option<SelectedSnapshot> possibleSnapshot = result.snapshot();
 182
 183         assertEquals("SelectedSnapshot present", TRUE, possibleSnapshot.nonEmpty());
 184         assertEquals("SelectedSnapshot metadata", metadata, possibleSnapshot.get().metadata());
 185         assertEquals("SelectedSnapshot snapshot", "one", possibleSnapshot.get().snapshot());
 186     }
 187
 188     private static void createSnapshotFile(final String persistenceId, final String payload, final int seqNr,
 189             final int timestamp) throws IOException {
 190         String name = toSnapshotName(persistenceId, seqNr, timestamp);
 191         try (FileOutputStream fos = new FileOutputStream(new File(SNAPSHOT_DIR, name))) {
 192             if (payload != null) {
 193                 fos.write(SerializationUtils.serialize(payload));
 194             }
 195         }
 196     }
 197
 198     private static String toSnapshotName(final String persistenceId, final int seqNr, final int timestamp)
 199             throws UnsupportedEncodingException {
 200         final String encodedPersistenceId = URLEncoder.encode(persistenceId, StandardCharsets.UTF_8.name());
 201         return "snapshot-" + encodedPersistenceId + "-" + seqNr + "-" + timestamp;
 202     }
 203 }