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