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