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 import static org.opendaylight.genius.infra.Datastore.CONFIGURATION;
  12
  13 import com.google.gson.Gson;
  14 import com.google.gson.GsonBuilder;
  15 import com.google.gson.JsonArray;
  16 import com.google.gson.JsonObject;
  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 import java.util.concurrent.ExecutionException;
  23 import java.util.concurrent.Future;
  24 import java.util.function.Supplier;
  25 import javax.annotation.PostConstruct;
  26 import javax.annotation.PreDestroy;
  27 import javax.inject.Inject;
  28 import javax.inject.Singleton;
  29 import org.apache.felix.service.command.CommandSession;
  30 import org.opendaylight.genius.infra.ManagedNewTransactionRunner;
  31 import org.opendaylight.genius.infra.ManagedNewTransactionRunnerImpl;
  32 import org.opendaylight.genius.interfacemanager.globals.IfmConstants;
  33 import org.opendaylight.genius.interfacemanager.globals.InterfaceInfo;
  34 import org.opendaylight.genius.interfacemanager.interfaces.IInterfaceManager;
  35 import org.opendaylight.infrautils.utils.concurrent.ListenableFutures;
  36 import org.opendaylight.mdsal.binding.api.DataBroker;
  37 import org.opendaylight.yang.gen.v1.urn.opendaylight.direct.statistics.rev160511.GetNodeConnectorStatisticsInputBuilder;
  38 import org.opendaylight.yang.gen.v1.urn.opendaylight.direct.statistics.rev160511.GetNodeConnectorStatisticsOutput;
  39 import org.opendaylight.yang.gen.v1.urn.opendaylight.direct.statistics.rev160511.OpendaylightDirectStatisticsService;
  40 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
  41 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
  42 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.Nodes;
  43 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
  44 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
  45 import org.opendaylight.yang.gen.v1.urn.opendaylight.netvirt.qosalert.config.rev170301.QosalertConfig;
  46 import org.opendaylight.yang.gen.v1.urn.opendaylight.netvirt.qosalert.config.rev170301.QosalertConfigBuilder;
  47 import org.opendaylight.yang.gen.v1.urn.opendaylight.neutron.ports.rev150712.ports.attributes.ports.Port;
  48 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.node.connector.statistics.and.port.number.map.NodeConnectorStatisticsAndPortNumberMap;
  49 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.node.connector.statistics.and.port.number.map.NodeConnectorStatisticsAndPortNumberMapKey;
  50 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  51 import org.opendaylight.yangtools.yang.common.RpcResult;
  52 import org.opendaylight.yangtools.yang.common.Uint64;
  53 import org.slf4j.Logger;
  54 import org.slf4j.LoggerFactory;
  55
  56
  57 @Singleton
  58 public final class QosAlertManager implements Runnable {
  59     private static final Logger LOG = LoggerFactory.getLogger(QosAlertManager.class);
  60
  61     private volatile boolean alertEnabled;
  62     private volatile int pollInterval;
  63     private volatile Thread thread;
  64     private volatile boolean statsPollThreadStart;
  65
  66     private final ManagedNewTransactionRunner txRunner;
  67     private final QosalertConfig defaultConfig;
  68     private final OpendaylightDirectStatisticsService odlDirectStatisticsService;
  69     private final QosNeutronUtils qosNeutronUtils;
  70     private final QosEosHandler qosEosHandler;
  71     private final IInterfaceManager interfaceManager;
  72     private final Set unprocessedInterfaceIds = ConcurrentHashMap.newKeySet();
  73     private final ConcurrentMap<Uint64, ConcurrentMap<String, QosAlertPortData>> qosAlertDpnPortNumberMap =
  74             new ConcurrentHashMap<>();
  75     private final AlertThresholdSupplier alertThresholdSupplier = new AlertThresholdSupplier();
  76
  77     @Inject
  78     public QosAlertManager(final DataBroker dataBroker,
  79             final OpendaylightDirectStatisticsService odlDirectStatisticsService, final QosalertConfig defaultConfig,
  80             final QosNeutronUtils qosNeutronUtils, final QosEosHandler qosEosHandler,
  81             final IInterfaceManager interfaceManager) {
  82         this.txRunner = new ManagedNewTransactionRunnerImpl(dataBroker);
  83         this.odlDirectStatisticsService = odlDirectStatisticsService;
  84         this.interfaceManager = interfaceManager;
  85         this.defaultConfig = defaultConfig;
  86         this.qosNeutronUtils = qosNeutronUtils;
  87         this.qosEosHandler = qosEosHandler;
  88         LOG.trace("QosAlert default config poll alertEnabled:{} threshold:{} pollInterval:{}",
  89                 defaultConfig.isQosAlertEnabled(), defaultConfig.getQosDropPacketThreshold(),
  90                 defaultConfig.getQosAlertPollInterval());
  91         getDefaultConfig();
  92     }
  93
  94     @PostConstruct
  95     public void init() {
  96         qosEosHandler.addLocalOwnershipChangedListener(this::setQosAlertOwner);
  97         qosAlertDpnPortNumberMap.clear();
  98         statsPollThreadStart = true;
  99         startStatsPollThread();
 100         LOG.trace("{} init done", getClass().getSimpleName());
 101     }
 102
 103     @PreDestroy
 104     public void close() {
 105         statsPollThreadStart = false;
 106         if (thread != null) {
 107             thread.interrupt();
 108         }
 109         LOG.trace("{} close done", getClass().getSimpleName());
 110     }
 111
 112     private void setQosAlertOwner(boolean isOwner) {
 113         LOG.trace("qos alert set owner : {}", isOwner);
 114         statsPollThreadStart = isOwner;
 115         if (thread != null) {
 116             thread.interrupt();
 117         } else {
 118             startStatsPollThread();
 119         }
 120     }
 121
 122     @Override
 123     public void run() {
 124         LOG.debug("Qos alert poll thread started");
 125         while (statsPollThreadStart && alertEnabled) {
 126             LOG.trace("Thread loop polling :{} threshold:{} pollInterval:{}",
 127                     alertEnabled, alertThresholdSupplier.get(), pollInterval);
 128
 129             try {
 130                 pollDirectStatisticsForAllNodes();
 131                 Thread.sleep(pollInterval * 60L * 1000L); // pollInterval in minutes
 132             } catch (final InterruptedException e) {
 133                 LOG.debug("Qos polling thread interrupted");
 134             }
 135         }
 136         thread = null;
 137         LOG.debug("Qos alert poll thread stopped");
 138     }
 139
 140     private void startStatsPollThread() {
 141         if (statsPollThreadStart && alertEnabled && thread == null) {
 142             initPortStatsData();
 143             thread = new Thread(this);
 144             thread.setDaemon(true);
 145             thread.start();
 146         }
 147     }
 148
 149     private void getDefaultConfig() {
 150         alertEnabled = defaultConfig.isQosAlertEnabled();
 151         pollInterval = defaultConfig.getQosAlertPollInterval().toJava();
 152
 153         alertThresholdSupplier.set(defaultConfig.getQosDropPacketThreshold().toJava());
 154     }
 155
 156     public void setQosalertConfig(QosalertConfig config) {
 157
 158         LOG.debug("New QoS alert config threshold:{} polling alertEnabled:{} interval:{}",
 159                 config.getQosDropPacketThreshold(), config.isQosAlertEnabled(),
 160                 config.getQosAlertPollInterval());
 161
 162         alertEnabled = config.isQosAlertEnabled().booleanValue();
 163         pollInterval = config.getQosAlertPollInterval().toJava();
 164
 165         alertThresholdSupplier.set(config.getQosDropPacketThreshold().shortValue());
 166
 167         if (thread != null) {
 168             thread.interrupt();
 169         } else {
 170             startStatsPollThread();
 171         }
 172
 173     }
 174
 175     public void restoreDefaultConfig() {
 176         LOG.debug("Restoring default configuration");
 177         getDefaultConfig();
 178         if (thread != null) {
 179             thread.interrupt();
 180         } else {
 181             startStatsPollThread();
 182         }
 183     }
 184
 185     public void setThreshold(short threshold) {
 186         LOG.debug("setting threshold {} in config data store", threshold);
 187         writeConfigDataStore(alertEnabled, threshold, pollInterval);
 188     }
 189
 190     public void setPollInterval(int pollInterval) {
 191         LOG.debug("setting interval {} in config data store", pollInterval);
 192         writeConfigDataStore(alertEnabled, alertThresholdSupplier.get().shortValue(), pollInterval);
 193     }
 194
 195     public void setEnable(boolean enable) {
 196         LOG.debug("setting QoS poll to {} in config data store", enable);
 197         writeConfigDataStore(enable, alertThresholdSupplier.get().shortValue(), pollInterval);
 198     }
 199
 200     public void addInterfaceIdInQoSAlertCache(String ifaceId) {
 201         LOG.trace("Adding interface id {} in cache", ifaceId);
 202         InterfaceInfo interfaceInfo =
 203                 interfaceManager.getInterfaceInfoFromOperationalDataStore(ifaceId);
 204         if (interfaceInfo == null) {
 205             LOG.debug("Interface not found {}. Added in cache now to process later ", ifaceId);
 206             unprocessedInterfaceIds.add(ifaceId);
 207         } else {
 208             addToQosAlertCache(interfaceInfo);
 209         }
 210     }
 211
 212     public void displayConfig(CommandSession session) {
 213
 214         session.getConsole().println("Qos Alert Configuration Details");
 215         session.getConsole().println("Threshold: " + alertThresholdSupplier.get().shortValue());
 216         session.getConsole().println("AlertEnabled: " + alertEnabled);
 217         session.getConsole().println("Poll Interval: " + pollInterval);
 218
 219         Uint64 dpnId;
 220         String portData;
 221         Gson gson = new GsonBuilder().setPrettyPrinting().create();
 222         if (qosAlertDpnPortNumberMap.isEmpty()) {
 223             session.getConsole().println("\nQosAlert Cache not found\n");
 224             return;
 225         } else {
 226             session.getConsole().println("\nDPN Map");
 227             JsonObject jsonObject;
 228             JsonArray jsonArray;
 229             JsonArray jsonArrayOuter = new JsonArray();
 230             for (Entry<Uint64, ConcurrentMap<String, QosAlertPortData>> dpnEntry
 231                     : qosAlertDpnPortNumberMap.entrySet()) {
 232                 dpnId = dpnEntry.getKey();
 233                 jsonObject = new JsonObject();
 234                 jsonObject.addProperty("DpnId", dpnId.toString());
 235                 ConcurrentMap<String, QosAlertPortData> portInnerMap = qosAlertDpnPortNumberMap.get(dpnId);
 236                 jsonArray = new JsonArray();
 237                 for (ConcurrentMap.Entry<String, QosAlertPortData> portEntry : portInnerMap.entrySet()) {
 238                     portData = "Port_number: " + portEntry.getKey() + ", " + portEntry.getValue();
 239                     jsonArray.add(portData);
 240                 }
 241                 jsonObject.add("QosAlertPortData Cache", jsonArray);
 242                 jsonArrayOuter.add(jsonObject);
 243             }
 244             session.getConsole().println(gson.toJson(jsonArrayOuter));
 245             session.getConsole().println();
 246         }
 247     }
 248
 249     public void processInterfaceUpEvent(String ifaceId) {
 250         LOG.trace("processInterfaceUpEvent {}", ifaceId);
 251         if (unprocessedInterfaceIds.remove(ifaceId)) {
 252             addInterfaceIdInQoSAlertCache(ifaceId);
 253         }
 254     }
 255
 256     private void addToQosAlertCache(InterfaceInfo interfaceInfo) {
 257         Uint64 dpnId = interfaceInfo.getDpId();
 258         if (dpnId.equals(Uint64.valueOf(0L))) {
 259             LOG.warn("Interface {} could not be added to Qos Alert Cache because Dpn Id is not found",
 260                     interfaceInfo.getInterfaceName());
 261             return;
 262         }
 263
 264         Port port = qosNeutronUtils.getNeutronPort(interfaceInfo.getInterfaceName());
 265         if (port == null) {
 266             LOG.warn("Port {} not added to Qos Alert Cache because it is not found", interfaceInfo.getInterfaceName());
 267             return;
 268         }
 269
 270         String portNumber = String.valueOf(interfaceInfo.getPortNo());
 271
 272         LOG.trace("Adding DPN ID {} with port {} port number {}", dpnId, port.getUuid(), portNumber);
 273
 274         qosAlertDpnPortNumberMap.computeIfAbsent(dpnId, key -> new ConcurrentHashMap<>())
 275                 .put(portNumber, new QosAlertPortData(port, qosNeutronUtils, alertThresholdSupplier));
 276     }
 277
 278     public void removeInterfaceIdFromQosAlertCache(String ifaceId) {
 279
 280         LOG.trace("If present, remove interface {} from cache", ifaceId);
 281         unprocessedInterfaceIds.remove(ifaceId);
 282         InterfaceInfo interfaceInfo =
 283                 interfaceManager.getInterfaceInfoFromOperationalDataStore(ifaceId);
 284         if (interfaceInfo == null) {
 285             return;
 286         }
 287         Uint64 dpnId = interfaceInfo.getDpId();
 288         String portNumber = String.valueOf(interfaceInfo.getPortNo());
 289         removeFromQosAlertCache(dpnId, portNumber);
 290     }
 291
 292     public void removeLowerLayerIfFromQosAlertCache(String lowerLayerIf) {
 293         LOG.trace("If present, remove lowerLayerIf {} from cache", lowerLayerIf);
 294         Uint64 dpnId = qosNeutronUtils.getDpnIdFromLowerLayerIf(lowerLayerIf);
 295         String portNumber = qosNeutronUtils.getPortNumberFromLowerLayerIf(lowerLayerIf);
 296         if (dpnId == null || portNumber == null) {
 297             LOG.warn("Interface {} not in openflow:dpnid:portnum format, could not remove from cache", lowerLayerIf);
 298             return;
 299         }
 300         removeFromQosAlertCache(dpnId, portNumber);
 301     }
 302
 303     private void removeFromQosAlertCache(Uint64 dpnId, String portNumber) {
 304         if (qosAlertDpnPortNumberMap.containsKey(dpnId)
 305                 && qosAlertDpnPortNumberMap.get(dpnId).containsKey(portNumber)) {
 306             qosAlertDpnPortNumberMap.get(dpnId).remove(portNumber);
 307             LOG.trace("Removed interace {}:{} from cache", dpnId, portNumber);
 308             if (qosAlertDpnPortNumberMap.get(dpnId).isEmpty()) {
 309                 LOG.trace("DPN {} empty. Removing dpn from cache", dpnId);
 310                 qosAlertDpnPortNumberMap.remove(dpnId);
 311             }
 312         }
 313     }
 314
 315     private void writeConfigDataStore(boolean qosAlertEnabled, short dropPacketThreshold, int alertPollInterval) {
 316
 317         InstanceIdentifier<QosalertConfig> path = InstanceIdentifier.builder(QosalertConfig.class).build();
 318
 319         QosalertConfig qosAlertConfig = new QosalertConfigBuilder()
 320                 .setQosDropPacketThreshold(dropPacketThreshold)
 321                 .setQosAlertEnabled(qosAlertEnabled)
 322                 .setQosAlertPollInterval(alertPollInterval)
 323                 .build();
 324
 325         ListenableFutures.addErrorLogging(txRunner.callWithNewWriteOnlyTransactionAndSubmit(CONFIGURATION,
 326             tx -> tx.mergeParentStructurePut(path,
 327                     qosAlertConfig)), LOG, "Error writing to the config data store");
 328     }
 329
 330     private void pollDirectStatisticsForAllNodes() {
 331         LOG.trace("Polling direct statistics from nodes");
 332
 333         for (Entry<Uint64, ConcurrentMap<String, QosAlertPortData>> entry : qosAlertDpnPortNumberMap.entrySet()) {
 334             Uint64 dpn = entry.getKey();
 335             LOG.trace("Polling DPN ID {}", dpn);
 336             GetNodeConnectorStatisticsInputBuilder input = new GetNodeConnectorStatisticsInputBuilder()
 337                     .setNode(new NodeRef(InstanceIdentifier.builder(Nodes.class)
 338                             .child(Node.class, new NodeKey(new NodeId(IfmConstants.OF_URI_PREFIX + dpn))).build()))
 339                     .setStoreStats(false);
 340             Future<RpcResult<GetNodeConnectorStatisticsOutput>> rpcResultFuture =
 341                     odlDirectStatisticsService.getNodeConnectorStatistics(input.build());
 342
 343             RpcResult<GetNodeConnectorStatisticsOutput> rpcResult = null;
 344             try {
 345                 rpcResult = rpcResultFuture.get();
 346             } catch (InterruptedException | ExecutionException e) {
 347                 if (LOG.isDebugEnabled()) {
 348                     LOG.debug("Could not get Direct-Statistics for node {} Exception occurred ", dpn, e);
 349                 } else {
 350                     LOG.info("Could not get Direct-Statistics for node {}", dpn);
 351                 }
 352             }
 353             if (rpcResult != null && rpcResult.isSuccessful() && rpcResult.getResult() != null) {
 354
 355                 GetNodeConnectorStatisticsOutput nodeConnectorStatisticsOutput = rpcResult.getResult();
 356
 357                 Map<NodeConnectorStatisticsAndPortNumberMapKey, NodeConnectorStatisticsAndPortNumberMap>
 358                         nodeConnectorStatisticsAndPortNumberMap =
 359                         nodeConnectorStatisticsOutput.nonnullNodeConnectorStatisticsAndPortNumberMap();
 360
 361                 ConcurrentMap<String, QosAlertPortData> portDataMap = entry.getValue();
 362                 for (NodeConnectorStatisticsAndPortNumberMap stats
 363                         : nodeConnectorStatisticsAndPortNumberMap.values()) {
 364                     QosAlertPortData portData = portDataMap.get(stats.getNodeConnectorId().getValue());
 365                     if (portData != null) {
 366                         portData.updatePortStatistics(stats);
 367                     }
 368                 }
 369             } else {
 370                 LOG.info("Direct-Statistics not available for node {}", dpn);
 371             }
 372
 373         }
 374     }
 375
 376     private void initPortStatsData() {
 377         qosAlertDpnPortNumberMap.values().forEach(portDataMap -> portDataMap.values()
 378                 .forEach(QosAlertPortData::initPortData));
 379     }
 380
 381     private static class AlertThresholdSupplier implements Supplier<Uint64> {
 382         private volatile Uint64 alertThreshold = Uint64.valueOf(0);
 383
 384         void set(short threshold) {
 385             alertThreshold = Uint64.valueOf(threshold);
 386         }
 387
 388         @Override
 389         public Uint64 get() {
 390             return alertThreshold;
 391         }
 392     }
 393 }