X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-akka-raft%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fexample%2FTestDriver.java;h=cd2e4a506ce5365bc45cc89f160695516846b4f2;hp=c2d0b3a6b7cf9a043554ee092de35fd22a8d293d;hb=0230f37066dfd974accaf36bc712d6f1e60637d0;hpb=0eb621d29daaf08979c356e2148e99c48458e169 diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/example/TestDriver.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/example/TestDriver.java index c2d0b3a6b7..cd2e4a506c 100644 --- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/example/TestDriver.java +++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/example/TestDriver.java @@ -2,39 +2,45 @@ package org.opendaylight.controller.cluster.example; import akka.actor.ActorRef; import akka.actor.ActorSystem; -import org.opendaylight.controller.cluster.example.messages.PrintRole; -import org.opendaylight.controller.cluster.example.messages.PrintState; -import org.opendaylight.controller.cluster.raft.client.messages.AddRaftPeer; -import org.opendaylight.controller.cluster.raft.client.messages.RemoveRaftPeer; - +import com.google.common.base.Optional; +import com.google.common.collect.Lists; +import com.typesafe.config.ConfigFactory; import java.io.BufferedReader; import java.io.InputStreamReader; import java.util.HashMap; +import java.util.List; import java.util.Map; -import java.util.Random; import java.util.concurrent.ConcurrentHashMap; +import org.opendaylight.controller.cluster.example.messages.PrintRole; +import org.opendaylight.controller.cluster.example.messages.PrintState; +import org.opendaylight.controller.cluster.raft.ConfigParams; /** * This is a test driver for testing akka-raft implementation * Its uses ExampleActors and threads to push content(key-vals) to these actors * Each ExampleActor can have one or more ClientActors. Each ClientActor spawns - * a thread and starts push logs to the actor its assignged to. + * a thread and starts push logs to the actor its assigned to. */ public class TestDriver { - private static final ActorSystem actorSystem = ActorSystem.create(); + private static Map allPeers = new HashMap<>(); private static Map clientActorRefs = new HashMap(); private static Map actorRefs = new HashMap(); - private static LogGenerator logGenerator = new LogGenerator();; + private static LogGenerator logGenerator = new LogGenerator(); + private int nameCounter = 0; + private static ConfigParams configParams = new ExampleConfigParamsImpl(); + + private static ActorSystem actorSystem; + private static ActorSystem listenerActorSystem; /** * Create nodes, add clients and start logging. * Commands * bye * createNodes:{num} - * addNodes:{num} * stopNode:{nodeName} + * reinstateNode:{nodeName} * addClients:{num} * addClientsToNode:{nodeName, num} * startLogging @@ -43,10 +49,22 @@ public class TestDriver { * stopLoggingForClient:{nodeName} * printNodes * printState + * + * Note: when run on IDE and on debug log level, the debug logs in + * AbstractUptypedActor and AbstractUptypedPersistentActor would need to be commented out. + * Also RaftActor handleCommand(), debug log which prints for every command other than AE/AER + * * @param args * @throws Exception */ public static void main(String[] args) throws Exception { + + actorSystem = ActorSystem.create("raft-test", ConfigFactory + .load().getConfig("raft-test")); + + listenerActorSystem = ActorSystem.create("raft-test-listener", ConfigFactory + .load().getConfig("raft-test-listener")); + TestDriver td = new TestDriver(); System.out.println("Enter command (type bye to exit):"); @@ -63,11 +81,6 @@ public class TestDriver { int n = Integer.parseInt(arr[1]); td.createNodes(n); - } else if (command.startsWith("addNodes")) { - String[] arr = command.split(":"); - int n = Integer.parseInt(arr[1]); - td.addNodes(n); - } else if (command.startsWith("addClients")) { String[] arr = command.split(":"); int n = Integer.parseInt(arr[1]); @@ -83,6 +96,10 @@ public class TestDriver { String[] arr = command.split(":"); td.stopNode(arr[1]); + } else if (command.startsWith("reinstateNode")) { + String[] arr = command.split(":"); + td.reinstateNode(arr[1]); + } else if (command.startsWith("startLogging")) { td.startAllLogging(); @@ -101,53 +118,44 @@ public class TestDriver { td.printState(); } else if (command.startsWith("printNodes")) { td.printNodes(); + } else { + System.out.println("Invalid command:" + command); } } } + // create the listener using a separate actor system for each example actor + private void createClusterRoleChangeListener(List memberIds) { + System.out.println("memberIds="+memberIds); + for (String memberId : memberIds) { + ActorRef listenerActor = listenerActorSystem.actorOf( + ExampleRoleChangeListener.getProps(memberId), memberId + "-role-change-listener"); + System.out.println("Role Change Listener created:" + listenerActor.path().toString()); + } + } + + public static ActorRef createExampleActor(String name) { + return actorSystem.actorOf(ExampleActor.props(name, withoutPeer(name), + Optional.of(configParams)), name); + } + public void createNodes(int num) { for (int i=0; i < num; i++) { - int rand = getUnusedRandom(num); - allPeers.put("example-"+rand, "akka://default/user/example-"+rand); + nameCounter = nameCounter + 1; + allPeers.put("example-"+nameCounter, "akka://raft-test/user/example-"+nameCounter); } for (String s : allPeers.keySet()) { - ActorRef exampleActor = actorSystem.actorOf( - ExampleActor.props(s, withoutPeer(s)), s); + ActorRef exampleActor = createExampleActor(s); actorRefs.put(s, exampleActor); System.out.println("Created node:"+s); } - } - - // add new nodes , pass in the count - public void addNodes(int num) { - Map newPeers = new HashMap<>(); - for (int i=0; i < num; i++) { - int rand = getUnusedRandom(num); - newPeers.put("example-"+rand, "akka://default/user/example-"+rand); - allPeers.put("example-"+rand, "akka://default/user/example-"+rand); - - } - Map newActorRefs = new HashMap(num); - for (Map.Entry entry : newPeers.entrySet()) { - ActorRef exampleActor = actorSystem.actorOf( - ExampleActor.props(entry.getKey(), withoutPeer(entry.getKey())), entry.getKey()); - newActorRefs.put(entry.getKey(), exampleActor); - - //now also add these new nodes as peers from the previous nodes - for (ActorRef actor : actorRefs.values()) { - actor.tell(new AddRaftPeer(entry.getKey(), entry.getValue()), null); - } - System.out.println("Added node:" + entry); - } - - actorRefs.putAll(newActorRefs); + createClusterRoleChangeListener(Lists.newArrayList(allPeers.keySet())); } - // add num clients to all nodes in the system public void addClients(int num) { for(Map.Entry actorRefEntry : actorRefs.entrySet()) { @@ -165,7 +173,7 @@ public class TestDriver { public void addClientsToNode(String actorName, int num) { ActorRef actorRef = actorRefs.get(actorName); for (int i=0; i < num; i++) { - String clientName = "client-" + i + "-" + actorRef; + String clientName = "client-" + i + "-" + actorName; clientActorRefs.put(clientName, actorSystem.actorOf(ClientActor.props(actorRef), clientName)); System.out.println("Added client-node:" + clientName); @@ -174,20 +182,26 @@ public class TestDriver { public void stopNode(String actorName) { ActorRef actorRef = actorRefs.get(actorName); - String clientName = "client-"+actorName; - if(clientActorRefs.containsKey(clientName)) { - actorSystem.stop(clientActorRefs.get(clientName)); - clientActorRefs.remove(clientName); + + for (Map.Entry entry : clientActorRefs.entrySet()) { + if (entry.getKey().endsWith(actorName)) { + actorSystem.stop(entry.getValue()); + } } + actorSystem.stop(actorRef); actorRefs.remove(actorName); + allPeers.remove(actorName); + } - for (ActorRef actor : actorRefs.values()) { - actor.tell(new RemoveRaftPeer(actorName), null); - } + public void reinstateNode(String actorName) { + String address = "akka://default/user/"+actorName; + allPeers.put(actorName, address); - allPeers.remove(actorName); + ActorRef exampleActor = createExampleActor(actorName); + actorRefs.put(actorName, exampleActor); + addClientsToNode(actorName, 1); } public void startAllLogging() { @@ -232,14 +246,6 @@ public class TestDriver { return null; } - private int getUnusedRandom(int num) { - int rand = -1; - do { - rand = (new Random()).nextInt(num * num); - } while (allPeers.keySet().contains("example-"+rand)); - - return rand; - } private static Map withoutPeer(String peerId) { Map without = new ConcurrentHashMap<>(allPeers);