2 * Copyright (c) 2014, 2015 Red Hat, Inc. and others. All rights reserved.
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
9 package org.opendaylight.ovsdb.lib.impl;
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.FutureCallback;
17 import com.google.common.util.concurrent.Futures;
18 import com.google.common.util.concurrent.ListenableFuture;
19 import com.google.common.util.concurrent.ThreadFactoryBuilder;
20 import io.netty.bootstrap.Bootstrap;
21 import io.netty.bootstrap.ServerBootstrap;
22 import io.netty.channel.AdaptiveRecvByteBufAllocator;
23 import io.netty.channel.Channel;
24 import io.netty.channel.ChannelFuture;
25 import io.netty.channel.ChannelInitializer;
26 import io.netty.channel.ChannelOption;
27 import io.netty.channel.EventLoopGroup;
28 import io.netty.channel.nio.NioEventLoopGroup;
29 import io.netty.channel.socket.SocketChannel;
30 import io.netty.channel.socket.nio.NioServerSocketChannel;
31 import io.netty.channel.socket.nio.NioSocketChannel;
32 import io.netty.handler.codec.string.StringEncoder;
33 import io.netty.handler.logging.LogLevel;
34 import io.netty.handler.logging.LoggingHandler;
35 import io.netty.handler.ssl.SslHandler;
36 import io.netty.handler.timeout.IdleStateHandler;
37 import io.netty.handler.timeout.ReadTimeoutHandler;
38 import io.netty.util.CharsetUtil;
39 import java.net.InetAddress;
40 import java.util.ArrayList;
41 import java.util.Arrays;
42 import java.util.Collection;
43 import java.util.List;
46 import java.util.concurrent.ExecutorService;
47 import java.util.concurrent.Executors;
48 import java.util.concurrent.ScheduledExecutorService;
49 import java.util.concurrent.ThreadFactory;
50 import java.util.concurrent.TimeUnit;
51 import javax.annotation.Nullable;
52 import javax.net.ssl.SSLContext;
53 import javax.net.ssl.SSLEngine;
54 import javax.net.ssl.SSLEngineResult.HandshakeStatus;
55 import javax.net.ssl.SSLPeerUnverifiedException;
56 import org.opendaylight.ovsdb.lib.OvsdbClient;
57 import org.opendaylight.ovsdb.lib.OvsdbConnection;
58 import org.opendaylight.ovsdb.lib.OvsdbConnectionInfo.ConnectionType;
59 import org.opendaylight.ovsdb.lib.OvsdbConnectionInfo.SocketConnectionType;
60 import org.opendaylight.ovsdb.lib.OvsdbConnectionListener;
61 import org.opendaylight.ovsdb.lib.jsonrpc.ExceptionHandler;
62 import org.opendaylight.ovsdb.lib.jsonrpc.JsonRpcDecoder;
63 import org.opendaylight.ovsdb.lib.jsonrpc.JsonRpcEndpoint;
64 import org.opendaylight.ovsdb.lib.jsonrpc.JsonRpcServiceBinderHandler;
65 import org.opendaylight.ovsdb.lib.message.OvsdbRPC;
66 import org.slf4j.Logger;
67 import org.slf4j.LoggerFactory;
70 * OvsDBConnectionService provides OVSDB connection management functionality which includes
71 * both Active and Passive connections.
72 * From the Library perspective, Active OVSDB connections are those that are initiated from
73 * the Controller towards the ovsdb-manager.
74 * While Passive OVSDB connections are those that are initiated from the ovs towards
77 * <p>Applications that use OvsDBConnectionService can use the OvsDBConnection class' connect APIs
78 * to initiate Active connections and can listen to the asynchronous Passive connections via
79 * registerConnectionListener listener API.
81 * <p>The library is designed as Java modular component that can work in both OSGi and non-OSGi
82 * environment. Hence a single instance of the service will be active (via Service Registry in OSGi)
83 * and a Singleton object in a non-OSGi environment.
85 public class OvsdbConnectionService implements AutoCloseable, OvsdbConnection {
86 private static final Logger LOG = LoggerFactory.getLogger(OvsdbConnectionService.class);
88 private static ThreadFactory passiveConnectionThreadFactory = new ThreadFactoryBuilder()
89 .setNameFormat("OVSDBPassiveConnServ-%d").build();
90 private static ScheduledExecutorService executorService
91 = Executors.newScheduledThreadPool(10, passiveConnectionThreadFactory);
93 private static ThreadFactory connectionNotifierThreadFactory = new ThreadFactoryBuilder()
94 .setNameFormat("OVSDBConnNotifSer-%d").build();
95 private static ExecutorService connectionNotifierService
96 = Executors.newCachedThreadPool(connectionNotifierThreadFactory);
98 private static Set<OvsdbConnectionListener> connectionListeners = Sets.newHashSet();
99 private static Map<OvsdbClient, Channel> connections = Maps.newHashMap();
100 private static OvsdbConnection connectionService;
101 private static volatile boolean singletonCreated = false;
102 private static final int IDLE_READER_TIMEOUT = 30;
103 private static final int READ_TIMEOUT = 180;
104 private static final String OVSDB_RPC_TASK_TIMEOUT_PARAM = "ovsdb-rpc-task-timeout";
106 private static final StalePassiveConnectionService STALE_PASSIVE_CONNECTION_SERVICE =
107 new StalePassiveConnectionService(executorService);
109 private static int retryPeriod = 100; // retry after 100 milliseconds
112 public static OvsdbConnection getService() {
113 if (connectionService == null) {
114 connectionService = new OvsdbConnectionService();
116 return connectionService;
120 public OvsdbClient connect(final InetAddress address, final int port) {
121 return connectWithSsl(address, port, null /* SslContext */);
125 public OvsdbClient connectWithSsl(final InetAddress address, final int port,
126 final SSLContext sslContext) {
128 Bootstrap bootstrap = new Bootstrap();
129 bootstrap.group(new NioEventLoopGroup());
130 bootstrap.channel(NioSocketChannel.class);
131 bootstrap.option(ChannelOption.TCP_NODELAY, true);
132 bootstrap.option(ChannelOption.RCVBUF_ALLOCATOR, new AdaptiveRecvByteBufAllocator(65535, 65535, 65535));
134 bootstrap.handler(new ChannelInitializer<SocketChannel>() {
136 public void initChannel(SocketChannel channel) throws Exception {
137 if (sslContext != null) {
138 /* First add ssl handler if ssl context is given */
140 sslContext.createSSLEngine(address.toString(), port);
141 engine.setUseClientMode(true);
142 channel.pipeline().addLast("ssl", new SslHandler(engine));
144 channel.pipeline().addLast(
145 //new LoggingHandler(LogLevel.INFO),
146 new JsonRpcDecoder(100000),
147 new StringEncoder(CharsetUtil.UTF_8),
148 new IdleStateHandler(IDLE_READER_TIMEOUT, 0, 0),
149 new ReadTimeoutHandler(READ_TIMEOUT),
150 new ExceptionHandler());
154 ChannelFuture future = bootstrap.connect(address, port).sync();
155 Channel channel = future.channel();
156 return getChannelClient(channel, ConnectionType.ACTIVE, SocketConnectionType.SSL);
157 } catch (InterruptedException e) {
158 LOG.warn("Failed to connect {}:{}", address, port, e);
164 public void disconnect(OvsdbClient client) {
165 if (client == null) {
168 Channel channel = connections.get(client);
169 if (channel != null) {
170 channel.disconnect();
172 connections.remove(client);
176 public void registerConnectionListener(OvsdbConnectionListener listener) {
177 LOG.info("registerConnectionListener: registering {}", listener.getClass().getSimpleName());
178 connectionListeners.add(listener);
182 public void unregisterConnectionListener(OvsdbConnectionListener listener) {
183 connectionListeners.remove(listener);
186 private static OvsdbClient getChannelClient(Channel channel, ConnectionType type,
187 SocketConnectionType socketConnType) {
188 ObjectMapper objectMapper = new ObjectMapper();
189 objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
190 objectMapper.setSerializationInclusion(Include.NON_NULL);
192 JsonRpcEndpoint factory = new JsonRpcEndpoint(objectMapper, channel);
193 JsonRpcServiceBinderHandler binderHandler = new JsonRpcServiceBinderHandler(factory);
194 binderHandler.setContext(channel);
195 channel.pipeline().addLast(binderHandler);
197 OvsdbRPC rpc = factory.getClient(channel, OvsdbRPC.class);
198 OvsdbClientImpl client = new OvsdbClientImpl(rpc, channel, type, socketConnType);
199 client.setConnectionPublished(true);
200 connections.put(client, channel);
201 ChannelFuture closeFuture = channel.closeFuture();
202 closeFuture.addListener(new ChannelConnectionHandler(client));
207 * Method that initiates the Passive OVSDB channel listening functionality.
208 * By default the ovsdb passive connection will listen in port 6640 which can
209 * be overridden using the ovsdb.listenPort system property.
212 public synchronized boolean startOvsdbManager(final int ovsdbListenPort) {
213 if (!singletonCreated) {
214 LOG.info("startOvsdbManager: Starting");
218 ovsdbManager(ovsdbListenPort);
221 singletonCreated = true;
229 * Method that initiates the Passive OVSDB channel listening functionality
230 * with ssl.By default the ovsdb passive connection will listen in port
231 * 6640 which can be overridden using the ovsdb.listenPort system property.
234 public synchronized boolean startOvsdbManagerWithSsl(final int ovsdbListenPort,
235 final SSLContext sslContext) {
236 if (!singletonCreated) {
240 ovsdbManagerWithSsl(ovsdbListenPort, sslContext);
243 singletonCreated = true;
251 * OVSDB Passive listening thread that uses Netty ServerBootstrap to open
252 * passive connection handle channel callbacks.
254 private static void ovsdbManager(int port) {
255 ovsdbManagerWithSsl(port, null /* SslContext */);
259 * OVSDB Passive listening thread that uses Netty ServerBootstrap to open
260 * passive connection with Ssl and handle channel callbacks.
262 private static void ovsdbManagerWithSsl(int port, final SSLContext sslContext) {
263 EventLoopGroup bossGroup = new NioEventLoopGroup();
264 EventLoopGroup workerGroup = new NioEventLoopGroup();
266 ServerBootstrap serverBootstrap = new ServerBootstrap();
267 serverBootstrap.group(bossGroup, workerGroup)
268 .channel(NioServerSocketChannel.class)
269 .option(ChannelOption.SO_BACKLOG, 100)
270 .handler(new LoggingHandler(LogLevel.INFO))
271 .childHandler(new ChannelInitializer<SocketChannel>() {
273 public void initChannel(SocketChannel channel) throws Exception {
274 LOG.debug("New Passive channel created : {}", channel);
275 if (sslContext != null) {
276 /* Add SSL handler first if SSL context is provided */
277 SSLEngine engine = sslContext.createSSLEngine();
278 engine.setUseClientMode(false); // work in a server mode
279 engine.setNeedClientAuth(true); // need client authentication
280 //Disable SSLv3, TLSv1 and enable all other supported protocols
281 String[] protocols = {"SSLv2Hello", "TLSv1.1", "TLSv1.2"};
282 LOG.debug("Set enable protocols {}", Arrays.toString(protocols));
283 engine.setEnabledProtocols(protocols);
284 LOG.debug("Supported ssl protocols {}",
285 Arrays.toString(engine.getSupportedProtocols()));
286 LOG.debug("Enabled ssl protocols {}",
287 Arrays.toString(engine.getEnabledProtocols()));
289 String[] cipherSuites = {"TLS_ECDH_RSA_WITH_AES_128_CBC_SHA256",
290 "TLS_ECDH_ECDSA_WITH_AES_128_CBC_SHA256",
291 "TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA256",
292 "TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA256",
293 "TLS_RSA_WITH_AES_128_CBC_SHA256"};
294 engine.setEnabledCipherSuites(cipherSuites);
295 LOG.debug("Enabled cipher suites {}",
296 Arrays.toString(engine.getEnabledCipherSuites()));
297 channel.pipeline().addLast("ssl", new SslHandler(engine));
300 channel.pipeline().addLast(
301 new JsonRpcDecoder(100000),
302 new StringEncoder(CharsetUtil.UTF_8),
303 new IdleStateHandler(IDLE_READER_TIMEOUT, 0, 0),
304 new ReadTimeoutHandler(READ_TIMEOUT),
305 new ExceptionHandler());
307 handleNewPassiveConnection(channel);
310 serverBootstrap.option(ChannelOption.TCP_NODELAY, true);
311 serverBootstrap.option(ChannelOption.RCVBUF_ALLOCATOR,
312 new AdaptiveRecvByteBufAllocator(65535, 65535, 65535));
314 ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
315 Channel serverListenChannel = channelFuture.channel();
316 // Wait until the server socket is closed.
317 serverListenChannel.closeFuture().sync();
318 } catch (InterruptedException e) {
319 LOG.error("Thread interrupted", e);
321 // Shut down all event loops to terminate all threads.
322 bossGroup.shutdownGracefully();
323 workerGroup.shutdownGracefully();
327 private static void handleNewPassiveConnection(OvsdbClient client) {
328 ListenableFuture<List<String>> echoFuture = client.echo();
329 LOG.debug("Send echo message to probe the OVSDB switch {}",client.getConnectionInfo());
330 Futures.addCallback(echoFuture, new FutureCallback<List<String>>() {
332 public void onSuccess(@Nullable List<String> result) {
333 LOG.debug("Probe was successful to OVSDB switch {}",client.getConnectionInfo());
334 List<OvsdbClient> clientsFromSameNode = getPassiveClientsFromSameNode(client);
335 if (clientsFromSameNode.size() == 0) {
336 notifyListenerForPassiveConnection(client);
338 STALE_PASSIVE_CONNECTION_SERVICE.handleNewPassiveConnection(client, clientsFromSameNode);
343 public void onFailure(Throwable failureException) {
344 LOG.error("Probe failed to OVSDB switch. Disconnecting the channel {}", client.getConnectionInfo());
347 }, connectionNotifierService);
350 private static void handleNewPassiveConnection(final Channel channel) {
351 if (!channel.isOpen()) {
352 LOG.warn("Channel {} is not open, skipped further processing of the connection.",channel);
355 SslHandler sslHandler = (SslHandler) channel.pipeline().get("ssl");
356 if (sslHandler != null) {
357 class HandleNewPassiveSslRunner implements Runnable {
358 public SslHandler sslHandler;
359 public final Channel channel;
360 private int retryTimes;
362 HandleNewPassiveSslRunner(Channel channel, SslHandler sslHandler) {
363 this.channel = channel;
364 this.sslHandler = sslHandler;
370 HandshakeStatus status = sslHandler.engine().getHandshakeStatus();
371 LOG.debug("Handshake status {}", status);
374 case NOT_HANDSHAKING:
375 if (sslHandler.engine().getSession().getCipherSuite()
376 .equals("SSL_NULL_WITH_NULL_NULL")) {
377 // Not begin handshake yet. Retry later.
378 LOG.debug("handshake not begin yet {}", status);
379 executorService.schedule(this, retryPeriod, TimeUnit.MILLISECONDS);
381 //Check if peer is trusted before notifying listeners
383 sslHandler.engine().getSession().getPeerCertificates();
384 //Handshake done. Notify listener.
385 OvsdbClient client = getChannelClient(channel, ConnectionType.PASSIVE,
386 SocketConnectionType.SSL);
387 handleNewPassiveConnection(client);
388 } catch (SSLPeerUnverifiedException e) {
389 //Trust manager is still checking peer certificate. Retry later
390 LOG.debug("Peer certifiacte is not verified yet {}", status);
391 executorService.schedule(this, retryPeriod, TimeUnit.MILLISECONDS);
398 //Handshake still ongoing. Retry later.
399 LOG.debug("handshake not done yet {}", status);
400 executorService.schedule(this, retryPeriod, TimeUnit.MILLISECONDS);
404 if (sslHandler.engine().getSession().getCipherSuite()
405 .equals("SSL_NULL_WITH_NULL_NULL")) {
406 /* peer not authenticated. No need to notify listener in this case. */
407 LOG.error("Ssl handshake fail. channel {}", channel);
410 * peer is authenticated. Give some time to wait for completion.
411 * If status is still NEED_WRAP, client might already disconnect.
412 * This happens when the first time client connects to controller in two-way handshake.
413 * After obtaining controller certificate, client will disconnect and start
414 * new connection with controller certificate it obtained.
415 * In this case no need to do anything for the first connection attempt. Just skip
416 * since client will reconnect later.
418 LOG.debug("handshake not done yet {}", status);
419 if (retryTimes > 0) {
420 executorService.schedule(this, retryPeriod, TimeUnit.MILLISECONDS);
422 LOG.debug("channel closed {}", channel);
429 LOG.error("unknown hadshake status {}", status);
434 executorService.schedule(new HandleNewPassiveSslRunner(channel, sslHandler),
435 retryPeriod, TimeUnit.MILLISECONDS);
437 executorService.execute(new Runnable() {
440 OvsdbClient client = getChannelClient(channel, ConnectionType.PASSIVE,
441 SocketConnectionType.NON_SSL);
442 handleNewPassiveConnection(client);
448 public static void channelClosed(final OvsdbClient client) {
449 LOG.info("Connection closed {}", client.getConnectionInfo().toString());
450 connections.remove(client);
451 if (client.isConnectionPublished()) {
452 for (OvsdbConnectionListener listener : connectionListeners) {
453 listener.disconnected(client);
456 STALE_PASSIVE_CONNECTION_SERVICE.clientDisconnected(client);
460 public Collection<OvsdbClient> getConnections() {
461 return connections.keySet();
465 public void close() throws Exception {
466 LOG.info("OvsdbConnectionService closed");
467 JsonRpcEndpoint.close();
471 public OvsdbClient getClient(Channel channel) {
472 for (OvsdbClient client : connections.keySet()) {
473 Channel ctx = connections.get(client);
474 if (ctx.equals(channel)) {
481 private static List<OvsdbClient> getPassiveClientsFromSameNode(OvsdbClient ovsdbClient) {
482 List<OvsdbClient> passiveClients = new ArrayList<>();
483 for (OvsdbClient client : connections.keySet()) {
484 if (!client.equals(ovsdbClient)
485 && client.getConnectionInfo().getRemoteAddress()
486 .equals(ovsdbClient.getConnectionInfo().getRemoteAddress())
487 && client.getConnectionInfo().getType() == ConnectionType.PASSIVE) {
488 passiveClients.add(client);
491 return passiveClients;
494 public static void notifyListenerForPassiveConnection(final OvsdbClient client) {
495 client.setConnectionPublished(true);
496 for (final OvsdbConnectionListener listener : connectionListeners) {
497 connectionNotifierService.submit(new Runnable() {
500 LOG.trace("Connection {} notified to listener {}", client.getConnectionInfo(), listener);
501 listener.connected(client);
507 public void setOvsdbRpcTaskTimeout(int timeout) {
508 JsonRpcEndpoint.setReaperInterval(timeout);
511 public void updateConfigParameter(Map<String, Object> configParameters) {
512 LOG.debug("Config parameters received : {}", configParameters.entrySet());
513 if (configParameters != null && !configParameters.isEmpty()) {
514 for (Map.Entry<String, Object> paramEntry : configParameters.entrySet()) {
515 if (paramEntry.getKey().equalsIgnoreCase(OVSDB_RPC_TASK_TIMEOUT_PARAM)) {
516 setOvsdbRpcTaskTimeout(Integer.parseInt((String)paramEntry.getValue()));
518 //Please remove the break if you add more config nobs.