opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/impl/StatisticsManagerImpl.java

   1 /**
   2  * Copyright (c) 2014 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.md.statistics.manager.impl;
  10
  11 import java.util.ArrayList;
  12 import java.util.Collections;
  13 import java.util.List;
  14 import java.util.concurrent.BlockingQueue;
  15 import java.util.concurrent.ExecutorService;
  16 import java.util.concurrent.Executors;
  17 import java.util.concurrent.LinkedBlockingDeque;
  18 import java.util.concurrent.ThreadFactory;
  19
  20 import org.opendaylight.controller.md.sal.binding.api.BindingTransactionChain;
  21 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  22 import org.opendaylight.controller.md.sal.binding.api.ReadWriteTransaction;
  23 import org.opendaylight.controller.md.sal.common.api.data.AsyncTransaction;
  24 import org.opendaylight.controller.md.sal.common.api.data.TransactionChain;
  25 import org.opendaylight.controller.md.statistics.manager.StatListeningCommiter;
  26 import org.opendaylight.controller.md.statistics.manager.StatNodeRegistration;
  27 import org.opendaylight.controller.md.statistics.manager.StatNotifyCommiter;
  28 import org.opendaylight.controller.md.statistics.manager.StatPermCollector;
  29 import org.opendaylight.controller.md.statistics.manager.StatPermCollector.StatCapabTypes;
  30 import org.opendaylight.controller.md.statistics.manager.StatRpcMsgManager;
  31 import org.opendaylight.controller.md.statistics.manager.StatisticsManager;
  32 import org.opendaylight.controller.sal.binding.api.NotificationProviderService;
  33 import org.opendaylight.controller.sal.binding.api.RpcConsumerRegistry;
  34 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.meters.Meter;
  35 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow;
  36 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.OpendaylightFlowStatisticsListener;
  37 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.OpendaylightFlowTableStatisticsListener;
  38 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.port.rev130925.queues.Queue;
  39 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.OpendaylightGroupStatisticsListener;
  40 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.groups.Group;
  41 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
  42 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.OpendaylightMeterStatisticsListener;
  43 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.OpendaylightPortStatisticsListener;
  44 import org.opendaylight.yang.gen.v1.urn.opendaylight.queue.statistics.rev131216.OpendaylightQueueStatisticsListener;
  45 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  46 import org.slf4j.Logger;
  47 import org.slf4j.LoggerFactory;
  48
  49 import com.google.common.base.Preconditions;
  50 import com.google.common.util.concurrent.ThreadFactoryBuilder;
  51
  52 /**
  53 * statistics-manager
  54 * org.opendaylight.controller.md.statistics.manager.impl
  55 *
  56 * StatisticsManagerImpl
  57 * It represent a central point for whole module. Implementation
  58 * {@link StatisticsManager} registers all Operation/DS {@link StatNotifyCommiter} and
  59 * Config/DS {@StatListeningCommiter}, as well as {@link StatPermCollector}
  60 * for statistic collecting and {@link StatRpcMsgManager} as Device RPCs provider.
  61 * In next, StatisticsManager provides all DS contact Transaction services.
  62 *
  63 * @author avishnoi@in.ibm.com <a href="mailto:vdemcak@cisco.com">Vaclav Demcak</a>
  64 *
  65 */
  66 public class StatisticsManagerImpl implements StatisticsManager, Runnable {
  67
  68    private final static Logger LOG = LoggerFactory.getLogger(StatisticsManagerImpl.class);
  69
  70    private static final int QUEUE_DEPTH = 5000;
  71    private static final int MAX_BATCH = 100;
  72
  73    private final BlockingQueue<StatDataStoreOperation> dataStoreOperQueue = new LinkedBlockingDeque<>(QUEUE_DEPTH);
  74
  75    private final DataBroker dataBroker;
  76    private final int maxNodesForCollectors;
  77    private long minReqNetMonitInt;
  78    private final ExecutorService statRpcMsgManagerExecutor;
  79    private final ExecutorService statDataStoreOperationServ;
  80    private StatRpcMsgManager rpcMsgManager;
  81    private List<StatPermCollector> statCollectors;
  82    private final Object statCollectorLock = new Object();
  83    private BindingTransactionChain txChain;
  84    private volatile boolean finishing = false;
  85
  86    private StatNodeRegistration nodeRegistrator;
  87    private StatListeningCommiter<Flow, OpendaylightFlowStatisticsListener> flowListeningCommiter;
  88    private StatListeningCommiter<Meter, OpendaylightMeterStatisticsListener> meterListeningCommiter;
  89    private StatListeningCommiter<Group, OpendaylightGroupStatisticsListener> groupListeningCommiter;
  90    private StatListeningCommiter<Queue, OpendaylightQueueStatisticsListener> queueNotifyCommiter;
  91    private StatNotifyCommiter<OpendaylightFlowTableStatisticsListener> tableNotifCommiter;
  92    private StatNotifyCommiter<OpendaylightPortStatisticsListener> portNotifyCommiter;
  93
  94    public StatisticsManagerImpl (final DataBroker dataBroker, final int maxNodesForCollector) {
  95        this.dataBroker = Preconditions.checkNotNull(dataBroker, "DataBroker can not be null!");
  96        ThreadFactory threadFact;
  97        threadFact = new ThreadFactoryBuilder().setNameFormat("odl-stat-rpc-oper-thread-%d").build();
  98        statRpcMsgManagerExecutor = Executors.newSingleThreadExecutor(threadFact);
  99        threadFact = new ThreadFactoryBuilder().setNameFormat("odl-stat-ds-oper-thread-%d").build();
 100        statDataStoreOperationServ = Executors.newSingleThreadExecutor(threadFact);
 101        maxNodesForCollectors = maxNodesForCollector;
 102        txChain =  dataBroker.createTransactionChain(this);
 103    }
 104
 105    @Override
 106    public void start(final NotificationProviderService notifService,
 107            final RpcConsumerRegistry rpcRegistry, final long minReqNetMonitInt) {
 108        Preconditions.checkArgument(rpcRegistry != null, "RpcConsumerRegistry can not be null !");
 109        this.minReqNetMonitInt = minReqNetMonitInt;
 110        rpcMsgManager = new StatRpcMsgManagerImpl(this, rpcRegistry, minReqNetMonitInt);
 111        statCollectors = Collections.emptyList();
 112        nodeRegistrator = new StatNodeRegistrationImpl(this, dataBroker, notifService);
 113        flowListeningCommiter = new StatListenCommitFlow(this, dataBroker, notifService);
 114        meterListeningCommiter = new StatListenCommitMeter(this, dataBroker, notifService);
 115        groupListeningCommiter = new StatListenCommitGroup(this, dataBroker, notifService);
 116        tableNotifCommiter = new StatNotifyCommitTable(this, notifService);
 117        portNotifyCommiter = new StatNotifyCommitPort(this, notifService);
 118        queueNotifyCommiter = new StatListenCommitQueue(this, dataBroker, notifService);
 119
 120        statRpcMsgManagerExecutor.execute(rpcMsgManager);
 121        statDataStoreOperationServ.execute(this);
 122        LOG.info("Statistics Manager started successfully!");
 123    }
 124
 125    @Override
 126    public void close() throws Exception {
 127        LOG.info("StatisticsManager close called");
 128        finishing = true;
 129        if (nodeRegistrator != null) {
 130            nodeRegistrator.close();
 131            nodeRegistrator = null;
 132        }
 133        if (flowListeningCommiter != null) {
 134            flowListeningCommiter.close();
 135            flowListeningCommiter = null;
 136        }
 137        if (meterListeningCommiter != null) {
 138            meterListeningCommiter.close();
 139            meterListeningCommiter = null;
 140        }
 141        if (groupListeningCommiter != null) {
 142            groupListeningCommiter.close();
 143            groupListeningCommiter = null;
 144        }
 145        if (tableNotifCommiter != null) {
 146            tableNotifCommiter.close();
 147            tableNotifCommiter = null;
 148        }
 149        if (portNotifyCommiter != null) {
 150            portNotifyCommiter.close();
 151            portNotifyCommiter = null;
 152        }
 153        if (queueNotifyCommiter != null) {
 154            queueNotifyCommiter.close();
 155            queueNotifyCommiter = null;
 156        }
 157        if (statCollectors != null) {
 158            for (StatPermCollector collector : statCollectors) {
 159                collector.close();
 160                collector = null;
 161            }
 162            statCollectors = null;
 163        }
 164        if (rpcMsgManager != null) {
 165            rpcMsgManager.close();
 166            rpcMsgManager = null;
 167        }
 168        statRpcMsgManagerExecutor.shutdown();
 169        statDataStoreOperationServ.shutdown();
 170        if (txChain != null) {
 171            txChain.close();
 172            txChain = null;
 173        }
 174    }
 175
 176    @Override
 177    public void enqueue(final StatDataStoreOperation op) {
 178        // we don't need to block anything - next statistics come soon
 179        final boolean success = dataStoreOperQueue.offer(op);
 180        if ( ! success) {
 181            LOG.debug("Stat DS/Operational submiter Queue is full!");
 182        }
 183    }
 184
 185    @Override
 186    public void run() {
 187        /* Neverending cyle - wait for finishing */
 188        while ( ! finishing) {
 189            try {
 190                StatDataStoreOperation op = dataStoreOperQueue.take();
 191                final ReadWriteTransaction tx = txChain.newReadWriteTransaction();
 192                LOG.trace("New operations available, starting transaction {}", tx.getIdentifier());
 193
 194                int ops = 0;
 195                do {
 196                    op.applyOperation(tx);
 197
 198                    ops++;
 199                    if (ops < MAX_BATCH) {
 200                        op = dataStoreOperQueue.poll();
 201                    } else {
 202                        op = null;
 203                    }
 204                } while (op != null);
 205
 206                LOG.trace("Processed {} operations, submitting transaction {}", ops, tx.getIdentifier());
 207
 208                    tx.submit().checkedGet();
 209            } catch (final InterruptedException e) {
 210                LOG.warn("Stat Manager DS Operation thread interupted!", e);
 211                finishing = true;
 212            } catch (final Exception e) {
 213                LOG.warn("Unhandled exception during processing statistics. Restarting transaction chain.", e);
 214                txChain.close();
 215                txChain = dataBroker.createTransactionChain(StatisticsManagerImpl.this);
 216                cleanDataStoreOperQueue();
 217            }
 218        }
 219        // Drain all events, making sure any blocked threads are unblocked
 220        cleanDataStoreOperQueue();
 221    }
 222
 223    private synchronized void cleanDataStoreOperQueue() {
 224        // Drain all events, making sure any blocked threads are unblocked
 225        while (! dataStoreOperQueue.isEmpty()) {
 226            dataStoreOperQueue.poll();
 227        }
 228    }
 229
 230    @Override
 231    public void onTransactionChainFailed(final TransactionChain<?, ?> chain, final AsyncTransaction<?, ?> transaction,
 232            final Throwable cause) {
 233        LOG.warn("Failed to export Flow Capable Statistics, Transaction {} failed.",transaction.getIdentifier(),cause);
 234    }
 235
 236    @Override
 237    public void onTransactionChainSuccessful(final TransactionChain<?, ?> chain) {
 238        // NOOP
 239    }
 240
 241    @Override
 242    public boolean isProvidedFlowNodeActive(final InstanceIdentifier<Node> nodeIdent) {
 243        for (final StatPermCollector collector : statCollectors) {
 244            if (collector.isProvidedFlowNodeActive(nodeIdent)) {
 245                return true;
 246            }
 247        }
 248        return false;
 249    }
 250
 251    @Override
 252    public void collectNextStatistics(final InstanceIdentifier<Node> nodeIdent) {
 253        for (final StatPermCollector collector : statCollectors) {
 254            if (collector.isProvidedFlowNodeActive(nodeIdent)) {
 255                collector.collectNextStatistics();
 256            }
 257        }
 258    }
 259
 260    @Override
 261    public void connectedNodeRegistration(final InstanceIdentifier<Node> nodeIdent,
 262            final List<StatCapabTypes> statTypes, final Short nrOfSwitchTables) {
 263        for (final StatPermCollector collector : statCollectors) {
 264            if (collector.connectedNodeRegistration(nodeIdent, statTypes, nrOfSwitchTables)) {
 265                return;
 266            }
 267        }
 268        synchronized (statCollectorLock) {
 269            for (final StatPermCollector collector : statCollectors) {
 270                if (collector.connectedNodeRegistration(nodeIdent, statTypes, nrOfSwitchTables)) {
 271                    return;
 272                }
 273            }
 274            final StatPermCollectorImpl newCollector = new StatPermCollectorImpl(this,
 275                    minReqNetMonitInt, statCollectors.size() + 1, maxNodesForCollectors);
 276            final List<StatPermCollector> statCollectorsNew = new ArrayList<>(statCollectors);
 277            newCollector.connectedNodeRegistration(nodeIdent, statTypes, nrOfSwitchTables);
 278            statCollectorsNew.add(newCollector);
 279            statCollectors = Collections.unmodifiableList(statCollectorsNew);
 280        }
 281    }
 282
 283    @Override
 284    public void disconnectedNodeUnregistration(final InstanceIdentifier<Node> nodeIdent) {
 285        flowListeningCommiter.cleanForDisconnect(nodeIdent);
 286
 287        for (final StatPermCollector collector : statCollectors) {
 288            if (collector.disconnectedNodeUnregistration(nodeIdent)) {
 289                if ( ! collector.hasActiveNodes()) {
 290                    synchronized (statCollectorLock) {
 291                        if (collector.hasActiveNodes()) {
 292                            return;
 293                        }
 294                        final List<StatPermCollector> newStatColl =
 295                                new ArrayList<>(statCollectors);
 296                        newStatColl.remove(collector);
 297                        statCollectors = Collections.unmodifiableList(newStatColl);
 298                    }
 299                }
 300                return;
 301            }
 302        }
 303        LOG.debug("Node {} has not removed.", nodeIdent);
 304    }
 305
 306    /* Getter internal Statistic Manager Job Classes */
 307    @Override
 308    public StatRpcMsgManager getRpcMsgManager() {
 309        return rpcMsgManager;
 310    }
 311
 312    @Override
 313    public StatNodeRegistration getNodeRegistrator() {
 314        return nodeRegistrator;
 315    }
 316
 317    @Override
 318    public StatListeningCommiter<Flow, OpendaylightFlowStatisticsListener> getFlowListenComit() {
 319        return flowListeningCommiter;
 320    }
 321
 322    @Override
 323    public StatListeningCommiter<Meter, OpendaylightMeterStatisticsListener> getMeterListenCommit() {
 324        return meterListeningCommiter;
 325    }
 326
 327    @Override
 328    public StatListeningCommiter<Group, OpendaylightGroupStatisticsListener> getGroupListenCommit() {
 329        return groupListeningCommiter;
 330    }
 331
 332    @Override
 333    public StatListeningCommiter<Queue, OpendaylightQueueStatisticsListener> getQueueNotifyCommit() {
 334        return queueNotifyCommiter;
 335    }
 336
 337
 338    @Override
 339    public StatNotifyCommiter<OpendaylightFlowTableStatisticsListener> getTableNotifCommit() {
 340        return tableNotifCommiter;
 341    }
 342
 343    @Override
 344    public StatNotifyCommiter<OpendaylightPortStatisticsListener> getPortNotifyCommit() {
 345        return portNotifyCommiter;
 346    }
 347 }
 348