This is an automated email from the ASF dual-hosted git repository. haonan pushed a commit to branch ssl_between_nodes in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit f5438c6f1ee835752dc7c5cec5519ab75650b865 Author: HTHou <[email protected]> AuthorDate: Mon Jun 16 18:44:22 2025 +0800 Support ratis --- .../apache/iotdb/consensus/config/RatisConfig.java | 49 +++++++++++++++++++++- .../iotdb/consensus/ratis/RatisConsensus.java | 4 +- .../apache/iotdb/consensus/ratis/utils/Utils.java | 45 +++++++++++++++++++- 3 files changed, 93 insertions(+), 5 deletions(-) diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/config/RatisConfig.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/config/RatisConfig.java index 3bb4e64e490..f5d8aa48fec 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/config/RatisConfig.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/config/RatisConfig.java @@ -704,18 +704,33 @@ public class RatisConfig { private final boolean asyncRequestThreadPoolCached; private final int asyncRequestThreadPoolSize; private final int leaderOutstandingAppendsMax; + private final boolean isEnableSSL; + private final String sslTrustStorePath; + private final String sslTrustStorePassword; + private final String sslKeyStorePath; + private final String sslKeyStorePassword; private Grpc( SizeInBytes messageSizeMax, SizeInBytes flowControlWindow, boolean asyncRequestThreadPoolCached, int asyncRequestThreadPoolSize, - int leaderOutstandingAppendsMax) { + int leaderOutstandingAppendsMax, + boolean isEnableSSL, + String sslTrustStorePath, + String sslTrustStorePassword, + String sslKeyStorePath, + String sslKeyStorePassword) { this.messageSizeMax = messageSizeMax; this.flowControlWindow = flowControlWindow; this.asyncRequestThreadPoolCached = asyncRequestThreadPoolCached; this.asyncRequestThreadPoolSize = asyncRequestThreadPoolSize; this.leaderOutstandingAppendsMax = leaderOutstandingAppendsMax; + this.isEnableSSL = isEnableSSL; + this.sslTrustStorePath = sslTrustStorePath; + this.sslTrustStorePassword = sslTrustStorePassword; + this.sslKeyStorePath = sslKeyStorePath; + this.sslKeyStorePassword = sslKeyStorePassword; } public SizeInBytes getMessageSizeMax() { @@ -738,6 +753,26 @@ public class RatisConfig { return leaderOutstandingAppendsMax; } + public boolean isEnableSSL() { + return isEnableSSL; + } + + public String getSslTrustStorePath() { + return sslTrustStorePath; + } + + public String getSslTrustStorePassword() { + return sslTrustStorePassword; + } + + public String getSslKeyStorePath() { + return sslKeyStorePath; + } + + public String getSslKeyStorePassword() { + return sslKeyStorePassword; + } + public static Grpc.Builder newBuilder() { return new Grpc.Builder(); } @@ -750,6 +785,11 @@ public class RatisConfig { Server.ASYNC_REQUEST_THREAD_POOL_CACHED_DEFAULT; private int asyncRequestThreadPoolSize = Server.ASYNC_REQUEST_THREAD_POOL_SIZE_DEFAULT; private int leaderOutstandingAppendsMax = Server.LEADER_OUTSTANDING_APPENDS_MAX_DEFAULT; + private boolean isEnableSSL = false; + private String sslTrustStorePath = ""; + private String sslTrustStorePassword = ""; + private String sslKeyStorePath = ""; + private String sslKeyStorePassword = ""; public Grpc build() { return new Grpc( @@ -757,7 +797,12 @@ public class RatisConfig { flowControlWindow, asyncRequestThreadPoolCached, asyncRequestThreadPoolSize, - leaderOutstandingAppendsMax); + leaderOutstandingAppendsMax, + isEnableSSL, + sslTrustStorePath, + sslTrustStorePassword, + sslKeyStorePath, + sslKeyStorePassword); } public Grpc.Builder setMessageSizeMax(SizeInBytes messageSizeMax) { diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java index ff16434df14..96779812c0a 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java @@ -158,7 +158,7 @@ class RatisConsensus implements IConsensus { RaftServerConfigKeys.setStorageDir(properties, Collections.singletonList(storageDir)); GrpcConfigKeys.Server.setPort(properties, config.getThisNodeEndPoint().getPort()); - Utils.initRatisConfig(properties, config.getRatisConfig()); + Parameters parameters = Utils.initRatisConfig(properties, config.getRatisConfig()); this.config = config.getRatisConfig(); this.readOption = this.config.getRead().getReadOption(); this.canServeStaleRead = @@ -211,7 +211,7 @@ class RatisConsensus implements IConsensus { new IClientManager.Factory<RaftGroup, RatisClient>() .createClientManager(new RatisClientPoolFactory(true)); - clientRpc = new GrpcFactory(new Parameters()).newRaftClientRpc(ClientId.randomId(), properties); + clientRpc = new GrpcFactory(parameters).newRaftClientRpc(ClientId.randomId(), properties); // do not build server in constructor in case stateMachine is not ready server = diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/utils/Utils.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/utils/Utils.java index 07dac6f8cf2..60ed5a87f10 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/utils/Utils.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/utils/Utils.java @@ -30,8 +30,10 @@ import org.apache.iotdb.consensus.config.RatisConfig; import org.apache.iotdb.rpc.AutoScalingBufferWriteTransport; import org.apache.ratis.client.RaftClientConfigKeys; +import org.apache.ratis.conf.Parameters; import org.apache.ratis.conf.RaftProperties; import org.apache.ratis.grpc.GrpcConfigKeys; +import org.apache.ratis.grpc.GrpcTlsConfig; import org.apache.ratis.proto.RaftProtos.RaftPeerProto; import org.apache.ratis.protocol.RaftGroupId; import org.apache.ratis.protocol.RaftPeer; @@ -47,8 +49,16 @@ import org.apache.thrift.TException; import org.apache.thrift.protocol.TCompactProtocol; import org.apache.thrift.transport.TByteBuffer; +import javax.net.ssl.KeyManager; +import javax.net.ssl.KeyManagerFactory; +import javax.net.ssl.TrustManager; +import javax.net.ssl.TrustManagerFactory; + import java.io.File; import java.nio.ByteBuffer; +import java.nio.file.Files; +import java.nio.file.Paths; +import java.security.KeyStore; import java.util.List; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; @@ -243,9 +253,10 @@ public class Utils { return TimeDuration.valueOf(maxWaitMs, TimeUnit.MILLISECONDS); } - public static void initRatisConfig(RaftProperties properties, RatisConfig config) { + public static Parameters initRatisConfig(RaftProperties properties, RatisConfig config) { GrpcConfigKeys.setMessageSizeMax(properties, config.getGrpc().getMessageSizeMax()); GrpcConfigKeys.setFlowControlWindow(properties, config.getGrpc().getFlowControlWindow()); + GrpcConfigKeys.Server.setAsyncRequestThreadPoolCached( properties, config.getGrpc().isAsyncRequestThreadPoolCached()); GrpcConfigKeys.Server.setAsyncRequestThreadPoolSize( @@ -345,6 +356,38 @@ public class Utils { final TimeDuration clientMaxRetryGap = getMaxRetrySleepTime(config.getClient()); RaftServerConfigKeys.RetryCache.setExpiryTime(properties, clientMaxRetryGap); + + Parameters parameters = new Parameters(); + if (config.getGrpc().isEnableSSL()) { + String keyStorePath = config.getGrpc().getSslKeyStorePath(); + String keyStorePassword = config.getGrpc().getSslKeyStorePassword(); + String trustStorePath = config.getGrpc().getSslTrustStorePath(); + String trustStorePassword = config.getGrpc().getSslTrustStorePassword(); + try { + // === 1) create KeyManager === + KeyStore keyStore = KeyStore.getInstance("JKS"); + keyStore.load( + Files.newInputStream(Paths.get(keyStorePath)), keyStorePassword.toCharArray()); + + KeyManagerFactory kmf = + KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm()); + kmf.init(keyStore, keyStorePassword.toCharArray()); + KeyManager keyManager = kmf.getKeyManagers()[0]; + + // === 2) create TrustManager === + KeyStore trustStore = KeyStore.getInstance("JKS"); + trustStore.load( + Files.newInputStream(Paths.get(trustStorePath)), trustStorePassword.toCharArray()); + + TrustManagerFactory tmf = + TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm()); + tmf.init(trustStore); + TrustManager trustManager = tmf.getTrustManagers()[0]; + GrpcConfigKeys.TLS.setConf(parameters, new GrpcTlsConfig(keyManager, trustManager, true)); + } catch (Exception ignored) { + } + } + return parameters; } public static boolean anyOf(BooleanSupplier... conditions) {
