2 * Copyright © 2014, 2017 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 static java.util.Objects.requireNonNull;
13 import com.google.common.util.concurrent.FutureCallback;
14 import com.google.common.util.concurrent.Futures;
15 import com.google.common.util.concurrent.ListenableFuture;
16 import com.google.common.util.concurrent.ThreadFactoryBuilder;
17 import io.netty.bootstrap.Bootstrap;
18 import io.netty.channel.Channel;
19 import io.netty.channel.ChannelFuture;
20 import io.netty.channel.ChannelFutureListener;
21 import io.netty.channel.ChannelInitializer;
22 import io.netty.channel.socket.SocketChannel;
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.handler.ssl.SslHandler;
27 import io.netty.handler.timeout.IdleStateHandler;
28 import io.netty.handler.timeout.ReadTimeoutHandler;
29 import java.net.InetAddress;
30 import java.nio.charset.StandardCharsets;
31 import java.util.ArrayList;
32 import java.util.Arrays;
33 import java.util.Collection;
34 import java.util.List;
36 import java.util.Map.Entry;
38 import java.util.concurrent.ConcurrentHashMap;
39 import java.util.concurrent.ExecutorService;
40 import java.util.concurrent.Executors;
41 import java.util.concurrent.ScheduledExecutorService;
42 import java.util.concurrent.TimeUnit;
43 import java.util.concurrent.atomic.AtomicBoolean;
44 import javax.annotation.Nullable;
45 import javax.inject.Inject;
46 import javax.inject.Singleton;
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.apache.aries.blueprint.annotation.service.Reference;
52 import org.apache.aries.blueprint.annotation.service.Service;
53 import org.opendaylight.aaa.cert.api.ICertificateManager;
54 import org.opendaylight.ovsdb.lib.OvsdbClient;
55 import org.opendaylight.ovsdb.lib.OvsdbConnection;
56 import org.opendaylight.ovsdb.lib.OvsdbConnectionInfo.ConnectionType;
57 import org.opendaylight.ovsdb.lib.OvsdbConnectionInfo.SocketConnectionType;
58 import org.opendaylight.ovsdb.lib.OvsdbConnectionListener;
59 import org.opendaylight.ovsdb.lib.jsonrpc.ExceptionHandler;
60 import org.opendaylight.ovsdb.lib.jsonrpc.JsonRpcDecoder;
61 import org.opendaylight.ovsdb.lib.jsonrpc.JsonRpcEndpoint;
62 import org.slf4j.Logger;
63 import org.slf4j.LoggerFactory;
66 * OvsDBConnectionService provides OVSDB connection management functionality which includes
67 * both Active and Passive connections.
68 * From the Library perspective, Active OVSDB connections are those that are initiated from
69 * the Controller towards the ovsdb-manager.
70 * While Passive OVSDB connections are those that are initiated from the ovs towards
73 * <p>Applications that use OvsDBConnectionService can use the OvsDBConnection class' connect APIs
74 * to initiate Active connections and can listen to the asynchronous Passive connections via
75 * registerConnectionListener listener API.
77 * <p>The library is designed as Java modular component that can work in both OSGi and non-OSGi
78 * environment. Hence a single instance of the service will be active (via Service Registry in OSGi)
79 * and a Singleton object in a non-OSGi environment.
82 @Service(classes = OvsdbConnection.class)
83 public class OvsdbConnectionService implements AutoCloseable, OvsdbConnection {
84 private static final Logger LOG = LoggerFactory.getLogger(OvsdbConnectionService.class);
85 private static final int IDLE_READER_TIMEOUT = 30;
86 private static final int READ_TIMEOUT = 180;
87 private static final String OVSDB_RPC_TASK_TIMEOUT_PARAM = "ovsdb-rpc-task-timeout";
88 private static final String USE_SSL = "use-ssl";
89 private static final int RETRY_PERIOD = 100; // retry after 100 milliseconds
91 private static final StringEncoder UTF8_ENCODER = new StringEncoder(StandardCharsets.UTF_8);
93 private static final ScheduledExecutorService EXECUTOR_SERVICE = Executors.newScheduledThreadPool(10,
94 new ThreadFactoryBuilder().setNameFormat("OVSDBPassiveConnServ-%d").build());
96 private static final ExecutorService CONNECTION_NOTIFIER_SERVICE = Executors
97 .newCachedThreadPool(new ThreadFactoryBuilder().setNameFormat("OVSDBConnNotifSer-%d").build());
99 private static final StalePassiveConnectionService STALE_PASSIVE_CONNECTION_SERVICE =
100 new StalePassiveConnectionService((client) -> {
101 notifyListenerForPassiveConnection(client);
105 private static final Set<OvsdbConnectionListener> CONNECTION_LISTENERS = ConcurrentHashMap.newKeySet();
106 private static final Map<OvsdbClient, Channel> CONNECTIONS = new ConcurrentHashMap<>();
108 private final NettyBootstrapFactory bootstrapFactory;
110 private volatile boolean useSSL = false;
111 private final ICertificateManager certManagerSrv;
113 private volatile int jsonRpcDecoderMaxFrameLength = 100000;
114 private volatile Channel serverChannel;
116 private final AtomicBoolean singletonCreated = new AtomicBoolean(false);
117 private volatile String listenerIp = "0.0.0.0";
118 private volatile int listenerPort = 6640;
121 public OvsdbConnectionService(final NettyBootstrapFactory bootstrapFactory,
122 @Reference(filter = "type=default-certificate-manager") final ICertificateManager certManagerSrv) {
123 this.bootstrapFactory = requireNonNull(bootstrapFactory);
124 this.certManagerSrv = certManagerSrv;
128 * If the SSL flag is enabled, the method internally will establish TLS communication using the default
129 * ODL certificateManager SSLContext and attributes.
132 public OvsdbClient connect(final InetAddress address, final int port) {
134 if (certManagerSrv == null) {
135 LOG.error("Certificate Manager service is not available cannot establish the SSL communication.");
138 return connectWithSsl(address, port, certManagerSrv);
140 return connectWithSsl(address, port, null /* SslContext */);
145 @SuppressWarnings("checkstyle:IllegalCatch")
146 public OvsdbClient connectWithSsl(final InetAddress address, final int port,
147 final ICertificateManager certificateManagerSrv) {
149 Bootstrap bootstrap = bootstrapFactory.newClient()
150 .handler(new ChannelInitializer<SocketChannel>() {
152 public void initChannel(final SocketChannel channel) throws Exception {
153 if (certificateManagerSrv != null && certificateManagerSrv.getServerContext() != null) {
154 SSLContext sslContext = certificateManagerSrv.getServerContext();
155 /* First add ssl handler if ssl context is given */
157 sslContext.createSSLEngine(address.toString(), port);
158 engine.setUseClientMode(true);
159 channel.pipeline().addLast("ssl", new SslHandler(engine));
161 channel.pipeline().addLast(
162 //new LoggingHandler(LogLevel.INFO),
163 new JsonRpcDecoder(jsonRpcDecoderMaxFrameLength),
165 new IdleStateHandler(IDLE_READER_TIMEOUT, 0, 0),
166 new ReadTimeoutHandler(READ_TIMEOUT),
167 new ExceptionHandler(OvsdbConnectionService.this));
172 ChannelFuture future = bootstrap.connect(address, port).sync();
173 Channel channel = future.channel();
174 return getChannelClient(channel, ConnectionType.ACTIVE, SocketConnectionType.SSL);
175 } catch (InterruptedException e) {
176 LOG.warn("Failed to connect {}:{}", address, port, e);
177 } catch (Throwable throwable) {
178 // sync() re-throws exceptions declared as Throwable, so the compiler doesn't see them
179 LOG.error("Error while binding to address {}, port {}", address, port, throwable);
186 public void disconnect(final OvsdbClient client) {
187 if (client == null) {
190 Channel channel = CONNECTIONS.get(client);
191 if (channel != null) {
192 //It's an explicit disconnect from user, so no need to notify back
193 //to user about the disconnect.
194 client.setConnectionPublished(false);
195 channel.disconnect();
197 CONNECTIONS.remove(client);
201 public void registerConnectionListener(final OvsdbConnectionListener listener) {
202 LOG.info("registerConnectionListener: registering {}", listener.getClass().getSimpleName());
203 CONNECTION_LISTENERS.add(listener);
204 notifyAlreadyExistingConnectionsToListener(listener);
207 private void notifyAlreadyExistingConnectionsToListener(final OvsdbConnectionListener listener) {
208 for (final OvsdbClient client : getConnections()) {
209 CONNECTION_NOTIFIER_SERVICE.execute(() -> {
210 LOG.trace("Connection {} notified to listener {}", client.getConnectionInfo(), listener);
211 listener.connected(client);
217 public void unregisterConnectionListener(final OvsdbConnectionListener listener) {
218 CONNECTION_LISTENERS.remove(listener);
221 private static OvsdbClient getChannelClient(final Channel channel, final ConnectionType type,
222 final SocketConnectionType socketConnType) {
224 JsonRpcEndpoint endpoint = new JsonRpcEndpoint(channel);
225 channel.pipeline().addLast(endpoint);
227 OvsdbClientImpl client = new OvsdbClientImpl(endpoint, channel, type, socketConnType);
228 client.setConnectionPublished(true);
229 CONNECTIONS.put(client, channel);
230 channel.closeFuture().addListener(new ChannelConnectionHandler(client));
235 * Method that initiates the Passive OVSDB channel listening functionality.
236 * By default the ovsdb passive connection will listen in port 6640 which can
237 * be overridden using the ovsdb.listenPort system property.
240 public synchronized boolean startOvsdbManager() {
241 final int ovsdbListenerPort = this.listenerPort;
242 final String ovsdbListenerIp = this.listenerIp;
243 if (singletonCreated.getAndSet(true)) {
247 LOG.info("startOvsdbManager: Starting");
248 ovsdbManager(ovsdbListenerIp, ovsdbListenerPort);
253 * Method that initiates the Passive OVSDB channel listening functionality
254 * with ssl.By default the ovsdb passive connection will listen in port
255 * 6640 which can be overridden using the ovsdb.listenPort system property.
258 public synchronized boolean startOvsdbManagerWithSsl(final String ovsdbListenIp, final int ovsdbListenPort,
259 final ICertificateManager certificateManagerSrv,
260 final String[] protocols, final String[] cipherSuites) {
261 if (!singletonCreated.getAndSet(true)) {
262 ovsdbManagerWithSsl(ovsdbListenIp, ovsdbListenPort, certificateManagerSrv, protocols, cipherSuites);
270 public synchronized boolean restartOvsdbManagerWithSsl(final String ovsdbListenIp, final int ovsdbListenPort,
271 final ICertificateManager certificateManagerSrv, final String[] protocols, final String[] cipherSuites) {
272 if (singletonCreated.getAndSet(false) && serverChannel != null) {
273 serverChannel.close();
274 LOG.info("Server channel closed");
276 serverChannel = null;
277 return startOvsdbManagerWithSsl(ovsdbListenIp, ovsdbListenPort,
278 certificateManagerSrv, protocols, cipherSuites);
282 * OVSDB Passive listening thread that uses Netty ServerBootstrap to open
283 * passive connection handle channel callbacks.
284 * If the SSL flag is enabled, the method internally will establish TLS communication using the default
285 * ODL certificateManager SSLContext and attributes.
287 private void ovsdbManager(final String ip, final int port) {
289 if (certManagerSrv == null) {
290 LOG.error("Certificate Manager service is not available cannot establish the SSL communication.");
293 ovsdbManagerWithSsl(ip, port, certManagerSrv, certManagerSrv.getTlsProtocols(),
294 certManagerSrv.getCipherSuites());
296 ovsdbManagerWithSsl(ip, port, null /* SslContext */, null, null);
301 * OVSDB Passive listening thread that uses Netty ServerBootstrap to open
302 * passive connection with Ssl and handle channel callbacks.
304 @SuppressWarnings("checkstyle:IllegalCatch")
305 private void ovsdbManagerWithSsl(final String ip, final int port, final ICertificateManager certificateManagerSrv,
306 final String[] protocols, final String[] cipherSuites) {
308 bootstrapFactory.newServer()
309 .handler(new LoggingHandler(LogLevel.INFO))
310 .childHandler(new ChannelInitializer<SocketChannel>() {
312 public void initChannel(final SocketChannel channel) throws Exception {
313 LOG.debug("New Passive channel created : {}", channel);
314 if (certificateManagerSrv != null) {
315 /* Add SSL handler first if SSL context is provided */
316 final SSLContext sslContext = certificateManagerSrv.getServerContext();
317 if (sslContext != null) {
318 SSLEngine engine = sslContext.createSSLEngine();
319 engine.setUseClientMode(false); // work in a server mode
320 engine.setNeedClientAuth(true); // need client authentication
321 if (protocols != null && protocols.length > 0) {
322 //Set supported protocols
323 engine.setEnabledProtocols(protocols);
324 LOG.debug("Supported ssl protocols {}",
325 Arrays.toString(engine.getSupportedProtocols()));
326 LOG.debug("Enabled ssl protocols {}",
327 Arrays.toString(engine.getEnabledProtocols()));
329 if (cipherSuites != null && cipherSuites.length > 0) {
330 //Set supported cipher suites
331 engine.setEnabledCipherSuites(cipherSuites);
332 LOG.debug("Enabled cipher suites {}",
333 Arrays.toString(engine.getEnabledCipherSuites()));
335 channel.pipeline().addLast("ssl", new SslHandler(engine));
339 channel.pipeline().addLast(
340 new JsonRpcDecoder(jsonRpcDecoderMaxFrameLength),
342 new IdleStateHandler(IDLE_READER_TIMEOUT, 0, 0),
343 new ReadTimeoutHandler(READ_TIMEOUT),
344 new ExceptionHandler(OvsdbConnectionService.this));
346 handleNewPassiveConnection(channel);
351 // Propagate the channel when its ready
352 .addListener((ChannelFutureListener) future -> {
353 if (future.isSuccess()) {
354 serverChannel = future.channel();
356 LOG.error("Error while binding to address {}, port {}", ip, port, future.cause());
361 private static void handleNewPassiveConnection(final OvsdbClient client) {
362 ListenableFuture<List<String>> echoFuture = client.echo();
363 LOG.debug("Send echo message to probe the OVSDB switch {}",client.getConnectionInfo());
364 Futures.addCallback(echoFuture, new FutureCallback<List<String>>() {
366 public void onSuccess(@Nullable final List<String> result) {
367 LOG.debug("Probe was successful to OVSDB switch {}",client.getConnectionInfo());
368 List<OvsdbClient> clientsFromSameNode = getPassiveClientsFromSameNode(client);
369 if (clientsFromSameNode.size() == 0) {
370 notifyListenerForPassiveConnection(client);
372 STALE_PASSIVE_CONNECTION_SERVICE.handleNewPassiveConnection(client, clientsFromSameNode);
377 public void onFailure(final Throwable failureException) {
378 LOG.error("Probe failed to OVSDB switch. Disconnecting the channel {}", client.getConnectionInfo());
381 }, CONNECTION_NOTIFIER_SERVICE);
384 private static void handleNewPassiveConnection(final Channel channel) {
385 if (!channel.isOpen()) {
386 LOG.warn("Channel {} is not open, skipped further processing of the connection.",channel);
389 SslHandler sslHandler = (SslHandler) channel.pipeline().get("ssl");
390 if (sslHandler != null) {
391 class HandleNewPassiveSslRunner implements Runnable {
392 private int retryTimes = 3;
394 private void retry() {
395 if (retryTimes > 0) {
396 EXECUTOR_SERVICE.schedule(this, RETRY_PERIOD, TimeUnit.MILLISECONDS);
398 LOG.debug("channel closed {}", channel);
399 channel.disconnect();
406 HandshakeStatus status = sslHandler.engine().getHandshakeStatus();
407 LOG.debug("Handshake status {}", status);
410 case NOT_HANDSHAKING:
411 if (sslHandler.engine().getSession().getCipherSuite()
412 .equals("SSL_NULL_WITH_NULL_NULL")) {
413 // Not begin handshake yet. Retry later.
414 LOG.debug("handshake not begin yet {}", status);
417 //Check if peer is trusted before notifying listeners
419 sslHandler.engine().getSession().getPeerCertificates();
420 //Handshake done. Notify listener.
421 OvsdbClient client = getChannelClient(channel, ConnectionType.PASSIVE,
422 SocketConnectionType.SSL);
423 handleNewPassiveConnection(client);
424 } catch (SSLPeerUnverifiedException e) {
425 //Trust manager is still checking peer certificate. Retry later
426 LOG.debug("Peer certifiacte is not verified yet {}", status);
434 //Handshake still ongoing. Retry later.
435 LOG.debug("handshake not done yet {}", status);
440 if (sslHandler.engine().getSession().getCipherSuite()
441 .equals("SSL_NULL_WITH_NULL_NULL")) {
442 /* peer not authenticated. No need to notify listener in this case. */
443 LOG.error("Ssl handshake fail. channel {}", channel);
444 channel.disconnect();
447 * peer is authenticated. Give some time to wait for completion.
448 * If status is still NEED_WRAP, client might already disconnect.
449 * This happens when the first time client connects to controller in two-way handshake.
450 * After obtaining controller certificate, client will disconnect and start
451 * new connection with controller certificate it obtained.
452 * In this case no need to do anything for the first connection attempt. Just skip
453 * since client will reconnect later.
455 LOG.debug("handshake not done yet {}", status);
461 LOG.error("unknown hadshake status {}", status);
466 EXECUTOR_SERVICE.schedule(new HandleNewPassiveSslRunner(),
467 RETRY_PERIOD, TimeUnit.MILLISECONDS);
469 EXECUTOR_SERVICE.execute(() -> {
470 OvsdbClient client = getChannelClient(channel, ConnectionType.PASSIVE,
471 SocketConnectionType.NON_SSL);
472 handleNewPassiveConnection(client);
477 public static void channelClosed(final OvsdbClient client) {
478 LOG.info("Connection closed {}", client.getConnectionInfo());
479 CONNECTIONS.remove(client);
480 if (client.isConnectionPublished()) {
481 for (OvsdbConnectionListener listener : CONNECTION_LISTENERS) {
482 listener.disconnected(client);
485 STALE_PASSIVE_CONNECTION_SERVICE.clientDisconnected(client);
489 public Collection<OvsdbClient> getConnections() {
490 return CONNECTIONS.keySet();
494 public void close() throws Exception {
495 LOG.info("OvsdbConnectionService closed");
496 JsonRpcEndpoint.close();
500 public OvsdbClient getClient(final Channel channel) {
501 for (Entry<OvsdbClient, Channel> entry : CONNECTIONS.entrySet()) {
502 OvsdbClient client = entry.getKey();
503 Channel ctx = entry.getValue();
504 if (ctx.equals(channel)) {
511 private static List<OvsdbClient> getPassiveClientsFromSameNode(final OvsdbClient ovsdbClient) {
512 List<OvsdbClient> passiveClients = new ArrayList<>();
513 for (OvsdbClient client : CONNECTIONS.keySet()) {
514 if (!client.equals(ovsdbClient)
515 && client.getConnectionInfo().getRemoteAddress()
516 .equals(ovsdbClient.getConnectionInfo().getRemoteAddress())
517 && client.getConnectionInfo().getType() == ConnectionType.PASSIVE) {
518 passiveClients.add(client);
521 return passiveClients;
524 public static void notifyListenerForPassiveConnection(final OvsdbClient client) {
525 client.setConnectionPublished(true);
526 for (final OvsdbConnectionListener listener : CONNECTION_LISTENERS) {
527 CONNECTION_NOTIFIER_SERVICE.execute(() -> {
528 LOG.trace("Connection {} notified to listener {}", client.getConnectionInfo(), listener);
529 listener.connected(client);
534 public void setOvsdbRpcTaskTimeout(final int timeout) {
535 JsonRpcEndpoint.setReaperInterval(timeout);
541 * @param flag boolean for using ssl
543 public void setUseSsl(final boolean flag) {
548 * Blueprint property setter method. Blueprint call this method and set the value of json rpc decoder
549 * max frame length to the value configured for config option (json-rpc-decoder-max-frame-length) in
550 * the configuration file. This option is only configured at the boot time of the controller. Any
551 * change at the run time will have no impact.
552 * @param maxFrameLength Max frame length (default : 100000)
554 public void setJsonRpcDecoderMaxFrameLength(final int maxFrameLength) {
555 jsonRpcDecoderMaxFrameLength = maxFrameLength;
556 LOG.info("Json Rpc Decoder Max Frame Length set to : {}", jsonRpcDecoderMaxFrameLength);
559 public void setOvsdbListenerIp(final String ip) {
560 LOG.info("OVSDB IP for listening connection is set to : {}", ip);
564 public void setOvsdbListenerPort(final int portNumber) {
565 LOG.info("OVSDB port for listening connection is set to : {}", portNumber);
566 listenerPort = portNumber;
569 public void updateConfigParameter(final Map<String, Object> configParameters) {
570 if (configParameters != null && !configParameters.isEmpty()) {
571 LOG.debug("Config parameters received : {}", configParameters.entrySet());
572 for (Map.Entry<String, Object> paramEntry : configParameters.entrySet()) {
573 if (paramEntry.getKey().equalsIgnoreCase(OVSDB_RPC_TASK_TIMEOUT_PARAM)) {
574 setOvsdbRpcTaskTimeout(Integer.parseInt((String)paramEntry.getValue()));
575 } else if (paramEntry.getKey().equalsIgnoreCase(USE_SSL)) {
576 useSSL = Boolean.parseBoolean(paramEntry.getValue().toString());