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=c7141da52bc8e95d05d56b4ec0ffd811be38e4b9;hb=6be16cf9eb42e8c9010b21ae983a3401e8aa88cd;hp=6b53858e870a6f4d5e7b80441b82a704fab2ebe7;hpb=8047885956fa765f6f5ea5eaf370eb65d7433d14;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 6b53858e87..c7141da52b 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,26 +8,36 @@ package org.opendaylight.openflowplugin.impl.device; import com.google.common.annotations.VisibleForTesting; -import com.google.common.base.Preconditions; +import com.google.common.base.Optional; import com.google.common.base.Verify; 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.Collection; 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 java.util.concurrent.TimeoutException; 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.ConnectionException; +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; @@ -37,10 +47,14 @@ 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.LifecycleService; +import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChainMastershipState; +import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChainState; +import org.opendaylight.openflowplugin.api.openflow.lifecycle.MastershipChangeListener; import org.opendaylight.openflowplugin.api.openflow.md.core.SwitchConnectionDistinguisher; import org.opendaylight.openflowplugin.api.openflow.md.core.TranslatorKey; +import org.opendaylight.openflowplugin.api.openflow.md.util.OpenflowVersion; import org.opendaylight.openflowplugin.api.openflow.registry.ItemLifeCycleRegistry; import org.opendaylight.openflowplugin.api.openflow.registry.flow.DeviceFlowRegistry; import org.opendaylight.openflowplugin.api.openflow.registry.flow.FlowDescriptor; @@ -56,52 +70,57 @@ import org.opendaylight.openflowplugin.extension.api.core.extension.ExtensionCon import org.opendaylight.openflowplugin.extension.api.exception.ConversionException; import org.opendaylight.openflowplugin.extension.api.path.MessagePath; import org.opendaylight.openflowplugin.impl.common.ItemLifeCycleSourceImpl; -import org.opendaylight.openflowplugin.impl.common.NodeStaticReplyTranslatorUtil; +import org.opendaylight.openflowplugin.impl.datastore.MultipartWriterProvider; +import org.opendaylight.openflowplugin.impl.datastore.MultipartWriterProviderFactory; +import org.opendaylight.openflowplugin.impl.device.initialization.AbstractDeviceInitializer; +import org.opendaylight.openflowplugin.impl.device.initialization.DeviceInitializerProvider; import org.opendaylight.openflowplugin.impl.device.listener.MultiMsgCollectorImpl; 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.util.DeviceInitializationUtils; +import org.opendaylight.openflowplugin.impl.rpc.AbstractRequestContext; import org.opendaylight.openflowplugin.openflow.md.core.sal.convertor.ConvertorExecutor; import org.opendaylight.openflowplugin.openflow.md.core.session.SwitchConnectionCookieOFImpl; +import org.opendaylight.openflowplugin.openflow.md.util.InventoryDataServiceUtil; 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; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNodeConnector; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowKey; -import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeConnectorId; import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef; import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnector; import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnectorBuilder; import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnectorKey; -import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.common.types.rev130731.PortReason; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.Error; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.ExperimenterMessage; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.FlowRemoved; -import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.MultipartReply; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.OfHeader; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PacketIn; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PacketInMessage; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PortGrouping; +import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PortStatus; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PortStatusMessage; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.experimenter.core.ExperimenterDataOfChoice; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.experimenter.types.rev151020.experimenter.core.message.ExperimenterMessageOfChoice; 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); @@ -112,15 +131,29 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi // TODO: high water mark factor should be parametrized private static final float HIGH_WATERMARK_FACTOR = 0.95f; - private ConnectionContext primaryConnectionContext; + // Timeout in seconds after what we will give up on propagating role + private static final int SET_ROLE_TIMEOUT = 10; + + // Timeout in milliseconds after what we will give up on initializing device + private static final int DEVICE_INIT_TIMEOUT = 9000; + + private static final int LOW_WATERMARK = 1000; + private static final int HIGH_WATERMARK = 2000; + private final MultipartWriterProvider writerProvider; + + private boolean initialized; + + private SalRoleService salRoleService = null; + private final HashedWheelTimer hashedWheelTimer; + private volatile 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 final PacketInRateLimiter packetInLimiter; + private TransactionChainManager transactionChainManager; + private DeviceFlowRegistry deviceFlowRegistry; + private DeviceGroupRegistry deviceGroupRegistry; + private DeviceMeterRegistry deviceMeterRegistry; + private PacketInRateLimiter packetInLimiter; private final MessageSpy messageSpy; private final ItemLifeCycleKeeper flowLifeCycleKeeper; private NotificationPublishService notificationPublishService; @@ -131,59 +164,66 @@ 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 DeviceInfo deviceInfo; private final ConvertorExecutor convertorExecutor; - private volatile CONTEXT_STATE state; - - public DeviceContextImpl( + private volatile ContextState state; + private ClusterInitializationPhaseHandler clusterInitializationPhaseHandler; + private final DeviceManager myManager; + private final DeviceInitializerProvider deviceInitializerProvider; + private final boolean useSingleLayerSerialization; + private boolean hasState; + private boolean isInitialTransactionSubmitted; + + DeviceContextImpl( @Nonnull final ConnectionContext primaryConnectionContext, @Nonnull final DataBroker dataBroker, @Nonnull final MessageSpy messageSpy, @Nonnull final TranslatorLibrary translatorLibrary, - @Nonnull final DeviceManager manager, + @Nonnull final DeviceManager contextManager, final ConvertorExecutor convertorExecutor, - final boolean skipTableFeatures) { - this.primaryConnectionContext = Preconditions.checkNotNull(primaryConnectionContext); + final boolean skipTableFeatures, + final HashedWheelTimer hashedWheelTimer, + final boolean useSingleLayerSerialization, + final DeviceInitializerProvider deviceInitializerProvider) { + + this.primaryConnectionContext = primaryConnectionContext; this.deviceInfo = primaryConnectionContext.getDeviceInfo(); + this.hashedWheelTimer = hashedWheelTimer; + this.deviceInitializerProvider = deviceInitializerProvider; + this.myManager = contextManager; this.deviceState = new DeviceStateImpl(); - this.dataBroker = Preconditions.checkNotNull(dataBroker); - this.transactionChainManager = new TransactionChainManager(dataBroker, deviceInfo); - auxiliaryConnectionContexts = new HashMap<>(); - deviceFlowRegistry = new DeviceFlowRegistryImpl(dataBroker, deviceInfo.getNodeInstanceIdentifier()); - deviceGroupRegistry = new DeviceGroupRegistryImpl(); - deviceMeterRegistry = new DeviceMeterRegistryImpl(); - this.messageSpy = Preconditions.checkNotNull(messageSpy); - this.deviceManager = Preconditions.checkNotNull(manager); - - packetInLimiter = new PacketInRateLimiter(primaryConnectionContext.getConnectionAdapter(), - /*initial*/ 1000, /*initial*/2000, this.messageSpy, REJECTED_DRAIN_FACTOR); + this.dataBroker = dataBroker; + this.auxiliaryConnectionContexts = new HashMap<>(); + this.messageSpy = messageSpy; + + this.packetInLimiter = new PacketInRateLimiter(primaryConnectionContext.getConnectionAdapter(), + /*initial*/ LOW_WATERMARK, /*initial*/HIGH_WATERMARK, 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); - this.state = CONTEXT_STATE.INITIALIZATION; + this.itemLifeCycleSourceRegistry = new ItemLifeCycleRegistryImpl(); + this.flowLifeCycleKeeper = new ItemLifeCycleSourceImpl(); + this.itemLifeCycleSourceRegistry.registerLifeCycleSource(flowLifeCycleKeeper); + this.state = ContextState.INITIALIZATION; this.convertorExecutor = convertorExecutor; this.skipTableFeatures = skipTableFeatures; + this.useSingleLayerSerialization = useSingleLayerSerialization; + this.initialized = false; + writerProvider = MultipartWriterProviderFactory.createDefaultProvider(this); } - /** - * 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). - */ @Override - public void initialSubmitTransaction() { - transactionChainManager.initialSubmitWriteTransaction(); + public boolean initialSubmitTransaction() { + return (initialized &&(isInitialTransactionSubmitted = + transactionChainManager.initialSubmitWriteTransaction())); } @Override @@ -214,28 +254,39 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi return dataBroker.newReadOnlyTransaction(); } + @Override + public boolean isTransactionsEnabled() { + return isInitialTransactionSubmitted; + } + @Override public void writeToTransaction(final LogicalDatastoreType store, final InstanceIdentifier path, final T data){ - transactionChainManager.writeToTransaction(store, path, data, false); + if (initialized) { + transactionChainManager.writeToTransaction(store, path, data, false); + } } @Override public void writeToTransactionWithParentsSlow(final LogicalDatastoreType store, final InstanceIdentifier path, final T data){ - transactionChainManager.writeToTransaction(store, path, data, true); + if (initialized) { + transactionChainManager.writeToTransaction(store, path, data, true); + } } @Override - public void addDeleteToTxChain(final LogicalDatastoreType store, final InstanceIdentifier path) throws TransactionChainClosedException { - transactionChainManager.addDeleteOperationTotTxChain(store, path); + public void addDeleteToTxChain(final LogicalDatastoreType store, final InstanceIdentifier path) { + if (initialized) { + transactionChainManager.addDeleteOperationTotTxChain(store, path); + } } @Override public boolean submitTransaction() { - return transactionChainManager.submitWriteTransaction(); + return initialized && transactionChainManager.submitWriteTransaction(); } @Override @@ -244,7 +295,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())); } @@ -265,18 +316,20 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi @Override public void processReply(final OfHeader ofHeader) { - if (ofHeader instanceof Error) { - messageSpy.spyMessage(ofHeader.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE); - } else { - messageSpy.spyMessage(ofHeader.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS); - } + messageSpy.spyMessage( + ofHeader.getImplementedInterface(), + (ofHeader instanceof Error) + ? MessageSpy.StatisticsGroup.FROM_SWITCH_PUBLISHED_FAILURE + : MessageSpy.StatisticsGroup.FROM_SWITCH_PUBLISHED_SUCCESS); } @Override - public void processReply(final Xid xid, final List ofHeaderList) { - for (final MultipartReply multipartReply : ofHeaderList) { - messageSpy.spyMessage(multipartReply.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE); - } + public void processReply(final Xid xid, final List ofHeaderList) { + ofHeaderList.forEach(header -> messageSpy.spyMessage( + header.getImplementedInterface(), + (header instanceof Error) + ? MessageSpy.StatisticsGroup.FROM_SWITCH_PUBLISHED_FAILURE + : MessageSpy.StatisticsGroup.FROM_SWITCH_PUBLISHED_SUCCESS)); } @Override @@ -285,19 +338,19 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi final org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.FlowRemoved flowRemovedNotification = flowRemovedTranslator.translate(flowRemoved, deviceInfo, null); - if(!deviceManager.getIsNotificationFlowRemovedOff()) { + if(myManager.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={} isNotificationFlowRemovedOn={}", getDeviceInfo().getLOGValue(), myManager.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); + final FlowDescriptor flowDescriptor = deviceFlowRegistry.retrieveDescriptor(flowRegKey); //4. if flowId present: if (flowDescriptor != null) { // a) construct flow path @@ -316,58 +369,70 @@ 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, getDeviceInfo(), 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); + messageSpy.spyMessage(portStatus.getImplementedInterface(), MessageSpy.StatisticsGroup.FROM_SWITCH_PUBLISHED_SUCCESS); + + if (initialized) { + try { + writePortStatusMessage(portStatus); + submitTransaction(); + } catch (final Exception e) { + LOG.warn("Error processing port status message for port {} on device {}", + portStatus.getPortNo(), getDeviceInfo().getLOGValue(), e); } - submitTransaction(); - } catch (final Exception e) { - LOG.warn("Error processing port status message: ", e); + } else if (!hasState) { + primaryConnectionContext.handlePortStatusMessage(portStatus); } } - private KeyedInstanceIdentifier provideIIToNodeConnector(final long portNo, final short version) { - 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)); + private void writePortStatusMessage(final PortStatus portStatusMessage) { + final FlowCapableNodeConnector flowCapableNodeConnector = portStatusTranslator + .translate(portStatusMessage, getDeviceInfo(), null); + + final KeyedInstanceIdentifier iiToNodeConnector = getDeviceInfo() + .getNodeInstanceIdentifier() + .child(NodeConnector.class, new NodeConnectorKey(InventoryDataServiceUtil + .nodeConnectorIdfromDatapathPortNo( + deviceInfo.getDatapathId(), + portStatusMessage.getPortNo(), + OpenflowVersion.get(deviceInfo.getVersion())))); + + if (PortReason.OFPPRADD.equals(portStatusMessage.getReason()) || PortReason.OFPPRMODIFY.equals(portStatusMessage.getReason())) { + // because of ADD status node connector has to be created + writeToTransaction(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector, new NodeConnectorBuilder() + .setKey(iiToNodeConnector.getKey()) + .addAugmentation(FlowCapableNodeConnectorStatisticsData.class, new FlowCapableNodeConnectorStatisticsDataBuilder().build()) + .addAugmentation(FlowCapableNodeConnector.class, flowCapableNodeConnector) + .build()); + } else if (PortReason.OFPPRDELETE.equals(portStatusMessage.getReason())) { + addDeleteToTxChain(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector); + } } @Override public void processPacketInMessage(final PacketInMessage packetInMessage) { - messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH); + messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.StatisticsGroup.FROM_SWITCH); final ConnectionAdapter connectionAdapter = getPrimaryConnectionContext().getConnectionAdapter(); final PacketReceived packetReceived = packetInTranslator.translate(packetInMessage, getDeviceInfo(), null); if (packetReceived == null) { LOG.debug("Received a null packet from switch {}", connectionAdapter.getRemoteAddress()); - messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_TRANSLATE_SRC_FAILURE); + messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.StatisticsGroup.FROM_SWITCH_TRANSLATE_SRC_FAILURE); return; } else { - messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_TRANSLATE_OUT_SUCCESS); + messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.StatisticsGroup.FROM_SWITCH_TRANSLATE_OUT_SUCCESS); } if (!packetInLimiter.acquirePermit()) { LOG.debug("Packet limited"); // TODO: save packet into emergency slot if possible - messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PACKET_IN_LIMIT_REACHED_AND_DROPPED); + messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.StatisticsGroup.FROM_SWITCH_PACKET_IN_LIMIT_REACHED_AND_DROPPED); return; } final ListenableFuture offerNotification = notificationPublishService.offerNotification(packetReceived); if (NotificationPublishService.REJECTED.equals(offerNotification)) { LOG.debug("notification offer rejected"); - messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_NOTIFICATION_REJECTED); + messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.StatisticsGroup.FROM_SWITCH_NOTIFICATION_REJECTED); packetInLimiter.drainLowWaterMark(); packetInLimiter.releasePermit(); return; @@ -376,13 +441,13 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi Futures.addCallback(offerNotification, new FutureCallback() { @Override public void onSuccess(final Object result) { - messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS); + messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.StatisticsGroup.FROM_SWITCH_PUBLISHED_SUCCESS); packetInLimiter.releasePermit(); } @Override public void onFailure(final Throwable t) { - messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_NOTIFICATION_REJECTED); + messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.StatisticsGroup.FROM_SWITCH_NOTIFICATION_REJECTED); LOG.debug("notification offer failed: {}", t.getMessage()); LOG.trace("notification offer failed..", t); packetInLimiter.releasePermit(); @@ -409,7 +474,7 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi try { messageOfChoice = messageConverter.convert(vendorData, MessagePath.MESSAGE_NOTIFICATION); final ExperimenterMessageFromDevBuilder experimenterMessageFromDevBld = new ExperimenterMessageFromDevBuilder() - .setNode(new NodeRef(getDeviceInfo().getNodeInstanceIdentifier())) + .setNode(new NodeRef(getDeviceInfo().getNodeInstanceIdentifier())) .setExperimenterMessageOfChoice(messageOfChoice); // publish notificationPublishService.offerNotification(experimenterMessageFromDevBld.build()); @@ -423,15 +488,6 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi 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; @@ -454,17 +510,14 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi @Override public void onPublished() { - Verify.verify(CONTEXT_STATE.INITIALIZATION.equals(getState())); - setState(CONTEXT_STATE.WORKING); + Verify.verify(ContextState.INITIALIZATION.equals(getState())); + this.state = ContextState.WORKING; primaryConnectionContext.getConnectionAdapter().setPacketInFiltering(false); - for (final ConnectionContext switchAuxConnectionContext : auxiliaryConnectionContexts.values()) { - switchAuxConnectionContext.getConnectionAdapter().setPacketInFiltering(false); - } } @Override - public MultiMsgCollector getMultiMsgCollector(final RequestContext> requestContext) { - return new MultiMsgCollectorImpl(this, requestContext); + public MultiMsgCollector getMultiMsgCollector(final RequestContext> requestContext) { + return new MultiMsgCollectorImpl<>(this, requestContext); } @Override @@ -489,33 +542,41 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi @Override public synchronized void shutdownConnection() { - LOG.debug("Shutdown method for node {}", getDeviceInfo().getLOGValue()); - if (CONTEXT_STATE.TERMINATION.equals(getState())) { + if (LOG.isDebugEnabled()) { + LOG.debug("Shutdown method for node {}", getDeviceInfo().getLOGValue()); + } + if (ContextState.TERMINATION.equals(getState())) { LOG.debug("DeviceCtx for Node {} is in termination process.", getDeviceInfo().getLOGValue()); return; } - setState(CONTEXT_STATE.TERMINATION); if (ConnectionContext.CONNECTION_STATE.RIP.equals(getPrimaryConnectionContext().getConnectionState())) { 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 @@ -529,26 +590,15 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi } @Override - public CONTEXT_STATE getState() { + public ContextState getState() { return this.state; } @Override - public void setState(CONTEXT_STATE state) { - this.state = state; - } - - @Override - public void startupClusterServices() throws ExecutionException, InterruptedException { - LOG.debug("Initializing transaction chain manager for node {}", getDeviceInfo().getLOGValue()); - this.transactionChainManager.activateTransactionManager(); - LOG.debug("Waiting to get node {} information", getDeviceInfo().getLOGValue()); - DeviceInitializationUtils.initializeNodeInformation(this, switchFeaturesMandatory, this.convertorExecutor); - } - - @Override - public ListenableFuture stopClusterServices(boolean deviceDisconnected) { - return this.transactionChainManager.deactivateTransactionManager(); + public ListenableFuture stopClusterServices() { + return initialized + ? transactionChainManager.deactivateTransactionManager() + : Futures.immediateFuture(null); } @Override @@ -562,20 +612,215 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi } @Override - public void putLifecycleServiceIntoTxChainManager(final LifecycleService lifecycleService){ - this.transactionChainManager.setLifecycleService(lifecycleService); + public void close() { + //NOOP } @Override - public void replaceConnectionContext(final ConnectionContext connectionContext){ - // Act like we are initializing the context - setState(CONTEXT_STATE.INITIALIZATION); - this.primaryConnectionContext = connectionContext; - this.onPublished(); + public boolean canUseSingleLayerSerialization() { + return useSingleLayerSerialization && getDeviceInfo().getVersion() >= OFConstants.OFP_VERSION_1_3; } @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 MastershipChangeListener mastershipChangeListener) { + + LOG.info("Starting device context cluster services for node {}", deviceInfo.getLOGValue()); + lazyTransactionManagerInitialization(); + + try { + final List portStatusMessages = primaryConnectionContext + .retrieveAndClearPortStatusMessages(); + + portStatusMessages.forEach(this::writePortStatusMessage); + submitTransaction(); + } catch (final Exception ex) { + LOG.warn("Error processing port status messages from device {}", getDeviceInfo().getLOGValue(), ex); + return false; + } + + try { + final java.util.Optional initializer = deviceInitializerProvider + .lookup(deviceInfo.getVersion()); + + if (initializer.isPresent()) { + initializer + .get() + .initialize(this, switchFeaturesMandatory, writerProvider, convertorExecutor) + .get(DEVICE_INIT_TIMEOUT, TimeUnit.MILLISECONDS); + } else { + throw new ExecutionException(new ConnectionException("Unsupported version " + deviceInfo.getVersion())); + } + } catch (ExecutionException | InterruptedException | TimeoutException ex) { + LOG.warn("Device {} cannot be initialized: ", deviceInfo.getLOGValue(), ex); + return false; + } + + Futures.addCallback(sendRoleChangeToDevice(OfpRole.BECOMEMASTER), + new RpcResultFutureCallback(mastershipChangeListener)); + + final ListenableFuture>> deviceFlowRegistryFill = getDeviceFlowRegistry().fill(); + Futures.addCallback(deviceFlowRegistryFill, + new DeviceFlowRegistryCallback(deviceFlowRegistryFill, mastershipChangeListener)); + + return this.clusterInitializationPhaseHandler.onContextInstantiateService(mastershipChangeListener); + } + + @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(deviceInfo.getVersion(), dataBroker, deviceInfo.getNodeInstanceIdentifier()); + this.deviceGroupRegistry = new DeviceGroupRegistryImpl(); + this.deviceMeterRegistry = new DeviceMeterRegistryImpl(); + this.transactionChainManager.activateTransactionManager(); + this.initialized = true; + } + } + + @Nullable + @Override + public RequestContext createRequestContext() { + return new AbstractRequestContext(deviceInfo.reserveXidForDeviceMessage()) { + @Override + public void close() { + } + }; + + } + + private 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); + } + + @Override + public void onStateAcquired(final ContextChainState state) { + hasState = true; + } + + private class RpcResultFutureCallback implements FutureCallback> { + + private final MastershipChangeListener mastershipChangeListener; + + RpcResultFutureCallback(final MastershipChangeListener mastershipChangeListener) { + this.mastershipChangeListener = mastershipChangeListener; + } + + @Override + public void onSuccess(@Nullable RpcResult setRoleOutputRpcResult) { + this.mastershipChangeListener.onMasterRoleAcquired( + deviceInfo, + ContextChainMastershipState.MASTER_ON_DEVICE + ); + if (LOG.isDebugEnabled()) { + LOG.debug("Role MASTER was successfully set on device, node {}", deviceInfo.getLOGValue()); + } + } + + @Override + public void onFailure(final Throwable throwable) { + mastershipChangeListener.onNotAbleToStartMastershipMandatory( + deviceInfo, + "Was not able to set MASTER role on device"); + } + } + + private class DeviceFlowRegistryCallback implements FutureCallback>> { + private final ListenableFuture>> deviceFlowRegistryFill; + private final MastershipChangeListener mastershipChangeListener; + + DeviceFlowRegistryCallback( + ListenableFuture>> deviceFlowRegistryFill, + MastershipChangeListener mastershipChangeListener) { + this.deviceFlowRegistryFill = deviceFlowRegistryFill; + this.mastershipChangeListener = mastershipChangeListener; + } + + @Override + public void onSuccess(@Nullable List> result) { + if (LOG.isDebugEnabled()) { + // Count all flows we read from datastore for debugging purposes. + // This number do not always represent how many flows were actually added + // to DeviceFlowRegistry, because of possible duplicates. + long flowCount = Optional.fromNullable(result).asSet().stream() + .flatMap(Collection::stream) + .filter(Objects::nonNull) + .flatMap(flowCapableNodeOptional -> flowCapableNodeOptional.asSet().stream()) + .filter(Objects::nonNull) + .filter(flowCapableNode -> Objects.nonNull(flowCapableNode.getTable())) + .flatMap(flowCapableNode -> flowCapableNode.getTable().stream()) + .filter(Objects::nonNull) + .filter(table -> Objects.nonNull(table.getFlow())) + .flatMap(table -> table.getFlow().stream()) + .filter(Objects::nonNull) + .count(); + + LOG.debug("Finished filling flow registry with {} flows for node: {}", flowCount, deviceInfo.getLOGValue()); + } + this.mastershipChangeListener.onMasterRoleAcquired(deviceInfo, ContextChainMastershipState.INITIAL_FLOW_REGISTRY_FILL); + } + + @Override + public void onFailure(Throwable t) { + if (deviceFlowRegistryFill.isCancelled()) { + if (LOG.isDebugEnabled()) { + LOG.debug("Cancelled filling flow registry with flows for node: {}", deviceInfo.getLOGValue()); + } + } else { + LOG.warn("Failed filling flow registry with flows for node: {} with exception: {}", deviceInfo.getLOGValue(), t); + } + mastershipChangeListener.onNotAbleToStartMastership( + deviceInfo, + "Was not able to fill flow registry on device", + false); + } + } + }