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;
10 import com.google.common.annotations.VisibleForTesting;
11 import com.google.common.util.concurrent.FutureCallback;
12 import com.google.common.util.concurrent.Futures;
13 import com.google.common.util.concurrent.ListenableFuture;
14 import com.google.common.util.concurrent.ListeningExecutorService;
15 import com.google.common.util.concurrent.MoreExecutors;
16 import com.google.common.util.concurrent.SettableFuture;
17 import io.netty.util.HashedWheelTimer;
18 import io.netty.util.Timer;
19 import java.lang.management.ManagementFactory;
20 import java.util.Collection;
21 import java.util.List;
22 import java.util.concurrent.ExecutionException;
23 import java.util.concurrent.ExecutorService;
24 import java.util.concurrent.Future;
25 import java.util.concurrent.SynchronousQueue;
26 import java.util.concurrent.TimeUnit;
27 import java.util.concurrent.TimeoutException;
28 import java.util.stream.Collectors;
29 import javax.annotation.PostConstruct;
30 import javax.annotation.PreDestroy;
31 import javax.inject.Inject;
32 import javax.inject.Singleton;
33 import javax.management.InstanceAlreadyExistsException;
34 import javax.management.InstanceNotFoundException;
35 import javax.management.MBeanRegistrationException;
36 import javax.management.MBeanServer;
37 import javax.management.MalformedObjectNameException;
38 import javax.management.NotCompliantMBeanException;
39 import javax.management.ObjectName;
40 import org.apache.aries.blueprint.annotation.service.Reference;
41 import org.apache.aries.blueprint.annotation.service.Service;
42 import org.opendaylight.infrautils.diagstatus.ServiceState;
43 import org.opendaylight.infrautils.ready.SystemReadyListener;
44 import org.opendaylight.infrautils.ready.SystemReadyMonitor;
45 import org.opendaylight.mdsal.binding.api.DataBroker;
46 import org.opendaylight.mdsal.binding.api.NotificationPublishService;
47 import org.opendaylight.mdsal.binding.api.RpcProviderService;
48 import org.opendaylight.mdsal.eos.binding.api.EntityOwnershipService;
49 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceProvider;
50 import org.opendaylight.openflowjava.protocol.api.connection.OpenflowDiagStatusProvider;
51 import org.opendaylight.openflowjava.protocol.spi.connection.SwitchConnectionProvider;
52 import org.opendaylight.openflowjava.protocol.spi.connection.SwitchConnectionProviderList;
53 import org.opendaylight.openflowplugin.api.openflow.OpenFlowPluginProvider;
54 import org.opendaylight.openflowplugin.api.openflow.configuration.ConfigurationService;
55 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionManager;
56 import org.opendaylight.openflowplugin.api.openflow.device.DeviceManager;
57 import org.opendaylight.openflowplugin.api.openflow.mastership.MastershipChangeServiceManager;
58 import org.opendaylight.openflowplugin.api.openflow.role.RoleManager;
59 import org.opendaylight.openflowplugin.api.openflow.rpc.RpcManager;
60 import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsManager;
61 import org.opendaylight.openflowplugin.api.openflow.statistics.ofpspecific.MessageIntelligenceAgency;
62 import org.opendaylight.openflowplugin.extension.api.ExtensionConverterProviderKeeper;
63 import org.opendaylight.openflowplugin.extension.api.ExtensionConverterRegistrator;
64 import org.opendaylight.openflowplugin.extension.api.OpenFlowPluginExtensionRegistratorProvider;
65 import org.opendaylight.openflowplugin.extension.api.core.extension.ExtensionConverterManager;
66 import org.opendaylight.openflowplugin.impl.configuration.OpenFlowProviderConfigImpl;
67 import org.opendaylight.openflowplugin.impl.connection.ConnectionManagerImpl;
68 import org.opendaylight.openflowplugin.impl.device.DeviceManagerImpl;
69 import org.opendaylight.openflowplugin.impl.device.initialization.DeviceInitializerProvider;
70 import org.opendaylight.openflowplugin.impl.device.initialization.DeviceInitializerProviderFactory;
71 import org.opendaylight.openflowplugin.impl.lifecycle.ContextChainHolderImpl;
72 import org.opendaylight.openflowplugin.impl.protocol.deserialization.DeserializerInjector;
73 import org.opendaylight.openflowplugin.impl.protocol.serialization.SerializerInjector;
74 import org.opendaylight.openflowplugin.impl.role.RoleManagerImpl;
75 import org.opendaylight.openflowplugin.impl.rpc.RpcManagerImpl;
76 import org.opendaylight.openflowplugin.impl.statistics.StatisticsManagerImpl;
77 import org.opendaylight.openflowplugin.impl.statistics.ofpspecific.MessageIntelligenceAgencyImpl;
78 import org.opendaylight.openflowplugin.impl.statistics.ofpspecific.MessageIntelligenceAgencyMXBean;
79 import org.opendaylight.openflowplugin.impl.util.ThreadPoolLoggingExecutor;
80 import org.opendaylight.openflowplugin.impl.util.TranslatorLibraryUtil;
81 import org.opendaylight.openflowplugin.openflow.md.core.extension.ExtensionConverterManagerImpl;
82 import org.opendaylight.openflowplugin.openflow.md.core.sal.convertor.ConvertorManager;
83 import org.opendaylight.openflowplugin.openflow.md.core.sal.convertor.ConvertorManagerFactory;
84 import org.opendaylight.openflowplugin.openflow.md.core.session.OFSessionUtil;
85 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflow.provider.config.rev160510.OpenflowProviderConfig;
86 import org.slf4j.Logger;
87 import org.slf4j.LoggerFactory;
90 @Service(classes = { OpenFlowPluginProvider.class, OpenFlowPluginExtensionRegistratorProvider.class })
91 public class OpenFlowPluginProviderImpl implements
92 OpenFlowPluginProvider,
93 OpenFlowPluginExtensionRegistratorProvider,
96 private static final Logger LOG = LoggerFactory.getLogger(OpenFlowPluginProviderImpl.class);
98 private static final int TICKS_PER_WHEEL = 500; // 0.5 sec.
99 private static final long TICK_DURATION = 10;
100 private static final String POOL_NAME = "ofppool";
102 private static final MessageIntelligenceAgency MESSAGE_INTELLIGENCE_AGENCY = new MessageIntelligenceAgencyImpl();
103 private static final String MESSAGE_INTELLIGENCE_AGENCY_MX_BEAN_NAME = String
104 .format("%s:type=%s",
105 MessageIntelligenceAgencyMXBean.class.getPackage().getName(),
106 MessageIntelligenceAgencyMXBean.class.getSimpleName());
108 private final HashedWheelTimer hashedWheelTimer =
109 new HashedWheelTimer(TICK_DURATION, TimeUnit.MILLISECONDS, TICKS_PER_WHEEL);
110 private final NotificationPublishService notificationPublishService;
111 private final ExtensionConverterManager extensionConverterManager;
112 private final DataBroker dataBroker;
113 private final Collection<SwitchConnectionProvider> switchConnectionProviders;
114 private final DeviceInitializerProvider deviceInitializerProvider;
115 private final ConvertorManager convertorManager;
116 private final RpcProviderService rpcProviderRegistry;
117 private final ClusterSingletonServiceProvider singletonServicesProvider;
118 private final OpenflowProviderConfig config;
119 private final EntityOwnershipService entityOwnershipService;
120 private final MastershipChangeServiceManager mastershipChangeServiceManager;
121 private DeviceManager deviceManager;
122 private RpcManager rpcManager;
123 private StatisticsManager statisticsManager;
124 private RoleManager roleManager;
125 private ConnectionManager connectionManager;
126 private ListeningExecutorService executorService;
127 private ContextChainHolderImpl contextChainHolder;
128 private final OpenflowDiagStatusProvider openflowDiagStatusProvider;
129 private final SystemReadyMonitor systemReadyMonitor;
130 private final SettableFuture<Void> fullyStarted = SettableFuture.create();
131 private static final String OPENFLOW_SERVICE_NAME = "OPENFLOW";
133 public static MessageIntelligenceAgency getMessageIntelligenceAgency() {
134 return MESSAGE_INTELLIGENCE_AGENCY;
138 public OpenFlowPluginProviderImpl(final ConfigurationService configurationService,
139 final SwitchConnectionProviderList switchConnectionProviders,
140 final PingPongDataBroker pingPongDataBroker,
141 final @Reference RpcProviderService rpcProviderRegistry,
142 final @Reference NotificationPublishService notificationPublishService,
143 final @Reference ClusterSingletonServiceProvider singletonServiceProvider,
144 final @Reference EntityOwnershipService entityOwnershipService,
145 final MastershipChangeServiceManager mastershipChangeServiceManager,
146 final @Reference OpenflowDiagStatusProvider openflowDiagStatusProvider,
147 final @Reference SystemReadyMonitor systemReadyMonitor) {
148 this.switchConnectionProviders = switchConnectionProviders;
149 this.dataBroker = pingPongDataBroker;
150 this.rpcProviderRegistry = rpcProviderRegistry;
151 this.notificationPublishService = notificationPublishService;
152 this.singletonServicesProvider = singletonServiceProvider;
153 this.entityOwnershipService = entityOwnershipService;
154 convertorManager = ConvertorManagerFactory.createDefaultManager();
155 extensionConverterManager = new ExtensionConverterManagerImpl();
156 deviceInitializerProvider = DeviceInitializerProviderFactory.createDefaultProvider();
157 config = new OpenFlowProviderConfigImpl(configurationService);
158 this.mastershipChangeServiceManager = mastershipChangeServiceManager;
159 this.openflowDiagStatusProvider = openflowDiagStatusProvider;
160 this.systemReadyMonitor = systemReadyMonitor;
164 public void onSystemBootReady() {
165 LOG.info("onSystemBootReady() received, starting the switch connections");
166 Futures.addCallback(Futures.allAsList(switchConnectionProviders.stream().map(switchConnectionProvider -> {
167 // Inject OpenFlowPlugin custom serializers and deserializers into OpenFlowJava
168 if (config.isUseSingleLayerSerialization()) {
169 SerializerInjector.injectSerializers(switchConnectionProvider,
170 switchConnectionProvider.getConfiguration().isGroupAddModEnabled());
171 DeserializerInjector.injectDeserializers(switchConnectionProvider);
173 DeserializerInjector.revertDeserializers(switchConnectionProvider);
176 // Set handler of incoming connections and start switch connection provider
177 switchConnectionProvider.setSwitchConnectionHandler(connectionManager);
178 return switchConnectionProvider.startup();
179 }).collect(Collectors.toSet())), new FutureCallback<List<Boolean>>() {
181 public void onSuccess(final List<Boolean> result) {
182 LOG.info("All switchConnectionProviders are up and running ({}).", result.size());
183 openflowDiagStatusProvider.reportStatus(OPENFLOW_SERVICE_NAME, ServiceState.OPERATIONAL);
184 fullyStarted.set(null);
188 public void onFailure(final Throwable throwable) {
189 LOG.warn("Some switchConnectionProviders failed to start.", throwable);
190 openflowDiagStatusProvider.reportStatus(OPENFLOW_SERVICE_NAME, throwable);
191 fullyStarted.setException(throwable);
193 }, MoreExecutors.directExecutor());
197 public Future<Void> getFullyStarted() {
201 private ListenableFuture<List<Boolean>> shutdownSwitchConnections() {
202 final ListenableFuture<List<Boolean>> listListenableFuture =
203 Futures.allAsList(switchConnectionProviders.stream().map(switchConnectionProvider -> {
204 // Revert deserializers to their original state
205 if (config.isUseSingleLayerSerialization()) {
206 DeserializerInjector.revertDeserializers(switchConnectionProvider);
209 // Shutdown switch connection provider
210 return switchConnectionProvider.shutdown();
211 }).collect(Collectors.toSet()));
213 Futures.addCallback(listListenableFuture, new FutureCallback<List<Boolean>>() {
215 public void onSuccess(final List<Boolean> result) {
216 LOG.info("All switchConnectionProviders were successfully shut down ({}).", result.size());
220 public void onFailure(final Throwable throwable) {
221 LOG.warn("Some switchConnectionProviders failed to shutdown.", throwable);
223 }, MoreExecutors.directExecutor());
225 return listListenableFuture;
230 public void initialize() {
231 registerMXBean(MESSAGE_INTELLIGENCE_AGENCY, MESSAGE_INTELLIGENCE_AGENCY_MX_BEAN_NAME);
233 // TODO: copied from OpenFlowPluginProvider (Helium) misusesing the old way of distributing extension converters
234 // TODO: rewrite later!
235 OFSessionUtil.getSessionManager().setExtensionConverterProvider(extensionConverterManager);
237 // Creates a thread pool that creates new threads as needed, but will reuse previously
238 // constructed threads when they are available.
239 // Threads that have not been used for x seconds are terminated and removed from the cache.
240 executorService = MoreExecutors.listeningDecorator(new ThreadPoolLoggingExecutor(
241 config.getThreadPoolMinThreads().toJava(),
242 config.getThreadPoolMaxThreads().getValue().toJava(),
243 config.getThreadPoolTimeout().toJava(),
244 TimeUnit.SECONDS, new SynchronousQueue<>(), POOL_NAME));
246 deviceManager = new DeviceManagerImpl(
249 getMessageIntelligenceAgency(),
250 notificationPublishService,
253 deviceInitializerProvider,
256 TranslatorLibraryUtil.injectBasicTranslatorLibrary(deviceManager, convertorManager);
257 ((ExtensionConverterProviderKeeper) deviceManager).setExtensionConverterProvider(extensionConverterManager);
259 rpcManager = new RpcManagerImpl(
262 extensionConverterManager,
264 notificationPublishService);
266 statisticsManager = new StatisticsManagerImpl(
272 roleManager = new RoleManagerImpl(hashedWheelTimer, config, executorService);
274 contextChainHolder = new ContextChainHolderImpl(
276 singletonServicesProvider,
277 entityOwnershipService,
278 mastershipChangeServiceManager,
281 contextChainHolder.addManager(deviceManager);
282 contextChainHolder.addManager(statisticsManager);
283 contextChainHolder.addManager(rpcManager);
284 contextChainHolder.addManager(roleManager);
286 connectionManager = new ConnectionManagerImpl(config, executorService, dataBroker, notificationPublishService);
287 connectionManager.setDeviceConnectedHandler(contextChainHolder);
288 connectionManager.setDeviceDisconnectedHandler(contextChainHolder);
290 deviceManager.setContextChainHolder(contextChainHolder);
291 deviceManager.initialize();
292 systemReadyMonitor.registerListener(this);
293 LOG.info("registered onSystemBootReady() listener for deferred startSwitchConnections()");
297 public ExtensionConverterRegistrator getExtensionConverterRegistrator() {
298 return extensionConverterManager;
303 @SuppressWarnings("checkstyle:IllegalCatch")
304 public void close() {
306 shutdownSwitchConnections().get(10, TimeUnit.SECONDS);
307 } catch (InterruptedException | ExecutionException | TimeoutException e) {
308 LOG.warn("Failed to shut down switch connections in time {}s", 10, e);
311 gracefulShutdown(contextChainHolder);
312 gracefulShutdown(connectionManager);
313 gracefulShutdown(deviceManager);
314 gracefulShutdown(rpcManager);
315 gracefulShutdown(statisticsManager);
316 gracefulShutdown(roleManager);
317 gracefulShutdown(executorService);
318 gracefulShutdown(hashedWheelTimer);
319 unregisterMXBean(MESSAGE_INTELLIGENCE_AGENCY_MX_BEAN_NAME);
320 openflowDiagStatusProvider.reportStatus(ServiceState.UNREGISTERED);
322 if (connectionManager != null) {
323 connectionManager.close();
324 connectionManager = null;
326 } catch (Exception e) {
327 LOG.error("Failed to close ConnectionManager", e);
331 @SuppressWarnings("checkstyle:IllegalCatch")
332 private static void gracefulShutdown(final AutoCloseable closeable) {
333 if (closeable != null) {
336 } catch (Exception e) {
337 LOG.warn("Failed to shutdown {} gracefully.", closeable);
342 private static void gracefulShutdown(final Timer timer) {
346 } catch (IllegalStateException e) {
347 LOG.warn("Failed to shutdown {} gracefully.", timer);
352 private static void gracefulShutdown(final ExecutorService executorService) {
353 if (executorService != null) {
355 executorService.shutdownNow();
356 } catch (SecurityException e) {
357 LOG.warn("Failed to shutdown {} gracefully.", executorService);
362 private static void registerMXBean(final Object bean, final String beanName) {
363 final MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
366 mbs.registerMBean(bean, new ObjectName(beanName));
367 } catch (MalformedObjectNameException
368 | NotCompliantMBeanException
369 | MBeanRegistrationException
370 | InstanceAlreadyExistsException e) {
371 LOG.warn("Error registering MBean {}", beanName, e);
375 private static void unregisterMXBean(final String beanName) {
376 final MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
379 mbs.unregisterMBean(new ObjectName(beanName));
380 } catch (InstanceNotFoundException
381 | MBeanRegistrationException
382 | MalformedObjectNameException e) {
383 LOG.warn("Error unregistering MBean {}", beanName, e);