2 * Copyright (c) 2015, 2017 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.collect.Iterators;
11 import com.google.common.util.concurrent.CheckedFuture;
12 import com.google.common.util.concurrent.FutureCallback;
13 import com.google.common.util.concurrent.Futures;
14 import io.netty.util.HashedWheelTimer;
15 import io.netty.util.internal.ConcurrentSet;
16 import java.util.Collections;
17 import java.util.Iterator;
18 import java.util.Optional;
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.Nonnull;
26 import javax.annotation.Nullable;
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.controller.md.sal.common.api.data.TransactionCommitFailedException;
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.TranslatorLibrary;
39 import org.opendaylight.openflowplugin.api.openflow.statistics.ofpspecific.MessageSpy;
40 import org.opendaylight.openflowplugin.extension.api.ExtensionConverterProviderKeeper;
41 import org.opendaylight.openflowplugin.extension.api.core.extension.ExtensionConverterProvider;
42 import org.opendaylight.openflowplugin.impl.connection.OutboundQueueProviderImpl;
43 import org.opendaylight.openflowplugin.impl.device.initialization.DeviceInitializerProvider;
44 import org.opendaylight.openflowplugin.impl.device.listener.OpenflowProtocolListenerFullImpl;
45 import org.opendaylight.openflowplugin.impl.services.sal.SalRoleServiceImpl;
46 import org.opendaylight.openflowplugin.openflow.md.core.sal.convertor.ConvertorExecutor;
47 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
48 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRemovedBuilder;
49 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeUpdatedBuilder;
50 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.Nodes;
51 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodesBuilder;
52 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
53 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
54 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
55 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
56 import org.slf4j.Logger;
57 import org.slf4j.LoggerFactory;
62 public class DeviceManagerImpl implements DeviceManager, ExtensionConverterProviderKeeper {
64 private static final Logger LOG = LoggerFactory.getLogger(DeviceManagerImpl.class);
66 private long globalNotificationQuota;
67 private boolean switchFeaturesMandatory;
68 private boolean isFlowRemovedNotificationOn;
69 private boolean skipTableFeatures;
70 private static final int SPY_RATE = 10;
72 private final DataBroker dataBroker;
73 private final DeviceInitializerProvider deviceInitializerProvider;
74 private final ConvertorExecutor convertorExecutor;
75 private TranslatorLibrary translatorLibrary;
77 private final ConcurrentMap<DeviceInfo, DeviceContext> deviceContexts = new ConcurrentHashMap<>();
78 private final Set<DeviceInfo> notificationCreateNodeSend = new ConcurrentSet<>();
80 private long barrierIntervalNanos;
81 private int barrierCountLimit;
83 private ExtensionConverterProvider extensionConverterProvider;
84 private ScheduledThreadPoolExecutor spyPool;
85 private final NotificationPublishService notificationPublishService;
86 private final MessageSpy messageSpy;
87 private final HashedWheelTimer hashedWheelTimer;
88 private final boolean useSingleLayerSerialization;
90 public DeviceManagerImpl(@Nonnull final DataBroker dataBroker,
91 @Nonnull final MessageSpy messageSpy,
92 @Nullable final NotificationPublishService notificationPublishService,
93 @Nonnull final HashedWheelTimer hashedWheelTimer,
94 @Nonnull final ConvertorExecutor convertorExecutor,
95 @Nonnull final DeviceInitializerProvider deviceInitializerProvider,
96 final boolean useSingleLayerSerialization) {
98 this.dataBroker = dataBroker;
99 this.deviceInitializerProvider = deviceInitializerProvider;
100 this.useSingleLayerSerialization = useSingleLayerSerialization;
102 /* merge empty nodes to oper DS to predict any problems with missing parent for Node */
103 final WriteTransaction tx = dataBroker.newWriteOnlyTransaction();
104 final NodesBuilder nodesBuilder = new NodesBuilder();
105 nodesBuilder.setNode(Collections.<Node>emptyList());
106 tx.merge(LogicalDatastoreType.OPERATIONAL, InstanceIdentifier.create(Nodes.class), nodesBuilder.build());
109 } catch (ExecutionException | InterruptedException e) {
110 LOG.error("Creation of node failed.", e);
111 throw new IllegalStateException(e);
114 this.convertorExecutor = convertorExecutor;
115 this.hashedWheelTimer = hashedWheelTimer;
116 this.spyPool = new ScheduledThreadPoolExecutor(1);
117 this.notificationPublishService = notificationPublishService;
118 this.messageSpy = messageSpy;
122 public TranslatorLibrary oook() {
123 return translatorLibrary;
127 public void setTranslatorLibrary(final TranslatorLibrary translatorLibrary) {
128 this.translatorLibrary = translatorLibrary;
132 public void close() {
133 for (final Iterator<DeviceContext> iterator = Iterators.consumingIterator(deviceContexts.values().iterator());
134 iterator.hasNext();) {
135 final DeviceContext deviceCtx = iterator.next();
136 deviceCtx.shutdownConnection();
137 deviceCtx.shuttingDownDataStoreTransactions();
140 Optional.ofNullable(spyPool).ifPresent(ScheduledThreadPoolExecutor::shutdownNow);
146 public void initialize() {
147 spyPool.scheduleAtFixedRate(messageSpy, SPY_RATE, SPY_RATE, TimeUnit.SECONDS);
151 public void setExtensionConverterProvider(final ExtensionConverterProvider extensionConverterProvider) {
152 this.extensionConverterProvider = extensionConverterProvider;
156 public ExtensionConverterProvider getExtensionConverterProvider() {
157 return extensionConverterProvider;
161 public void setFlowRemovedNotificationOn(boolean isNotificationFlowRemovedOff) {
162 this.isFlowRemovedNotificationOn = isNotificationFlowRemovedOff;
166 public boolean isFlowRemovedNotificationOn() {
167 return this.isFlowRemovedNotificationOn;
171 public void setGlobalNotificationQuota(final long globalNotificationQuota) {
172 this.globalNotificationQuota = globalNotificationQuota;
176 public void setSwitchFeaturesMandatory(final boolean switchFeaturesMandatory) {
177 this.switchFeaturesMandatory = switchFeaturesMandatory;
181 public void setSkipTableFeatures(boolean skipTableFeaturesValue) {
182 skipTableFeatures = skipTableFeaturesValue;
186 public void setBarrierCountLimit(final int barrierCountLimit) {
187 this.barrierCountLimit = barrierCountLimit;
191 public void setBarrierInterval(final long barrierTimeoutLimit) {
192 this.barrierIntervalNanos = TimeUnit.MILLISECONDS.toNanos(barrierTimeoutLimit);
196 public CheckedFuture<Void, TransactionCommitFailedException> removeDeviceFromOperationalDS(final KeyedInstanceIdentifier<Node, NodeKey> ii) {
197 final WriteTransaction delWtx = dataBroker.newWriteOnlyTransaction();
198 delWtx.delete(LogicalDatastoreType.OPERATIONAL, ii);
199 final CheckedFuture<Void, TransactionCommitFailedException> delFuture = delWtx.submit();
201 Futures.addCallback(delFuture, new FutureCallback<Void>() {
203 public void onSuccess(final Void result) {
204 if (LOG.isDebugEnabled()) {
205 LOG.debug("Delete Node {} was successful", ii);
210 public void onFailure(@Nonnull final Throwable t) {
211 LOG.warn("Delete node {} failed with exception {}", ii, t);
219 public CheckedFuture<Void, TransactionCommitFailedException> removeDeviceFromOperationalDS(final DeviceInfo deviceInfo) {
220 return this.removeDeviceFromOperationalDS(deviceInfo.getNodeInstanceIdentifier());
223 public DeviceContext createContext(@Nonnull final ConnectionContext connectionContext) {
225 LOG.info("ConnectionEvent: Device connected to controller, Device:{}, NodeId:{}",
226 connectionContext.getConnectionAdapter().getRemoteAddress(),
227 connectionContext.getDeviceInfo().getNodeId());
229 connectionContext.getConnectionAdapter().setPacketInFiltering(true);
231 final OutboundQueueProvider outboundQueueProvider
232 = new OutboundQueueProviderImpl(connectionContext.getDeviceInfo().getVersion());
234 connectionContext.setOutboundQueueProvider(outboundQueueProvider);
235 final OutboundQueueHandlerRegistration<OutboundQueueProvider> outboundQueueHandlerRegistration =
236 connectionContext.getConnectionAdapter().registerOutboundQueueHandler(
237 outboundQueueProvider,
239 barrierIntervalNanos);
240 connectionContext.setOutboundQueueHandleRegistration(outboundQueueHandlerRegistration);
243 final DeviceContext deviceContext = new DeviceContextImpl(
252 useSingleLayerSerialization,
253 deviceInitializerProvider);
255 deviceContext.setSalRoleService(new SalRoleServiceImpl(deviceContext, deviceContext));
256 deviceContext.setSwitchFeaturesMandatory(switchFeaturesMandatory);
257 ((ExtensionConverterProviderKeeper) deviceContext).setExtensionConverterProvider(extensionConverterProvider);
258 deviceContext.setNotificationPublishService(notificationPublishService);
260 deviceContexts.put(connectionContext.getDeviceInfo(), deviceContext);
261 updatePacketInRateLimiters();
263 final OpenflowProtocolListenerFullImpl messageListener = new OpenflowProtocolListenerFullImpl(
264 connectionContext.getConnectionAdapter(), deviceContext);
266 connectionContext.getConnectionAdapter().setMessageListener(messageListener);
268 return deviceContext;
271 private void updatePacketInRateLimiters() {
272 synchronized (deviceContexts) {
273 final int deviceContextsSize = deviceContexts.size();
274 if (deviceContextsSize > 0) {
275 long freshNotificationLimit = globalNotificationQuota / deviceContextsSize;
276 if (freshNotificationLimit < 100) {
277 freshNotificationLimit = 100;
279 if (LOG.isDebugEnabled()) {
280 LOG.debug("fresh notification limit = {}", freshNotificationLimit);
282 for (final DeviceContext deviceContext : deviceContexts.values()) {
283 deviceContext.updatePacketInRateLimit(freshNotificationLimit);
290 public void sendNodeRemovedNotification(@Nonnull final DeviceInfo deviceInfo) {
291 notificationCreateNodeSend.remove(deviceInfo);
292 NodeRemovedBuilder builder = new NodeRemovedBuilder();
293 builder.setNodeRef(new NodeRef(deviceInfo.getNodeInstanceIdentifier()));
294 if (LOG.isDebugEnabled()) {
295 LOG.debug("Publishing node removed notification for {}", deviceInfo.getLOGValue());
297 notificationPublishService.offerNotification(builder.build());
302 public void onDeviceRemoved(final DeviceInfo deviceInfo) {
303 deviceContexts.remove(deviceInfo);
304 if (LOG.isDebugEnabled()) {
305 LOG.debug("Device context removed for node {}", deviceInfo.getLOGValue());
307 if (deviceContexts.size() > 0) {
308 this.updatePacketInRateLimiters();
313 public void sendNodeAddedNotification(@Nonnull final DeviceInfo deviceInfo) {
314 if (!notificationCreateNodeSend.contains(deviceInfo)) {
315 notificationCreateNodeSend.add(deviceInfo);
316 NodeUpdatedBuilder builder = new NodeUpdatedBuilder();
317 builder.setId(deviceInfo.getNodeId());
318 builder.setNodeRef(new NodeRef(deviceInfo.getNodeInstanceIdentifier()));
319 if (LOG.isDebugEnabled()) {
320 LOG.debug("Publishing node added notification for {}", deviceInfo.getLOGValue());
322 notificationPublishService.offerNotification(builder.build());