ef2af7530e7a5ec32d49d4efdb5f588b4326c5ac
[ovsdb.git] / 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 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         connectionListeners.add(listener);
153     }
154
155     @Override
156     public void unregisterConnectionListener(OvsdbConnectionListener listener) {
157         connectionListeners.remove(listener);
158     }
159
160     private static OvsdbClient getChannelClient(Channel channel, ConnectionType type,
161                                                 ExecutorService executorService) {
162         ObjectMapper objectMapper = new ObjectMapper();
163         objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
164         objectMapper.setSerializationInclusion(Include.NON_NULL);
165
166         JsonRpcEndpoint factory = new JsonRpcEndpoint(objectMapper, channel);
167         JsonRpcServiceBinderHandler binderHandler = new JsonRpcServiceBinderHandler(factory);
168         binderHandler.setContext(channel);
169         channel.pipeline().addLast(binderHandler);
170
171         OvsdbRPC rpc = factory.getClient(channel, OvsdbRPC.class);
172         OvsdbClientImpl client = new OvsdbClientImpl(rpc, channel, type, executorService);
173         connections.put(client, channel);
174         ChannelFuture closeFuture = channel.closeFuture();
175         closeFuture.addListener(new ChannelConnectionHandler(client));
176         return client;
177     }
178
179     /**
180      * Method that initiates the Passive OVSDB channel listening functionality.
181      * By default the ovsdb passive connection will listen in port 6640 which can
182      * be overridden using the ovsdb.listenPort system property.
183      */
184     @Override
185     public synchronized boolean startOvsdbManager(final int ovsdbListenPort) {
186         if (!singletonCreated) {
187             new Thread() {
188                 @Override
189                 public void run() {
190                     ovsdbManager(ovsdbListenPort);
191                 }
192             }.start();
193             singletonCreated = true;
194             return true;
195         } else {
196             return false;
197         }
198     }
199
200     /**
201      * Method that initiates the Passive OVSDB channel listening functionality
202      * with ssl.By default the ovsdb passive connection will listen in port
203      * 6640 which can be overridden using the ovsdb.listenPort system property.
204      */
205     @Override
206     public synchronized boolean startOvsdbManagerWithSsl(final int ovsdbListenPort,
207                                      final SSLContext sslContext) {
208         if (!singletonCreated) {
209             new Thread() {
210                 @Override
211                 public void run() {
212                     ovsdbManagerWithSsl(ovsdbListenPort, sslContext);
213                 }
214             }.start();
215             singletonCreated = true;
216             return true;
217         } else {
218             return false;
219         }
220     }
221
222     /**
223      * OVSDB Passive listening thread that uses Netty ServerBootstrap to open
224      * passive connection handle channel callbacks.
225      */
226     private static void ovsdbManager(int port) {
227         ovsdbManagerWithSsl(port, null /* SslContext */);
228     }
229
230     /**
231      * OVSDB Passive listening thread that uses Netty ServerBootstrap to open
232      * passive connection with Ssl and handle channel callbacks.
233      */
234     private static void ovsdbManagerWithSsl(int port, final SSLContext sslContext) {
235         EventLoopGroup bossGroup = new NioEventLoopGroup();
236         EventLoopGroup workerGroup = new NioEventLoopGroup();
237         try {
238             ServerBootstrap serverBootstrap = new ServerBootstrap();
239             serverBootstrap.group(bossGroup, workerGroup)
240                     .channel(NioServerSocketChannel.class)
241                     .option(ChannelOption.SO_BACKLOG, 100)
242                     .handler(new LoggingHandler(LogLevel.INFO))
243                     .childHandler(new ChannelInitializer<SocketChannel>() {
244                         @Override
245                         public void initChannel(SocketChannel channel) throws Exception {
246                             LOG.debug("New Passive channel created : {}", channel);
247                             if (sslContext != null) {
248                                 /* Add SSL handler first if SSL context is provided */
249                                 SSLEngine engine = sslContext.createSSLEngine();
250                                 engine.setUseClientMode(false); // work in a server mode
251                                 engine.setNeedClientAuth(true); // need client authentication
252                                 //Disable SSLv3 and enable all other supported protocols
253                                 String[] protocols = {"SSLv2Hello", "TLSv1", "TLSv1.1", "TLSv1.2"};
254                                 LOG.debug("Set enable protocols {}", Arrays.toString(protocols));
255                                 engine.setEnabledProtocols(protocols);
256                                 LOG.debug("Supported ssl protocols {}",
257                                         Arrays.toString(engine.getSupportedProtocols()));
258                                 LOG.debug("Enabled ssl protocols {}",
259                                         Arrays.toString(engine.getEnabledProtocols()));
260                                 //Set cipher suites
261                                 String[] cipherSuites = {"TLS_RSA_WITH_AES_128_CBC_SHA"};
262                                 LOG.debug("Set enable cipher cuites {}", Arrays.toString(cipherSuites));
263                                 engine.setEnabledCipherSuites(cipherSuites);
264                                 LOG.debug("Enabled cipher suites {}",
265                                         Arrays.toString(engine.getEnabledCipherSuites()));
266                                 channel.pipeline().addLast("ssl", new SslHandler(engine));
267                             }
268
269                             channel.pipeline().addLast(
270                                  new JsonRpcDecoder(100000),
271                                  new StringEncoder(CharsetUtil.UTF_8),
272                                  new ExceptionHandler());
273
274                             handleNewPassiveConnection(channel);
275                         }
276                     });
277             serverBootstrap.option(ChannelOption.TCP_NODELAY, true);
278             serverBootstrap.option(ChannelOption.RCVBUF_ALLOCATOR,
279                     new AdaptiveRecvByteBufAllocator(65535, 65535, 65535));
280             // Start the server.
281             ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
282             Channel serverListenChannel = channelFuture.channel();
283             // Wait until the server socket is closed.
284             serverListenChannel.closeFuture().sync();
285         } catch (InterruptedException e) {
286             LOG.error("Thread interrupted", e);
287         } finally {
288             // Shut down all event loops to terminate all threads.
289             bossGroup.shutdownGracefully();
290             workerGroup.shutdownGracefully();
291         }
292     }
293
294     private static ScheduledExecutorService executorService = Executors.newScheduledThreadPool(10);
295     private static int retryPeriod = 100; // retry after 100 milliseconds
296     private static void handleNewPassiveConnection(final Channel channel) {
297         SslHandler sslHandler = (SslHandler) channel.pipeline().get("ssl");
298         if (sslHandler != null) {
299             class HandleNewPassiveSslRunner implements Runnable {
300                 public SslHandler sslHandler;
301                 public final Channel channel;
302                 private int retryTimes;
303
304                 public HandleNewPassiveSslRunner(Channel channel, SslHandler sslHandler) {
305                     this.channel = channel;
306                     this.sslHandler = sslHandler;
307                     this.retryTimes = 3;
308                 }
309                 @Override
310                 public void run() {
311                     HandshakeStatus status = sslHandler.engine().getHandshakeStatus();
312                     LOG.debug("Handshake status {}", status);
313                     switch (status) {
314                         case FINISHED:
315                         case NOT_HANDSHAKING:
316                             //Handshake done. Notify listener.
317                             OvsdbClient client = getChannelClient(channel, ConnectionType.PASSIVE,
318                                                  Executors.newFixedThreadPool(NUM_THREADS));
319
320                             LOG.debug("Notify listener");
321                             for (OvsdbConnectionListener listener : connectionListeners) {
322                                 listener.connected(client);
323                             }
324                             break;
325
326                         case NEED_UNWRAP:
327                         case NEED_TASK:
328                             //Handshake still ongoing. Retry later.
329                             LOG.debug("handshake not done yet {}", status);
330                             executorService.schedule(this,  retryPeriod, TimeUnit.MILLISECONDS);
331                             break;
332
333                         case NEED_WRAP:
334                             if (sslHandler.engine().getSession().getCipherSuite()
335                                     .equals("SSL_NULL_WITH_NULL_NULL")) {
336                                 /* peer not authenticated. No need to notify listener in this case. */
337                                 LOG.error("Ssl handshake fail. channel {}", channel);
338                             } else {
339                                 /*
340                                  * peer is authenticated. Give some time to wait for completion.
341                                  * If status is still NEED_WRAP, client might already disconnect.
342                                  * This happens when the first time client connects to controller in two-way handshake.
343                                  * After obtaining controller certificate, client will disconnect and start
344                                  * new connection with controller certificate it obtained.
345                                  * In this case no need to do anything for the first connection attempt. Just skip
346                                  * since client will reconnect later.
347                                  */
348                                 LOG.debug("handshake not done yet {}", status);
349                                 if (retryTimes > 0) {
350                                     executorService.schedule(this,  retryPeriod, TimeUnit.MILLISECONDS);
351                                 } else {
352                                     LOG.debug("channel closed {}", channel);
353                                 }
354                                 retryTimes--;
355                             }
356                             break;
357
358                         default:
359                             LOG.error("unknown hadshake status {}", status);
360                     }
361                 }
362             }
363             executorService.schedule(new HandleNewPassiveSslRunner(channel, sslHandler),
364                     retryPeriod, TimeUnit.MILLISECONDS);
365         } else {
366             executorService.execute(new Runnable() {
367                 @Override
368                 public void run() {
369                     OvsdbClient client = getChannelClient(channel, ConnectionType.PASSIVE,
370                             Executors.newFixedThreadPool(NUM_THREADS));
371
372                     LOG.debug("Notify listener");
373                     for (OvsdbConnectionListener listener : connectionListeners) {
374                         listener.connected(client);
375                     }
376                 }
377             });
378         }
379     }
380
381     public static void channelClosed(final OvsdbClient client) {
382         LOG.info("Connection closed {}", client.getConnectionInfo().toString());
383         connections.remove(client);
384         for (OvsdbConnectionListener listener : connectionListeners) {
385             listener.disconnected(client);
386         }
387     }
388     @Override
389     public Collection<OvsdbClient> getConnections() {
390         return connections.keySet();
391     }
392 }