vpnservice/qosservice/impl/src/main/java/org/opendaylight/netvirt/qosservice/QosAlertManager.java

   1 /*
   2  * Copyright (c) 2017 Ericsson India Global Services Pvt Ltd. 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.netvirt.qosservice;
  10
  11
  12 import com.google.common.base.Optional;
  13 import com.google.common.util.concurrent.FutureCallback;
  14 import com.google.common.util.concurrent.Futures;
  15 import java.math.BigInteger;
  16 import java.util.List;
  17 import java.util.concurrent.ConcurrentHashMap;
  18 import java.util.concurrent.ExecutionException;
  19 import java.util.concurrent.Future;
  20 import javax.annotation.PostConstruct;
  21 import javax.annotation.PreDestroy;
  22 import javax.inject.Inject;
  23 import javax.inject.Singleton;
  24 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  25 import org.opendaylight.controller.md.sal.binding.api.WriteTransaction;
  26 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  27 import org.opendaylight.genius.interfacemanager.globals.IfmConstants;
  28 import org.opendaylight.genius.mdsalutil.MDSALUtil;
  29 import org.opendaylight.netvirt.neutronvpn.interfaces.INeutronVpnManager;
  30 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.yang.types.rev130715.Uuid;
  31 import org.opendaylight.yang.gen.v1.urn.opendaylight.direct.statistics.rev160511.GetNodeConnectorStatisticsInputBuilder;
  32 import org.opendaylight.yang.gen.v1.urn.opendaylight.direct.statistics.rev160511.GetNodeConnectorStatisticsOutput;
  33 import org.opendaylight.yang.gen.v1.urn.opendaylight.direct.statistics.rev160511.OpendaylightDirectStatisticsService;
  34 import org.opendaylight.yang.gen.v1.urn.opendaylight.genius.interfacemanager.rpcs.rev160406.OdlInterfaceRpcService;
  35 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeConnectorId;
  36 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
  37 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
  38 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.Nodes;
  39 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
  40 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
  41 import org.opendaylight.yang.gen.v1.urn.opendaylight.netvirt.qosalert.config.rev170301.QosalertConfig;
  42 import org.opendaylight.yang.gen.v1.urn.opendaylight.netvirt.qosalert.config.rev170301.QosalertConfigBuilder;
  43 import org.opendaylight.yang.gen.v1.urn.opendaylight.neutron.networks.rev150712.networks.attributes.networks.Network;
  44 import org.opendaylight.yang.gen.v1.urn.opendaylight.neutron.ports.rev150712.ports.attributes.ports.Port;
  45 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.node.connector.statistics.and.port.number.map.NodeConnectorStatisticsAndPortNumberMap;
  46 import org.opendaylight.yangtools.yang.binding.DataObject;
  47 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  48 import org.opendaylight.yangtools.yang.common.RpcResult;
  49 import org.slf4j.Logger;
  50 import org.slf4j.LoggerFactory;
  51
  52
  53 @Singleton
  54 public final class QosAlertManager implements Runnable {
  55
  56     private short threshold;
  57     private boolean alertEnabled;
  58     private int pollInterval;
  59     private final QosalertConfig defaultConfig;
  60     private boolean statsPollThreadStart;
  61     private static DataBroker dataBroker;
  62     private static OpendaylightDirectStatisticsService odlDirectStatisticsService;
  63     private static INeutronVpnManager neutronVpnManager;
  64     private Thread thread;
  65     private static OdlInterfaceRpcService odlInterfaceRpcService;
  66     private static ConcurrentHashMap<BigInteger, ConcurrentHashMap<String, QosAlertPortData>>
  67                                                          qosAlertDpnPortNumberMap = new ConcurrentHashMap<>();
  68     private static final Logger LOG = LoggerFactory.getLogger(QosAlertManager.class);
  69
  70     private static final FutureCallback<Void> DEFAULT_FUTURE_CALLBACK;
  71
  72     static {
  73         DEFAULT_FUTURE_CALLBACK = new FutureCallback<Void>() {
  74
  75             @Override
  76             public void onSuccess(Void result) {
  77                 LOG.info("Datastore operation completed successfully");
  78             }
  79
  80             @Override
  81             public void onFailure(Throwable error) {
  82                 LOG.error("Error in datastore operation {}", error);
  83             }
  84
  85         };
  86     }
  87
  88     @Inject
  89     public QosAlertManager(final DataBroker dataBroker,
  90                            final OpendaylightDirectStatisticsService odlDirectStatisticsService,
  91                            final QosalertConfig defaultConfig,
  92                            final OdlInterfaceRpcService odlInterfaceRpcService,
  93                            final INeutronVpnManager neutronVpnManager) {
  94
  95         LOG.info("{} created",  getClass().getSimpleName());
  96         this.dataBroker = dataBroker;
  97         this.odlDirectStatisticsService = odlDirectStatisticsService;
  98         this.odlInterfaceRpcService = odlInterfaceRpcService;
  99         this.neutronVpnManager =  neutronVpnManager;
 100         this.defaultConfig = defaultConfig;
 101         thread = null;
 102         LOG.info("QosAlert default config poll alertEnabled:{} threshold:{} pollInterval:{}",
 103                 defaultConfig.isQosAlertEnabled(), defaultConfig.getQosDropPacketThreshold(),
 104                 defaultConfig.getQosAlertPollInterval());
 105         getDefaultConfig();
 106     }
 107
 108     @PostConstruct
 109     public void init() {
 110         qosAlertDpnPortNumberMap.clear();
 111         QosAlertPortData.setAlertThreshold(threshold);
 112         statsPollThreadStart = true;
 113         startStatsPollThread();
 114         LOG.info("{} init done", getClass().getSimpleName());
 115     }
 116
 117     @PreDestroy
 118     public void close() {
 119         statsPollThreadStart = false;
 120         if (thread != null) {
 121             thread.interrupt();
 122         }
 123         LOG.info("{} close done", getClass().getSimpleName());
 124     }
 125
 126     public void setQosAlertOwner(boolean isOwner) {
 127         LOG.trace("qos alert set owner : {}", isOwner);
 128         statsPollThreadStart = isOwner;
 129         if (thread != null) {
 130             thread.interrupt();
 131         } else {
 132             startStatsPollThread();
 133         }
 134     }
 135
 136     @Override
 137     public void run() {
 138         LOG.info("Qos alert poll thread started");
 139
 140         while (statsPollThreadStart && alertEnabled) {
 141             LOG.debug("Thread loop polling :{} threshold:{} pollInterval:{}", alertEnabled, threshold,
 142                     pollInterval);
 143
 144             try {
 145                 pollDirectStatisticsForAllNodes();
 146                 Thread.sleep(pollInterval * 60 * 1000); // pollInterval in minutes
 147             } catch (final InterruptedException e) {
 148                 LOG.debug("Qos polling thread interrupted");
 149             }
 150         }
 151         thread = null;
 152         LOG.info("Qos alert poll thread stopped");
 153     }
 154
 155     private void startStatsPollThread() {
 156         if (statsPollThreadStart && alertEnabled && (thread == null)) {
 157             thread = new Thread(this);
 158             thread.setDaemon(true);
 159             thread.start();
 160         }
 161     }
 162
 163     private void getDefaultConfig() {
 164         alertEnabled = defaultConfig.isQosAlertEnabled();
 165         threshold = defaultConfig.getQosDropPacketThreshold();
 166         pollInterval = defaultConfig.getQosAlertPollInterval();
 167     }
 168
 169     public void setQosalertConfig(QosalertConfig config) {
 170
 171         LOG.info("New QoS alert config threshold:{} polling alertEnabled:{} interval:{}",
 172                 config.getQosDropPacketThreshold(), config.isQosAlertEnabled(),
 173                 config.getQosAlertPollInterval());
 174
 175         threshold = config.getQosDropPacketThreshold().shortValue();
 176         alertEnabled = config.isQosAlertEnabled().booleanValue();
 177         pollInterval = config.getQosAlertPollInterval();
 178
 179         QosAlertPortData.setAlertThreshold(threshold);
 180
 181         if (thread != null) {
 182             thread.interrupt();
 183         } else {
 184             startStatsPollThread();
 185         }
 186
 187     }
 188
 189     public void restoreDefaultConfig() {
 190         LOG.info("Restoring default configuration");
 191         getDefaultConfig();
 192         QosAlertPortData.setAlertThreshold(threshold);
 193         if (thread != null) {
 194             thread.interrupt();
 195         } else {
 196             startStatsPollThread();
 197         }
 198     }
 199
 200     public void setThreshold(short threshold) {
 201         LOG.info("setting threshold {} in config data store", threshold);
 202         this.threshold = threshold;
 203         writeConfigDataStore();
 204     }
 205
 206     public void setPollInterval(int pollInterval) {
 207         LOG.info("setting interval {} in config data store", pollInterval);
 208         this.pollInterval = pollInterval;
 209         writeConfigDataStore();
 210     }
 211
 212     public void setEnable(boolean alertEnabled) {
 213         LOG.info("setting QoS poll to {} in config data store", alertEnabled);
 214         this.alertEnabled = alertEnabled;
 215         writeConfigDataStore();
 216     }
 217
 218     public static void addToQosAlertCache(Port port) {
 219         LOG.trace("Adding port {} in cache", port.getUuid());
 220
 221         BigInteger dpnId = QosNeutronUtils.getDpnForInterface(odlInterfaceRpcService, port.getUuid().getValue());
 222
 223         if (dpnId.equals(BigInteger.ZERO)) {
 224             LOG.debug("DPN ID for port {} not found", port.getUuid());
 225             return;
 226         }
 227
 228         String portNumber = QosNeutronUtils.getPortNumberForInterface(odlInterfaceRpcService,
 229                                                                                  port.getUuid().getValue());
 230
 231         if (qosAlertDpnPortNumberMap.containsKey(dpnId)) {
 232             LOG.trace("Adding port {}  port number {} in DPN {}", port.getUuid(), portNumber, dpnId);
 233             qosAlertDpnPortNumberMap.get(dpnId).put(portNumber, new QosAlertPortData(port, neutronVpnManager));
 234         } else {
 235             LOG.trace("Adding DPN ID {} with port {} port number {}", dpnId, port.getUuid(), portNumber);
 236             ConcurrentHashMap<String, QosAlertPortData> portDataMap = new ConcurrentHashMap<>();
 237             portDataMap.put(portNumber, new QosAlertPortData(port, neutronVpnManager));
 238             qosAlertDpnPortNumberMap.put(dpnId, portDataMap);
 239         }
 240     }
 241
 242     public static void addToQosAlertCache(Network network) {
 243         LOG.trace("Adding network {} in cache", network.getUuid());
 244
 245         List<Uuid> subnetIds = QosNeutronUtils.getSubnetIdsFromNetworkId(dataBroker, network.getUuid());
 246
 247         if (subnetIds != null) {
 248             for (Uuid subnetId : subnetIds) {
 249                 List<Uuid> portIds = QosNeutronUtils.getPortIdsFromSubnetId(dataBroker, subnetId);
 250                 if (portIds != null) {
 251                     for (Uuid portId : portIds) {
 252                         Port port = neutronVpnManager.getNeutronPort(portId);
 253                         if (port != null) {
 254                             if (!QosNeutronUtils.portHasQosPolicy(neutronVpnManager, port)) {
 255                                 LOG.trace("Adding network {} port {} in cache", network.getUuid(), port.getUuid());
 256                                 addToQosAlertCache(port);
 257                             }
 258                         }
 259                     }
 260                 }
 261             }
 262         }
 263     }
 264
 265     public static void removeFromQosAlertCache(Port port) {
 266         LOG.trace("Removing port {} from cache", port.getUuid());
 267
 268         BigInteger dpnId = QosNeutronUtils.getDpnForInterface(odlInterfaceRpcService, port.getUuid().getValue());
 269
 270         if (dpnId.equals(BigInteger.ZERO)) {
 271             LOG.debug("DPN ID for port {} not found", port.getUuid());
 272             return;
 273         }
 274
 275         String portNumber = QosNeutronUtils.getPortNumberForInterface(odlInterfaceRpcService,
 276                                                                                   port.getUuid().getValue());
 277
 278         if (qosAlertDpnPortNumberMap.containsKey(dpnId)
 279                                          && qosAlertDpnPortNumberMap.get(dpnId).containsKey(portNumber)) {
 280             qosAlertDpnPortNumberMap.get(dpnId).remove(portNumber);
 281             LOG.trace("Removed DPN {} port {} port number {} from cache", dpnId, port.getUuid(), portNumber);
 282             if (qosAlertDpnPortNumberMap.get(dpnId).isEmpty()) {
 283                 LOG.trace("DPN {} empty. Removing from cache", dpnId);
 284                 qosAlertDpnPortNumberMap.remove(dpnId);
 285             }
 286         } else {
 287             LOG.trace("DPN {} port {} port number {} not found in cache", dpnId, port.getUuid(), portNumber);
 288         }
 289
 290     }
 291
 292     public static void removeFromQosAlertCache(NodeConnectorId nodeConnectorId) {
 293         LOG.trace("Removing node connector {} from cache", nodeConnectorId.getValue());
 294
 295         long nodeId = MDSALUtil.getDpnIdFromPortName(nodeConnectorId);
 296
 297         if (nodeId == -1) {
 298             LOG.debug("Node ID for node connector {} not found", nodeConnectorId.getValue());
 299             return;
 300         }
 301
 302         BigInteger dpnId = new BigInteger(String.valueOf(nodeId));
 303
 304         long portId = MDSALUtil.getOfPortNumberFromPortName(nodeConnectorId);
 305
 306         String portNumber = String.valueOf(portId);
 307
 308         if (qosAlertDpnPortNumberMap.containsKey(dpnId)
 309                              && qosAlertDpnPortNumberMap.get(dpnId).containsKey(portNumber)) {
 310             qosAlertDpnPortNumberMap.get(dpnId).remove(portNumber);
 311             LOG.trace("Removed DPN {} port number {} from cache", dpnId, portNumber);
 312         } else {
 313             LOG.trace("DPN {} port number {} not found in cache", dpnId, portNumber);
 314         }
 315
 316     }
 317
 318     public static void removeFromQosAlertCache(Network network) {
 319         LOG.trace("Removing network {} from cache", network.getUuid());
 320
 321         List<Uuid> subnetIds = QosNeutronUtils.getSubnetIdsFromNetworkId(dataBroker, network.getUuid());
 322
 323         if (subnetIds != null) {
 324             for (Uuid subnetId : subnetIds) {
 325                 List<Uuid> portIds = QosNeutronUtils.getPortIdsFromSubnetId(dataBroker, subnetId);
 326                 if (portIds != null) {
 327                     for (Uuid portId : portIds) {
 328                         Port port = neutronVpnManager.getNeutronPort(portId);
 329                         if (port != null) {
 330                             if (!QosNeutronUtils.portHasQosPolicy(neutronVpnManager, port)) {
 331                                 LOG.trace("Removing network {} port {} from cache", network.getUuid(), port.getUuid());
 332                                 removeFromQosAlertCache(port);
 333                             }
 334                         }
 335                     }
 336                 }
 337             }
 338         }
 339     }
 340
 341     private static <T extends DataObject> void asyncWrite(LogicalDatastoreType datastoreType,
 342                                                           InstanceIdentifier<T> path, T data, DataBroker broker,
 343                                                           FutureCallback<Void> callback) {
 344         WriteTransaction tx = broker.newWriteOnlyTransaction();
 345         tx.put(datastoreType, path, data, true);
 346         Futures.addCallback(tx.submit(), callback);
 347     }
 348
 349     private void writeConfigDataStore() {
 350
 351         InstanceIdentifier<QosalertConfig> path = InstanceIdentifier.builder(QosalertConfig.class).build();
 352
 353         QosalertConfig qosAlertConfig = new QosalertConfigBuilder()
 354                 .setQosDropPacketThreshold(threshold)
 355                 .setQosAlertEnabled(alertEnabled)
 356                 .setQosAlertPollInterval(pollInterval)
 357                 .build();
 358
 359         asyncWrite(LogicalDatastoreType.CONFIGURATION, path, qosAlertConfig, dataBroker, DEFAULT_FUTURE_CALLBACK);
 360     }
 361
 362     private void pollDirectStatisticsForAllNodes() {
 363         LOG.trace("Polling direct statistics from nodes");
 364
 365         for (BigInteger dpn : qosAlertDpnPortNumberMap.keySet()) {
 366             LOG.trace("Polling DPN ID {}", dpn);
 367             GetNodeConnectorStatisticsInputBuilder input = new GetNodeConnectorStatisticsInputBuilder()
 368                     .setNode(new NodeRef(InstanceIdentifier.builder(Nodes.class)
 369                             .child(Node.class, new NodeKey(new NodeId(IfmConstants.OF_URI_PREFIX + dpn))).build()))
 370                     .setStoreStats(false);
 371             Future<RpcResult<GetNodeConnectorStatisticsOutput>> rpcResultFuture =
 372                     odlDirectStatisticsService.getNodeConnectorStatistics(input.build());
 373
 374             RpcResult<GetNodeConnectorStatisticsOutput> rpcResult = null;
 375             try {
 376                 rpcResult = rpcResultFuture.get();
 377             } catch (InterruptedException | ExecutionException e) {
 378                 LOG.error("Exception {} occurred with node {} Direct-Statistics get", e, dpn);
 379             }
 380             if (Optional.fromNullable(rpcResult).isPresent() && rpcResult.isSuccessful()
 381                     && Optional.fromNullable(rpcResult.getResult()).isPresent()) {
 382
 383                 GetNodeConnectorStatisticsOutput nodeConnectorStatisticsOutput = rpcResult.getResult();
 384
 385                 List<NodeConnectorStatisticsAndPortNumberMap> nodeConnectorStatisticsAndPortNumberMapList =
 386                         nodeConnectorStatisticsOutput.getNodeConnectorStatisticsAndPortNumberMap();
 387
 388                 ConcurrentHashMap<String, QosAlertPortData> portDataMap = qosAlertDpnPortNumberMap.get(dpn);
 389                 for (NodeConnectorStatisticsAndPortNumberMap stats : nodeConnectorStatisticsAndPortNumberMapList) {
 390                     QosAlertPortData portData = portDataMap.get(stats.getNodeConnectorId().getValue());
 391                     if (portData != null) {
 392                         portData.updatePortStatistics(stats);
 393                     }
 394                 }
 395             } else {
 396                 LOG.error("Direct-Statistics not available for node {}", dpn);
 397             }
 398
 399         }
 400     }
 401
 402 }