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.ListenableFuture;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.atomic.AtomicLong;
-import javax.annotation.CheckForNull;
import javax.annotation.Nonnull;
import org.opendaylight.controller.md.sal.binding.api.DataBroker;
import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService;
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.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.MultiMsgCollector;
+import org.opendaylight.openflowplugin.api.openflow.lifecycle.LifecycleConductor;
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;
import org.opendaylight.openflowplugin.api.openflow.registry.group.DeviceGroupRegistry;
import org.opendaylight.openflowplugin.api.openflow.registry.meter.DeviceMeterRegistry;
import org.opendaylight.openflowplugin.api.openflow.rpc.ItemLifeCycleKeeper;
-import org.opendaylight.openflowplugin.api.openflow.rpc.RpcContext;
import org.opendaylight.openflowplugin.api.openflow.rpc.listener.ItemLifecycleListener;
-import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsContext;
import org.opendaylight.openflowplugin.api.openflow.statistics.ofpspecific.MessageSpy;
import org.opendaylight.openflowplugin.extension.api.ConvertorMessageFromOFJava;
import org.opendaylight.openflowplugin.extension.api.ExtensionConverterProviderKeeper;
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.RpcContextImpl;
-import org.opendaylight.openflowplugin.impl.util.DeviceInitializationUtils;
-import org.opendaylight.openflowplugin.impl.util.MdSalRegistrationUtils;
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;
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.NodeConnectorRef;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
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.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.yangtools.yang.binding.DataObject;
import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
private final MessageTranslator<PacketInMessage, PacketReceived> packetInTranslator;
private final MessageTranslator<FlowRemoved, org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.FlowRemoved> flowRemovedTranslator;
private final TranslatorLibrary translatorLibrary;
- private final Map<Long, NodeConnectorRef> nodeConnectorCache;
private final ItemLifeCycleRegistry itemLifeCycleSourceRegistry;
- private RpcContext rpcContext;
private ExtensionConverterProvider extensionConverterProvider;
- private final boolean switchFeaturesMandatory;
- private StatisticsContext statisticsContext;
-
- private final NodeId nodeId;
+ private final DeviceInfo deviceInfo;
private volatile DEVICE_CONTEXT_STATE deviceCtxState;
- private boolean isStatisticsRpcEnabled;
-
@VisibleForTesting
DeviceContextImpl(@Nonnull final ConnectionContext primaryConnectionContext,
@Nonnull final DeviceState deviceState,
@Nonnull final DataBroker dataBroker,
- @Nonnull final MessageSpy _messageSpy,
+ @Nonnull final LifecycleConductor conductor,
@Nonnull final OutboundQueueProvider outboundQueueProvider,
- @Nonnull final TranslatorLibrary translatorLibrary,
- final boolean switchFeaturesMandatory) {
- this.switchFeaturesMandatory = switchFeaturesMandatory;
+ @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);
- this.transactionChainManager = new TransactionChainManager(dataBroker, deviceState);
+ deviceInfo = primaryConnectionContext.getDeviceInfo();
+ this.transactionChainManager = new TransactionChainManager(dataBroker, deviceInfo, conductor);
auxiliaryConnectionContexts = new HashMap<>();
deviceFlowRegistry = new DeviceFlowRegistryImpl();
deviceGroupRegistry = new DeviceGroupRegistryImpl();
deviceMeterRegistry = new DeviceMeterRegistryImpl();
- messageSpy = _messageSpy;
+ messageSpy = conductor.getMessageIntelligenceAgency();
+
packetInLimiter = new PacketInRateLimiter(primaryConnectionContext.getConnectionAdapter(),
/*initial*/ 1000, /*initial*/2000, messageSpy, REJECTED_DRAIN_FACTOR);
this.translatorLibrary = translatorLibrary;
portStatusTranslator = translatorLibrary.lookupTranslator(
- new TranslatorKey(deviceState.getVersion(), PortGrouping.class.getName()));
+ new TranslatorKey(deviceInfo.getVersion(), PortGrouping.class.getName()));
packetInTranslator = translatorLibrary.lookupTranslator(
- new TranslatorKey(deviceState.getVersion(), PacketIn.class.getName()));
+ new TranslatorKey(deviceInfo.getVersion(), PacketIn.class.getName()));
flowRemovedTranslator = translatorLibrary.lookupTranslator(
- new TranslatorKey(deviceState.getVersion(), FlowRemoved.class.getName()));
-
-
- nodeConnectorCache = new ConcurrentHashMap<>();
+ new TranslatorKey(deviceInfo.getVersion(), FlowRemoved.class.getName()));
itemLifeCycleSourceRegistry = new ItemLifeCycleRegistryImpl();
flowLifeCycleKeeper = new ItemLifeCycleSourceImpl();
itemLifeCycleSourceRegistry.registerLifeCycleSource(flowLifeCycleKeeper);
deviceCtxState = DEVICE_CONTEXT_STATE.INITIALIZATION;
-
- nodeId = primaryConnectionContext.getNodeId();
}
/**
public void removeAuxiliaryConnectionContext(final ConnectionContext connectionContext) {
final SwitchConnectionDistinguisher connectionDistinguisher = createConnectionDistinguisher(connectionContext);
LOG.debug("auxiliary connection dropped: {}, nodeId:{}", connectionContext.getConnectionAdapter()
- .getRemoteAddress(), nodeId);
+ .getRemoteAddress(), deviceInfo.getNodeId());
auxiliaryConnectionContexts.remove(connectionDistinguisher);
}
}
@Override
- public ReadOnlyTransaction getReadTransaction() {
- return dataBroker.newReadOnlyTransaction();
- }
-
- @Override
- public ListenableFuture<Void> onClusterRoleChange(final OfpRole oldRole, @CheckForNull final OfpRole role) {
- LOG.trace("onClusterRoleChange {} for node:", role, nodeId);
- Preconditions.checkArgument(role != null);
- if (role.equals(oldRole)) {
- LOG.debug("Demanded role change for device {} is not changed. OldRole: {}, NewRole {}", nodeId, oldRole, role);
- return Futures.immediateFuture(null);
- }
- if (OfpRole.BECOMEMASTER.equals(role)) {
- return onDeviceTakeClusterLeadership();
- } else if (OfpRole.BECOMESLAVE.equals(role)) {
- return onDeviceLostClusterLeadership();
- } else {
- LOG.warn("Unknown OFCluster Role {} for Node {}", role, nodeId);
- if (null != rpcContext) {
- MdSalRegistrationUtils.unregisterServices(rpcContext);
- }
- return transactionChainManager.deactivateTransactionManager();
- }
+ public DeviceInfo getDeviceInfo() {
+ return this.deviceInfo;
}
@Override
- public ListenableFuture<Void> onDeviceLostClusterLeadership() {
- LOG.trace("onDeviceLostClusterLeadership for node: {}", nodeId);
- if (null != rpcContext) {
- MdSalRegistrationUtils.registerSlaveServices(rpcContext, OfpRole.BECOMESLAVE);
- }
- return transactionChainManager.deactivateTransactionManager();
+ public ReadOnlyTransaction getReadTransaction() {
+ return dataBroker.newReadOnlyTransaction();
}
@Override
- public ListenableFuture<Void> onDeviceTakeClusterLeadership() {
- LOG.trace("onDeviceTakeClusterLeadership for node: {}", nodeId);
- /* validation */
- if (statisticsContext == null) {
- final String errMsg = String.format("DeviceCtx %s is up but we are missing StatisticsContext", nodeId);
- LOG.warn(errMsg);
- return Futures.immediateFailedFuture(new IllegalStateException(errMsg));
- }
- if (rpcContext == null) {
- final String errMsg = String.format("DeviceCtx %s is up but we are missing RpcContext", nodeId);
- LOG.warn(errMsg);
- return Futures.immediateFailedFuture(new IllegalStateException(errMsg));
- }
- /* Routed RPC registration */
- MdSalRegistrationUtils.registerMasterServices(getRpcContext(), DeviceContextImpl.this, OfpRole.BECOMEMASTER);
-
- if (isStatisticsRpcEnabled) {
- MdSalRegistrationUtils.registerStatCompatibilityServices(getRpcContext(), this,
- notificationPublishService, new AtomicLong());
- }
-
- /* Prepare init info collecting */
- getDeviceState().setDeviceSynchronized(false);
- transactionChainManager.activateTransactionManager();
- /* Init Collecting NodeInfo */
- final ListenableFuture<Void> initCollectingDeviceInfo = DeviceInitializationUtils.initializeNodeInformation(
- DeviceContextImpl.this, switchFeaturesMandatory);
- /* Init Collecting StatInfo */
- final ListenableFuture<Boolean> statPollFuture = Futures.transform(initCollectingDeviceInfo,
- new AsyncFunction<Void, Boolean>() {
-
- @Override
- public ListenableFuture<Boolean> apply(@Nonnull final Void input) throws Exception {
- getStatisticsContext().statListForCollectingInitialization();
- return getStatisticsContext().gatherDynamicData();
- }
- });
-
- return Futures.transform(statPollFuture, new Function<Boolean, Void>() {
-
- @Override
- public Void apply(final Boolean input) {
- if (ConnectionContext.CONNECTION_STATE.RIP.equals(getPrimaryConnectionContext().getConnectionState())) {
- final String errMsg = String.format("We lost connection for Device %s, context has to be closed.",
- getDeviceState().getNodeId());
- LOG.warn(errMsg);
- throw new IllegalStateException(errMsg);
- }
- if (!input) {
- final String errMsg = String.format("Get Initial Device %s information fails",
- getDeviceState().getNodeId());
- LOG.warn(errMsg);
- throw new IllegalStateException(errMsg);
- }
- LOG.debug("Get Initial Device {} information is successful", nodeId);
- getDeviceState().setDeviceSynchronized(true);
- initialSubmitTransaction();
- getDeviceState().setStatisticsPollingEnabledProp(true);
- return null;
- }
- });
+ public <T extends DataObject> void writeToTransaction(final LogicalDatastoreType store,
+ final InstanceIdentifier<T> path, final T data) throws Exception {
+ transactionChainManager.writeToTransaction(store, path, data, false);
}
@Override
- public <T extends DataObject> void writeToTransaction(final LogicalDatastoreType store,
- final InstanceIdentifier<T> path, final T data) throws Exception {
- transactionChainManager.writeToTransaction(store, path, data);
+ public <T extends DataObject> void writeToTransactionWithParentsSlow(LogicalDatastoreType store, InstanceIdentifier<T> path, T data) throws Exception {
+ transactionChainManager.writeToTransaction(store, path, data, true);
}
@Override
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, this, null);
+ flowRemovedTranslator.translate(flowRemoved, deviceInfo, null);
//2. create registry key
final FlowRegistryKey flowRegKey = FlowRegistryKeyFactory.create(flowRemovedNotification);
//3. lookup flowId
//4. if flowId present:
if (flowDescriptor != null) {
// a) construct flow path
- final KeyedInstanceIdentifier<Flow, FlowKey> flowPath = getDeviceState().getNodeInstanceIdentifier()
+ final KeyedInstanceIdentifier<Flow, FlowKey> flowPath = getDeviceInfo().getNodeInstanceIdentifier()
.augmentation(FlowCapableNode.class)
.child(Table.class, flowDescriptor.getTableKey())
.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={}",
- getDeviceState().getNodeId(), flowRegKey.getTableId(), flowRemovedNotification.getPriority());
+ getDeviceInfo().getNodeId(), flowRegKey.getTableId(), flowRemovedNotification.getPriority());
}
}
}
@Override
public void processPortStatusMessage(final PortStatusMessage portStatus) {
messageSpy.spyMessage(portStatus.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
- final FlowCapableNodeConnector flowCapableNodeConnector = portStatusTranslator.translate(portStatus, this, null);
+ final FlowCapableNodeConnector flowCapableNodeConnector = portStatusTranslator.translate(portStatus, deviceInfo, null);
final KeyedInstanceIdentifier<NodeConnector, NodeConnectorKey> iiToNodeConnector = provideIIToNodeConnector(portStatus.getPortNo(), portStatus.getVersion());
try {
}
private KeyedInstanceIdentifier<NodeConnector, NodeConnectorKey> provideIIToNodeConnector(final long portNo, final short version) {
- final InstanceIdentifier<Node> iiToNodes = deviceState.getNodeInstanceIdentifier();
- final BigInteger dataPathId = deviceState.getFeatures().getDatapathId();
+ final InstanceIdentifier<Node> iiToNodes = deviceInfo.getNodeInstanceIdentifier();
+ final BigInteger dataPathId = deviceInfo.getDatapathId();
final NodeConnectorId nodeConnectorId = NodeStaticReplyTranslatorUtil.nodeConnectorId(dataPathId.toString(), portNo, version);
return iiToNodes.child(NodeConnector.class, new NodeConnectorKey(nodeConnectorId));
}
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, this, null);
+ final PacketReceived packetReceived = packetInTranslator.translate(packetInMessage, deviceInfo, null);
if (packetReceived == null) {
LOG.debug("Received a null packet from switch {}", connectionAdapter.getRemoteAddress());
// lookup converter
final ExperimenterDataOfChoice vendorData = notification.getExperimenterDataOfChoice();
final MessageTypeKey<? extends ExperimenterDataOfChoice> key = new MessageTypeKey<>(
- deviceState.getVersion(),
+ deviceInfo.getVersion(),
(Class<? extends ExperimenterDataOfChoice>) vendorData.getImplementedInterface());
final ConvertorMessageFromOFJava<ExperimenterDataOfChoice, MessagePath> messageConverter = extensionConverterProvider.getMessageConverter(key);
if (messageConverter == null) {
LOG.warn("custom converter for {}[OF:{}] not found",
notification.getExperimenterDataOfChoice().getImplementedInterface(),
- deviceState.getVersion());
+ deviceInfo.getVersion());
return;
}
// build notification
try {
messageOfChoice = messageConverter.convert(vendorData, MessagePath.MESSAGE_NOTIFICATION);
final ExperimenterMessageFromDevBuilder experimenterMessageFromDevBld = new ExperimenterMessageFromDevBuilder()
- .setNode(new NodeRef(deviceState.getNodeInstanceIdentifier()))
+ .setNode(new NodeRef(deviceInfo.getNodeInstanceIdentifier()))
.setExperimenterMessageOfChoice(messageOfChoice);
// publish
notificationPublishService.offerNotification(experimenterMessageFromDevBld.build());
public synchronized void close() {
LOG.debug("closing deviceContext: {}, nodeId:{}",
getPrimaryConnectionContext().getConnectionAdapter().getRemoteAddress(),
- getDeviceState().getNodeId());
+ getDeviceInfo().getNodeId());
// NOOP
throw new UnsupportedOperationException("Autocloseble.close will be removed soon");
}
return new MultiMsgCollectorImpl(this, requestContext);
}
- @Override
- public NodeConnectorRef lookupNodeConnectorRef(final Long portNumber) {
- return nodeConnectorCache.get(portNumber);
- }
-
- @Override
- public void storeNodeConnectorRef(@Nonnull final Long portNumber, @Nonnull final NodeConnectorRef nodeConnectorRef) {
- nodeConnectorCache.put(
- Preconditions.checkNotNull(portNumber),
- Preconditions.checkNotNull(nodeConnectorRef));
- }
-
@Override
public void updatePacketInRateLimit(final long upperBound) {
packetInLimiter.changeWaterMarks((int) (LOW_WATERMARK_FACTOR * upperBound), (int) (HIGH_WATERMARK_FACTOR * upperBound));
return itemLifeCycleSourceRegistry;
}
- @Override
- public void setRpcContext(final RpcContext rpcContext) {
- this.rpcContext = rpcContext;
- }
-
- @Override
- public RpcContext getRpcContext() {
- return rpcContext;
- }
-
@Override
public void setExtensionConverterProvider(final ExtensionConverterProvider extensionConverterProvider) {
this.extensionConverterProvider = extensionConverterProvider;
return extensionConverterProvider;
}
- @Override
- public void setStatisticsContext(final StatisticsContext statisticsContext) {
- this.statisticsContext = statisticsContext;
- }
-
- @Override
- public StatisticsContext getStatisticsContext() {
- return statisticsContext;
- }
-
@Override
public synchronized void shutdownConnection() {
- LOG.debug("Shutdown method for node {}", nodeId);
- deviceState.setValid(false);
+ LOG.debug("Shutdown method for node {}", deviceInfo.getNodeId());
if (DEVICE_CONTEXT_STATE.TERMINATION.equals(deviceCtxState)) {
- LOG.debug("DeviceCtx for Node {} is in termination process.", nodeId);
+ LOG.debug("DeviceCtx for Node {} is in termination process.", deviceInfo.getNodeId());
return;
}
deviceCtxState = DEVICE_CONTEXT_STATE.TERMINATION;
if (ConnectionContext.CONNECTION_STATE.RIP.equals(getPrimaryConnectionContext().getConnectionState())) {
- LOG.debug("ConnectionCtx for Node {} is in RIP state.", deviceState.getNodeId());
+ LOG.debug("ConnectionCtx for Node {} is in RIP state.", deviceInfo.getNodeId());
return;
}
/* Terminate Auxiliary Connection */
deviceMeterRegistry.close();
}
- @Override
- public void setStatisticsRpcEnabled(boolean isStatisticsRpcEnabled) {
- this.isStatisticsRpcEnabled = isStatisticsRpcEnabled;
- }
-
@Override
public DEVICE_CONTEXT_STATE getDeviceContextState() {
return deviceCtxState;
@Override
public ListenableFuture<Void> shuttingDownDataStoreTransactions() {
- deviceState.setValid(false);
return transactionChainManager.shuttingDown();
}
+
+ @VisibleForTesting
+ TransactionChainManager getTransactionChainManager() {
+ return this.transactionChainManager;
+ }
}
Code Review / openflowplugin.git / blobdiff