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

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