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                         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 }

©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.