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

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.