opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/core/internal/Controller.java

   1 /*
   2  * Copyright (c) 2013 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.protocol_plugin.openflow.core.internal;
  10
  11 import java.io.IOException;
  12 import java.nio.channels.SelectionKey;
  13 import java.nio.channels.Selector;
  14 import java.nio.channels.ServerSocketChannel;
  15 import java.nio.channels.SocketChannel;
  16 import java.util.Comparator;
  17 import java.util.Date;
  18 import java.util.Iterator;
  19 import java.util.Map;
  20 import java.util.Map.Entry;
  21 import java.util.Set;
  22 import java.util.concurrent.ConcurrentHashMap;
  23 import java.util.concurrent.ConcurrentMap;
  24 import java.util.concurrent.PriorityBlockingQueue;
  25 import java.util.concurrent.atomic.AtomicInteger;
  26
  27 import org.eclipse.osgi.framework.console.CommandInterpreter;
  28 import org.eclipse.osgi.framework.console.CommandProvider;
  29 import org.opendaylight.controller.protocol_plugin.openflow.core.IController;
  30 import org.opendaylight.controller.protocol_plugin.openflow.core.IMessageListener;
  31 import org.opendaylight.controller.protocol_plugin.openflow.core.ISwitch;
  32 import org.opendaylight.controller.protocol_plugin.openflow.core.ISwitchStateListener;
  33 import org.opendaylight.controller.sal.connection.ConnectionConstants;
  34 import org.opendaylight.controller.sal.connection.IPluginInConnectionService;
  35 import org.opendaylight.controller.sal.core.Node;
  36 import org.opendaylight.controller.sal.utils.Status;
  37 import org.opendaylight.controller.sal.utils.StatusCode;
  38 import org.openflow.protocol.OFMessage;
  39 import org.openflow.protocol.OFType;
  40 import org.openflow.util.HexString;
  41 import org.osgi.framework.BundleContext;
  42 import org.osgi.framework.FrameworkUtil;
  43 import org.slf4j.Logger;
  44 import org.slf4j.LoggerFactory;
  45
  46 public class Controller implements IController, CommandProvider, IPluginInConnectionService {
  47     private static final Logger logger = LoggerFactory
  48             .getLogger(Controller.class);
  49     private ControllerIO controllerIO;
  50     private Thread switchEventThread;
  51     private ConcurrentHashMap<Long, ISwitch> switches;
  52     private PriorityBlockingQueue<SwitchEvent> switchEvents;
  53     // only 1 message listener per OFType
  54     private ConcurrentMap<OFType, IMessageListener> messageListeners;
  55     // only 1 switch state listener
  56     private ISwitchStateListener switchStateListener;
  57     private AtomicInteger switchInstanceNumber;
  58     private int MAXQUEUESIZE = 50000;
  59
  60     private static enum SwitchEventPriority { LOW, NORMAL, HIGH }
  61
  62     /*
  63      * this thread monitors the switchEvents queue for new incoming events from
  64      * switch
  65      */
  66     private class EventHandler implements Runnable {
  67         @Override
  68         public void run() {
  69
  70             while (true) {
  71                 try {
  72                     SwitchEvent ev = switchEvents.take();
  73                     SwitchEvent.SwitchEventType eType = ev.getEventType();
  74                     ISwitch sw = ev.getSwitch();
  75                     switch (eType) {
  76                     case SWITCH_ADD:
  77                         Long sid = sw.getId();
  78                         ISwitch existingSwitch = switches.get(sid);
  79                         if (existingSwitch != null) {
  80                             logger.info("Replacing existing {} with New {}",
  81                                     existingSwitch, sw);
  82                             disconnectSwitch(existingSwitch);
  83                         }
  84                         switches.put(sid, sw);
  85                         notifySwitchAdded(sw);
  86                         break;
  87                     case SWITCH_DELETE:
  88                         disconnectSwitch(sw);
  89                         break;
  90                     case SWITCH_ERROR:
  91                         disconnectSwitch(sw);
  92                         break;
  93                     case SWITCH_MESSAGE:
  94                         OFMessage msg = ev.getMsg();
  95                         if (msg != null) {
  96                             IMessageListener listener = messageListeners
  97                                     .get(msg.getType());
  98                             if (listener != null) {
  99                                 listener.receive(sw, msg);
 100                             }
 101                         }
 102                         break;
 103                     default:
 104                         logger.error("Unknown switch event {}", eType.ordinal());
 105                     }
 106                 } catch (InterruptedException e) {
 107                     switchEvents.clear();
 108                     return;
 109                 }
 110             }
 111         }
 112
 113     }
 114
 115     /**
 116      * Function called by the dependency manager when all the required
 117      * dependencies are satisfied
 118      *
 119      */
 120     public void init() {
 121         logger.debug("Initializing!");
 122         this.switches = new ConcurrentHashMap<Long, ISwitch>();
 123         this.switchEvents = new PriorityBlockingQueue<SwitchEvent>(MAXQUEUESIZE, new Comparator<SwitchEvent>() {
 124             @Override
 125             public int compare(SwitchEvent p1, SwitchEvent p2) {
 126                 return p2.getPriority() - p1.getPriority();
 127             }
 128         });
 129         this.messageListeners = new ConcurrentHashMap<OFType, IMessageListener>();
 130         this.switchStateListener = null;
 131         this.switchInstanceNumber = new AtomicInteger(0);
 132         registerWithOSGIConsole();
 133     }
 134
 135     /**
 136      * Function called by dependency manager after "init ()" is called and after
 137      * the services provided by the class are registered in the service registry
 138      *
 139      */
 140     public void start() {
 141         logger.debug("Starting!");
 142         /*
 143          * start a thread to handle event coming from the switch
 144          */
 145         switchEventThread = new Thread(new EventHandler(), "SwitchEvent Thread");
 146         switchEventThread.start();
 147
 148         // spawn a thread to start to listen on the open flow port
 149         controllerIO = new ControllerIO(this);
 150         try {
 151             controllerIO.start();
 152         } catch (IOException ex) {
 153             logger.error("Caught exception while starting:", ex);
 154         }
 155     }
 156
 157     /**
 158      * Function called by the dependency manager before the services exported by
 159      * the component are unregistered, this will be followed by a "destroy ()"
 160      * calls
 161      *
 162      */
 163     public void stop() {
 164         for (Iterator<Entry<Long, ISwitch>> it = switches.entrySet().iterator(); it
 165                 .hasNext();) {
 166             Entry<Long, ISwitch> entry = it.next();
 167             ((SwitchHandler) entry.getValue()).stop();
 168             it.remove();
 169         }
 170         switchEventThread.interrupt();
 171         try {
 172             controllerIO.shutDown();
 173         } catch (IOException ex) {
 174             logger.error("Caught exception while stopping:", ex);
 175         }
 176     }
 177
 178     /**
 179      * Function called by the dependency manager when at least one dependency
 180      * become unsatisfied or when the component is shutting down because for
 181      * example bundle is being stopped.
 182      *
 183      */
 184     public void destroy() {
 185     }
 186
 187     @Override
 188     public void addMessageListener(OFType type, IMessageListener listener) {
 189         IMessageListener currentListener = this.messageListeners.get(type);
 190         if (currentListener != null) {
 191             logger.warn("{} is already listened by {}", type,
 192                     currentListener);
 193         }
 194         this.messageListeners.put(type, listener);
 195         logger.debug("{} is now listened by {}", type, listener);
 196     }
 197
 198     @Override
 199     public void removeMessageListener(OFType type, IMessageListener listener) {
 200         IMessageListener currentListener = this.messageListeners.get(type);
 201         if ((currentListener != null) && (currentListener == listener)) {
 202             logger.debug("{} listener {} is Removed", type, listener);
 203             this.messageListeners.remove(type);
 204         }
 205     }
 206
 207     @Override
 208     public void addSwitchStateListener(ISwitchStateListener listener) {
 209         if (this.switchStateListener != null) {
 210             logger.warn("Switch events are already listened by {}",
 211                     this.switchStateListener);
 212         }
 213         this.switchStateListener = listener;
 214         logger.debug("Switch events are now listened by {}", listener);
 215     }
 216
 217     @Override
 218     public void removeSwitchStateListener(ISwitchStateListener listener) {
 219         if ((this.switchStateListener != null)
 220                 && (this.switchStateListener == listener)) {
 221             logger.debug("SwitchStateListener {} is Removed", listener);
 222             this.switchStateListener = null;
 223         }
 224     }
 225
 226     public void handleNewConnection(Selector selector,
 227             SelectionKey serverSelectionKey) {
 228         ServerSocketChannel ssc = (ServerSocketChannel) serverSelectionKey
 229                 .channel();
 230         SocketChannel sc = null;
 231         try {
 232             sc = ssc.accept();
 233             // create new switch
 234             int i = this.switchInstanceNumber.addAndGet(1);
 235             String instanceName = "SwitchHandler-" + i;
 236             SwitchHandler switchHandler = new SwitchHandler(this, sc, instanceName);
 237             switchHandler.start();
 238             if (sc.isConnected()) {
 239                 logger.info("Switch:{} is connected to the Controller",
 240                         sc.socket().getRemoteSocketAddress()
 241                         .toString().split("/")[1]);
 242             }
 243
 244         } catch (IOException e) {
 245             return;
 246         }
 247     }
 248
 249     private void disconnectSwitch(ISwitch sw) {
 250         if (((SwitchHandler) sw).isOperational()) {
 251             Long sid = sw.getId();
 252             if (this.switches.remove(sid, sw)) {
 253                 logger.info("{} is removed", sw);
 254                 notifySwitchDeleted(sw);
 255             }
 256         }
 257         ((SwitchHandler) sw).stop();
 258         sw = null;
 259     }
 260
 261     private void notifySwitchAdded(ISwitch sw) {
 262         if (switchStateListener != null) {
 263             switchStateListener.switchAdded(sw);
 264         }
 265     }
 266
 267     private void notifySwitchDeleted(ISwitch sw) {
 268         if (switchStateListener != null) {
 269             switchStateListener.switchDeleted(sw);
 270         }
 271     }
 272
 273     private synchronized void addSwitchEvent(SwitchEvent event) {
 274         this.switchEvents.put(event);
 275     }
 276
 277     public void takeSwitchEventAdd(ISwitch sw) {
 278         SwitchEvent ev = new SwitchEvent(SwitchEvent.SwitchEventType.SWITCH_ADD, sw, null,
 279                 SwitchEventPriority.HIGH.ordinal());
 280         addSwitchEvent(ev);
 281     }
 282
 283     public void takeSwitchEventDelete(ISwitch sw) {
 284         SwitchEvent ev = new SwitchEvent(SwitchEvent.SwitchEventType.SWITCH_DELETE, sw, null,
 285                 SwitchEventPriority.HIGH.ordinal());
 286         addSwitchEvent(ev);
 287     }
 288
 289     public void takeSwitchEventError(ISwitch sw) {
 290         SwitchEvent ev = new SwitchEvent(SwitchEvent.SwitchEventType.SWITCH_ERROR, sw, null,
 291                 SwitchEventPriority.NORMAL.ordinal());
 292         addSwitchEvent(ev);
 293     }
 294
 295     public void takeSwitchEventMsg(ISwitch sw, OFMessage msg) {
 296         if (messageListeners.get(msg.getType()) != null) {
 297             SwitchEvent ev = new SwitchEvent(SwitchEvent.SwitchEventType.SWITCH_MESSAGE, sw, msg,
 298                     SwitchEventPriority.LOW.ordinal());
 299             addSwitchEvent(ev);
 300         }
 301     }
 302
 303     @Override
 304     public Map<Long, ISwitch> getSwitches() {
 305         return this.switches;
 306     }
 307
 308     @Override
 309     public ISwitch getSwitch(Long switchId) {
 310         return this.switches.get(switchId);
 311     }
 312
 313     public void _controllerShowQueueSize(CommandInterpreter ci) {
 314         ci.print("switchEvents queue size: " + switchEvents.size() + "\n");
 315     }
 316
 317     public void _controllerShowSwitches(CommandInterpreter ci) {
 318         Set<Long> sids = switches.keySet();
 319         StringBuffer s = new StringBuffer();
 320         int size = sids.size();
 321         if (size == 0) {
 322             ci.print("switches: empty");
 323             return;
 324         }
 325         Iterator<Long> iter = sids.iterator();
 326         s.append("Total: " + size + " switches\n");
 327         while (iter.hasNext()) {
 328             Long sid = iter.next();
 329             Date date = switches.get(sid).getConnectedDate();
 330             String switchInstanceName = ((SwitchHandler) switches.get(sid))
 331                     .getInstanceName();
 332             s.append(switchInstanceName + "/" + HexString.toHexString(sid)
 333                     + " connected since " + date.toString() + "\n");
 334         }
 335         ci.print(s.toString());
 336         return;
 337     }
 338
 339     public void _controllerReset(CommandInterpreter ci) {
 340         ci.print("...Disconnecting the communication to all switches...\n");
 341         stop();
 342         try {
 343             Thread.sleep(1000);
 344         } catch (InterruptedException ie) {
 345         } finally {
 346             ci.print("...start to accept connections from switches...\n");
 347             start();
 348         }
 349     }
 350
 351     public void _controllerShowConnConfig(CommandInterpreter ci) {
 352         String str = System.getProperty("secureChannelEnabled");
 353         if ((str != null) && (str.trim().equalsIgnoreCase("true"))) {
 354             ci.print("The Controller and Switch should communicate through TLS connetion.\n");
 355
 356             String keyStoreFile = System.getProperty("controllerKeyStore");
 357             String trustStoreFile = System.getProperty("controllerTrustStore");
 358             if ((keyStoreFile == null) || keyStoreFile.trim().isEmpty()) {
 359                 ci.print("controllerKeyStore not specified\n");
 360             } else {
 361                 ci.print("controllerKeyStore=" + keyStoreFile + "\n");
 362             }
 363             if ((trustStoreFile == null) || trustStoreFile.trim().isEmpty()) {
 364                 ci.print("controllerTrustStore not specified\n");
 365             } else {
 366                 ci.print("controllerTrustStore=" + trustStoreFile + "\n");
 367             }
 368         } else {
 369             ci.print("The Controller and Switch should communicate through TCP connetion.\n");
 370         }
 371     }
 372
 373     private void registerWithOSGIConsole() {
 374         BundleContext bundleContext = FrameworkUtil.getBundle(this.getClass())
 375                 .getBundleContext();
 376         bundleContext.registerService(CommandProvider.class.getName(), this,
 377                 null);
 378     }
 379
 380     @Override
 381     public String getHelp() {
 382         StringBuffer help = new StringBuffer();
 383         help.append("---Open Flow Controller---\n");
 384         help.append("\t controllerShowSwitches\n");
 385         help.append("\t controllerReset\n");
 386         help.append("\t controllerShowConnConfig\n");
 387         help.append("\t controllerShowQueueSize\n");
 388         return help.toString();
 389     }
 390
 391     @Override
 392     public Status disconnect(Node node) {
 393         ISwitch sw = getSwitch((Long) node.getID());
 394         if (sw != null) disconnectSwitch(sw);
 395         return new Status(StatusCode.SUCCESS);
 396     }
 397
 398     @Override
 399     public Node connect(String connectionIdentifier, Map<ConnectionConstants, String> params) {
 400         return null;
 401     }
 402
 403     /**
 404      * View Change notification
 405      */
 406     public void notifyClusterViewChanged() {
 407         for (ISwitch sw : switches.values()) {
 408             notifySwitchAdded(sw);
 409         }
 410     }
 411
 412     /**
 413      * Node Disconnected from the node's master controller.
 414      */
 415     @Override
 416     public void notifyNodeDisconnectFromMaster(Node node) {
 417         ISwitch sw = switches.get((Long)node.getID());
 418         if (sw != null) notifySwitchAdded(sw);
 419     }
 420 }