package org.opendaylight.openflowplugin.impl.device;
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Function;
import com.google.common.base.Preconditions;
import com.google.common.base.Verify;
+import com.google.common.util.concurrent.AsyncFunction;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.JdkFutureAdapters;
import com.google.common.util.concurrent.ListenableFuture;
+import io.netty.util.HashedWheelTimer;
import io.netty.util.Timeout;
+import io.netty.util.TimerTask;
import java.math.BigInteger;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Objects;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
import org.opendaylight.controller.md.sal.binding.api.DataBroker;
import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService;
import org.opendaylight.controller.md.sal.binding.api.ReadOnlyTransaction;
import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
-import org.opendaylight.controller.md.sal.common.api.data.TransactionChainClosedException;
import org.opendaylight.mdsal.singleton.common.api.ServiceGroupIdentifier;
import org.opendaylight.openflowjava.protocol.api.connection.ConnectionAdapter;
import org.opendaylight.openflowjava.protocol.api.keys.MessageTypeKey;
+import org.opendaylight.openflowplugin.api.OFConstants;
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.impl.registry.flow.FlowRegistryKeyFactory;
import org.opendaylight.openflowplugin.impl.registry.group.DeviceGroupRegistryImpl;
import org.opendaylight.openflowplugin.impl.registry.meter.DeviceMeterRegistryImpl;
+import org.opendaylight.openflowplugin.impl.rpc.AbstractRequestContext;
import org.opendaylight.openflowplugin.impl.util.DeviceInitializationUtils;
import org.opendaylight.openflowplugin.openflow.md.core.sal.convertor.ConvertorExecutor;
import org.opendaylight.openflowplugin.openflow.md.core.session.SwitchConnectionCookieOFImpl;
import org.opendaylight.yang.gen.v1.urn.opendaylight.packet.service.rev130709.PacketReceived;
import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.FlowCapableNodeConnectorStatisticsData;
import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.FlowCapableNodeConnectorStatisticsDataBuilder;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.role.service.rev150727.OfpRole;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.role.service.rev150727.SalRoleService;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.role.service.rev150727.SetRoleInput;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.role.service.rev150727.SetRoleInputBuilder;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.role.service.rev150727.SetRoleOutput;
import org.opendaylight.yangtools.yang.binding.DataObject;
import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
+import org.opendaylight.yangtools.yang.common.RpcResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-/**
- *
- */
-public class DeviceContextImpl implements DeviceContext, ExtensionConverterProviderKeeper{
+public class DeviceContextImpl implements DeviceContext, ExtensionConverterProviderKeeper {
private static final Logger LOG = LoggerFactory.getLogger(DeviceContextImpl.class);
private static final float LOW_WATERMARK_FACTOR = 0.75f;
// TODO: high water mark factor should be parametrized
private static final float HIGH_WATERMARK_FACTOR = 0.95f;
+
+ // Timeout in seconds after what we will give up on propagating role
+ private static final int SET_ROLE_TIMEOUT = 10;
+
private boolean initialized;
+ private SalRoleService salRoleService = null;
+ private final HashedWheelTimer hashedWheelTimer;
private ConnectionContext primaryConnectionContext;
private final DeviceState deviceState;
private final DataBroker dataBroker;
private final ConvertorExecutor convertorExecutor;
private volatile CONTEXT_STATE state;
private ClusterInitializationPhaseHandler clusterInitializationPhaseHandler;
+ private final DeviceManager myManager;
DeviceContextImpl(
@Nonnull final ConnectionContext primaryConnectionContext,
@Nonnull final TranslatorLibrary translatorLibrary,
@Nonnull final DeviceManager manager,
final ConvertorExecutor convertorExecutor,
- final boolean skipTableFeatures) {
+ final boolean skipTableFeatures,
+ final HashedWheelTimer hashedWheelTimer,
+ final DeviceManager myManager) {
this.primaryConnectionContext = primaryConnectionContext;
this.deviceInfo = primaryConnectionContext.getDeviceInfo();
+ this.hashedWheelTimer = hashedWheelTimer;
+ this.myManager = myManager;
this.deviceState = new DeviceStateImpl();
this.dataBroker = dataBroker;
this.auxiliaryConnectionContexts = new HashMap<>();
@Override
public void initialSubmitTransaction() {
- transactionChainManager.initialSubmitWriteTransaction();
+ if (initialized) {
+ transactionChainManager.initialSubmitWriteTransaction();
+ }
}
@Override
public <T extends DataObject> void writeToTransaction(final LogicalDatastoreType store,
final InstanceIdentifier<T> path,
final T data){
- if (Objects.nonNull(transactionChainManager)) {
+ if (initialized) {
transactionChainManager.writeToTransaction(store, path, data, false);
}
}
public <T extends DataObject> void writeToTransactionWithParentsSlow(final LogicalDatastoreType store,
final InstanceIdentifier<T> path,
final T data){
- if (Objects.nonNull(transactionChainManager)) {
+ if (initialized) {
transactionChainManager.writeToTransaction(store, path, data, true);
}
}
@Override
- public <T extends DataObject> void addDeleteToTxChain(final LogicalDatastoreType store, final InstanceIdentifier<T> path) throws TransactionChainClosedException {
- if (Objects.nonNull(transactionChainManager)) {
+ public <T extends DataObject> void addDeleteToTxChain(final LogicalDatastoreType store, final InstanceIdentifier<T> path) {
+ if (initialized) {
transactionChainManager.addDeleteOperationTotTxChain(store, path);
}
}
@Override
public boolean submitTransaction() {
- return Objects.nonNull(transactionChainManager) && transactionChainManager.submitWriteTransaction();
+ return initialized && transactionChainManager.submitWriteTransaction();
}
@Override
final org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.FlowRemoved flowRemovedNotification =
flowRemovedTranslator.translate(flowRemoved, deviceInfo, null);
- if(!deviceManager.getIsNotificationFlowRemovedOff()) {
+ if(deviceManager.isFlowRemovedNotificationOn()) {
// Trigger off a notification
notificationPublishService.offerNotification(flowRemovedNotification);
} else if(LOG.isDebugEnabled()) {
- LOG.debug("For nodeId={} isNotificationFlowRemovedOff={}", getDeviceInfo().getLOGValue(), deviceManager.getIsNotificationFlowRemovedOff());
+ LOG.debug("For nodeId={} isFlowRemovedNotificationOn={}", getDeviceInfo().getLOGValue(), deviceManager.isFlowRemovedNotificationOn());
}
final ItemLifecycleListener itemLifecycleListener = flowLifeCycleKeeper.getItemLifecycleListener();
if (itemLifecycleListener != null) {
//2. create registry key
- final FlowRegistryKey flowRegKey = FlowRegistryKeyFactory.create(flowRemovedNotification);
+ final FlowRegistryKey flowRegKey = FlowRegistryKeyFactory.create(getDeviceInfo().getVersion(), flowRemovedNotification);
//3. lookup flowId
final FlowDescriptor flowDescriptor = deviceFlowRegistry.retrieveIdForFlow(flowRegKey);
//4. if flowId present:
}
submitTransaction();
} catch (final Exception e) {
- LOG.warn("Error processing port status message: ", e);
+ LOG.warn("Error processing port status message for port {} on device {} : {}", portStatus.getPortNo(),
+ getDeviceInfo().getNodeId().toString(), e);
}
}
return translatorLibrary;
}
- @Override
- public synchronized void close() {
- LOG.debug("closing deviceContext: {}, nodeId:{}",
- getPrimaryConnectionContext().getConnectionAdapter().getRemoteAddress(),
- getDeviceInfo().getLOGValue());
- // NOOP
- throw new UnsupportedOperationException("Autocloseble.close will be removed soon");
- }
-
@Override
public void setCurrentBarrierTimeout(final Timeout timeout) {
barrierTaskTimeout = timeout;
@Override
public void onPublished() {
Verify.verify(CONTEXT_STATE.INITIALIZATION.equals(getState()));
- setState(CONTEXT_STATE.WORKING);
+ this.state = CONTEXT_STATE.WORKING;
primaryConnectionContext.getConnectionAdapter().setPacketInFiltering(false);
for (final ConnectionContext switchAuxConnectionContext : auxiliaryConnectionContexts.values()) {
switchAuxConnectionContext.getConnectionAdapter().setPacketInFiltering(false);
LOG.debug("ConnectionCtx for Node {} is in RIP state.", getDeviceInfo().getLOGValue());
return;
}
- /* Terminate Auxiliary Connection */
+
+ // Terminate Auxiliary Connection
for (final ConnectionContext connectionContext : auxiliaryConnectionContexts.values()) {
LOG.debug("Closing auxiliary connection {}", connectionContext.getNodeId());
connectionContext.closeConnection(false);
}
- /* Terminate Primary Connection */
+
+ // Terminate Primary Connection
getPrimaryConnectionContext().closeConnection(true);
- /* Close all Group Registry */
- deviceGroupRegistry.close();
- deviceFlowRegistry.close();
- deviceMeterRegistry.close();
+
+ // Close all datastore registries
+ if (initialized) {
+ deviceGroupRegistry.close();
+ deviceFlowRegistry.close();
+ deviceMeterRegistry.close();
+ }
}
@Override
public ListenableFuture<Void> shuttingDownDataStoreTransactions() {
- ListenableFuture<Void> future = Futures.immediateFuture(null);
- if (Objects.nonNull(this.transactionChainManager)) {
- future = this.transactionChainManager.shuttingDown();
- }
- return future;
+ return initialized
+ ? this.transactionChainManager.shuttingDown()
+ : Futures.immediateFuture(null);
}
@VisibleForTesting
}
@Override
- public void setState(CONTEXT_STATE state) {
- this.state = state;
- }
-
- @Override
- public ListenableFuture<Void> stopClusterServices(boolean deviceDisconnected) {
- ListenableFuture<Void> future = Futures.immediateFuture(null);
- if (Objects.nonNull(this.transactionChainManager)) {
- future = this.transactionChainManager.deactivateTransactionManager();
+ public ListenableFuture<Void> stopClusterServices(boolean connectionInterrupted) {
+ final ListenableFuture<Void> deactivateTxManagerFuture = initialized
+ ? transactionChainManager.deactivateTransactionManager()
+ : Futures.immediateFuture(null);
+
+ if (!connectionInterrupted) {
+ final ListenableFuture<Void> makeSlaveFuture = Futures.transform(makeDeviceSlave(), new Function<RpcResult<SetRoleOutput>, Void>() {
+ @Nullable
+ @Override
+ public Void apply(@Nullable RpcResult<SetRoleOutput> setRoleOutputRpcResult) {
+ return null;
+ }
+ });
+
+ Futures.addCallback(makeSlaveFuture, new FutureCallback<Void>() {
+ @Override
+ public void onSuccess(@Nullable Void aVoid) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Role SLAVE was successfully propagated on device, node {}", deviceInfo.getLOGValue());
+ }
+ }
+
+ @Override
+ public void onFailure(final Throwable throwable) {
+ LOG.warn("Was not able to set role SLAVE to device on node {} ", deviceInfo.getLOGValue());
+ LOG.trace("Error occurred on device role setting, probably connection loss: ", throwable);
+ }
+ });
+
+ return Futures.transform(deactivateTxManagerFuture, new AsyncFunction<Void, Void>() {
+ @Override
+ public ListenableFuture<Void> apply(Void aVoid) throws Exception {
+ // Add fallback to remove device from operational DS if setting slave fails
+ return Futures.withFallback(makeSlaveFuture, t ->
+ myManager.removeDeviceFromOperationalDS(deviceInfo));
+ }
+ });
+ } else {
+ return Futures.transform(deactivateTxManagerFuture, new AsyncFunction<Void, Void>() {
+ @Override
+ public ListenableFuture<Void> apply(Void aVoid) throws Exception {
+ return myManager.removeDeviceFromOperationalDS(deviceInfo);
+ }
+ });
}
- return future;
}
@Override
return this.deviceInfo;
}
+ @Override
+ public void close() {
+ if (CONTEXT_STATE.TERMINATION.equals(getState())){
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("DeviceContext for node {} is already in TERMINATION state.", getDeviceInfo().getLOGValue());
+ }
+ } else {
+ this.state = CONTEXT_STATE.TERMINATION;
+ }
+ }
+
@Override
public void putLifecycleServiceIntoTxChainManager(final LifecycleService lifecycleService){
- if (Objects.nonNull(this.transactionChainManager)) {
+ if (initialized) {
this.transactionChainManager.setLifecycleService(lifecycleService);
}
}
@Override
public void replaceConnectionContext(final ConnectionContext connectionContext){
// Act like we are initializing the context
- setState(CONTEXT_STATE.INITIALIZATION);
+ this.state = CONTEXT_STATE.INITIALIZATION;
this.primaryConnectionContext = connectionContext;
this.onPublished();
}
return this.skipTableFeatures;
}
+ @Override
+ public void setSalRoleService(@Nonnull SalRoleService salRoleService) {
+ this.salRoleService = salRoleService;
+ }
+
@Override
public void setLifecycleInitializationPhaseHandler(final ClusterInitializationPhaseHandler handler) {
this.clusterInitializationPhaseHandler = handler;
return false;
}
+ Futures.addCallback(sendRoleChangeToDevice(OfpRole.BECOMEMASTER), new RpcResultFutureCallback());
+
return this.clusterInitializationPhaseHandler.onContextInstantiateService(getPrimaryConnectionContext());
}
LOG.debug("Transaction chain manager for node {} created", deviceInfo.getLOGValue());
}
this.transactionChainManager = new TransactionChainManager(dataBroker, deviceInfo);
- this.deviceFlowRegistry = new DeviceFlowRegistryImpl(dataBroker, deviceInfo.getNodeInstanceIdentifier());
+ this.deviceFlowRegistry = new DeviceFlowRegistryImpl(deviceInfo.getVersion(), dataBroker, deviceInfo.getNodeInstanceIdentifier());
this.deviceGroupRegistry = new DeviceGroupRegistryImpl();
this.deviceMeterRegistry = new DeviceMeterRegistryImpl();
this.initialized = true;
}
}
+
+ @Nullable
+ @Override
+ public <T> RequestContext<T> createRequestContext() {
+ return new AbstractRequestContext<T>(deviceInfo.reserveXidForDeviceMessage()) {
+ @Override
+ public void close() {
+ }
+ };
+
+ }
+
+ ListenableFuture<RpcResult<SetRoleOutput>> sendRoleChangeToDevice(final OfpRole newRole) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Sending new role {} to device {}", newRole, deviceInfo.getNodeId());
+ }
+
+ final Future<RpcResult<SetRoleOutput>> setRoleOutputFuture;
+
+ if (deviceInfo.getVersion() >= OFConstants.OFP_VERSION_1_3) {
+ final SetRoleInput setRoleInput = (new SetRoleInputBuilder()).setControllerRole(newRole)
+ .setNode(new NodeRef(deviceInfo.getNodeInstanceIdentifier())).build();
+
+ setRoleOutputFuture = this.salRoleService.setRole(setRoleInput);
+
+ final TimerTask timerTask = timeout -> {
+ if (!setRoleOutputFuture.isDone()) {
+ LOG.warn("New role {} was not propagated to device {} during {} sec", newRole, deviceInfo.getLOGValue(), SET_ROLE_TIMEOUT);
+ setRoleOutputFuture.cancel(true);
+ }
+ };
+
+ hashedWheelTimer.newTimeout(timerTask, SET_ROLE_TIMEOUT, TimeUnit.SECONDS);
+ } else {
+ LOG.info("Device: {} with version: {} does not support role", deviceInfo.getLOGValue(), deviceInfo.getVersion());
+ return Futures.immediateFuture(null);
+ }
+
+ return JdkFutureAdapters.listenInPoolThread(setRoleOutputFuture);
+ }
+
+ @Override
+ public ListenableFuture<RpcResult<SetRoleOutput>> makeDeviceSlave() {
+ return sendRoleChangeToDevice(OfpRole.BECOMESLAVE);
+ }
+
+ private class RpcResultFutureCallback implements FutureCallback<RpcResult<SetRoleOutput>> {
+ @Override
+ public void onSuccess(@Nullable RpcResult<SetRoleOutput> setRoleOutputRpcResult) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Role MASTER was successfully set on device, node {}", deviceInfo.getLOGValue());
+ }
+ }
+
+ @Override
+ public void onFailure(final Throwable throwable) {
+ LOG.warn("Was not able to set MASTER role on device, node {}", deviceInfo.getLOGValue());
+ shutdownConnection();
+ }
+ }
}
Code Review / openflowplugin.git / blobdiff