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 org.opendaylight.controller.cluster.example.messages.PrintRole;
   7 import org.opendaylight.controller.cluster.example.messages.PrintState;
   8 import org.opendaylight.controller.cluster.raft.ConfigParams;
   9 import org.opendaylight.controller.cluster.raft.client.messages.AddRaftPeer;
  10 import org.opendaylight.controller.cluster.raft.client.messages.RemoveRaftPeer;
  11
  12 import java.io.BufferedReader;
  13 import java.io.InputStreamReader;
  14 import java.util.HashMap;
  15 import java.util.Map;
  16 import java.util.concurrent.ConcurrentHashMap;
  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     private static final ActorSystem actorSystem = ActorSystem.create();
  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     /**
  35      * Create nodes, add clients and start logging.
  36      * Commands
  37      *  bye
  38      *  createNodes:{num}
  39      *  addNodes:{num}
  40      *  stopNode:{nodeName}
  41      *  reinstateNode:{nodeName}
  42      *  addClients:{num}
  43      *  addClientsToNode:{nodeName, num}
  44      *  startLogging
  45      *  stopLogging
  46      *  startLoggingForClient:{nodeName}
  47      *  stopLoggingForClient:{nodeName}
  48      *  printNodes
  49      *  printState
  50      * @param args
  51      * @throws Exception
  52      */
  53     public static void main(String[] args) throws Exception {
  54         TestDriver td = new TestDriver();
  55
  56         System.out.println("Enter command (type bye to exit):");
  57
  58
  59         BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
  60         while(true) {
  61             String command = br.readLine();
  62             if (command.startsWith("bye")) {
  63                 System.exit(0);
  64
  65             } else if (command.startsWith("createNodes")) {
  66                 String[] arr = command.split(":");
  67                 int n = Integer.parseInt(arr[1]);
  68                 td.createNodes(n);
  69
  70             } else if (command.startsWith("addNodes")) {
  71                 String[] arr = command.split(":");
  72                 int n = Integer.parseInt(arr[1]);
  73                 td.addNodes(n);
  74
  75             } else if (command.startsWith("addClients")) {
  76                 String[] arr = command.split(":");
  77                 int n = Integer.parseInt(arr[1]);
  78                 td.addClients(n);
  79
  80             } else if (command.startsWith("addClientsToNode")) {
  81                 String[] arr = command.split(":");
  82                 String nodeName = arr[1];
  83                 int n = Integer.parseInt(arr[1]);
  84                 td.addClientsToNode(nodeName, n);
  85
  86             } else if (command.startsWith("stopNode")) {
  87                 String[] arr = command.split(":");
  88                 td.stopNode(arr[1]);
  89
  90             } else if (command.startsWith("reinstateNode")) {
  91                 String[] arr = command.split(":");
  92                 td.reinstateNode(arr[1]);
  93
  94             } else if (command.startsWith("startLogging")) {
  95                 td.startAllLogging();
  96
  97             } else if (command.startsWith("startLoggingForClient")) {
  98                 String[] arr = command.split(":");
  99                 td.startLoggingForClient(clientActorRefs.get(arr[1]));
 100
 101             } else if (command.startsWith("stopLogging")) {
 102                 td.stopAllLogging();
 103
 104             } else if (command.startsWith("stopLoggingForClient")) {
 105                 String[] arr = command.split(":");
 106                 td.stopLoggingForClient(clientActorRefs.get(arr[1]));
 107
 108             } else if (command.startsWith("printState")) {
 109                 td.printState();
 110             } else if (command.startsWith("printNodes")) {
 111                 td.printNodes();
 112             } else {
 113                 System.out.println("Invalid command:" + command);
 114             }
 115
 116         }
 117     }
 118
 119     public static ActorRef createExampleActor(String name) {
 120         return actorSystem.actorOf(ExampleActor.props(name, withoutPeer(name),
 121             Optional.of(configParams)), name);
 122     }
 123
 124     public void createNodes(int num) {
 125         for (int i=0; i < num; i++)  {
 126             nameCounter = nameCounter + 1;
 127             allPeers.put("example-"+nameCounter, "akka://default/user/example-"+nameCounter);
 128         }
 129
 130         for (String s : allPeers.keySet())  {
 131             ActorRef exampleActor = createExampleActor(s);
 132             actorRefs.put(s, exampleActor);
 133             System.out.println("Created node:"+s);
 134
 135         }
 136     }
 137
 138     // add new nodes , pass in the count
 139     public void addNodes(int num) {
 140         Map<String, String> newPeers = new HashMap<>();
 141         for (int i=0; i < num; i++)  {
 142             nameCounter = nameCounter + 1;
 143             newPeers.put("example-"+nameCounter, "akka://default/user/example-"+nameCounter);
 144             allPeers.put("example-"+nameCounter, "akka://default/user/example-"+nameCounter);
 145
 146         }
 147         Map<String, ActorRef> newActorRefs = new HashMap<String, ActorRef>(num);
 148         for (Map.Entry<String, String> entry : newPeers.entrySet())  {
 149             ActorRef exampleActor = createExampleActor(entry.getKey());
 150             newActorRefs.put(entry.getKey(), exampleActor);
 151
 152             //now also add these new nodes as peers from the previous nodes
 153             for (ActorRef actor : actorRefs.values()) {
 154                 actor.tell(new AddRaftPeer(entry.getKey(), entry.getValue()), null);
 155             }
 156
 157             System.out.println("Added node:" + entry);
 158         }
 159
 160         actorRefs.putAll(newActorRefs);
 161     }
 162
 163
 164     // add num clients to all nodes in the system
 165     public void addClients(int num) {
 166         for(Map.Entry<String,ActorRef> actorRefEntry : actorRefs.entrySet()) {
 167             for (int i=0; i < num; i++) {
 168                 String clientName = "client-" + i + "-" + actorRefEntry.getKey();
 169                 ActorRef clientActor = actorSystem.actorOf(
 170                     ClientActor.props(actorRefEntry.getValue()), clientName);
 171                 clientActorRefs.put(clientName, clientActor);
 172                 System.out.println("Created client-node:" + clientName);
 173             }
 174         }
 175     }
 176
 177     // add num clients to a node
 178     public void addClientsToNode(String actorName, int num) {
 179         ActorRef actorRef = actorRefs.get(actorName);
 180         for (int i=0; i < num; i++) {
 181             String clientName = "client-" + i + "-" + actorName;
 182             clientActorRefs.put(clientName,
 183                 actorSystem.actorOf(ClientActor.props(actorRef), clientName));
 184             System.out.println("Added client-node:" + clientName);
 185         }
 186     }
 187
 188     public void stopNode(String actorName) {
 189         ActorRef actorRef = actorRefs.get(actorName);
 190
 191         for (Map.Entry<String,ActorRef> entry : clientActorRefs.entrySet()) {
 192             if (entry.getKey().endsWith(actorName)) {
 193                 actorSystem.stop(entry.getValue());
 194             }
 195         }
 196
 197         actorSystem.stop(actorRef);
 198         actorRefs.remove(actorName);
 199
 200         for (ActorRef actor : actorRefs.values()) {
 201             actor.tell(new RemoveRaftPeer(actorName), null);
 202         }
 203
 204         allPeers.remove(actorName);
 205     }
 206
 207     public void reinstateNode(String actorName) {
 208         String address = "akka://default/user/"+actorName;
 209         allPeers.put(actorName, address);
 210
 211         ActorRef exampleActor = createExampleActor(actorName);
 212
 213         for (ActorRef actor : actorRefs.values()) {
 214             actor.tell(new AddRaftPeer(actorName, address), null);
 215         }
 216
 217         actorRefs.put(actorName, exampleActor);
 218
 219         addClientsToNode(actorName, 1);
 220     }
 221
 222     public void startAllLogging() {
 223         if(!clientActorRefs.isEmpty()) {
 224             for(Map.Entry<String,ActorRef> client : clientActorRefs.entrySet()) {
 225                 logGenerator.startLoggingForClient(client.getValue());
 226                 System.out.println("Started logging for client:"+client.getKey());
 227             }
 228         } else {
 229             System.out.println("There are no clients for any nodes. First create clients using commands- addClients:<num> or addClientsToNode:<nodename>:<num>");
 230         }
 231
 232     }
 233
 234     public void startLoggingForClient(ActorRef client) {
 235         logGenerator.startLoggingForClient(client);
 236     }
 237
 238     public void stopAllLogging() {
 239         for(Map.Entry<String,ActorRef> client : clientActorRefs.entrySet()) {
 240             logGenerator.stopLoggingForClient(client.getValue());
 241         }
 242     }
 243
 244     public void stopLoggingForClient(ActorRef client) {
 245         logGenerator.stopLoggingForClient(client);
 246     }
 247
 248     public void printState() {
 249         for (ActorRef ref : actorRefs.values()) {
 250             ref.tell(new PrintState(), null);
 251         }
 252     }
 253
 254     public void printNodes() {
 255         for (ActorRef ref : actorRefs.values()) {
 256             ref.tell(new PrintRole(), null);
 257         }
 258     }
 259
 260     public ActorRef getLeader() {
 261         return null;
 262     }
 263
 264
 265     private static Map<String, String> withoutPeer(String peerId) {
 266         Map<String, String> without = new ConcurrentHashMap<>(allPeers);
 267         without.remove(peerId);
 268
 269         return without;
 270     }
 271 }
 272