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.ConnectionException;
  34 import org.opendaylight.openflowplugin.api.openflow.OFPContext;
  35 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
  36 import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo;
  37 import org.opendaylight.openflowplugin.api.openflow.device.DeviceState;
  38 import org.opendaylight.openflowplugin.api.openflow.device.handlers.DeviceInitializationPhaseHandler;
  39 import org.opendaylight.openflowplugin.api.openflow.device.handlers.DeviceTerminationPhaseHandler;
  40 import org.opendaylight.openflowplugin.api.openflow.lifecycle.LifecycleService;
  41 import org.opendaylight.openflowplugin.api.openflow.rpc.ItemLifeCycleSource;
  42 import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsContext;
  43 import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsManager;
  44 import org.opendaylight.openflowplugin.openflow.md.core.sal.convertor.ConvertorExecutor;
  45 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflowplugin.sm.control.rev150812.ChangeStatisticsWorkModeInput;
  46 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflowplugin.sm.control.rev150812.GetStatisticsWorkModeOutput;
  47 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflowplugin.sm.control.rev150812.GetStatisticsWorkModeOutputBuilder;
  48 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflowplugin.sm.control.rev150812.StatisticsManagerControlService;
  49 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflowplugin.sm.control.rev150812.StatisticsWorkMode;
  50 import org.opendaylight.yangtools.yang.common.RpcError;
  51 import org.opendaylight.yangtools.yang.common.RpcResult;
  52 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
  53 import org.slf4j.Logger;
  54 import org.slf4j.LoggerFactory;
  55
  56 public class StatisticsManagerImpl implements StatisticsManager, StatisticsManagerControlService {
  57
  58     private static final Logger LOG = LoggerFactory.getLogger(StatisticsManagerImpl.class);
  59
  60     private static final long DEFAULT_STATS_TIMEOUT_SEC = 50L;
  61     private final ConvertorExecutor converterExecutor;
  62
  63     private DeviceInitializationPhaseHandler deviceInitPhaseHandler;
  64     private DeviceTerminationPhaseHandler deviceTerminationPhaseHandler;
  65
  66     private final ConcurrentMap<DeviceInfo, StatisticsContext> contexts = new ConcurrentHashMap<>();
  67
  68     private static long basicTimerDelay;
  69     private static long currentTimerDelay;
  70     private static long maximumTimerDelay; //wait time for next statistics
  71
  72     private StatisticsWorkMode workMode = StatisticsWorkMode.COLLECTALL;
  73     private final Semaphore workModeGuard = new Semaphore(1, true);
  74     private boolean isStatisticsPollingOn;
  75     private BindingAwareBroker.RpcRegistration<StatisticsManagerControlService> controlServiceRegistration;
  76
  77     private final HashedWheelTimer hashedWheelTimer;
  78
  79     @Override
  80     public void setDeviceInitializationPhaseHandler(final DeviceInitializationPhaseHandler handler) {
  81         deviceInitPhaseHandler = handler;
  82     }
  83
  84     public StatisticsManagerImpl(final RpcProviderRegistry rpcProviderRegistry,
  85                                  final boolean isStatisticsPollingOn,
  86                                  final HashedWheelTimer hashedWheelTimer,
  87                                  final ConvertorExecutor convertorExecutor,
  88                                  final long basicTimerDelay,
  89                                  final long maximumTimerDelay) {
  90         Preconditions.checkArgument(rpcProviderRegistry != null);
  91             this.converterExecutor = convertorExecutor;
  92         this.controlServiceRegistration = Preconditions.checkNotNull(
  93                 rpcProviderRegistry.addRpcImplementation(StatisticsManagerControlService.class, this)
  94         );
  95         this.isStatisticsPollingOn = isStatisticsPollingOn;
  96         this.basicTimerDelay = basicTimerDelay;
  97         this.currentTimerDelay = basicTimerDelay;
  98         this.maximumTimerDelay = maximumTimerDelay;
  99         this.hashedWheelTimer = hashedWheelTimer;
 100     }
 101
 102     @Override
 103     public void onDeviceContextLevelUp(final DeviceInfo deviceInfo,
 104                                        final LifecycleService lifecycleService) throws Exception {
 105
 106         final StatisticsContext statisticsContext =
 107                 new StatisticsContextImpl(
 108                         deviceInfo,
 109                         isStatisticsPollingOn,
 110                         lifecycleService,
 111                         converterExecutor,
 112                         this);
 113
 114         Verify.verify(
 115                 contexts.putIfAbsent(deviceInfo, statisticsContext) == null,
 116                 "StatisticsCtx still not closed for Node {}", deviceInfo.getLOGValue()
 117         );
 118
 119         lifecycleService.setStatContext(statisticsContext);
 120         lifecycleService.registerDeviceRemovedHandler(this);
 121         deviceInitPhaseHandler.onDeviceContextLevelUp(deviceInfo, lifecycleService);
 122     }
 123
 124     @VisibleForTesting
 125     void pollStatistics(final DeviceState deviceState,
 126                         final StatisticsContext statisticsContext,
 127                         final TimeCounter timeCounter,
 128                         final DeviceInfo deviceInfo) {
 129
 130         if (!statisticsContext.isSchedulingEnabled()) {
 131             if (LOG.isDebugEnabled()) {
 132                 LOG.debug("Disabled statistics scheduling for device: {}", deviceInfo.getNodeId().getValue());
 133             }
 134             return;
 135         }
 136
 137         if (LOG.isDebugEnabled()) {
 138             LOG.debug("POLLING ALL STATISTICS for device: {}", deviceInfo.getNodeId());
 139         }
 140
 141         timeCounter.markStart();
 142         final ListenableFuture<Boolean> deviceStatisticsCollectionFuture = statisticsContext.gatherDynamicData();
 143         Futures.addCallback(deviceStatisticsCollectionFuture, new FutureCallback<Boolean>() {
 144             @Override
 145             public void onSuccess(final Boolean o) {
 146                 timeCounter.addTimeMark();
 147                 calculateTimerDelay(timeCounter);
 148                 scheduleNextPolling(deviceState, deviceInfo, statisticsContext, timeCounter);
 149             }
 150
 151             @Override
 152             public void onFailure(@Nonnull final Throwable throwable) {
 153                 timeCounter.addTimeMark();
 154                 LOG.warn("Statistics gathering for single node {} was not successful: {}", deviceInfo.getLOGValue(),
 155                         throwable.getMessage());
 156                 if (LOG.isTraceEnabled()) {
 157                     LOG.trace("Gathering for node {} failure: ", deviceInfo.getLOGValue(), throwable);
 158                 }
 159                 calculateTimerDelay(timeCounter);
 160                 if (throwable instanceof ConnectionException) {
 161                     // ConnectionException is raised by StatisticsContextImpl class when the connections
 162                     // move to RIP state. In this particular case, there is no need to reschedule
 163                     // because this statistics manager should be closed soon
 164                     LOG.warn("Node {} is no more connected, stopping the statistics collection",
 165                             deviceInfo.getLOGValue(),throwable);
 166                     stopScheduling(deviceInfo);
 167                 } else {
 168                     if (!(throwable instanceof CancellationException)) {
 169                         LOG.warn("Unexpected error occurred during statistics collection for node {}, rescheduling " +
 170                                 "statistics collections", deviceInfo.getLOGValue(),throwable);
 171                     }
 172                     scheduleNextPolling(deviceState, deviceInfo, statisticsContext, timeCounter);
 173                 }
 174             }
 175         });
 176
 177         final long averageTime = TimeUnit.MILLISECONDS.toSeconds(timeCounter.getAverageTimeBetweenMarks());
 178         final long statsTimeoutSec = averageTime > 0 ? 3 * averageTime : DEFAULT_STATS_TIMEOUT_SEC;
 179         final TimerTask timerTask = timeout -> {
 180             if (!deviceStatisticsCollectionFuture.isDone()) {
 181                 LOG.info("Statistics collection for node {} still in progress even after {} secs", deviceInfo.getLOGValue(), statsTimeoutSec);
 182                 deviceStatisticsCollectionFuture.cancel(true);
 183             }
 184         };
 185
 186         hashedWheelTimer.newTimeout(timerTask, statsTimeoutSec, TimeUnit.SECONDS);
 187     }
 188
 189     private void scheduleNextPolling(final DeviceState deviceState,
 190                                      final DeviceInfo deviceInfo,
 191                                      final StatisticsContext statisticsContext,
 192                                      final TimeCounter timeCounter) {
 193         if (LOG.isDebugEnabled()) {
 194             LOG.debug("SCHEDULING NEXT STATISTICS POLLING for device: {}", deviceInfo.getNodeId());
 195         }
 196         if (isStatisticsPollingOn) {
 197             final Timeout pollTimeout = hashedWheelTimer.newTimeout(
 198                     timeout -> pollStatistics(
 199                             deviceState,
 200                             statisticsContext,
 201                             timeCounter,
 202                             deviceInfo),
 203                     currentTimerDelay,
 204                     TimeUnit.MILLISECONDS);
 205             statisticsContext.setPollTimeout(pollTimeout);
 206         }
 207     }
 208
 209     @VisibleForTesting
 210     void calculateTimerDelay(final TimeCounter timeCounter) {
 211         final long averageStatisticsGatheringTime = timeCounter.getAverageTimeBetweenMarks();
 212         if (averageStatisticsGatheringTime > currentTimerDelay) {
 213             currentTimerDelay *= 2;
 214             if (currentTimerDelay > maximumTimerDelay) {
 215                 currentTimerDelay = maximumTimerDelay;
 216             }
 217         } else {
 218             if (currentTimerDelay > basicTimerDelay) {
 219                 currentTimerDelay /= 2;
 220             } else {
 221                 currentTimerDelay = basicTimerDelay;
 222             }
 223         }
 224     }
 225
 226     @VisibleForTesting
 227     static long getCurrentTimerDelay() {
 228         return currentTimerDelay;
 229     }
 230
 231     @Override
 232     public void onDeviceContextLevelDown(final DeviceInfo deviceInfo) {
 233         Optional.ofNullable(contexts.get(deviceInfo)).ifPresent(OFPContext::close);
 234         deviceTerminationPhaseHandler.onDeviceContextLevelDown(deviceInfo);
 235     }
 236
 237     @Override
 238     public Future<RpcResult<GetStatisticsWorkModeOutput>> getStatisticsWorkMode() {
 239         final GetStatisticsWorkModeOutputBuilder smModeOutputBld = new GetStatisticsWorkModeOutputBuilder();
 240         smModeOutputBld.setMode(workMode);
 241         return RpcResultBuilder.success(smModeOutputBld.build()).buildFuture();
 242     }
 243
 244     @Override
 245     public Future<RpcResult<Void>> changeStatisticsWorkMode(ChangeStatisticsWorkModeInput input) {
 246         final Future<RpcResult<Void>> result;
 247         // acquire exclusive access
 248         if (workModeGuard.tryAcquire()) {
 249             final StatisticsWorkMode targetWorkMode = input.getMode();
 250             if (!workMode.equals(targetWorkMode)) {
 251                 isStatisticsPollingOn = !(StatisticsWorkMode.FULLYDISABLED.equals(targetWorkMode));
 252                 // iterate through stats-ctx: propagate mode
 253                 for (Map.Entry<DeviceInfo, StatisticsContext> entry : contexts.entrySet()) {
 254                     final DeviceInfo deviceInfo = entry.getKey();
 255                     final StatisticsContext statisticsContext = entry.getValue();
 256                     final DeviceContext deviceContext = statisticsContext.gainDeviceContext();
 257                     switch (targetWorkMode) {
 258                         case COLLECTALL:
 259                             scheduleNextPolling(statisticsContext.gainDeviceState(), deviceInfo, statisticsContext, new TimeCounter());
 260                             for (final ItemLifeCycleSource lifeCycleSource : deviceContext.getItemLifeCycleSourceRegistry().getLifeCycleSources()) {
 261                                 lifeCycleSource.setItemLifecycleListener(null);
 262                             }
 263                             break;
 264                         case FULLYDISABLED:
 265                             final Optional<Timeout> pollTimeout = statisticsContext.getPollTimeout();
 266                             if (pollTimeout.isPresent()) {
 267                                 pollTimeout.get().cancel();
 268                             }
 269                             for (final ItemLifeCycleSource lifeCycleSource : deviceContext.getItemLifeCycleSourceRegistry().getLifeCycleSources()) {
 270                                 lifeCycleSource.setItemLifecycleListener(statisticsContext.getItemLifeCycleListener());
 271                             }
 272                             break;
 273                         default:
 274                             LOG.warn("Statistics work mode not supported: {}", targetWorkMode);
 275                     }
 276                 }
 277                 workMode = targetWorkMode;
 278             }
 279             workModeGuard.release();
 280             result = RpcResultBuilder.<Void>success().buildFuture();
 281         } else {
 282             result = RpcResultBuilder.<Void>failed()
 283                     .withError(RpcError.ErrorType.APPLICATION, "mode change already in progress")
 284                     .buildFuture();
 285         }
 286         return result;
 287     }
 288
 289     @Override
 290     public void startScheduling(final DeviceInfo deviceInfo) {
 291         if (!isStatisticsPollingOn) {
 292             LOG.info("Statistics are shutdown for device: {}", deviceInfo.getNodeId());
 293             return;
 294         }
 295
 296         final StatisticsContext statisticsContext = contexts.get(deviceInfo);
 297
 298         if (statisticsContext == null) {
 299             LOG.warn("Statistics context not found for device: {}", deviceInfo.getNodeId());
 300             return;
 301         }
 302
 303         if (statisticsContext.isSchedulingEnabled()) {
 304             LOG.debug("Statistics scheduling is already enabled for device: {}", deviceInfo.getNodeId());
 305             return;
 306         }
 307
 308         LOG.info("Scheduling statistics poll for device: {}", deviceInfo.getNodeId());
 309
 310         statisticsContext.setSchedulingEnabled(true);
 311         scheduleNextPolling(
 312                 statisticsContext.gainDeviceState(),
 313                 deviceInfo,
 314                 statisticsContext,
 315                 new TimeCounter()
 316         );
 317     }
 318
 319     @Override
 320     public void stopScheduling(final DeviceInfo deviceInfo) {
 321         if (LOG.isDebugEnabled()) {
 322             LOG.debug("Stopping statistics scheduling for device: {}", deviceInfo.getNodeId());
 323         }
 324
 325         final StatisticsContext statisticsContext = contexts.get(deviceInfo);
 326
 327         if (statisticsContext == null) {
 328             LOG.warn("Statistics context not found for device: {}", deviceInfo.getNodeId());
 329             return;
 330         }
 331
 332         statisticsContext.setSchedulingEnabled(false);
 333     }
 334
 335     @Override
 336     public void close() {
 337         if (controlServiceRegistration != null) {
 338             controlServiceRegistration.close();
 339             controlServiceRegistration = null;
 340         }
 341
 342         for (final Iterator<StatisticsContext> iterator = Iterators.consumingIterator(contexts.values().iterator());
 343                 iterator.hasNext();) {
 344             iterator.next().close();
 345         }
 346     }
 347
 348     @Override
 349     public void setDeviceTerminationPhaseHandler(final DeviceTerminationPhaseHandler handler) {
 350         this.deviceTerminationPhaseHandler = handler;
 351     }
 352
 353     @Override
 354     public void setIsStatisticsPollingOn(boolean isStatisticsPollingOn){
 355         this.isStatisticsPollingOn = isStatisticsPollingOn;
 356     }
 357
 358     public void onDeviceRemoved(DeviceInfo deviceInfo) {
 359         contexts.remove(deviceInfo);
 360         LOG.debug("Statistics context removed for node {}", deviceInfo.getLOGValue());
 361     }
 362
 363     @Override
 364     public void setBasicTimerDelay(final long basicTimerDelay) {
 365         this.basicTimerDelay = basicTimerDelay;
 366     }
 367
 368     @Override
 369     public void setMaximumTimerDelay(final long maximumTimerDelay) {
 370         this.maximumTimerDelay = maximumTimerDelay;
 371     }
 372 }