X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=openflowplugin-impl%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fopenflowplugin%2Fimpl%2Fdevice%2FDeviceContextImpl.java;h=302deb93e5c1a720630427708f71f5c79d4a84f9;hb=5ea63885f11c0b0a9af7f83ffa03f62ebc9c1b4f;hp=bcf4259fc9632f3fcdb507737f14fc0255bc380d;hpb=31772f5ccce34997b0dc1031b441af4c3ef21557;p=openflowplugin.git diff --git a/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/device/DeviceContextImpl.java b/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/device/DeviceContextImpl.java index bcf4259fc9..302deb93e5 100644 --- a/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/device/DeviceContextImpl.java +++ b/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/device/DeviceContextImpl.java @@ -8,35 +8,49 @@ 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.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +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.connection.OutboundQueue; 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.connection.OutboundQueueProvider; import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext; import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo; +import org.opendaylight.openflowplugin.api.openflow.device.DeviceManager; import org.opendaylight.openflowplugin.api.openflow.device.DeviceState; import org.opendaylight.openflowplugin.api.openflow.device.MessageTranslator; import org.opendaylight.openflowplugin.api.openflow.device.RequestContext; import org.opendaylight.openflowplugin.api.openflow.device.TranslatorLibrary; import org.opendaylight.openflowplugin.api.openflow.device.Xid; +import org.opendaylight.openflowplugin.api.openflow.device.handlers.ClusterInitializationPhaseHandler; import org.opendaylight.openflowplugin.api.openflow.device.handlers.MultiMsgCollector; -import org.opendaylight.openflowplugin.api.openflow.lifecycle.LifecycleConductor; +import org.opendaylight.openflowplugin.api.openflow.lifecycle.LifecycleService; import org.opendaylight.openflowplugin.api.openflow.md.core.SwitchConnectionDistinguisher; import org.opendaylight.openflowplugin.api.openflow.md.core.TranslatorKey; import org.opendaylight.openflowplugin.api.openflow.registry.ItemLifeCycleRegistry; @@ -60,6 +74,9 @@ import org.opendaylight.openflowplugin.impl.registry.flow.DeviceFlowRegistryImpl 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.experimenter.message.service.rev151020.ExperimenterMessageFromDevBuilder; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode; @@ -88,15 +105,18 @@ import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.experimenter 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 { private static final Logger LOG = LoggerFactory.getLogger(DeviceContextImpl.class); @@ -108,19 +128,27 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi // TODO: high water mark factor should be parametrized private static final float HIGH_WATERMARK_FACTOR = 0.95f; - private final ConnectionContext primaryConnectionContext; + // Timeout in seconds after what we will give up on propagating role + private static final int SET_ROLE_TIMEOUT = 10; + + private boolean initialized; + private static final Long RETRY_DELAY = 100L; + private static final int RETRY_COUNT = 3; + + private SalRoleService salRoleService = null; + private final HashedWheelTimer hashedWheelTimer; + private ConnectionContext primaryConnectionContext; private final DeviceState deviceState; private final DataBroker dataBroker; private final Map auxiliaryConnectionContexts; - private final TransactionChainManager transactionChainManager; - private final DeviceFlowRegistry deviceFlowRegistry; - private final DeviceGroupRegistry deviceGroupRegistry; - private final DeviceMeterRegistry deviceMeterRegistry; + private TransactionChainManager transactionChainManager; + private DeviceFlowRegistry deviceFlowRegistry; + private DeviceGroupRegistry deviceGroupRegistry; + private DeviceMeterRegistry deviceMeterRegistry; private final PacketInRateLimiter packetInLimiter; private final MessageSpy messageSpy; private final ItemLifeCycleKeeper flowLifeCycleKeeper; private NotificationPublishService notificationPublishService; - private final OutboundQueue outboundQueueProvider; private Timeout barrierTaskTimeout; private final MessageTranslator portStatusTranslator; private final MessageTranslator packetInTranslator; @@ -128,60 +156,60 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi private final TranslatorLibrary translatorLibrary; private final ItemLifeCycleRegistry itemLifeCycleSourceRegistry; private ExtensionConverterProvider extensionConverterProvider; - + private final DeviceManager deviceManager; + private boolean skipTableFeatures; + private boolean switchFeaturesMandatory; private final DeviceInfo deviceInfo; - - private volatile CONTEXT_STATE contextState; - - @VisibleForTesting - DeviceContextImpl(@Nonnull final ConnectionContext primaryConnectionContext, - @Nonnull final DeviceState deviceState, - @Nonnull final DataBroker dataBroker, - @Nonnull final LifecycleConductor conductor, - @Nonnull final OutboundQueueProvider outboundQueueProvider, - @Nonnull final TranslatorLibrary translatorLibrary) { - this.primaryConnectionContext = Preconditions.checkNotNull(primaryConnectionContext); - this.deviceState = Preconditions.checkNotNull(deviceState); - this.dataBroker = Preconditions.checkNotNull(dataBroker); - Preconditions.checkNotNull(conductor); - this.outboundQueueProvider = Preconditions.checkNotNull(outboundQueueProvider); - deviceInfo = primaryConnectionContext.getDeviceInfo(); - this.transactionChainManager = new TransactionChainManager(dataBroker, deviceInfo, conductor); - auxiliaryConnectionContexts = new HashMap<>(); - deviceFlowRegistry = new DeviceFlowRegistryImpl(); - deviceGroupRegistry = new DeviceGroupRegistryImpl(); - deviceMeterRegistry = new DeviceMeterRegistryImpl(); - messageSpy = conductor.getMessageIntelligenceAgency(); - - - packetInLimiter = new PacketInRateLimiter(primaryConnectionContext.getConnectionAdapter(), - /*initial*/ 1000, /*initial*/2000, messageSpy, REJECTED_DRAIN_FACTOR); + private final ConvertorExecutor convertorExecutor; + private volatile CONTEXT_STATE state; + private ClusterInitializationPhaseHandler clusterInitializationPhaseHandler; + private final DeviceManager myManager; + + DeviceContextImpl( + @Nonnull final ConnectionContext primaryConnectionContext, + @Nonnull final DataBroker dataBroker, + @Nonnull final MessageSpy messageSpy, + @Nonnull final TranslatorLibrary translatorLibrary, + @Nonnull final DeviceManager manager, + final ConvertorExecutor convertorExecutor, + 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<>(); + this.messageSpy = Preconditions.checkNotNull(messageSpy); + this.deviceManager = manager; + + this.packetInLimiter = new PacketInRateLimiter(primaryConnectionContext.getConnectionAdapter(), + /*initial*/ 1000, /*initial*/2000, this.messageSpy, REJECTED_DRAIN_FACTOR); this.translatorLibrary = translatorLibrary; - portStatusTranslator = translatorLibrary.lookupTranslator( + this.portStatusTranslator = translatorLibrary.lookupTranslator( new TranslatorKey(deviceInfo.getVersion(), PortGrouping.class.getName())); - packetInTranslator = translatorLibrary.lookupTranslator( + this.packetInTranslator = translatorLibrary.lookupTranslator( new TranslatorKey(deviceInfo.getVersion(), PacketIn.class.getName())); - flowRemovedTranslator = translatorLibrary.lookupTranslator( + this.flowRemovedTranslator = translatorLibrary.lookupTranslator( new TranslatorKey(deviceInfo.getVersion(), FlowRemoved.class.getName())); - itemLifeCycleSourceRegistry = new ItemLifeCycleRegistryImpl(); - flowLifeCycleKeeper = new ItemLifeCycleSourceImpl(); - itemLifeCycleSourceRegistry.registerLifeCycleSource(flowLifeCycleKeeper); - contextState = CONTEXT_STATE.INITIALIZATION; - } - - /** - * This method is called from {@link DeviceManagerImpl} only. So we could say "posthandshake process finish" - * and we are able to set a scheduler for an automatic transaction submitting by time (0,5sec). - */ - void initialSubmitTransaction() { - transactionChainManager.initialSubmitWriteTransaction(); + this.itemLifeCycleSourceRegistry = new ItemLifeCycleRegistryImpl(); + this.flowLifeCycleKeeper = new ItemLifeCycleSourceImpl(); + this.itemLifeCycleSourceRegistry.registerLifeCycleSource(flowLifeCycleKeeper); + this.state = CONTEXT_STATE.INITIALIZATION; + this.convertorExecutor = convertorExecutor; + this.skipTableFeatures = skipTableFeatures; + this.initialized = false; } @Override - public Long reserveXidForDeviceMessage() { - return outboundQueueProvider.reserveEntry(); + public void initialSubmitTransaction() { + if (initialized) { + transactionChainManager.initialSubmitWriteTransaction(); + } } @Override @@ -198,7 +226,7 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi public void removeAuxiliaryConnectionContext(final ConnectionContext connectionContext) { final SwitchConnectionDistinguisher connectionDistinguisher = createConnectionDistinguisher(connectionContext); LOG.debug("auxiliary connection dropped: {}, nodeId:{}", connectionContext.getConnectionAdapter() - .getRemoteAddress(), deviceInfo.getNodeId()); + .getRemoteAddress(), getDeviceInfo().getLOGValue()); auxiliaryConnectionContexts.remove(connectionDistinguisher); } @@ -207,11 +235,6 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi return deviceState; } - @Override - public DeviceInfo getDeviceInfo() { - return this.deviceInfo; - } - @Override public ReadOnlyTransaction getReadTransaction() { return dataBroker.newReadOnlyTransaction(); @@ -219,23 +242,32 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi @Override public void writeToTransaction(final LogicalDatastoreType store, - final InstanceIdentifier path, final T data) throws Exception { - transactionChainManager.writeToTransaction(store, path, data, false); + final InstanceIdentifier path, + final T data){ + if (initialized) { + transactionChainManager.writeToTransaction(store, path, data, false); + } } @Override - public void writeToTransactionWithParentsSlow(LogicalDatastoreType store, InstanceIdentifier path, T data) throws Exception { - transactionChainManager.writeToTransaction(store, path, data, true); + public void writeToTransactionWithParentsSlow(final LogicalDatastoreType store, + final InstanceIdentifier path, + final T data){ + if (initialized) { + transactionChainManager.writeToTransaction(store, path, data, true); + } } @Override - public void addDeleteToTxChain(final LogicalDatastoreType store, final InstanceIdentifier path) throws Exception { - transactionChainManager.addDeleteOperationTotTxChain(store, path); + public void addDeleteToTxChain(final LogicalDatastoreType store, final InstanceIdentifier path) throws TransactionChainClosedException { + if (initialized) { + transactionChainManager.addDeleteOperationTotTxChain(store, path); + } } @Override public boolean submitTransaction() { - return transactionChainManager.submitWriteTransaction(); + return initialized && transactionChainManager.submitWriteTransaction(); } @Override @@ -244,7 +276,7 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi } @Override - public ConnectionContext getAuxiliaryConnectiobContexts(final BigInteger cookie) { + public ConnectionContext getAuxiliaryConnectionContexts(final BigInteger cookie) { return auxiliaryConnectionContexts.get(new SwitchConnectionCookieOFImpl(cookie.longValue())); } @@ -281,11 +313,19 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi @Override public void processFlowRemovedMessage(final FlowRemoved flowRemoved) { + //1. translate to general flow (table, priority, match, cookie) + final org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.FlowRemoved flowRemovedNotification = + flowRemovedTranslator.translate(flowRemoved, deviceInfo, null); + + if(deviceManager.isFlowRemovedNotificationOn()) { + // Trigger off a notification + notificationPublishService.offerNotification(flowRemovedNotification); + } else if(LOG.isDebugEnabled()) { + LOG.debug("For nodeId={} isFlowRemovedNotificationOn={}", getDeviceInfo().getLOGValue(), deviceManager.isFlowRemovedNotificationOn()); + } + final ItemLifecycleListener itemLifecycleListener = flowLifeCycleKeeper.getItemLifecycleListener(); if (itemLifecycleListener != null) { - //1. translate to general flow (table, priority, match, cookie) - final org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.FlowRemoved flowRemovedNotification = - flowRemovedTranslator.translate(flowRemoved, deviceInfo, null); //2. create registry key final FlowRegistryKey flowRegKey = FlowRegistryKeyFactory.create(flowRemovedNotification); //3. lookup flowId @@ -299,8 +339,6 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi .child(Flow.class, new FlowKey(flowDescriptor.getFlowId())); // b) notify listener itemLifecycleListener.onRemoved(flowPath); - // c) trigger off a notification - notificationPublishService.offerNotification(flowRemovedNotification); } else { LOG.debug("flow id not found: nodeId={} tableId={}, priority={}", getDeviceInfo().getNodeId(), flowRegKey.getTableId(), flowRemovedNotification.getPriority()); @@ -311,28 +349,77 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi @Override public void processPortStatusMessage(final PortStatusMessage portStatus) { messageSpy.spyMessage(portStatus.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS); - final FlowCapableNodeConnector flowCapableNodeConnector = portStatusTranslator.translate(portStatus, deviceInfo, null); - - final KeyedInstanceIdentifier iiToNodeConnector = provideIIToNodeConnector(portStatus.getPortNo(), portStatus.getVersion()); try { - if (portStatus.getReason().equals(PortReason.OFPPRADD) || portStatus.getReason().equals(PortReason.OFPPRMODIFY)) { - // because of ADD status node connector has to be created - final NodeConnectorBuilder nConnectorBuilder = new NodeConnectorBuilder().setKey(iiToNodeConnector.getKey()); - nConnectorBuilder.addAugmentation(FlowCapableNodeConnectorStatisticsData.class, new FlowCapableNodeConnectorStatisticsDataBuilder().build()); - nConnectorBuilder.addAugmentation(FlowCapableNodeConnector.class, flowCapableNodeConnector); - writeToTransaction(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector, nConnectorBuilder.build()); - } else if (portStatus.getReason().equals(PortReason.OFPPRDELETE)) { - addDeleteToTxChain(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector); - } - submitTransaction(); + updatePortStatusMessage(portStatus); } catch (final Exception e) { - LOG.warn("Error processing port status message: {}", e.getMessage()); + LOG.warn("Error processing port status message for port {} on device {} : {}", portStatus.getPortNo().toString(), + getDeviceInfo().getNodeId().toString(), e); + retryProcessPortStatusMessage(portStatus,RETRY_COUNT); + } + } + + private void retryProcessPortStatusMessage(final PortStatusMessage portStatus, final int retryCount){ + Thread thread = new Thread(new PortStatusRetryRunnable(portStatus, retryCount)); + thread.start(); + } + + private class PortStatusRetryRunnable implements Runnable { + private final PortStatusMessage portStatusMessage; + private final int retryCount; + private int count ; + + PortStatusRetryRunnable (final PortStatusMessage portStatus,final int retry){ + portStatusMessage = portStatus; + retryCount = retry; + count = 0; + } + @Override + public void run() { + while(count < retryCount) { + try { + TimeUnit.MILLISECONDS.sleep(RETRY_DELAY); + } catch (InterruptedException e) { + LOG.warn("Caught interrupted exception while sleeping during a port retry task"); + } + + try { + updatePortStatusMessage(portStatusMessage); + break; + } catch (final Exception e) { + LOG.warn("Error processing port status message for {} on port {} on device {} : {}",count, + portStatusMessage.getPortNo().toString(),getDeviceInfo().getNodeId().toString(), e); + count ++ ; + } + } + LOG.warn("Failed to update port status for port {} on device {} even after 3 retries", + portStatusMessage.getPortNo().toString(),getDeviceInfo().getNodeId().toString()); + } + } + + void updatePortStatusMessage(final PortStatusMessage portStatusMessage){ + final FlowCapableNodeConnector flowCapableNodeConnector = portStatusTranslator.translate(portStatusMessage, + getDeviceInfo(), null); + + final KeyedInstanceIdentifier iiToNodeConnector = + provideIIToNodeConnector(portStatusMessage.getPortNo(), portStatusMessage.getVersion()); + + if (portStatusMessage.getReason().equals(PortReason.OFPPRADD) || portStatusMessage.getReason(). + equals(PortReason.OFPPRMODIFY)) { + // because of ADD status node connector has to be created + final NodeConnectorBuilder nConnectorBuilder = new NodeConnectorBuilder().setKey(iiToNodeConnector.getKey()); + nConnectorBuilder.addAugmentation(FlowCapableNodeConnectorStatisticsData.class, + new FlowCapableNodeConnectorStatisticsDataBuilder().build()); + nConnectorBuilder.addAugmentation(FlowCapableNodeConnector.class, flowCapableNodeConnector); + writeToTransaction(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector, nConnectorBuilder.build()); + } else if (portStatusMessage.getReason().equals(PortReason.OFPPRDELETE)) { + addDeleteToTxChain(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector); } + submitTransaction(); } private KeyedInstanceIdentifier provideIIToNodeConnector(final long portNo, final short version) { - final InstanceIdentifier iiToNodes = deviceInfo.getNodeInstanceIdentifier(); - final BigInteger dataPathId = deviceInfo.getDatapathId(); + final InstanceIdentifier iiToNodes = getDeviceInfo().getNodeInstanceIdentifier(); + final BigInteger dataPathId = getDeviceInfo().getDatapathId(); final NodeConnectorId nodeConnectorId = NodeStaticReplyTranslatorUtil.nodeConnectorId(dataPathId.toString(), portNo, version); return iiToNodes.child(NodeConnector.class, new NodeConnectorKey(nodeConnectorId)); } @@ -341,7 +428,7 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi public void processPacketInMessage(final PacketInMessage packetInMessage) { messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH); final ConnectionAdapter connectionAdapter = getPrimaryConnectionContext().getConnectionAdapter(); - final PacketReceived packetReceived = packetInTranslator.translate(packetInMessage, deviceInfo, null); + final PacketReceived packetReceived = packetInTranslator.translate(packetInMessage, getDeviceInfo(), null); if (packetReceived == null) { LOG.debug("Received a null packet from switch {}", connectionAdapter.getRemoteAddress()); @@ -389,13 +476,13 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi // lookup converter final ExperimenterDataOfChoice vendorData = notification.getExperimenterDataOfChoice(); final MessageTypeKey key = new MessageTypeKey<>( - deviceInfo.getVersion(), + getDeviceInfo().getVersion(), (Class) vendorData.getImplementedInterface()); final ConvertorMessageFromOFJava messageConverter = extensionConverterProvider.getMessageConverter(key); if (messageConverter == null) { LOG.warn("custom converter for {}[OF:{}] not found", notification.getExperimenterDataOfChoice().getImplementedInterface(), - deviceInfo.getVersion()); + getDeviceInfo().getVersion()); return; } // build notification @@ -403,7 +490,7 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi try { messageOfChoice = messageConverter.convert(vendorData, MessagePath.MESSAGE_NOTIFICATION); final ExperimenterMessageFromDevBuilder experimenterMessageFromDevBld = new ExperimenterMessageFromDevBuilder() - .setNode(new NodeRef(deviceInfo.getNodeInstanceIdentifier())) + .setNode(new NodeRef(getDeviceInfo().getNodeInstanceIdentifier())) .setExperimenterMessageOfChoice(messageOfChoice); // publish notificationPublishService.offerNotification(experimenterMessageFromDevBld.build()); @@ -417,15 +504,6 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi return translatorLibrary; } - @Override - public synchronized void close() { - LOG.debug("closing deviceContext: {}, nodeId:{}", - getPrimaryConnectionContext().getConnectionAdapter().getRemoteAddress(), - getDeviceInfo().getNodeId()); - // NOOP - throw new UnsupportedOperationException("Autocloseble.close will be removed soon"); - } - @Override public void setCurrentBarrierTimeout(final Timeout timeout) { barrierTaskTimeout = timeout; @@ -448,8 +526,8 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi @Override public void onPublished() { - Verify.verify(CONTEXT_STATE.INITIALIZATION.equals(contextState)); - contextState = CONTEXT_STATE.WORKING; + Verify.verify(CONTEXT_STATE.INITIALIZATION.equals(getState())); + this.state = CONTEXT_STATE.WORKING; primaryConnectionContext.getConnectionAdapter().setPacketInFiltering(false); for (final ConnectionContext switchAuxConnectionContext : auxiliaryConnectionContexts.values()) { switchAuxConnectionContext.getConnectionAdapter().setPacketInFiltering(false); @@ -483,33 +561,41 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi @Override public synchronized void shutdownConnection() { - LOG.debug("Shutdown method for node {}", deviceInfo.getNodeId()); - if (CONTEXT_STATE.TERMINATION.equals(contextState)) { - LOG.debug("DeviceCtx for Node {} is in termination process.", deviceInfo.getNodeId()); + if (LOG.isDebugEnabled()) { + LOG.debug("Shutdown method for node {}", getDeviceInfo().getLOGValue()); + } + if (CONTEXT_STATE.TERMINATION.equals(getState())) { + LOG.debug("DeviceCtx for Node {} is in termination process.", getDeviceInfo().getLOGValue()); return; } - contextState = CONTEXT_STATE.TERMINATION; if (ConnectionContext.CONNECTION_STATE.RIP.equals(getPrimaryConnectionContext().getConnectionState())) { - LOG.debug("ConnectionCtx for Node {} is in RIP state.", deviceInfo.getNodeId()); + 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 shuttingDownDataStoreTransactions() { - return transactionChainManager.shuttingDown(); + return initialized + ? this.transactionChainManager.shuttingDown() + : Futures.immediateFuture(null); } @VisibleForTesting @@ -517,8 +603,212 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi return this.transactionChainManager; } + @Override + public void setSwitchFeaturesMandatory(boolean switchFeaturesMandatory) { + this.switchFeaturesMandatory = switchFeaturesMandatory; + } + @Override public CONTEXT_STATE getState() { - return this.contextState; + return this.state; + } + + @Override + public ListenableFuture stopClusterServices(boolean connectionInterrupted) { + final ListenableFuture deactivateTxManagerFuture = initialized + ? transactionChainManager.deactivateTransactionManager() + : Futures.immediateFuture(null); + + if (!connectionInterrupted) { + final ListenableFuture makeSlaveFuture = Futures.transform(makeDeviceSlave(), new Function, Void>() { + @Nullable + @Override + public Void apply(@Nullable RpcResult setRoleOutputRpcResult) { + return null; + } + }); + + Futures.addCallback(makeSlaveFuture, new FutureCallback() { + @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() { + @Override + public ListenableFuture 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() { + @Override + public ListenableFuture apply(Void aVoid) throws Exception { + return myManager.removeDeviceFromOperationalDS(deviceInfo); + } + }); + } + } + + @Override + public ServiceGroupIdentifier getServiceIdentifier() { + return this.deviceInfo.getServiceIdentifier(); + } + + @Override + public DeviceInfo getDeviceInfo() { + 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 (initialized) { + this.transactionChainManager.setLifecycleService(lifecycleService); + } + } + + @Override + public void replaceConnectionContext(final ConnectionContext connectionContext){ + // Act like we are initializing the context + this.state = CONTEXT_STATE.INITIALIZATION; + this.primaryConnectionContext = connectionContext; + this.onPublished(); + } + + @Override + public boolean isSkipTableFeatures() { + return this.skipTableFeatures; + } + + @Override + public void setSalRoleService(@Nonnull SalRoleService salRoleService) { + this.salRoleService = salRoleService; + } + + @Override + public void setLifecycleInitializationPhaseHandler(final ClusterInitializationPhaseHandler handler) { + this.clusterInitializationPhaseHandler = handler; + } + + @Override + public boolean onContextInstantiateService(final ConnectionContext connectionContext) { + + if (getPrimaryConnectionContext().getConnectionState().equals(ConnectionContext.CONNECTION_STATE.RIP)) { + LOG.warn("Connection on device {} was interrupted, will stop starting master services.", deviceInfo.getLOGValue()); + return false; + } + + LOG.info("Starting device context cluster services for node {}", deviceInfo.getLOGValue()); + + lazyTransactionManagerInitialization(); + + this.transactionChainManager.activateTransactionManager(); + + try { + DeviceInitializationUtils.initializeNodeInformation(this, switchFeaturesMandatory, this.convertorExecutor); + } catch (ExecutionException | InterruptedException e) { + LOG.warn("Device {} cannot be initialized: ", deviceInfo.getLOGValue(), e); + return false; + } + + Futures.addCallback(sendRoleChangeToDevice(OfpRole.BECOMEMASTER), new RpcResultFutureCallback()); + + return this.clusterInitializationPhaseHandler.onContextInstantiateService(getPrimaryConnectionContext()); + } + + @VisibleForTesting + void lazyTransactionManagerInitialization() { + if (!this.initialized) { + if (LOG.isDebugEnabled()) { + LOG.debug("Transaction chain manager for node {} created", deviceInfo.getLOGValue()); + } + this.transactionChainManager = new TransactionChainManager(dataBroker, deviceInfo); + this.deviceFlowRegistry = new DeviceFlowRegistryImpl(dataBroker, deviceInfo.getNodeInstanceIdentifier()); + this.deviceGroupRegistry = new DeviceGroupRegistryImpl(); + this.deviceMeterRegistry = new DeviceMeterRegistryImpl(); + this.initialized = true; + } + } + + @Nullable + @Override + public RequestContext createRequestContext() { + return new AbstractRequestContext(deviceInfo.reserveXidForDeviceMessage()) { + @Override + public void close() { + } + }; + + } + + ListenableFuture> sendRoleChangeToDevice(final OfpRole newRole) { + if (LOG.isDebugEnabled()) { + LOG.debug("Sending new role {} to device {}", newRole, deviceInfo.getNodeId()); + } + + final Future> 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> makeDeviceSlave() { + return sendRoleChangeToDevice(OfpRole.BECOMESLAVE); + } + + private class RpcResultFutureCallback implements FutureCallback> { + @Override + public void onSuccess(@Nullable RpcResult 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(); + } } }