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.ListenableFuture;
11 import io.netty.util.HashedWheelTimer;
12 import java.util.Optional;
14 import java.util.concurrent.ConcurrentHashMap;
15 import java.util.concurrent.ConcurrentMap;
16 import java.util.concurrent.ScheduledThreadPoolExecutor;
17 import java.util.concurrent.TimeUnit;
18 import javax.annotation.Nonnull;
19 import org.opendaylight.mdsal.binding.api.DataBroker;
20 import org.opendaylight.mdsal.binding.api.NotificationPublishService;
21 import org.opendaylight.mdsal.binding.api.WriteTransaction;
22 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
23 import org.opendaylight.openflowjava.protocol.api.connection.OutboundQueueHandlerRegistration;
24 import org.opendaylight.openflowplugin.api.openflow.OFPContext;
25 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
26 import org.opendaylight.openflowplugin.api.openflow.connection.OutboundQueueProvider;
27 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
28 import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo;
29 import org.opendaylight.openflowplugin.api.openflow.device.DeviceManager;
30 import org.opendaylight.openflowplugin.api.openflow.device.TranslatorLibrary;
31 import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChainHolder;
32 import org.opendaylight.openflowplugin.api.openflow.statistics.ofpspecific.MessageSpy;
33 import org.opendaylight.openflowplugin.extension.api.ExtensionConverterProviderKeeper;
34 import org.opendaylight.openflowplugin.extension.api.core.extension.ExtensionConverterProvider;
35 import org.opendaylight.openflowplugin.impl.connection.OutboundQueueProviderImpl;
36 import org.opendaylight.openflowplugin.impl.device.initialization.DeviceInitializerProvider;
37 import org.opendaylight.openflowplugin.impl.device.listener.OpenflowProtocolListenerFullImpl;
38 import org.opendaylight.openflowplugin.impl.util.DeviceInitializationUtil;
39 import org.opendaylight.openflowplugin.openflow.md.core.sal.convertor.ConvertorExecutor;
40 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
41 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
42 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRemovedBuilder;
43 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeUpdatedBuilder;
44 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
45 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
46 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflow.provider.config.rev160510.OpenflowProviderConfig;
47 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
48 import org.slf4j.Logger;
49 import org.slf4j.LoggerFactory;
51 public class DeviceManagerImpl implements DeviceManager, ExtensionConverterProviderKeeper {
53 private static final Logger LOG = LoggerFactory.getLogger(DeviceManagerImpl.class);
54 private static final int SPY_RATE = 10;
56 private final OpenflowProviderConfig config;
57 private final DataBroker dataBroker;
58 private final DeviceInitializerProvider deviceInitializerProvider;
59 private final ConvertorExecutor convertorExecutor;
60 private final ConcurrentMap<DeviceInfo, DeviceContext> deviceContexts = new ConcurrentHashMap<>();
61 private final Set<KeyedInstanceIdentifier<Node, NodeKey>> notificationCreateNodeSend =
62 ConcurrentHashMap.newKeySet();
63 private final NotificationPublishService notificationPublishService;
64 private final MessageSpy messageSpy;
65 private final HashedWheelTimer hashedWheelTimer;
66 private final Object updatePacketInRateLimitersLock = new Object();
67 private TranslatorLibrary translatorLibrary;
68 private ExtensionConverterProvider extensionConverterProvider;
69 private ScheduledThreadPoolExecutor spyPool;
70 private ContextChainHolder contextChainHolder;
72 public DeviceManagerImpl(@Nonnull final OpenflowProviderConfig config,
73 @Nonnull final DataBroker dataBroker,
74 @Nonnull final MessageSpy messageSpy,
75 @Nonnull final NotificationPublishService notificationPublishService,
76 @Nonnull final HashedWheelTimer hashedWheelTimer,
77 @Nonnull final ConvertorExecutor convertorExecutor,
78 @Nonnull final DeviceInitializerProvider deviceInitializerProvider) {
80 this.dataBroker = dataBroker;
81 this.deviceInitializerProvider = deviceInitializerProvider;
82 this.convertorExecutor = convertorExecutor;
83 this.hashedWheelTimer = hashedWheelTimer;
84 this.spyPool = new ScheduledThreadPoolExecutor(1);
85 this.notificationPublishService = notificationPublishService;
86 this.messageSpy = messageSpy;
87 DeviceInitializationUtil.makeEmptyNodes(dataBroker);
91 public TranslatorLibrary oook() {
92 return translatorLibrary;
96 public void setTranslatorLibrary(final TranslatorLibrary translatorLibrary) {
97 this.translatorLibrary = translatorLibrary;
101 public void close() {
102 deviceContexts.values().forEach(OFPContext::close);
103 deviceContexts.clear();
104 Optional.ofNullable(spyPool).ifPresent(ScheduledThreadPoolExecutor::shutdownNow);
110 public void initialize() {
111 spyPool.scheduleAtFixedRate(messageSpy, SPY_RATE, SPY_RATE, TimeUnit.SECONDS);
115 public void setExtensionConverterProvider(final ExtensionConverterProvider extensionConverterProvider) {
116 this.extensionConverterProvider = extensionConverterProvider;
120 public ExtensionConverterProvider getExtensionConverterProvider() {
121 return extensionConverterProvider;
125 public ListenableFuture<?> removeDeviceFromOperationalDS(@Nonnull final KeyedInstanceIdentifier<Node, NodeKey> ii) {
127 final WriteTransaction delWtx = dataBroker.newWriteOnlyTransaction();
128 delWtx.delete(LogicalDatastoreType.OPERATIONAL, ii);
129 return delWtx.commit();
134 public DeviceContext createContext(@Nonnull final ConnectionContext connectionContext) {
136 LOG.info("ConnectionEvent: Device connected to controller, Device:{}, NodeId:{}",
137 connectionContext.getConnectionAdapter().getRemoteAddress(),
138 connectionContext.getDeviceInfo().getNodeId());
140 connectionContext.getConnectionAdapter().setPacketInFiltering(true);
142 final OutboundQueueProvider outboundQueueProvider
143 = new OutboundQueueProviderImpl(connectionContext.getDeviceInfo().getVersion());
145 connectionContext.setOutboundQueueProvider(outboundQueueProvider);
146 final OutboundQueueHandlerRegistration<OutboundQueueProvider> outboundQueueHandlerRegistration =
147 connectionContext.getConnectionAdapter().registerOutboundQueueHandler(
148 outboundQueueProvider,
149 config.getBarrierCountLimit().getValue(),
150 TimeUnit.MILLISECONDS.toNanos(config.getBarrierIntervalTimeoutLimit().getValue()));
151 connectionContext.setOutboundQueueHandleRegistration(outboundQueueHandlerRegistration);
154 final DeviceContext deviceContext = new DeviceContextImpl(
160 config.isSkipTableFeatures(),
162 config.isUseSingleLayerSerialization(),
163 deviceInitializerProvider,
164 config.isEnableFlowRemovedNotification(),
165 config.isSwitchFeaturesMandatory(),
168 ((ExtensionConverterProviderKeeper) deviceContext).setExtensionConverterProvider(extensionConverterProvider);
169 deviceContext.setNotificationPublishService(notificationPublishService);
171 deviceContexts.put(connectionContext.getDeviceInfo(), deviceContext);
172 updatePacketInRateLimiters();
174 final OpenflowProtocolListenerFullImpl messageListener = new OpenflowProtocolListenerFullImpl(
175 connectionContext.getConnectionAdapter(), deviceContext);
177 connectionContext.getConnectionAdapter().setMessageListener(messageListener);
178 connectionContext.getConnectionAdapter().setAlienMessageListener(messageListener);
180 return deviceContext;
183 private void updatePacketInRateLimiters() {
184 synchronized (updatePacketInRateLimitersLock) {
185 final int deviceContextsSize = deviceContexts.size();
186 if (deviceContextsSize > 0) {
187 long freshNotificationLimit = config.getGlobalNotificationQuota() / deviceContextsSize;
188 if (freshNotificationLimit < 100) {
189 freshNotificationLimit = 100;
191 if (LOG.isDebugEnabled()) {
192 LOG.debug("fresh notification limit = {}", freshNotificationLimit);
194 for (final DeviceContext deviceContext : deviceContexts.values()) {
195 deviceContext.updatePacketInRateLimit(freshNotificationLimit);
202 public void onDeviceRemoved(final DeviceInfo deviceInfo) {
203 deviceContexts.remove(deviceInfo);
204 if (LOG.isDebugEnabled()) {
205 LOG.debug("Device context removed for node {}", deviceInfo);
208 this.updatePacketInRateLimiters();
212 public void sendNodeRemovedNotification(@Nonnull final KeyedInstanceIdentifier<Node, NodeKey> instanceIdentifier) {
213 if (notificationCreateNodeSend.remove(instanceIdentifier)) {
214 NodeRemovedBuilder builder = new NodeRemovedBuilder();
215 builder.setNodeRef(new NodeRef(instanceIdentifier));
216 LOG.info("Publishing node removed notification for {}", instanceIdentifier.firstKeyOf(Node.class).getId());
217 notificationPublishService.offerNotification(builder.build());
222 public void setContextChainHolder(@Nonnull ContextChainHolder contextChainHolder) {
223 this.contextChainHolder = contextChainHolder;
227 public void sendNodeAddedNotification(@Nonnull final KeyedInstanceIdentifier<Node, NodeKey> instanceIdentifier) {
228 if (!notificationCreateNodeSend.contains(instanceIdentifier)) {
229 notificationCreateNodeSend.add(instanceIdentifier);
230 final NodeId id = instanceIdentifier.firstKeyOf(Node.class).getId();
231 NodeUpdatedBuilder builder = new NodeUpdatedBuilder();
233 builder.setNodeRef(new NodeRef(instanceIdentifier));
234 LOG.info("Publishing node added notification for {}", id);
235 notificationPublishService.offerNotification(builder.build());