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=06dc64de2688f7183a14cf35543128e70912402d;hb=d1af0fd5a4053a10917f631bae42970c1960fd20;hp=b6dbb56c61d8bd33b153c2e47a1a8c289d63caf0;hpb=86c852307c5484cc6a5545453b8309315a389af7;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 b6dbb56c61..06dc64de26 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 @@ -9,26 +9,39 @@ package org.opendaylight.openflowplugin.impl.statistics; import com.google.common.annotations.VisibleForTesting; -import com.google.common.base.Optional; import com.google.common.base.Preconditions; import com.google.common.base.Verify; 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; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.CancellationException; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.Future; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; +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.ConnectionException; +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.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId; +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; @@ -40,89 +53,91 @@ import org.opendaylight.yangtools.yang.common.RpcResultBuilder; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import javax.annotation.CheckForNull; -import javax.annotation.Nonnull; -import java.util.Iterator; -import java.util.concurrent.CancellationException; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.Future; -import java.util.concurrent.Semaphore; -import java.util.concurrent.TimeUnit; - public class StatisticsManagerImpl implements StatisticsManager, StatisticsManagerControlService { private static final Logger LOG = LoggerFactory.getLogger(StatisticsManagerImpl.class); private static final long DEFAULT_STATS_TIMEOUT_SEC = 50L; + private final ConvertorExecutor converterExecutor; private DeviceInitializationPhaseHandler deviceInitPhaseHandler; - private DeviceTerminationPhaseHandler deviceTerminPhaseHandler; + private DeviceTerminationPhaseHandler deviceTerminationPhaseHandler; - private final ConcurrentMap contexts = new ConcurrentHashMap<>(); + private final ConcurrentMap contexts = new ConcurrentHashMap<>(); - private static final long basicTimerDelay = 3000; - private static long currentTimerDelay = basicTimerDelay; - private static final long maximumTimerDelay = 900000; //wait max 15 minutes for next statistics + private static long basicTimerDelay; + private static long currentTimerDelay; + private static long maximumTimerDelay; //wait time for next statistics private StatisticsWorkMode workMode = StatisticsWorkMode.COLLECTALL; private final Semaphore workModeGuard = new Semaphore(1, true); - private boolean shuttingDownStatisticsPolling; + private boolean isStatisticsPollingOn; 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 isStatisticsPollingOn, + final HashedWheelTimer hashedWheelTimer, + final ConvertorExecutor convertorExecutor, + final long basicTimerDelay, + final long maximumTimerDelay) { Preconditions.checkArgument(rpcProviderRegistry != null); - this.controlServiceRegistration = Preconditions.checkNotNull(rpcProviderRegistry.addRpcImplementation( - StatisticsManagerControlService.class, this)); - this.shuttingDownStatisticsPolling = shuttingDownStatisticsPolling; - this.conductor = lifecycleConductor; + this.converterExecutor = convertorExecutor; + this.controlServiceRegistration = Preconditions.checkNotNull( + rpcProviderRegistry.addRpcImplementation(StatisticsManagerControlService.class, this) + ); + this.isStatisticsPollingOn = isStatisticsPollingOn; + this.basicTimerDelay = basicTimerDelay; + this.currentTimerDelay = basicTimerDelay; + this.maximumTimerDelay = maximumTimerDelay; + this.hashedWheelTimer = hashedWheelTimer; } @Override - public void onDeviceContextLevelUp(final DeviceInfo deviceInfo) throws Exception { - - final DeviceContext deviceContext = Preconditions.checkNotNull(conductor.getDeviceContext(deviceInfo.getNodeId())); - - final StatisticsContext statisticsContext = new StatisticsContextImpl(deviceInfo.getNodeId(), shuttingDownStatisticsPolling, conductor); - Verify.verify(contexts.putIfAbsent(deviceInfo.getNodeId(), statisticsContext) == null, "StatisticsCtx still not closed for Node {}", deviceInfo.getNodeId()); - - deviceContext.getDeviceState().setDeviceSynchronized(true); - deviceInitPhaseHandler.onDeviceContextLevelUp(deviceInfo); + public void onDeviceContextLevelUp(final DeviceInfo deviceInfo, + final LifecycleService lifecycleService) throws Exception { + + final StatisticsContext statisticsContext = + new StatisticsContextImpl( + deviceInfo, + isStatisticsPollingOn, + lifecycleService, + converterExecutor, + this); + + Verify.verify( + contexts.putIfAbsent(deviceInfo, statisticsContext) == null, + "StatisticsCtx still not closed for Node {}", deviceInfo.getLOGValue() + ); + + lifecycleService.setStatContext(statisticsContext); + lifecycleService.registerDeviceRemovedHandler(this); + deviceInitPhaseHandler.onDeviceContextLevelUp(deviceInfo, lifecycleService); } @VisibleForTesting - void pollStatistics(final DeviceContext deviceContext, - final StatisticsContext statisticsContext, - final TimeCounter timeCounter) { - - final NodeId nodeId = deviceContext.getDeviceState().getNodeId(); + void pollStatistics(final DeviceState deviceState, + final StatisticsContext statisticsContext, + final TimeCounter timeCounter, + final DeviceInfo deviceInfo) { if (!statisticsContext.isSchedulingEnabled()) { - LOG.debug("Disabling statistics scheduling for device: {}", nodeId); - return; - } - - if (!deviceContext.getDeviceState().isValid()) { - LOG.debug("Session is not valid for device: {}", nodeId); + if (LOG.isDebugEnabled()) { + LOG.debug("Disabled statistics scheduling for device: {}", deviceInfo.getNodeId().getValue()); + } return; } - if (!deviceContext.getDeviceState().isStatisticsPollingEnabled()) { - LOG.debug("Statistics polling is currently disabled for device: {}", nodeId); - scheduleNextPolling(deviceContext, statisticsContext, timeCounter); - return; + if (LOG.isDebugEnabled()) { + LOG.debug("POLLING ALL STATISTICS for device: {}", deviceInfo.getNodeId()); } - LOG.debug("POLLING ALL STATISTICS for device: {}", nodeId); timeCounter.markStart(); final ListenableFuture deviceStatisticsCollectionFuture = statisticsContext.gatherDynamicData(); Futures.addCallback(deviceStatisticsCollectionFuture, new FutureCallback() { @@ -130,51 +145,63 @@ public class StatisticsManagerImpl implements StatisticsManager, StatisticsManag public void onSuccess(final Boolean o) { timeCounter.addTimeMark(); calculateTimerDelay(timeCounter); - scheduleNextPolling(deviceContext, statisticsContext, timeCounter); + scheduleNextPolling(deviceState, deviceInfo, statisticsContext, timeCounter); } @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(deviceContext.getDeviceState().getNodeId()); + if (throwable instanceof ConnectionException) { + // ConnectionException is raised by StatisticsContextImpl class when the connections + // move to RIP state. In this particular case, there is no need to reschedule + // because this statistics manager should be closed soon + LOG.warn("Node {} is no more connected, stopping the statistics collection", + deviceInfo.getLOGValue(),throwable); + stopScheduling(deviceInfo); } else { - scheduleNextPolling(deviceContext, statisticsContext, timeCounter); + if (!(throwable instanceof CancellationException)) { + LOG.warn("Unexpected error occurred during statistics collection for node {}, rescheduling " + + "statistics collections", deviceInfo.getLOGValue(),throwable); + } + 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 TimerTask timerTask = new TimerTask() { - - @Override - public void run(final Timeout timeout) throws Exception { - if (!deviceStatisticsCollectionFuture.isDone()) { - LOG.info("Statistics collection for node {} still in progress even after {} secs", nodeId, STATS_TIMEOUT_SEC); - deviceStatisticsCollectionFuture.cancel(true); - } + 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.getLOGValue(), statsTimeoutSec); + deviceStatisticsCollectionFuture.cancel(true); } }; - conductor.newTimeout(timerTask, STATS_TIMEOUT_SEC, TimeUnit.SECONDS); + hashedWheelTimer.newTimeout(timerTask, statsTimeoutSec, TimeUnit.SECONDS); } - private void scheduleNextPolling(final DeviceContext deviceContext, + private void scheduleNextPolling(final DeviceState deviceState, + final DeviceInfo deviceInfo, final StatisticsContext statisticsContext, final TimeCounter timeCounter) { - LOG.debug("SCHEDULING NEXT STATISTICS POLLING for device: {}", deviceContext.getDeviceState().getNodeId()); - if (!shuttingDownStatisticsPolling) { - final Timeout pollTimeout = conductor.newTimeout(new TimerTask() { - @Override - public void run(final Timeout timeout) throws Exception { - pollStatistics(deviceContext, statisticsContext, timeCounter); - } - }, currentTimerDelay, TimeUnit.MILLISECONDS); + if (LOG.isDebugEnabled()) { + LOG.debug("SCHEDULING NEXT STATISTICS POLLING for device: {}", deviceInfo.getNodeId()); + } + if (isStatisticsPollingOn) { + final Timeout pollTimeout = hashedWheelTimer.newTimeout( + timeout -> pollStatistics( + deviceState, + statisticsContext, + timeCounter, + deviceInfo), + currentTimerDelay, + TimeUnit.MILLISECONDS); statisticsContext.setPollTimeout(pollTimeout); } } @@ -202,13 +229,9 @@ public class StatisticsManagerImpl implements StatisticsManager, StatisticsManag } @Override - public void onDeviceContextLevelDown(final DeviceContext deviceContext) { - final StatisticsContext statisticsContext = contexts.remove(deviceContext.getDeviceState().getNodeId()); - if (null != statisticsContext) { - LOG.trace("Removing device context from stack. No more statistics gathering for device: {}", deviceContext.getDeviceState().getNodeId()); - statisticsContext.close(); - } - deviceTerminPhaseHandler.onDeviceContextLevelDown(deviceContext); + public void onDeviceContextLevelDown(final DeviceInfo deviceInfo) { + Optional.ofNullable(contexts.get(deviceInfo)).ifPresent(OFPContext::close); + deviceTerminationPhaseHandler.onDeviceContextLevelDown(deviceInfo); } @Override @@ -225,13 +248,15 @@ public class StatisticsManagerImpl implements StatisticsManager, StatisticsManag if (workModeGuard.tryAcquire()) { final StatisticsWorkMode targetWorkMode = input.getMode(); if (!workMode.equals(targetWorkMode)) { - shuttingDownStatisticsPolling = StatisticsWorkMode.FULLYDISABLED.equals(targetWorkMode); + isStatisticsPollingOn = !(StatisticsWorkMode.FULLYDISABLED.equals(targetWorkMode)); // iterate through stats-ctx: propagate mode - for (final StatisticsContext statisticsContext : contexts.values()) { - final DeviceContext deviceContext = statisticsContext.getDeviceContext(); + for (Map.Entry entry : contexts.entrySet()) { + final DeviceInfo deviceInfo = entry.getKey(); + final StatisticsContext statisticsContext = entry.getValue(); + final DeviceContext deviceContext = statisticsContext.gainDeviceContext(); switch (targetWorkMode) { case COLLECTALL: - scheduleNextPolling(deviceContext, statisticsContext, new TimeCounter()); + scheduleNextPolling(statisticsContext.gainDeviceState(), deviceInfo, statisticsContext, new TimeCounter()); for (final ItemLifeCycleSource lifeCycleSource : deviceContext.getItemLifeCycleSourceRegistry().getLifeCycleSources()) { lifeCycleSource.setItemLifecycleListener(null); } @@ -262,43 +287,45 @@ public class StatisticsManagerImpl implements StatisticsManager, StatisticsManag } @Override - public void startScheduling(final NodeId nodeId) { - if (shuttingDownStatisticsPolling) { - LOG.info("Statistics are shut down for device: {}", nodeId); + public void startScheduling(final DeviceInfo deviceInfo) { + if (!isStatisticsPollingOn) { + LOG.info("Statistics are shutdown for device: {}", deviceInfo.getNodeId()); return; } - final StatisticsContext statisticsContext = contexts.get(nodeId); + final StatisticsContext statisticsContext = contexts.get(deviceInfo); if (statisticsContext == null) { - LOG.warn("Statistics context not found for device: {}", nodeId); + LOG.warn("Statistics context not found for device: {}", deviceInfo.getNodeId()); return; } if (statisticsContext.isSchedulingEnabled()) { - LOG.debug("Statistics scheduling is already enabled for device: {}", nodeId); + LOG.debug("Statistics scheduling is already enabled for device: {}", deviceInfo.getNodeId()); return; } - LOG.info("Scheduling statistics poll for device: {}", nodeId); - final DeviceContext deviceContext = conductor.getDeviceContext(nodeId); - - if (deviceContext == null) { - LOG.warn("Device context not found for device: {}", nodeId); - return; - } + LOG.info("Scheduling statistics poll for device: {}", deviceInfo.getNodeId()); statisticsContext.setSchedulingEnabled(true); - scheduleNextPolling(deviceContext, statisticsContext, new TimeCounter()); + scheduleNextPolling( + statisticsContext.gainDeviceState(), + deviceInfo, + statisticsContext, + new TimeCounter() + ); } @Override - public void stopScheduling(final NodeId nodeId) { - LOG.debug("Stopping statistics scheduling for device: {}", nodeId); - final StatisticsContext statisticsContext = contexts.get(nodeId); + public void stopScheduling(final DeviceInfo deviceInfo) { + if (LOG.isDebugEnabled()) { + LOG.debug("Stopping statistics scheduling for device: {}", deviceInfo.getNodeId()); + } + + final StatisticsContext statisticsContext = contexts.get(deviceInfo); if (statisticsContext == null) { - LOG.warn("Statistics context not found for device: {}", nodeId); + LOG.warn("Statistics context not found for device: {}", deviceInfo.getNodeId()); return; } @@ -311,6 +338,7 @@ public class StatisticsManagerImpl implements StatisticsManager, StatisticsManag controlServiceRegistration.close(); controlServiceRegistration = null; } + for (final Iterator iterator = Iterators.consumingIterator(contexts.values().iterator()); iterator.hasNext();) { iterator.next().close(); @@ -319,6 +347,26 @@ public class StatisticsManagerImpl implements StatisticsManager, StatisticsManag @Override public void setDeviceTerminationPhaseHandler(final DeviceTerminationPhaseHandler handler) { - this.deviceTerminPhaseHandler = handler; + this.deviceTerminationPhaseHandler = handler; + } + + @Override + public void setIsStatisticsPollingOn(boolean isStatisticsPollingOn){ + this.isStatisticsPollingOn = isStatisticsPollingOn; + } + + public void onDeviceRemoved(DeviceInfo deviceInfo) { + contexts.remove(deviceInfo); + LOG.debug("Statistics context removed for node {}", deviceInfo.getLOGValue()); + } + + @Override + public void setBasicTimerDelay(final long basicTimerDelay) { + this.basicTimerDelay = basicTimerDelay; + } + + @Override + public void setMaximumTimerDelay(final long maximumTimerDelay) { + this.maximumTimerDelay = maximumTimerDelay; } }