library/src/main/java/org/opendaylight/ovsdb/lib/impl/OvsdbConnectionService.java

   1 /*
   2  * Copyright (C) 2014 Red Hat, Inc.
   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  * Authors : Madhu Venugopal
   9  */
  10
  11 package org.opendaylight.ovsdb.lib.impl;
  12
  13 import io.netty.bootstrap.Bootstrap;
  14 import io.netty.bootstrap.ServerBootstrap;
  15 import io.netty.channel.AdaptiveRecvByteBufAllocator;
  16 import io.netty.channel.Channel;
  17 import io.netty.channel.ChannelFuture;
  18 import io.netty.channel.ChannelInitializer;
  19 import io.netty.channel.ChannelOption;
  20 import io.netty.channel.EventLoopGroup;
  21 import io.netty.channel.nio.NioEventLoopGroup;
  22 import io.netty.channel.socket.SocketChannel;
  23 import io.netty.channel.socket.nio.NioServerSocketChannel;
  24 import io.netty.channel.socket.nio.NioSocketChannel;
  25 import io.netty.handler.codec.string.StringEncoder;
  26 import io.netty.handler.logging.LogLevel;
  27 import io.netty.handler.logging.LoggingHandler;
  28 import io.netty.util.CharsetUtil;
  29 import io.netty.handler.ssl.SslHandler;
  30
  31 import javax.net.ssl.SSLContext;
  32 import javax.net.ssl.SSLEngine;
  33
  34 import java.net.InetAddress;
  35 import java.util.Arrays;
  36 import java.util.Collection;
  37 import java.util.Map;
  38 import java.util.Set;
  39 import java.util.concurrent.ExecutorService;
  40 import java.util.concurrent.Executors;
  41
  42 import org.opendaylight.ovsdb.lib.OvsdbClient;
  43 import org.opendaylight.ovsdb.lib.OvsdbConnection;
  44 import org.opendaylight.ovsdb.lib.OvsdbConnectionInfo.ConnectionType;
  45 import org.opendaylight.ovsdb.lib.OvsdbConnectionListener;
  46 import org.opendaylight.ovsdb.lib.jsonrpc.ExceptionHandler;
  47 import org.opendaylight.ovsdb.lib.jsonrpc.JsonRpcDecoder;
  48 import org.opendaylight.ovsdb.lib.jsonrpc.JsonRpcEndpoint;
  49 import org.opendaylight.ovsdb.lib.jsonrpc.JsonRpcServiceBinderHandler;
  50 import org.opendaylight.ovsdb.lib.message.OvsdbRPC;
  51 import org.slf4j.Logger;
  52 import org.slf4j.LoggerFactory;
  53
  54 import com.fasterxml.jackson.annotation.JsonInclude.Include;
  55 import com.fasterxml.jackson.databind.DeserializationFeature;
  56 import com.fasterxml.jackson.databind.ObjectMapper;
  57 import com.google.common.collect.Maps;
  58 import com.google.common.collect.Sets;
  59
  60 /**
  61  * OvsDBConnectionService provides OVSDB connection management functionality which includes
  62  * both Active and Passive connections.
  63  * From the Library perspective, Active OVSDB connections are those that are initiated from
  64  * the Controller towards the ovsdb-manager.
  65  * While Passive OVSDB connections are those that are initiated from the ovs towards
  66  * the controller.
  67  *
  68  * Applications that use OvsDBConnectionService can use the OvsDBConnection class' connect APIs
  69  * to initiate Active connections and can listen to the asynchronous Passive connections via
  70  * registerConnectionListener listener API.
  71  *
  72  * The library is designed as Java modular component that can work in both OSGi and non-OSGi
  73  * environment. Hence a single instance of the service will be active (via Service Registry in OSGi)
  74  * and a Singleton object in a non-OSGi environment.
  75  */
  76 public class OvsdbConnectionService implements OvsdbConnection {
  77     private static final Logger logger = LoggerFactory.getLogger(OvsdbConnectionService.class);
  78     private static final int NUM_THREADS = 3;
  79
  80     // Singleton Service object that can be used in Non-OSGi environment
  81     private static Set<OvsdbConnectionListener> connectionListeners = Sets.newHashSet();
  82     private static Map<OvsdbClient, Channel> connections = Maps.newHashMap();
  83     private static OvsdbConnection connectionService;
  84     private static volatile boolean singletonCreated = false;
  85
  86     public static OvsdbConnection getService() {
  87         if (connectionService == null) {
  88             connectionService = new OvsdbConnectionService();
  89         }
  90         return connectionService;
  91     }
  92     @Override
  93     public OvsdbClient connect(final InetAddress address, final int port) {
  94         return connectWithSsl(address, port, null /* SslContext */);
  95     }
  96     @Override
  97     public OvsdbClient connectWithSsl(final InetAddress address, final int port,
  98                                final SSLContext sslContext) {
  99         try {
 100             Bootstrap bootstrap = new Bootstrap();
 101             bootstrap.group(new NioEventLoopGroup());
 102             bootstrap.channel(NioSocketChannel.class);
 103             bootstrap.option(ChannelOption.TCP_NODELAY, true);
 104             bootstrap.option(ChannelOption.RCVBUF_ALLOCATOR, new AdaptiveRecvByteBufAllocator(65535, 65535, 65535));
 105
 106             bootstrap.handler(new ChannelInitializer<SocketChannel>() {
 107                 @Override
 108                 public void initChannel(SocketChannel channel) throws Exception {
 109                     if (sslContext != null) {
 110                         /* First add ssl handler if ssl context is given */
 111                         SSLEngine engine =
 112                             sslContext.createSSLEngine(address.toString(), port);
 113                         engine.setUseClientMode(true);
 114                         channel.pipeline().addLast("ssl", new SslHandler(engine));
 115                     }
 116                     channel.pipeline().addLast(
 117                             //new LoggingHandler(LogLevel.INFO),
 118                             new JsonRpcDecoder(100000),
 119                             new StringEncoder(CharsetUtil.UTF_8),
 120                             new ExceptionHandler());
 121                 }
 122             });
 123
 124             ChannelFuture future = bootstrap.connect(address, port).sync();
 125             Channel channel = future.channel();
 126             OvsdbClient client = getChannelClient(channel, ConnectionType.ACTIVE,
 127                     Executors.newFixedThreadPool(NUM_THREADS));
 128             return client;
 129         } catch (InterruptedException e) {
 130             logger.warn("Thread was interrupted during connect", e);
 131         } catch (Exception e) {
 132             logger.warn("bootstrap.connect failed", e);
 133         }
 134         return null;
 135     }
 136
 137     @Override
 138     public void disconnect(OvsdbClient client) {
 139         if (client == null) {
 140             return;
 141         }
 142         Channel channel = connections.get(client);
 143         if (channel != null) {
 144             channel.disconnect();
 145         }
 146         connections.remove(client);
 147     }
 148
 149     @Override
 150     public void registerConnectionListener(OvsdbConnectionListener listener) {
 151         connectionListeners.add(listener);
 152     }
 153
 154     @Override
 155     public void unregisterConnectionListener(OvsdbConnectionListener listener) {
 156         connectionListeners.remove(listener);
 157     }
 158
 159     private static OvsdbClient getChannelClient(Channel channel, ConnectionType type,
 160                                                 ExecutorService executorService) {
 161         ObjectMapper objectMapper = new ObjectMapper();
 162         objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
 163         objectMapper.setSerializationInclusion(Include.NON_NULL);
 164
 165         JsonRpcEndpoint factory = new JsonRpcEndpoint(objectMapper, channel);
 166         JsonRpcServiceBinderHandler binderHandler = new JsonRpcServiceBinderHandler(factory);
 167         binderHandler.setContext(channel);
 168         channel.pipeline().addLast(binderHandler);
 169
 170         OvsdbRPC rpc = factory.getClient(channel, OvsdbRPC.class);
 171         OvsdbClientImpl client = new OvsdbClientImpl(rpc, channel, type, executorService);
 172         connections.put(client, channel);
 173         ChannelFuture closeFuture = channel.closeFuture();
 174         closeFuture.addListener(new ChannelConnectionHandler(client));
 175         return client;
 176     }
 177
 178     /**
 179      * Method that initiates the Passive OVSDB channel listening functionality.
 180      * By default the ovsdb passive connection will listen in port 6640 which can
 181      * be overridden using the ovsdb.listenPort system property.
 182      */
 183     @Override
 184     public synchronized boolean startOvsdbManager(final int ovsdbListenPort) {
 185         if (!singletonCreated) {
 186             new Thread() {
 187                 @Override
 188                 public void run() {
 189                     ovsdbManager(ovsdbListenPort);
 190                 }
 191             }.start();
 192             singletonCreated = true;
 193             return true;
 194         } else {
 195             return false;
 196         }
 197     }
 198
 199     /**
 200      * Method that initiates the Passive OVSDB channel listening functionality
 201      * with ssl.By default the ovsdb passive connection will listen in port
 202      * 6640 which can be overridden using the ovsdb.listenPort system property.
 203      */
 204     @Override
 205     public synchronized boolean startOvsdbManagerWithSsl(final int ovsdbListenPort,
 206                                      final SSLContext sslContext) {
 207         if (!singletonCreated) {
 208             new Thread() {
 209                 @Override
 210                 public void run() {
 211                     ovsdbManagerWithSsl(ovsdbListenPort, sslContext);
 212                 }
 213             }.start();
 214             singletonCreated = true;
 215             return true;
 216         } else {
 217             return false;
 218         }
 219     }
 220
 221     /**
 222      * OVSDB Passive listening thread that uses Netty ServerBootstrap to open
 223      * passive connection handle channel callbacks.
 224      */
 225     private static void ovsdbManager(int port) {
 226         ovsdbManagerWithSsl(port, null /* SslContext */);
 227     }
 228
 229     /**
 230      * OVSDB Passive listening thread that uses Netty ServerBootstrap to open
 231      * passive connection with Ssl and handle channel callbacks.
 232      */
 233     private static void ovsdbManagerWithSsl(int port, final SSLContext sslContext) {
 234         EventLoopGroup bossGroup = new NioEventLoopGroup();
 235         EventLoopGroup workerGroup = new NioEventLoopGroup();
 236         try {
 237             ServerBootstrap serverBootstrap = new ServerBootstrap();
 238             serverBootstrap.group(bossGroup, workerGroup)
 239                     .channel(NioServerSocketChannel.class)
 240                     .option(ChannelOption.SO_BACKLOG, 100)
 241                     .handler(new LoggingHandler(LogLevel.INFO))
 242                     .childHandler(new ChannelInitializer<SocketChannel>() {
 243                         @Override
 244                         public void initChannel(SocketChannel channel) throws Exception {
 245                             logger.debug("New Passive channel created : {}", channel);
 246                             if (sslContext != null) {
 247                                 /* Add SSL handler first if SSL context is provided */
 248                                 SSLEngine engine = sslContext.createSSLEngine();
 249                                 engine.setUseClientMode(false); // work in a server mode
 250                                 engine.setNeedClientAuth(true); // need client authentication
 251                                 //Disable SSLv3 and enable all other supported protocols
 252                                 String[] protocols = {"SSLv2Hello", "TLSv1", "TLSv1.1", "TLSv1.2"};
 253                                 logger.debug("Set enable protocols {}", Arrays.toString(protocols));
 254                                 engine.setEnabledProtocols(protocols);
 255                                 logger.debug("Supported ssl protocols {}",
 256                                         Arrays.toString(engine.getSupportedProtocols()));
 257                                 logger.debug("Enabled ssl protocols {}",
 258                                         Arrays.toString(engine.getEnabledProtocols()));
 259                                 //Set cipher suites
 260                                 String[] cipherSuites = {"TLS_RSA_WITH_AES_128_CBC_SHA"};
 261                                 logger.debug("Set enable cipher cuites {}", Arrays.toString(cipherSuites));
 262                                 engine.setEnabledCipherSuites(cipherSuites);
 263                                 logger.debug("Enabled cipher suites {}",
 264                                         Arrays.toString(engine.getEnabledCipherSuites()));
 265                                 channel.pipeline().addLast("ssl", new SslHandler(engine));
 266                             }
 267
 268                             channel.pipeline().addLast(
 269                                  new JsonRpcDecoder(100000),
 270                                  new StringEncoder(CharsetUtil.UTF_8),
 271                                  new ExceptionHandler());
 272
 273                             handleNewPassiveConnection(channel);
 274                         }
 275                     });
 276             serverBootstrap.option(ChannelOption.TCP_NODELAY, true);
 277             serverBootstrap.option(ChannelOption.RCVBUF_ALLOCATOR,
 278                     new AdaptiveRecvByteBufAllocator(65535, 65535, 65535));
 279             // Start the server.
 280             ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
 281             Channel serverListenChannel = channelFuture.channel();
 282             // Wait until the server socket is closed.
 283             serverListenChannel.closeFuture().sync();
 284         } catch (InterruptedException e) {
 285             logger.error("Thread interrupted", e);
 286         } finally {
 287             // Shut down all event loops to terminate all threads.
 288             bossGroup.shutdownGracefully();
 289             workerGroup.shutdownGracefully();
 290         }
 291     }
 292
 293     private static ExecutorService executorService = Executors.newFixedThreadPool(10);
 294     private static void handleNewPassiveConnection(final Channel channel) {
 295         executorService.execute(new Runnable() {
 296             @Override
 297             public void run() {
 298                 OvsdbClient client = getChannelClient(channel, ConnectionType.PASSIVE,
 299                         Executors.newFixedThreadPool(NUM_THREADS));
 300
 301                 SslHandler sslHandler = (SslHandler) channel.pipeline().get("ssl");
 302                 if (sslHandler != null) {
 303                     //Wait until ssl handshake is complete
 304                     int count = 0;
 305                     logger.debug("Check if ssl handshake is done");
 306                     while (sslHandler.engine().getSession().getCipherSuite()
 307                                             .equals("SSL_NULL_WITH_NULL_NULL")
 308                                             && count < 10) {
 309                         try {
 310                             Thread.sleep(100);
 311                         } catch (InterruptedException e) {
 312                             logger.error("Exception while checking if ssl handshake is done", e);
 313                         }
 314                         count++;
 315                     }
 316                     if (sslHandler.engine().getSession().getCipherSuite()
 317                                            .equals("SSL_NULL_WITH_NULL_NULL")) {
 318                         logger.debug("Ssl hanshake is not compelete yet");
 319                         return;
 320                     }
 321                 }
 322                 logger.debug("Notify listener");
 323                 for (OvsdbConnectionListener listener : connectionListeners) {
 324                     listener.connected(client);
 325                 }
 326             }
 327         });
 328     }
 329
 330     public static void channelClosed(final OvsdbClient client) {
 331         logger.info("Connection closed {}", client.getConnectionInfo().toString());
 332         connections.remove(client);
 333         for (OvsdbConnectionListener listener : connectionListeners) {
 334             listener.disconnected(client);
 335         }
 336     }
 337     @Override
 338     public Collection<OvsdbClient> getConnections() {
 339         return connections.keySet();
 340     }
 341 }