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

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