opendaylight/topologymanager/implementation/src/main/java/org/opendaylight/controller/topologymanager/internal/TopologyManagerImpl.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.topologymanager.internal;
  10
  11 import java.io.FileNotFoundException;
  12 import java.io.IOException;
  13 import java.io.ObjectInputStream;
  14 import java.util.ArrayList;
  15 import java.util.Dictionary;
  16 import java.util.EnumSet;
  17 import java.util.HashMap;
  18 import java.util.HashSet;
  19 import java.util.Iterator;
  20 import java.util.LinkedList;
  21 import java.util.List;
  22 import java.util.Map;
  23 import java.util.Set;
  24 import java.util.concurrent.BlockingQueue;
  25 import java.util.concurrent.ConcurrentHashMap;
  26 import java.util.concurrent.ConcurrentMap;
  27 import java.util.concurrent.CopyOnWriteArraySet;
  28 import java.util.concurrent.LinkedBlockingQueue;
  29
  30 import org.apache.commons.lang3.tuple.ImmutablePair;
  31 import org.apache.felix.dm.Component;
  32 import org.eclipse.osgi.framework.console.CommandInterpreter;
  33 import org.eclipse.osgi.framework.console.CommandProvider;
  34 import org.opendaylight.controller.clustering.services.CacheConfigException;
  35 import org.opendaylight.controller.clustering.services.CacheExistException;
  36 import org.opendaylight.controller.clustering.services.ICacheUpdateAware;
  37 import org.opendaylight.controller.clustering.services.IClusterContainerServices;
  38 import org.opendaylight.controller.clustering.services.IClusterServices;
  39 import org.opendaylight.controller.configuration.ConfigurationObject;
  40 import org.opendaylight.controller.configuration.IConfigurationContainerAware;
  41 import org.opendaylight.controller.configuration.IConfigurationContainerService;
  42 import org.opendaylight.controller.sal.core.Edge;
  43 import org.opendaylight.controller.sal.core.Host;
  44 import org.opendaylight.controller.sal.core.Node;
  45 import org.opendaylight.controller.sal.core.NodeConnector;
  46 import org.opendaylight.controller.sal.core.Property;
  47 import org.opendaylight.controller.sal.core.TimeStamp;
  48 import org.opendaylight.controller.sal.core.UpdateType;
  49 import org.opendaylight.controller.sal.topology.IListenTopoUpdates;
  50 import org.opendaylight.controller.sal.topology.ITopologyService;
  51 import org.opendaylight.controller.sal.topology.TopoEdgeUpdate;
  52 import org.opendaylight.controller.sal.utils.IObjectReader;
  53 import org.opendaylight.controller.sal.utils.NodeConnectorCreator;
  54 import org.opendaylight.controller.sal.utils.Status;
  55 import org.opendaylight.controller.sal.utils.StatusCode;
  56 import org.opendaylight.controller.switchmanager.ISwitchManager;
  57 import org.opendaylight.controller.topologymanager.ITopologyManager;
  58 import org.opendaylight.controller.topologymanager.ITopologyManagerAware;
  59 import org.opendaylight.controller.topologymanager.ITopologyManagerClusterWideAware;
  60 import org.opendaylight.controller.topologymanager.TopologyUserLinkConfig;
  61 import org.osgi.framework.BundleContext;
  62 import org.osgi.framework.FrameworkUtil;
  63 import org.slf4j.Logger;
  64 import org.slf4j.LoggerFactory;
  65
  66 /**
  67  * The class describes TopologyManager which is the central repository of the
  68  * network topology. It provides service for applications to interact with
  69  * topology database and notifies all the listeners of topology changes.
  70  */
  71 public class TopologyManagerImpl implements
  72         ICacheUpdateAware<Object, Object>,
  73         ITopologyManager,
  74         IConfigurationContainerAware,
  75         IListenTopoUpdates,
  76         IObjectReader,
  77         CommandProvider {
  78     protected static final String TOPOEDGESDB = "topologymanager.edgesDB";
  79     protected static final String TOPOHOSTSDB = "topologymanager.hostsDB";
  80     protected static final String TOPONODECONNECTORDB = "topologymanager.nodeConnectorDB";
  81     protected static final String TOPOUSERLINKSDB = "topologymanager.userLinksDB";
  82     private static final String USER_LINKS_FILE_NAME = "userTopology.conf";
  83     private static final Logger log = LoggerFactory.getLogger(TopologyManagerImpl.class);
  84     private ITopologyService topoService;
  85     private IClusterContainerServices clusterContainerService;
  86     private IConfigurationContainerService configurationService;
  87     private ISwitchManager switchManager;
  88     // DB of all the Edges with properties which constitute our topology
  89     private ConcurrentMap<Edge, Set<Property>> edgesDB;
  90     // DB of all NodeConnector which are part of ISL Edges, meaning they
  91     // are connected to another NodeConnector on the other side of an ISL link.
  92     // NodeConnector of a Production Edge is not part of this DB.
  93     private ConcurrentMap<NodeConnector, Set<Property>> nodeConnectorsDB;
  94     // DB of all the NodeConnectors with an Host attached to it
  95     private ConcurrentMap<NodeConnector, Set<ImmutablePair<Host, Set<Property>>>> hostsDB;
  96     // Topology Manager Aware listeners
  97     private Set<ITopologyManagerAware> topologyManagerAware = new CopyOnWriteArraySet<ITopologyManagerAware>();
  98     // Topology Manager Aware listeners - for clusterwide updates
  99     private Set<ITopologyManagerClusterWideAware> topologyManagerClusterWideAware =
 100             new CopyOnWriteArraySet<ITopologyManagerClusterWideAware>();
 101     private ConcurrentMap<String, TopologyUserLinkConfig> userLinksDB;
 102     private BlockingQueue<TopoEdgeUpdate> notifyQ = new LinkedBlockingQueue<TopoEdgeUpdate>();
 103     private volatile Boolean shuttingDown = false;
 104     private Thread notifyThread;
 105
 106
 107     void nonClusterObjectCreate() {
 108         edgesDB = new ConcurrentHashMap<Edge, Set<Property>>();
 109         hostsDB = new ConcurrentHashMap<NodeConnector, Set<ImmutablePair<Host, Set<Property>>>>();
 110         nodeConnectorsDB = new ConcurrentHashMap<NodeConnector, Set<Property>>();
 111         userLinksDB = new ConcurrentHashMap<String, TopologyUserLinkConfig>();
 112     }
 113
 114     void setTopologyManagerAware(ITopologyManagerAware s) {
 115         if (this.topologyManagerAware != null) {
 116             log.debug("Adding ITopologyManagerAware: {}", s);
 117             this.topologyManagerAware.add(s);
 118         }
 119     }
 120
 121     void unsetTopologyManagerAware(ITopologyManagerAware s) {
 122         if (this.topologyManagerAware != null) {
 123             log.debug("Removing ITopologyManagerAware: {}", s);
 124             this.topologyManagerAware.remove(s);
 125         }
 126     }
 127
 128     void setTopologyManagerClusterWideAware(ITopologyManagerClusterWideAware s) {
 129         if (this.topologyManagerClusterWideAware != null) {
 130             log.debug("Adding ITopologyManagerClusterWideAware: {}", s);
 131             this.topologyManagerClusterWideAware.add(s);
 132         }
 133     }
 134
 135     void unsetTopologyManagerClusterWideAware(ITopologyManagerClusterWideAware s) {
 136         if (this.topologyManagerClusterWideAware != null) {
 137             log.debug("Removing ITopologyManagerClusterWideAware: {}", s);
 138             this.topologyManagerClusterWideAware.remove(s);
 139         }
 140     }
 141
 142     void setTopoService(ITopologyService s) {
 143         log.debug("Adding ITopologyService: {}", s);
 144         this.topoService = s;
 145     }
 146
 147     void unsetTopoService(ITopologyService s) {
 148         if (this.topoService == s) {
 149             log.debug("Removing ITopologyService: {}", s);
 150             this.topoService = null;
 151         }
 152     }
 153
 154     void setClusterContainerService(IClusterContainerServices s) {
 155         log.debug("Cluster Service set");
 156         this.clusterContainerService = s;
 157     }
 158
 159     void unsetClusterContainerService(IClusterContainerServices s) {
 160         if (this.clusterContainerService == s) {
 161             log.debug("Cluster Service removed!");
 162             this.clusterContainerService = null;
 163         }
 164     }
 165
 166     public void setConfigurationContainerService(IConfigurationContainerService service) {
 167         log.trace("Got configuration service set request {}", service);
 168         this.configurationService = service;
 169     }
 170
 171     public void unsetConfigurationContainerService(IConfigurationContainerService service) {
 172         log.trace("Got configuration service UNset request");
 173         this.configurationService = null;
 174     }
 175
 176     void setSwitchManager(ISwitchManager s) {
 177         log.debug("Adding ISwitchManager: {}", s);
 178         this.switchManager = s;
 179     }
 180
 181     void unsetSwitchManager(ISwitchManager s) {
 182         if (this.switchManager == s) {
 183             log.debug("Removing ISwitchManager: {}", s);
 184             this.switchManager = null;
 185         }
 186     }
 187
 188     /**
 189      * Function called by the dependency manager when all the required
 190      * dependencies are satisfied
 191      *
 192      */
 193     void init(Component c) {
 194         allocateCaches();
 195         retrieveCaches();
 196         String containerName = null;
 197         Dictionary<?, ?> props = c.getServiceProperties();
 198         if (props != null) {
 199             containerName = (String) props.get("containerName");
 200         } else {
 201             // In the Global instance case the containerName is empty
 202             containerName = "UNKNOWN";
 203         }
 204
 205         registerWithOSGIConsole();
 206         loadConfiguration();
 207
 208         // Restore the shuttingDown status on init of the component
 209         shuttingDown = false;
 210         notifyThread = new Thread(new TopologyNotify(notifyQ));
 211     }
 212
 213     @SuppressWarnings({ "unchecked" })
 214     private void allocateCaches() {
 215             this.edgesDB =
 216                     (ConcurrentMap<Edge, Set<Property>>) allocateCache(TOPOEDGESDB,EnumSet.of(IClusterServices.cacheMode.TRANSACTIONAL));
 217
 218             this.hostsDB =
 219                     (ConcurrentMap<NodeConnector, Set<ImmutablePair<Host, Set<Property>>>>) allocateCache(TOPOHOSTSDB, EnumSet.of(IClusterServices.cacheMode.TRANSACTIONAL));
 220
 221             this.nodeConnectorsDB =
 222                     (ConcurrentMap<NodeConnector, Set<Property>>) allocateCache(
 223                             TOPONODECONNECTORDB, EnumSet.of(IClusterServices.cacheMode.TRANSACTIONAL));
 224             this.userLinksDB =
 225                     (ConcurrentMap<String, TopologyUserLinkConfig>) allocateCache(
 226                             TOPOUSERLINKSDB, EnumSet.of(IClusterServices.cacheMode.TRANSACTIONAL));
 227     }
 228
 229     private ConcurrentMap<?, ?> allocateCache(String cacheName, Set<IClusterServices.cacheMode> cacheModes) {
 230         ConcurrentMap<?, ?> cache = null;
 231         try {
 232             cache = this.clusterContainerService.createCache(cacheName, cacheModes);
 233         } catch (CacheExistException e) {
 234             log.debug(cacheName + " cache already exists - destroy and recreate if needed");
 235         } catch (CacheConfigException e) {
 236             log.error(cacheName + " cache configuration invalid - check cache mode");
 237         }
 238         return cache;
 239     }
 240
 241     @SuppressWarnings({ "unchecked" })
 242     private void retrieveCaches() {
 243         if (this.clusterContainerService == null) {
 244             log.error("Cluster Services is null, can't retrieve caches.");
 245             return;
 246         }
 247
 248         this.edgesDB = (ConcurrentMap<Edge, Set<Property>>) this.clusterContainerService.getCache(TOPOEDGESDB);
 249         if (edgesDB == null) {
 250             log.error("Failed to get cache for " + TOPOEDGESDB);
 251         }
 252
 253         this.hostsDB =
 254                 (ConcurrentMap<NodeConnector, Set<ImmutablePair<Host, Set<Property>>>>) this.clusterContainerService.getCache(TOPOHOSTSDB);
 255         if (hostsDB == null) {
 256             log.error("Failed to get cache for " + TOPOHOSTSDB);
 257         }
 258
 259         this.nodeConnectorsDB =
 260                 (ConcurrentMap<NodeConnector, Set<Property>>) this.clusterContainerService.getCache(TOPONODECONNECTORDB);
 261         if (nodeConnectorsDB == null) {
 262             log.error("Failed to get cache for " + TOPONODECONNECTORDB);
 263         }
 264
 265         this.userLinksDB =
 266                 (ConcurrentMap<String, TopologyUserLinkConfig>) this.clusterContainerService.getCache(TOPOUSERLINKSDB);
 267         if (userLinksDB == null) {
 268             log.error("Failed to get cache for " + TOPOUSERLINKSDB);
 269         }
 270     }
 271
 272     /**
 273      * Function called after the topology manager has registered the service in
 274      * OSGi service registry.
 275      *
 276      */
 277     void started() {
 278         // Start the batcher thread for the cluster wide topology updates
 279         notifyThread.start();
 280         // SollicitRefresh MUST be called here else if called at init
 281         // time it may sollicit refresh too soon.
 282         log.debug("Sollicit topology refresh");
 283         topoService.sollicitRefresh();
 284     }
 285
 286     void stop() {
 287         shuttingDown = true;
 288         notifyThread.interrupt();
 289     }
 290
 291     /**
 292      * Function called by the dependency manager when at least one dependency
 293      * become unsatisfied or when the component is shutting down because for
 294      * example bundle is being stopped.
 295      *
 296      */
 297     void destroy() {
 298         notifyQ.clear();
 299         notifyThread = null;
 300     }
 301
 302     private void loadConfiguration() {
 303         for (ConfigurationObject conf : configurationService.retrieveConfiguration(this, USER_LINKS_FILE_NAME)) {
 304             addUserLink((TopologyUserLinkConfig) conf);
 305         }
 306     }
 307
 308     @Override
 309     public Status saveConfig() {
 310         return saveConfigInternal();
 311     }
 312
 313     public Status saveConfigInternal() {
 314         Status saveStatus = configurationService.persistConfiguration(
 315                 new ArrayList<ConfigurationObject>(userLinksDB.values()), USER_LINKS_FILE_NAME);
 316
 317         if (!saveStatus.isSuccess()) {
 318             return new Status(StatusCode.INTERNALERROR, "Topology save failed: " + saveStatus.getDescription());
 319         }
 320         return saveStatus;
 321     }
 322
 323     @Override
 324     public Map<Node, Set<Edge>> getNodeEdges() {
 325         if (this.edgesDB == null) {
 326             return null;
 327         }
 328
 329         Map<Node, Set<Edge>> res = new HashMap<Node, Set<Edge>>();
 330         for (Edge edge : this.edgesDB.keySet()) {
 331             // Lets analyze the tail
 332             Node node = edge.getTailNodeConnector().getNode();
 333             Set<Edge> nodeEdges = res.get(node);
 334             if (nodeEdges == null) {
 335                 nodeEdges = new HashSet<Edge>();
 336                 res.put(node, nodeEdges);
 337             }
 338             nodeEdges.add(edge);
 339
 340             // Lets analyze the head
 341             node = edge.getHeadNodeConnector().getNode();
 342             nodeEdges = res.get(node);
 343             if (nodeEdges == null) {
 344                 nodeEdges = new HashSet<Edge>();
 345                 res.put(node, nodeEdges);
 346             }
 347             nodeEdges.add(edge);
 348         }
 349
 350         return res;
 351     }
 352
 353     @Override
 354     public boolean isInternal(NodeConnector p) {
 355         if (this.nodeConnectorsDB == null) {
 356             return false;
 357         }
 358
 359         // This is an internal NodeConnector if is connected to
 360         // another Node i.e it's part of the nodeConnectorsDB
 361         return (this.nodeConnectorsDB.get(p) != null);
 362     }
 363
 364     /**
 365      * This method returns true if the edge is an ISL link.
 366      *
 367      * @param e
 368      *            The edge
 369      * @return true if it is an ISL link
 370      */
 371     public boolean isISLink(Edge e) {
 372         return (!isProductionLink(e));
 373     }
 374
 375     /**
 376      * This method returns true if the edge is a production link.
 377      *
 378      * @param e
 379      *            The edge
 380      * @return true if it is a production link
 381      */
 382     public boolean isProductionLink(Edge e) {
 383         return (e.getHeadNodeConnector().getType().equals(NodeConnector.NodeConnectorIDType.PRODUCTION)
 384                 || e.getTailNodeConnector().getType().equals(NodeConnector.NodeConnectorIDType.PRODUCTION));
 385     }
 386
 387     /**
 388      * This method cross checks the determination of nodeConnector type by Discovery Service
 389      * against the information in SwitchManager and updates it accordingly.
 390      * @param e
 391      *          The edge
 392      */
 393     private void crossCheckNodeConnectors(Edge e) {
 394         NodeConnector nc;
 395         if (e.getHeadNodeConnector().getType().equals(NodeConnector.NodeConnectorIDType.PRODUCTION)) {
 396             nc = updateNCTypeFromSwitchMgr(e.getHeadNodeConnector());
 397             if (nc != null) {
 398                 e.setHeadNodeConnector(nc);
 399             }
 400         }
 401         if (e.getTailNodeConnector().getType().equals(NodeConnector.NodeConnectorIDType.PRODUCTION)) {
 402             nc = updateNCTypeFromSwitchMgr(e.getTailNodeConnector());
 403             if (nc != null) {
 404                 e.setTailNodeConnector(nc);
 405             }
 406         }
 407     }
 408
 409     /**
 410      * A NodeConnector may have been categorized as of type Production by Discovery Service.
 411      * But at the time when this determination was made, only OF nodes were known to Discovery
 412      * Service. This method checks if the node of nodeConnector is known to SwitchManager. If
 413      * so, then it returns a new NodeConnector with correct type.
 414      *
 415      * @param nc
 416      *       NodeConnector as passed on in the edge
 417      * @return
 418      *       If Node of the NodeConnector is in SwitchManager, then return a new NodeConnector
 419      *       with correct type, null otherwise
 420      */
 421
 422     private NodeConnector updateNCTypeFromSwitchMgr(NodeConnector nc) {
 423
 424         for (Node node : switchManager.getNodes()) {
 425             String nodeName = node.getNodeIDString();
 426             log.trace("Switch Manager Node Name: {}, NodeConnector Node Name: {}", nodeName,
 427                     nc.getNode().getNodeIDString());
 428             if (nodeName.equals(nc.getNode().getNodeIDString())) {
 429                 NodeConnector nodeConnector = NodeConnectorCreator
 430                         .createNodeConnector(node.getType(), nc.getID(), node);
 431                 return nodeConnector;
 432             }
 433         }
 434         return null;
 435     }
 436
 437     /**
 438      * The Map returned is a copy of the current topology hence if the topology
 439      * changes the copy doesn't
 440      *
 441      * @return A Map representing the current topology expressed as edges of the
 442      *         network
 443      */
 444     @Override
 445     public Map<Edge, Set<Property>> getEdges() {
 446         if (this.edgesDB == null) {
 447             return null;
 448         }
 449
 450         Map<Edge, Set<Property>> edgeMap = new HashMap<Edge, Set<Property>>();
 451         Set<Property> props;
 452         for (Map.Entry<Edge, Set<Property>> edgeEntry : edgesDB.entrySet()) {
 453             // Sets of props are copied because the composition of
 454             // those properties could change with time
 455             props = new HashSet<Property>(edgeEntry.getValue());
 456             // We can simply reuse the key because the object is
 457             // immutable so doesn't really matter that we are
 458             // referencing the only owned by a different table, the
 459             // meaning is the same because doesn't change with time.
 460             edgeMap.put(edgeEntry.getKey(), props);
 461         }
 462
 463         return edgeMap;
 464     }
 465
 466     @Override
 467     public Set<NodeConnector> getNodeConnectorWithHost() {
 468         if (this.hostsDB == null) {
 469             return null;
 470         }
 471
 472         return (new HashSet<NodeConnector>(this.hostsDB.keySet()));
 473     }
 474
 475     @Override
 476     public Map<Node, Set<NodeConnector>> getNodesWithNodeConnectorHost() {
 477         if (this.hostsDB == null) {
 478             return null;
 479         }
 480         HashMap<Node, Set<NodeConnector>> res = new HashMap<Node, Set<NodeConnector>>();
 481         Node node;
 482         Set<NodeConnector> portSet;
 483         for (NodeConnector nc : this.hostsDB.keySet()) {
 484             node = nc.getNode();
 485             portSet = res.get(node);
 486             if (portSet == null) {
 487                 // Create the HashSet if null
 488                 portSet = new HashSet<NodeConnector>();
 489                 res.put(node, portSet);
 490             }
 491
 492             // Keep updating the HashSet, given this is not a
 493             // clustered map we can just update the set without
 494             // worrying to update the hashmap.
 495             portSet.add(nc);
 496         }
 497
 498         return (res);
 499     }
 500
 501     @Override
 502     public Host getHostAttachedToNodeConnector(NodeConnector port) {
 503         List<Host> hosts = getHostsAttachedToNodeConnector(port);
 504         if(hosts != null && !hosts.isEmpty()){
 505             return hosts.get(0);
 506         }
 507         return null;
 508     }
 509
 510     @Override
 511     public List<Host> getHostsAttachedToNodeConnector(NodeConnector p) {
 512         Set<ImmutablePair<Host, Set<Property>>> hosts;
 513         if (this.hostsDB == null || (hosts = this.hostsDB.get(p)) == null) {
 514             return null;
 515         }
 516         // create a list of hosts
 517         List<Host> retHosts = new LinkedList<Host>();
 518         for(ImmutablePair<Host, Set<Property>> host : hosts) {
 519             retHosts.add(host.getLeft());
 520         }
 521         return retHosts;
 522     }
 523
 524     @Override
 525     public synchronized void updateHostLink(NodeConnector port, Host h, UpdateType t, Set<Property> props) {
 526
 527         // Clone the property set in case non null else just
 528         // create an empty one. Caches allocated via infinispan
 529         // don't allow null values
 530         if (props == null) {
 531             props = new HashSet<Property>();
 532         } else {
 533             props = new HashSet<Property>(props);
 534         }
 535         ImmutablePair<Host, Set<Property>> thisHost = new ImmutablePair<Host, Set<Property>>(h, props);
 536
 537         // get the host list
 538         Set<ImmutablePair<Host, Set<Property>>> hostSet = this.hostsDB.get(port);
 539         if(hostSet == null) {
 540             hostSet = new HashSet<ImmutablePair<Host, Set<Property>>>();
 541         }
 542         switch (t) {
 543         case ADDED:
 544         case CHANGED:
 545             hostSet.add(thisHost);
 546             this.hostsDB.put(port, hostSet);
 547             break;
 548         case REMOVED:
 549             hostSet.remove(thisHost);
 550             if(hostSet.isEmpty()) {
 551                 //remove only if hasn't been concurrently modified
 552                 this.hostsDB.remove(port, hostSet);
 553             } else {
 554                 this.hostsDB.put(port, hostSet);
 555             }
 556             break;
 557         }
 558     }
 559
 560     private boolean headNodeConnectorExist(Edge e) {
 561         /*
 562          * Only check the head end point which is supposed to be part of a
 563          * network node we control (present in our inventory). If we checked the
 564          * tail end point as well, we would not store the edges that connect to
 565          * a non sdn enable port on a non sdn capable production switch. We want
 566          * to be able to see these switches on the topology.
 567          */
 568         NodeConnector head = e.getHeadNodeConnector();
 569         return (switchManager.doesNodeConnectorExist(head));
 570     }
 571
 572     private TopoEdgeUpdate edgeUpdate(Edge e, UpdateType type, Set<Property> props) {
 573         switch (type) {
 574         case ADDED:
 575
 576
 577             if (this.edgesDB.containsKey(e)) {
 578                 // Avoid redundant updates (e.g. cluster switch-over) as notifications trigger expensive tasks
 579                 log.trace("Skipping redundant edge addition: {}", e);
 580                 return null;
 581             }
 582
 583             // Make sure the props are non-null or create a copy
 584             if (props == null) {
 585                 props = new HashSet<Property>();
 586             } else {
 587                 props = new HashSet<Property>(props);
 588             }
 589
 590
 591             // Ensure that head node connector exists
 592             if (!headNodeConnectorExist(e)) {
 593                 log.warn("Ignore edge that contains invalid node connector: {}", e);
 594                 return null;
 595             }
 596
 597             // Check if nodeConnectors of the edge were correctly categorized
 598             // by protocol plugin
 599             crossCheckNodeConnectors(e);
 600
 601             // Now make sure there is the creation timestamp for the
 602             // edge, if not there, stamp with the first update
 603             boolean found_create = false;
 604             for (Property prop : props) {
 605                 if (prop instanceof TimeStamp) {
 606                     TimeStamp t = (TimeStamp) prop;
 607                     if (t.getTimeStampName().equals("creation")) {
 608                         found_create = true;
 609                         break;
 610                     }
 611                 }
 612             }
 613
 614             if (!found_create) {
 615                 TimeStamp t = new TimeStamp(System.currentTimeMillis(), "creation");
 616                 props.add(t);
 617             }
 618
 619             // Now add this in the database eventually overriding
 620             // something that may have been already existing
 621             this.edgesDB.put(e, props);
 622
 623             // Now populate the DB of NodeConnectors
 624             // NOTE WELL: properties are empty sets, not really needed
 625             // for now.
 626             // The DB only contains ISL ports
 627             if (isISLink(e)) {
 628                 this.nodeConnectorsDB.put(e.getHeadNodeConnector(), new HashSet<Property>(1));
 629                 this.nodeConnectorsDB.put(e.getTailNodeConnector(), new HashSet<Property>(1));
 630             }
 631             log.trace("Edge {}  {}", e.toString(), type.name());
 632             break;
 633         case REMOVED:
 634             // Now remove the edge from edgesDB
 635             this.edgesDB.remove(e);
 636
 637             // Now lets update the NodeConnectors DB, the assumption
 638             // here is that two NodeConnector are exclusively
 639             // connected by 1 and only 1 edge, this is reasonable in
 640             // the same plug (virtual of phisical) we can assume two
 641             // cables won't be plugged. This could break only in case
 642             // of devices in the middle that acts as hubs, but it
 643             // should be safe to assume that won't happen.
 644             this.nodeConnectorsDB.remove(e.getHeadNodeConnector());
 645             this.nodeConnectorsDB.remove(e.getTailNodeConnector());
 646             log.trace("Edge {}  {}", e.toString(), type.name());
 647             break;
 648         case CHANGED:
 649             Set<Property> oldProps = this.edgesDB.get(e);
 650
 651             // When property(s) changes lets make sure we can change it
 652             // all except the creation time stamp because that should
 653             // be set only when the edge is created
 654             TimeStamp timeStamp = null;
 655             for (Property prop : oldProps) {
 656                 if (prop instanceof TimeStamp) {
 657                     TimeStamp tsProp = (TimeStamp) prop;
 658                     if (tsProp.getTimeStampName().equals("creation")) {
 659                         timeStamp = tsProp;
 660                         break;
 661                     }
 662                 }
 663             }
 664
 665             // Now lets make sure new properties are non-null
 666             if (props == null) {
 667                 props = new HashSet<Property>();
 668             } else {
 669                 // Copy the set so noone is going to change the content
 670                 props = new HashSet<Property>(props);
 671             }
 672
 673             // Now lets remove the creation property if exist in the
 674             // new props
 675             for (Iterator<Property> i = props.iterator(); i.hasNext();) {
 676                 Property prop = i.next();
 677                 if (prop instanceof TimeStamp) {
 678                     TimeStamp t = (TimeStamp) prop;
 679                     if (t.getTimeStampName().equals("creation")) {
 680                         i.remove();
 681                         break;
 682                     }
 683                 }
 684             }
 685
 686             // Now lets add the creation timestamp in it
 687             if (timeStamp != null) {
 688                 props.add(timeStamp);
 689             }
 690
 691             // Finally update
 692             this.edgesDB.put(e, props);
 693             log.trace("Edge {}  {}", e.toString(), type.name());
 694             break;
 695         }
 696         return new TopoEdgeUpdate(e, props, type);
 697     }
 698
 699     @Override
 700     public void edgeUpdate(List<TopoEdgeUpdate> topoedgeupdateList) {
 701         List<TopoEdgeUpdate> teuList = new ArrayList<TopoEdgeUpdate>();
 702         for (int i = 0; i < topoedgeupdateList.size(); i++) {
 703             Edge e = topoedgeupdateList.get(i).getEdge();
 704             Set<Property> p = topoedgeupdateList.get(i).getProperty();
 705             UpdateType type = topoedgeupdateList.get(i).getUpdateType();
 706             TopoEdgeUpdate teu = edgeUpdate(e, type, p);
 707             if (teu != null) {
 708                 teuList.add(teu);
 709             }
 710         }
 711
 712         if (!teuList.isEmpty()) {
 713             // Now update the listeners
 714             for (ITopologyManagerAware s : this.topologyManagerAware) {
 715                 try {
 716                     s.edgeUpdate(teuList);
 717                 } catch (Exception exc) {
 718                     log.error("Exception on edge update:", exc);
 719                 }
 720             }
 721         }
 722     }
 723
 724     private Edge getReverseLinkTuple(TopologyUserLinkConfig link) {
 725         TopologyUserLinkConfig rLink = new TopologyUserLinkConfig(
 726                 link.getName(), link.getDstNodeConnector(), link.getSrcNodeConnector());
 727         return getLinkTuple(rLink);
 728     }
 729
 730
 731     private Edge getLinkTuple(TopologyUserLinkConfig link) {
 732         NodeConnector srcNodeConnector = NodeConnector.fromString(link.getSrcNodeConnector());
 733         NodeConnector dstNodeConnector = NodeConnector.fromString(link.getDstNodeConnector());
 734         try {
 735             return new Edge(srcNodeConnector, dstNodeConnector);
 736         } catch (Exception e) {
 737             return null;
 738         }
 739     }
 740
 741     @Override
 742     public ConcurrentMap<String, TopologyUserLinkConfig> getUserLinks() {
 743         return new ConcurrentHashMap<String, TopologyUserLinkConfig>(userLinksDB);
 744     }
 745
 746     @Override
 747     public Status addUserLink(TopologyUserLinkConfig userLink) {
 748         if (!userLink.isValid()) {
 749             return new Status(StatusCode.BADREQUEST,
 750                     "User link configuration invalid.");
 751         }
 752         userLink.setStatus(TopologyUserLinkConfig.STATUS.LINKDOWN);
 753
 754         //Check if this link already configured
 755         //NOTE: infinispan cache doesn't support Map.containsValue()
 756         // (which is linear time in most ConcurrentMap impl anyway)
 757         for (TopologyUserLinkConfig existingLink : userLinksDB.values()) {
 758             if (existingLink.equals(userLink)) {
 759                 return new Status(StatusCode.CONFLICT, "Link configuration exists");
 760             }
 761         }
 762         //attempt put, if mapping for this key already existed return conflict
 763         if (userLinksDB.putIfAbsent(userLink.getName(), userLink) != null) {
 764             return new Status(StatusCode.CONFLICT, "Link with name : " + userLink.getName()
 765                     + " already exists. Please use another name");
 766         }
 767
 768         Edge linkTuple = getLinkTuple(userLink);
 769         if (linkTuple != null) {
 770             if (!isProductionLink(linkTuple)) {
 771                 TopoEdgeUpdate teu = edgeUpdate(linkTuple, UpdateType.ADDED,
 772                                                 new HashSet<Property>());
 773                 if (teu == null) {
 774                     userLinksDB.remove(userLink.getName());
 775                     return new Status(StatusCode.NOTFOUND,
 776                            "Link configuration contains invalid node connector: "
 777                            + userLink);
 778                 }
 779             }
 780
 781             linkTuple = getReverseLinkTuple(userLink);
 782             if (linkTuple != null) {
 783                 userLink.setStatus(TopologyUserLinkConfig.STATUS.SUCCESS);
 784                 if (!isProductionLink(linkTuple)) {
 785                     edgeUpdate(linkTuple, UpdateType.ADDED, new HashSet<Property>());
 786                 }
 787             }
 788         }
 789         return new Status(StatusCode.SUCCESS);
 790     }
 791
 792     @Override
 793     public Status deleteUserLink(String linkName) {
 794         if (linkName == null) {
 795             return new Status(StatusCode.BADREQUEST, "User link name cannot be null.");
 796         }
 797
 798         TopologyUserLinkConfig link = userLinksDB.remove(linkName);
 799         Edge linkTuple;
 800         if ((link != null) && ((linkTuple = getLinkTuple(link)) != null)) {
 801             if (! isProductionLink(linkTuple)) {
 802                 edgeUpdate(linkTuple, UpdateType.REMOVED, null);
 803             }
 804
 805             linkTuple = getReverseLinkTuple(link);
 806             if (! isProductionLink(linkTuple)) {
 807                 edgeUpdate(linkTuple, UpdateType.REMOVED, null);
 808             }
 809         }
 810         return new Status(StatusCode.SUCCESS);
 811     }
 812
 813     private void registerWithOSGIConsole() {
 814         BundleContext bundleContext = FrameworkUtil.getBundle(this.getClass())
 815                 .getBundleContext();
 816         bundleContext.registerService(CommandProvider.class.getName(), this,
 817                 null);
 818     }
 819
 820     @Override
 821     public String getHelp() {
 822         StringBuffer help = new StringBuffer();
 823         help.append("---Topology Manager---\n");
 824         help.append("\t addUserLink <name> <node connector string> <node connector string>\n");
 825         help.append("\t deleteUserLink <name>\n");
 826         help.append("\t printUserLink\n");
 827         help.append("\t printNodeEdges\n");
 828         return help.toString();
 829     }
 830
 831     public void _printUserLink(CommandInterpreter ci) {
 832         for (String name : this.userLinksDB.keySet()) {
 833             TopologyUserLinkConfig linkConfig = userLinksDB.get(name);
 834             ci.println("Name : " + name);
 835             ci.println(linkConfig);
 836             ci.println("Edge " + getLinkTuple(linkConfig));
 837             ci.println("Reverse Edge " + getReverseLinkTuple(linkConfig));
 838         }
 839     }
 840
 841     public void _addUserLink(CommandInterpreter ci) {
 842         String name = ci.nextArgument();
 843         if ((name == null)) {
 844             ci.println("Please enter a valid Name");
 845             return;
 846         }
 847
 848         String ncStr1 = ci.nextArgument();
 849         if (ncStr1 == null) {
 850             ci.println("Please enter two node connector strings");
 851             return;
 852         }
 853         String ncStr2 = ci.nextArgument();
 854         if (ncStr2 == null) {
 855             ci.println("Please enter second node connector string");
 856             return;
 857         }
 858
 859         NodeConnector nc1 = NodeConnector.fromString(ncStr1);
 860         if (nc1 == null) {
 861             ci.println("Invalid input node connector 1 string: " + ncStr1);
 862             return;
 863         }
 864         NodeConnector nc2 = NodeConnector.fromString(ncStr2);
 865         if (nc2 == null) {
 866             ci.println("Invalid input node connector 2 string: " + ncStr2);
 867             return;
 868         }
 869
 870         TopologyUserLinkConfig config = new TopologyUserLinkConfig(name, ncStr1, ncStr2);
 871         ci.println(this.addUserLink(config));
 872     }
 873
 874     public void _deleteUserLink(CommandInterpreter ci) {
 875         String name = ci.nextArgument();
 876         if ((name == null)) {
 877             ci.println("Please enter a valid Name");
 878             return;
 879         }
 880         this.deleteUserLink(name);
 881     }
 882
 883     public void _printNodeEdges(CommandInterpreter ci) {
 884         Map<Node, Set<Edge>> nodeEdges = getNodeEdges();
 885         if (nodeEdges == null) {
 886             return;
 887         }
 888         Set<Node> nodeSet = nodeEdges.keySet();
 889         if (nodeSet == null) {
 890             return;
 891         }
 892         ci.println("        Node                                         Edge");
 893         for (Node node : nodeSet) {
 894             Set<Edge> edgeSet = nodeEdges.get(node);
 895             if (edgeSet == null) {
 896                 continue;
 897             }
 898             for (Edge edge : edgeSet) {
 899                 ci.println(node + "             " + edge);
 900             }
 901         }
 902     }
 903
 904     @Override
 905     public Object readObject(ObjectInputStream ois)
 906             throws FileNotFoundException, IOException, ClassNotFoundException {
 907         return ois.readObject();
 908     }
 909
 910     @Override
 911     public Status saveConfiguration() {
 912         return saveConfig();
 913     }
 914
 915     @Override
 916     public void edgeOverUtilized(Edge edge) {
 917         log.warn("Link Utilization above normal: {}", edge);
 918     }
 919
 920     @Override
 921     public void edgeUtilBackToNormal(Edge edge) {
 922         log.warn("Link Utilization back to normal: {}", edge);
 923     }
 924
 925     private void edgeUpdateClusterWide(Edge e, UpdateType type, Set<Property> props, boolean isLocal) {
 926         TopoEdgeUpdate upd = new TopoEdgeUpdate(e, props, type);
 927         upd.setLocal(isLocal);
 928         notifyQ.add(upd);
 929     }
 930
 931     @Override
 932     public void entryCreated(final Object key, final String cacheName, final boolean originLocal) {
 933         if (cacheName.equals(TOPOEDGESDB)) {
 934             // This is the case of an Edge being added to the topology DB
 935             final Edge e = (Edge) key;
 936             log.trace("Edge {} CREATED isLocal:{}", e, originLocal);
 937             edgeUpdateClusterWide(e, UpdateType.ADDED, null, originLocal);
 938         }
 939     }
 940
 941     @Override
 942     public void entryUpdated(final Object key, final Object new_value, final String cacheName, final boolean originLocal) {
 943         if (cacheName.equals(TOPOEDGESDB)) {
 944             final Edge e = (Edge) key;
 945             log.trace("Edge {} UPDATED isLocal:{}", e, originLocal);
 946             final Set<Property> props = (Set<Property>) new_value;
 947             edgeUpdateClusterWide(e, UpdateType.CHANGED, props, originLocal);
 948         }
 949     }
 950
 951     @Override
 952     public void entryDeleted(final Object key, final String cacheName, final boolean originLocal) {
 953         if (cacheName.equals(TOPOEDGESDB)) {
 954             final Edge e = (Edge) key;
 955             log.trace("Edge {} DELETED isLocal:{}", e, originLocal);
 956             edgeUpdateClusterWide(e, UpdateType.REMOVED, null, originLocal);
 957         }
 958     }
 959
 960     class TopologyNotify implements Runnable {
 961         private final BlockingQueue<TopoEdgeUpdate> notifyQ;
 962         private TopoEdgeUpdate entry;
 963         private List<TopoEdgeUpdate> teuList = new ArrayList<TopoEdgeUpdate>();
 964         private boolean notifyListeners;
 965
 966         TopologyNotify(BlockingQueue<TopoEdgeUpdate> notifyQ) {
 967             this.notifyQ = notifyQ;
 968         }
 969
 970         @Override
 971         public void run() {
 972             while (true) {
 973                 try {
 974                     log.trace("New run of TopologyNotify");
 975                     notifyListeners = false;
 976                     // First we block waiting for an element to get in
 977                     entry = notifyQ.take();
 978                     // Then we drain the whole queue if elements are
 979                     // in it without getting into any blocking condition
 980                     for (; entry != null; entry = notifyQ.poll()) {
 981                         teuList.add(entry);
 982                         notifyListeners = true;
 983                     }
 984
 985                     // Notify listeners only if there were updates drained else
 986                     // give up
 987                     if (notifyListeners) {
 988                         log.trace("Notifier thread, notified a listener");
 989                         // Now update the listeners
 990                         for (ITopologyManagerClusterWideAware s : topologyManagerClusterWideAware) {
 991                             try {
 992                                 s.edgeUpdate(teuList);
 993                             } catch (Exception exc) {
 994                                 log.error("Exception on edge update:", exc);
 995                             }
 996                         }
 997                     }
 998                     teuList.clear();
 999
1000                     // Lets sleep for sometime to allow aggregation of event
1001                     Thread.sleep(100);
1002                 } catch (InterruptedException e1) {
1003                     if (shuttingDown) {
1004                         return;
1005                     }
1006                     log.warn("TopologyNotify interrupted {}", e1.getMessage());
1007                 } catch (Exception e2) {
1008                     log.error("", e2);
1009                 }
1010             }
1011         }
1012     }
1013 }