* terms of the Eclipse Public License v1.0 which accompanies this distribution,
* and is available at http://www.eclipse.org/legal/epl-v10.html
*/
-
package org.opendaylight.openflowplugin.impl;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
+import com.google.common.util.concurrent.SettableFuture;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timer;
-import java.lang.management.ManagementFactory;
import java.util.Collection;
import java.util.List;
-import java.util.Objects;
+import java.util.Map;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
import java.util.concurrent.SynchronousQueue;
-import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;
-import javax.annotation.Nonnull;
-import javax.management.InstanceAlreadyExistsException;
-import javax.management.InstanceNotFoundException;
-import javax.management.MBeanRegistrationException;
-import javax.management.MBeanServer;
-import javax.management.MalformedObjectNameException;
-import javax.management.NotCompliantMBeanException;
-import javax.management.ObjectName;
-import org.opendaylight.controller.md.sal.binding.api.DataBroker;
-import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService;
-import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipService;
-import org.opendaylight.controller.sal.binding.api.RpcProviderRegistry;
-import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceProvider;
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+import javax.inject.Inject;
+import javax.inject.Singleton;
+import org.opendaylight.infrautils.diagstatus.ServiceState;
+import org.opendaylight.infrautils.ready.SystemReadyListener;
+import org.opendaylight.infrautils.ready.SystemReadyMonitor;
+import org.opendaylight.mdsal.binding.api.DataBroker;
+import org.opendaylight.mdsal.binding.api.NotificationPublishService;
+import org.opendaylight.mdsal.binding.api.RpcProviderService;
+import org.opendaylight.mdsal.eos.binding.api.EntityOwnershipService;
+import org.opendaylight.mdsal.singleton.api.ClusterSingletonServiceProvider;
import org.opendaylight.openflowjava.protocol.spi.connection.SwitchConnectionProvider;
+import org.opendaylight.openflowplugin.api.openflow.FlowGroupInfoHistories;
+import org.opendaylight.openflowplugin.api.openflow.FlowGroupInfoHistory;
import org.opendaylight.openflowplugin.api.openflow.OpenFlowPluginProvider;
import org.opendaylight.openflowplugin.api.openflow.configuration.ConfigurationService;
import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionManager;
import org.opendaylight.openflowplugin.api.openflow.device.DeviceManager;
import org.opendaylight.openflowplugin.api.openflow.mastership.MastershipChangeServiceManager;
+import org.opendaylight.openflowplugin.api.openflow.role.RoleManager;
import org.opendaylight.openflowplugin.api.openflow.rpc.RpcManager;
import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsManager;
import org.opendaylight.openflowplugin.api.openflow.statistics.ofpspecific.MessageIntelligenceAgency;
import org.opendaylight.openflowplugin.impl.lifecycle.ContextChainHolderImpl;
import org.opendaylight.openflowplugin.impl.protocol.deserialization.DeserializerInjector;
import org.opendaylight.openflowplugin.impl.protocol.serialization.SerializerInjector;
+import org.opendaylight.openflowplugin.impl.role.RoleManagerImpl;
import org.opendaylight.openflowplugin.impl.rpc.RpcManagerImpl;
import org.opendaylight.openflowplugin.impl.statistics.StatisticsManagerImpl;
-import org.opendaylight.openflowplugin.impl.statistics.ofpspecific.MessageIntelligenceAgencyImpl;
-import org.opendaylight.openflowplugin.impl.statistics.ofpspecific.MessageIntelligenceAgencyMXBean;
+import org.opendaylight.openflowplugin.impl.util.ThreadPoolLoggingExecutor;
import org.opendaylight.openflowplugin.impl.util.TranslatorLibraryUtil;
-import org.opendaylight.openflowplugin.openflow.md.core.ThreadPoolLoggingExecutor;
import org.opendaylight.openflowplugin.openflow.md.core.extension.ExtensionConverterManagerImpl;
import org.opendaylight.openflowplugin.openflow.md.core.sal.convertor.ConvertorManager;
import org.opendaylight.openflowplugin.openflow.md.core.sal.convertor.ConvertorManagerFactory;
import org.opendaylight.openflowplugin.openflow.md.core.session.OFSessionUtil;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflow.provider.config.rev160510.OpenflowProviderConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+@Singleton
public class OpenFlowPluginProviderImpl implements
OpenFlowPluginProvider,
- OpenFlowPluginExtensionRegistratorProvider {
+ OpenFlowPluginExtensionRegistratorProvider,
+ FlowGroupInfoHistories,
+ SystemReadyListener {
private static final Logger LOG = LoggerFactory.getLogger(OpenFlowPluginProviderImpl.class);
private static final long TICK_DURATION = 10;
private static final String POOL_NAME = "ofppool";
- private static final MessageIntelligenceAgency MESSAGE_INTELLIGENCE_AGENCY = new MessageIntelligenceAgencyImpl();
- private static final String MESSAGE_INTELLIGENCE_AGENCY_MX_BEAN_NAME = String
- .format("%s:type=%s",
- MessageIntelligenceAgencyMXBean.class.getPackage().getName(),
- MessageIntelligenceAgencyMXBean.class.getSimpleName());
-
- private final HashedWheelTimer hashedWheelTimer = new HashedWheelTimer(TICK_DURATION, TimeUnit.MILLISECONDS, TICKS_PER_WHEEL);
+ // TODO: Split this out into a separate component, which requires proper timer cancellation from all users. But is
+ // that worth the complications?
+ private final HashedWheelTimer hashedWheelTimer =
+ new HashedWheelTimer(TICK_DURATION, TimeUnit.MILLISECONDS, TICKS_PER_WHEEL);
private final NotificationPublishService notificationPublishService;
private final ExtensionConverterManager extensionConverterManager;
private final DataBroker dataBroker;
private final Collection<SwitchConnectionProvider> switchConnectionProviders;
private final DeviceInitializerProvider deviceInitializerProvider;
private final ConvertorManager convertorManager;
- private final RpcProviderRegistry rpcProviderRegistry;
+ private final RpcProviderService rpcProviderRegistry;
private final ClusterSingletonServiceProvider singletonServicesProvider;
private final OpenflowProviderConfig config;
private final EntityOwnershipService entityOwnershipService;
private DeviceManager deviceManager;
private RpcManager rpcManager;
private StatisticsManager statisticsManager;
+ private RoleManager roleManager;
private ConnectionManager connectionManager;
- private ThreadPoolExecutor threadPool;
+ private ExecutorService executorService;
private ContextChainHolderImpl contextChainHolder;
-
- public static MessageIntelligenceAgency getMessageIntelligenceAgency() {
- return MESSAGE_INTELLIGENCE_AGENCY;
- }
-
- OpenFlowPluginProviderImpl(final ConfigurationService configurationService,
+ private final MessageIntelligenceAgency messageIntelligenceAgency;
+ private final DiagStatusProvider diagStatusProvider;
+ private final SystemReadyMonitor systemReadyMonitor;
+ private final SettableFuture<Void> fullyStarted = SettableFuture.create();
+ private static final String OPENFLOW_SERVICE_NAME = "OPENFLOW";
+
+ @Inject
+ public OpenFlowPluginProviderImpl(final ConfigurationService configurationService,
final List<SwitchConnectionProvider> switchConnectionProviders,
final DataBroker dataBroker,
- final RpcProviderRegistry rpcProviderRegistry,
+ final RpcProviderService rpcProviderRegistry,
final NotificationPublishService notificationPublishService,
final ClusterSingletonServiceProvider singletonServiceProvider,
final EntityOwnershipService entityOwnershipService,
- final MastershipChangeServiceManager mastershipChangeServiceManager) {
+ final MastershipChangeServiceManager mastershipChangeServiceManager,
+ final MessageIntelligenceAgency messageIntelligenceAgency,
+ final DiagStatusProvider diagStatusProvider,
+ final SystemReadyMonitor systemReadyMonitor) {
this.switchConnectionProviders = switchConnectionProviders;
- this.dataBroker = dataBroker;
+ this.dataBroker = new PingPongDataBroker(dataBroker);
this.rpcProviderRegistry = rpcProviderRegistry;
this.notificationPublishService = notificationPublishService;
- this.singletonServicesProvider = singletonServiceProvider;
+ singletonServicesProvider = singletonServiceProvider;
this.entityOwnershipService = entityOwnershipService;
convertorManager = ConvertorManagerFactory.createDefaultManager();
extensionConverterManager = new ExtensionConverterManagerImpl();
deviceInitializerProvider = DeviceInitializerProviderFactory.createDefaultProvider();
config = new OpenFlowProviderConfigImpl(configurationService);
this.mastershipChangeServiceManager = mastershipChangeServiceManager;
+ this.messageIntelligenceAgency = messageIntelligenceAgency;
+ this.diagStatusProvider = diagStatusProvider;
+ this.systemReadyMonitor = systemReadyMonitor;
}
-
- private void startSwitchConnections() {
+ @Override
+ public void onSystemBootReady() {
+ LOG.info("onSystemBootReady() received, starting the switch connections");
Futures.addCallback(Futures.allAsList(switchConnectionProviders.stream().map(switchConnectionProvider -> {
// Inject OpenFlowPlugin custom serializers and deserializers into OpenFlowJava
- if (config.isUseSingleLayerSerialization()) {
- SerializerInjector.injectSerializers(switchConnectionProvider);
+ if (config.getUseSingleLayerSerialization()) {
+ SerializerInjector.injectSerializers(switchConnectionProvider,
+ switchConnectionProvider.getConfiguration().isGroupAddModEnabled());
DeserializerInjector.injectDeserializers(switchConnectionProvider);
} else {
DeserializerInjector.revertDeserializers(switchConnectionProvider);
@Override
public void onSuccess(final List<Boolean> result) {
LOG.info("All switchConnectionProviders are up and running ({}).", result.size());
+ diagStatusProvider.reportStatus(ServiceState.OPERATIONAL);
+ fullyStarted.set(null);
}
@Override
- public void onFailure(@Nonnull final Throwable throwable) {
+ public void onFailure(final Throwable throwable) {
LOG.warn("Some switchConnectionProviders failed to start.", throwable);
+ diagStatusProvider.reportStatus(ServiceState.ERROR, throwable);
+ fullyStarted.setException(throwable);
}
- });
+ }, MoreExecutors.directExecutor());
+ }
+
+ @VisibleForTesting
+ public Future<Void> getFullyStarted() {
+ return fullyStarted;
}
private ListenableFuture<List<Boolean>> shutdownSwitchConnections() {
- final ListenableFuture<List<Boolean>> listListenableFuture = Futures.allAsList(switchConnectionProviders.stream().map(switchConnectionProvider -> {
- // Revert deserializers to their original state
- if (config.isUseSingleLayerSerialization()) {
- DeserializerInjector.revertDeserializers(switchConnectionProvider);
- }
+ final ListenableFuture<List<Boolean>> listListenableFuture =
+ Futures.allAsList(switchConnectionProviders.stream().map(switchConnectionProvider -> {
+ // Revert deserializers to their original state
+ if (config.getUseSingleLayerSerialization()) {
+ DeserializerInjector.revertDeserializers(switchConnectionProvider);
+ }
- // Shutdown switch connection provider
- return switchConnectionProvider.shutdown();
- }).collect(Collectors.toSet()));
+ // Shutdown switch connection provider
+ return switchConnectionProvider.shutdown();
+ }).collect(Collectors.toSet()));
Futures.addCallback(listListenableFuture, new FutureCallback<List<Boolean>>() {
@Override
}
@Override
- public void onFailure(@Nonnull final Throwable throwable) {
+ public void onFailure(final Throwable throwable) {
LOG.warn("Some switchConnectionProviders failed to shutdown.", throwable);
}
- });
+ }, MoreExecutors.directExecutor());
return listListenableFuture;
}
@Override
+ @PostConstruct
public void initialize() {
- registerMXBean(MESSAGE_INTELLIGENCE_AGENCY, MESSAGE_INTELLIGENCE_AGENCY_MX_BEAN_NAME);
-
// TODO: copied from OpenFlowPluginProvider (Helium) misusesing the old way of distributing extension converters
// TODO: rewrite later!
OFSessionUtil.getSessionManager().setExtensionConverterProvider(extensionConverterManager);
// Creates a thread pool that creates new threads as needed, but will reuse previously
// constructed threads when they are available.
// Threads that have not been used for x seconds are terminated and removed from the cache.
- threadPool = new ThreadPoolLoggingExecutor(
- config.getThreadPoolMinThreads(),
- config.getThreadPoolMaxThreads().getValue(),
- config.getThreadPoolTimeout(),
+ executorService = new ThreadPoolLoggingExecutor(
+ config.getThreadPoolMinThreads().toJava(),
+ config.getThreadPoolMaxThreads().getValue().toJava(),
+ config.getThreadPoolTimeout().toJava(),
TimeUnit.SECONDS, new SynchronousQueue<>(), POOL_NAME);
deviceManager = new DeviceManagerImpl(
config,
dataBroker,
- getMessageIntelligenceAgency(),
+ messageIntelligenceAgency,
notificationPublishService,
hashedWheelTimer,
convertorManager,
- deviceInitializerProvider);
+ deviceInitializerProvider,
+ executorService);
TranslatorLibraryUtil.injectBasicTranslatorLibrary(deviceManager, convertorManager);
((ExtensionConverterProviderKeeper) deviceManager).setExtensionConverterProvider(extensionConverterManager);
statisticsManager = new StatisticsManagerImpl(
config,
rpcProviderRegistry,
- hashedWheelTimer,
- convertorManager);
+ convertorManager,
+ executorService);
+
+ roleManager = new RoleManagerImpl(hashedWheelTimer, config, executorService);
contextChainHolder = new ContextChainHolderImpl(
- hashedWheelTimer,
- threadPool,
+ executorService,
singletonServicesProvider,
entityOwnershipService,
- mastershipChangeServiceManager
- );
-
- statisticsManager.setReconciliationFrameworkRegistrar(mastershipChangeServiceManager);
+ mastershipChangeServiceManager,
+ config);
contextChainHolder.addManager(deviceManager);
contextChainHolder.addManager(statisticsManager);
contextChainHolder.addManager(rpcManager);
+ contextChainHolder.addManager(roleManager);
- connectionManager = new ConnectionManagerImpl(config, threadPool);
+ connectionManager = new ConnectionManagerImpl(config, executorService, dataBroker, notificationPublishService);
connectionManager.setDeviceConnectedHandler(contextChainHolder);
connectionManager.setDeviceDisconnectedHandler(contextChainHolder);
+ deviceManager.setContextChainHolder(contextChainHolder);
deviceManager.initialize();
- startSwitchConnections();
+ systemReadyMonitor.registerListener(this);
+ LOG.info("registered onSystemBootReady() listener for OpenFlowPluginProvider");
}
@Override
}
@Override
+ public Map<NodeId, FlowGroupInfoHistory> getAllFlowGroupHistories() {
+ return deviceManager.getAllFlowGroupHistories();
+ }
+
+ @Override
+ public FlowGroupInfoHistory getFlowGroupHistory(final NodeId nodeId) {
+ return deviceManager.getFlowGroupHistory(nodeId);
+ }
+
+ @Override
+ @PreDestroy
+ @SuppressWarnings("checkstyle:IllegalCatch")
public void close() {
try {
shutdownSwitchConnections().get(10, TimeUnit.SECONDS);
} catch (InterruptedException | ExecutionException | TimeoutException e) {
- LOG.warn("Failed to shut down switch connections in time {}s, error: {}", 10, e);
+ LOG.warn("Failed to shut down switch connections in time {}s", 10, e);
}
gracefulShutdown(contextChainHolder);
+ gracefulShutdown(connectionManager);
gracefulShutdown(deviceManager);
gracefulShutdown(rpcManager);
gracefulShutdown(statisticsManager);
- gracefulShutdown(threadPool);
+ gracefulShutdown(roleManager);
+ gracefulShutdown(executorService);
gracefulShutdown(hashedWheelTimer);
- unregisterMXBean(MESSAGE_INTELLIGENCE_AGENCY_MX_BEAN_NAME);
- }
-
- private static void gracefulShutdown(final AutoCloseable closeable) {
- if (Objects.isNull(closeable)) {
- return;
- }
-
+ diagStatusProvider.reportStatus(ServiceState.UNREGISTERED);
try {
- closeable.close();
+ if (connectionManager != null) {
+ connectionManager.close();
+ connectionManager = null;
+ }
} catch (Exception e) {
- LOG.warn("Failed to shutdown {} gracefully.", closeable);
+ LOG.error("Failed to close ConnectionManager", e);
}
}
- private static void gracefulShutdown(final Timer timer) {
- if (Objects.isNull(timer)) {
- return;
- }
-
- try {
- timer.stop();
- } catch (Exception e) {
- LOG.warn("Failed to shutdown {} gracefully.", timer);
+ @SuppressWarnings("checkstyle:IllegalCatch")
+ private static void gracefulShutdown(final AutoCloseable closeable) {
+ if (closeable != null) {
+ try {
+ closeable.close();
+ } catch (Exception e) {
+ LOG.warn("Failed to shutdown {} gracefully.", closeable);
+ }
}
}
- private static void gracefulShutdown(final ThreadPoolExecutor threadPoolExecutor) {
- if (Objects.isNull(threadPoolExecutor)) {
- return;
- }
-
- try {
- threadPoolExecutor.shutdownNow();
- } catch (Exception e) {
- LOG.warn("Failed to shutdown {} gracefully.", threadPoolExecutor);
+ private static void gracefulShutdown(final Timer timer) {
+ if (timer != null) {
+ try {
+ timer.stop();
+ } catch (IllegalStateException e) {
+ LOG.warn("Failed to shutdown {} gracefully.", timer);
+ }
}
}
- private static void registerMXBean(final Object bean, final String beanName) {
- final MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
-
- try {
- mbs.registerMBean(bean, new ObjectName(beanName));
- } catch (MalformedObjectNameException
- | NotCompliantMBeanException
- | MBeanRegistrationException
- | InstanceAlreadyExistsException e) {
- LOG.warn("Error registering MBean {}", e);
+ private static void gracefulShutdown(final ExecutorService executorService) {
+ if (executorService != null) {
+ try {
+ executorService.shutdownNow();
+ } catch (SecurityException e) {
+ LOG.warn("Failed to shutdown {} gracefully.", executorService);
+ }
}
}
- private static void unregisterMXBean(final String beanName) {
- final MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
- try {
- mbs.unregisterMBean(new ObjectName(beanName));
- } catch (InstanceNotFoundException
- | MBeanRegistrationException
- | MalformedObjectNameException e) {
- LOG.warn("Error unregistering MBean {}", e);
- }
- }
}
Code Review / openflowplugin.git / blobdiff