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