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
8 package org.opendaylight.ovsdb.lib.impl;
10 import static java.util.Objects.requireNonNull;
12 import com.google.common.util.concurrent.FutureCallback;
13 import com.google.common.util.concurrent.Futures;
14 import com.google.common.util.concurrent.ListenableFuture;
15 import com.google.common.util.concurrent.ThreadFactoryBuilder;
16 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
17 import io.netty.channel.Channel;
18 import io.netty.channel.ChannelFuture;
19 import io.netty.channel.ChannelFutureListener;
20 import io.netty.channel.ChannelInitializer;
21 import io.netty.channel.socket.SocketChannel;
22 import io.netty.handler.codec.string.StringEncoder;
23 import io.netty.handler.logging.LogLevel;
24 import io.netty.handler.logging.LoggingHandler;
25 import io.netty.handler.ssl.SslHandler;
26 import io.netty.handler.timeout.IdleStateHandler;
27 import io.netty.handler.timeout.ReadTimeoutHandler;
28 import java.net.InetAddress;
29 import java.nio.charset.StandardCharsets;
30 import java.util.ArrayList;
31 import java.util.Arrays;
32 import java.util.Collection;
33 import java.util.List;
35 import java.util.Map.Entry;
37 import java.util.concurrent.ConcurrentHashMap;
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 import java.util.concurrent.atomic.AtomicBoolean;
43 import javax.inject.Inject;
44 import javax.inject.Singleton;
45 import javax.net.ssl.SSLContext;
46 import javax.net.ssl.SSLEngine;
47 import javax.net.ssl.SSLEngineResult.HandshakeStatus;
48 import javax.net.ssl.SSLPeerUnverifiedException;
49 import org.apache.aries.blueprint.annotation.service.Reference;
50 import org.apache.aries.blueprint.annotation.service.Service;
51 import org.eclipse.jdt.annotation.Nullable;
52 import org.opendaylight.aaa.cert.api.ICertificateManager;
53 import org.opendaylight.ovsdb.lib.OvsdbClient;
54 import org.opendaylight.ovsdb.lib.OvsdbConnection;
55 import org.opendaylight.ovsdb.lib.OvsdbConnectionInfo.ConnectionType;
56 import org.opendaylight.ovsdb.lib.OvsdbConnectionInfo.SocketConnectionType;
57 import org.opendaylight.ovsdb.lib.OvsdbConnectionListener;
58 import org.opendaylight.ovsdb.lib.jsonrpc.ExceptionHandler;
59 import org.opendaylight.ovsdb.lib.jsonrpc.JsonRpcDecoder;
60 import org.opendaylight.ovsdb.lib.jsonrpc.JsonRpcEndpoint;
61 import org.slf4j.Logger;
62 import org.slf4j.LoggerFactory;
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
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.
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.
81 @Service(classes = OvsdbConnection.class)
82 public class OvsdbConnectionService implements AutoCloseable, OvsdbConnection {
83 private class ClientChannelInitializer extends ChannelInitializer<SocketChannel> {
85 public void initChannel(final SocketChannel channel) throws Exception {
86 channel.pipeline().addLast(
87 //new LoggingHandler(LogLevel.INFO),
88 new JsonRpcDecoder(jsonRpcDecoderMaxFrameLength),
90 new IdleStateHandler(IDLE_READER_TIMEOUT, 0, 0),
91 new ReadTimeoutHandler(READ_TIMEOUT),
92 new ExceptionHandler(OvsdbConnectionService.this));
96 private class SslClientChannelInitializer extends ClientChannelInitializer {
97 private final ICertificateManager certManagerSrv;
98 private final InetAddress address;
99 private final int port;
101 SslClientChannelInitializer(final ICertificateManager certManagerSrv, final InetAddress address,
103 this.certManagerSrv = requireNonNull(certManagerSrv);
104 this.address = requireNonNull(address);
109 public void initChannel(final SocketChannel channel) throws Exception {
110 SSLContext sslContext = certManagerSrv.getServerContext();
111 if (sslContext != null) {
112 /* First add ssl handler if ssl context is given */
113 SSLEngine engine = sslContext.createSSLEngine(address.toString(), port);
114 engine.setUseClientMode(true);
115 channel.pipeline().addLast("ssl", new SslHandler(engine));
118 super.initChannel(channel);
122 private class ServerChannelInitializer extends ChannelInitializer<SocketChannel> {
124 public final void initChannel(final SocketChannel channel) {
125 LOG.debug("New Passive channel created : {}", channel);
126 initChannelImpl(channel);
129 void initChannelImpl(final SocketChannel channel) {
130 channel.pipeline().addLast(
131 new JsonRpcDecoder(jsonRpcDecoderMaxFrameLength),
133 new IdleStateHandler(IDLE_READER_TIMEOUT, 0, 0),
134 new ReadTimeoutHandler(READ_TIMEOUT),
135 new ExceptionHandler(OvsdbConnectionService.this));
136 handleNewPassiveConnection(channel);
140 private final class SslServerChannelInitializer extends ServerChannelInitializer {
141 private final ICertificateManager certManagerSrv;
142 private final String[] protocols;
143 private final String[] cipherSuites;
145 SslServerChannelInitializer(final ICertificateManager certManagerSrv, final String[] protocols,
146 final String[] cipherSuites) {
147 this.certManagerSrv = requireNonNull(certManagerSrv);
148 this.protocols = requireNonNull(protocols);
149 this.cipherSuites = requireNonNull(cipherSuites);
153 SslServerChannelInitializer(final ICertificateManager certManagerSrv) {
154 this(certManagerSrv, certManagerSrv.getTlsProtocols(), certManagerSrv.getCipherSuites());
158 void initChannelImpl(final SocketChannel channel) {
159 /* Add SSL handler first if SSL context is provided */
160 final SSLContext sslContext = certManagerSrv.getServerContext();
161 if (sslContext != null) {
162 SSLEngine engine = sslContext.createSSLEngine();
163 engine.setUseClientMode(false); // work in a server mode
164 engine.setNeedClientAuth(true); // need client authentication
165 if (protocols != null && protocols.length > 0) {
166 //Set supported protocols
167 engine.setEnabledProtocols(protocols);
168 LOG.debug("Supported ssl protocols {}",
169 Arrays.toString(engine.getSupportedProtocols()));
170 LOG.debug("Enabled ssl protocols {}",
171 Arrays.toString(engine.getEnabledProtocols()));
173 if (cipherSuites != null && cipherSuites.length > 0) {
174 //Set supported cipher suites
175 engine.setEnabledCipherSuites(cipherSuites);
176 LOG.debug("Enabled cipher suites {}",
177 Arrays.toString(engine.getEnabledCipherSuites()));
179 channel.pipeline().addLast("ssl", new SslHandler(engine));
181 super.initChannelImpl(channel);
185 private static final Logger LOG = LoggerFactory.getLogger(OvsdbConnectionService.class);
186 private static final int IDLE_READER_TIMEOUT = 30;
187 private static final int READ_TIMEOUT = 180;
188 private static final String OVSDB_RPC_TASK_TIMEOUT_PARAM = "ovsdb-rpc-task-timeout";
189 private static final String USE_SSL = "use-ssl";
190 private static final int RETRY_PERIOD = 100; // retry after 100 milliseconds
192 private static final StringEncoder UTF8_ENCODER = new StringEncoder(StandardCharsets.UTF_8);
194 private static final ScheduledExecutorService EXECUTOR_SERVICE = Executors.newScheduledThreadPool(10,
195 new ThreadFactoryBuilder().setNameFormat("OVSDBPassiveConnServ-%d").build());
197 private static final ExecutorService CONNECTION_NOTIFIER_SERVICE = Executors
198 .newCachedThreadPool(new ThreadFactoryBuilder().setNameFormat("OVSDBConnNotifSer-%d").build());
200 private static final StalePassiveConnectionService STALE_PASSIVE_CONNECTION_SERVICE =
201 new StalePassiveConnectionService((client) -> {
202 notifyListenerForPassiveConnection(client);
206 private static final Set<OvsdbConnectionListener> CONNECTION_LISTENERS = ConcurrentHashMap.newKeySet();
207 private static final Map<OvsdbClient, Channel> CONNECTIONS = new ConcurrentHashMap<>();
209 private final NettyBootstrapFactory bootstrapFactory;
211 private volatile boolean useSSL = false;
212 private final ICertificateManager certManagerSrv;
214 private volatile int jsonRpcDecoderMaxFrameLength = 100000;
215 private volatile Channel serverChannel;
217 private final AtomicBoolean singletonCreated = new AtomicBoolean(false);
218 private volatile String listenerIp = "0.0.0.0";
219 private volatile int listenerPort = 6640;
222 public OvsdbConnectionService(final NettyBootstrapFactory bootstrapFactory,
223 @Reference(filter = "type=default-certificate-manager") final ICertificateManager certManagerSrv) {
224 this.bootstrapFactory = requireNonNull(bootstrapFactory);
225 this.certManagerSrv = certManagerSrv;
229 * If the SSL flag is enabled, the method internally will establish TLS communication using the default
230 * ODL certificateManager SSLContext and attributes.
233 public OvsdbClient connect(final InetAddress address, final int port) {
235 if (certManagerSrv == null) {
236 LOG.error("Certificate Manager service is not available cannot establish the SSL communication.");
239 return connectWithSsl(address, port, certManagerSrv);
241 return connectWithSsl(address, port, null /* SslContext */);
246 @SuppressWarnings("checkstyle:IllegalCatch")
247 public OvsdbClient connectWithSsl(final InetAddress address, final int port,
248 final ICertificateManager certificateManagerSrv) {
250 final ChannelFuture future = bootstrapFactory.newClient()
251 .handler(certificateManagerSrv == null ? new ClientChannelInitializer() :
252 new SslClientChannelInitializer(certificateManagerSrv, address, port))
253 .connect(address, port);
257 } catch (InterruptedException e) {
258 LOG.warn("Failed to connect {}:{}", address, port, e);
260 } catch (Throwable throwable) {
261 // sync() re-throws exceptions declared as Throwable, so the compiler doesn't see them
262 LOG.error("Error while binding to address {}, port {}", address, port, throwable);
266 return getChannelClient(future.channel(), ConnectionType.ACTIVE, SocketConnectionType.SSL);
270 public void disconnect(final OvsdbClient client) {
271 if (client == null) {
274 Channel channel = CONNECTIONS.get(client);
275 if (channel != null) {
276 //It's an explicit disconnect from user, so no need to notify back
277 //to user about the disconnect.
278 client.setConnectionPublished(false);
279 channel.disconnect();
281 CONNECTIONS.remove(client);
285 public void registerConnectionListener(final OvsdbConnectionListener listener) {
286 LOG.info("registerConnectionListener: registering {}", listener.getClass().getSimpleName());
287 if (CONNECTION_LISTENERS.add(listener)) {
288 LOG.info("registerConnectionListener: registered {} notifying exisitng connections",
289 listener.getClass().getSimpleName());
290 //notify only the first time if called multiple times
291 notifyAlreadyExistingConnectionsToListener(listener);
295 private void notifyAlreadyExistingConnectionsToListener(final OvsdbConnectionListener listener) {
296 for (final OvsdbClient client : getConnections()) {
297 CONNECTION_NOTIFIER_SERVICE.execute(() -> {
298 LOG.trace("Connection {} notified to listener {}", client.getConnectionInfo(), listener);
299 listener.connected(client);
305 public void unregisterConnectionListener(final OvsdbConnectionListener listener) {
306 CONNECTION_LISTENERS.remove(listener);
309 private static OvsdbClient getChannelClient(final Channel channel, final ConnectionType type,
310 final SocketConnectionType socketConnType) {
312 JsonRpcEndpoint endpoint = new JsonRpcEndpoint(channel);
313 channel.pipeline().addLast(endpoint);
315 OvsdbClientImpl client = new OvsdbClientImpl(endpoint, channel, type, socketConnType);
316 client.setConnectionPublished(true);
317 CONNECTIONS.put(client, channel);
318 channel.closeFuture().addListener(new ChannelConnectionHandler(client));
323 * Method that initiates the Passive OVSDB channel listening functionality.
324 * By default the ovsdb passive connection will listen in port 6640 which can
325 * be overridden using the ovsdb.listenPort system property.
328 public synchronized boolean startOvsdbManager() {
329 final int ovsdbListenerPort = this.listenerPort;
330 final String ovsdbListenerIp = this.listenerIp;
331 if (singletonCreated.getAndSet(true)) {
335 LOG.info("startOvsdbManager: Starting");
336 ovsdbManager(ovsdbListenerIp, ovsdbListenerPort);
341 * Method that initiates the Passive OVSDB channel listening functionality
342 * with ssl.By default the ovsdb passive connection will listen in port
343 * 6640 which can be overridden using the ovsdb.listenPort system property.
346 public synchronized boolean startOvsdbManagerWithSsl(final String ovsdbListenIp, final int ovsdbListenPort,
347 final ICertificateManager certificateManagerSrv,
348 final String[] protocols, final String[] cipherSuites) {
349 if (!singletonCreated.getAndSet(true)) {
350 ovsdbManagerWithSsl(ovsdbListenIp, ovsdbListenPort,
351 certificateManagerSrv == null ? new ServerChannelInitializer()
352 : new SslServerChannelInitializer(certificateManagerSrv, protocols, cipherSuites));
360 public synchronized boolean restartOvsdbManagerWithSsl(final String ovsdbListenIp, final int ovsdbListenPort,
361 final ICertificateManager certificateManagerSrv, final String[] protocols, final String[] cipherSuites) {
362 if (singletonCreated.getAndSet(false) && serverChannel != null) {
363 serverChannel.close();
364 LOG.info("Server channel closed");
366 serverChannel = null;
367 return startOvsdbManagerWithSsl(ovsdbListenIp, ovsdbListenPort,
368 certificateManagerSrv, protocols, cipherSuites);
372 * OVSDB Passive listening thread that uses Netty ServerBootstrap to open
373 * passive connection handle channel callbacks.
374 * If the SSL flag is enabled, the method internally will establish TLS communication using the default
375 * ODL certificateManager SSLContext and attributes.
377 private void ovsdbManager(final String ip, final int port) {
379 if (certManagerSrv == null) {
380 LOG.error("Certificate Manager service is not available cannot establish the SSL communication.");
383 ovsdbManagerWithSsl(ip, port, new SslServerChannelInitializer(certManagerSrv));
385 ovsdbManagerWithSsl(ip, port, new ServerChannelInitializer());
390 * OVSDB Passive listening thread that uses Netty ServerBootstrap to open
391 * passive connection with Ssl and handle channel callbacks.
393 @SuppressWarnings("checkstyle:IllegalCatch")
394 private void ovsdbManagerWithSsl(final String ip, final int port, final ServerChannelInitializer channelHandler) {
395 bootstrapFactory.newServer()
396 .handler(new LoggingHandler(LogLevel.INFO))
397 .childHandler(channelHandler)
400 // Propagate the channel when its ready
401 .addListener((ChannelFutureListener) future -> {
402 if (future.isSuccess()) {
403 serverChannel = future.channel();
405 LOG.error("Error while binding to address {}, port {}", ip, port, future.cause());
410 @SuppressWarnings("checkstyle:IllegalCatch")
411 private static void handleNewPassiveConnection(final OvsdbClient client) {
412 ListenableFuture<List<String>> echoFuture = client.echo();
413 LOG.debug("Send echo message to probe the OVSDB switch {}",client.getConnectionInfo());
414 Futures.addCallback(echoFuture, new FutureCallback<List<String>>() {
416 public void onSuccess(@Nullable List<String> result) {
417 LOG.info("Probe was successful to OVSDB switch {}",client.getConnectionInfo());
418 //List<OvsdbClient> clientsFromSameNode = getPassiveClientsFromSameNode(client);
420 getPassiveClientsFromSameNode(client);
421 } catch (Throwable throwable) {
422 LOG.error("Failed to get passive clients from same node", throwable);
424 notifyListenerForPassiveConnection(client);
426 if (clientsFromSameNode.size() == 0) {
427 notifyListenerForPassiveConnection(client);
429 STALE_PASSIVE_CONNECTION_SERVICE.handleNewPassiveConnection(client, clientsFromSameNode);
435 public void onFailure(final Throwable failureException) {
436 LOG.error("Probe failed to OVSDB switch. Disconnecting the channel {}", client.getConnectionInfo());
439 }, CONNECTION_NOTIFIER_SERVICE);
442 @SuppressFBWarnings(value = "UPM_UNCALLED_PRIVATE_METHOD",
443 justification = "https://github.com/spotbugs/spotbugs/issues/811")
444 private static void handleNewPassiveConnection(final Channel channel) {
445 if (!channel.isOpen()) {
446 LOG.warn("Channel {} is not open, skipped further processing of the connection.",channel);
449 SslHandler sslHandler = (SslHandler) channel.pipeline().get("ssl");
450 if (sslHandler != null) {
451 class HandleNewPassiveSslRunner implements Runnable {
452 private int retryTimes = 3;
454 private void retry() {
455 if (retryTimes > 0) {
456 EXECUTOR_SERVICE.schedule(this, RETRY_PERIOD, TimeUnit.MILLISECONDS);
458 LOG.debug("channel closed {}", channel);
459 channel.disconnect();
466 HandshakeStatus status = sslHandler.engine().getHandshakeStatus();
467 LOG.debug("Handshake status {}", status);
470 case NOT_HANDSHAKING:
471 if (sslHandler.engine().getSession().getCipherSuite()
472 .equals("SSL_NULL_WITH_NULL_NULL")) {
473 // Not begin handshake yet. Retry later.
474 LOG.debug("handshake not begin yet {}", status);
477 //Check if peer is trusted before notifying listeners
479 sslHandler.engine().getSession().getPeerCertificates();
480 //Handshake done. Notify listener.
481 OvsdbClient client = getChannelClient(channel, ConnectionType.PASSIVE,
482 SocketConnectionType.SSL);
483 handleNewPassiveConnection(client);
484 } catch (SSLPeerUnverifiedException e) {
485 //Trust manager is still checking peer certificate. Retry later
486 LOG.debug("Peer certifiacte is not verified yet {}", status);
494 //Handshake still ongoing. Retry later.
495 LOG.debug("handshake not done yet {}", status);
500 if (sslHandler.engine().getSession().getCipherSuite()
501 .equals("SSL_NULL_WITH_NULL_NULL")) {
502 /* peer not authenticated. No need to notify listener in this case. */
503 LOG.error("Ssl handshake fail. channel {}", channel);
504 channel.disconnect();
507 * peer is authenticated. Give some time to wait for completion.
508 * If status is still NEED_WRAP, client might already disconnect.
509 * This happens when the first time client connects to controller in two-way handshake.
510 * After obtaining controller certificate, client will disconnect and start
511 * new connection with controller certificate it obtained.
512 * In this case no need to do anything for the first connection attempt. Just skip
513 * since client will reconnect later.
515 LOG.debug("handshake not done yet {}", status);
521 LOG.error("unknown hadshake status {}", status);
526 EXECUTOR_SERVICE.schedule(new HandleNewPassiveSslRunner(),
527 RETRY_PERIOD, TimeUnit.MILLISECONDS);
529 EXECUTOR_SERVICE.execute(() -> {
530 OvsdbClient client = getChannelClient(channel, ConnectionType.PASSIVE,
531 SocketConnectionType.NON_SSL);
532 handleNewPassiveConnection(client);
537 public static void channelClosed(final OvsdbClient client) {
538 LOG.info("Connection closed {}", client.getConnectionInfo());
539 CONNECTIONS.remove(client);
540 if (client.isConnectionPublished()) {
541 for (OvsdbConnectionListener listener : CONNECTION_LISTENERS) {
542 listener.disconnected(client);
545 STALE_PASSIVE_CONNECTION_SERVICE.clientDisconnected(client);
549 public Collection<OvsdbClient> getConnections() {
550 return CONNECTIONS.keySet();
554 public void close() throws Exception {
555 LOG.info("OvsdbConnectionService closed");
556 JsonRpcEndpoint.close();
560 public OvsdbClient getClient(final Channel channel) {
561 for (Entry<OvsdbClient, Channel> entry : CONNECTIONS.entrySet()) {
562 OvsdbClient client = entry.getKey();
563 Channel ctx = entry.getValue();
564 if (ctx.equals(channel)) {
571 @SuppressFBWarnings(value = "UPM_UNCALLED_PRIVATE_METHOD",
572 justification = "https://github.com/spotbugs/spotbugs/issues/811")
573 private static List<OvsdbClient> getPassiveClientsFromSameNode(final OvsdbClient ovsdbClient) {
574 List<OvsdbClient> passiveClients = new ArrayList<>();
575 for (OvsdbClient client : CONNECTIONS.keySet()) {
576 if (!client.equals(ovsdbClient)
577 && client.getConnectionInfo().getRemoteAddress()
578 .equals(ovsdbClient.getConnectionInfo().getRemoteAddress())
579 && client.getConnectionInfo().getType() == ConnectionType.PASSIVE) {
580 passiveClients.add(client);
583 return passiveClients;
586 public static void notifyListenerForPassiveConnection(final OvsdbClient client) {
587 client.setConnectionPublished(true);
588 for (final OvsdbConnectionListener listener : CONNECTION_LISTENERS) {
589 CONNECTION_NOTIFIER_SERVICE.execute(() -> {
590 LOG.trace("Connection {} notified to listener {}", client.getConnectionInfo(), listener);
591 listener.connected(client);
596 public void setOvsdbRpcTaskTimeout(final int timeout) {
597 JsonRpcEndpoint.setReaperInterval(timeout);
603 * @param flag boolean for using ssl
605 public void setUseSsl(final boolean flag) {
610 * Blueprint property setter method. Blueprint call this method and set the value of json rpc decoder
611 * max frame length to the value configured for config option (json-rpc-decoder-max-frame-length) in
612 * the configuration file. This option is only configured at the boot time of the controller. Any
613 * change at the run time will have no impact.
614 * @param maxFrameLength Max frame length (default : 100000)
616 public void setJsonRpcDecoderMaxFrameLength(final int maxFrameLength) {
617 jsonRpcDecoderMaxFrameLength = maxFrameLength;
618 LOG.info("Json Rpc Decoder Max Frame Length set to : {}", jsonRpcDecoderMaxFrameLength);
621 public void setOvsdbListenerIp(final String ip) {
622 LOG.info("OVSDB IP for listening connection is set to : {}", ip);
626 public void setOvsdbListenerPort(final int portNumber) {
627 LOG.info("OVSDB port for listening connection is set to : {}", portNumber);
628 listenerPort = portNumber;
631 public void updateConfigParameter(final Map<String, Object> configParameters) {
632 if (configParameters != null && !configParameters.isEmpty()) {
633 LOG.debug("Config parameters received : {}", configParameters.entrySet());
634 for (Map.Entry<String, Object> paramEntry : configParameters.entrySet()) {
635 if (paramEntry.getKey().equalsIgnoreCase(OVSDB_RPC_TASK_TIMEOUT_PARAM)) {
636 setOvsdbRpcTaskTimeout(Integer.parseInt((String)paramEntry.getValue()));
637 } else if (paramEntry.getKey().equalsIgnoreCase(USE_SSL)) {
638 useSSL = Boolean.parseBoolean(paramEntry.getValue().toString());