openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/OpenFlowPluginProviderImpl.java

   1 /*
   2  * Copyright (c) 2015 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   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
   7  */
   8
   9 package org.opendaylight.openflowplugin.impl;
  10
  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 io.netty.util.HashedWheelTimer;
  15 import io.netty.util.Timer;
  16 import java.lang.management.ManagementFactory;
  17 import java.util.Collection;
  18 import java.util.List;
  19 import java.util.Objects;
  20 import java.util.concurrent.ExecutionException;
  21 import java.util.concurrent.SynchronousQueue;
  22 import java.util.concurrent.ThreadPoolExecutor;
  23 import java.util.concurrent.TimeUnit;
  24 import java.util.concurrent.TimeoutException;
  25 import java.util.stream.Collectors;
  26 import javax.annotation.Nonnull;
  27 import javax.management.InstanceAlreadyExistsException;
  28 import javax.management.InstanceNotFoundException;
  29 import javax.management.MBeanRegistrationException;
  30 import javax.management.MBeanServer;
  31 import javax.management.MalformedObjectNameException;
  32 import javax.management.NotCompliantMBeanException;
  33 import javax.management.ObjectName;
  34 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  35 import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService;
  36 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipService;
  37 import org.opendaylight.controller.sal.binding.api.RpcProviderRegistry;
  38 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceProvider;
  39 import org.opendaylight.openflowjava.protocol.spi.connection.SwitchConnectionProvider;
  40 import org.opendaylight.openflowplugin.api.openflow.OpenFlowPluginProvider;
  41 import org.opendaylight.openflowplugin.api.openflow.configuration.ConfigurationService;
  42 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionManager;
  43 import org.opendaylight.openflowplugin.api.openflow.device.DeviceManager;
  44 import org.opendaylight.openflowplugin.api.openflow.lifecycle.OwnershipChangeListener;
  45 import org.opendaylight.openflowplugin.api.openflow.lifecycle.ReconciliationFrameworkRegistrar;
  46 import org.opendaylight.openflowplugin.api.openflow.mastership.MastershipChangeServiceManager;
  47 import org.opendaylight.openflowplugin.api.openflow.rpc.RpcManager;
  48 import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsManager;
  49 import org.opendaylight.openflowplugin.api.openflow.statistics.ofpspecific.MessageIntelligenceAgency;
  50 import org.opendaylight.openflowplugin.extension.api.ExtensionConverterProviderKeeper;
  51 import org.opendaylight.openflowplugin.extension.api.ExtensionConverterRegistrator;
  52 import org.opendaylight.openflowplugin.extension.api.OpenFlowPluginExtensionRegistratorProvider;
  53 import org.opendaylight.openflowplugin.extension.api.core.extension.ExtensionConverterManager;
  54 import org.opendaylight.openflowplugin.impl.configuration.OpenFlowProviderConfigImpl;
  55 import org.opendaylight.openflowplugin.impl.connection.ConnectionManagerImpl;
  56 import org.opendaylight.openflowplugin.impl.device.DeviceManagerImpl;
  57 import org.opendaylight.openflowplugin.impl.device.initialization.DeviceInitializerProvider;
  58 import org.opendaylight.openflowplugin.impl.device.initialization.DeviceInitializerProviderFactory;
  59 import org.opendaylight.openflowplugin.impl.lifecycle.ContextChainHolderImpl;
  60 import org.opendaylight.openflowplugin.impl.protocol.deserialization.DeserializerInjector;
  61 import org.opendaylight.openflowplugin.impl.protocol.serialization.SerializerInjector;
  62 import org.opendaylight.openflowplugin.impl.rpc.RpcManagerImpl;
  63 import org.opendaylight.openflowplugin.impl.statistics.StatisticsManagerImpl;
  64 import org.opendaylight.openflowplugin.impl.statistics.ofpspecific.MessageIntelligenceAgencyImpl;
  65 import org.opendaylight.openflowplugin.impl.statistics.ofpspecific.MessageIntelligenceAgencyMXBean;
  66 import org.opendaylight.openflowplugin.impl.util.TranslatorLibraryUtil;
  67 import org.opendaylight.openflowplugin.openflow.md.core.ThreadPoolLoggingExecutor;
  68 import org.opendaylight.openflowplugin.openflow.md.core.extension.ExtensionConverterManagerImpl;
  69 import org.opendaylight.openflowplugin.openflow.md.core.sal.convertor.ConvertorManager;
  70 import org.opendaylight.openflowplugin.openflow.md.core.sal.convertor.ConvertorManagerFactory;
  71 import org.opendaylight.openflowplugin.openflow.md.core.session.OFSessionUtil;
  72 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflow.provider.config.rev160510.OpenflowProviderConfig;
  73 import org.slf4j.Logger;
  74 import org.slf4j.LoggerFactory;
  75
  76 public class OpenFlowPluginProviderImpl implements
  77         OpenFlowPluginProvider,
  78         OpenFlowPluginExtensionRegistratorProvider {
  79
  80     private static final Logger LOG = LoggerFactory.getLogger(OpenFlowPluginProviderImpl.class);
  81
  82     private static final int TICKS_PER_WHEEL = 500; // 0.5 sec.
  83     private static final long TICK_DURATION = 10;
  84     private static final String POOL_NAME = "ofppool";
  85
  86     private static final MessageIntelligenceAgency MESSAGE_INTELLIGENCE_AGENCY = new MessageIntelligenceAgencyImpl();
  87     private static final String MESSAGE_INTELLIGENCE_AGENCY_MX_BEAN_NAME = String
  88             .format("%s:type=%s",
  89                     MessageIntelligenceAgencyMXBean.class.getPackage().getName(),
  90                     MessageIntelligenceAgencyMXBean.class.getSimpleName());
  91
  92     private final HashedWheelTimer hashedWheelTimer = new HashedWheelTimer(TICK_DURATION, TimeUnit.MILLISECONDS, TICKS_PER_WHEEL);
  93     private final NotificationPublishService notificationPublishService;
  94     private final ExtensionConverterManager extensionConverterManager;
  95     private final DataBroker dataBroker;
  96     private final Collection<SwitchConnectionProvider> switchConnectionProviders;
  97     private final DeviceInitializerProvider deviceInitializerProvider;
  98     private final ConvertorManager convertorManager;
  99     private final RpcProviderRegistry rpcProviderRegistry;
 100     private final ClusterSingletonServiceProvider singletonServicesProvider;
 101     private final OpenflowProviderConfig config;
 102     private final EntityOwnershipService entityOwnershipService;
 103     private final MastershipChangeServiceManager mastershipChangeServiceManager;
 104     private DeviceManager deviceManager;
 105     private RpcManager rpcManager;
 106     private StatisticsManager statisticsManager;
 107     private ConnectionManager connectionManager;
 108     private ThreadPoolExecutor threadPool;
 109     private ContextChainHolderImpl contextChainHolder;
 110
 111     public static MessageIntelligenceAgency getMessageIntelligenceAgency() {
 112         return MESSAGE_INTELLIGENCE_AGENCY;
 113     }
 114
 115     OpenFlowPluginProviderImpl(final ConfigurationService configurationService,
 116                                final List<SwitchConnectionProvider> switchConnectionProviders,
 117                                final DataBroker dataBroker,
 118                                final RpcProviderRegistry rpcProviderRegistry,
 119                                final NotificationPublishService notificationPublishService,
 120                                final ClusterSingletonServiceProvider singletonServiceProvider,
 121                                final EntityOwnershipService entityOwnershipService,
 122                                final MastershipChangeServiceManager mastershipChangeServiceManager) {
 123         this.switchConnectionProviders = switchConnectionProviders;
 124         this.dataBroker = dataBroker;
 125         this.rpcProviderRegistry = rpcProviderRegistry;
 126         this.notificationPublishService = notificationPublishService;
 127         this.singletonServicesProvider = singletonServiceProvider;
 128         this.entityOwnershipService = entityOwnershipService;
 129         convertorManager = ConvertorManagerFactory.createDefaultManager();
 130         extensionConverterManager = new ExtensionConverterManagerImpl();
 131         deviceInitializerProvider = DeviceInitializerProviderFactory.createDefaultProvider();
 132         config = new OpenFlowProviderConfigImpl(configurationService);
 133         this.mastershipChangeServiceManager = mastershipChangeServiceManager;
 134     }
 135
 136
 137     private void startSwitchConnections() {
 138         Futures.addCallback(Futures.allAsList(switchConnectionProviders.stream().map(switchConnectionProvider -> {
 139             // Inject OpenFlowPlugin custom serializers and deserializers into OpenFlowJava
 140             if (config.isUseSingleLayerSerialization()) {
 141                 SerializerInjector.injectSerializers(switchConnectionProvider);
 142                 DeserializerInjector.injectDeserializers(switchConnectionProvider);
 143             } else {
 144                 DeserializerInjector.revertDeserializers(switchConnectionProvider);
 145             }
 146
 147             // Set handler of incoming connections and start switch connection provider
 148             switchConnectionProvider.setSwitchConnectionHandler(connectionManager);
 149             return switchConnectionProvider.startup();
 150         }).collect(Collectors.toSet())), new FutureCallback<List<Boolean>>() {
 151             @Override
 152             public void onSuccess(final List<Boolean> result) {
 153                 LOG.info("All switchConnectionProviders are up and running ({}).", result.size());
 154             }
 155
 156             @Override
 157             public void onFailure(@Nonnull final Throwable throwable) {
 158                 LOG.warn("Some switchConnectionProviders failed to start.", throwable);
 159             }
 160         });
 161     }
 162
 163     private ListenableFuture<List<Boolean>> shutdownSwitchConnections() {
 164         final ListenableFuture<List<Boolean>> listListenableFuture = Futures.allAsList(switchConnectionProviders.stream().map(switchConnectionProvider -> {
 165             // Revert deserializers to their original state
 166             if (config.isUseSingleLayerSerialization()) {
 167                 DeserializerInjector.revertDeserializers(switchConnectionProvider);
 168             }
 169
 170             // Shutdown switch connection provider
 171             return switchConnectionProvider.shutdown();
 172         }).collect(Collectors.toSet()));
 173
 174         Futures.addCallback(listListenableFuture, new FutureCallback<List<Boolean>>() {
 175             @Override
 176             public void onSuccess(final List<Boolean> result) {
 177                 LOG.info("All switchConnectionProviders were successfully shut down ({}).", result.size());
 178             }
 179
 180             @Override
 181             public void onFailure(@Nonnull final Throwable throwable) {
 182                 LOG.warn("Some switchConnectionProviders failed to shutdown.", throwable);
 183             }
 184         });
 185
 186         return listListenableFuture;
 187     }
 188
 189     @Override
 190     public void initialize() {
 191         registerMXBean(MESSAGE_INTELLIGENCE_AGENCY, MESSAGE_INTELLIGENCE_AGENCY_MX_BEAN_NAME);
 192
 193         // TODO: copied from OpenFlowPluginProvider (Helium) misusesing the old way of distributing extension converters
 194         // TODO: rewrite later!
 195         OFSessionUtil.getSessionManager().setExtensionConverterProvider(extensionConverterManager);
 196
 197         // Creates a thread pool that creates new threads as needed, but will reuse previously
 198         // constructed threads when they are available.
 199         // Threads that have not been used for x seconds are terminated and removed from the cache.
 200         threadPool = new ThreadPoolLoggingExecutor(
 201                 config.getThreadPoolMinThreads(),
 202                 config.getThreadPoolMaxThreads().getValue(),
 203                 config.getThreadPoolTimeout(),
 204                 TimeUnit.SECONDS, new SynchronousQueue<>(), POOL_NAME);
 205
 206         deviceManager = new DeviceManagerImpl(
 207                 config,
 208                 dataBroker,
 209                 getMessageIntelligenceAgency(),
 210                 notificationPublishService,
 211                 hashedWheelTimer,
 212                 convertorManager,
 213                 deviceInitializerProvider);
 214
 215         TranslatorLibraryUtil.injectBasicTranslatorLibrary(deviceManager, convertorManager);
 216         ((ExtensionConverterProviderKeeper) deviceManager).setExtensionConverterProvider(extensionConverterManager);
 217
 218         rpcManager = new RpcManagerImpl(
 219                 config,
 220                 rpcProviderRegistry,
 221                 extensionConverterManager,
 222                 convertorManager,
 223                 notificationPublishService);
 224
 225         statisticsManager = new StatisticsManagerImpl(
 226                 config,
 227                 rpcProviderRegistry,
 228                 hashedWheelTimer,
 229                 convertorManager);
 230
 231         contextChainHolder = new ContextChainHolderImpl(
 232                 hashedWheelTimer,
 233                 threadPool,
 234                 singletonServicesProvider,
 235                 entityOwnershipService,
 236                 mastershipChangeServiceManager
 237         );
 238
 239         statisticsManager.setReconciliationFrameworkRegistrar(mastershipChangeServiceManager);
 240
 241         contextChainHolder.addManager(deviceManager);
 242         contextChainHolder.addManager(statisticsManager);
 243         contextChainHolder.addManager(rpcManager);
 244
 245         connectionManager = new ConnectionManagerImpl(config, threadPool);
 246         connectionManager.setDeviceConnectedHandler(contextChainHolder);
 247         connectionManager.setDeviceDisconnectedHandler(contextChainHolder);
 248
 249         deviceManager.initialize();
 250         startSwitchConnections();
 251     }
 252
 253     @Override
 254     public ExtensionConverterRegistrator getExtensionConverterRegistrator() {
 255         return extensionConverterManager;
 256     }
 257
 258     @Override
 259     public void close() {
 260         try {
 261             shutdownSwitchConnections().get(10, TimeUnit.SECONDS);
 262         } catch (InterruptedException | ExecutionException | TimeoutException e) {
 263             LOG.warn("Failed to shut down switch connections in time {}s, error: {}", 10, e);
 264         }
 265
 266         gracefulShutdown(contextChainHolder);
 267         gracefulShutdown(deviceManager);
 268         gracefulShutdown(rpcManager);
 269         gracefulShutdown(statisticsManager);
 270         gracefulShutdown(threadPool);
 271         gracefulShutdown(hashedWheelTimer);
 272         unregisterMXBean(MESSAGE_INTELLIGENCE_AGENCY_MX_BEAN_NAME);
 273     }
 274
 275     private static void gracefulShutdown(final AutoCloseable closeable) {
 276         if (Objects.isNull(closeable)) {
 277             return;
 278         }
 279
 280         try {
 281             closeable.close();
 282         } catch (Exception e) {
 283             LOG.warn("Failed to shutdown {} gracefully.", closeable);
 284         }
 285     }
 286
 287     private static void gracefulShutdown(final Timer timer) {
 288         if (Objects.isNull(timer)) {
 289             return;
 290         }
 291
 292         try {
 293             timer.stop();
 294         } catch (Exception e) {
 295             LOG.warn("Failed to shutdown {} gracefully.", timer);
 296         }
 297     }
 298
 299     private static void gracefulShutdown(final ThreadPoolExecutor threadPoolExecutor) {
 300         if (Objects.isNull(threadPoolExecutor)) {
 301             return;
 302         }
 303
 304         try {
 305             threadPoolExecutor.shutdownNow();
 306         } catch (Exception e) {
 307             LOG.warn("Failed to shutdown {} gracefully.", threadPoolExecutor);
 308         }
 309     }
 310
 311     private static void registerMXBean(final Object bean, final String beanName) {
 312         final MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
 313
 314         try {
 315             mbs.registerMBean(bean, new ObjectName(beanName));
 316         } catch (MalformedObjectNameException
 317                 | NotCompliantMBeanException
 318                 | MBeanRegistrationException
 319                 | InstanceAlreadyExistsException e) {
 320             LOG.warn("Error registering MBean {}", e);
 321         }
 322     }
 323
 324     private static void unregisterMXBean(final String beanName) {
 325         final MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
 326
 327         try {
 328             mbs.unregisterMBean(new ObjectName(beanName));
 329         } catch (InstanceNotFoundException
 330                 | MBeanRegistrationException
 331                 | MalformedObjectNameException e) {
 332             LOG.warn("Error unregistering MBean {}", e);
 333         }
 334     }
 335 }