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