import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
-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 com.google.common.util.concurrent.SettableFuture;
-import io.netty.util.Timeout;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
-import java.util.Iterator;
import java.util.List;
+import java.util.Objects;
import java.util.Optional;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
-import javax.annotation.concurrent.GuardedBy;
import org.opendaylight.mdsal.common.api.TransactionChainClosedException;
import org.opendaylight.mdsal.singleton.common.api.ServiceGroupIdentifier;
+import org.opendaylight.openflowplugin.api.ConnectionException;
import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
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.RequestContext;
-import org.opendaylight.openflowplugin.api.openflow.device.handlers.ClusterInitializationPhaseHandler;
-import org.opendaylight.openflowplugin.api.openflow.lifecycle.LifecycleService;
-import org.opendaylight.openflowplugin.api.openflow.rpc.listener.ItemLifecycleListener;
+import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChainMastershipState;
+import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChainMastershipWatcher;
import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsContext;
-import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsManager;
+import org.opendaylight.openflowplugin.impl.datastore.MultipartWriterProvider;
import org.opendaylight.openflowplugin.impl.rpc.AbstractRequestContext;
-import org.opendaylight.openflowplugin.impl.rpc.listener.ItemLifecycleListenerImpl;
-import org.opendaylight.openflowplugin.impl.services.RequestContextUtil;
+import org.opendaylight.openflowplugin.impl.services.util.RequestContextUtil;
import org.opendaylight.openflowplugin.impl.statistics.services.dedicated.StatisticsGatheringOnTheFlyService;
import org.opendaylight.openflowplugin.impl.statistics.services.dedicated.StatisticsGatheringService;
import org.opendaylight.openflowplugin.openflow.md.core.sal.convertor.ConvertorExecutor;
import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.common.types.rev130731.MultipartType;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.OfHeader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-class StatisticsContextImpl implements StatisticsContext {
+class StatisticsContextImpl<T extends OfHeader> implements StatisticsContext {
private static final Logger LOG = LoggerFactory.getLogger(StatisticsContextImpl.class);
private static final String CONNECTION_CLOSED = "Connection closed.";
- private final ItemLifecycleListener itemLifeCycleListener;
private final Collection<RequestContext<?>> requestContexts = new HashSet<>();
private final DeviceContext deviceContext;
private final DeviceState devState;
- private final ListenableFuture<Boolean> emptyFuture;
- private final boolean shuttingDownStatisticsPolling;
- private final SinglePurposeMultipartReplyTranslator multipartReplyTranslator;
- private final Object collectionStatTypeLock = new Object();
- @GuardedBy("collectionStatTypeLock")
- private List<MultipartType> collectingStatType;
-
- private StatisticsGatheringService statisticsGatheringService;
- private StatisticsGatheringOnTheFlyService statisticsGatheringOnTheFlyService;
- private Timeout pollTimeout;
+ private final ListeningExecutorService executorService;
+ private final boolean isStatisticsPollingOn;
+ private final ConvertorExecutor convertorExecutor;
+ private final MultipartWriterProvider statisticsWriterProvider;
private final DeviceInfo deviceInfo;
- private final StatisticsManager myManager;
- private final LifecycleService lifecycleService;
-
- private volatile boolean schedulingEnabled;
- private volatile CONTEXT_STATE state;
- private ClusterInitializationPhaseHandler clusterInitializationPhaseHandler;
- private ClusterInitializationPhaseHandler initialSubmitHandler;
-
- StatisticsContextImpl(@Nonnull final DeviceInfo deviceInfo,
- final boolean shuttingDownStatisticsPolling,
- @Nonnull final LifecycleService lifecycleService,
- @Nonnull final ConvertorExecutor convertorExecutor,
- @Nonnull final StatisticsManager myManager) {
- this.lifecycleService = lifecycleService;
- this.deviceContext = lifecycleService.getDeviceContext();
+ private final TimeCounter timeCounter = new TimeCounter();
+ private final long statisticsPollingInterval;
+ private final long maximumPollingDelay;
+ private final boolean isUsingReconciliationFramework;
+ private final AtomicBoolean schedulingEnabled = new AtomicBoolean(true);
+ private final AtomicReference<ListenableFuture<Boolean>> lastDataGatheringRef = new AtomicReference<>();
+ private final AtomicReference<StatisticsPollingService> statisticsPollingServiceRef = new AtomicReference<>();
+ private List<MultipartType> collectingStatType;
+ private StatisticsGatheringService<T> statisticsGatheringService;
+ private StatisticsGatheringOnTheFlyService<T> statisticsGatheringOnTheFlyService;
+ private ContextChainMastershipWatcher contextChainMastershipWatcher;
+
+ StatisticsContextImpl(@Nonnull final DeviceContext deviceContext,
+ @Nonnull final ConvertorExecutor convertorExecutor,
+ @Nonnull final MultipartWriterProvider statisticsWriterProvider,
+ @Nonnull final ListeningExecutorService executorService, boolean isStatisticsPollingOn,
+ boolean isUsingReconciliationFramework, long statisticsPollingInterval,
+ long maximumPollingDelay) {
+ this.deviceContext = deviceContext;
this.devState = Preconditions.checkNotNull(deviceContext.getDeviceState());
- this.shuttingDownStatisticsPolling = shuttingDownStatisticsPolling;
- multipartReplyTranslator = new SinglePurposeMultipartReplyTranslator(convertorExecutor);
- emptyFuture = Futures.immediateFuture(false);
- statisticsGatheringService = new StatisticsGatheringService(this, deviceContext);
- statisticsGatheringOnTheFlyService = new StatisticsGatheringOnTheFlyService(this, deviceContext, convertorExecutor);
- itemLifeCycleListener = new ItemLifecycleListenerImpl(deviceContext);
- statListForCollectingInitialization();
- setState(CONTEXT_STATE.INITIALIZATION);
- this.deviceInfo = deviceInfo;
- this.myManager = myManager;
- }
+ this.executorService = executorService;
+ this.isStatisticsPollingOn = isStatisticsPollingOn;
+ this.convertorExecutor = convertorExecutor;
+ this.deviceInfo = deviceContext.getDeviceInfo();
+ this.statisticsPollingInterval = statisticsPollingInterval;
+ this.maximumPollingDelay = maximumPollingDelay;
+ this.statisticsWriterProvider = statisticsWriterProvider;
+ this.isUsingReconciliationFramework = isUsingReconciliationFramework;
- @Override
- public void statListForCollectingInitialization() {
- synchronized (collectionStatTypeLock) {
- final List<MultipartType> statListForCollecting = new ArrayList<>();
- if (devState.isTableStatisticsAvailable()) {
- statListForCollecting.add(MultipartType.OFPMPTABLE);
- }
- if (devState.isFlowStatisticsAvailable()) {
- statListForCollecting.add(MultipartType.OFPMPFLOW);
- }
- if (devState.isGroupAvailable()) {
- statListForCollecting.add(MultipartType.OFPMPGROUPDESC);
- statListForCollecting.add(MultipartType.OFPMPGROUP);
- }
- if (devState.isMetersAvailable()) {
- statListForCollecting.add(MultipartType.OFPMPMETERCONFIG);
- statListForCollecting.add(MultipartType.OFPMPMETER);
- }
- if (devState.isPortStatisticsAvailable()) {
- statListForCollecting.add(MultipartType.OFPMPPORTSTATS);
- }
- if (devState.isQueueStatisticsAvailable()) {
- statListForCollecting.add(MultipartType.OFPMPQUEUE);
- }
- collectingStatType = ImmutableList.<MultipartType>copyOf(statListForCollecting);
- }
+ statisticsGatheringService = new StatisticsGatheringService<>(this, deviceContext);
+ statisticsGatheringOnTheFlyService = new StatisticsGatheringOnTheFlyService<>(this, deviceContext,
+ convertorExecutor,
+ statisticsWriterProvider);
}
-
@Override
- public ListenableFuture<Boolean> initialGatherDynamicData() {
- return gatherDynamicData(true);
+ public DeviceInfo getDeviceInfo() {
+ return this.deviceInfo;
}
+ @Nonnull
@Override
- public ListenableFuture<Boolean> gatherDynamicData(){
- return gatherDynamicData(false);
- }
-
- private ListenableFuture<Boolean> gatherDynamicData(final boolean initial) {
- if (shuttingDownStatisticsPolling) {
- LOG.debug("Statistics for device {} is not enabled.", getDeviceInfo().getNodeId().getValue());
- return Futures.immediateFuture(Boolean.TRUE);
- }
- final ListenableFuture<Boolean> errorResultFuture = deviceConnectionCheck();
- if (errorResultFuture != null) {
- return errorResultFuture;
- }
- synchronized (collectionStatTypeLock) {
- final Iterator<MultipartType> statIterator = collectingStatType.iterator();
- final SettableFuture<Boolean> settableStatResultFuture = SettableFuture.create();
-
- // write start timestamp to state snapshot container
- StatisticsGatheringUtils.markDeviceStateSnapshotStart(deviceContext);
-
- statChainFuture(statIterator, settableStatResultFuture, initial);
-
- // write end timestamp to state snapshot container
- Futures.addCallback(settableStatResultFuture, new FutureCallback<Boolean>() {
- @Override
- public void onSuccess(@Nullable final Boolean result) {
- StatisticsGatheringUtils.markDeviceStateSnapshotEnd(deviceContext, true);
- }
- @Override
- public void onFailure(final Throwable t) {
- if (!(t instanceof TransactionChainClosedException)) {
- StatisticsGatheringUtils.markDeviceStateSnapshotEnd(deviceContext, false);
- }
- }
- });
- return settableStatResultFuture;
- }
+ public ServiceGroupIdentifier getIdentifier() {
+ return deviceInfo.getServiceIdentifier();
}
- private ListenableFuture<Boolean> chooseStat(final MultipartType multipartType, final boolean initial){
- ListenableFuture<Boolean> result = Futures.immediateCheckedFuture(Boolean.TRUE);
-
- switch (multipartType) {
- case OFPMPFLOW:
- result = collectFlowStatistics(multipartType, initial);
- break;
- case OFPMPTABLE:
- result = collectTableStatistics(multipartType);
- break;
- case OFPMPPORTSTATS:
- result = collectPortStatistics(multipartType);
- break;
- case OFPMPQUEUE:
- result = collectQueueStatistics(multipartType);
- break;
- case OFPMPGROUPDESC:
- result = collectGroupDescStatistics(multipartType);
- break;
- case OFPMPGROUP:
- result = collectGroupStatistics(multipartType);
- break;
- case OFPMPMETERCONFIG:
- result = collectMeterConfigStatistics(multipartType);
- break;
- case OFPMPMETER:
- result = collectMeterStatistics(multipartType);
- break;
- default:
- LOG.warn("Unsupported Statistics type {}", multipartType);
- }
-
- return result;
+ @Override
+ public void registerMastershipWatcher(@Nonnull final ContextChainMastershipWatcher newWatcher) {
+ this.contextChainMastershipWatcher = newWatcher;
}
-
@Override
- public <T> RequestContext<T> createRequestContext() {
- final AbstractRequestContext<T> ret = new AbstractRequestContext<T>(deviceInfo.reserveXidForDeviceMessage()) {
+ public <O> RequestContext<O> createRequestContext() {
+ final AbstractRequestContext<O> ret = new AbstractRequestContext<O>(deviceInfo.reserveXidForDeviceMessage()) {
@Override
public void close() {
requestContexts.remove(this);
}
};
+
requestContexts.add(ret);
return ret;
}
@Override
- public void close() {
- if (CONTEXT_STATE.TERMINATION.equals(getState())) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Statistics context is already in state TERMINATION.");
- }
- } else {
- setState(CONTEXT_STATE.TERMINATION);
- schedulingEnabled = false;
- for (final Iterator<RequestContext<?>> iterator = Iterators.consumingIterator(requestContexts.iterator());
- iterator.hasNext(); ) {
- RequestContextUtil.closeRequestContextWithRpcError(iterator.next(), CONNECTION_CLOSED);
- }
- if (null != pollTimeout && !pollTimeout.isExpired()) {
- pollTimeout.cancel();
- }
- }
+ public void enableGathering() {
+ this.schedulingEnabled.set(true);
}
@Override
- public void setSchedulingEnabled(final boolean schedulingEnabled) {
- this.schedulingEnabled = schedulingEnabled;
+ public void disableGathering() {
+ this.schedulingEnabled.set(false);
}
@Override
- public boolean isSchedulingEnabled() {
- return schedulingEnabled;
- }
+ public void continueInitializationAfterReconciliation() {
+ if (deviceContext.initialSubmitTransaction()) {
+ contextChainMastershipWatcher.onMasterRoleAcquired(deviceInfo, ContextChainMastershipState.INITIAL_SUBMIT);
- @Override
- public void setPollTimeout(final Timeout pollTimeout) {
- this.pollTimeout = pollTimeout;
+ startGatheringData();
+ } else {
+ contextChainMastershipWatcher
+ .onNotAbleToStartMastershipMandatory(deviceInfo, "Initial transaction cannot be submitted.");
+ }
}
@Override
- public Optional<Timeout> getPollTimeout() {
- return Optional.ofNullable(pollTimeout);
- }
+ public void instantiateServiceInstance() {
+ final List<MultipartType> statListForCollecting = new ArrayList<>();
- private void statChainFuture(final Iterator<MultipartType> iterator, final SettableFuture<Boolean> resultFuture, final boolean initial) {
- if (ConnectionContext.CONNECTION_STATE.RIP.equals(deviceContext.getPrimaryConnectionContext().getConnectionState())) {
- final String errMsg = String.format("Device connection is closed for Node : %s.",
- getDeviceInfo().getNodeId());
- LOG.debug(errMsg);
- resultFuture.setException(new IllegalStateException(errMsg));
- return;
+ if (devState.isTableStatisticsAvailable()) {
+ statListForCollecting.add(MultipartType.OFPMPTABLE);
}
- if ( ! iterator.hasNext()) {
- resultFuture.set(Boolean.TRUE);
- LOG.debug("Stats collection successfully finished for node {}", getDeviceInfo().getNodeId());
- return;
+
+ if (devState.isFlowStatisticsAvailable()) {
+ statListForCollecting.add(MultipartType.OFPMPFLOW);
+ }
+
+ if (devState.isGroupAvailable()) {
+ statListForCollecting.add(MultipartType.OFPMPGROUPDESC);
+ statListForCollecting.add(MultipartType.OFPMPGROUP);
}
- final MultipartType nextType = iterator.next();
- LOG.debug("Stats iterating to next type for node {} of type {}", getDeviceInfo().getNodeId(), nextType);
+ if (devState.isMetersAvailable()) {
+ statListForCollecting.add(MultipartType.OFPMPMETERCONFIG);
+ statListForCollecting.add(MultipartType.OFPMPMETER);
+ }
+
+ if (devState.isPortStatisticsAvailable()) {
+ statListForCollecting.add(MultipartType.OFPMPPORTSTATS);
+ }
- final ListenableFuture<Boolean> deviceStatisticsCollectionFuture = chooseStat(nextType, initial);
- Futures.addCallback(deviceStatisticsCollectionFuture, new FutureCallback<Boolean>() {
+ if (devState.isQueueStatisticsAvailable()) {
+ statListForCollecting.add(MultipartType.OFPMPQUEUE);
+ }
+
+ collectingStatType = ImmutableList.copyOf(statListForCollecting);
+ Futures.addCallback(gatherDynamicData(), new InitialSubmitCallback(), MoreExecutors.directExecutor());
+ }
+
+ @Override
+ public ListenableFuture<Void> closeServiceInstance() {
+ return stopGatheringData();
+ }
+
+ @Override
+ public void close() {
+ Futures.addCallback(stopGatheringData(), new FutureCallback<Void>() {
@Override
- public void onSuccess(final Boolean result) {
- statChainFuture(iterator, resultFuture, initial);
+ public void onSuccess(@Nullable final Void result) {
+ requestContexts.forEach(requestContext -> RequestContextUtil
+ .closeRequestContextWithRpcError(requestContext, CONNECTION_CLOSED));
}
+
@Override
- public void onFailure(@Nonnull final Throwable t) {
- resultFuture.setException(t);
+ public void onFailure(final Throwable throwable) {
+ requestContexts.forEach(requestContext -> RequestContextUtil
+ .closeRequestContextWithRpcError(requestContext, CONNECTION_CLOSED));
}
- });
+ }, MoreExecutors.directExecutor());
}
- /**
- * Method checks a device state. It returns null for be able continue. Otherwise it returns immediateFuture
- * which has to be returned from caller too
- *
- * @return
- */
- @VisibleForTesting
- ListenableFuture<Boolean> deviceConnectionCheck() {
- if (!ConnectionContext.CONNECTION_STATE.WORKING.equals(deviceContext.getPrimaryConnectionContext().getConnectionState())) {
- ListenableFuture<Boolean> resultingFuture;
- switch (deviceContext.getPrimaryConnectionContext().getConnectionState()) {
- case RIP:
- final String errMsg = String.format("Device connection doesn't exist anymore. Primary connection status : %s",
- deviceContext.getPrimaryConnectionContext().getConnectionState());
- resultingFuture = Futures.immediateFailedFuture(new Throwable(errMsg));
- break;
- default:
- resultingFuture = Futures.immediateCheckedFuture(Boolean.TRUE);
- break;
- }
- return resultingFuture;
+ private ListenableFuture<Boolean> gatherDynamicData() {
+ if (!isStatisticsPollingOn || !schedulingEnabled.get()) {
+ LOG.debug("Statistics for device {} are not enabled.", getDeviceInfo().getNodeId().getValue());
+ return Futures.immediateFuture(Boolean.TRUE);
}
- return null;
- }
- //TODO: Refactor twice sending deviceContext into gatheringStatistics
- private ListenableFuture<Boolean> collectFlowStatistics(final MultipartType multipartType, final boolean initial) {
- return devState.isFlowStatisticsAvailable() ? StatisticsGatheringUtils.gatherStatistics(
- statisticsGatheringOnTheFlyService,
- getDeviceInfo(),
- /*MultipartType.OFPMPFLOW*/ multipartType,
- deviceContext,
- deviceContext,
- initial, multipartReplyTranslator) : emptyFuture;
- }
+ return this.lastDataGatheringRef.updateAndGet(future -> {
+ // write start timestamp to state snapshot container
+ StatisticsGatheringUtils.markDeviceStateSnapshotStart(deviceInfo, deviceContext);
- private ListenableFuture<Boolean> collectTableStatistics(final MultipartType multipartType) {
- return devState.isTableStatisticsAvailable() ? StatisticsGatheringUtils.gatherStatistics(
- statisticsGatheringService,
- getDeviceInfo(),
- /*MultipartType.OFPMPTABLE*/ multipartType,
- deviceContext,
- deviceContext,
- false, multipartReplyTranslator) : emptyFuture;
- }
+ // recreate gathering future if it should be recreated
+ final ListenableFuture<Boolean> lastDataGathering =
+ Objects.isNull(future) || future.isCancelled() || future.isDone() ? Futures
+ .immediateFuture(Boolean.TRUE) : future;
- private ListenableFuture<Boolean> collectPortStatistics(final MultipartType multipartType) {
- return devState.isPortStatisticsAvailable() ? StatisticsGatheringUtils.gatherStatistics(
- statisticsGatheringService,
- getDeviceInfo(),
- /*MultipartType.OFPMPPORTSTATS*/ multipartType,
- deviceContext,
- deviceContext,
- false, multipartReplyTranslator) : emptyFuture;
- }
+ // build statistics gathering future
+ final ListenableFuture<Boolean> newDataGathering = collectingStatType.stream()
+ .reduce(lastDataGathering, this::statChainFuture,
+ (listenableFuture, asyn) -> Futures.transformAsync(listenableFuture, result -> asyn,
+ MoreExecutors.directExecutor()));
- private ListenableFuture<Boolean> collectQueueStatistics(final MultipartType multipartType) {
- return !devState.isQueueStatisticsAvailable() ? emptyFuture : StatisticsGatheringUtils.gatherStatistics(
- statisticsGatheringService,
- getDeviceInfo(),
- /*MultipartType.OFPMPQUEUE*/ multipartType,
- deviceContext,
- deviceContext,
- false, multipartReplyTranslator);
- }
+ // write end timestamp to state snapshot container
+ Futures.addCallback(newDataGathering, new FutureCallback<Boolean>() {
+ @Override
+ public void onSuccess(final Boolean result) {
+ StatisticsGatheringUtils.markDeviceStateSnapshotEnd(deviceInfo, deviceContext, result);
+ }
- private ListenableFuture<Boolean> collectGroupDescStatistics(final MultipartType multipartType) {
- return devState.isGroupAvailable() ? StatisticsGatheringUtils.gatherStatistics(
- statisticsGatheringService,
- getDeviceInfo(),
- /*MultipartType.OFPMPGROUPDESC*/ multipartType,
- deviceContext,
- deviceContext,
- false, multipartReplyTranslator) : emptyFuture;
- }
+ @Override
+ public void onFailure(final Throwable throwable) {
+ if (!(throwable instanceof TransactionChainClosedException)) {
+ StatisticsGatheringUtils.markDeviceStateSnapshotEnd(deviceInfo, deviceContext, false);
+ }
+ }
+ }, MoreExecutors.directExecutor());
- private ListenableFuture<Boolean> collectGroupStatistics(final MultipartType multipartType) {
- return devState.isGroupAvailable() ? StatisticsGatheringUtils.gatherStatistics(
- statisticsGatheringService,
- getDeviceInfo(),
- /*MultipartType.OFPMPGROUP*/ multipartType,
- deviceContext,
- deviceContext,
- false, multipartReplyTranslator) : emptyFuture;
+ return newDataGathering;
+ });
}
- private ListenableFuture<Boolean> collectMeterConfigStatistics(final MultipartType multipartType) {
- return devState.isMetersAvailable() ? StatisticsGatheringUtils.gatherStatistics(
- statisticsGatheringService,
- getDeviceInfo(),
- /*MultipartType.OFPMPMETERCONFIG*/ multipartType,
- deviceContext,
- deviceContext,
- false, multipartReplyTranslator) : emptyFuture;
- }
+ private ListenableFuture<Boolean> statChainFuture(final ListenableFuture<Boolean> prevFuture,
+ final MultipartType multipartType) {
+ if (ConnectionContext.CONNECTION_STATE.RIP
+ .equals(deviceContext.getPrimaryConnectionContext().getConnectionState())) {
+ final String errMsg = String
+ .format("Device connection for node %s doesn't exist anymore. Primary connection status : %s",
+ getDeviceInfo().getNodeId(),
+ deviceContext.getPrimaryConnectionContext().getConnectionState());
- private ListenableFuture<Boolean> collectMeterStatistics(final MultipartType multipartType) {
- return devState.isMetersAvailable() ? StatisticsGatheringUtils.gatherStatistics(
- statisticsGatheringService,
- getDeviceInfo(),
- /*MultipartType.OFPMPMETER*/ multipartType,
- deviceContext,
- deviceContext,
- false, multipartReplyTranslator) : emptyFuture;
- }
+ return Futures.immediateFailedFuture(new ConnectionException(errMsg));
+ }
- @VisibleForTesting
- void setStatisticsGatheringService(final StatisticsGatheringService statisticsGatheringService) {
- this.statisticsGatheringService = statisticsGatheringService;
- }
+ return Futures.transformAsync(prevFuture, result -> {
+ LOG.debug("Status of previous stat iteration for node {}: {}", deviceInfo, result);
+ LOG.debug("Stats iterating to next type for node {} of type {}", deviceInfo, multipartType);
+ final boolean onTheFly = MultipartType.OFPMPFLOW.equals(multipartType);
+ final boolean supported = collectingStatType.contains(multipartType);
- @VisibleForTesting
- void setStatisticsGatheringOnTheFlyService(final StatisticsGatheringOnTheFlyService
- statisticsGatheringOnTheFlyService) {
- this.statisticsGatheringOnTheFlyService = statisticsGatheringOnTheFlyService;
+ // TODO: Refactor twice sending deviceContext into gatheringStatistics
+ return supported ? StatisticsGatheringUtils
+ .gatherStatistics(onTheFly ? statisticsGatheringOnTheFlyService : statisticsGatheringService,
+ getDeviceInfo(), multipartType, deviceContext, deviceContext, convertorExecutor,
+ statisticsWriterProvider, executorService) : Futures
+ .immediateFuture(Boolean.FALSE);
+ }, MoreExecutors.directExecutor());
}
- @Override
- public ItemLifecycleListener getItemLifeCycleListener () {
- return itemLifeCycleListener;
- }
+ private void startGatheringData() {
+ if (!isStatisticsPollingOn) {
+ return;
+ }
- @Override
- public CONTEXT_STATE getState() {
- return this.state;
- }
+ LOG.info("Starting statistics gathering for node {}", deviceInfo);
+ final StatisticsPollingService statisticsPollingService =
+ new StatisticsPollingService(timeCounter,
+ statisticsPollingInterval,
+ maximumPollingDelay,
+ StatisticsContextImpl.this::gatherDynamicData);
- @Override
- public void setState(CONTEXT_STATE state) {
- this.state = state;
+ schedulingEnabled.set(true);
+ statisticsPollingService.startAsync();
+ this.statisticsPollingServiceRef.set(statisticsPollingService);
}
- @Override
- public ServiceGroupIdentifier getServiceIdentifier() {
- return this.deviceInfo.getServiceIdentifier();
- }
+ private ListenableFuture<Void> stopGatheringData() {
+ LOG.info("Stopping running statistics gathering for node {}", deviceInfo);
+ cancelLastDataGathering();
- @Override
- public DeviceInfo getDeviceInfo() {
- return this.deviceInfo;
+ return Optional.ofNullable(statisticsPollingServiceRef.getAndSet(null)).map(StatisticsPollingService::stop)
+ .orElseGet(() -> Futures.immediateFuture(null));
}
- @Override
- public ListenableFuture<Void> stopClusterServices(boolean deviceDisconnected) {
- myManager.stopScheduling(deviceInfo);
- return Futures.immediateFuture(null);
+ private void cancelLastDataGathering() {
+ final ListenableFuture<Boolean> future = lastDataGatheringRef.getAndSet(null);
+
+ if (Objects.nonNull(future) && !future.isDone() && !future.isCancelled()) {
+ future.cancel(true);
+ }
}
- @Override
- public LifecycleService getLifecycleService() {
- return lifecycleService;
+ @VisibleForTesting
+ void setStatisticsGatheringService(final StatisticsGatheringService<T> statisticsGatheringService) {
+ this.statisticsGatheringService = statisticsGatheringService;
}
- @Override
- public void setLifecycleInitializationPhaseHandler(final ClusterInitializationPhaseHandler handler) {
- this.clusterInitializationPhaseHandler = handler;
+ @VisibleForTesting
+ void setStatisticsGatheringOnTheFlyService(
+ final StatisticsGatheringOnTheFlyService<T> statisticsGatheringOnTheFlyService) {
+ this.statisticsGatheringOnTheFlyService = statisticsGatheringOnTheFlyService;
}
- @Override
- public boolean onContextInstantiateService(final ConnectionContext connectionContext) {
+ private final class InitialSubmitCallback implements FutureCallback<Boolean> {
+ @Override
+ public void onSuccess(@Nullable final Boolean result) {
+ contextChainMastershipWatcher
+ .onMasterRoleAcquired(deviceInfo, ContextChainMastershipState.INITIAL_GATHERING);
- if (connectionContext.getConnectionState().equals(ConnectionContext.CONNECTION_STATE.RIP)) {
- LOG.warn("Connection on device {} was interrupted, will stop starting master services.", deviceInfo.getLOGValue());
- return false;
+ if (!isUsingReconciliationFramework) {
+ continueInitializationAfterReconciliation();
+ }
}
- if (!this.shuttingDownStatisticsPolling) {
-
- LOG.info("Starting statistics context cluster services for node {}", deviceInfo.getLOGValue());
-
- this.statListForCollectingInitialization();
- Futures.addCallback(this.initialGatherDynamicData(), new FutureCallback<Boolean>() {
-
- @Override
- public void onSuccess(@Nullable Boolean aBoolean) {
- initialSubmitHandler.initialSubmitTransaction();
- }
-
- @Override
- public void onFailure(Throwable throwable) {
- LOG.warn("Initial gathering statistics unsuccessful for node {}", deviceInfo.getLOGValue());
- lifecycleService.closeConnection();
- }
- });
-
- myManager.startScheduling(deviceInfo);
-
+ @Override
+ public void onFailure(@Nonnull final Throwable throwable) {
+ contextChainMastershipWatcher.onNotAbleToStartMastershipMandatory(deviceInfo,
+ "Initial gathering statistics "
+ + "unsuccessful: "
+ + throwable.getMessage());
}
-
- return this.clusterInitializationPhaseHandler.onContextInstantiateService(connectionContext);
- }
-
- @Override
- public void setInitialSubmitHandler(final ClusterInitializationPhaseHandler initialSubmitHandler) {
- this.initialSubmitHandler = initialSubmitHandler;
}
}
Code Review / openflowplugin.git / blobdiff