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.util.concurrent.CheckedFuture;
11 import com.google.common.util.concurrent.FutureCallback;
12 import com.google.common.util.concurrent.Futures;
13 import io.netty.util.HashedWheelTimer;
14 import io.netty.util.internal.ConcurrentSet;
15 import java.util.Optional;
17 import java.util.concurrent.ConcurrentHashMap;
18 import java.util.concurrent.ConcurrentMap;
19 import java.util.concurrent.ScheduledThreadPoolExecutor;
20 import java.util.concurrent.TimeUnit;
21 import javax.annotation.Nonnull;
22 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
23 import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService;
24 import org.opendaylight.controller.md.sal.binding.api.WriteTransaction;
25 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
26 import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
27 import org.opendaylight.openflowjava.protocol.api.connection.OutboundQueueHandlerRegistration;
28 import org.opendaylight.openflowplugin.api.openflow.OFPContext;
29 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
30 import org.opendaylight.openflowplugin.api.openflow.connection.OutboundQueueProvider;
31 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
32 import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo;
33 import org.opendaylight.openflowplugin.api.openflow.device.DeviceManager;
34 import org.opendaylight.openflowplugin.api.openflow.device.TranslatorLibrary;
35 import org.opendaylight.openflowplugin.api.openflow.statistics.ofpspecific.MessageSpy;
36 import org.opendaylight.openflowplugin.extension.api.ExtensionConverterProviderKeeper;
37 import org.opendaylight.openflowplugin.extension.api.core.extension.ExtensionConverterProvider;
38 import org.opendaylight.openflowplugin.impl.connection.OutboundQueueProviderImpl;
39 import org.opendaylight.openflowplugin.impl.device.initialization.DeviceInitializerProvider;
40 import org.opendaylight.openflowplugin.impl.device.listener.OpenflowProtocolListenerFullImpl;
41 import org.opendaylight.openflowplugin.impl.services.sal.SalRoleServiceImpl;
42 import org.opendaylight.openflowplugin.impl.util.DeviceInitializationUtil;
43 import org.opendaylight.openflowplugin.openflow.md.core.sal.convertor.ConvertorExecutor;
44 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
45 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
46 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRemovedBuilder;
47 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeUpdatedBuilder;
48 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
49 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
50 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflow.provider.config.rev160510.OpenflowProviderConfig;
51 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
52 import org.slf4j.Logger;
53 import org.slf4j.LoggerFactory;
58 public class DeviceManagerImpl implements DeviceManager, ExtensionConverterProviderKeeper {
60 private static final Logger LOG = LoggerFactory.getLogger(DeviceManagerImpl.class);
61 private static final int SPY_RATE = 10;
63 private final OpenflowProviderConfig config;
64 private final DataBroker dataBroker;
65 private final DeviceInitializerProvider deviceInitializerProvider;
66 private final ConvertorExecutor convertorExecutor;
67 private final ConcurrentMap<DeviceInfo, DeviceContext> deviceContexts = new ConcurrentHashMap<>();
68 private final Set<KeyedInstanceIdentifier<Node, NodeKey>> notificationCreateNodeSend = new ConcurrentSet<>();
69 private final NotificationPublishService notificationPublishService;
70 private final MessageSpy messageSpy;
71 private final HashedWheelTimer hashedWheelTimer;
72 private TranslatorLibrary translatorLibrary;
73 private ExtensionConverterProvider extensionConverterProvider;
74 private ScheduledThreadPoolExecutor spyPool;
76 public DeviceManagerImpl(@Nonnull final OpenflowProviderConfig config,
77 @Nonnull final DataBroker dataBroker,
78 @Nonnull final MessageSpy messageSpy,
79 @Nonnull final NotificationPublishService notificationPublishService,
80 @Nonnull final HashedWheelTimer hashedWheelTimer,
81 @Nonnull final ConvertorExecutor convertorExecutor,
82 @Nonnull final DeviceInitializerProvider deviceInitializerProvider) {
84 this.dataBroker = dataBroker;
85 this.deviceInitializerProvider = deviceInitializerProvider;
86 this.convertorExecutor = convertorExecutor;
87 this.hashedWheelTimer = hashedWheelTimer;
88 this.spyPool = new ScheduledThreadPoolExecutor(1);
89 this.notificationPublishService = notificationPublishService;
90 this.messageSpy = messageSpy;
91 DeviceInitializationUtil.makeEmptyNodes(dataBroker);
95 public TranslatorLibrary oook() {
96 return translatorLibrary;
100 public void setTranslatorLibrary(final TranslatorLibrary translatorLibrary) {
101 this.translatorLibrary = translatorLibrary;
105 public void close() {
106 deviceContexts.values().forEach(OFPContext::close);
107 deviceContexts.clear();
108 Optional.ofNullable(spyPool).ifPresent(ScheduledThreadPoolExecutor::shutdownNow);
114 public void initialize() {
115 spyPool.scheduleAtFixedRate(messageSpy, SPY_RATE, SPY_RATE, TimeUnit.SECONDS);
119 public void setExtensionConverterProvider(final ExtensionConverterProvider extensionConverterProvider) {
120 this.extensionConverterProvider = extensionConverterProvider;
124 public ExtensionConverterProvider getExtensionConverterProvider() {
125 return extensionConverterProvider;
129 public CheckedFuture<Void, TransactionCommitFailedException> removeDeviceFromOperationalDS(final KeyedInstanceIdentifier<Node, NodeKey> ii) {
130 final WriteTransaction delWtx = dataBroker.newWriteOnlyTransaction();
131 delWtx.delete(LogicalDatastoreType.OPERATIONAL, ii);
132 final CheckedFuture<Void, TransactionCommitFailedException> delFuture = delWtx.submit();
134 Futures.addCallback(delFuture, new FutureCallback<Void>() {
136 public void onSuccess(final Void result) {
137 if (LOG.isDebugEnabled()) {
138 LOG.debug("Delete Node {} was successful", ii);
143 public void onFailure(@Nonnull final Throwable t) {
144 LOG.warn("Delete node {} failed with exception {}", ii, t);
151 public DeviceContext createContext(@Nonnull final ConnectionContext connectionContext) {
153 LOG.info("ConnectionEvent: Device connected to controller, Device:{}, NodeId:{}",
154 connectionContext.getConnectionAdapter().getRemoteAddress(),
155 connectionContext.getDeviceInfo().getNodeId());
157 connectionContext.getConnectionAdapter().setPacketInFiltering(true);
159 final OutboundQueueProvider outboundQueueProvider
160 = new OutboundQueueProviderImpl(connectionContext.getDeviceInfo().getVersion());
162 connectionContext.setOutboundQueueProvider(outboundQueueProvider);
163 final OutboundQueueHandlerRegistration<OutboundQueueProvider> outboundQueueHandlerRegistration =
164 connectionContext.getConnectionAdapter().registerOutboundQueueHandler(
165 outboundQueueProvider,
166 config.getBarrierCountLimit().getValue(),
167 TimeUnit.MILLISECONDS.toNanos(config.getBarrierIntervalTimeoutLimit().getValue()));
168 connectionContext.setOutboundQueueHandleRegistration(outboundQueueHandlerRegistration);
171 final DeviceContext deviceContext = new DeviceContextImpl(
177 config.isSkipTableFeatures(),
179 config.isUseSingleLayerSerialization(),
180 deviceInitializerProvider,
181 config.isEnableFlowRemovedNotification(),
182 config.isSwitchFeaturesMandatory());
184 deviceContext.setSalRoleService(new SalRoleServiceImpl(deviceContext, deviceContext));
185 ((ExtensionConverterProviderKeeper) deviceContext).setExtensionConverterProvider(extensionConverterProvider);
186 deviceContext.setNotificationPublishService(notificationPublishService);
188 deviceContexts.put(connectionContext.getDeviceInfo(), deviceContext);
189 updatePacketInRateLimiters();
191 final OpenflowProtocolListenerFullImpl messageListener = new OpenflowProtocolListenerFullImpl(
192 connectionContext.getConnectionAdapter(), deviceContext);
194 connectionContext.getConnectionAdapter().setMessageListener(messageListener);
195 connectionContext.getConnectionAdapter().setAlienMessageListener(messageListener);
197 return deviceContext;
200 private void updatePacketInRateLimiters() {
201 synchronized (deviceContexts) {
202 final int deviceContextsSize = deviceContexts.size();
203 if (deviceContextsSize > 0) {
204 long freshNotificationLimit = config.getGlobalNotificationQuota() / deviceContextsSize;
205 if (freshNotificationLimit < 100) {
206 freshNotificationLimit = 100;
208 if (LOG.isDebugEnabled()) {
209 LOG.debug("fresh notification limit = {}", freshNotificationLimit);
211 for (final DeviceContext deviceContext : deviceContexts.values()) {
212 deviceContext.updatePacketInRateLimit(freshNotificationLimit);
219 public void onDeviceRemoved(final DeviceInfo deviceInfo) {
220 deviceContexts.remove(deviceInfo);
221 if (LOG.isDebugEnabled()) {
222 LOG.debug("Device context removed for node {}", deviceInfo.getLOGValue());
224 if (deviceContexts.size() > 0) {
225 this.updatePacketInRateLimiters();
230 public void sendNodeRemovedNotification(@Nonnull final KeyedInstanceIdentifier<Node, NodeKey> instanceIdentifier) {
231 if (notificationCreateNodeSend.remove(instanceIdentifier)) {
232 NodeRemovedBuilder builder = new NodeRemovedBuilder();
233 builder.setNodeRef(new NodeRef(instanceIdentifier));
234 LOG.info("Publishing node removed notification for {}", instanceIdentifier.firstKeyOf(Node.class).getId());
235 notificationPublishService.offerNotification(builder.build());
240 public void sendNodeAddedNotification(@Nonnull final KeyedInstanceIdentifier<Node, NodeKey> instanceIdentifier) {
241 if (!notificationCreateNodeSend.contains(instanceIdentifier)) {
242 notificationCreateNodeSend.add(instanceIdentifier);
243 final NodeId id = instanceIdentifier.firstKeyOf(Node.class).getId();
244 NodeUpdatedBuilder builder = new NodeUpdatedBuilder();
246 builder.setNodeRef(new NodeRef(instanceIdentifier));
247 LOG.info("Publishing node added notification for {}", id);
248 notificationPublishService.offerNotification(builder.build());