2 * Copyright (c) 2015 Cisco Systems, Inc. and others. All rights reserved.
4 * This program and the accompanying materials are made available under the
5 * terms of the Eclipse Public License v1.0 which accompanies this distribution,
6 * and is available at http://www.eclipse.org/legal/epl-v10.html
8 package org.opendaylight.openflowplugin.impl.device;
10 import com.google.common.annotations.VisibleForTesting;
11 import com.google.common.base.Preconditions;
12 import com.google.common.base.Verify;
13 import com.google.common.collect.Iterators;
14 import com.google.common.util.concurrent.FutureCallback;
15 import com.google.common.util.concurrent.Futures;
16 import com.google.common.util.concurrent.ListenableFuture;
17 import io.netty.util.TimerTask;
18 import java.util.Collections;
19 import java.util.Iterator;
20 import java.util.concurrent.ConcurrentHashMap;
21 import java.util.concurrent.ConcurrentMap;
22 import java.util.concurrent.ExecutionException;
23 import java.util.concurrent.ScheduledThreadPoolExecutor;
24 import java.util.concurrent.TimeUnit;
25 import javax.annotation.CheckForNull;
26 import javax.annotation.Nonnull;
27 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
28 import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService;
29 import org.opendaylight.controller.md.sal.binding.api.WriteTransaction;
30 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
31 import org.opendaylight.openflowjava.protocol.api.connection.ConnectionAdapter;
32 import org.opendaylight.openflowjava.protocol.api.connection.OutboundQueueHandlerRegistration;
33 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
34 import org.opendaylight.openflowplugin.api.openflow.connection.OutboundQueueProvider;
35 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
36 import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo;
37 import org.opendaylight.openflowplugin.api.openflow.device.DeviceManager;
38 import org.opendaylight.openflowplugin.api.openflow.device.DeviceState;
39 import org.opendaylight.openflowplugin.api.openflow.device.TranslatorLibrary;
40 import org.opendaylight.openflowplugin.api.openflow.device.handlers.DeviceInitializationPhaseHandler;
41 import org.opendaylight.openflowplugin.api.openflow.device.handlers.DeviceTerminationPhaseHandler;
42 import org.opendaylight.openflowplugin.api.openflow.lifecycle.LifecycleConductor;
43 import org.opendaylight.openflowplugin.extension.api.ExtensionConverterProviderKeeper;
44 import org.opendaylight.openflowplugin.extension.api.core.extension.ExtensionConverterProvider;
45 import org.opendaylight.openflowplugin.impl.connection.OutboundQueueProviderImpl;
46 import org.opendaylight.openflowplugin.impl.device.listener.OpenflowProtocolListenerFullImpl;
47 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.Nodes;
48 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodesBuilder;
49 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
50 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
51 import org.slf4j.Logger;
52 import org.slf4j.LoggerFactory;
57 public class DeviceManagerImpl implements DeviceManager, ExtensionConverterProviderKeeper {
59 private static final Logger LOG = LoggerFactory.getLogger(DeviceManagerImpl.class);
61 private final long globalNotificationQuota;
62 private final boolean switchFeaturesMandatory;
64 private final int spyRate = 10;
66 private final DataBroker dataBroker;
67 private TranslatorLibrary translatorLibrary;
68 private DeviceInitializationPhaseHandler deviceInitPhaseHandler;
69 private DeviceTerminationPhaseHandler deviceTerminPhaseHandler;
70 private NotificationPublishService notificationPublishService;
72 private final ConcurrentMap<DeviceInfo, DeviceContext> deviceContexts = new ConcurrentHashMap<>();
74 private final long barrierIntervalNanos;
75 private final int barrierCountLimit;
76 private ExtensionConverterProvider extensionConverterProvider;
77 private ScheduledThreadPoolExecutor spyPool;
79 private final LifecycleConductor conductor;
80 private boolean isStatisticsRpcEnabled;
82 public DeviceManagerImpl(@Nonnull final DataBroker dataBroker,
83 final long globalNotificationQuota, final boolean switchFeaturesMandatory,
84 final long barrierInterval, final int barrierCountLimit,
85 final LifecycleConductor lifecycleConductor) {
86 this.switchFeaturesMandatory = switchFeaturesMandatory;
87 this.globalNotificationQuota = globalNotificationQuota;
88 this.dataBroker = Preconditions.checkNotNull(dataBroker);
89 /* merge empty nodes to oper DS to predict any problems with missing parent for Node */
90 final WriteTransaction tx = dataBroker.newWriteOnlyTransaction();
92 final NodesBuilder nodesBuilder = new NodesBuilder();
93 nodesBuilder.setNode(Collections.<Node>emptyList());
94 tx.merge(LogicalDatastoreType.OPERATIONAL, InstanceIdentifier.create(Nodes.class), nodesBuilder.build());
97 } catch (ExecutionException | InterruptedException e) {
98 LOG.error("Creation of node failed.", e);
99 throw new IllegalStateException(e);
102 this.barrierIntervalNanos = TimeUnit.MILLISECONDS.toNanos(barrierInterval);
103 this.barrierCountLimit = barrierCountLimit;
105 this.conductor = lifecycleConductor;
106 spyPool = new ScheduledThreadPoolExecutor(1);
111 public void setDeviceInitializationPhaseHandler(final DeviceInitializationPhaseHandler handler) {
112 this.deviceInitPhaseHandler = handler;
116 public void onDeviceContextLevelUp(@CheckForNull DeviceInfo deviceInfo) throws Exception {
117 // final phase - we have to add new Device to MD-SAL DataStore
118 LOG.debug("Final phase of DeviceContextLevelUp for Node: {} ", deviceInfo.getNodeId());
119 DeviceContext deviceContext = Preconditions.checkNotNull(deviceContexts.get(deviceInfo));
120 ((DeviceContextImpl) deviceContext).initialSubmitTransaction();
121 deviceContext.onPublished();
125 public boolean deviceConnected(@CheckForNull final ConnectionContext connectionContext) throws Exception {
126 Preconditions.checkArgument(connectionContext != null);
128 DeviceInfo deviceInfo = connectionContext.getDeviceInfo();
130 * This part prevent destroy another device context. Throwing here an exception result to propagate close connection
131 * in {@link org.opendaylight.openflowplugin.impl.connection.org.opendaylight.openflowplugin.impl.connection.HandshakeContextImpl}
132 * If context already exist we are in state closing process (connection flapping) and we should not propagate connection close
134 if (deviceContexts.containsKey(deviceInfo)) {
135 LOG.warn("Rejecting connection from node which is already connected and there exist deviceContext for it: {}", connectionContext.getNodeId());
139 LOG.info("ConnectionEvent: Device connected to controller, Device:{}, NodeId:{}",
140 connectionContext.getConnectionAdapter().getRemoteAddress(), deviceInfo.getNodeId());
142 // Add Disconnect handler
143 connectionContext.setDeviceDisconnectedHandler(DeviceManagerImpl.this);
144 // Cache this for clarity
145 final ConnectionAdapter connectionAdapter = connectionContext.getConnectionAdapter();
147 //FIXME: as soon as auxiliary connection are fully supported then this is needed only before device context published
148 connectionAdapter.setPacketInFiltering(true);
150 final Short version = connectionContext.getFeatures().getVersion();
151 final OutboundQueueProvider outboundQueueProvider = new OutboundQueueProviderImpl(version);
153 connectionContext.setOutboundQueueProvider(outboundQueueProvider);
154 final OutboundQueueHandlerRegistration<OutboundQueueProvider> outboundQueueHandlerRegistration =
155 connectionAdapter.registerOutboundQueueHandler(outboundQueueProvider, barrierCountLimit, barrierIntervalNanos);
156 connectionContext.setOutboundQueueHandleRegistration(outboundQueueHandlerRegistration);
158 final DeviceState deviceState = new DeviceStateImpl();
159 final DeviceContext deviceContext = new DeviceContextImpl(connectionContext,
163 outboundQueueProvider,
165 switchFeaturesMandatory);
167 Verify.verify(deviceContexts.putIfAbsent(deviceInfo, deviceContext) == null, "DeviceCtx still not closed.");
169 ((ExtensionConverterProviderKeeper) deviceContext).setExtensionConverterProvider(extensionConverterProvider);
170 deviceContext.setStatisticsRpcEnabled(isStatisticsRpcEnabled);
171 deviceContext.setNotificationPublishService(notificationPublishService);
173 updatePacketInRateLimiters();
175 final OpenflowProtocolListenerFullImpl messageListener = new OpenflowProtocolListenerFullImpl(
176 connectionAdapter, deviceContext);
177 connectionAdapter.setMessageListener(messageListener);
178 deviceState.setValid(true);
180 deviceInitPhaseHandler.onDeviceContextLevelUp(connectionContext.getDeviceInfo());
185 private void updatePacketInRateLimiters() {
186 synchronized (deviceContexts) {
187 final int deviceContextsSize = deviceContexts.size();
188 if (deviceContextsSize > 0) {
189 long freshNotificationLimit = globalNotificationQuota / deviceContextsSize;
190 if (freshNotificationLimit < 100) {
191 freshNotificationLimit = 100;
193 LOG.debug("fresh notification limit = {}", freshNotificationLimit);
194 for (final DeviceContext deviceContext : deviceContexts.values()) {
195 deviceContext.updatePacketInRateLimit(freshNotificationLimit);
202 public TranslatorLibrary oook() {
203 return translatorLibrary;
207 public void setTranslatorLibrary(final TranslatorLibrary translatorLibrary) {
208 this.translatorLibrary = translatorLibrary;
212 public void setNotificationPublishService(final NotificationPublishService notificationService) {
213 notificationPublishService = notificationService;
217 public void close() {
218 for (final Iterator<DeviceContext> iterator = Iterators.consumingIterator(deviceContexts.values().iterator());
219 iterator.hasNext();) {
220 final DeviceContext deviceCtx = iterator.next();
221 deviceCtx.shutdownConnection();
222 deviceCtx.shuttingDownDataStoreTransactions();
225 if (spyPool != null) {
226 spyPool.shutdownNow();
232 public void onDeviceContextLevelDown(final DeviceInfo deviceInfo) {
233 LOG.debug("onDeviceContextClosed for Node {}", deviceInfo.getNodeId());
234 deviceContexts.remove(deviceInfo);
235 updatePacketInRateLimiters();
239 public void initialize() {
240 spyPool.scheduleAtFixedRate(conductor.getMessageIntelligenceAgency(), spyRate, spyRate, TimeUnit.SECONDS);
244 public DeviceContext getDeviceContextFromNodeId(DeviceInfo deviceInfo) {
245 return deviceContexts.get(deviceInfo);
249 public void setStatisticsRpcEnabled(boolean isStatisticsRpcEnabled) {
250 this.isStatisticsRpcEnabled = isStatisticsRpcEnabled;
254 public void setExtensionConverterProvider(final ExtensionConverterProvider extensionConverterProvider) {
255 this.extensionConverterProvider = extensionConverterProvider;
259 public ExtensionConverterProvider getExtensionConverterProvider() {
260 return extensionConverterProvider;
264 public void setDeviceTerminationPhaseHandler(final DeviceTerminationPhaseHandler handler) {
265 this.deviceTerminPhaseHandler = handler;
269 public void onDeviceDisconnected(final ConnectionContext connectionContext) {
270 LOG.trace("onDeviceDisconnected method call for Node: {}", connectionContext.getNodeId());
271 final DeviceInfo deviceInfo = connectionContext.getDeviceInfo();
272 final DeviceContext deviceCtx = this.deviceContexts.get(deviceInfo.getNodeId());
274 if (null == deviceCtx) {
275 LOG.info("DeviceContext for Node {} was not found. Connection is terminated without OFP context suite.", deviceInfo.getNodeId());
279 if (!connectionContext.equals(deviceCtx.getPrimaryConnectionContext())) {
280 /* Connection is not PrimaryConnection so try to remove from Auxiliary Connections */
281 deviceCtx.removeAuxiliaryConnectionContext(connectionContext);
283 /* Device is disconnected and so we need to close TxManager */
284 final ListenableFuture<Void> future = deviceCtx.shuttingDownDataStoreTransactions();
285 Futures.addCallback(future, new FutureCallback<Void>() {
288 public void onSuccess(final Void result) {
289 LOG.debug("TxChainManager for device {} is closed successful.", deviceInfo.getNodeId());
290 deviceTerminPhaseHandler.onDeviceContextLevelDown(deviceInfo);
294 public void onFailure(final Throwable t) {
295 LOG.warn("TxChainManager for device {} failed by closing.", deviceInfo.getNodeId(), t);
296 deviceTerminPhaseHandler.onDeviceContextLevelDown(deviceInfo);
299 /* Add timer for Close TxManager because it could fain ind cluster without notification */
300 final TimerTask timerTask = timeout -> {
301 if (!future.isDone()) {
302 LOG.info("Shutting down TxChain for node {} not completed during 10 sec. Continue anyway.", deviceInfo.getNodeId());
303 future.cancel(false);
306 conductor.newTimeout(timerTask, 10, TimeUnit.SECONDS);
311 void addDeviceContextToMap(final DeviceInfo deviceInfo, final DeviceContext deviceContext){
312 deviceContexts.put(deviceInfo, deviceContext);