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.collect.Iterators;
  14 import com.google.common.util.concurrent.FutureCallback;
  15 import com.google.common.util.concurrent.Futures;
  16 import com.google.common.util.concurrent.ListenableFuture;
  17 import io.netty.util.HashedWheelTimer;
  18 import io.netty.util.Timeout;
  19 import io.netty.util.TimerTask;
  20 import java.util.Iterator;
  21 import java.util.Map;
  22 import java.util.concurrent.CancellationException;
  23 import java.util.concurrent.ConcurrentHashMap;
  24 import java.util.concurrent.ConcurrentMap;
  25 import java.util.concurrent.Future;
  26 import java.util.concurrent.Semaphore;
  27 import java.util.concurrent.TimeUnit;
  28 import javax.annotation.Nonnull;
  29 import org.opendaylight.controller.sal.binding.api.BindingAwareBroker;
  30 import org.opendaylight.controller.sal.binding.api.RpcProviderRegistry;
  31 import org.opendaylight.openflowplugin.api.ConnectionException;
  32 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
  33 import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo;
  34 import org.opendaylight.openflowplugin.api.openflow.device.DeviceState;
  35 import org.opendaylight.openflowplugin.api.openflow.rpc.ItemLifeCycleSource;
  36 import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsContext;
  37 import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsManager;
  38 import org.opendaylight.openflowplugin.impl.datastore.MultipartWriterProvider;
  39 import org.opendaylight.openflowplugin.impl.datastore.MultipartWriterProviderFactory;
  40 import org.opendaylight.openflowplugin.openflow.md.core.sal.convertor.ConvertorExecutor;
  41 import org.opendaylight.yang.gen.v1.urn.opendaylight.multipart.types.rev170112.MultipartReply;
  42 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflow.provider.config.rev160510.OpenflowProviderConfig;
  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     @Nonnull
  60     private final OpenflowProviderConfig config;
  61     private final ConvertorExecutor converterExecutor;
  62
  63     private final ConcurrentMap<DeviceInfo, StatisticsContext> contexts = new ConcurrentHashMap<>();
  64
  65     private long currentTimerDelay;
  66     private StatisticsWorkMode workMode = StatisticsWorkMode.COLLECTALL;
  67     private final Semaphore workModeGuard = new Semaphore(1, true);
  68     private BindingAwareBroker.RpcRegistration<StatisticsManagerControlService> controlServiceRegistration;
  69     private final HashedWheelTimer hashedWheelTimer;
  70     private boolean istStatisticsFullyDisabled;
  71
  72     public StatisticsManagerImpl(@Nonnull final OpenflowProviderConfig config,
  73                                  @Nonnull final RpcProviderRegistry rpcProviderRegistry,
  74                                  final HashedWheelTimer hashedWheelTimer,
  75                                  final ConvertorExecutor convertorExecutor) {
  76         this.config = config;
  77         currentTimerDelay = config.getBasicTimerDelay().getValue();
  78         this.converterExecutor = convertorExecutor;
  79         this.controlServiceRegistration = Preconditions.checkNotNull(rpcProviderRegistry
  80                 .addRpcImplementation(StatisticsManagerControlService.class, this));
  81
  82         this.hashedWheelTimer = hashedWheelTimer;
  83     }
  84
  85     @VisibleForTesting
  86     void pollStatistics(final DeviceState deviceState,
  87                         final StatisticsContext statisticsContext,
  88                         final TimeCounter timeCounter,
  89                         final DeviceInfo deviceInfo) {
  90
  91         if (!statisticsContext.isSchedulingEnabled()) {
  92             if (LOG.isDebugEnabled()) {
  93                 LOG.debug("Disabled statistics scheduling for device: {}", deviceInfo.getNodeId().getValue());
  94             }
  95             return;
  96         }
  97
  98         if (LOG.isDebugEnabled()) {
  99             LOG.debug("POLLING ALL STATISTICS for device: {}", deviceInfo.getNodeId());
 100         }
 101
 102         timeCounter.markStart();
 103         final ListenableFuture<Boolean> deviceStatisticsCollectionFuture = statisticsContext.gatherDynamicData();
 104         Futures.addCallback(deviceStatisticsCollectionFuture, new FutureCallback<Boolean>() {
 105             @Override
 106             public void onSuccess(final Boolean o) {
 107                 timeCounter.addTimeMark();
 108                 calculateTimerDelay(timeCounter);
 109                 scheduleNextPolling(deviceState, deviceInfo, statisticsContext, timeCounter);
 110             }
 111
 112             @Override
 113             public void onFailure(@Nonnull final Throwable throwable) {
 114                 timeCounter.addTimeMark();
 115                 calculateTimerDelay(timeCounter);
 116
 117                 if (throwable instanceof ConnectionException) {
 118                     // ConnectionException is raised by StatisticsContextImpl class when the connections
 119                     // move to RIP state. In this particular case, there is no need to reschedule
 120                     // because this statistics manager should be closed soon
 121                     LOG.warn("Device {} is no more connected, stopping the statistics collection. Reason: {}",
 122                             deviceInfo, throwable.getMessage());
 123                     stopScheduling(deviceInfo);
 124                 } else if (throwable instanceof CancellationException) {
 125                     LOG.info("Statistics gathering for device {} was cancelled.", deviceInfo);
 126                 } else {
 127                     LOG.warn("Unexpected error occurred during statistics collection for device {}, rescheduling " +
 128                             "statistics collections", deviceInfo, throwable);
 129
 130                     scheduleNextPolling(deviceState, deviceInfo, statisticsContext, timeCounter);
 131                 }
 132             }
 133         });
 134
 135         final long averageTime = TimeUnit.MILLISECONDS.toSeconds(timeCounter.getAverageTimeBetweenMarks());
 136         final long statsTimeoutSec = averageTime > 0 ? 3 * averageTime : DEFAULT_STATS_TIMEOUT_SEC;
 137         final TimerTask timerTask = timeout -> {
 138             if (!deviceStatisticsCollectionFuture.isDone()) {
 139                 LOG.info("Statistics collection for node {} still in progress even after {} secs", deviceInfo.getLOGValue(), statsTimeoutSec);
 140                 deviceStatisticsCollectionFuture.cancel(true);
 141             }
 142         };
 143
 144         hashedWheelTimer.newTimeout(timerTask, statsTimeoutSec, TimeUnit.SECONDS);
 145     }
 146
 147     private void scheduleNextPolling(final DeviceState deviceState,
 148                                      final DeviceInfo deviceInfo,
 149                                      final StatisticsContext statisticsContext,
 150                                      final TimeCounter timeCounter) {
 151         if (LOG.isDebugEnabled()) {
 152             LOG.debug("SCHEDULING NEXT STATISTICS POLLING for device: {}", deviceInfo.getNodeId());
 153         }
 154
 155         if (isStatisticsEnabled()) {
 156             final Timeout pollTimeout = hashedWheelTimer.newTimeout(
 157                     timeout -> pollStatistics(
 158                             deviceState,
 159                             statisticsContext,
 160                             timeCounter,
 161                             deviceInfo),
 162                     currentTimerDelay,
 163                     TimeUnit.MILLISECONDS);
 164
 165             statisticsContext.setPollTimeout(pollTimeout);
 166         }
 167     }
 168
 169     private boolean isStatisticsEnabled() {
 170         return !istStatisticsFullyDisabled && config.isIsStatisticsPollingOn();
 171     }
 172
 173     @VisibleForTesting
 174     void calculateTimerDelay(final TimeCounter timeCounter) {
 175         final long averageStatisticsGatheringTime = timeCounter.getAverageTimeBetweenMarks();
 176         if (averageStatisticsGatheringTime > currentTimerDelay) {
 177             currentTimerDelay *= 2;
 178             if (currentTimerDelay > config.getMaximumTimerDelay().getValue()) {
 179                 currentTimerDelay = config.getMaximumTimerDelay().getValue();
 180             }
 181         } else {
 182             if (currentTimerDelay > config.getBasicTimerDelay().getValue()) {
 183                 currentTimerDelay /= 2;
 184             } else {
 185                 currentTimerDelay = config.getBasicTimerDelay().getValue();
 186             }
 187         }
 188     }
 189
 190     @VisibleForTesting
 191     long getCurrentTimerDelay() {
 192         return currentTimerDelay;
 193     }
 194
 195     @Override
 196     public Future<RpcResult<GetStatisticsWorkModeOutput>> getStatisticsWorkMode() {
 197         final GetStatisticsWorkModeOutputBuilder smModeOutputBld = new GetStatisticsWorkModeOutputBuilder();
 198         smModeOutputBld.setMode(workMode);
 199         return RpcResultBuilder.success(smModeOutputBld.build()).buildFuture();
 200     }
 201
 202     @Override
 203     public Future<RpcResult<Void>> changeStatisticsWorkMode(ChangeStatisticsWorkModeInput input) {
 204         final Future<RpcResult<Void>> result;
 205         // acquire exclusive access
 206         if (workModeGuard.tryAcquire()) {
 207             final StatisticsWorkMode targetWorkMode = input.getMode();
 208             if (!workMode.equals(targetWorkMode)) {
 209                 istStatisticsFullyDisabled = StatisticsWorkMode.FULLYDISABLED.equals(targetWorkMode);
 210
 211                 // iterate through stats-ctx: propagate mode
 212                 for (Map.Entry<DeviceInfo, StatisticsContext> entry : contexts.entrySet()) {
 213                     final DeviceInfo deviceInfo = entry.getKey();
 214                     final StatisticsContext statisticsContext = entry.getValue();
 215                     final DeviceContext deviceContext = statisticsContext.gainDeviceContext();
 216                     switch (targetWorkMode) {
 217                         case COLLECTALL:
 218                             scheduleNextPolling(statisticsContext.gainDeviceState(), deviceInfo, statisticsContext, new TimeCounter());
 219                             for (final ItemLifeCycleSource lifeCycleSource : deviceContext.getItemLifeCycleSourceRegistry().getLifeCycleSources()) {
 220                                 lifeCycleSource.setItemLifecycleListener(null);
 221                             }
 222                             break;
 223                         case FULLYDISABLED:
 224                             statisticsContext.stopGatheringData();
 225                             for (final ItemLifeCycleSource lifeCycleSource : deviceContext.getItemLifeCycleSourceRegistry().getLifeCycleSources()) {
 226                                 lifeCycleSource.setItemLifecycleListener(statisticsContext.getItemLifeCycleListener());
 227                             }
 228                             break;
 229                         default:
 230                             LOG.warn("Statistics work mode not supported: {}", targetWorkMode);
 231                     }
 232                 }
 233                 workMode = targetWorkMode;
 234             }
 235             workModeGuard.release();
 236             result = RpcResultBuilder.<Void>success().buildFuture();
 237         } else {
 238             result = RpcResultBuilder.<Void>failed()
 239                     .withError(RpcError.ErrorType.APPLICATION, "mode change already in progress")
 240                     .buildFuture();
 241         }
 242         return result;
 243     }
 244
 245     @Override
 246     public void startScheduling(final DeviceInfo deviceInfo) {
 247         if (!isStatisticsEnabled()) {
 248             LOG.info("Statistics are shutdown for device: {}", deviceInfo.getNodeId());
 249             return;
 250         }
 251
 252         final StatisticsContext statisticsContext = contexts.get(deviceInfo);
 253
 254         if (statisticsContext == null) {
 255             LOG.warn("Statistics context not found for device: {}", deviceInfo.getNodeId());
 256             return;
 257         }
 258
 259         if (statisticsContext.isSchedulingEnabled()) {
 260             LOG.debug("Statistics scheduling is already enabled for device: {}", deviceInfo.getNodeId());
 261             return;
 262         }
 263
 264         LOG.info("Scheduling statistics poll for device: {}", deviceInfo.getNodeId());
 265
 266         statisticsContext.setSchedulingEnabled(true);
 267         scheduleNextPolling(
 268                 statisticsContext.gainDeviceState(),
 269                 deviceInfo,
 270                 statisticsContext,
 271                 new TimeCounter()
 272         );
 273     }
 274
 275     @Override
 276     public void stopScheduling(final DeviceInfo deviceInfo) {
 277         if (LOG.isDebugEnabled()) {
 278             LOG.debug("Stopping statistics scheduling for device: {}", deviceInfo.getNodeId());
 279         }
 280
 281         final StatisticsContext statisticsContext = contexts.get(deviceInfo);
 282
 283         if (statisticsContext == null) {
 284             LOG.warn("Statistics context not found for device: {}", deviceInfo.getNodeId());
 285             return;
 286         }
 287
 288         statisticsContext.setSchedulingEnabled(false);
 289     }
 290
 291     @Override
 292     public boolean isUsingReconciliationFramework() {
 293         return config.isUsingReconciliationFramework();
 294     }
 295
 296     @Override
 297     public void close() {
 298         istStatisticsFullyDisabled = true;
 299
 300         if (controlServiceRegistration != null) {
 301             controlServiceRegistration.close();
 302             controlServiceRegistration = null;
 303         }
 304
 305         for (final Iterator<StatisticsContext> iterator = Iterators.consumingIterator(contexts.values().iterator());
 306              iterator.hasNext();) {
 307             iterator.next().close();
 308         }
 309     }
 310
 311     @Override
 312     public StatisticsContext createContext(@Nonnull final DeviceContext deviceContext) {
 313
 314         final MultipartWriterProvider statisticsWriterProvider = MultipartWriterProviderFactory
 315                 .createDefaultProvider(deviceContext);
 316
 317         final StatisticsContext statisticsContext =
 318                 deviceContext.canUseSingleLayerSerialization() ?
 319                         new StatisticsContextImpl<MultipartReply>(
 320                                 isStatisticsEnabled(),
 321                                 deviceContext,
 322                                 converterExecutor,
 323                                 this,
 324                                 statisticsWriterProvider) :
 325                         new StatisticsContextImpl<org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731
 326                                 .MultipartReply>(
 327                                 isStatisticsEnabled(),
 328                                 deviceContext,
 329                                 converterExecutor,
 330                                 this,
 331                                 statisticsWriterProvider);
 332         contexts.putIfAbsent(deviceContext.getDeviceInfo(), statisticsContext);
 333
 334         return statisticsContext;
 335     }
 336
 337     @Override
 338     public void onDeviceRemoved(final DeviceInfo deviceInfo) {
 339         contexts.remove(deviceInfo);
 340         if (LOG.isDebugEnabled()) {
 341             LOG.debug("Statistics context removed for node {}", deviceInfo.getLOGValue());
 342         }
 343     }
 344 }