openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/statistics/StatisticsManagerImpl.java

   1 /*
   2  * Copyright (c) 2015 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.impl.statistics;
  10
  11 import com.google.common.annotations.VisibleForTesting;
  12 import com.google.common.base.Preconditions;
  13 import com.google.common.base.Verify;
  14 import com.google.common.collect.Iterators;
  15 import com.google.common.util.concurrent.FutureCallback;
  16 import com.google.common.util.concurrent.Futures;
  17 import com.google.common.util.concurrent.ListenableFuture;
  18 import io.netty.util.HashedWheelTimer;
  19 import io.netty.util.Timeout;
  20 import io.netty.util.TimerTask;
  21 import java.util.Iterator;
  22 import java.util.Map;
  23 import java.util.Optional;
  24 import java.util.concurrent.CancellationException;
  25 import java.util.concurrent.ConcurrentHashMap;
  26 import java.util.concurrent.ConcurrentMap;
  27 import java.util.concurrent.Future;
  28 import java.util.concurrent.Semaphore;
  29 import java.util.concurrent.TimeUnit;
  30 import javax.annotation.Nonnull;
  31 import org.opendaylight.controller.sal.binding.api.BindingAwareBroker;
  32 import org.opendaylight.controller.sal.binding.api.RpcProviderRegistry;
  33 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
  34 import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo;
  35 import org.opendaylight.openflowplugin.api.openflow.device.DeviceState;
  36 import org.opendaylight.openflowplugin.api.openflow.device.handlers.DeviceInitializationPhaseHandler;
  37 import org.opendaylight.openflowplugin.api.openflow.device.handlers.DeviceTerminationPhaseHandler;
  38 import org.opendaylight.openflowplugin.api.openflow.lifecycle.LifecycleService;
  39 import org.opendaylight.openflowplugin.api.openflow.rpc.ItemLifeCycleSource;
  40 import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsContext;
  41 import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsManager;
  42 import org.opendaylight.openflowplugin.openflow.md.core.sal.convertor.ConvertorExecutor;
  43 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflowplugin.sm.control.rev150812.ChangeStatisticsWorkModeInput;
  44 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflowplugin.sm.control.rev150812.GetStatisticsWorkModeOutput;
  45 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflowplugin.sm.control.rev150812.GetStatisticsWorkModeOutputBuilder;
  46 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflowplugin.sm.control.rev150812.StatisticsManagerControlService;
  47 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflowplugin.sm.control.rev150812.StatisticsWorkMode;
  48 import org.opendaylight.yangtools.yang.common.RpcError;
  49 import org.opendaylight.yangtools.yang.common.RpcResult;
  50 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
  51 import org.slf4j.Logger;
  52 import org.slf4j.LoggerFactory;
  53
  54 public class StatisticsManagerImpl implements StatisticsManager, StatisticsManagerControlService {
  55
  56     private static final Logger LOG = LoggerFactory.getLogger(StatisticsManagerImpl.class);
  57
  58     private static final long DEFAULT_STATS_TIMEOUT_SEC = 50L;
  59     private final ConvertorExecutor convertorExecutor;
  60
  61     private DeviceInitializationPhaseHandler deviceInitPhaseHandler;
  62     private DeviceTerminationPhaseHandler deviceTerminPhaseHandler;
  63
  64     private final ConcurrentMap<DeviceInfo, StatisticsContext> contexts = new ConcurrentHashMap<>();
  65
  66     private static final long basicTimerDelay = 3000;
  67     private static long currentTimerDelay = basicTimerDelay;
  68     private static final long maximumTimerDelay = 900000; //wait max 15 minutes for next statistics
  69
  70     private StatisticsWorkMode workMode = StatisticsWorkMode.COLLECTALL;
  71     private final Semaphore workModeGuard = new Semaphore(1, true);
  72     private boolean isStatisticsPollingEnabled;
  73     private BindingAwareBroker.RpcRegistration<StatisticsManagerControlService> controlServiceRegistration;
  74
  75     private final HashedWheelTimer hashedWheelTimer;
  76
  77     @Override
  78     public void setDeviceInitializationPhaseHandler(final DeviceInitializationPhaseHandler handler) {
  79         deviceInitPhaseHandler = handler;
  80     }
  81
  82     public StatisticsManagerImpl(final RpcProviderRegistry rpcProviderRegistry,
  83                                  final boolean isStatisticsPollingEnabled,
  84                                  final HashedWheelTimer hashedWheelTimer,
  85                                  final ConvertorExecutor convertorExecutor) {
  86         Preconditions.checkArgument(rpcProviderRegistry != null);
  87         this.convertorExecutor = convertorExecutor;
  88         this.controlServiceRegistration = Preconditions.checkNotNull(rpcProviderRegistry.addRpcImplementation(
  89                 StatisticsManagerControlService.class, this));
  90         this.isStatisticsPollingEnabled = isStatisticsPollingEnabled;
  91         this.hashedWheelTimer = hashedWheelTimer;
  92     }
  93
  94     @Override
  95     public void onDeviceContextLevelUp(final DeviceInfo deviceInfo, final LifecycleService lifecycleService) throws Exception {
  96
  97         final StatisticsContext statisticsContext = new StatisticsContextImpl(deviceInfo, isStatisticsPollingEnabled, lifecycleService, convertorExecutor, this);
  98         Verify.verify(contexts.putIfAbsent(deviceInfo, statisticsContext) == null, "StatisticsCtx still not closed for Node {}", deviceInfo.getLOGValue());
  99         lifecycleService.setStatContext(statisticsContext);
 100         deviceInitPhaseHandler.onDeviceContextLevelUp(deviceInfo, lifecycleService);
 101     }
 102
 103     @VisibleForTesting
 104     void pollStatistics(final DeviceState deviceState,
 105                         final StatisticsContext statisticsContext,
 106                         final TimeCounter timeCounter,
 107                         final DeviceInfo deviceInfo) {
 108
 109         if (!statisticsContext.isSchedulingEnabled()) {
 110             if (LOG.isDebugEnabled()) {
 111                 LOG.debug("Disabled statistics scheduling for device: {}", deviceInfo.getNodeId().getValue());
 112             }
 113             return;
 114         }
 115
 116         if (LOG.isDebugEnabled()) {
 117             LOG.debug("POLLING ALL STATISTICS for device: {}", deviceInfo.getNodeId());
 118         }
 119         timeCounter.markStart();
 120         final ListenableFuture<Boolean> deviceStatisticsCollectionFuture = statisticsContext.gatherDynamicData();
 121         Futures.addCallback(deviceStatisticsCollectionFuture, new FutureCallback<Boolean>() {
 122             @Override
 123             public void onSuccess(final Boolean o) {
 124                 timeCounter.addTimeMark();
 125                 calculateTimerDelay(timeCounter);
 126                 scheduleNextPolling(deviceState, deviceInfo, statisticsContext, timeCounter);
 127             }
 128
 129             @Override
 130             public void onFailure(@Nonnull final Throwable throwable) {
 131                 timeCounter.addTimeMark();
 132                 LOG.warn("Statistics gathering for single node {} was not successful: ", deviceInfo.getLOGValue(), throwable.getMessage());
 133                 if (LOG.isTraceEnabled()) {
 134                     LOG.trace("Gathering for node {} failure: ", deviceInfo.getLOGValue(), throwable);
 135                 }
 136                 calculateTimerDelay(timeCounter);
 137                 if (throwable instanceof CancellationException) {
 138                     /* This often happens when something wrong with akka or DS, so closing connection will help to restart device **/
 139                     contexts.get(deviceInfo).getLifecycleService().closeConnection();
 140                 } else {
 141                     if (throwable instanceof IllegalStateException) {
 142                         stopScheduling(deviceInfo);
 143                     } else {
 144                         scheduleNextPolling(deviceState, deviceInfo, statisticsContext, timeCounter);
 145                     }
 146                 }
 147             }
 148         });
 149
 150         final long averageTime = TimeUnit.MILLISECONDS.toSeconds(timeCounter.getAverageTimeBetweenMarks());
 151         final long STATS_TIMEOUT_SEC = averageTime > 0 ? 3 * averageTime : DEFAULT_STATS_TIMEOUT_SEC;
 152         final TimerTask timerTask = timeout -> {
 153             if (!deviceStatisticsCollectionFuture.isDone()) {
 154                 LOG.info("Statistics collection for node {} still in progress even after {} secs", deviceInfo.getLOGValue(), STATS_TIMEOUT_SEC);
 155                 deviceStatisticsCollectionFuture.cancel(true);
 156             }
 157         };
 158
 159         hashedWheelTimer.newTimeout(timerTask, STATS_TIMEOUT_SEC, TimeUnit.SECONDS);
 160     }
 161
 162     private void scheduleNextPolling(final DeviceState deviceState,
 163                                      final DeviceInfo deviceInfo,
 164                                      final StatisticsContext statisticsContext,
 165                                      final TimeCounter timeCounter) {
 166         if (LOG.isDebugEnabled()) {
 167             LOG.debug("SCHEDULING NEXT STATISTICS POLLING for device: {}", deviceInfo.getNodeId());
 168         }
 169         if (!isStatisticsPollingEnabled) {
 170             final Timeout pollTimeout = hashedWheelTimer.newTimeout(
 171                     timeout -> pollStatistics(
 172                             deviceState,
 173                             statisticsContext,
 174                             timeCounter,
 175                             deviceInfo),
 176                     currentTimerDelay,
 177                     TimeUnit.MILLISECONDS);
 178             statisticsContext.setPollTimeout(pollTimeout);
 179         }
 180     }
 181
 182     @VisibleForTesting
 183     void calculateTimerDelay(final TimeCounter timeCounter) {
 184         final long averageStatisticsGatheringTime = timeCounter.getAverageTimeBetweenMarks();
 185         if (averageStatisticsGatheringTime > currentTimerDelay) {
 186             currentTimerDelay *= 2;
 187             if (currentTimerDelay > maximumTimerDelay) {
 188                 currentTimerDelay = maximumTimerDelay;
 189             }
 190         } else {
 191             if (currentTimerDelay > basicTimerDelay) {
 192                 currentTimerDelay /= 2;
 193             } else {
 194                 currentTimerDelay = basicTimerDelay;
 195             }
 196         }
 197     }
 198
 199     @VisibleForTesting
 200     static long getCurrentTimerDelay() {
 201         return currentTimerDelay;
 202     }
 203
 204     @Override
 205     public void onDeviceContextLevelDown(final DeviceInfo deviceInfo) {
 206         final StatisticsContext statisticsContext = contexts.remove(deviceInfo);
 207         if (null != statisticsContext) {
 208             LOG.debug("Removing device context from stack. No more statistics gathering for device: {}", deviceInfo.getLOGValue());
 209             statisticsContext.close();
 210         }
 211         deviceTerminPhaseHandler.onDeviceContextLevelDown(deviceInfo);
 212     }
 213
 214     @Override
 215     public Future<RpcResult<GetStatisticsWorkModeOutput>> getStatisticsWorkMode() {
 216         final GetStatisticsWorkModeOutputBuilder smModeOutputBld = new GetStatisticsWorkModeOutputBuilder();
 217         smModeOutputBld.setMode(workMode);
 218         return RpcResultBuilder.success(smModeOutputBld.build()).buildFuture();
 219     }
 220
 221     @Override
 222     public Future<RpcResult<Void>> changeStatisticsWorkMode(ChangeStatisticsWorkModeInput input) {
 223         final Future<RpcResult<Void>> result;
 224         // acquire exclusive access
 225         if (workModeGuard.tryAcquire()) {
 226             final StatisticsWorkMode targetWorkMode = input.getMode();
 227             if (!workMode.equals(targetWorkMode)) {
 228                 isStatisticsPollingEnabled = StatisticsWorkMode.FULLYDISABLED.equals(targetWorkMode);
 229                 // iterate through stats-ctx: propagate mode
 230                 for (Map.Entry<DeviceInfo, StatisticsContext> entry : contexts.entrySet()) {
 231                     final DeviceInfo deviceInfo = entry.getKey();
 232                     final StatisticsContext statisticsContext = entry.getValue();
 233                     final DeviceContext deviceContext = statisticsContext.getLifecycleService().getDeviceContext();
 234                     switch (targetWorkMode) {
 235                         case COLLECTALL:
 236                             scheduleNextPolling(deviceContext.getDeviceState(), deviceInfo, statisticsContext, new TimeCounter());
 237                             for (final ItemLifeCycleSource lifeCycleSource : deviceContext.getItemLifeCycleSourceRegistry().getLifeCycleSources()) {
 238                                 lifeCycleSource.setItemLifecycleListener(null);
 239                             }
 240                             break;
 241                         case FULLYDISABLED:
 242                             final Optional<Timeout> pollTimeout = statisticsContext.getPollTimeout();
 243                             if (pollTimeout.isPresent()) {
 244                                 pollTimeout.get().cancel();
 245                             }
 246                             for (final ItemLifeCycleSource lifeCycleSource : deviceContext.getItemLifeCycleSourceRegistry().getLifeCycleSources()) {
 247                                 lifeCycleSource.setItemLifecycleListener(statisticsContext.getItemLifeCycleListener());
 248                             }
 249                             break;
 250                         default:
 251                             LOG.warn("Statistics work mode not supported: {}", targetWorkMode);
 252                     }
 253                 }
 254                 workMode = targetWorkMode;
 255             }
 256             workModeGuard.release();
 257             result = RpcResultBuilder.<Void>success().buildFuture();
 258         } else {
 259             result = RpcResultBuilder.<Void>failed()
 260                     .withError(RpcError.ErrorType.APPLICATION, "mode change already in progress")
 261                     .buildFuture();
 262         }
 263         return result;
 264     }
 265
 266     @Override
 267     public void startScheduling(final DeviceInfo deviceInfo) {
 268         if (isStatisticsPollingEnabled) {
 269             LOG.info("Statistics are shutdown for device: {}", deviceInfo.getNodeId());
 270             return;
 271         }
 272
 273         final StatisticsContext statisticsContext = contexts.get(deviceInfo);
 274
 275         if (statisticsContext == null) {
 276             LOG.warn("Statistics context not found for device: {}", deviceInfo.getNodeId());
 277             return;
 278         }
 279
 280         if (statisticsContext.isSchedulingEnabled()) {
 281             LOG.debug("Statistics scheduling is already enabled for device: {}", deviceInfo.getNodeId());
 282             return;
 283         }
 284
 285         LOG.info("Scheduling statistics poll for device: {}", deviceInfo.getNodeId());
 286
 287         statisticsContext.setSchedulingEnabled(true);
 288         final DeviceState deviceState = contexts.get(deviceInfo).getLifecycleService().getDeviceContext().getDeviceState();
 289         scheduleNextPolling(deviceState, deviceInfo, statisticsContext, new TimeCounter());
 290     }
 291
 292     @Override
 293     public void stopScheduling(final DeviceInfo deviceInfo) {
 294         if (LOG.isDebugEnabled()) {
 295             LOG.debug("Stopping statistics scheduling for device: {}", deviceInfo.getNodeId());
 296         }
 297         final StatisticsContext statisticsContext = contexts.get(deviceInfo);
 298
 299         if (statisticsContext == null) {
 300             LOG.warn("Statistics context not found for device: {}", deviceInfo.getNodeId());
 301             return;
 302         }
 303
 304         statisticsContext.setSchedulingEnabled(false);
 305     }
 306
 307     @Override
 308     public void close() {
 309         if (controlServiceRegistration != null) {
 310             controlServiceRegistration.close();
 311             controlServiceRegistration = null;
 312         }
 313         for (final Iterator<StatisticsContext> iterator = Iterators.consumingIterator(contexts.values().iterator());
 314                 iterator.hasNext();) {
 315             iterator.next().close();
 316         }
 317     }
 318
 319     @Override
 320     public void setDeviceTerminationPhaseHandler(final DeviceTerminationPhaseHandler handler) {
 321         this.deviceTerminPhaseHandler = handler;
 322     }
 323
 324 }