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