X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=openflowplugin-impl%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fopenflowplugin%2Fimpl%2Fstatistics%2FStatisticsManagerImpl.java;h=7ed60e5823d2399230256049a14690b5936be7b8;hb=refs%2Fchanges%2F64%2F44664%2F10;hp=8aa3c7aa537d248bc8b58a1cb78991f180fc450b;hpb=31772f5ccce34997b0dc1031b441af4c3ef21557;p=openflowplugin.git diff --git a/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/statistics/StatisticsManagerImpl.java b/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/statistics/StatisticsManagerImpl.java index 8aa3c7aa53..7ed60e5823 100644 --- a/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/statistics/StatisticsManagerImpl.java +++ b/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/statistics/StatisticsManagerImpl.java @@ -15,6 +15,7 @@ import com.google.common.collect.Iterators; import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; +import io.netty.util.HashedWheelTimer; import io.netty.util.Timeout; import io.netty.util.TimerTask; import java.util.Iterator; @@ -26,20 +27,19 @@ import java.util.concurrent.ConcurrentMap; import java.util.concurrent.Future; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; -import javax.annotation.CheckForNull; import javax.annotation.Nonnull; import org.opendaylight.controller.sal.binding.api.BindingAwareBroker; import org.opendaylight.controller.sal.binding.api.RpcProviderRegistry; -import org.opendaylight.openflowplugin.api.openflow.OFPContext; import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext; import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo; import org.opendaylight.openflowplugin.api.openflow.device.DeviceState; import org.opendaylight.openflowplugin.api.openflow.device.handlers.DeviceInitializationPhaseHandler; import org.opendaylight.openflowplugin.api.openflow.device.handlers.DeviceTerminationPhaseHandler; -import org.opendaylight.openflowplugin.api.openflow.lifecycle.LifecycleConductor; +import org.opendaylight.openflowplugin.api.openflow.lifecycle.LifecycleService; import org.opendaylight.openflowplugin.api.openflow.rpc.ItemLifeCycleSource; import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsContext; import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsManager; +import org.opendaylight.openflowplugin.openflow.md.core.sal.convertor.ConvertorExecutor; import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflowplugin.sm.control.rev150812.ChangeStatisticsWorkModeInput; import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflowplugin.sm.control.rev150812.GetStatisticsWorkModeOutput; import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflowplugin.sm.control.rev150812.GetStatisticsWorkModeOutputBuilder; @@ -56,6 +56,7 @@ public class StatisticsManagerImpl implements StatisticsManager, StatisticsManag private static final Logger LOG = LoggerFactory.getLogger(StatisticsManagerImpl.class); private static final long DEFAULT_STATS_TIMEOUT_SEC = 50L; + private final ConvertorExecutor convertorExecutor; private DeviceInitializationPhaseHandler deviceInitPhaseHandler; private DeviceTerminationPhaseHandler deviceTerminPhaseHandler; @@ -68,35 +69,35 @@ public class StatisticsManagerImpl implements StatisticsManager, StatisticsManag private StatisticsWorkMode workMode = StatisticsWorkMode.COLLECTALL; private final Semaphore workModeGuard = new Semaphore(1, true); - private boolean shuttingDownStatisticsPolling; + private boolean isStatisticsPollingEnabled; private BindingAwareBroker.RpcRegistration controlServiceRegistration; - private final LifecycleConductor conductor; + private final HashedWheelTimer hashedWheelTimer; @Override public void setDeviceInitializationPhaseHandler(final DeviceInitializationPhaseHandler handler) { deviceInitPhaseHandler = handler; } - public StatisticsManagerImpl(@CheckForNull final RpcProviderRegistry rpcProviderRegistry, - final boolean shuttingDownStatisticsPolling, - final LifecycleConductor lifecycleConductor) { + public StatisticsManagerImpl(final RpcProviderRegistry rpcProviderRegistry, + final boolean isStatisticsPollingEnabled, + final HashedWheelTimer hashedWheelTimer, + final ConvertorExecutor convertorExecutor) { Preconditions.checkArgument(rpcProviderRegistry != null); + this.convertorExecutor = convertorExecutor; this.controlServiceRegistration = Preconditions.checkNotNull(rpcProviderRegistry.addRpcImplementation( StatisticsManagerControlService.class, this)); - this.shuttingDownStatisticsPolling = shuttingDownStatisticsPolling; - this.conductor = lifecycleConductor; + this.isStatisticsPollingEnabled = isStatisticsPollingEnabled; + this.hashedWheelTimer = hashedWheelTimer; } @Override - public void onDeviceContextLevelUp(final DeviceInfo deviceInfo) throws Exception { + public void onDeviceContextLevelUp(final DeviceInfo deviceInfo, final LifecycleService lifecycleService) throws Exception { - final DeviceContext deviceContext = Preconditions.checkNotNull(conductor.getDeviceContext(deviceInfo)); - - final StatisticsContext statisticsContext = new StatisticsContextImpl(deviceInfo, shuttingDownStatisticsPolling, conductor); - Verify.verify(contexts.putIfAbsent(deviceInfo, statisticsContext) == null, "StatisticsCtx still not closed for Node {}", deviceInfo.getNodeId()); - - deviceInitPhaseHandler.onDeviceContextLevelUp(deviceInfo); + final StatisticsContext statisticsContext = new StatisticsContextImpl(deviceInfo, isStatisticsPollingEnabled, lifecycleService, convertorExecutor, this); + Verify.verify(contexts.putIfAbsent(deviceInfo, statisticsContext) == null, "StatisticsCtx still not closed for Node {}", deviceInfo.getLOGValue()); + lifecycleService.setStatContext(statisticsContext); + deviceInitPhaseHandler.onDeviceContextLevelUp(deviceInfo, lifecycleService); } @VisibleForTesting @@ -106,22 +107,15 @@ public class StatisticsManagerImpl implements StatisticsManager, StatisticsManag final DeviceInfo deviceInfo) { if (!statisticsContext.isSchedulingEnabled()) { - LOG.debug("Disabling statistics scheduling for device: {}", deviceInfo.getNodeId()); - return; - } - - if (!deviceState.isValid()) { - LOG.debug("Session is not valid for device: {}", deviceInfo.getNodeId()); + if (LOG.isDebugEnabled()) { + LOG.debug("Disabled statistics scheduling for device: {}", deviceInfo.getNodeId().getValue()); + } return; } - if (!deviceState.isStatisticsPollingEnabled()) { - LOG.debug("Statistics polling is currently disabled for device: {}", deviceInfo.getNodeId()); - scheduleNextPolling(deviceState, deviceInfo, statisticsContext, timeCounter); - return; + if (LOG.isDebugEnabled()) { + LOG.debug("POLLING ALL STATISTICS for device: {}", deviceInfo.getNodeId()); } - - LOG.debug("POLLING ALL STATISTICS for device: {}", deviceInfo.getNodeId()); timeCounter.markStart(); final ListenableFuture deviceStatisticsCollectionFuture = statisticsContext.gatherDynamicData(); Futures.addCallback(deviceStatisticsCollectionFuture, new FutureCallback() { @@ -135,37 +129,52 @@ public class StatisticsManagerImpl implements StatisticsManager, StatisticsManag @Override public void onFailure(@Nonnull final Throwable throwable) { timeCounter.addTimeMark(); - LOG.warn("Statistics gathering for single node was not successful: {}", throwable.getMessage()); - LOG.trace("Statistics gathering for single node was not successful.. ", throwable); + LOG.warn("Statistics gathering for single node {} was not successful: ", deviceInfo.getLOGValue(), throwable.getMessage()); + if (LOG.isTraceEnabled()) { + LOG.trace("Gathering for node {} failure: ", deviceInfo.getLOGValue(), throwable); + } calculateTimerDelay(timeCounter); if (throwable instanceof CancellationException) { - /** This often happens when something wrong with akka or DS, so closing connection will help to restart device **/ - conductor.closeConnection(deviceInfo); + /* This often happens when something wrong with akka or DS, so closing connection will help to restart device **/ + contexts.get(deviceInfo).getLifecycleService().closeConnection(); } else { - scheduleNextPolling(deviceState, deviceInfo, statisticsContext, timeCounter); + if (throwable instanceof IllegalStateException) { + stopScheduling(deviceInfo); + } else { + scheduleNextPolling(deviceState, deviceInfo, statisticsContext, timeCounter); + } } } }); final long averageTime = TimeUnit.MILLISECONDS.toSeconds(timeCounter.getAverageTimeBetweenMarks()); - final long STATS_TIMEOUT_SEC = averageTime > 0 ? 3 * averageTime : DEFAULT_STATS_TIMEOUT_SEC; + final long statsTimeoutSec = averageTime > 0 ? 3 * averageTime : DEFAULT_STATS_TIMEOUT_SEC; final TimerTask timerTask = timeout -> { if (!deviceStatisticsCollectionFuture.isDone()) { - LOG.info("Statistics collection for node {} still in progress even after {} secs", deviceInfo.getNodeId(), STATS_TIMEOUT_SEC); + LOG.info("Statistics collection for node {} still in progress even after {} secs", deviceInfo.getLOGValue(), statsTimeoutSec); deviceStatisticsCollectionFuture.cancel(true); } }; - conductor.newTimeout(timerTask, STATS_TIMEOUT_SEC, TimeUnit.SECONDS); + hashedWheelTimer.newTimeout(timerTask, statsTimeoutSec, TimeUnit.SECONDS); } private void scheduleNextPolling(final DeviceState deviceState, final DeviceInfo deviceInfo, final StatisticsContext statisticsContext, final TimeCounter timeCounter) { - LOG.debug("SCHEDULING NEXT STATISTICS POLLING for device: {}", deviceInfo.getNodeId()); - if (!shuttingDownStatisticsPolling) { - final Timeout pollTimeout = conductor.newTimeout(timeout -> pollStatistics(deviceState, statisticsContext, timeCounter, deviceInfo), currentTimerDelay, TimeUnit.MILLISECONDS); + if (LOG.isDebugEnabled()) { + LOG.debug("SCHEDULING NEXT STATISTICS POLLING for device: {}", deviceInfo.getNodeId()); + } + if (!isStatisticsPollingEnabled) { + final Timeout pollTimeout = hashedWheelTimer.newTimeout( + timeout -> pollStatistics( + deviceState, + statisticsContext, + timeCounter, + deviceInfo), + currentTimerDelay, + TimeUnit.MILLISECONDS); statisticsContext.setPollTimeout(pollTimeout); } } @@ -196,7 +205,7 @@ public class StatisticsManagerImpl implements StatisticsManager, StatisticsManag public void onDeviceContextLevelDown(final DeviceInfo deviceInfo) { final StatisticsContext statisticsContext = contexts.remove(deviceInfo); if (null != statisticsContext) { - LOG.trace("Removing device context from stack. No more statistics gathering for device: {}", deviceInfo.getNodeId()); + LOG.debug("Removing device context from stack. No more statistics gathering for device: {}", deviceInfo.getLOGValue()); statisticsContext.close(); } deviceTerminPhaseHandler.onDeviceContextLevelDown(deviceInfo); @@ -216,23 +225,26 @@ public class StatisticsManagerImpl implements StatisticsManager, StatisticsManag if (workModeGuard.tryAcquire()) { final StatisticsWorkMode targetWorkMode = input.getMode(); if (!workMode.equals(targetWorkMode)) { - shuttingDownStatisticsPolling = StatisticsWorkMode.FULLYDISABLED.equals(targetWorkMode); + isStatisticsPollingEnabled = StatisticsWorkMode.FULLYDISABLED.equals(targetWorkMode); // iterate through stats-ctx: propagate mode for (Map.Entry entry : contexts.entrySet()) { + final DeviceInfo deviceInfo = entry.getKey(); + final StatisticsContext statisticsContext = entry.getValue(); + final DeviceContext deviceContext = statisticsContext.getLifecycleService().getDeviceContext(); switch (targetWorkMode) { case COLLECTALL: - scheduleNextPolling(conductor.getDeviceContext(entry.getKey()).getDeviceState(), entry.getKey(), entry.getValue(), new TimeCounter()); - for (final ItemLifeCycleSource lifeCycleSource : conductor.getDeviceContext(entry.getKey()).getItemLifeCycleSourceRegistry().getLifeCycleSources()) { + scheduleNextPolling(deviceContext.getDeviceState(), deviceInfo, statisticsContext, new TimeCounter()); + for (final ItemLifeCycleSource lifeCycleSource : deviceContext.getItemLifeCycleSourceRegistry().getLifeCycleSources()) { lifeCycleSource.setItemLifecycleListener(null); } break; case FULLYDISABLED: - final Optional pollTimeout = entry.getValue().getPollTimeout(); + final Optional pollTimeout = statisticsContext.getPollTimeout(); if (pollTimeout.isPresent()) { pollTimeout.get().cancel(); } - for (final ItemLifeCycleSource lifeCycleSource : conductor.getDeviceContext(entry.getKey()).getItemLifeCycleSourceRegistry().getLifeCycleSources()) { - lifeCycleSource.setItemLifecycleListener(entry.getValue().getItemLifeCycleListener()); + for (final ItemLifeCycleSource lifeCycleSource : deviceContext.getItemLifeCycleSourceRegistry().getLifeCycleSources()) { + lifeCycleSource.setItemLifecycleListener(statisticsContext.getItemLifeCycleListener()); } break; default: @@ -253,8 +265,8 @@ public class StatisticsManagerImpl implements StatisticsManager, StatisticsManag @Override public void startScheduling(final DeviceInfo deviceInfo) { - if (shuttingDownStatisticsPolling) { - LOG.info("Statistics are shut down for device: {}", deviceInfo.getNodeId()); + if (isStatisticsPollingEnabled) { + LOG.info("Statistics are shutdown for device: {}", deviceInfo.getNodeId()); return; } @@ -273,12 +285,15 @@ public class StatisticsManagerImpl implements StatisticsManager, StatisticsManag LOG.info("Scheduling statistics poll for device: {}", deviceInfo.getNodeId()); statisticsContext.setSchedulingEnabled(true); - scheduleNextPolling(conductor.getDeviceContext(deviceInfo).getDeviceState(), deviceInfo, statisticsContext, new TimeCounter()); + final DeviceState deviceState = contexts.get(deviceInfo).getLifecycleService().getDeviceContext().getDeviceState(); + scheduleNextPolling(deviceState, deviceInfo, statisticsContext, new TimeCounter()); } @Override public void stopScheduling(final DeviceInfo deviceInfo) { - LOG.debug("Stopping statistics scheduling for device: {}", deviceInfo.getNodeId()); + if (LOG.isDebugEnabled()) { + LOG.debug("Stopping statistics scheduling for device: {}", deviceInfo.getNodeId()); + } final StatisticsContext statisticsContext = contexts.get(deviceInfo); if (statisticsContext == null) { @@ -306,8 +321,4 @@ public class StatisticsManagerImpl implements StatisticsManager, StatisticsManag this.deviceTerminPhaseHandler = handler; } - @Override - public T gainContext(DeviceInfo deviceInfo) { - return (T) contexts.get(deviceInfo); - } }