opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/TopologyServiceShim.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.internal;
  10
  11 import java.util.ArrayList;
  12 import java.util.Collection;
  13 import java.util.HashMap;
  14 import java.util.HashSet;
  15 import java.util.List;
  16 import java.util.Map;
  17 import java.util.Set;
  18 import java.util.Timer;
  19 import java.util.TimerTask;
  20 import java.util.concurrent.BlockingQueue;
  21 import java.util.concurrent.ConcurrentHashMap;
  22 import java.util.concurrent.ConcurrentMap;
  23 import java.util.concurrent.CopyOnWriteArrayList;
  24 import java.util.concurrent.LinkedBlockingQueue;
  25
  26 import org.apache.commons.lang3.tuple.ImmutablePair;
  27 import org.apache.commons.lang3.tuple.Pair;
  28 import org.eclipse.osgi.framework.console.CommandInterpreter;
  29 import org.eclipse.osgi.framework.console.CommandProvider;
  30 import org.opendaylight.controller.protocol_plugin.openflow.IDiscoveryListener;
  31 import org.opendaylight.controller.protocol_plugin.openflow.IInventoryShimExternalListener;
  32 import org.opendaylight.controller.protocol_plugin.openflow.IOFStatisticsManager;
  33 import org.opendaylight.controller.protocol_plugin.openflow.IRefreshInternalProvider;
  34 import org.opendaylight.controller.protocol_plugin.openflow.ITopologyServiceShimListener;
  35 import org.opendaylight.controller.sal.core.Bandwidth;
  36 import org.opendaylight.controller.sal.core.Config;
  37 import org.opendaylight.controller.sal.core.ContainerFlow;
  38 import org.opendaylight.controller.sal.core.Edge;
  39 import org.opendaylight.controller.sal.core.IContainerAware;
  40 import org.opendaylight.controller.sal.core.IContainerListener;
  41 import org.opendaylight.controller.sal.core.Node;
  42 import org.opendaylight.controller.sal.core.NodeConnector;
  43 import org.opendaylight.controller.sal.core.Property;
  44 import org.opendaylight.controller.sal.core.State;
  45 import org.opendaylight.controller.sal.core.UpdateType;
  46 import org.opendaylight.controller.sal.topology.TopoEdgeUpdate;
  47 import org.opendaylight.controller.sal.utils.GlobalConstants;
  48 import org.osgi.framework.BundleContext;
  49 import org.osgi.framework.FrameworkUtil;
  50 import org.slf4j.Logger;
  51 import org.slf4j.LoggerFactory;
  52
  53 /**
  54  * The class describes a shim layer that relays the topology events from
  55  * OpenFlow core to various listeners. The notifications are filtered based on
  56  * container configurations.
  57  */
  58 public class TopologyServiceShim implements IDiscoveryListener,
  59         IContainerListener, CommandProvider, IRefreshInternalProvider,
  60         IInventoryShimExternalListener, IContainerAware {
  61     protected static final Logger logger = LoggerFactory
  62             .getLogger(TopologyServiceShim.class);
  63     private ConcurrentMap<String, ITopologyServiceShimListener> topologyServiceShimListeners = new ConcurrentHashMap<String, ITopologyServiceShimListener>();
  64     private ConcurrentMap<NodeConnector, List<String>> containerMap = new ConcurrentHashMap<NodeConnector, List<String>>();
  65     private ConcurrentMap<String, ConcurrentMap<NodeConnector, Pair<Edge, Set<Property>>>> edgeMap = new ConcurrentHashMap<String, ConcurrentMap<NodeConnector, Pair<Edge, Set<Property>>>>();
  66
  67     private BlockingQueue<NotifyEntry> notifyQ;
  68     private Thread notifyThread;
  69     private BlockingQueue<String> bulkNotifyQ;
  70     private Thread ofPluginTopoBulkUpdate;
  71     private volatile Boolean shuttingDown = false;
  72     private IOFStatisticsManager statsMgr;
  73     private Timer pollTimer;
  74     private TimerTask txRatePoller;
  75     private Thread bwUtilNotifyThread;
  76     private BlockingQueue<UtilizationUpdate> bwUtilNotifyQ;
  77     private List<NodeConnector> connectorsOverUtilized;
  78     private float bwThresholdFactor = (float) 0.8; // Threshold = 80% of link
  79                                                    // bandwidth
  80
  81     class NotifyEntry {
  82         String container;
  83         List<TopoEdgeUpdate> teuList;
  84
  85         public NotifyEntry(String container, TopoEdgeUpdate teu) {
  86             this.container = container;
  87             this.teuList = new ArrayList<TopoEdgeUpdate>();
  88             if (teu != null) {
  89                 this.teuList.add(teu);
  90             }
  91         }
  92
  93         public NotifyEntry(String container, List<TopoEdgeUpdate> teuList) {
  94             this.container = container;
  95             this.teuList = new ArrayList<TopoEdgeUpdate>();
  96             if (teuList != null) {
  97                 this.teuList.addAll(teuList);
  98             }
  99         }
 100     }
 101
 102     class TopologyNotify implements Runnable {
 103         private final BlockingQueue<NotifyEntry> notifyQ;
 104         private NotifyEntry entry;
 105         private Map<String, List<TopoEdgeUpdate>> teuMap = new HashMap<String, List<TopoEdgeUpdate>>();
 106         private List<TopoEdgeUpdate> teuList;
 107         private boolean notifyListeners;
 108
 109         TopologyNotify(BlockingQueue<NotifyEntry> notifyQ) {
 110             this.notifyQ = notifyQ;
 111         }
 112
 113         @Override
 114         public void run() {
 115             while (true) {
 116                 try {
 117                     teuMap.clear();
 118                     notifyListeners = false;
 119                     while (!notifyQ.isEmpty()) {
 120                         entry = notifyQ.take();
 121                         teuList = teuMap.get(entry.container);
 122                         if (teuList == null) {
 123                             teuList = new ArrayList<TopoEdgeUpdate>();
 124                         }
 125                         // group all the updates together
 126                         teuList.addAll(entry.teuList);
 127                         teuMap.put(entry.container, teuList);
 128                         notifyListeners = true;
 129                     }
 130
 131                     if (notifyListeners) {
 132                         for (String container : teuMap.keySet()) {
 133                             // notify the listener
 134                             ITopologyServiceShimListener l = topologyServiceShimListeners.get(container);
 135                             // container topology service may not have come up yet
 136                             if (l != null) {
 137                                 l.edgeUpdate(teuMap.get(container));
 138                             }
 139                         }
 140                     }
 141
 142                     Thread.sleep(100);
 143                 } catch (InterruptedException e1) {
 144                     logger.warn("TopologyNotify interrupted {}",
 145                             e1.getMessage());
 146                     if (shuttingDown) {
 147                         return;
 148                     }
 149                 } catch (Exception e2) {
 150                     logger.error("", e2);
 151                 }
 152             }
 153         }
 154     }
 155
 156     class UtilizationUpdate {
 157         NodeConnector connector;
 158         UpdateType type;
 159
 160         UtilizationUpdate(NodeConnector connector, UpdateType type) {
 161             this.connector = connector;
 162             this.type = type;
 163         }
 164     }
 165
 166     class BwUtilizationNotify implements Runnable {
 167         private final BlockingQueue<UtilizationUpdate> notifyQ;
 168
 169         BwUtilizationNotify(BlockingQueue<UtilizationUpdate> notifyQ) {
 170             this.notifyQ = notifyQ;
 171         }
 172
 173         @Override
 174         public void run() {
 175             while (true) {
 176                 try {
 177                     UtilizationUpdate update = notifyQ.take();
 178                     NodeConnector connector = update.connector;
 179                     Set<String> containerList = edgeMap.keySet();
 180                     for (String container : containerList) {
 181                         Map<NodeConnector, Pair<Edge, Set<Property>>> edgePropsMap = edgeMap
 182                                 .get(container);
 183                         Edge edge = edgePropsMap.get(connector).getLeft();
 184                         if (edge.getTailNodeConnector().equals(connector)) {
 185                             ITopologyServiceShimListener topologServiceShimListener = topologyServiceShimListeners
 186                                     .get(container);
 187                             if (update.type == UpdateType.ADDED) {
 188                                 topologServiceShimListener
 189                                         .edgeOverUtilized(edge);
 190                             } else {
 191                                 topologServiceShimListener
 192                                         .edgeUtilBackToNormal(edge);
 193                             }
 194                         }
 195                     }
 196                 } catch (InterruptedException e1) {
 197                     logger.warn(
 198                             "Edge Bandwidth Utilization Notify Thread interrupted {}",
 199                             e1.getMessage());
 200                     if (shuttingDown) {
 201                         return;
 202                     }
 203                 } catch (Exception e2) {
 204                     logger.error("", e2);
 205                 }
 206             }
 207         }
 208     }
 209
 210     /**
 211      * Function called by the dependency manager when all the required
 212      * dependencies are satisfied
 213      *
 214      */
 215     void init() {
 216         logger.trace("Init called");
 217         connectorsOverUtilized = new ArrayList<NodeConnector>();
 218         notifyQ = new LinkedBlockingQueue<NotifyEntry>();
 219         notifyThread = new Thread(new TopologyNotify(notifyQ));
 220         bwUtilNotifyQ = new LinkedBlockingQueue<UtilizationUpdate>();
 221         bwUtilNotifyThread = new Thread(new BwUtilizationNotify(bwUtilNotifyQ));
 222         bulkNotifyQ = new LinkedBlockingQueue<String>();
 223         ofPluginTopoBulkUpdate = new Thread(new Runnable() {
 224             @Override
 225             public void run() {
 226                 while (true) {
 227                     try {
 228                         String containerName = bulkNotifyQ.take();
 229                         logger.debug("Bulk Notify container:{}", containerName);
 230                         TopologyBulkUpdate(containerName);
 231                     } catch (InterruptedException e) {
 232                         logger.warn("Topology Bulk update thread interrupted");
 233                         if (shuttingDown) {
 234                             return;
 235                         }
 236                     }
 237                 }
 238             }
 239         }, "Topology Bulk Update");
 240
 241         // Initialize node connector tx bit rate poller timer
 242         pollTimer = new Timer();
 243         txRatePoller = new TimerTask() {
 244             @Override
 245             public void run() {
 246                 pollTxBitRates();
 247             }
 248         };
 249
 250         registerWithOSGIConsole();
 251     }
 252
 253     /**
 254      * Continuously polls the transmit bit rate for all the node connectors from
 255      * statistics manager and trigger the warning notification upward when the
 256      * transmit rate is above a threshold which is a percentage of the edge
 257      * bandwidth
 258      */
 259     protected void pollTxBitRates() {
 260         Map<NodeConnector, Pair<Edge, Set<Property>>> globalContainerEdges = edgeMap
 261                 .get(GlobalConstants.DEFAULT.toString());
 262         if (globalContainerEdges == null) {
 263             return;
 264         }
 265
 266         for (NodeConnector connector : globalContainerEdges.keySet()) {
 267             // Skip if node connector belongs to production switch
 268             if (connector.getType().equals(
 269                     NodeConnector.NodeConnectorIDType.PRODUCTION)) {
 270                 continue;
 271             }
 272
 273             // Get edge for which this node connector is head
 274             Pair<Edge, Set<Property>> props = this.edgeMap.get(
 275                     GlobalConstants.DEFAULT.toString()).get(connector);
 276             // On switch mgr restart the props get reset
 277             if (props == null) {
 278                 continue;
 279             }
 280             Set<Property> propSet = props.getRight();
 281             if (propSet == null) {
 282                 continue;
 283             }
 284
 285             float bw = 0;
 286             for (Property prop : propSet) {
 287                 if (prop instanceof Bandwidth) {
 288                     bw = ((Bandwidth) prop).getValue();
 289                     break;
 290                 }
 291             }
 292
 293             // Skip if agent did not provide a bandwidth info for the edge
 294             if (bw == 0) {
 295                 continue;
 296             }
 297
 298             // Compare bandwidth usage
 299             Long switchId = (Long) connector.getNode().getID();
 300             Short port = (Short) connector.getID();
 301             if (statsMgr != null) {
 302                 float rate = statsMgr.getTransmitRate(switchId, port);
 303                 if (rate > bwThresholdFactor * bw) {
 304                     if (!connectorsOverUtilized.contains(connector)) {
 305                         connectorsOverUtilized.add(connector);
 306                         this.bwUtilNotifyQ.add(new UtilizationUpdate(connector, UpdateType.ADDED));
 307                     }
 308                 } else {
 309                     if (connectorsOverUtilized.contains(connector)) {
 310                         connectorsOverUtilized.remove(connector);
 311                         this.bwUtilNotifyQ.add(new UtilizationUpdate(connector, UpdateType.REMOVED));
 312                     }
 313                 }
 314             }
 315         }
 316
 317     }
 318
 319     /**
 320      * Function called by the dependency manager when at least one dependency
 321      * become unsatisfied or when the component is shutting down because for
 322      * example bundle is being stopped.
 323      *
 324      */
 325     void destroy() {
 326         logger.trace("DESTROY called!");
 327         notifyQ = null;
 328         notifyThread = null;
 329     }
 330
 331     /**
 332      * Function called by dependency manager after "init ()" is called and after
 333      * the services provided by the class are registered in the service registry
 334      *
 335      */
 336     void start() {
 337         logger.trace("START called!");
 338         notifyThread.start();
 339         bwUtilNotifyThread.start();
 340         ofPluginTopoBulkUpdate.start();
 341         pollTimer.scheduleAtFixedRate(txRatePoller, 10000, 5000);
 342     }
 343
 344     /**
 345      * Function called by the dependency manager before the services exported by
 346      * the component are unregistered, this will be followed by a "destroy ()"
 347      * calls
 348      *
 349      */
 350     void stop() {
 351         logger.trace("STOP called!");
 352         shuttingDown = true;
 353         notifyThread.interrupt();
 354     }
 355
 356     void setTopologyServiceShimListener(Map<?, ?> props,
 357             ITopologyServiceShimListener s) {
 358         if (props == null) {
 359             logger.error("Didn't receive the service properties");
 360             return;
 361         }
 362         String containerName = (String) props.get("containerName");
 363         if (containerName == null) {
 364             logger.error("containerName not supplied");
 365             return;
 366         }
 367         if ((this.topologyServiceShimListeners != null)
 368                 && !this.topologyServiceShimListeners
 369                         .containsKey(containerName)) {
 370             this.topologyServiceShimListeners.put(containerName, s);
 371             logger.trace("Added topologyServiceShimListener for container: {}",
 372                     containerName);
 373         }
 374     }
 375
 376     void unsetTopologyServiceShimListener(Map<?, ?> props,
 377             ITopologyServiceShimListener s) {
 378         if (props == null) {
 379             logger.error("Didn't receive the service properties");
 380             return;
 381         }
 382         String containerName = (String) props.get("containerName");
 383         if (containerName == null) {
 384             logger.error("containerName not supplied");
 385             return;
 386         }
 387         if ((this.topologyServiceShimListeners != null)
 388                 && this.topologyServiceShimListeners.containsKey(containerName)
 389                 && this.topologyServiceShimListeners.get(containerName).equals(
 390                         s)) {
 391             this.topologyServiceShimListeners.remove(containerName);
 392             logger.trace(
 393                     "Removed topologyServiceShimListener for container: {}",
 394                     containerName);
 395         }
 396     }
 397
 398     void setStatisticsManager(IOFStatisticsManager s) {
 399         this.statsMgr = s;
 400     }
 401
 402     void unsetStatisticsManager(IOFStatisticsManager s) {
 403         if (this.statsMgr == s) {
 404             this.statsMgr = null;
 405         }
 406     }
 407
 408     private void updateContainerMap(List<String> containers, NodeConnector p) {
 409         if (containers.isEmpty()) {
 410             // Do cleanup to reduce memory footprint if no
 411             // elements to be tracked
 412             this.containerMap.remove(p);
 413         } else {
 414             this.containerMap.put(p, containers);
 415         }
 416     }
 417
 418     /**
 419      * From a given edge map, retrieve the edge sourced by the port and update
 420      * the local cache in the container
 421      *
 422      * @param container
 423      *            the container name
 424      * @param nodeConnector
 425      *            the node connector
 426      * @param edges
 427      *            the given edge map
 428      * @return the found edge
 429      */
 430     private Edge addEdge(String container, NodeConnector nodeConnector,
 431             Map<NodeConnector, Pair<Edge, Set<Property>>> edges) {
 432         logger.debug("Search edge sourced by port {} in container {}", nodeConnector, container);
 433
 434         // Retrieve the associated edge
 435         Pair<Edge, Set<Property>> edgeProps = edges.get(nodeConnector);
 436         if (edgeProps == null) {
 437             logger.debug("edgePros is null for port {} in container {}", nodeConnector, container);
 438             return null;
 439         }
 440
 441         Edge edge = edgeProps.getLeft();
 442         if (edge == null) {
 443             logger.debug("edge is null for port {} in container {}", nodeConnector, container);
 444             return null;
 445         }
 446
 447         // Make sure the peer port is in the same container
 448         NodeConnector peerConnector = edge.getHeadNodeConnector();
 449         List<String> containers = this.containerMap.get(peerConnector);
 450         if ((containers == null) || !containers.contains(container)) {
 451             logger.debug("peer port {} of edge {} is not part of the container {}", new Object[] { peerConnector, edge,
 452                     container });
 453             return null;
 454         }
 455
 456         // Update the local cache
 457         updateLocalEdgeMap(container, edge, UpdateType.ADDED, edgeProps.getRight());
 458         logger.debug("Added edge {} to local cache in container {}", edge, container);
 459
 460         return edge;
 461     }
 462
 463     private void addNodeConnector(String container,
 464             NodeConnector nodeConnector) {
 465         // Use the global edge map for the newly added port in a container
 466         Map<NodeConnector, Pair<Edge, Set<Property>>> globalEdgeMap = edgeMap.get(GlobalConstants.DEFAULT
 467                 .toString());
 468         if (globalEdgeMap == null) {
 469             return;
 470         }
 471
 472         // Get the edge and update local cache in the container
 473         Edge edge1, edge2;
 474         edge1 = addEdge(container, nodeConnector, globalEdgeMap);
 475         if (edge1 == null) {
 476             return;
 477         }
 478
 479         // Get the edge in reverse direction and update local cache in the container
 480         NodeConnector peerConnector = edge1.getHeadNodeConnector();
 481         edge2 = addEdge(container, peerConnector, globalEdgeMap);
 482
 483         // Send notification upwards in one shot
 484         List<TopoEdgeUpdate> teuList = new ArrayList<TopoEdgeUpdate>();
 485         teuList.add(new TopoEdgeUpdate(edge1, null, UpdateType.ADDED));
 486         logger.debug("Notify edge1: {} in container {}", edge1, container);
 487         if (edge2 != null) {
 488             teuList.add(new TopoEdgeUpdate(edge2, null, UpdateType.ADDED));
 489             logger.debug("Notify edge2: {} in container {}", edge2, container);
 490         }
 491         notifyEdge(container, teuList);
 492     }
 493
 494     private void removeNodeConnector(String container,
 495             NodeConnector nodeConnector) {
 496         List<TopoEdgeUpdate> teuList = new ArrayList<TopoEdgeUpdate>();
 497         Map<NodeConnector, Pair<Edge, Set<Property>>> edgePropsMap = edgeMap
 498                 .get(container);
 499         if (edgePropsMap == null) {
 500             return;
 501         }
 502
 503         // Remove edge in one direction
 504         Pair<Edge, Set<Property>> edgeProps = edgePropsMap.get(nodeConnector);
 505         if (edgeProps == null) {
 506             return;
 507         }
 508         teuList.add(new TopoEdgeUpdate(edgeProps.getLeft(), null,
 509                 UpdateType.REMOVED));
 510
 511         // Remove edge in another direction
 512         edgeProps = edgePropsMap
 513                 .get(edgeProps.getLeft().getHeadNodeConnector());
 514         if (edgeProps == null) {
 515             return;
 516         }
 517         teuList.add(new TopoEdgeUpdate(edgeProps.getLeft(), null,
 518                 UpdateType.REMOVED));
 519
 520         // Update in one shot
 521         notifyEdge(container, teuList);
 522     }
 523
 524     /**
 525      * Update local cache and return true if it needs to notify upper layer
 526      * Topology listeners.
 527      *
 528      * @param container
 529      *            The network container
 530      * @param edge
 531      *            The edge
 532      * @param type
 533      *            The update type
 534      * @param props
 535      *            The edge properties
 536      * @return true if it needs to notify upper layer Topology listeners
 537      */
 538     private boolean updateLocalEdgeMap(String container, Edge edge,
 539             UpdateType type, Set<Property> props) {
 540         ConcurrentMap<NodeConnector, Pair<Edge, Set<Property>>> edgePropsMap = edgeMap
 541                 .get(container);
 542         NodeConnector src = edge.getTailNodeConnector();
 543         Pair<Edge, Set<Property>> edgeProps = new ImmutablePair<Edge, Set<Property>>(
 544                 edge, props);
 545         boolean rv = false;
 546
 547         switch (type) {
 548         case ADDED:
 549         case CHANGED:
 550             if (edgePropsMap == null) {
 551                 edgePropsMap = new ConcurrentHashMap<NodeConnector, Pair<Edge, Set<Property>>>();
 552                 rv = true;
 553             } else {
 554                 if (edgePropsMap.containsKey(src)
 555                         && edgePropsMap.get(src).equals(edgeProps)) {
 556                     // Entry already exists. No update.
 557                     rv = false;
 558                 } else {
 559                     rv = true;
 560                 }
 561             }
 562             if (rv) {
 563                 edgePropsMap.put(src, edgeProps);
 564                 edgeMap.put(container, edgePropsMap);
 565             }
 566             break;
 567         case REMOVED:
 568             if ((edgePropsMap != null) && edgePropsMap.containsKey(src)) {
 569                 edgePropsMap.remove(src);
 570                 if (edgePropsMap.isEmpty()) {
 571                     edgeMap.remove(container);
 572                 } else {
 573                     edgeMap.put(container, edgePropsMap);
 574                 }
 575                 rv = true;
 576             }
 577             break;
 578         default:
 579             logger.debug(
 580                     "notifyLocalEdgeMap: invalid {} for Edge {} in container {}",
 581                     new Object[] { type.getName(), edge, container });
 582         }
 583
 584         if (rv) {
 585             logger.debug(
 586                     "notifyLocalEdgeMap: {} for Edge {} in container {}",
 587                     new Object[] { type.getName(), edge, container });
 588         }
 589
 590         return rv;
 591     }
 592
 593     private void notifyEdge(String container, Edge edge, UpdateType type,
 594             Set<Property> props) {
 595         boolean notifyListeners;
 596
 597         // Update local cache
 598         notifyListeners = updateLocalEdgeMap(container, edge, type, props);
 599
 600         // Prepare to update TopologyService
 601         if (notifyListeners) {
 602             notifyQ.add(new NotifyEntry(container, new TopoEdgeUpdate(edge, props,
 603                     type)));
 604             logger.debug("notifyEdge: {} Edge {} in container {}",
 605                     new Object[] { type.getName(), edge, container });
 606         }
 607     }
 608
 609     private void notifyEdge(String container, List<TopoEdgeUpdate> etuList) {
 610         if (etuList == null) {
 611             return;
 612         }
 613
 614         Edge edge;
 615         UpdateType type;
 616         List<TopoEdgeUpdate> etuNotifyList = new ArrayList<TopoEdgeUpdate>();
 617         boolean notifyListeners = false, rv;
 618
 619         for (TopoEdgeUpdate etu : etuList) {
 620             edge = etu.getEdge();
 621             type = etu.getUpdateType();
 622
 623             // Update local cache
 624             rv = updateLocalEdgeMap(container, edge, type, etu.getProperty());
 625             if (rv) {
 626                 if (!notifyListeners) {
 627                     notifyListeners = true;
 628                 }
 629                 etuNotifyList.add(etu);
 630                 logger.debug(
 631                         "notifyEdge(TopoEdgeUpdate): {} Edge {} in container {}",
 632                         new Object[] { type.getName(), edge, container });
 633             }
 634         }
 635
 636         // Prepare to update TopologyService
 637         if (notifyListeners) {
 638             notifyQ.add(new NotifyEntry(container, etuNotifyList));
 639             logger.debug("notifyEdge(TopoEdgeUpdate): add notifyQ");
 640         }
 641     }
 642
 643     @Override
 644     public void notifyEdge(Edge edge, UpdateType type, Set<Property> props) {
 645         if ((edge == null) || (type == null)) {
 646             return;
 647         }
 648
 649         // Notify default container
 650         notifyEdge(GlobalConstants.DEFAULT.toString(), edge, type, props);
 651
 652         // Notify the corresponding containers
 653         List<String> containers = getEdgeContainers(edge);
 654         if (containers != null) {
 655             for (String container : containers) {
 656                 notifyEdge(container, edge, type, props);
 657             }
 658         }
 659     }
 660
 661     /*
 662      * Return a list of containers the edge associated with
 663      */
 664     private List<String> getEdgeContainers(Edge edge) {
 665         NodeConnector src = edge.getTailNodeConnector(), dst = edge
 666                 .getHeadNodeConnector();
 667
 668         if (!src.getType().equals(NodeConnector.NodeConnectorIDType.PRODUCTION)) {
 669             /* Find the common containers for both ends */
 670             List<String> srcContainers = this.containerMap.get(src), dstContainers = this.containerMap
 671                     .get(dst), cmnContainers = null;
 672             if ((srcContainers != null) && (dstContainers != null)) {
 673                 cmnContainers = new ArrayList<String>(srcContainers);
 674                 cmnContainers.retainAll(dstContainers);
 675             }
 676             return cmnContainers;
 677         } else {
 678             /*
 679              * If the neighbor is part of a monitored production network, get
 680              * the containers that the edge port belongs to
 681              */
 682             return this.containerMap.get(dst);
 683         }
 684     }
 685
 686     @Override
 687     public void tagUpdated(String containerName, Node n, short oldTag,
 688             short newTag, UpdateType t) {
 689     }
 690
 691     @Override
 692     public void containerFlowUpdated(String containerName,
 693             ContainerFlow previousFlow, ContainerFlow currentFlow, UpdateType t) {
 694     }
 695
 696     @Override
 697     public void nodeConnectorUpdated(String containerName, NodeConnector p,
 698             UpdateType t) {
 699         if (this.containerMap == null) {
 700             logger.error("containerMap is NULL");
 701             return;
 702         }
 703         List<String> containers = this.containerMap.get(p);
 704         if (containers == null) {
 705             containers = new CopyOnWriteArrayList<String>();
 706         }
 707         switch (t) {
 708         case ADDED:
 709             if (!containers.contains(containerName)) {
 710                 containers.add(containerName);
 711                 updateContainerMap(containers, p);
 712                 addNodeConnector(containerName, p);
 713             }
 714             break;
 715         case REMOVED:
 716             if (containers.contains(containerName)) {
 717                 containers.remove(containerName);
 718                 updateContainerMap(containers, p);
 719                 removeNodeConnector(containerName, p);
 720             }
 721             break;
 722         case CHANGED:
 723             break;
 724         }
 725     }
 726
 727     @Override
 728     public void containerModeUpdated(UpdateType t) {
 729         // do nothing
 730     }
 731
 732     private void registerWithOSGIConsole() {
 733         BundleContext bundleContext = FrameworkUtil.getBundle(this.getClass())
 734                 .getBundleContext();
 735         bundleContext.registerService(CommandProvider.class.getName(), this,
 736                 null);
 737     }
 738
 739     @Override
 740     public String getHelp() {
 741         StringBuffer help = new StringBuffer();
 742         help.append("---Topology Service Shim---\n");
 743         help.append("\t pem [container]               - Print edgeMap entries");
 744         help.append(" for a given container\n");
 745         return help.toString();
 746     }
 747
 748     public void _pem(CommandInterpreter ci) {
 749         String container = ci.nextArgument();
 750         if (container == null) {
 751             container = GlobalConstants.DEFAULT.toString();
 752         }
 753
 754         ci.println("Container: " + container);
 755         ci.println("                             Edge                                          Bandwidth");
 756
 757         Map<NodeConnector, Pair<Edge, Set<Property>>> edgePropsMap = edgeMap
 758                 .get(container);
 759         if (edgePropsMap == null) {
 760             return;
 761         }
 762         int count = 0;
 763         for (Pair<Edge, Set<Property>> edgeProps : edgePropsMap.values()) {
 764             if (edgeProps == null) {
 765                 continue;
 766             }
 767
 768             long bw = 0;
 769             Set<Property> props = edgeProps.getRight();
 770             if (props != null) {
 771                 for (Property prop : props) {
 772                     if (prop.getName().equals(Bandwidth.BandwidthPropName)) {
 773                         bw = ((Bandwidth) prop).getValue();
 774                     }
 775                 }
 776             }
 777             count++;
 778             ci.println(edgeProps.getLeft() + "          " + bw);
 779         }
 780         ci.println("Total number of Edges: " + count);
 781     }
 782
 783     public void _bwfactor(CommandInterpreter ci) {
 784         String factorString = ci.nextArgument();
 785         if (factorString == null) {
 786             ci.println("Bw threshold: " + this.bwThresholdFactor);
 787             ci.println("Insert a non null bw threshold");
 788             return;
 789         }
 790         bwThresholdFactor = Float.parseFloat(factorString);
 791         ci.println("New Bw threshold: " + this.bwThresholdFactor);
 792     }
 793
 794     /**
 795      * This method will trigger topology updates to be sent toward SAL. SAL then
 796      * pushes the updates to ALL the applications that have registered as
 797      * listeners for this service. SAL has no way of knowing which application
 798      * requested for the refresh.
 799      *
 800      * As an example of this case, is stopping and starting the Topology
 801      * Manager. When the topology Manager is stopped, and restarted, it will no
 802      * longer have the latest topology. Hence, a request is sent here.
 803      *
 804      * @param containerName
 805      * @return void
 806      */
 807     @Override
 808     public void requestRefresh(String containerName) {
 809         // wake up a bulk update thread and exit
 810         // the thread will execute the bulkUpdate()
 811         bulkNotifyQ.add(containerName);
 812     }
 813
 814     /**
 815      * Retrieve the edges for a given container
 816      *
 817      * @param containerName
 818      *            the container name
 819      * @return the edges and their properties
 820      */
 821     private Collection<Pair<Edge, Set<Property>>> getEdgeProps(String containerName) {
 822         Map<NodeConnector, Pair<Edge, Set<Property>>> edgePropMap = null;
 823         edgePropMap = edgeMap.get(containerName);
 824         if (edgePropMap == null) {
 825             return null;
 826         }
 827         return edgePropMap.values();
 828     }
 829
 830     /**
 831      * Reading the current topology database, the method will replay all the
 832      * edge updates for the ITopologyServiceShimListener instance in the given
 833      * container, which will in turn publish them toward SAL.
 834      *
 835      * @param containerName
 836      *            the container name
 837      */
 838     private void TopologyBulkUpdate(String containerName) {
 839         Collection<Pair<Edge, Set<Property>>> edgeProps = null;
 840
 841         logger.debug("Try bulk update for container:{}", containerName);
 842         edgeProps = getEdgeProps(containerName);
 843         if (edgeProps == null) {
 844             logger.debug("No edges known for container:{}", containerName);
 845             return;
 846         }
 847         ITopologyServiceShimListener topologServiceShimListener = topologyServiceShimListeners
 848                 .get(containerName);
 849         if (topologServiceShimListener == null) {
 850             logger.debug("No topology service shim listener for container:{}",
 851                     containerName);
 852             return;
 853         }
 854         int i = 0;
 855         List<TopoEdgeUpdate> teuList = new ArrayList<TopoEdgeUpdate>();
 856         for (Pair<Edge, Set<Property>> edgeProp : edgeProps) {
 857             if (edgeProp != null) {
 858                 i++;
 859                 teuList.add(new TopoEdgeUpdate(edgeProp.getLeft(), edgeProp
 860                         .getRight(), UpdateType.ADDED));
 861                 logger.trace("Add edge {}", edgeProp.getLeft());
 862             }
 863         }
 864         if (i > 0) {
 865             topologServiceShimListener.edgeUpdate(teuList);
 866         }
 867         logger.debug("Sent {} updates", i);
 868     }
 869
 870     @Override
 871     public void updateNode(Node node, UpdateType type, Set<Property> props) {
 872     }
 873
 874     @Override
 875     public void updateNodeConnector(NodeConnector nodeConnector,
 876             UpdateType type, Set<Property> props) {
 877         List<String> containers = new ArrayList<String>();
 878         List<String> conList = this.containerMap.get(nodeConnector);
 879
 880         containers.add(GlobalConstants.DEFAULT.toString());
 881         if (conList != null) {
 882             containers.addAll(conList);
 883         }
 884
 885         switch (type) {
 886         case ADDED:
 887             break;
 888         case CHANGED:
 889             if (props == null) {
 890                 break;
 891             }
 892
 893             boolean rmEdge = false;
 894             for (Property prop : props) {
 895                 if (((prop instanceof Config) && (((Config) prop).getValue() != Config.ADMIN_UP))
 896                         || ((prop instanceof State) && (((State) prop)
 897                                 .getValue() != State.EDGE_UP))) {
 898                     /*
 899                      * If port admin down or link down, remove the edges
 900                      * associated with the port
 901                      */
 902                     rmEdge = true;
 903                     break;
 904                 }
 905             }
 906
 907             if (rmEdge) {
 908                 for (String cName : containers) {
 909                     removeNodeConnector(cName, nodeConnector);
 910                 }
 911             }
 912             break;
 913         case REMOVED:
 914             for (String cName : containers) {
 915                 removeNodeConnector(cName, nodeConnector);
 916             }
 917             break;
 918         default:
 919             break;
 920         }
 921     }
 922
 923     @Override
 924     public void containerCreate(String containerName) {
 925         // do nothing
 926     }
 927
 928     @Override
 929     public void containerDestroy(String containerName) {
 930         Set<NodeConnector> removeNodeConnectorSet = new HashSet<NodeConnector>();
 931         for (Map.Entry<NodeConnector, List<String>> entry : containerMap.entrySet()) {
 932             List<String> ncContainers = entry.getValue();
 933             if (ncContainers.contains(containerName)) {
 934                 NodeConnector nodeConnector = entry.getKey();
 935                 removeNodeConnectorSet.add(nodeConnector);
 936             }
 937         }
 938         for (NodeConnector nodeConnector : removeNodeConnectorSet) {
 939             List<String> ncContainers = containerMap.get(nodeConnector);
 940             ncContainers.remove(containerName);
 941             if (ncContainers.isEmpty()) {
 942                 containerMap.remove(nodeConnector);
 943             }
 944         }
 945         edgeMap.remove(containerName);
 946     }
 947 }