opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/example/TestDriver.java

   1 package org.opendaylight.controller.cluster.example;
   2
   3 import akka.actor.ActorRef;
   4 import akka.actor.ActorSystem;
   5 import com.google.common.base.Optional;
   6 import com.google.common.collect.Lists;
   7 import com.typesafe.config.ConfigFactory;
   8 import java.io.BufferedReader;
   9 import java.io.InputStreamReader;
  10 import java.util.HashMap;
  11 import java.util.List;
  12 import java.util.Map;
  13 import java.util.concurrent.ConcurrentHashMap;
  14 import org.opendaylight.controller.cluster.example.messages.PrintRole;
  15 import org.opendaylight.controller.cluster.example.messages.PrintState;
  16 import org.opendaylight.controller.cluster.raft.ConfigParams;
  17
  18 /**
  19  * This is a test driver for testing akka-raft implementation
  20  * Its uses ExampleActors and threads to push content(key-vals) to these actors
  21  * Each ExampleActor can have one or more ClientActors. Each ClientActor spawns
  22  * a thread and starts push logs to the actor its assigned to.
  23  */
  24 public class TestDriver {
  25
  26
  27     private static Map<String, String> allPeers = new HashMap<>();
  28     private static Map<String, ActorRef> clientActorRefs  = new HashMap<String, ActorRef>();
  29     private static Map<String, ActorRef> actorRefs = new HashMap<String, ActorRef>();
  30     private static LogGenerator logGenerator = new LogGenerator();
  31     private int nameCounter = 0;
  32     private static ConfigParams configParams = new ExampleConfigParamsImpl();
  33
  34     private static ActorSystem actorSystem;
  35     private static ActorSystem listenerActorSystem;
  36
  37     /**
  38      * Create nodes, add clients and start logging.
  39      * Commands
  40      *  bye
  41      *  createNodes:{num}
  42      *  stopNode:{nodeName}
  43      *  reinstateNode:{nodeName}
  44      *  addClients:{num}
  45      *  addClientsToNode:{nodeName, num}
  46      *  startLogging
  47      *  stopLogging
  48      *  startLoggingForClient:{nodeName}
  49      *  stopLoggingForClient:{nodeName}
  50      *  printNodes
  51      *  printState
  52      *
  53      *  Note: when run on IDE and on debug log level, the debug logs in
  54      *  AbstractUptypedActor and AbstractUptypedPersistentActor would need to be commented out.
  55      *  Also RaftActor handleCommand(), debug log which prints for every command other than AE/AER
  56      *
  57      * @param args
  58      * @throws Exception
  59      */
  60     public static void main(String[] args) throws Exception {
  61
  62         actorSystem = ActorSystem.create("raft-test", ConfigFactory
  63             .load().getConfig("raft-test"));
  64
  65         listenerActorSystem = ActorSystem.create("raft-test-listener", ConfigFactory
  66             .load().getConfig("raft-test-listener"));
  67
  68         TestDriver td = new TestDriver();
  69
  70         System.out.println("Enter command (type bye to exit):");
  71
  72
  73         BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
  74         while(true) {
  75             String command = br.readLine();
  76             if (command.startsWith("bye")) {
  77                 System.exit(0);
  78
  79             } else if (command.startsWith("createNodes")) {
  80                 String[] arr = command.split(":");
  81                 int n = Integer.parseInt(arr[1]);
  82                 td.createNodes(n);
  83
  84             } else if (command.startsWith("addClients")) {
  85                 String[] arr = command.split(":");
  86                 int n = Integer.parseInt(arr[1]);
  87                 td.addClients(n);
  88
  89             } else if (command.startsWith("addClientsToNode")) {
  90                 String[] arr = command.split(":");
  91                 String nodeName = arr[1];
  92                 int n = Integer.parseInt(arr[1]);
  93                 td.addClientsToNode(nodeName, n);
  94
  95             } else if (command.startsWith("stopNode")) {
  96                 String[] arr = command.split(":");
  97                 td.stopNode(arr[1]);
  98
  99             } else if (command.startsWith("reinstateNode")) {
 100                 String[] arr = command.split(":");
 101                 td.reinstateNode(arr[1]);
 102
 103             } else if (command.startsWith("startLogging")) {
 104                 td.startAllLogging();
 105
 106             } else if (command.startsWith("startLoggingForClient")) {
 107                 String[] arr = command.split(":");
 108                 td.startLoggingForClient(clientActorRefs.get(arr[1]));
 109
 110             } else if (command.startsWith("stopLogging")) {
 111                 td.stopAllLogging();
 112
 113             } else if (command.startsWith("stopLoggingForClient")) {
 114                 String[] arr = command.split(":");
 115                 td.stopLoggingForClient(clientActorRefs.get(arr[1]));
 116
 117             } else if (command.startsWith("printState")) {
 118                 td.printState();
 119             } else if (command.startsWith("printNodes")) {
 120                 td.printNodes();
 121             } else {
 122                 System.out.println("Invalid command:" + command);
 123             }
 124
 125         }
 126     }
 127
 128     // create the listener using a separate actor system for each example actor
 129     private void createClusterRoleChangeListener(List<String> memberIds) {
 130         System.out.println("memberIds="+memberIds);
 131         for (String memberId : memberIds) {
 132             ActorRef listenerActor = listenerActorSystem.actorOf(
 133                 ExampleRoleChangeListener.getProps(memberId), memberId + "-role-change-listener");
 134             System.out.println("Role Change Listener created:" + listenerActor.path().toString());
 135         }
 136     }
 137
 138     public static ActorRef createExampleActor(String name) {
 139         return actorSystem.actorOf(ExampleActor.props(name, withoutPeer(name),
 140             Optional.of(configParams)), name);
 141     }
 142
 143     public void createNodes(int num) {
 144         for (int i=0; i < num; i++)  {
 145             nameCounter = nameCounter + 1;
 146             allPeers.put("example-"+nameCounter, "akka://raft-test/user/example-"+nameCounter);
 147         }
 148
 149         for (String s : allPeers.keySet())  {
 150             ActorRef exampleActor = createExampleActor(s);
 151             actorRefs.put(s, exampleActor);
 152             System.out.println("Created node:"+s);
 153
 154         }
 155
 156         createClusterRoleChangeListener(Lists.newArrayList(allPeers.keySet()));
 157     }
 158
 159     // add num clients to all nodes in the system
 160     public void addClients(int num) {
 161         for(Map.Entry<String,ActorRef> actorRefEntry : actorRefs.entrySet()) {
 162             for (int i=0; i < num; i++) {
 163                 String clientName = "client-" + i + "-" + actorRefEntry.getKey();
 164                 ActorRef clientActor = actorSystem.actorOf(
 165                     ClientActor.props(actorRefEntry.getValue()), clientName);
 166                 clientActorRefs.put(clientName, clientActor);
 167                 System.out.println("Created client-node:" + clientName);
 168             }
 169         }
 170     }
 171
 172     // add num clients to a node
 173     public void addClientsToNode(String actorName, int num) {
 174         ActorRef actorRef = actorRefs.get(actorName);
 175         for (int i=0; i < num; i++) {
 176             String clientName = "client-" + i + "-" + actorName;
 177             clientActorRefs.put(clientName,
 178                 actorSystem.actorOf(ClientActor.props(actorRef), clientName));
 179             System.out.println("Added client-node:" + clientName);
 180         }
 181     }
 182
 183     public void stopNode(String actorName) {
 184         ActorRef actorRef = actorRefs.get(actorName);
 185
 186         for (Map.Entry<String,ActorRef> entry : clientActorRefs.entrySet()) {
 187             if (entry.getKey().endsWith(actorName)) {
 188                 actorSystem.stop(entry.getValue());
 189             }
 190         }
 191
 192         actorSystem.stop(actorRef);
 193         actorRefs.remove(actorName);
 194         allPeers.remove(actorName);
 195     }
 196
 197     public void reinstateNode(String actorName) {
 198         String address = "akka://default/user/"+actorName;
 199         allPeers.put(actorName, address);
 200
 201         ActorRef exampleActor = createExampleActor(actorName);
 202         actorRefs.put(actorName, exampleActor);
 203
 204         addClientsToNode(actorName, 1);
 205     }
 206
 207     public void startAllLogging() {
 208         if(!clientActorRefs.isEmpty()) {
 209             for(Map.Entry<String,ActorRef> client : clientActorRefs.entrySet()) {
 210                 logGenerator.startLoggingForClient(client.getValue());
 211                 System.out.println("Started logging for client:"+client.getKey());
 212             }
 213         } else {
 214             System.out.println("There are no clients for any nodes. First create clients using commands- addClients:<num> or addClientsToNode:<nodename>:<num>");
 215         }
 216
 217     }
 218
 219     public void startLoggingForClient(ActorRef client) {
 220         logGenerator.startLoggingForClient(client);
 221     }
 222
 223     public void stopAllLogging() {
 224         for(Map.Entry<String,ActorRef> client : clientActorRefs.entrySet()) {
 225             logGenerator.stopLoggingForClient(client.getValue());
 226         }
 227     }
 228
 229     public void stopLoggingForClient(ActorRef client) {
 230         logGenerator.stopLoggingForClient(client);
 231     }
 232
 233     public void printState() {
 234         for (ActorRef ref : actorRefs.values()) {
 235             ref.tell(new PrintState(), null);
 236         }
 237     }
 238
 239     public void printNodes() {
 240         for (ActorRef ref : actorRefs.values()) {
 241             ref.tell(new PrintRole(), null);
 242         }
 243     }
 244
 245     public ActorRef getLeader() {
 246         return null;
 247     }
 248
 249
 250     private static Map<String, String> withoutPeer(String peerId) {
 251         Map<String, String> without = new ConcurrentHashMap<>(allPeers);
 252         without.remove(peerId);
 253
 254         return without;
 255     }
 256 }
 257