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