This is an automated email from the ASF dual-hosted git repository.
tanxinyu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 7c5feb13613 datanode's dn_rpc_port supports ssl (#11218)
7c5feb13613 is described below
commit 7c5feb13613f157636caa465bfd676e5bd073f37
Author: CloudWise-Lukemiao
<[email protected]>
AuthorDate: Mon Oct 30 18:59:04 2023 +0800
datanode's dn_rpc_port supports ssl (#11218)
Co-authored-by: Cloudwise_Luke <[email protected]>
---
.../java/org/apache/iotdb/cli/AbstractCli.java | 42 +++++++
.../src/main/java/org/apache/iotdb/cli/Cli.java | 23 +++-
.../main/java/org/apache/iotdb/jdbc/Config.java | 6 +
.../org/apache/iotdb/jdbc/IoTDBConnection.java | 17 ++-
.../apache/iotdb/jdbc/IoTDBConnectionParams.java | 28 +++++
.../src/main/java/org/apache/iotdb/jdbc/Utils.java | 15 +++
.../org/apache/iotdb/rpc/RpcTransportFactory.java | 11 ++
.../java/org/apache/iotdb/session/Session.java | 72 ++++++-----
.../apache/iotdb/session/SessionConnection.java | 27 ++--
.../org/apache/iotdb/session/pool/SessionPool.java | 138 ++++++++++++++++-----
.../resources/conf/iotdb-datanode.properties | 55 +++++++-
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 33 +++++
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 11 ++
.../iotdb/db/conf/rest/IoTDBRestServiceConfig.java | 2 +-
.../org/apache/iotdb/db/service/RPCService.java | 40 ++++--
.../resources/conf/iotdb-common.properties | 46 -------
.../service/AbstractThriftServiceThread.java | 36 ++++++
.../iotdb/commons/service/ThriftServiceThread.java | 28 +++++
18 files changed, 496 insertions(+), 134 deletions(-)
diff --git
a/iotdb-client/cli/src/main/java/org/apache/iotdb/cli/AbstractCli.java
b/iotdb-client/cli/src/main/java/org/apache/iotdb/cli/AbstractCli.java
index b9663cb3514..12816536eb7 100644
--- a/iotdb-client/cli/src/main/java/org/apache/iotdb/cli/AbstractCli.java
+++ b/iotdb-client/cli/src/main/java/org/apache/iotdb/cli/AbstractCli.java
@@ -70,7 +70,18 @@ public abstract class AbstractCli {
static final String USERNAME_NAME = "username";
private static final String EXECUTE_ARGS = "e";
+
+ static final String USE_SSL_ARGS = "usessl";
+ static final String TRUST_STORE_ARGS = "ts";
+
+ static final String TRUST_STORE_PWD_ARGS = "tpw";
+
private static final String EXECUTE_NAME = "execute";
+
+ private static final String USE_SSL = "use_ssl";
+ private static final String TRUST_STORE = "trust_store";
+
+ private static final String TRUST_STORE_PWD = "trust_store_pwd";
private static final String NULL = "null";
static final int CODE_OK = 0;
@@ -113,6 +124,10 @@ public abstract class AbstractCli {
static String port = "6667";
static String username;
static String password;
+ static String useSsl;
+ static String trustStore;
+ static String trustStorePwd;
+
static String execute;
static boolean hasExecuteSQL = false;
@@ -130,6 +145,9 @@ public abstract class AbstractCli {
keywordSet.add("-" + PORT_ARGS);
keywordSet.add("-" + PW_ARGS);
keywordSet.add("-" + USERNAME_ARGS);
+ keywordSet.add("-" + USE_SSL_ARGS);
+ keywordSet.add("-" + TRUST_STORE_ARGS);
+ keywordSet.add("-" + TRUST_STORE_PWD_ARGS);
keywordSet.add("-" + EXECUTE_ARGS);
keywordSet.add("-" + ISO8601_ARGS);
keywordSet.add("-" + RPC_COMPRESS_ARGS);
@@ -174,6 +192,30 @@ public abstract class AbstractCli {
Option.builder(PW_ARGS).argName(PW_NAME).hasArg().desc("password
(optional)").build();
options.addOption(password);
+ Option useSSL =
+ Option.builder(USE_SSL_ARGS)
+ .argName(USE_SSL)
+ .hasArg()
+ .desc("use_ssl statement (optional)")
+ .build();
+ options.addOption(useSSL);
+
+ Option trustStore =
+ Option.builder(TRUST_STORE_ARGS)
+ .argName(TRUST_STORE)
+ .hasArg()
+ .desc("trust_store statement (optional)")
+ .build();
+ options.addOption(trustStore);
+
+ Option trustStorePwd =
+ Option.builder(TRUST_STORE_PWD_ARGS)
+ .argName(TRUST_STORE_PWD)
+ .hasArg()
+ .desc("trust_store_pwd statement (optional)")
+ .build();
+ options.addOption(trustStorePwd);
+
Option execute =
Option.builder(EXECUTE_ARGS)
.argName(EXECUTE_NAME)
diff --git a/iotdb-client/cli/src/main/java/org/apache/iotdb/cli/Cli.java
b/iotdb-client/cli/src/main/java/org/apache/iotdb/cli/Cli.java
index 54c7f223991..9d433231c91 100644
--- a/iotdb-client/cli/src/main/java/org/apache/iotdb/cli/Cli.java
+++ b/iotdb-client/cli/src/main/java/org/apache/iotdb/cli/Cli.java
@@ -39,6 +39,7 @@ import org.jline.reader.UserInterruptException;
import java.io.IOException;
import java.sql.DriverManager;
import java.sql.SQLException;
+import java.util.Properties;
import static org.apache.iotdb.cli.utils.IoTPrinter.println;
import static org.apache.iotdb.jdbc.Config.IOTDB_ERROR_PREFIX;
@@ -48,6 +49,8 @@ public class Cli extends AbstractCli {
private static CommandLine commandLine;
private static LineReader lineReader;
+ private static Properties info = new Properties();
+
/**
* IoTDB Client main function.
*
@@ -93,6 +96,16 @@ public class Cli extends AbstractCli {
serve();
}
+ private static void constructProperties() {
+ if (useSsl != null && Boolean.parseBoolean(useSsl)) {
+ info.setProperty("use_ssl", useSsl);
+ info.setProperty("trust_store", trustStore);
+ info.setProperty("trust_store_pwd", trustStorePwd);
+ }
+ info.setProperty("user", username);
+ info.setProperty("password", password);
+ }
+
private static boolean parseCommandLine(Options options, String[] newArgs,
HelpFormatter hf) {
try {
CommandLineParser parser = new DefaultParser();
@@ -128,7 +141,11 @@ public class Cli extends AbstractCli {
private static void serve() {
try {
+ useSsl = commandLine.getOptionValue(USE_SSL_ARGS);
+ trustStore = commandLine.getOptionValue(TRUST_STORE_ARGS);
+ trustStorePwd = commandLine.getOptionValue(TRUST_STORE_PWD_ARGS);
password = commandLine.getOptionValue(PW_ARGS);
+ constructProperties();
if (hasExecuteSQL && password != null) {
executeSql();
}
@@ -145,8 +162,7 @@ public class Cli extends AbstractCli {
private static void executeSql() throws TException {
try (IoTDBConnection connection =
(IoTDBConnection)
- DriverManager.getConnection(
- Config.IOTDB_URL_PREFIX + host + ":" + port + "/", username,
password)) {
+ DriverManager.getConnection(Config.IOTDB_URL_PREFIX + host + ":" +
port + "/", info)) {
connection.setQueryTimeout(queryTimeout);
properties = connection.getServerProperties();
timestampPrecision = properties.getTimestampPrecision();
@@ -162,8 +178,7 @@ public class Cli extends AbstractCli {
private static void receiveCommands(LineReader reader) throws TException {
try (IoTDBConnection connection =
(IoTDBConnection)
- DriverManager.getConnection(
- Config.IOTDB_URL_PREFIX + host + ":" + port + "/", username,
password)) {
+ DriverManager.getConnection(Config.IOTDB_URL_PREFIX + host + ":" +
port + "/", info)) {
connection.setQueryTimeout(queryTimeout);
properties = connection.getServerProperties();
AGGREGRATE_TIME_LIST.addAll(properties.getSupportedTimeAggregationOperations());
diff --git a/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/Config.java
b/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/Config.java
index 9728fb47239..1e98bc56f2e 100644
--- a/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/Config.java
+++ b/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/Config.java
@@ -69,4 +69,10 @@ public class Config {
/** Key of connection's time zone. */
public static final String TIME_ZONE = "time_zone";
+
+ public static final String USE_SSL = "use_ssl";
+
+ public static final String TRUST_STORE = "trust_store";
+
+ public static final String TRUST_STORE_PWD = "trust_store_pwd";
}
diff --git
a/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBConnection.java
b/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBConnection.java
index c5db129216f..bec0678db82 100644
--- a/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBConnection.java
+++ b/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBConnection.java
@@ -469,9 +469,20 @@ public class IoTDBConnection implements Connection {
private void openTransport() throws TTransportException {
RpcTransportFactory.setDefaultBufferCapacity(params.getThriftDefaultBufferSize());
RpcTransportFactory.setThriftMaxFrameSize(params.getThriftMaxFrameSize());
- transport =
- RpcTransportFactory.INSTANCE.getTransport(
- params.getHost(), params.getPort(), getNetworkTimeout());
+
+ if (params.isUseSSL()) {
+ transport =
+ RpcTransportFactory.INSTANCE.getTransport(
+ params.getHost(),
+ params.getPort(),
+ getNetworkTimeout(),
+ params.getTrustStore(),
+ params.getTrustStorePwd());
+ } else {
+ transport =
+ RpcTransportFactory.INSTANCE.getTransport(
+ params.getHost(), params.getPort(), getNetworkTimeout());
+ }
if (!transport.isOpen()) {
transport.open();
}
diff --git
a/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBConnectionParams.java
b/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBConnectionParams.java
index b820600adad..d1523691f79 100644
---
a/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBConnectionParams.java
+++
b/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBConnectionParams.java
@@ -41,6 +41,10 @@ public class IoTDBConnectionParams {
private String timeZone = ZoneId.systemDefault().toString();
+ private boolean useSSL = false;
+ private String trustStore;
+ private String trustStorePwd;
+
public IoTDBConnectionParams(String url) {
this.jdbcUriString = url;
}
@@ -136,4 +140,28 @@ public class IoTDBConnectionParams {
public String getTimeZone() {
return this.timeZone;
}
+
+ public boolean isUseSSL() {
+ return useSSL;
+ }
+
+ public void setUseSSL(boolean useSSL) {
+ this.useSSL = useSSL;
+ }
+
+ public String getTrustStore() {
+ return trustStore;
+ }
+
+ public void setTrustStore(String trustStore) {
+ this.trustStore = trustStore;
+ }
+
+ public String getTrustStorePwd() {
+ return trustStorePwd;
+ }
+
+ public void setTrustStorePwd(String trustStorePwd) {
+ this.trustStorePwd = trustStorePwd;
+ }
}
diff --git a/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/Utils.java
b/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/Utils.java
index 6200e79e5fa..c40b057bfda 100644
--- a/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/Utils.java
+++ b/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/Utils.java
@@ -86,6 +86,18 @@ public class Utils {
params.setTimeZone(info.getProperty(Config.TIME_ZONE));
}
+ if (info.containsKey(Config.USE_SSL)) {
+ params.setUseSSL(Boolean.parseBoolean(info.getProperty(Config.USE_SSL)));
+ }
+
+ if (info.containsKey(Config.TRUST_STORE)) {
+ params.setTrustStore(info.getProperty(Config.TRUST_STORE));
+ }
+
+ if (info.containsKey(Config.TRUST_STORE_PWD)) {
+ params.setTrustStorePwd(info.getProperty(Config.TRUST_STORE_PWD));
+ }
+
return params;
}
@@ -118,6 +130,9 @@ public class Utils {
return false;
}
break;
+ case Config.USE_SSL:
+ case Config.TRUST_STORE:
+ case Config.TRUST_STORE_PWD:
case Config.VERSION:
case Config.NETWORK_TIMEOUT:
info.put(key, value);
diff --git
a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/RpcTransportFactory.java
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/RpcTransportFactory.java
index c7aeef8ec36..191f0a42f95 100644
---
a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/RpcTransportFactory.java
+++
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/RpcTransportFactory.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.rpc;
import org.apache.thrift.transport.TMemoryInputTransport;
+import org.apache.thrift.transport.TSSLTransportFactory;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;
@@ -78,6 +79,16 @@ public class RpcTransportFactory extends TTransportFactory {
return inner.getTransport(new
TSocket(TConfigurationConst.defaultTConfiguration, ip, port));
}
+ public TTransport getTransport(
+ String ip, int port, int timeout, String trustStore, String
trustStorePwd)
+ throws TTransportException {
+ TSSLTransportFactory.TSSLTransportParameters params =
+ new TSSLTransportFactory.TSSLTransportParameters();
+ params.setTrustStore(trustStore, trustStorePwd);
+ TTransport transport = TSSLTransportFactory.getClientSocket(ip, port,
timeout, params);
+ return inner.getTransport(transport);
+ }
+
public TTransport getTransport(String ip, int port, int timeout) throws
TTransportException {
return inner.getTransport(
new TSocket(TConfigurationConst.defaultTConfiguration, ip, port,
timeout));
diff --git
a/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java
b/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java
index 54f6d718101..4fcef0e0c03 100644
--- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java
+++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java
@@ -117,6 +117,9 @@ public class Session implements ISession {
protected String username;
protected String password;
protected int fetchSize;
+ protected boolean useSSL;
+ protected String trustStore;
+ protected String trustStorePwd;
/**
* Timeout of query can be set by users. A negative number means using the
default configuration
* of server. And value 0 will disable the function of query timeout.
@@ -381,6 +384,26 @@ public class Session implements ISession {
this.version = version;
}
+ public Session(Builder builder) {
+ if (builder.nodeUrls != null && builder.nodeUrls.size() > 0) {
+ this.nodeUrls = builder.nodeUrls;
+ this.enableRedirection = true;
+ } else {
+ this.defaultEndPoint = new TEndPoint(builder.host, builder.rpcPort);
+ this.enableRedirection = builder.enableRedirection;
+ }
+ this.username = builder.username;
+ this.password = builder.pw;
+ this.fetchSize = builder.fetchSize;
+ this.zoneId = builder.zoneId;
+ this.thriftDefaultBufferSize = builder.thriftDefaultBufferSize;
+ this.thriftMaxFrameSize = builder.thriftMaxFrameSize;
+ this.version = builder.version;
+ this.useSSL = builder.useSSL;
+ this.trustStore = builder.trustStore;
+ this.trustStorePwd = builder.trustStorePwd;
+ }
+
@Override
public void setFetchSize(int fetchSize) {
this.fetchSize = fetchSize;
@@ -3453,6 +3476,25 @@ public class Session implements ISession {
private Version version = SessionConfig.DEFAULT_VERSION;
private long timeOut = SessionConfig.DEFAULT_QUERY_TIME_OUT;
+ private boolean useSSL = false;
+ private String trustStore;
+ private String trustStorePwd;
+
+ public Builder useSSL(boolean useSSL) {
+ this.useSSL = useSSL;
+ return this;
+ }
+
+ public Builder trustStore(String keyStore) {
+ this.trustStore = keyStore;
+ return this;
+ }
+
+ public Builder trustStorePwd(String keyStorePwd) {
+ this.trustStorePwd = keyStorePwd;
+ return this;
+ }
+
private List<String> nodeUrls = null;
public Builder host(String host) {
@@ -3521,34 +3563,8 @@ public class Session implements ISession {
throw new IllegalArgumentException(
"You should specify either nodeUrls or (host + rpcPort), but not
both");
}
-
- if (nodeUrls != null) {
- Session newSession =
- new Session(
- nodeUrls,
- username,
- pw,
- fetchSize,
- zoneId,
- thriftDefaultBufferSize,
- thriftMaxFrameSize,
- enableRedirection,
- version);
- newSession.setEnableQueryRedirection(true);
- return newSession;
- }
-
- return new Session(
- host,
- rpcPort,
- username,
- pw,
- fetchSize,
- zoneId,
- thriftDefaultBufferSize,
- thriftMaxFrameSize,
- enableRedirection,
- version);
+ Session newSession = new Session(this);
+ return newSession;
}
}
}
diff --git
a/iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java
b/iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java
index ef03d825214..417b60b8650 100644
---
a/iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java
+++
b/iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java
@@ -105,7 +105,7 @@ public class SessionConnection {
endPointList.add(endPoint);
this.zoneId = zoneId == null ? ZoneId.systemDefault() : zoneId;
try {
- init(endPoint);
+ init(endPoint, session.useSSL, session.trustStore,
session.trustStorePwd);
} catch (IoTDBConnectionException e) {
throw new IoTDBConnectionException(logForReconnectionFailure());
}
@@ -118,14 +118,25 @@ public class SessionConnection {
initClusterConn();
}
- private void init(TEndPoint endPoint) throws IoTDBConnectionException {
+ private void init(TEndPoint endPoint, boolean useSSL, String trustStore,
String trustStorePwd)
+ throws IoTDBConnectionException {
RpcTransportFactory.setDefaultBufferCapacity(session.thriftDefaultBufferSize);
RpcTransportFactory.setThriftMaxFrameSize(session.thriftMaxFrameSize);
try {
- transport =
- RpcTransportFactory.INSTANCE.getTransport(
- // as there is a try-catch already, we do not need to use
TSocket.wrap
- endPoint.getIp(), endPoint.getPort(),
session.connectionTimeoutInMs);
+ if (useSSL) {
+ transport =
+ RpcTransportFactory.INSTANCE.getTransport(
+ endPoint.getIp(),
+ endPoint.getPort(),
+ session.connectionTimeoutInMs,
+ trustStore,
+ trustStorePwd);
+ } else {
+ transport =
+ RpcTransportFactory.INSTANCE.getTransport(
+ // as there is a try-catch already, we do not need to use
TSocket.wrap
+ endPoint.getIp(), endPoint.getPort(),
session.connectionTimeoutInMs);
+ }
if (!transport.isOpen()) {
transport.open();
}
@@ -180,7 +191,7 @@ public class SessionConnection {
for (TEndPoint tEndPoint : endPointList) {
try {
session.defaultEndPoint = tEndPoint;
- init(tEndPoint);
+ init(tEndPoint, session.useSSL, session.trustStore,
session.trustStorePwd);
} catch (IoTDBConnectionException e) {
if (!reconnect()) {
logger.error("Cluster has no nodes to connect");
@@ -958,7 +969,7 @@ public class SessionConnection {
}
tryHostNum++;
try {
- init(endPoint);
+ init(endPoint, session.useSSL, session.trustStore,
session.trustStorePwd);
connectedSuccess = true;
} catch (IoTDBConnectionException e) {
logger.error("The current node may have been down {},try next
node", endPoint);
diff --git
a/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java
b/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java
index 363351f1283..bd814dcd0bd 100644
---
a/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java
+++
b/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java
@@ -97,6 +97,12 @@ public class SessionPool implements ISessionPool {
private final String user;
private final String password;
private int fetchSize;
+
+ private boolean useSSL;
+
+ private String trustStore;
+
+ private String trustStorePwd;
private ZoneId zoneId;
private boolean enableRedirection;
private boolean enableQueryRedirection = false;
@@ -352,6 +358,48 @@ public class SessionPool implements ISessionPool {
this.formattedNodeUrls = String.format("%s:%s", host, port);
}
+ public SessionPool(
+ String host,
+ int port,
+ String user,
+ String password,
+ int maxSize,
+ int fetchSize,
+ long waitToGetSessionTimeoutInMs,
+ boolean enableCompression,
+ ZoneId zoneId,
+ boolean enableRedirection,
+ int connectionTimeoutInMs,
+ Version version,
+ int thriftDefaultBufferSize,
+ int thriftMaxFrameSize,
+ boolean useSSL,
+ String trustStore,
+ String trustStorePwd) {
+ this.maxSize = maxSize;
+ this.host = host;
+ this.port = port;
+ this.nodeUrls = null;
+ this.user = user;
+ this.password = password;
+ this.fetchSize = fetchSize;
+ this.waitToGetSessionTimeoutInMs = waitToGetSessionTimeoutInMs;
+ this.enableCompression = enableCompression;
+ this.zoneId = zoneId;
+ this.enableRedirection = enableRedirection;
+ if (this.enableRedirection) {
+ deviceIdToEndpoint = new ConcurrentHashMap<>();
+ }
+ this.connectionTimeoutInMs = connectionTimeoutInMs;
+ this.version = version;
+ this.thriftDefaultBufferSize = thriftDefaultBufferSize;
+ this.thriftMaxFrameSize = thriftMaxFrameSize;
+ this.formattedNodeUrls = String.format("%s:%s", host, port);
+ this.useSSL = useSSL;
+ this.trustStore = trustStore;
+ this.trustStorePwd = trustStorePwd;
+ }
+
@SuppressWarnings("squid:S107") // ignore Methods should not have too many
parameters
public SessionPool(
List<String> nodeUrls,
@@ -388,6 +436,38 @@ public class SessionPool implements ISessionPool {
this.formattedNodeUrls = nodeUrls.toString();
}
+ public SessionPool(Builder builder) {
+ this.maxSize = builder.maxSize;
+ this.user = builder.user;
+ this.password = builder.pw;
+ this.fetchSize = builder.fetchSize;
+ this.waitToGetSessionTimeoutInMs = builder.waitToGetSessionTimeoutInMs;
+ this.enableCompression = builder.enableCompression;
+ this.zoneId = builder.zoneId;
+ this.enableRedirection = builder.enableRedirection;
+ if (this.enableRedirection) {
+ deviceIdToEndpoint = new ConcurrentHashMap<>();
+ }
+ this.connectionTimeoutInMs = builder.connectionTimeoutInMs;
+ this.version = builder.version;
+ this.thriftDefaultBufferSize = builder.thriftDefaultBufferSize;
+ this.thriftMaxFrameSize = builder.thriftMaxFrameSize;
+ if (builder.nodeUrls != null && builder.nodeUrls.size() > 0) {
+ this.nodeUrls = builder.nodeUrls;
+ this.host = null;
+ this.port = -1;
+ this.formattedNodeUrls = builder.nodeUrls.toString();
+ } else {
+ this.host = builder.host;
+ this.port = builder.port;
+ this.nodeUrls = null;
+ this.formattedNodeUrls = String.format("%s:%s", host, port);
+ }
+ this.useSSL = builder.useSSL;
+ this.trustStore = builder.trustStore;
+ this.trustStorePwd = builder.trustStorePwd;
+ }
+
private Session constructNewSession() {
Session session;
if (nodeUrls == null) {
@@ -404,6 +484,9 @@ public class SessionPool implements ISessionPool {
.thriftMaxFrameSize(thriftMaxFrameSize)
.enableRedirection(enableRedirection)
.version(version)
+ .useSSL(useSSL)
+ .trustStore(trustStore)
+ .trustStorePwd(trustStorePwd)
.build();
} else {
// Construct redirect-able Session
@@ -418,6 +501,9 @@ public class SessionPool implements ISessionPool {
.thriftMaxFrameSize(thriftMaxFrameSize)
.enableRedirection(enableRedirection)
.version(version)
+ .useSSL(useSSL)
+ .trustStore(trustStore)
+ .trustStorePwd(trustStorePwd)
.build();
}
session.setEnableQueryRedirection(enableQueryRedirection);
@@ -3441,6 +3527,25 @@ public class SessionPool implements ISessionPool {
private int connectionTimeoutInMs =
SessionConfig.DEFAULT_CONNECTION_TIMEOUT_MS;
private Version version = SessionConfig.DEFAULT_VERSION;
+ private boolean useSSL = false;
+ private String trustStore;
+ private String trustStorePwd;
+
+ public Builder useSSL(boolean useSSL) {
+ this.useSSL = useSSL;
+ return this;
+ }
+
+ public Builder trustStore(String keyStore) {
+ this.trustStore = keyStore;
+ return this;
+ }
+
+ public Builder trustStorePwd(String keyStorePwd) {
+ this.trustStorePwd = keyStorePwd;
+ return this;
+ }
+
public Builder host(String host) {
this.host = host;
return this;
@@ -3517,38 +3622,7 @@ public class SessionPool implements ISessionPool {
}
public SessionPool build() {
- if (nodeUrls == null) {
- return new SessionPool(
- host,
- port,
- user,
- pw,
- maxSize,
- fetchSize,
- waitToGetSessionTimeoutInMs,
- enableCompression,
- zoneId,
- enableRedirection,
- connectionTimeoutInMs,
- version,
- thriftDefaultBufferSize,
- thriftMaxFrameSize);
- } else {
- return new SessionPool(
- nodeUrls,
- user,
- pw,
- maxSize,
- fetchSize,
- waitToGetSessionTimeoutInMs,
- enableCompression,
- zoneId,
- enableRedirection,
- connectionTimeoutInMs,
- version,
- thriftDefaultBufferSize,
- thriftMaxFrameSize);
- }
+ return new SessionPool(this);
}
}
}
diff --git
a/iotdb-core/datanode/src/assembly/resources/conf/iotdb-datanode.properties
b/iotdb-core/datanode/src/assembly/resources/conf/iotdb-datanode.properties
index e616df68b70..d27772a66a3 100644
--- a/iotdb-core/datanode/src/assembly/resources/conf/iotdb-datanode.properties
+++ b/iotdb-core/datanode/src/assembly/resources/conf/iotdb-datanode.properties
@@ -60,6 +60,19 @@ dn_data_region_consensus_port=10760
# The time of data node waiting for the next retry to join into the cluster.
# dn_join_cluster_retry_interval_ms=5000
+# Does dn_rpc_port enable SSL
+# enable_thrift_ssl=false
+
+####################
+### SSL Configuration
+####################
+
+# SSL key store path
+# key_store_path=""
+
+# SSL key store password
+# key_store_pwd=""
+
####################
### Seed ConfigNode
####################
@@ -266,4 +279,44 @@ dn_seed_config_node=127.0.0.1:10710
# The type of internal reporter in metric module, used for checking flushed
point number
# Options: [MEMORY, IOTDB]
# Datatype: String
-# dn_metric_internal_reporter_type=MEMORY
\ No newline at end of file
+# dn_metric_internal_reporter_type=MEMORY
+
+####################
+### REST Service Configuration
+####################
+
+# Is the REST service enabled
+# enable_rest_service=false
+
+# the binding port of the REST service
+# rest_service_port=18080
+
+# Whether to display rest service interface information through swagger. eg:
http://ip:port/swagger.json
+# enable_swagger=false
+
+# the default row limit to a REST query response when the rowSize parameter is
not given in request
+# rest_query_default_row_size_limit=10000
+
+# the expiration time of the user login information cache (in seconds)
+# cache_expire_in_seconds=28800
+
+# maximum number of users can be stored in the user login cache.
+# cache_max_num=100
+
+# init capacity of users can be stored in the user login cache.
+# cache_init_num=10
+
+# is SSL enabled
+# enable_https=false
+
+# Is client authentication required
+# client_auth=false
+
+# SSL trust store path
+# trust_store_path=""
+
+# SSL trust store password.
+# trust_store_pwd=""
+
+# SSL timeout (in seconds)
+# idle_timeout_in_seconds=50000
\ No newline at end of file
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
index 15aa68d6d9f..394ba55fe9a 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
@@ -124,6 +124,15 @@ public class IoTDBConfig {
/** Port which the JDBC server listens to. */
private int rpcPort = 6667;
+ /** Enable the thrift rpcPort Service ssl. */
+ private boolean enableSSL = false;
+
+ /** ssl key Store Path. */
+ private String keyStorePath = "";
+
+ /** ssl key Store password. */
+ private String keyStorePwd = "";
+
/** Rpc Selector thread num */
private int rpcSelectorThreadCount = 1;
@@ -1159,6 +1168,30 @@ public class IoTDBConfig {
this.udfCollectorMemoryBudgetInMB = udfCollectorMemoryBudgetInMB;
}
+ public boolean isEnableSSL() {
+ return enableSSL;
+ }
+
+ public void setEnableSSL(boolean enableSSL) {
+ this.enableSSL = enableSSL;
+ }
+
+ public String getKeyStorePath() {
+ return keyStorePath;
+ }
+
+ public void setKeyStorePath(String keyStorePath) {
+ this.keyStorePath = keyStorePath;
+ }
+
+ public String getKeyStorePwd() {
+ return keyStorePwd;
+ }
+
+ public void setKeyStorePwd(String keyStorePwd) {
+ this.keyStorePwd = keyStorePwd;
+ }
+
public int getUdfInitialByteArrayLengthForMemoryControl() {
return udfInitialByteArrayLengthForMemoryControl;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
index e4ab5e2d392..e9c315cd8b1 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
@@ -1011,6 +1011,9 @@ public class IoTDBDescriptor {
// UDF
loadUDFProps(properties);
+ // thrift ssl
+ initThriftSSL(properties);
+
// trigger
loadTriggerProps(properties);
@@ -1949,6 +1952,14 @@ public class IoTDBDescriptor {
}
}
+ private void initThriftSSL(Properties properties) {
+ conf.setEnableSSL(
+ Boolean.parseBoolean(
+ properties.getProperty("enable_thrift_ssl",
Boolean.toString(conf.isEnableSSL()))));
+ conf.setKeyStorePath(properties.getProperty("key_store_path",
conf.getKeyStorePath()).trim());
+ conf.setKeyStorePwd(properties.getProperty("key_store_pwd",
conf.getKeyStorePath()).trim());
+ }
+
private void loadTriggerProps(Properties properties) {
conf.setTriggerDir(properties.getProperty("trigger_lib_dir",
conf.getTriggerDir()));
conf.setRetryNumToFindStatefulTrigger(
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/rest/IoTDBRestServiceConfig.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/rest/IoTDBRestServiceConfig.java
index e376decf442..dd984815e76 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/rest/IoTDBRestServiceConfig.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/rest/IoTDBRestServiceConfig.java
@@ -20,7 +20,7 @@
package org.apache.iotdb.db.conf.rest;
public class IoTDBRestServiceConfig {
- static final String CONFIG_NAME = "iotdb-common.properties";
+ static final String CONFIG_NAME = "iotdb-datanode.properties";
/** If the enableRestService is true, we will start REST Service. */
private boolean enableRestService = false;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/RPCService.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/RPCService.java
index 012602313e8..af747cb1dd0 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/RPCService.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/RPCService.java
@@ -59,17 +59,35 @@ public class RPCService extends ThriftService implements
RPCServiceMBean {
public void initThriftServiceThread() throws IllegalAccessException {
IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
try {
- thriftServiceThread =
- new ThriftServiceThread(
- processor,
- getID().getName(),
- ThreadName.CLIENT_RPC_PROCESSOR.getName(),
- config.getRpcAddress(),
- config.getRpcPort(),
- config.getRpcMaxConcurrentClientNum(),
- config.getThriftServerAwaitTimeForStopService(),
- new RPCServiceThriftHandler(impl),
-
IoTDBDescriptor.getInstance().getConfig().isRpcThriftCompressionEnable());
+ if (config.isEnableSSL()) {
+ thriftServiceThread =
+ new ThriftServiceThread(
+ processor,
+ getID().getName(),
+ ThreadName.CLIENT_RPC_PROCESSOR.getName(),
+ config.getRpcAddress(),
+ config.getRpcPort(),
+ config.getRpcMaxConcurrentClientNum(),
+ config.getThriftServerAwaitTimeForStopService(),
+ new RPCServiceThriftHandler(impl),
+
IoTDBDescriptor.getInstance().getConfig().isRpcThriftCompressionEnable(),
+ config.getKeyStorePath(),
+ config.getKeyStorePwd(),
+ config.getConnectionTimeoutInMS());
+ } else {
+ thriftServiceThread =
+ new ThriftServiceThread(
+ processor,
+ getID().getName(),
+ ThreadName.CLIENT_RPC_PROCESSOR.getName(),
+ config.getRpcAddress(),
+ config.getRpcPort(),
+ config.getRpcMaxConcurrentClientNum(),
+ config.getThriftServerAwaitTimeForStopService(),
+ new RPCServiceThriftHandler(impl),
+
IoTDBDescriptor.getInstance().getConfig().isRpcThriftCompressionEnable());
+ }
+
} catch (RPCServiceException e) {
throw new IllegalAccessException(e.getMessage());
}
diff --git
a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-common.properties
b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-common.properties
index 1291424562b..cb10c55e781 100644
---
a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-common.properties
+++
b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-common.properties
@@ -1084,52 +1084,6 @@ timestamp_precision=ms
# Datatype: int
# mqtt_max_message_size=1048576
-####################
-### REST Service Configuration
-####################
-
-# Is the REST service enabled
-# enable_rest_service=false
-
-# the binding port of the REST service
-# rest_service_port=18080
-
-# Whether to display rest service interface information through swagger. eg:
http://ip:port/swagger.json
-# enable_swagger=false
-
-# the default row limit to a REST query response when the rowSize parameter is
not given in request
-# rest_query_default_row_size_limit=10000
-
-# the expiration time of the user login information cache (in seconds)
-# cache_expire_in_seconds=28800
-
-# maximum number of users can be stored in the user login cache.
-# cache_max_num=100
-
-# init capacity of users can be stored in the user login cache.
-# cache_init_num=10
-
-# is SSL enabled
-# enable_https=false
-
-# SSL key store path
-# key_store_path=
-
-# SSL key store password
-# key_store_pwd=
-
-# Is client authentication required
-# client_auth=false
-
-# SSL trust store path
-# trust_store_path=
-
-# SSL trust store password.
-# trust_store_pwd=
-
-# SSL timeout (in seconds)
-# idle_timeout_in_seconds=50000
-
####################
### IoTDB-ML Configuration
####################
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/AbstractThriftServiceThread.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/AbstractThriftServiceThread.java
index 3e315e4b486..12eea072f32 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/AbstractThriftServiceThread.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/AbstractThriftServiceThread.java
@@ -36,6 +36,7 @@ import org.apache.thrift.server.TThreadPoolServer;
import org.apache.thrift.server.TThreadedSelectorServer;
import org.apache.thrift.transport.TNonblockingServerSocket;
import org.apache.thrift.transport.TNonblockingServerTransport;
+import org.apache.thrift.transport.TSSLTransportFactory;
import org.apache.thrift.transport.TServerSocket;
import org.apache.thrift.transport.TServerTransport;
import org.apache.thrift.transport.TTransportException;
@@ -154,6 +155,41 @@ public abstract class AbstractThriftServiceThread extends
Thread {
}
/** for synced ThriftServiceThread */
+ @SuppressWarnings("squid:S107")
+ protected AbstractThriftServiceThread(
+ TProcessor processor,
+ String serviceName,
+ String threadsName,
+ String bindAddress,
+ int port,
+ int maxWorkerThreads,
+ int timeoutSecond,
+ TServerEventHandler serverEventHandler,
+ boolean compress,
+ String keyStorePath,
+ String keyStorePwd,
+ int clientTimeout) {
+ initProtocolFactory(compress);
+ this.serviceName = serviceName;
+
+ try {
+ TSSLTransportFactory.TSSLTransportParameters params =
+ new TSSLTransportFactory.TSSLTransportParameters();
+ params.setKeyStore(keyStorePath, keyStorePwd);
+ params.requireClientAuth(false);
+ InetSocketAddress socketAddress = new InetSocketAddress(bindAddress,
port);
+ serverTransport =
+ TSSLTransportFactory.getServerSocket(
+ socketAddress.getPort(), clientTimeout,
socketAddress.getAddress(), params);
+ TThreadPoolServer.Args poolArgs =
+ initSyncedPoolArgs(processor, threadsName, maxWorkerThreads,
timeoutSecond);
+ poolServer = new TThreadPoolServer(poolArgs);
+ poolServer.setServerEventHandler(serverEventHandler);
+ } catch (TTransportException e) {
+ catchFailedInitialization(e);
+ }
+ }
+
@SuppressWarnings("squid:S107")
protected AbstractThriftServiceThread(
TProcessor processor,
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/ThriftServiceThread.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/ThriftServiceThread.java
index f7c7f05e24b..ad5db5504e5 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/ThriftServiceThread.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/ThriftServiceThread.java
@@ -64,6 +64,34 @@ public class ThriftServiceThread extends
AbstractThriftServiceThread {
/** for synced ThriftServiceThread */
@SuppressWarnings("squid:S107")
+ public ThriftServiceThread(
+ TProcessor processor,
+ String serviceName,
+ String threadsName,
+ String bindAddress,
+ int port,
+ int maxWorkerThreads,
+ int timeoutSecond,
+ TServerEventHandler serverEventHandler,
+ boolean compress,
+ String keyStorePath,
+ String keyStorePwd,
+ int clientTimeout) {
+ super(
+ processor,
+ serviceName,
+ threadsName,
+ bindAddress,
+ port,
+ maxWorkerThreads,
+ timeoutSecond,
+ serverEventHandler,
+ compress,
+ keyStorePath,
+ keyStorePwd,
+ clientTimeout);
+ }
+
public ThriftServiceThread(
TProcessor processor,
String serviceName,
