opendaylight/statisticsmanager/implementation/src/main/java/org/opendaylight/controller/statisticsmanager/internal/StatisticsManager.java

   1
   2 /*
   3  * Copyright (c) 2013 Cisco Systems, Inc. and others.  All rights reserved.
   4  *
   5  * This program and the accompanying materials are made available under the
   6  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   7  * and is available at http://www.eclipse.org/legal/epl-v10.html
   8  */
   9
  10 package org.opendaylight.controller.statisticsmanager.internal;
  11
  12 import java.util.ArrayList;
  13 import java.util.EnumSet;
  14 import java.util.HashMap;
  15 import java.util.HashSet;
  16 import java.util.List;
  17 import java.util.Map;
  18 import java.util.Map.Entry;
  19 import java.util.Set;
  20 import java.util.concurrent.ConcurrentHashMap;
  21 import java.util.concurrent.ConcurrentMap;
  22
  23 import org.opendaylight.controller.clustering.services.CacheConfigException;
  24 import org.opendaylight.controller.clustering.services.CacheExistException;
  25 import org.opendaylight.controller.clustering.services.IClusterContainerServices;
  26 import org.opendaylight.controller.clustering.services.IClusterServices;
  27 import org.opendaylight.controller.forwardingrulesmanager.FlowEntry;
  28 import org.opendaylight.controller.sal.core.IContainer;
  29 import org.opendaylight.controller.sal.core.Node;
  30 import org.opendaylight.controller.sal.core.NodeConnector;
  31 import org.opendaylight.controller.sal.core.NodeTable;
  32 import org.opendaylight.controller.sal.core.Property;
  33 import org.opendaylight.controller.sal.core.UpdateType;
  34 import org.opendaylight.controller.sal.flowprogrammer.Flow;
  35 import org.opendaylight.controller.sal.inventory.IListenInventoryUpdates;
  36 import org.opendaylight.controller.sal.reader.FlowOnNode;
  37 import org.opendaylight.controller.sal.reader.IReadService;
  38 import org.opendaylight.controller.sal.reader.IReadServiceListener;
  39 import org.opendaylight.controller.sal.reader.NodeConnectorStatistics;
  40 import org.opendaylight.controller.sal.reader.NodeDescription;
  41 import org.opendaylight.controller.sal.reader.NodeTableStatistics;
  42 import org.opendaylight.controller.sal.utils.ServiceHelper;
  43 import org.opendaylight.controller.statisticsmanager.IStatisticsManager;
  44 import org.opendaylight.controller.switchmanager.ISwitchManager;
  45 import org.slf4j.Logger;
  46 import org.slf4j.LoggerFactory;
  47
  48 /**
  49  * The class caches latest network nodes statistics as notified by reader
  50  * services and provides API to retrieve them.
  51  */
  52 public class StatisticsManager implements IStatisticsManager, IReadServiceListener, IListenInventoryUpdates {
  53     private static final Logger log = LoggerFactory.getLogger(StatisticsManager.class);
  54     private IContainer container;
  55     private IClusterContainerServices clusterContainerService;
  56     private IReadService reader;
  57     //statistics caches
  58     private ConcurrentMap<Node, List<FlowOnNode>> flowStatistics;
  59     private ConcurrentMap<Node, List<NodeConnectorStatistics>> nodeConnectorStatistics;
  60     private ConcurrentMap<Node, List<NodeTableStatistics>> tableStatistics;
  61     private ConcurrentMap<Node, NodeDescription> descriptionStatistics;
  62
  63     private void nonClusterObjectCreate() {
  64         flowStatistics = new ConcurrentHashMap<Node, List<FlowOnNode>>();
  65         nodeConnectorStatistics = new ConcurrentHashMap<Node, List<NodeConnectorStatistics>>();
  66         tableStatistics = new ConcurrentHashMap<Node, List<NodeTableStatistics>>();
  67         descriptionStatistics = new ConcurrentHashMap<Node, NodeDescription>();
  68     }
  69
  70     @SuppressWarnings("deprecation")
  71     private void allocateCaches() {
  72         if (clusterContainerService == null) {
  73             nonClusterObjectCreate();
  74             log.error("Clustering service unavailable. Allocated non-cluster statistics manager cache.");
  75             return;
  76         }
  77
  78         try {
  79             clusterContainerService.createCache("statisticsmanager.flowStatistics",
  80                     EnumSet.of(IClusterServices.cacheMode.NON_TRANSACTIONAL));
  81             clusterContainerService.createCache("statisticsmanager.nodeConnectorStatistics",
  82                     EnumSet.of(IClusterServices.cacheMode.NON_TRANSACTIONAL));
  83             clusterContainerService.createCache("statisticsmanager.tableStatistics",
  84                     EnumSet.of(IClusterServices.cacheMode.NON_TRANSACTIONAL));
  85             clusterContainerService.createCache("statisticsmanager.descriptionStatistics",
  86                     EnumSet.of(IClusterServices.cacheMode.NON_TRANSACTIONAL));
  87
  88         } catch (CacheConfigException cce) {
  89             log.error("Statistics cache configuration invalid - check cache mode");
  90         } catch (CacheExistException ce) {
  91             log.debug("Skipping statistics cache creation - already present");
  92         }
  93     }
  94     @SuppressWarnings({ "unchecked", "deprecation" })
  95     private void retrieveCaches() {
  96         ConcurrentMap<?, ?> map;
  97
  98         if (this.clusterContainerService == null) {
  99             log.warn("Can't retrieve statistics manager cache, Clustering service unavailable.");
 100             return;
 101         }
 102
 103         log.debug("Statistics Manager - retrieveCaches for Container {}", container);
 104
 105         map = clusterContainerService.getCache("statisticsmanager.flowStatistics");
 106         if (map != null) {
 107             this.flowStatistics = (ConcurrentMap<Node, List<FlowOnNode>>) map;
 108         } else {
 109             log.error("Cache allocation failed for statisticsmanager.flowStatistics in container {}", container.getName());
 110         }
 111
 112         map = clusterContainerService.getCache("statisticsmanager.nodeConnectorStatistics");
 113         if (map != null) {
 114             this.nodeConnectorStatistics = (ConcurrentMap<Node, List<NodeConnectorStatistics>>) map;
 115         } else {
 116             log.error("Cache allocation failed for statisticsmanager.nodeConnectorStatistics in container {}", container.getName());
 117         }
 118
 119         map = clusterContainerService.getCache("statisticsmanager.tableStatistics");
 120         if (map != null) {
 121             this.tableStatistics = (ConcurrentMap<Node, List<NodeTableStatistics>>) map;
 122         } else {
 123             log.error("Cache allocation failed for statisticsmanager.tableStatistics in container {}", container.getName());
 124         }
 125
 126         map = clusterContainerService.getCache("statisticsmanager.descriptionStatistics");
 127         if (map != null) {
 128             this.descriptionStatistics = (ConcurrentMap<Node, NodeDescription>) map;
 129         } else {
 130             log.error("Cache allocation failed for statisticsmanager.descriptionStatistics in container {}", container.getName());
 131         }
 132     }
 133
 134     /**
 135      * Function called by the dependency manager when all the required
 136      * dependencies are satisfied
 137      *
 138      */
 139     void init() {
 140         log.debug("INIT called!");
 141         allocateCaches();
 142         retrieveCaches();
 143
 144     }
 145
 146     /**
 147      * Function called by the dependency manager when at least one
 148      * dependency become unsatisfied or when the component is shutting
 149      * down because for example bundle is being stopped.
 150      *
 151      */
 152     void destroy() {
 153         log.debug("DESTROY called!");
 154     }
 155
 156     /**
 157      * Function called by dependency manager after "init ()" is called
 158      * and after the services provided by the class are registered in
 159      * the service registry
 160      *
 161      */
 162     void start() {
 163         log.debug("START called!");
 164     }
 165
 166     /**
 167      * Function called after registering the service in OSGi service registry.
 168      */
 169     void started(){
 170         // Retrieve current statistics so we don't have to wait for next refresh
 171         ISwitchManager switchManager = (ISwitchManager) ServiceHelper.getInstance(
 172                 ISwitchManager.class, container.getName(), this);
 173         if (reader != null && switchManager != null) {
 174             Set<Node> nodeSet = switchManager.getNodes();
 175             for (Node node : nodeSet) {
 176                 flowStatistics.put(node, reader.readAllFlows(node));
 177                 descriptionStatistics.put(node, reader.readDescription(node));
 178                 tableStatistics.put(node, reader.readNodeTable(node));
 179                 nodeConnectorStatistics.put(node, reader.readNodeConnectors(node));
 180             }
 181
 182         } else {
 183             log.trace("Failed to retrieve current statistics. Statistics will not be immediately available!");
 184         }
 185     }
 186
 187     /**
 188      * Function called by the dependency manager before the services
 189      * exported by the component are unregistered, this will be
 190      * followed by a "destroy ()" calls
 191      *
 192      */
 193     void stop() {
 194         log.debug("STOP called!");
 195     }
 196
 197     void setClusterContainerService(IClusterContainerServices s) {
 198         log.debug("Cluster Service set for Statistics Mgr");
 199         this.clusterContainerService = s;
 200     }
 201
 202     void unsetClusterContainerService(IClusterContainerServices s) {
 203         if (this.clusterContainerService == s) {
 204             log.debug("Cluster Service removed for Statistics Mgr!");
 205             this.clusterContainerService = null;
 206         }
 207     }
 208     void setIContainer(IContainer c){
 209         container = c;
 210     }
 211     public void unsetIContainer(IContainer s) {
 212         if (this.container == s) {
 213             this.container = null;
 214         }
 215     }
 216
 217     public void setReaderService(IReadService service) {
 218         log.debug("Got inventory service set request {}", service);
 219         this.reader = service;
 220     }
 221
 222     public void unsetReaderService(IReadService service) {
 223         log.debug("Got a service UNset request {}", service);
 224         this.reader = null;
 225     }
 226
 227     @Override
 228     public List<FlowOnNode> getFlows(Node node) {
 229         if (node == null) {
 230             return null;
 231         }
 232
 233         List<FlowOnNode> flowList = new ArrayList<FlowOnNode>();
 234         List<FlowOnNode> cachedList = flowStatistics.get(node);
 235         if (cachedList != null){
 236             flowList.addAll(cachedList);
 237         }
 238         return flowList;
 239     }
 240
 241     @Override
 242     public Map<Node, List<FlowOnNode>> getFlowStatisticsForFlowList(List<FlowEntry> flowList) {
 243         Map<Node, List<FlowOnNode>> statMapOutput = new HashMap<Node, List<FlowOnNode>>();
 244
 245         if (flowList == null || flowList.isEmpty()){
 246             return statMapOutput;
 247         }
 248
 249         Node node;
 250         // Index FlowEntries' flows by node so we don't traverse entire flow list for each flowEntry
 251         Map<Node, Set<Flow>> index = new HashMap<Node, Set<Flow>>();
 252         for (FlowEntry flowEntry : flowList) {
 253             node = flowEntry.getNode();
 254             Set<Flow> set = (index.containsKey(node) ? index.get(node) : new HashSet<Flow>());
 255             set.add(flowEntry.getFlow());
 256             index.put(node, set);
 257         }
 258
 259         // Iterate over flows per indexed node and add to output
 260         for (Entry<Node, Set<Flow>> indexEntry : index.entrySet()) {
 261             node = indexEntry.getKey();
 262             List<FlowOnNode> flowsPerNode = flowStatistics.get(node);
 263
 264             if (flowsPerNode != null && !flowsPerNode.isEmpty()){
 265                 List<FlowOnNode> filteredFlows = statMapOutput.containsKey(node) ?
 266                         statMapOutput.get(node) : new ArrayList<FlowOnNode>();
 267
 268                 for (FlowOnNode flowOnNode : flowsPerNode) {
 269                     if (indexEntry.getValue().contains(flowOnNode.getFlow())) {
 270                         filteredFlows.add(flowOnNode);
 271                     }
 272                 }
 273                 statMapOutput.put(node, filteredFlows);
 274             }
 275         }
 276         return statMapOutput;
 277     }
 278
 279     @Override
 280     public int getFlowsNumber(Node node) {
 281         List<FlowOnNode> l;
 282         if (node == null || (l = flowStatistics.get(node)) == null){
 283             return -1;
 284         }
 285         return l.size();
 286     }
 287
 288     @Override
 289     public NodeDescription getNodeDescription(Node node) {
 290         if (node == null){
 291             return null;
 292         }
 293         NodeDescription nd = descriptionStatistics.get(node);
 294         return nd != null? nd.clone() : null;
 295     }
 296
 297     @Override
 298     public NodeConnectorStatistics getNodeConnectorStatistics(NodeConnector nodeConnector) {
 299         if (nodeConnector == null){
 300             return null;
 301         }
 302
 303         List<NodeConnectorStatistics> statList = nodeConnectorStatistics.get(nodeConnector.getNode());
 304         if (statList != null){
 305             for (NodeConnectorStatistics stat : statList) {
 306                 if (stat.getNodeConnector().equals(nodeConnector)){
 307                     return stat;
 308                 }
 309             }
 310         }
 311         return null;
 312     }
 313
 314     @Override
 315     public List<NodeConnectorStatistics> getNodeConnectorStatistics(Node node) {
 316         if (node == null){
 317             return null;
 318         }
 319
 320         List<NodeConnectorStatistics> statList = new ArrayList<NodeConnectorStatistics>();
 321         List<NodeConnectorStatistics> cachedList = nodeConnectorStatistics.get(node);
 322         if (cachedList != null) {
 323             statList.addAll(cachedList);
 324         }
 325         return statList;
 326     }
 327
 328     @Override
 329     public NodeTableStatistics getNodeTableStatistics(NodeTable nodeTable) {
 330         if (nodeTable == null){
 331             return null;
 332         }
 333         List<NodeTableStatistics> statList = tableStatistics.get(nodeTable.getNode());
 334         if (statList != null){
 335             for (NodeTableStatistics stat : statList) {
 336                 if (stat.getNodeTable().getID().equals(nodeTable.getID())){
 337                     return stat;
 338                 }
 339             }
 340         }
 341         return null;
 342     }
 343
 344     @Override
 345     public List<NodeTableStatistics> getNodeTableStatistics(Node node){
 346         if (node == null){
 347             return null;
 348         }
 349         List<NodeTableStatistics> statList = new ArrayList<NodeTableStatistics>();
 350         List<NodeTableStatistics> cachedList = tableStatistics.get(node);
 351         if (cachedList != null) {
 352             statList.addAll(cachedList);
 353         }
 354         return statList;
 355     }
 356
 357     @Override
 358     public void nodeFlowStatisticsUpdated(Node node, List<FlowOnNode> flowStatsList) {
 359         List<FlowOnNode> currentStat = this.flowStatistics.get(node);
 360         // Update cache only if changed to avoid unnecessary cache sync operations
 361         if (! flowStatsList.equals(currentStat)){
 362             this.flowStatistics.put(node, flowStatsList);
 363         }
 364     }
 365
 366     @Override
 367     public void nodeConnectorStatisticsUpdated(Node node, List<NodeConnectorStatistics> ncStatsList) {
 368         List<NodeConnectorStatistics> currentStat = this.nodeConnectorStatistics.get(node);
 369         if (! ncStatsList.equals(currentStat)){
 370             this.nodeConnectorStatistics.put(node, ncStatsList);
 371         }
 372     }
 373
 374     @Override
 375     public void nodeTableStatisticsUpdated(Node node, List<NodeTableStatistics> tableStatsList) {
 376         List<NodeTableStatistics> currentStat = this.tableStatistics.get(node);
 377         if (! tableStatsList.equals(currentStat)) {
 378             this.tableStatistics.put(node, tableStatsList);
 379         }
 380     }
 381
 382     @Override
 383     public void descriptionStatisticsUpdated(Node node, NodeDescription nodeDescription) {
 384         NodeDescription currentDesc = this.descriptionStatistics.get(node);
 385         if (! nodeDescription.equals(currentDesc)){
 386             this.descriptionStatistics.put(node, nodeDescription);
 387         }
 388     }
 389
 390     @Override
 391     public void updateNode(Node node, UpdateType type, Set<Property> props) {
 392         // If node is removed, clean up stats mappings
 393         if (type == UpdateType.REMOVED) {
 394             flowStatistics.remove(node);
 395             nodeConnectorStatistics.remove(node);
 396             tableStatistics.remove(node);
 397             descriptionStatistics.remove(node);
 398         }
 399     }
 400
 401     @Override
 402     public void updateNodeConnector(NodeConnector nodeConnector, UpdateType type, Set<Property> props) {
 403         // Not interested in this update
 404     }
 405 }