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.sal.common.api.data.TransactionCommitFailedException;
  26 import org.opendaylight.controller.md.statistics.manager.StatListeningCommiter;
  27 import org.opendaylight.controller.md.statistics.manager.StatNodeRegistration;
  28 import org.opendaylight.controller.md.statistics.manager.StatNotifyCommiter;
  29 import org.opendaylight.controller.md.statistics.manager.StatPermCollector;
  30 import org.opendaylight.controller.md.statistics.manager.StatPermCollector.StatCapabTypes;
  31 import org.opendaylight.controller.md.statistics.manager.StatRpcMsgManager;
  32 import org.opendaylight.controller.md.statistics.manager.StatisticsManager;
  33 import org.opendaylight.controller.sal.binding.api.NotificationProviderService;
  34 import org.opendaylight.controller.sal.binding.api.RpcConsumerRegistry;
  35 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.meters.Meter;
  36 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow;
  37 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.OpendaylightFlowStatisticsListener;
  38 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.OpendaylightFlowTableStatisticsListener;
  39 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.port.rev130925.queues.Queue;
  40 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.OpendaylightGroupStatisticsListener;
  41 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.groups.Group;
  42 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
  43 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.OpendaylightMeterStatisticsListener;
  44 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.OpendaylightPortStatisticsListener;
  45 import org.opendaylight.yang.gen.v1.urn.opendaylight.queue.statistics.rev131216.OpendaylightQueueStatisticsListener;
  46 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  47 import org.slf4j.Logger;
  48 import org.slf4j.LoggerFactory;
  49
  50 import com.google.common.base.Preconditions;
  51 import com.google.common.util.concurrent.ThreadFactoryBuilder;
  52
  53 /**
  54 * statistics-manager
  55 * org.opendaylight.controller.md.statistics.manager.impl
  56 *
  57 * StatisticsManagerImpl
  58 * It represent a central point for whole module. Implementation
  59 * {@link StatisticsManager} registers all Operation/DS {@link StatNotifyCommiter} and
  60 * Config/DS {@StatListeningCommiter}, as well as {@link StatPermCollector}
  61 * for statistic collecting and {@link StatRpcMsgManager} as Device RPCs provider.
  62 * In next, StatisticsManager provides all DS contact Transaction services.
  63 *
  64 * @author avishnoi@in.ibm.com <a href="mailto:vdemcak@cisco.com">Vaclav Demcak</a>
  65 *
  66 */
  67 public class StatisticsManagerImpl implements StatisticsManager, Runnable {
  68
  69    private final static Logger LOG = LoggerFactory.getLogger(StatisticsManagerImpl.class);
  70
  71    private static final int QUEUE_DEPTH = 500;
  72    private static final int MAX_BATCH = 1;
  73
  74    private final BlockingQueue<StatDataStoreOperation> dataStoreOperQueue = new LinkedBlockingDeque<>(QUEUE_DEPTH);
  75
  76    private final DataBroker dataBroker;
  77    private final int maxNodesForCollectors;
  78    private long minReqNetMonitInt;
  79    private final ExecutorService statRpcMsgManagerExecutor;
  80    private final ExecutorService statDataStoreOperationServ;
  81    private StatRpcMsgManager rpcMsgManager;
  82    private List<StatPermCollector> statCollectors;
  83    private final Object statCollectorLock = new Object();
  84    private BindingTransactionChain txChain;
  85    private volatile boolean finishing = false;
  86
  87    private StatNodeRegistration nodeRegistrator;
  88    private StatListeningCommiter<Flow, OpendaylightFlowStatisticsListener> flowListeningCommiter;
  89    private StatListeningCommiter<Meter, OpendaylightMeterStatisticsListener> meterListeningCommiter;
  90    private StatListeningCommiter<Group, OpendaylightGroupStatisticsListener> groupListeningCommiter;
  91    private StatListeningCommiter<Queue, OpendaylightQueueStatisticsListener> queueNotifyCommiter;
  92    private StatNotifyCommiter<OpendaylightFlowTableStatisticsListener> tableNotifCommiter;
  93    private StatNotifyCommiter<OpendaylightPortStatisticsListener> portNotifyCommiter;
  94
  95    public StatisticsManagerImpl (final DataBroker dataBroker, final int maxNodesForCollector) {
  96        this.dataBroker = Preconditions.checkNotNull(dataBroker, "DataBroker can not be null!");
  97        ThreadFactory threadFact;
  98        threadFact = new ThreadFactoryBuilder().setNameFormat("odl-stat-rpc-oper-thread-%d").build();
  99        statRpcMsgManagerExecutor = Executors.newSingleThreadExecutor(threadFact);
 100        threadFact = new ThreadFactoryBuilder().setNameFormat("odl-stat-ds-oper-thread-%d").build();
 101        statDataStoreOperationServ = Executors.newSingleThreadExecutor(threadFact);
 102        maxNodesForCollectors = maxNodesForCollector;
 103        txChain =  dataBroker.createTransactionChain(this);
 104    }
 105
 106    @Override
 107    public void start(final NotificationProviderService notifService,
 108            final RpcConsumerRegistry rpcRegistry, final long minReqNetMonitInt) {
 109        Preconditions.checkArgument(rpcRegistry != null, "RpcConsumerRegistry can not be null !");
 110        this.minReqNetMonitInt = minReqNetMonitInt;
 111        rpcMsgManager = new StatRpcMsgManagerImpl(this, rpcRegistry, minReqNetMonitInt);
 112        statCollectors = Collections.emptyList();
 113        nodeRegistrator = new StatNodeRegistrationImpl(this, dataBroker, notifService);
 114        flowListeningCommiter = new StatListenCommitFlow(this, dataBroker, notifService);
 115        meterListeningCommiter = new StatListenCommitMeter(this, dataBroker, notifService);
 116        groupListeningCommiter = new StatListenCommitGroup(this, dataBroker, notifService);
 117        tableNotifCommiter = new StatNotifyCommitTable(this, notifService);
 118        portNotifyCommiter = new StatNotifyCommitPort(this, notifService);
 119        queueNotifyCommiter = new StatListenCommitQueue(this, dataBroker, notifService);
 120
 121        statRpcMsgManagerExecutor.execute(rpcMsgManager);
 122        statDataStoreOperationServ.execute(this);
 123        LOG.info("Statistics Manager started successfully!");
 124    }
 125
 126    @Override
 127    public void close() throws Exception {
 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                try {
 209                    tx.submit().checkedGet();
 210                } catch (final TransactionCommitFailedException e) {
 211                    LOG.warn("Stat DataStoreOperation unexpected State!", e);
 212                    txChain.close();
 213                    txChain = dataBroker.createTransactionChain(StatisticsManagerImpl.this);
 214                    cleanDataStoreOperQueue();
 215                }
 216            }
 217            catch (final IllegalStateException e) {
 218                LOG.warn("Stat DataStoreOperation unexpected State!", e);
 219            }
 220            catch (final InterruptedException e) {
 221                LOG.warn("Stat Manager DS Operation thread interupted!", e);
 222                finishing = true;
 223            }
 224            catch (final Exception e) {
 225                LOG.warn("Stat DataStore Operation executor fail!", e);
 226            }
 227        }
 228        // Drain all events, making sure any blocked threads are unblocked
 229        cleanDataStoreOperQueue();
 230    }
 231
 232    private void cleanDataStoreOperQueue() {
 233        // Drain all events, making sure any blocked threads are unblocked
 234        while (! dataStoreOperQueue.isEmpty()) {
 235            dataStoreOperQueue.poll();
 236        }
 237    }
 238
 239    @Override
 240    public void onTransactionChainFailed(final TransactionChain<?, ?> chain, final AsyncTransaction<?, ?> transaction,
 241            final Throwable cause) {
 242        LOG.warn("Failed to export Flow Capable Statistics, Transaction {} failed.",transaction.getIdentifier(),cause);
 243        txChain.close();
 244        txChain = dataBroker.createTransactionChain(StatisticsManagerImpl.this);
 245        cleanDataStoreOperQueue();
 246    }
 247
 248    @Override
 249    public void onTransactionChainSuccessful(final TransactionChain<?, ?> chain) {
 250        // NOOP
 251    }
 252
 253    @Override
 254    public boolean isProvidedFlowNodeActive(final InstanceIdentifier<Node> nodeIdent) {
 255        for (final StatPermCollector collector : statCollectors) {
 256            if (collector.isProvidedFlowNodeActive(nodeIdent)) {
 257                return true;
 258            }
 259        }
 260        return false;
 261    }
 262
 263    @Override
 264    public void collectNextStatistics(final InstanceIdentifier<Node> nodeIdent) {
 265        for (final StatPermCollector collector : statCollectors) {
 266            if (collector.isProvidedFlowNodeActive(nodeIdent)) {
 267                collector.collectNextStatistics();
 268            }
 269        }
 270    }
 271
 272    @Override
 273    public void connectedNodeRegistration(final InstanceIdentifier<Node> nodeIdent,
 274            final List<StatCapabTypes> statTypes, final Short nrOfSwitchTables) {
 275        for (final StatPermCollector collector : statCollectors) {
 276            if (collector.connectedNodeRegistration(nodeIdent, statTypes, nrOfSwitchTables)) {
 277                return;
 278            }
 279        }
 280        synchronized (statCollectorLock) {
 281            for (final StatPermCollector collector : statCollectors) {
 282                if (collector.connectedNodeRegistration(nodeIdent, statTypes, nrOfSwitchTables)) {
 283                    return;
 284                }
 285            }
 286            final StatPermCollectorImpl newCollector = new StatPermCollectorImpl(this,
 287                    minReqNetMonitInt, statCollectors.size() + 1, maxNodesForCollectors);
 288            final List<StatPermCollector> statCollectorsNew = new ArrayList<>(statCollectors);
 289            newCollector.connectedNodeRegistration(nodeIdent, statTypes, nrOfSwitchTables);
 290            statCollectorsNew.add(newCollector);
 291            statCollectors = Collections.unmodifiableList(statCollectorsNew);
 292        }
 293    }
 294
 295    @Override
 296    public void disconnectedNodeUnregistration(final InstanceIdentifier<Node> nodeIdent) {
 297        for (final StatPermCollector collector : statCollectors) {
 298            if (collector.disconnectedNodeUnregistration(nodeIdent)) {
 299                if ( ! collector.hasActiveNodes()) {
 300                    synchronized (statCollectorLock) {
 301                        if (collector.hasActiveNodes()) {
 302                            return;
 303                        }
 304                        final List<StatPermCollector> newStatColl =
 305                                new ArrayList<>(statCollectors);
 306                        newStatColl.remove(collector);
 307                        statCollectors = Collections.unmodifiableList(newStatColl);
 308                    }
 309                }
 310                return;
 311            }
 312        }
 313        LOG.debug("Node {} has not removed.", nodeIdent);
 314    }
 315
 316    /* Getter internal Statistic Manager Job Classes */
 317    @Override
 318    public StatRpcMsgManager getRpcMsgManager() {
 319        return rpcMsgManager;
 320    }
 321
 322    @Override
 323    public StatNodeRegistration getNodeRegistrator() {
 324        return nodeRegistrator;
 325    }
 326
 327    @Override
 328    public StatListeningCommiter<Flow, OpendaylightFlowStatisticsListener> getFlowListenComit() {
 329        return flowListeningCommiter;
 330    }
 331
 332    @Override
 333    public StatListeningCommiter<Meter, OpendaylightMeterStatisticsListener> getMeterListenCommit() {
 334        return meterListeningCommiter;
 335    }
 336
 337    @Override
 338    public StatListeningCommiter<Group, OpendaylightGroupStatisticsListener> getGroupListenCommit() {
 339        return groupListeningCommiter;
 340    }
 341
 342    @Override
 343    public StatListeningCommiter<Queue, OpendaylightQueueStatisticsListener> getQueueNotifyCommit() {
 344        return queueNotifyCommiter;
 345    }
 346
 347
 348    @Override
 349    public StatNotifyCommiter<OpendaylightFlowTableStatisticsListener> getTableNotifCommit() {
 350        return tableNotifCommiter;
 351    }
 352
 353    @Override
 354    public StatNotifyCommiter<OpendaylightPortStatisticsListener> getPortNotifyCommit() {
 355        return portNotifyCommiter;
 356    }
 357 }
 358