/*
- * Copyright (c) 2014, 2015 Red Hat, Inc. and others. All rights reserved.
+ * Copyright © 2014, 2017 Red Hat, Inc. and others. All rights reserved.
*
* This program and the accompanying materials are made available under the
* terms of the Eclipse Public License v1.0 which accompanies this distribution,
import com.fasterxml.jackson.annotation.JsonInclude.Include;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
-import com.google.common.collect.Maps;
-import com.google.common.collect.Sets;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import java.util.Collection;
import java.util.List;
import java.util.Map;
+import java.util.Map.Entry;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
-import javax.annotation.Nullable;
+import java.util.concurrent.atomic.AtomicBoolean;
+import javax.inject.Inject;
+import javax.inject.Singleton;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLEngine;
import javax.net.ssl.SSLEngineResult.HandshakeStatus;
import javax.net.ssl.SSLPeerUnverifiedException;
+import org.apache.aries.blueprint.annotation.service.Reference;
+import org.apache.aries.blueprint.annotation.service.Service;
+import org.eclipse.jdt.annotation.Nullable;
+import org.opendaylight.aaa.cert.api.ICertificateManager;
import org.opendaylight.ovsdb.lib.OvsdbClient;
import org.opendaylight.ovsdb.lib.OvsdbConnection;
import org.opendaylight.ovsdb.lib.OvsdbConnectionInfo.ConnectionType;
* environment. Hence a single instance of the service will be active (via Service Registry in OSGi)
* and a Singleton object in a non-OSGi environment.
*/
+@Singleton
+@Service(classes = OvsdbConnection.class)
public class OvsdbConnectionService implements AutoCloseable, OvsdbConnection {
private static final Logger LOG = LoggerFactory.getLogger(OvsdbConnectionService.class);
-
- private static ThreadFactory passiveConnectionThreadFactory = new ThreadFactoryBuilder()
- .setNameFormat("OVSDBPassiveConnServ-%d").build();
- private static ScheduledExecutorService executorService
- = Executors.newScheduledThreadPool(10, passiveConnectionThreadFactory);
-
- private static ThreadFactory connectionNotifierThreadFactory = new ThreadFactoryBuilder()
- .setNameFormat("OVSDBConnNotifSer-%d").build();
- private static ExecutorService connectionNotifierService
- = Executors.newCachedThreadPool(connectionNotifierThreadFactory);
-
- private static Set<OvsdbConnectionListener> connectionListeners = Sets.newHashSet();
- private static Map<OvsdbClient, Channel> connections = Maps.newHashMap();
- private static OvsdbConnection connectionService;
- private static volatile boolean singletonCreated = false;
private static final int IDLE_READER_TIMEOUT = 30;
private static final int READ_TIMEOUT = 180;
private static final String OVSDB_RPC_TASK_TIMEOUT_PARAM = "ovsdb-rpc-task-timeout";
+ private static final String USE_SSL = "use-ssl";
+ private static final int RETRY_PERIOD = 100; // retry after 100 milliseconds
+
+ private static final ScheduledExecutorService EXECUTOR_SERVICE = Executors.newScheduledThreadPool(10,
+ new ThreadFactoryBuilder().setNameFormat("OVSDBPassiveConnServ-%d").build());
+
+ private static final ExecutorService CONNECTION_NOTIFIER_SERVICE = Executors
+ .newCachedThreadPool(new ThreadFactoryBuilder().setNameFormat("OVSDBConnNotifSer-%d").build());
private static final StalePassiveConnectionService STALE_PASSIVE_CONNECTION_SERVICE =
- new StalePassiveConnectionService(executorService);
+ new StalePassiveConnectionService((client) -> {
+ notifyListenerForPassiveConnection(client);
+ return null;
+ });
- private static int retryPeriod = 100; // retry after 100 milliseconds
+ private static final Set<OvsdbConnectionListener> CONNECTION_LISTENERS = ConcurrentHashMap.newKeySet();
+ private static final Map<OvsdbClient, Channel> CONNECTIONS = new ConcurrentHashMap<>();
+ private volatile boolean useSSL = false;
+ private final ICertificateManager certManagerSrv;
- public static OvsdbConnection getService() {
- if (connectionService == null) {
- connectionService = new OvsdbConnectionService();
- }
- return connectionService;
+ private volatile int jsonRpcDecoderMaxFrameLength = 100000;
+ private volatile Channel serverChannel;
+
+ private final AtomicBoolean singletonCreated = new AtomicBoolean(false);
+ private volatile String listenerIp = "0.0.0.0";
+ private volatile int listenerPort = 6640;
+
+ @Inject
+ public OvsdbConnectionService(@Reference(filter = "type=default-certificate-manager")
+ ICertificateManager certManagerSrv) {
+ this.certManagerSrv = certManagerSrv;
}
+ /**
+ * If the SSL flag is enabled, the method internally will establish TLS communication using the default
+ * ODL certificateManager SSLContext and attributes.
+ */
@Override
public OvsdbClient connect(final InetAddress address, final int port) {
- return connectWithSsl(address, port, null /* SslContext */);
+ if (useSSL) {
+ if (certManagerSrv == null) {
+ LOG.error("Certificate Manager service is not available cannot establish the SSL communication.");
+ return null;
+ }
+ return connectWithSsl(address, port, certManagerSrv);
+ } else {
+ return connectWithSsl(address, port, null /* SslContext */);
+ }
}
@Override
+ @SuppressWarnings("checkstyle:IllegalCatch")
public OvsdbClient connectWithSsl(final InetAddress address, final int port,
- final SSLContext sslContext) {
+ final ICertificateManager certificateManagerSrv) {
try {
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(new NioEventLoopGroup());
bootstrap.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel channel) throws Exception {
- if (sslContext != null) {
+ if (certificateManagerSrv != null && certificateManagerSrv.getServerContext() != null) {
+ SSLContext sslContext = certificateManagerSrv.getServerContext();
/* First add ssl handler if ssl context is given */
SSLEngine engine =
sslContext.createSSLEngine(address.toString(), port);
}
channel.pipeline().addLast(
//new LoggingHandler(LogLevel.INFO),
- new JsonRpcDecoder(100000),
+ new JsonRpcDecoder(jsonRpcDecoderMaxFrameLength),
new StringEncoder(CharsetUtil.UTF_8),
new IdleStateHandler(IDLE_READER_TIMEOUT, 0, 0),
new ReadTimeoutHandler(READ_TIMEOUT),
- new ExceptionHandler());
+ new ExceptionHandler(OvsdbConnectionService.this));
}
});
return getChannelClient(channel, ConnectionType.ACTIVE, SocketConnectionType.SSL);
} catch (InterruptedException e) {
LOG.warn("Failed to connect {}:{}", address, port, e);
+ } catch (Throwable throwable) {
+ // sync() re-throws exceptions declared as Throwable, so the compiler doesn't see them
+ LOG.error("Error while binding to address {}, port {}", address, port, throwable);
+ throw throwable;
}
return null;
}
if (client == null) {
return;
}
- Channel channel = connections.get(client);
+ Channel channel = CONNECTIONS.get(client);
if (channel != null) {
+ //It's an explicit disconnect from user, so no need to notify back
+ //to user about the disconnect.
+ client.setConnectionPublished(false);
channel.disconnect();
}
- connections.remove(client);
+ CONNECTIONS.remove(client);
}
@Override
public void registerConnectionListener(OvsdbConnectionListener listener) {
LOG.info("registerConnectionListener: registering {}", listener.getClass().getSimpleName());
- connectionListeners.add(listener);
+ CONNECTION_LISTENERS.add(listener);
+ notifyAlreadyExistingConnectionsToListener(listener);
+ }
+
+ private void notifyAlreadyExistingConnectionsToListener(final OvsdbConnectionListener listener) {
+ for (final OvsdbClient client : getConnections()) {
+ CONNECTION_NOTIFIER_SERVICE.execute(() -> {
+ LOG.trace("Connection {} notified to listener {}", client.getConnectionInfo(), listener);
+ listener.connected(client);
+ });
+ }
}
@Override
public void unregisterConnectionListener(OvsdbConnectionListener listener) {
- connectionListeners.remove(listener);
+ CONNECTION_LISTENERS.remove(listener);
}
private static OvsdbClient getChannelClient(Channel channel, ConnectionType type,
OvsdbRPC rpc = factory.getClient(channel, OvsdbRPC.class);
OvsdbClientImpl client = new OvsdbClientImpl(rpc, channel, type, socketConnType);
client.setConnectionPublished(true);
- connections.put(client, channel);
+ CONNECTIONS.put(client, channel);
ChannelFuture closeFuture = channel.closeFuture();
closeFuture.addListener(new ChannelConnectionHandler(client));
return client;
* be overridden using the ovsdb.listenPort system property.
*/
@Override
- public synchronized boolean startOvsdbManager(final int ovsdbListenPort) {
- if (!singletonCreated) {
+ public synchronized boolean startOvsdbManager() {
+ final int ovsdbListenerPort = this.listenerPort;
+ final String ovsdbListenerIp = this.listenerIp;
+ if (!singletonCreated.getAndSet(true)) {
LOG.info("startOvsdbManager: Starting");
- new Thread() {
- @Override
- public void run() {
- ovsdbManager(ovsdbListenPort);
- }
- }.start();
- singletonCreated = true;
+ new Thread(() -> ovsdbManager(ovsdbListenerIp, ovsdbListenerPort)).start();
return true;
} else {
return false;
* 6640 which can be overridden using the ovsdb.listenPort system property.
*/
@Override
- public synchronized boolean startOvsdbManagerWithSsl(final int ovsdbListenPort,
- final SSLContext sslContext) {
- if (!singletonCreated) {
- new Thread() {
- @Override
- public void run() {
- ovsdbManagerWithSsl(ovsdbListenPort, sslContext);
- }
- }.start();
- singletonCreated = true;
+ public synchronized boolean startOvsdbManagerWithSsl(final String ovsdbListenIp, final int ovsdbListenPort,
+ final ICertificateManager certificateManagerSrv,
+ String[] protocols, String[] cipherSuites) {
+ if (!singletonCreated.getAndSet(true)) {
+ new Thread(() -> ovsdbManagerWithSsl(ovsdbListenIp, ovsdbListenPort,
+ certificateManagerSrv, protocols, cipherSuites)).start();
return true;
} else {
return false;
}
}
+ @Override
+ public synchronized boolean restartOvsdbManagerWithSsl(final String ovsdbListenIp,
+ final int ovsdbListenPort,
+ final ICertificateManager certificateManagerSrv,
+ final String[] protocols,
+ final String[] cipherSuites) {
+ if (singletonCreated.getAndSet(false) && serverChannel != null) {
+ serverChannel.close();
+ LOG.info("Server channel closed");
+ }
+ serverChannel = null;
+ return startOvsdbManagerWithSsl(ovsdbListenIp, ovsdbListenPort,
+ certificateManagerSrv, protocols, cipherSuites);
+ }
+
/**
* OVSDB Passive listening thread that uses Netty ServerBootstrap to open
* passive connection handle channel callbacks.
+ * If the SSL flag is enabled, the method internally will establish TLS communication using the default
+ * ODL certificateManager SSLContext and attributes.
*/
- private static void ovsdbManager(int port) {
- ovsdbManagerWithSsl(port, null /* SslContext */);
+ private void ovsdbManager(String ip, int port) {
+ if (useSSL) {
+ if (certManagerSrv == null) {
+ LOG.error("Certificate Manager service is not available cannot establish the SSL communication.");
+ return;
+ }
+ ovsdbManagerWithSsl(ip, port, certManagerSrv, certManagerSrv.getTlsProtocols(),
+ certManagerSrv.getCipherSuites());
+ } else {
+ ovsdbManagerWithSsl(ip, port, null /* SslContext */, null, null);
+ }
}
/**
* OVSDB Passive listening thread that uses Netty ServerBootstrap to open
* passive connection with Ssl and handle channel callbacks.
*/
- private static void ovsdbManagerWithSsl(int port, final SSLContext sslContext) {
+ @SuppressWarnings("checkstyle:IllegalCatch")
+ private void ovsdbManagerWithSsl(String ip, int port, final ICertificateManager certificateManagerSrv,
+ final String[] protocols, final String[] cipherSuites) {
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
@Override
public void initChannel(SocketChannel channel) throws Exception {
LOG.debug("New Passive channel created : {}", channel);
- if (sslContext != null) {
+ if (certificateManagerSrv != null && certificateManagerSrv.getServerContext() != null) {
/* Add SSL handler first if SSL context is provided */
+ SSLContext sslContext = certificateManagerSrv.getServerContext();
SSLEngine engine = sslContext.createSSLEngine();
engine.setUseClientMode(false); // work in a server mode
engine.setNeedClientAuth(true); // need client authentication
- //Disable SSLv3 and enable all other supported protocols
- String[] protocols = {"SSLv2Hello", "TLSv1", "TLSv1.1", "TLSv1.2"};
- LOG.debug("Set enable protocols {}", Arrays.toString(protocols));
- engine.setEnabledProtocols(protocols);
- LOG.debug("Supported ssl protocols {}",
- Arrays.toString(engine.getSupportedProtocols()));
- LOG.debug("Enabled ssl protocols {}",
- Arrays.toString(engine.getEnabledProtocols()));
- //Set cipher suites
- String[] cipherSuites = {"TLS_RSA_WITH_AES_128_CBC_SHA"};
- LOG.debug("Set enable cipher cuites {}", Arrays.toString(cipherSuites));
- engine.setEnabledCipherSuites(cipherSuites);
- LOG.debug("Enabled cipher suites {}",
- Arrays.toString(engine.getEnabledCipherSuites()));
+ if (protocols != null && protocols.length > 0) {
+ //Set supported protocols
+ engine.setEnabledProtocols(protocols);
+ LOG.debug("Supported ssl protocols {}",
+ Arrays.toString(engine.getSupportedProtocols()));
+ LOG.debug("Enabled ssl protocols {}",
+ Arrays.toString(engine.getEnabledProtocols()));
+ }
+ if (cipherSuites != null && cipherSuites.length > 0) {
+ //Set supported cipher suites
+ engine.setEnabledCipherSuites(cipherSuites);
+ LOG.debug("Enabled cipher suites {}",
+ Arrays.toString(engine.getEnabledCipherSuites()));
+ }
channel.pipeline().addLast("ssl", new SslHandler(engine));
}
channel.pipeline().addLast(
- new JsonRpcDecoder(100000),
+ new JsonRpcDecoder(jsonRpcDecoderMaxFrameLength),
new StringEncoder(CharsetUtil.UTF_8),
new IdleStateHandler(IDLE_READER_TIMEOUT, 0, 0),
new ReadTimeoutHandler(READ_TIMEOUT),
- new ExceptionHandler());
+ new ExceptionHandler(OvsdbConnectionService.this));
handleNewPassiveConnection(channel);
}
serverBootstrap.option(ChannelOption.RCVBUF_ALLOCATOR,
new AdaptiveRecvByteBufAllocator(65535, 65535, 65535));
// Start the server.
- ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
+ ChannelFuture channelFuture = serverBootstrap.bind(ip, port).sync();
Channel serverListenChannel = channelFuture.channel();
+ serverChannel = serverListenChannel;
// Wait until the server socket is closed.
serverListenChannel.closeFuture().sync();
} catch (InterruptedException e) {
LOG.error("Thread interrupted", e);
+ } catch (Throwable throwable) {
+ // sync() re-throws exceptions declared as Throwable, so the compiler doesn't see them
+ LOG.error("Error while binding to address {}, port {}", ip, port, throwable);
+ throw throwable;
} finally {
// Shut down all event loops to terminate all threads.
bossGroup.shutdownGracefully();
LOG.error("Probe failed to OVSDB switch. Disconnecting the channel {}", client.getConnectionInfo());
client.disconnect();
}
- }, connectionNotifierService);
+ }, CONNECTION_NOTIFIER_SERVICE);
}
private static void handleNewPassiveConnection(final Channel channel) {
SslHandler sslHandler = (SslHandler) channel.pipeline().get("ssl");
if (sslHandler != null) {
class HandleNewPassiveSslRunner implements Runnable {
- public SslHandler sslHandler;
- public final Channel channel;
- private int retryTimes;
-
- HandleNewPassiveSslRunner(Channel channel, SslHandler sslHandler) {
- this.channel = channel;
- this.sslHandler = sslHandler;
- this.retryTimes = 3;
+ private int retryTimes = 3;
+
+ private void retry() {
+ if (retryTimes > 0) {
+ EXECUTOR_SERVICE.schedule(this, RETRY_PERIOD, TimeUnit.MILLISECONDS);
+ } else {
+ LOG.debug("channel closed {}", channel);
+ channel.disconnect();
+ }
+ retryTimes--;
}
@Override
.equals("SSL_NULL_WITH_NULL_NULL")) {
// Not begin handshake yet. Retry later.
LOG.debug("handshake not begin yet {}", status);
- executorService.schedule(this, retryPeriod, TimeUnit.MILLISECONDS);
+ retry();
} else {
//Check if peer is trusted before notifying listeners
try {
} catch (SSLPeerUnverifiedException e) {
//Trust manager is still checking peer certificate. Retry later
LOG.debug("Peer certifiacte is not verified yet {}", status);
- executorService.schedule(this, retryPeriod, TimeUnit.MILLISECONDS);
+ retry();
}
}
break;
case NEED_TASK:
//Handshake still ongoing. Retry later.
LOG.debug("handshake not done yet {}", status);
- executorService.schedule(this, retryPeriod, TimeUnit.MILLISECONDS);
+ retry();
break;
case NEED_WRAP:
.equals("SSL_NULL_WITH_NULL_NULL")) {
/* peer not authenticated. No need to notify listener in this case. */
LOG.error("Ssl handshake fail. channel {}", channel);
+ channel.disconnect();
} else {
/*
* peer is authenticated. Give some time to wait for completion.
* since client will reconnect later.
*/
LOG.debug("handshake not done yet {}", status);
- if (retryTimes > 0) {
- executorService.schedule(this, retryPeriod, TimeUnit.MILLISECONDS);
- } else {
- LOG.debug("channel closed {}", channel);
- }
- retryTimes--;
+ retry();
}
break;
}
}
- executorService.schedule(new HandleNewPassiveSslRunner(channel, sslHandler),
- retryPeriod, TimeUnit.MILLISECONDS);
+ EXECUTOR_SERVICE.schedule(new HandleNewPassiveSslRunner(),
+ RETRY_PERIOD, TimeUnit.MILLISECONDS);
} else {
- executorService.execute(new Runnable() {
- @Override
- public void run() {
- OvsdbClient client = getChannelClient(channel, ConnectionType.PASSIVE,
- SocketConnectionType.NON_SSL);
- handleNewPassiveConnection(client);
- }
+ EXECUTOR_SERVICE.execute(() -> {
+ OvsdbClient client = getChannelClient(channel, ConnectionType.PASSIVE,
+ SocketConnectionType.NON_SSL);
+ handleNewPassiveConnection(client);
});
}
}
public static void channelClosed(final OvsdbClient client) {
- LOG.info("Connection closed {}", client.getConnectionInfo().toString());
- connections.remove(client);
+ LOG.info("Connection closed {}", client.getConnectionInfo());
+ CONNECTIONS.remove(client);
if (client.isConnectionPublished()) {
- for (OvsdbConnectionListener listener : connectionListeners) {
+ for (OvsdbConnectionListener listener : CONNECTION_LISTENERS) {
listener.disconnected(client);
}
}
@Override
public Collection<OvsdbClient> getConnections() {
- return connections.keySet();
+ return CONNECTIONS.keySet();
}
@Override
public void close() throws Exception {
LOG.info("OvsdbConnectionService closed");
+ JsonRpcEndpoint.close();
}
@Override
public OvsdbClient getClient(Channel channel) {
- for (OvsdbClient client : connections.keySet()) {
- Channel ctx = connections.get(client);
+ for (Entry<OvsdbClient, Channel> entry : CONNECTIONS.entrySet()) {
+ OvsdbClient client = entry.getKey();
+ Channel ctx = entry.getValue();
if (ctx.equals(channel)) {
return client;
}
private static List<OvsdbClient> getPassiveClientsFromSameNode(OvsdbClient ovsdbClient) {
List<OvsdbClient> passiveClients = new ArrayList<>();
- for (OvsdbClient client : connections.keySet()) {
+ for (OvsdbClient client : CONNECTIONS.keySet()) {
if (!client.equals(ovsdbClient)
&& client.getConnectionInfo().getRemoteAddress()
.equals(ovsdbClient.getConnectionInfo().getRemoteAddress())
public static void notifyListenerForPassiveConnection(final OvsdbClient client) {
client.setConnectionPublished(true);
- for (final OvsdbConnectionListener listener : connectionListeners) {
- connectionNotifierService.submit(new Runnable() {
- @Override
- public void run() {
- LOG.trace("Connection {} notified to listener {}", client.getConnectionInfo(), listener);
- listener.connected(client);
- }
+ for (final OvsdbConnectionListener listener : CONNECTION_LISTENERS) {
+ CONNECTION_NOTIFIER_SERVICE.execute(() -> {
+ LOG.trace("Connection {} notified to listener {}", client.getConnectionInfo(), listener);
+ listener.connected(client);
});
}
}
+ public void setOvsdbRpcTaskTimeout(int timeout) {
+ JsonRpcEndpoint.setReaperInterval(timeout);
+ }
+
+ /**
+ * Set useSSL flag.
+ *
+ * @param flag boolean for using ssl
+ */
+ public void setUseSsl(boolean flag) {
+ useSSL = flag;
+ }
+
+ /**
+ * Blueprint property setter method. Blueprint call this method and set the value of json rpc decoder
+ * max frame length to the value configured for config option (json-rpc-decoder-max-frame-length) in
+ * the configuration file. This option is only configured at the boot time of the controller. Any
+ * change at the run time will have no impact.
+ * @param maxFrameLength Max frame length (default : 100000)
+ */
+ public void setJsonRpcDecoderMaxFrameLength(int maxFrameLength) {
+ jsonRpcDecoderMaxFrameLength = maxFrameLength;
+ LOG.info("Json Rpc Decoder Max Frame Length set to : {}", jsonRpcDecoderMaxFrameLength);
+ }
+
+ public void setOvsdbListenerIp(String ip) {
+ LOG.info("OVSDB IP for listening connection is set to : {}", ip);
+ listenerIp = ip;
+ }
+
+ public void setOvsdbListenerPort(int portNumber) {
+ LOG.info("OVSDB port for listening connection is set to : {}", portNumber);
+ listenerPort = portNumber;
+ }
+
public void updateConfigParameter(Map<String, Object> configParameters) {
- LOG.debug("Config parameters received : {}", configParameters.entrySet());
if (configParameters != null && !configParameters.isEmpty()) {
+ LOG.debug("Config parameters received : {}", configParameters.entrySet());
for (Map.Entry<String, Object> paramEntry : configParameters.entrySet()) {
if (paramEntry.getKey().equalsIgnoreCase(OVSDB_RPC_TASK_TIMEOUT_PARAM)) {
- JsonRpcEndpoint.setReaperInterval(Integer.parseInt((String)paramEntry.getValue()));
-
- //Please remove the break if you add more config nobs.
- break;
+ setOvsdbRpcTaskTimeout(Integer.parseInt((String)paramEntry.getValue()));
+ } else if (paramEntry.getKey().equalsIgnoreCase(USE_SSL)) {
+ useSSL = Boolean.parseBoolean(paramEntry.getValue().toString());
}
}
}
Code Review / ovsdb.git / blobdiff