X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Ftest%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2Fadmin%2FClusterAdminRpcServiceTest.java;h=9ac5c7af61d635a71cf040094efb3605760e0cc0;hb=9412c00daf7c8c3f108d1284a6288753ba67f6ac;hp=765d8a1033e59c75e500b3041cb55d311995b709;hpb=0b5ba62b192d1f1e1c5be6e884f2bacb40a5c2c3;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/admin/ClusterAdminRpcServiceTest.java b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/admin/ClusterAdminRpcServiceTest.java index 765d8a1033..9ac5c7af61 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/admin/ClusterAdminRpcServiceTest.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/admin/ClusterAdminRpcServiceTest.java @@ -7,13 +7,19 @@ */ package org.opendaylight.controller.cluster.datastore.admin; +import static org.hamcrest.CoreMatchers.anyOf; +import static org.hamcrest.CoreMatchers.containsString; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertThat; import static org.junit.Assert.fail; +import static org.opendaylight.controller.cluster.datastore.MemberNode.verifyNoShardPresent; import static org.opendaylight.controller.cluster.datastore.MemberNode.verifyRaftPeersPresent; import static org.opendaylight.controller.cluster.datastore.MemberNode.verifyRaftState; import akka.actor.ActorRef; import akka.actor.PoisonPill; +import akka.actor.Status.Success; import akka.cluster.Cluster; import com.google.common.base.Optional; import com.google.common.collect.ImmutableMap; @@ -21,18 +27,26 @@ import com.google.common.collect.Iterables; import com.google.common.collect.Sets; import java.io.File; import java.io.FileInputStream; +import java.net.URI; import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashMap; import java.util.HashSet; import java.util.List; +import java.util.Map; import java.util.Set; import java.util.concurrent.TimeUnit; import org.apache.commons.lang3.SerializationUtils; import org.junit.After; import org.junit.Before; import org.junit.Test; +import org.opendaylight.controller.cluster.datastore.DatastoreContext; import org.opendaylight.controller.cluster.datastore.DistributedDataStore; import org.opendaylight.controller.cluster.datastore.MemberNode; import org.opendaylight.controller.cluster.datastore.MemberNode.RaftStateVerifier; +import org.opendaylight.controller.cluster.datastore.Shard; +import org.opendaylight.controller.cluster.datastore.config.ModuleShardConfiguration; +import org.opendaylight.controller.cluster.datastore.messages.CreateShard; import org.opendaylight.controller.cluster.datastore.messages.DatastoreSnapshot; import org.opendaylight.controller.cluster.raft.client.messages.OnDemandRaftState; import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal; @@ -40,9 +54,15 @@ import org.opendaylight.controller.cluster.raft.utils.InMemorySnapshotStore; import org.opendaylight.controller.md.cluster.datastore.model.CarsModel; import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort; import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction; +import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddReplicasForAllShardsOutput; import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddShardReplicaInputBuilder; import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.BackupDatastoreInputBuilder; import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.DataStoreType; +import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveAllShardReplicasInputBuilder; +import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveAllShardReplicasOutput; +import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveShardReplicaInputBuilder; +import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.shard.result.output.ShardResult; +import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.shard.result.output.ShardResultBuilder; import org.opendaylight.yangtools.yang.common.RpcError; import org.opendaylight.yangtools.yang.common.RpcResult; import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode; @@ -89,8 +109,8 @@ public class ClusterAdminRpcServiceTest { ImmutableMap map = ImmutableMap.of(snapshots.get(0).getType(), snapshots.get(0), snapshots.get(1).getType(), snapshots.get(1)); - verifyDatastoreSnapshot(node.configDataStore().getActorContext().getDataStoreType(), - map.get(node.configDataStore().getActorContext().getDataStoreType()), "cars", "people"); + verifyDatastoreSnapshot(node.configDataStore().getActorContext().getDataStoreName(), + map.get(node.configDataStore().getActorContext().getDataStoreName()), "cars", "people"); } finally { new File(fileName).delete(); } @@ -260,7 +280,7 @@ public class ClusterAdminRpcServiceTest { service.close(); } - private void verifySuccessfulRpcResult(RpcResult rpcResult) { + private T verifySuccessfulRpcResult(RpcResult rpcResult) { if(!rpcResult.isSuccessful()) { if(rpcResult.getErrors().size() > 0) { RpcError error = Iterables.getFirst(rpcResult.getErrors(), null); @@ -269,6 +289,8 @@ public class ClusterAdminRpcServiceTest { fail("Rpc failed with no error"); } + + return rpcResult.getResult(); } private void verifyFailedRpcResult(RpcResult rpcResult) { @@ -279,18 +301,223 @@ public class ClusterAdminRpcServiceTest { } @Test - public void testRemoveShardReplica() { - // TODO implement + public void testRemoveShardReplica() throws Exception { + String name = "testRemoveShardReplicaLocal"; + String moduleShardsConfig = "module-shards-member1-and-2-and-3.conf"; + MemberNode leaderNode1 = MemberNode.builder(memberNodes).akkaConfig("Member1").testName(name ). + moduleShardsConfig(moduleShardsConfig). + datastoreContextBuilder(DatastoreContext.newBuilder(). + shardHeartbeatIntervalInMillis(300).shardElectionTimeoutFactor(1)).build(); + + MemberNode replicaNode2 = MemberNode.builder(memberNodes).akkaConfig("Member2").testName(name). + moduleShardsConfig(moduleShardsConfig).build(); + + MemberNode replicaNode3 = MemberNode.builder(memberNodes).akkaConfig("Member3").testName(name). + moduleShardsConfig(moduleShardsConfig).build(); + + leaderNode1.configDataStore().waitTillReady(); + verifyRaftPeersPresent(leaderNode1.configDataStore(), "cars", "member-2", "member-3"); + verifyRaftPeersPresent(replicaNode2.configDataStore(), "cars", "member-1", "member-3"); + verifyRaftPeersPresent(replicaNode3.configDataStore(), "cars", "member-1", "member-2"); + + // Invoke RPC service on member-3 to remove it's local shard + + ClusterAdminRpcService service3 = new ClusterAdminRpcService(replicaNode3.configDataStore(), + replicaNode3.operDataStore()); + + RpcResult rpcResult = service3.removeShardReplica(new RemoveShardReplicaInputBuilder(). + setShardName("cars").setMemberName("member-3").setDataStoreType(DataStoreType.Config).build()). + get(10, TimeUnit.SECONDS); + verifySuccessfulRpcResult(rpcResult); + service3.close(); + + verifyRaftPeersPresent(leaderNode1.configDataStore(), "cars", "member-2"); + verifyRaftPeersPresent(replicaNode2.configDataStore(), "cars", "member-1"); + verifyNoShardPresent(replicaNode3.configDataStore(), "cars"); + + // Restart member-2 and verify member-3 isn't present. + + Cluster.get(leaderNode1.kit().getSystem()).down(Cluster.get(replicaNode2.kit().getSystem()).selfAddress()); + replicaNode2.cleanup(); + + replicaNode2 = MemberNode.builder(memberNodes).akkaConfig("Member2").testName(name). + moduleShardsConfig(moduleShardsConfig).build(); + + verifyRaftPeersPresent(replicaNode2.configDataStore(), "cars", "member-1"); + + // Invoke RPC service on member-1 to remove member-2 + + ClusterAdminRpcService service1 = new ClusterAdminRpcService(leaderNode1.configDataStore(), + leaderNode1.operDataStore()); + + rpcResult = service1.removeShardReplica(new RemoveShardReplicaInputBuilder(). + setShardName("cars").setMemberName("member-2").setDataStoreType(DataStoreType.Config).build()). + get(10, TimeUnit.SECONDS); + verifySuccessfulRpcResult(rpcResult); + service1.close(); + + verifyRaftPeersPresent(leaderNode1.configDataStore(), "cars"); + verifyNoShardPresent(replicaNode2.configDataStore(), "cars"); } @Test - public void testAddReplicasForAllShards() { - // TODO implement + public void testRemoveShardLeaderReplica() throws Exception { + String name = "testRemoveShardLeaderReplica"; + String moduleShardsConfig = "module-shards-member1-and-2-and-3.conf"; + MemberNode leaderNode1 = MemberNode.builder(memberNodes).akkaConfig("Member1").testName(name ). + moduleShardsConfig(moduleShardsConfig). + datastoreContextBuilder(DatastoreContext.newBuilder(). + shardHeartbeatIntervalInMillis(300).shardElectionTimeoutFactor(1)).build(); + + MemberNode replicaNode2 = MemberNode.builder(memberNodes).akkaConfig("Member2").testName(name). + moduleShardsConfig(moduleShardsConfig).build(); + + MemberNode replicaNode3 = MemberNode.builder(memberNodes).akkaConfig("Member3").testName(name). + moduleShardsConfig(moduleShardsConfig).build(); + + leaderNode1.configDataStore().waitTillReady(); + verifyRaftPeersPresent(leaderNode1.configDataStore(), "cars", "member-2", "member-3"); + verifyRaftPeersPresent(replicaNode2.configDataStore(), "cars", "member-1", "member-3"); + verifyRaftPeersPresent(replicaNode3.configDataStore(), "cars", "member-1", "member-2"); + + replicaNode2.waitForMembersUp("member-1", "member-3"); + replicaNode2.waitForMembersUp("member-1", "member-2"); + + // Invoke RPC service on leader member-1 to remove it's local shard + + ClusterAdminRpcService service1 = new ClusterAdminRpcService(leaderNode1.configDataStore(), + leaderNode1.operDataStore()); + + RpcResult rpcResult = service1.removeShardReplica(new RemoveShardReplicaInputBuilder(). + setShardName("cars").setMemberName("member-1").setDataStoreType(DataStoreType.Config).build()). + get(10, TimeUnit.SECONDS); + verifySuccessfulRpcResult(rpcResult); + service1.close(); + + verifyRaftState(replicaNode2.configDataStore(), "cars", new RaftStateVerifier() { + @Override + public void verify(OnDemandRaftState raftState) { + assertThat("Leader Id", raftState.getLeader(), anyOf(containsString("member-2"), + containsString("member-3"))); + } + }); + + verifyRaftPeersPresent(replicaNode2.configDataStore(), "cars", "member-3"); + verifyRaftPeersPresent(replicaNode3.configDataStore(), "cars", "member-2"); + verifyNoShardPresent(leaderNode1.configDataStore(), "cars"); } @Test - public void testRemoveAllShardReplicas() { - // TODO implement + public void testAddReplicasForAllShards() throws Exception { + String name = "testAddReplicasForAllShards"; + String moduleShardsConfig = "module-shards-member1.conf"; + MemberNode leaderNode1 = MemberNode.builder(memberNodes).akkaConfig("Member1").testName(name ). + moduleShardsConfig(moduleShardsConfig).waitForShardLeader("cars", "people").build(); + + ModuleShardConfiguration petsModuleConfig = new ModuleShardConfiguration(URI.create("pets-ns"), "pets-module", + "pets", null, Arrays.asList("member-1")); + leaderNode1.configDataStore().getActorContext().getShardManager().tell( + new CreateShard(petsModuleConfig, Shard.builder(), null), leaderNode1.kit().getRef()); + leaderNode1.kit().expectMsgClass(Success.class); + leaderNode1.kit().waitUntilLeader(leaderNode1.configDataStore().getActorContext(), "pets"); + + MemberNode newReplicaNode2 = MemberNode.builder(memberNodes).akkaConfig("Member2").testName(name). + moduleShardsConfig(moduleShardsConfig).build(); + + leaderNode1.waitForMembersUp("member-2"); + newReplicaNode2.waitForMembersUp("member-1"); + + newReplicaNode2.configDataStore().getActorContext().getShardManager().tell( + new CreateShard(petsModuleConfig, Shard.builder(), null), newReplicaNode2.kit().getRef()); + newReplicaNode2.kit().expectMsgClass(Success.class); + + newReplicaNode2.operDataStore().getActorContext().getShardManager().tell( + new CreateShard(new ModuleShardConfiguration(URI.create("no-leader-ns"), "no-leader-module", + "no-leader", null, Arrays.asList("member-1")), Shard.builder(), null), + newReplicaNode2.kit().getRef()); + newReplicaNode2.kit().expectMsgClass(Success.class); + + ClusterAdminRpcService service = new ClusterAdminRpcService(newReplicaNode2.configDataStore(), + newReplicaNode2.operDataStore()); + + RpcResult rpcResult = service.addReplicasForAllShards().get(10, TimeUnit.SECONDS); + AddReplicasForAllShardsOutput result = verifySuccessfulRpcResult(rpcResult); + verifyShardResults(result.getShardResult(), successShardResult("cars", DataStoreType.Config), + successShardResult("people", DataStoreType.Config), + successShardResult("pets", DataStoreType.Config), + successShardResult("cars", DataStoreType.Operational), + successShardResult("people", DataStoreType.Operational), + failedShardResult("no-leader", DataStoreType.Operational)); + + verifyRaftPeersPresent(newReplicaNode2.configDataStore(), "cars", "member-1"); + verifyRaftPeersPresent(newReplicaNode2.configDataStore(), "people", "member-1"); + verifyRaftPeersPresent(newReplicaNode2.configDataStore(), "pets", "member-1"); + verifyRaftPeersPresent(newReplicaNode2.operDataStore(), "cars", "member-1"); + verifyRaftPeersPresent(newReplicaNode2.operDataStore(), "people", "member-1"); + + service.close(); + } + + @Test + public void testRemoveAllShardReplicas() throws Exception { + String name = "testRemoveAllShardReplicas"; + String moduleShardsConfig = "module-shards-member1-and-2-and-3.conf"; + MemberNode leaderNode1 = MemberNode.builder(memberNodes).akkaConfig("Member1").testName(name ). + moduleShardsConfig(moduleShardsConfig).datastoreContextBuilder(DatastoreContext.newBuilder(). + shardHeartbeatIntervalInMillis(300).shardElectionTimeoutFactor(1)).build(); + + MemberNode replicaNode2 = MemberNode.builder(memberNodes).akkaConfig("Member2").testName(name). + moduleShardsConfig(moduleShardsConfig).build(); + + MemberNode replicaNode3 = MemberNode.builder(memberNodes).akkaConfig("Member3").testName(name). + moduleShardsConfig(moduleShardsConfig).build(); + + leaderNode1.configDataStore().waitTillReady(); + verifyRaftPeersPresent(leaderNode1.configDataStore(), "cars", "member-2", "member-3"); + verifyRaftPeersPresent(replicaNode2.configDataStore(), "cars", "member-1", "member-3"); + verifyRaftPeersPresent(replicaNode3.configDataStore(), "cars", "member-1", "member-2"); + + ModuleShardConfiguration petsModuleConfig = new ModuleShardConfiguration(URI.create("pets-ns"), "pets-module", + "pets", null, Arrays.asList("member-1", "member-2", "member-3")); + leaderNode1.configDataStore().getActorContext().getShardManager().tell( + new CreateShard(petsModuleConfig, Shard.builder(), null), leaderNode1.kit().getRef()); + leaderNode1.kit().expectMsgClass(Success.class); + + replicaNode2.configDataStore().getActorContext().getShardManager().tell( + new CreateShard(petsModuleConfig, Shard.builder(), null), replicaNode2.kit().getRef()); + replicaNode2.kit().expectMsgClass(Success.class); + + replicaNode3.configDataStore().getActorContext().getShardManager().tell( + new CreateShard(petsModuleConfig, Shard.builder(), null), replicaNode3.kit().getRef()); + replicaNode3.kit().expectMsgClass(Success.class); + + verifyRaftPeersPresent(leaderNode1.configDataStore(), "pets", "member-2", "member-3"); + verifyRaftPeersPresent(replicaNode2.configDataStore(), "pets", "member-1", "member-3"); + verifyRaftPeersPresent(replicaNode3.configDataStore(), "pets", "member-1", "member-2"); + + ClusterAdminRpcService service3 = new ClusterAdminRpcService(replicaNode3.configDataStore(), + replicaNode3.operDataStore()); + + RpcResult rpcResult = service3.removeAllShardReplicas( + new RemoveAllShardReplicasInputBuilder().setMemberName("member-3").build()).get(10, TimeUnit.SECONDS); + RemoveAllShardReplicasOutput result = verifySuccessfulRpcResult(rpcResult); + verifyShardResults(result.getShardResult(), successShardResult("cars", DataStoreType.Config), + successShardResult("people", DataStoreType.Config), + successShardResult("pets", DataStoreType.Config), + successShardResult("cars", DataStoreType.Operational), + successShardResult("people", DataStoreType.Operational)); + + verifyRaftPeersPresent(leaderNode1.configDataStore(), "cars", "member-2"); + verifyRaftPeersPresent(leaderNode1.configDataStore(), "people", "member-2"); + verifyRaftPeersPresent(leaderNode1.configDataStore(), "pets", "member-2"); + verifyRaftPeersPresent(replicaNode2.configDataStore(), "cars", "member-1"); + verifyRaftPeersPresent(replicaNode2.configDataStore(), "people", "member-1"); + verifyRaftPeersPresent(replicaNode2.configDataStore(), "pets", "member-1"); + verifyNoShardPresent(replicaNode3.configDataStore(), "cars"); + verifyNoShardPresent(replicaNode3.configDataStore(), "people"); + verifyNoShardPresent(replicaNode3.configDataStore(), "pets"); + + service3.close(); } @Test @@ -302,4 +529,35 @@ public class ClusterAdminRpcServiceTest { public void testConvertMembersToNonvotingForAllShards() { // TODO implement } + + private void verifyShardResults(List shardResults, ShardResult... expShardResults) { + Map expResultsMap = new HashMap<>(); + for(ShardResult r: expShardResults) { + expResultsMap.put(r.getShardName() + "-" + r.getDataStoreType(), r); + } + + for(ShardResult result: shardResults) { + ShardResult exp = expResultsMap.remove(result.getShardName() + "-" + result.getDataStoreType()); + assertNotNull(String.format("Unexpected result for shard %s, type %s", result.getShardName(), + result.getDataStoreType()), exp); + assertEquals("isSucceeded", exp.isSucceeded(), result.isSucceeded()); + if(exp.isSucceeded()) { + assertNull("Expected null error message", result.getErrorMessage()); + } else { + assertNotNull("Expected error message", result.getErrorMessage()); + } + } + + if(!expResultsMap.isEmpty()) { + fail("Missing shard results for " + expResultsMap.keySet()); + } + } + + private ShardResult successShardResult(String shardName, DataStoreType type) { + return new ShardResultBuilder().setDataStoreType(type).setShardName(shardName).setSucceeded(true).build(); + } + + private ShardResult failedShardResult(String shardName, DataStoreType type) { + return new ShardResultBuilder().setDataStoreType(type).setShardName(shardName).setSucceeded(false).build(); + } }