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                 List<FlowOnNode> flows = reader.readAllFlows(node);
 177                 if (flows != null) {
 178                     flowStatistics.put(node, flows);
 179                 }
 180                 NodeDescription descr = reader.readDescription(node);
 181                 if (descr != null) {
 182                     descriptionStatistics.put(node, descr);
 183                 }
 184                 List<NodeTableStatistics> tableStats = reader.readNodeTable(node);
 185                 if (tableStats != null) {
 186                     tableStatistics.put(node, tableStats);
 187                 }
 188                 List<NodeConnectorStatistics> ncStats = reader.readNodeConnectors(node);
 189                 if (ncStats != null) {
 190                     nodeConnectorStatistics.put(node, ncStats);
 191                 }
 192             }
 193
 194         } else {
 195             log.trace("Failed to retrieve current statistics. Statistics will not be immediately available!");
 196         }
 197     }
 198
 199     /**
 200      * Function called by the dependency manager before the services
 201      * exported by the component are unregistered, this will be
 202      * followed by a "destroy ()" calls
 203      *
 204      */
 205     void stop() {
 206         log.debug("STOP called!");
 207     }
 208
 209     void setClusterContainerService(IClusterContainerServices s) {
 210         log.debug("Cluster Service set for Statistics Mgr");
 211         this.clusterContainerService = s;
 212     }
 213
 214     void unsetClusterContainerService(IClusterContainerServices s) {
 215         if (this.clusterContainerService == s) {
 216             log.debug("Cluster Service removed for Statistics Mgr!");
 217             this.clusterContainerService = null;
 218         }
 219     }
 220     void setIContainer(IContainer c){
 221         container = c;
 222     }
 223     public void unsetIContainer(IContainer s) {
 224         if (this.container == s) {
 225             this.container = null;
 226         }
 227     }
 228
 229     public void setReaderService(IReadService service) {
 230         log.debug("Got inventory service set request {}", service);
 231         this.reader = service;
 232     }
 233
 234     public void unsetReaderService(IReadService service) {
 235         log.debug("Got a service UNset request {}", service);
 236         this.reader = null;
 237     }
 238
 239     @Override
 240     public List<FlowOnNode> getFlows(Node node) {
 241         if (node == null) {
 242             return null;
 243         }
 244
 245         List<FlowOnNode> flowList = new ArrayList<FlowOnNode>();
 246         List<FlowOnNode> cachedList = flowStatistics.get(node);
 247         if (cachedList != null){
 248             flowList.addAll(cachedList);
 249         }
 250         return flowList;
 251     }
 252
 253     @Override
 254     public Map<Node, List<FlowOnNode>> getFlowStatisticsForFlowList(List<FlowEntry> flowList) {
 255         Map<Node, List<FlowOnNode>> statMapOutput = new HashMap<Node, List<FlowOnNode>>();
 256
 257         if (flowList == null || flowList.isEmpty()){
 258             return statMapOutput;
 259         }
 260
 261         Node node;
 262         // Index FlowEntries' flows by node so we don't traverse entire flow list for each flowEntry
 263         Map<Node, Set<Flow>> index = new HashMap<Node, Set<Flow>>();
 264         for (FlowEntry flowEntry : flowList) {
 265             node = flowEntry.getNode();
 266             Set<Flow> set = (index.containsKey(node) ? index.get(node) : new HashSet<Flow>());
 267             set.add(flowEntry.getFlow());
 268             index.put(node, set);
 269         }
 270
 271         // Iterate over flows per indexed node and add to output
 272         for (Entry<Node, Set<Flow>> indexEntry : index.entrySet()) {
 273             node = indexEntry.getKey();
 274             List<FlowOnNode> flowsPerNode = flowStatistics.get(node);
 275
 276             if (flowsPerNode != null && !flowsPerNode.isEmpty()){
 277                 List<FlowOnNode> filteredFlows = statMapOutput.containsKey(node) ?
 278                         statMapOutput.get(node) : new ArrayList<FlowOnNode>();
 279
 280                 for (FlowOnNode flowOnNode : flowsPerNode) {
 281                     if (indexEntry.getValue().contains(flowOnNode.getFlow())) {
 282                         filteredFlows.add(flowOnNode);
 283                     }
 284                 }
 285                 statMapOutput.put(node, filteredFlows);
 286             }
 287         }
 288         return statMapOutput;
 289     }
 290
 291     @Override
 292     public int getFlowsNumber(Node node) {
 293         List<FlowOnNode> l;
 294         if (node == null || (l = flowStatistics.get(node)) == null){
 295             return -1;
 296         }
 297         return l.size();
 298     }
 299
 300     @Override
 301     public NodeDescription getNodeDescription(Node node) {
 302         if (node == null){
 303             return null;
 304         }
 305         NodeDescription nd = descriptionStatistics.get(node);
 306         return nd != null? nd.clone() : null;
 307     }
 308
 309     @Override
 310     public NodeConnectorStatistics getNodeConnectorStatistics(NodeConnector nodeConnector) {
 311         if (nodeConnector == null){
 312             return null;
 313         }
 314
 315         List<NodeConnectorStatistics> statList = nodeConnectorStatistics.get(nodeConnector.getNode());
 316         if (statList != null){
 317             for (NodeConnectorStatistics stat : statList) {
 318                 if (stat.getNodeConnector().equals(nodeConnector)){
 319                     return stat;
 320                 }
 321             }
 322         }
 323         return null;
 324     }
 325
 326     @Override
 327     public List<NodeConnectorStatistics> getNodeConnectorStatistics(Node node) {
 328         if (node == null){
 329             return null;
 330         }
 331
 332         List<NodeConnectorStatistics> statList = new ArrayList<NodeConnectorStatistics>();
 333         List<NodeConnectorStatistics> cachedList = nodeConnectorStatistics.get(node);
 334         if (cachedList != null) {
 335             statList.addAll(cachedList);
 336         }
 337         return statList;
 338     }
 339
 340     @Override
 341     public NodeTableStatistics getNodeTableStatistics(NodeTable nodeTable) {
 342         if (nodeTable == null){
 343             return null;
 344         }
 345         List<NodeTableStatistics> statList = tableStatistics.get(nodeTable.getNode());
 346         if (statList != null){
 347             for (NodeTableStatistics stat : statList) {
 348                 if (stat.getNodeTable().getID().equals(nodeTable.getID())){
 349                     return stat;
 350                 }
 351             }
 352         }
 353         return null;
 354     }
 355
 356     @Override
 357     public List<NodeTableStatistics> getNodeTableStatistics(Node node){
 358         if (node == null){
 359             return null;
 360         }
 361         List<NodeTableStatistics> statList = new ArrayList<NodeTableStatistics>();
 362         List<NodeTableStatistics> cachedList = tableStatistics.get(node);
 363         if (cachedList != null) {
 364             statList.addAll(cachedList);
 365         }
 366         return statList;
 367     }
 368
 369     @Override
 370     public void nodeFlowStatisticsUpdated(Node node, List<FlowOnNode> flowStatsList) {
 371         List<FlowOnNode> currentStat = this.flowStatistics.get(node);
 372         // Update cache only if changed to avoid unnecessary cache sync operations
 373         if (! flowStatsList.equals(currentStat)){
 374             this.flowStatistics.put(node, flowStatsList);
 375         }
 376     }
 377
 378     @Override
 379     public void nodeConnectorStatisticsUpdated(Node node, List<NodeConnectorStatistics> ncStatsList) {
 380         List<NodeConnectorStatistics> currentStat = this.nodeConnectorStatistics.get(node);
 381         if (! ncStatsList.equals(currentStat)){
 382             this.nodeConnectorStatistics.put(node, ncStatsList);
 383         }
 384     }
 385
 386     @Override
 387     public void nodeTableStatisticsUpdated(Node node, List<NodeTableStatistics> tableStatsList) {
 388         List<NodeTableStatistics> currentStat = this.tableStatistics.get(node);
 389         if (! tableStatsList.equals(currentStat)) {
 390             this.tableStatistics.put(node, tableStatsList);
 391         }
 392     }
 393
 394     @Override
 395     public void descriptionStatisticsUpdated(Node node, NodeDescription nodeDescription) {
 396         NodeDescription currentDesc = this.descriptionStatistics.get(node);
 397         if (! nodeDescription.equals(currentDesc)){
 398             this.descriptionStatistics.put(node, nodeDescription);
 399         }
 400     }
 401
 402     @Override
 403     public void updateNode(Node node, UpdateType type, Set<Property> props) {
 404         // If node is removed, clean up stats mappings
 405         if (type == UpdateType.REMOVED) {
 406             flowStatistics.remove(node);
 407             nodeConnectorStatistics.remove(node);
 408             tableStatistics.remove(node);
 409             descriptionStatistics.remove(node);
 410         }
 411     }
 412
 413     @Override
 414     public void updateNodeConnector(NodeConnector nodeConnector, UpdateType type, Set<Property> props) {
 415         // Not interested in this update
 416     }
 417 }