This is an automated email from the ASF dual-hosted git repository. HTHou pushed a commit to branch codex/trusted-channel-failure-hooks in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit e5d3cc52f940e4e635fe855e254a133c56a661c9 Author: HTHou <[email protected]> AuthorDate: Fri Jul 24 14:38:19 2026 +0800 Add trusted channel failure reporting hooks --- external-service-impl/rest/pom.xml | 4 + .../java/org/apache/iotdb/rest/RestService.java | 4 + .../rest/TrustedChannelAuditHandshakeListener.java | 73 +++++++++++ .../TrustedChannelAuditHandshakeListenerTest.java | 79 ++++++++++++ .../iotdb/confignode/audit/CNAuditLogger.java | 9 ++ .../async/AsyncDataNodeHeartbeatClientPool.java | 13 ++ .../CnToDnInternalServiceAsyncRequestManager.java | 13 ++ .../client/sync/SyncDataNodeClientPool.java | 16 +++ .../service/thrift/ConfigNodeRPCService.java | 8 +- .../iotdb/consensus/config/ConsensusConfig.java | 23 +++- .../apache/iotdb/consensus/iot/IoTConsensus.java | 4 +- .../iot/service/IoTConsensusRPCService.java | 14 ++- .../iotdb/consensus/pipe/IoTConsensusV2.java | 4 +- .../pipe/service/IoTConsensusV2RPCService.java | 14 ++- .../org/apache/iotdb/db/audit/DNAuditLogger.java | 14 +++ .../db/consensus/DataRegionConsensusImpl.java | 3 + .../iotdb/db/protocol/client/ConfigNodeClient.java | 10 ++ .../iotdb/db/protocol/client/an/AINodeClient.java | 11 ++ .../client/dn/AsyncTSStatusRPCHandler.java | 3 + .../execution/exchange/MPPDataExchangeService.java | 7 +- .../execution/exchange/sink/SinkChannel.java | 5 + .../execution/exchange/source/SourceHandle.java | 7 ++ .../plan/scheduler/AsyncPlanNodeSender.java | 3 +- .../plan/scheduler/AsyncSendPlanNodeHandler.java | 8 +- .../scheduler/FragmentInstanceDispatcherImpl.java | 3 + .../db/service/DataNodeInternalRPCService.java | 8 +- .../iotdb/db/service/ExternalRPCService.java | 14 ++- .../apache/iotdb/commons/i18n/CommonMessages.java | 2 + .../apache/iotdb/commons/i18n/CommonMessages.java | 2 + .../iotdb/commons/audit/AbstractAuditLogger.java | 71 +++++++++++ .../apache/iotdb/commons/audit/AuditEventType.java | 2 +- ...Type.java => TrustedChannelFailureHandler.java} | 38 +----- .../client/request/AsyncRequestManager.java | 5 + .../TrustedChannelAuditServerEventHandler.java | 135 +++++++++++++++++++++ .../commons/audit/AbstractAuditLoggerTest.java | 114 +++++++++++++++++ .../TrustedChannelAuditServerEventHandlerTest.java | 121 ++++++++++++++++++ pom.xml | 5 + 37 files changed, 821 insertions(+), 48 deletions(-) diff --git a/external-service-impl/rest/pom.xml b/external-service-impl/rest/pom.xml index f6cedd00796..fc064d28254 100644 --- a/external-service-impl/rest/pom.xml +++ b/external-service-impl/rest/pom.xml @@ -98,6 +98,10 @@ <groupId>org.eclipse.jetty</groupId> <artifactId>jetty-http</artifactId> </dependency> + <dependency> + <groupId>org.eclipse.jetty</groupId> + <artifactId>jetty-io</artifactId> + </dependency> <dependency> <groupId>org.apache.iotdb</groupId> <artifactId>node-commons</artifactId> diff --git a/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/RestService.java b/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/RestService.java index 506facdb4ce..6ecba9aece8 100644 --- a/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/RestService.java +++ b/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/RestService.java @@ -16,6 +16,7 @@ */ package org.apache.iotdb.rest; +import org.apache.iotdb.db.audit.DNAuditLogger; import org.apache.iotdb.db.conf.rest.IoTDBRestServiceConfig; import org.apache.iotdb.db.conf.rest.IoTDBRestServiceDescriptor; import org.apache.iotdb.externalservice.api.IExternalService; @@ -79,6 +80,9 @@ public class RestService implements IExternalService { new HttpConnectionFactory(httpsConfig)); httpsConnector.setPort(port); httpsConnector.setIdleTimeout(idleTime); + httpsConnector.addBean( + new TrustedChannelAuditHandshakeListener( + DNAuditLogger.getInstance()::recordTrustedChannelFailureAuditLogIfNecessary)); server.addConnector(httpsConnector); server.setHandler(constructServletContextHandler()); diff --git a/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/TrustedChannelAuditHandshakeListener.java b/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/TrustedChannelAuditHandshakeListener.java new file mode 100644 index 00000000000..3ba484a5ff7 --- /dev/null +++ b/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/TrustedChannelAuditHandshakeListener.java @@ -0,0 +1,73 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to You under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.iotdb.rest; + +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.commons.audit.TrustedChannelFailureHandler; + +import org.eclipse.jetty.io.EndPoint; +import org.eclipse.jetty.io.ssl.SslHandshakeListener; + +import java.net.InetSocketAddress; +import java.net.SocketAddress; +import java.util.Objects; + +final class TrustedChannelAuditHandshakeListener implements SslHandshakeListener { + + private final TrustedChannelFailureHandler failureHandler; + + TrustedChannelAuditHandshakeListener(TrustedChannelFailureHandler failureHandler) { + this.failureHandler = Objects.requireNonNull(failureHandler); + } + + @Override + public void handshakeFailed(Event event, Throwable failure) { + recordHandshakeFailure(event.getEndPoint(), failure); + } + + void recordHandshakeFailure(EndPoint endPoint, Throwable failure) { + if (endPoint == null) { + return; + } + + TEndPoint initiator = toEndPoint(endPoint.getRemoteSocketAddress()); + TEndPoint target = toEndPoint(endPoint.getLocalSocketAddress()); + if (initiator == null || target == null) { + return; + } + + try { + failureHandler.onFailure(failure, initiator, target); + } catch (RuntimeException auditFailure) { + failure.addSuppressed(auditFailure); + } + } + + private static TEndPoint toEndPoint(SocketAddress socketAddress) { + if (!(socketAddress instanceof InetSocketAddress)) { + return null; + } + InetSocketAddress inetSocketAddress = (InetSocketAddress) socketAddress; + String host = + inetSocketAddress.getAddress() == null + ? inetSocketAddress.getHostString() + : inetSocketAddress.getAddress().getHostAddress(); + return new TEndPoint(host, inetSocketAddress.getPort()); + } +} diff --git a/external-service-impl/rest/src/test/java/org/apache/iotdb/rest/TrustedChannelAuditHandshakeListenerTest.java b/external-service-impl/rest/src/test/java/org/apache/iotdb/rest/TrustedChannelAuditHandshakeListenerTest.java new file mode 100644 index 00000000000..55ce2b7a194 --- /dev/null +++ b/external-service-impl/rest/src/test/java/org/apache/iotdb/rest/TrustedChannelAuditHandshakeListenerTest.java @@ -0,0 +1,79 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.iotdb.rest; + +import org.apache.iotdb.common.rpc.thrift.TEndPoint; + +import org.eclipse.jetty.io.EndPoint; +import org.junit.Test; + +import javax.net.ssl.SSLHandshakeException; + +import java.lang.reflect.Proxy; +import java.net.InetSocketAddress; +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertSame; + +public class TrustedChannelAuditHandshakeListenerTest { + + @Test + public void testRecordSocketEndpointsOnHandshakeFailure() { + AtomicReference<Throwable> actualFailure = new AtomicReference<>(); + AtomicReference<TEndPoint> actualInitiator = new AtomicReference<>(); + AtomicReference<TEndPoint> actualTarget = new AtomicReference<>(); + TrustedChannelAuditHandshakeListener listener = + new TrustedChannelAuditHandshakeListener( + (failure, initiator, target) -> { + actualFailure.set(failure); + actualInitiator.set(initiator); + actualTarget.set(target); + }); + SSLHandshakeException failure = new SSLHandshakeException("test"); + EndPoint endPoint = + newEndPoint( + new InetSocketAddress("192.0.2.10", 45123), new InetSocketAddress("192.0.2.20", 18080)); + + listener.recordHandshakeFailure(endPoint, failure); + + assertSame(failure, actualFailure.get()); + assertEquals(new TEndPoint("192.0.2.10", 45123), actualInitiator.get()); + assertEquals(new TEndPoint("192.0.2.20", 18080), actualTarget.get()); + } + + private static EndPoint newEndPoint( + InetSocketAddress remoteAddress, InetSocketAddress localAddress) { + return (EndPoint) + Proxy.newProxyInstance( + EndPoint.class.getClassLoader(), + new Class<?>[] {EndPoint.class}, + (proxy, method, args) -> { + switch (method.getName()) { + case "getRemoteAddress": + case "getRemoteSocketAddress": + return remoteAddress; + case "getLocalAddress": + case "getLocalSocketAddress": + return localAddress; + default: + return null; + } + }); + } +} diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/audit/CNAuditLogger.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/audit/CNAuditLogger.java index ccc1008eec3..8c739c54c9d 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/audit/CNAuditLogger.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/audit/CNAuditLogger.java @@ -19,8 +19,10 @@ package org.apache.iotdb.confignode.audit; +import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.commons.audit.AbstractAuditLogger; import org.apache.iotdb.commons.audit.IAuditEntity; +import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.confignode.conf.ConfigNodeConfig; import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor; import org.apache.iotdb.confignode.manager.ConfigManager; @@ -43,4 +45,11 @@ public class CNAuditLogger extends AbstractAuditLogger { @Override public void log(IAuditEntity auditLogFields, Supplier<String> log) {} + + public void recordTrustedChannelFailureAuditLogIfNecessary(Throwable failure, TEndPoint target) { + if (CommonDescriptor.getInstance().getConfig().isEnableInternalSSL()) { + recordTrustedChannelFailureAuditLogIfNecessary( + failure, new TEndPoint(CONF.getInternalAddress(), CONF.getInternalPort()), target); + } + } } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/AsyncDataNodeHeartbeatClientPool.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/AsyncDataNodeHeartbeatClientPool.java index 516596a61b8..0c137615986 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/AsyncDataNodeHeartbeatClientPool.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/AsyncDataNodeHeartbeatClientPool.java @@ -27,6 +27,7 @@ import org.apache.iotdb.commons.client.async.AsyncDataNodeInternalServiceClient; import org.apache.iotdb.confignode.client.async.handlers.audit.DataNodeWriteAuditLogHandler; import org.apache.iotdb.confignode.client.async.handlers.heartbeat.DataNodeHeartbeatHandler; import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor; +import org.apache.iotdb.confignode.service.ConfigNode; import org.apache.iotdb.mpp.rpc.thrift.TAuditLogReq; import org.apache.iotdb.mpp.rpc.thrift.TDataNodeHeartbeatReq; @@ -57,6 +58,7 @@ public class AsyncDataNodeHeartbeatClientPool { client.getDataNodeHeartBeat(req, handler); dispatched = true; } catch (Exception e) { + recordTrustedChannelFailureAuditLogIfNecessary(e, endPoint); handleError(handler, e); } finally { returnClientIfNotDispatched(endPoint, client, dispatched); @@ -105,6 +107,17 @@ public class AsyncDataNodeHeartbeatClientPool { } } + private void recordTrustedChannelFailureAuditLogIfNecessary( + Throwable failure, TEndPoint targetEndPoint) { + if (ConfigNode.getInstance() == null || ConfigNode.getInstance().getConfigManager() == null) { + return; + } + ConfigNode.getInstance() + .getConfigManager() + .getAuditLogger() + .recordTrustedChannelFailureAuditLogIfNecessary(failure, targetEndPoint); + } + private static class AsyncDataNodeHeartbeatClientPoolHolder { private static final AsyncDataNodeHeartbeatClientPool INSTANCE = diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnInternalServiceAsyncRequestManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnInternalServiceAsyncRequestManager.java index 9c90c469137..a651fd39b58 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnInternalServiceAsyncRequestManager.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnInternalServiceAsyncRequestManager.java @@ -20,6 +20,7 @@ package org.apache.iotdb.confignode.client.async; import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation; +import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.common.rpc.thrift.TFlushReq; import org.apache.iotdb.common.rpc.thrift.TNodeLocations; import org.apache.iotdb.common.rpc.thrift.TSetConfigurationReq; @@ -50,6 +51,7 @@ import org.apache.iotdb.confignode.client.async.handlers.rpc.subscription.PullCo import org.apache.iotdb.confignode.client.async.handlers.rpc.subscription.TopicPushMetaRPCHandler; import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor; import org.apache.iotdb.confignode.i18n.ConfigNodeMessages; +import org.apache.iotdb.confignode.service.ConfigNode; import org.apache.iotdb.mpp.rpc.thrift.TActiveTriggerInstanceReq; import org.apache.iotdb.mpp.rpc.thrift.TAlterEncodingCompressorReq; import org.apache.iotdb.mpp.rpc.thrift.TAlterTimeSeriesReq; @@ -557,6 +559,17 @@ public class CnToDnInternalServiceAsyncRequestManager } } + @Override + protected void onRequestFailure(Exception failure, TEndPoint targetEndPoint) { + if (ConfigNode.getInstance() == null || ConfigNode.getInstance().getConfigManager() == null) { + return; + } + ConfigNode.getInstance() + .getConfigManager() + .getAuditLogger() + .recordTrustedChannelFailureAuditLogIfNecessary(failure, targetEndPoint); + } + private static class ClientPoolHolder { private static final CnToDnInternalServiceAsyncRequestManager INSTANCE = diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/SyncDataNodeClientPool.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/SyncDataNodeClientPool.java index f35b6d906a5..4611636f51f 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/SyncDataNodeClientPool.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/SyncDataNodeClientPool.java @@ -29,6 +29,7 @@ import org.apache.iotdb.commons.client.exception.ClientManagerException; import org.apache.iotdb.commons.client.sync.SyncDataNodeInternalServiceClient; import org.apache.iotdb.commons.exception.UncheckedStartupException; import org.apache.iotdb.confignode.i18n.ConfigNodeMessages; +import org.apache.iotdb.confignode.service.ConfigNode; import org.apache.iotdb.mpp.rpc.thrift.TCleanDataNodeCacheReq; import org.apache.iotdb.mpp.rpc.thrift.TCreateDataRegionReq; import org.apache.iotdb.mpp.rpc.thrift.TCreatePeerReq; @@ -177,6 +178,7 @@ public class SyncDataNodeClientPool { return executeSyncRequest(requestType, client, req); } catch (Exception e) { lastException = e; + recordTrustedChannelFailureAuditLogIfNecessary(e, endPoint); if (retry != DEFAULT_RETRY_NUM - 1) { LOGGER.warn( ConfigNodeMessages.FAILED_ON_DATANODE_RETRYING, requestType, endPoint, retry + 1); @@ -197,6 +199,7 @@ public class SyncDataNodeClientPool { return executeSyncRequest(requestType, client, req); } catch (Exception e) { lastException = e; + recordTrustedChannelFailureAuditLogIfNecessary(e, endPoint); if (retry != retryNum - 1) { LOGGER.warn( ConfigNodeMessages.FAILED_ON_DATANODE_RETRYING, requestType, endPoint, retry + 1); @@ -248,16 +251,29 @@ public class SyncDataNodeClientPool { return client.changeRegionLeader(req); } catch (ClientManagerException e) { LOGGER.error(ConfigNodeMessages.CAN_T_CONNECT_TO_DATA_NODE, dataNode, e); + recordTrustedChannelFailureAuditLogIfNecessary(e, dataNode); status = new TSStatus(TSStatusCode.CAN_NOT_CONNECT_DATANODE.getStatusCode()); status.setMessage(e.getMessage()); } catch (TException e) { LOGGER.error(ConfigNodeMessages.CHANGE_REGIONS_LEADER_ERROR_ON_DATE_NODE, dataNode, e); + recordTrustedChannelFailureAuditLogIfNecessary(e, dataNode); status = new TSStatus(TSStatusCode.REGION_LEADER_CHANGE_ERROR.getStatusCode()); status.setMessage(e.getMessage()); } return new TRegionLeaderChangeResp(status, -1L); } + private void recordTrustedChannelFailureAuditLogIfNecessary( + Throwable failure, TEndPoint targetEndPoint) { + if (ConfigNode.getInstance() == null || ConfigNode.getInstance().getConfigManager() == null) { + return; + } + ConfigNode.getInstance() + .getConfigManager() + .getAuditLogger() + .recordTrustedChannelFailureAuditLogIfNecessary(failure, targetEndPoint); + } + private static class ClientPoolHolder { private static final SyncDataNodeClientPool INSTANCE = new SyncDataNodeClientPool(); diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCService.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCService.java index 15be2e90d43..f325da85ab0 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCService.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCService.java @@ -18,6 +18,7 @@ */ package org.apache.iotdb.confignode.service.thrift; +import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.commons.concurrent.ThreadName; import org.apache.iotdb.commons.conf.CommonConfig; import org.apache.iotdb.commons.conf.CommonDescriptor; @@ -25,6 +26,7 @@ import org.apache.iotdb.commons.exception.runtime.RPCServiceException; import org.apache.iotdb.commons.service.ServiceType; import org.apache.iotdb.commons.service.ThriftService; import org.apache.iotdb.commons.service.ThriftServiceThread; +import org.apache.iotdb.commons.service.TrustedChannelAuditServerEventHandler; import org.apache.iotdb.commons.service.metric.MetricService; import org.apache.iotdb.confignode.conf.ConfigNodeConfig; import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor; @@ -70,7 +72,11 @@ public class ConfigNodeRPCService extends ThriftService implements ConfigNodeRPC getBindPort(), configConf.getCnRpcMaxConcurrentClientNum(), configConf.getThriftServerAwaitTimeForStopService(), - new ConfigNodeRPCServiceHandler(), + new TrustedChannelAuditServerEventHandler( + new ConfigNodeRPCServiceHandler(), + new TEndPoint(getBindIP(), getBindPort()), + configNodeRPCServiceProcessor.configManager.getAuditLogger() + ::recordTrustedChannelFailureAuditLogIfNecessary), commonConfig.isRpcThriftCompressionEnabled(), commonConfig.getKeyStorePath(), commonConfig.getKeyStorePwd(), diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/config/ConsensusConfig.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/config/ConsensusConfig.java index d54299992fe..114a8aed4fc 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/config/ConsensusConfig.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/config/ConsensusConfig.java @@ -21,6 +21,7 @@ package org.apache.iotdb.consensus.config; import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType; import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.commons.audit.TrustedChannelFailureHandler; import org.apache.iotdb.commons.disk.strategy.DirectoryStrategyType; import java.util.List; @@ -37,6 +38,7 @@ public class ConsensusConfig { private final IoTConsensusConfig iotConsensusConfig; private final IoTConsensusV2Config iotConsensusV2Config; private final DirectoryStrategyType directoryStrategyType; + private final TrustedChannelFailureHandler trustedChannelFailureHandler; private ConsensusConfig( TEndPoint thisNode, @@ -47,7 +49,8 @@ public class ConsensusConfig { RatisConfig ratisConfig, IoTConsensusConfig iotConsensusConfig, IoTConsensusV2Config iotConsensusV2Config, - DirectoryStrategyType directoryStrategyType) { + DirectoryStrategyType directoryStrategyType, + TrustedChannelFailureHandler trustedChannelFailureHandler) { this.thisNodeEndPoint = thisNode; this.thisNodeId = thisNodeId; this.storageDir = storageDir; @@ -57,6 +60,7 @@ public class ConsensusConfig { this.iotConsensusConfig = iotConsensusConfig; this.iotConsensusV2Config = iotConsensusV2Config; this.directoryStrategyType = directoryStrategyType; + this.trustedChannelFailureHandler = trustedChannelFailureHandler; } public TEndPoint getThisNodeEndPoint() { @@ -95,6 +99,10 @@ public class ConsensusConfig { return directoryStrategyType; } + public TrustedChannelFailureHandler getTrustedChannelFailureHandler() { + return trustedChannelFailureHandler; + } + public static ConsensusConfig.Builder newBuilder() { return new ConsensusConfig.Builder(); } @@ -111,6 +119,8 @@ public class ConsensusConfig { private IoTConsensusV2Config iotConsensusV2Config; private DirectoryStrategyType directoryStrategyType = DirectoryStrategyType.MIN_FOLDER_OCCUPIED_SPACE_FIRST_STRATEGY; + private TrustedChannelFailureHandler trustedChannelFailureHandler = + TrustedChannelFailureHandler.NO_OP; public ConsensusConfig build() { return new ConsensusConfig( @@ -124,7 +134,8 @@ public class ConsensusConfig { .orElseGet(() -> IoTConsensusConfig.newBuilder().build()), Optional.ofNullable(iotConsensusV2Config) .orElseGet(() -> IoTConsensusV2Config.newBuilder().build()), - directoryStrategyType); + directoryStrategyType, + trustedChannelFailureHandler); } public Builder setThisNode(TEndPoint thisNode) { @@ -171,5 +182,13 @@ public class ConsensusConfig { this.directoryStrategyType = directoryStrategyType; return this; } + + public Builder setTrustedChannelFailureHandler( + TrustedChannelFailureHandler trustedChannelFailureHandler) { + this.trustedChannelFailureHandler = + Optional.ofNullable(trustedChannelFailureHandler) + .orElse(TrustedChannelFailureHandler.NO_OP); + return this; + } } } diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java index 1e247ad599e..ce21fa0c8db 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java @@ -133,7 +133,9 @@ public class IoTConsensus implements IConsensus { this.recvFolderStrategyType = config.getDirectoryStrategyType(); this.config = config.getIotConsensusConfig(); this.registry = registry; - this.service = new IoTConsensusRPCService(thisNode, config.getIotConsensusConfig()); + this.service = + new IoTConsensusRPCService( + thisNode, config.getIotConsensusConfig(), config.getTrustedChannelFailureHandler()); this.clientManager = new IClientManager.Factory<TEndPoint, AsyncIoTConsensusServiceClient>() .createClientManager( diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCService.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCService.java index cfa175b2de5..15590dcd2ba 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCService.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCService.java @@ -20,11 +20,13 @@ package org.apache.iotdb.consensus.iot.service; import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.commons.audit.TrustedChannelFailureHandler; import org.apache.iotdb.commons.concurrent.ThreadName; import org.apache.iotdb.commons.exception.runtime.RPCServiceException; import org.apache.iotdb.commons.service.ServiceType; import org.apache.iotdb.commons.service.ThriftService; import org.apache.iotdb.commons.service.ThriftServiceThread; +import org.apache.iotdb.commons.service.TrustedChannelAuditServerEventHandler; import org.apache.iotdb.consensus.config.IoTConsensusConfig; import org.apache.iotdb.consensus.iot.thrift.IoTConsensusIService; import org.apache.iotdb.rpc.ZeroCopyRpcTransportFactory; @@ -40,11 +42,16 @@ public class IoTConsensusRPCService extends ThriftService implements IoTConsensu private final TEndPoint thisNode; private final IoTConsensusConfig config; + private final TrustedChannelFailureHandler trustedChannelFailureHandler; private IoTConsensusRPCServiceProcessor iotConsensusRPCServiceProcessor; - public IoTConsensusRPCService(TEndPoint thisNode, IoTConsensusConfig config) { + public IoTConsensusRPCService( + TEndPoint thisNode, + IoTConsensusConfig config, + TrustedChannelFailureHandler trustedChannelFailureHandler) { this.thisNode = thisNode; this.config = config; + this.trustedChannelFailureHandler = trustedChannelFailureHandler; } @Override @@ -83,7 +90,10 @@ public class IoTConsensusRPCService extends ThriftService implements IoTConsensu getBindPort(), config.getRpc().getRpcMaxConcurrentClientNum(), config.getRpc().getThriftServerAwaitTimeForStopService(), - new IoTConsensusRPCServiceHandler(iotConsensusRPCServiceProcessor), + new TrustedChannelAuditServerEventHandler( + new IoTConsensusRPCServiceHandler(iotConsensusRPCServiceProcessor), + thisNode, + trustedChannelFailureHandler), config.getRpc().isRpcThriftCompressionEnabled(), config.getRpc().getSslKeyStorePath(), config.getRpc().getSslKeyStorePassword(), diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/IoTConsensusV2.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/IoTConsensusV2.java index cf1d4056398..b98b7e5f3f1 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/IoTConsensusV2.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/IoTConsensusV2.java @@ -107,7 +107,9 @@ public class IoTConsensusV2 implements IConsensus { this.storageDir = new File(config.getStorageDir()); this.config = config.getIoTConsensusV2Config(); this.registry = registry; - this.rpcService = new IoTConsensusV2RPCService(thisNode, config.getIoTConsensusV2Config()); + this.rpcService = + new IoTConsensusV2RPCService( + thisNode, config.getIoTConsensusV2Config(), config.getTrustedChannelFailureHandler()); this.asyncClientManager = IoTV2GlobalComponentContainer.getInstance().getGlobalAsyncClientManager(); this.syncClientManager = diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/service/IoTConsensusV2RPCService.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/service/IoTConsensusV2RPCService.java index c218c8051e8..22667813076 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/service/IoTConsensusV2RPCService.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/service/IoTConsensusV2RPCService.java @@ -20,11 +20,13 @@ package org.apache.iotdb.consensus.pipe.service; import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.commons.audit.TrustedChannelFailureHandler; import org.apache.iotdb.commons.concurrent.ThreadName; import org.apache.iotdb.commons.exception.runtime.RPCServiceException; import org.apache.iotdb.commons.service.ServiceType; import org.apache.iotdb.commons.service.ThriftService; import org.apache.iotdb.commons.service.ThriftServiceThread; +import org.apache.iotdb.commons.service.TrustedChannelAuditServerEventHandler; import org.apache.iotdb.consensus.config.IoTConsensusV2Config; import org.apache.iotdb.consensus.iotconsensusv2.thrift.IoTConsensusV2IService; import org.apache.iotdb.rpc.ZeroCopyRpcTransportFactory; @@ -34,11 +36,16 @@ public class IoTConsensusV2RPCService extends ThriftService private final TEndPoint thisNode; private final IoTConsensusV2Config config; + private final TrustedChannelFailureHandler trustedChannelFailureHandler; private IoTConsensusV2RPCServiceProcessor iotConsensusV2RPCServiceProcessor; - public IoTConsensusV2RPCService(TEndPoint thisNode, IoTConsensusV2Config config) { + public IoTConsensusV2RPCService( + TEndPoint thisNode, + IoTConsensusV2Config config, + TrustedChannelFailureHandler trustedChannelFailureHandler) { this.thisNode = thisNode; this.config = config; + this.trustedChannelFailureHandler = trustedChannelFailureHandler; } @Override @@ -71,7 +78,10 @@ public class IoTConsensusV2RPCService extends ThriftService getBindPort(), config.getRpc().getRpcMaxConcurrentClientNum(), config.getRpc().getThriftServerAwaitTimeForStopService(), - new IoTConsensusV2RPCServiceHandler(iotConsensusV2RPCServiceProcessor), + new TrustedChannelAuditServerEventHandler( + new IoTConsensusV2RPCServiceHandler(iotConsensusV2RPCServiceProcessor), + thisNode, + trustedChannelFailureHandler), config.getRpc().isRpcThriftCompressionEnabled(), config.getRpc().getSslKeyStorePath(), config.getRpc().getSslKeyStorePassword(), diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/audit/DNAuditLogger.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/audit/DNAuditLogger.java index 90026e37ba0..1ba8309feeb 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/audit/DNAuditLogger.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/audit/DNAuditLogger.java @@ -19,11 +19,14 @@ package org.apache.iotdb.db.audit; +import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.commons.audit.AbstractAuditLogger; import org.apache.iotdb.commons.audit.AuditLogFields; import org.apache.iotdb.commons.audit.IAuditEntity; +import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.commons.exception.IllegalPathException; import org.apache.iotdb.commons.path.PartialPath; +import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.queryengine.plan.Coordinator; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowStatement; @@ -58,6 +61,17 @@ public class DNAuditLogger extends AbstractAuditLogger { @Override public synchronized void log(IAuditEntity auditLogFields, Supplier<String> log) {} + public void recordTrustedChannelFailureAuditLogIfNecessary(Throwable failure, TEndPoint target) { + if (CommonDescriptor.getInstance().getConfig().isEnableInternalSSL()) { + recordTrustedChannelFailureAuditLogIfNecessary( + failure, + new TEndPoint( + IoTDBDescriptor.getInstance().getConfig().getInternalAddress(), + IoTDBDescriptor.getInstance().getConfig().getInternalPort()), + target); + } + } + public void logFromCN(AuditLogFields auditLogFields, String log, int nodeId) throws IllegalPathException {} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/DataRegionConsensusImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/DataRegionConsensusImpl.java index a4c8e00f1ad..fbda403b320 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/DataRegionConsensusImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/DataRegionConsensusImpl.java @@ -39,6 +39,7 @@ import org.apache.iotdb.consensus.config.IoTConsensusV2Config; import org.apache.iotdb.consensus.config.IoTConsensusV2Config.ReplicateMode; import org.apache.iotdb.consensus.config.RatisConfig; import org.apache.iotdb.consensus.config.RatisConfig.Snapshot; +import org.apache.iotdb.db.audit.DNAuditLogger; import org.apache.iotdb.db.conf.DataNodeMemoryConfig; import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; @@ -140,6 +141,8 @@ public class DataRegionConsensusImpl { return ConsensusConfig.newBuilder() .setThisNodeId(CONF.getDataNodeId()) .setThisNode(new TEndPoint(CONF.getInternalAddress(), CONF.getDataRegionConsensusPort())) + .setTrustedChannelFailureHandler( + DNAuditLogger.getInstance()::recordTrustedChannelFailureAuditLogIfNecessary) .setStorageDir(CONF.getDataRegionConsensusDir()) .setRecvSnapshotDirs(Arrays.asList(CONF.getLocalDataDirs())) // IoTConsensus always balances received snapshot files by least occupied space, diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java index a9fe5a60cd4..572cd65ca53 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java @@ -190,6 +190,7 @@ import org.apache.iotdb.confignode.rpc.thrift.TTestOperation; import org.apache.iotdb.confignode.rpc.thrift.TThrottleQuotaResp; import org.apache.iotdb.confignode.rpc.thrift.TUnsetSchemaTemplateReq; import org.apache.iotdb.confignode.rpc.thrift.TUnsubscribeReq; +import org.apache.iotdb.db.audit.DNAuditLogger; import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.i18n.DataNodeMiscMessages; @@ -325,6 +326,7 @@ public class ConfigNodeClient implements IConfigNodeRPCService.Iface, ThriftClie return; } catch (TException e) { logger.warn(DataNodeMiscMessages.NODE_LEADER_MAY_DOWN_TRY_NEXT, configLeader); + recordTrustedChannelFailureAuditLogIfNecessary(e, configLeader); configLeader = null; exception = e; } @@ -347,6 +349,7 @@ public class ConfigNodeClient implements IConfigNodeRPCService.Iface, ThriftClie return; } catch (TException e) { logger.warn(DataNodeMiscMessages.NODE_MAY_DOWN_TRY_NEXT, tryEndpoint); + recordTrustedChannelFailureAuditLogIfNecessary(e, tryEndpoint); exception = e; } } @@ -461,6 +464,7 @@ public class ConfigNodeClient implements IConfigNodeRPCService.Iface, ThriftClie Thread.currentThread().getStackTrace()[2].getMethodName()); logger.warn(message, e); configLeader = null; + recordTrustedChannelFailureAuditLogIfNecessary(e, configNode); if (e.getCause() != null && e.getCause() instanceof SSLHandshakeException) { throw e; } @@ -485,6 +489,12 @@ public class ConfigNodeClient implements IConfigNodeRPCService.Iface, ThriftClie throw new TException(MSG_RECONNECTION_FAIL); } + private void recordTrustedChannelFailureAuditLogIfNecessary(Throwable failure, TEndPoint target) { + if (commonConfig.isEnableInternalSSL()) { + DNAuditLogger.getInstance().recordTrustedChannelFailureAuditLogIfNecessary(failure, target); + } + } + @FunctionalInterface private interface Operation<T> { T execute() throws TException; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/an/AINodeClient.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/an/AINodeClient.java index 8815acc036e..72f1886310b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/an/AINodeClient.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/an/AINodeClient.java @@ -50,6 +50,7 @@ import org.apache.iotdb.commons.conf.CommonConfig; import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.commons.consensus.ConfigRegionId; import org.apache.iotdb.confignode.rpc.thrift.TGetAINodeLocationResp; +import org.apache.iotdb.db.audit.DNAuditLogger; import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.i18n.DataNodeMiscMessages; @@ -177,6 +178,9 @@ public class AINodeClient implements IAINodeRPCService.Iface, AutoCloseable, Thr IOTDB_CONFIG.getAddressAndPort(), Thread.currentThread().getStackTrace()[2].getMethodName()); LOGGER.warn(message, e); + final TAINodeLocation currentLocation = CURRENT_LOCATION.get(); + recordTrustedChannelFailureAuditLogIfNecessary( + e, currentLocation == null ? null : currentLocation.getInternalEndPoint()); CURRENT_LOCATION.set(null); if (e.getCause() != null && e.getCause() instanceof SSLHandshakeException) { throw e; @@ -202,6 +206,7 @@ public class AINodeClient implements IAINodeRPCService.Iface, AutoCloseable, Thr return; } catch (TException e) { LOGGER.warn(DataNodeMiscMessages.AINODE_MAY_DOWN, endpoint, e); + recordTrustedChannelFailureAuditLogIfNecessary(e, endpoint); CURRENT_LOCATION.set(null); } } else { @@ -233,6 +238,12 @@ public class AINodeClient implements IAINodeRPCService.Iface, AutoCloseable, Thr client = new IAINodeRPCService.Client(property.getProtocolFactory().getProtocol(transport)); } + private void recordTrustedChannelFailureAuditLogIfNecessary(Throwable failure, TEndPoint target) { + if (COMMON_CONFIG.isEnableInternalSSL()) { + DNAuditLogger.getInstance().recordTrustedChannelFailureAuditLogIfNecessary(failure, target); + } + } + public TEndPoint getCurrentEndpoint() { TAINodeLocation loc = CURRENT_LOCATION.get(); if (loc == null) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/dn/AsyncTSStatusRPCHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/dn/AsyncTSStatusRPCHandler.java index c69f6ea4e40..fefb4983dd9 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/dn/AsyncTSStatusRPCHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/dn/AsyncTSStatusRPCHandler.java @@ -21,6 +21,7 @@ package org.apache.iotdb.db.protocol.client.dn; import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation; import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.db.audit.DNAuditLogger; import org.apache.iotdb.db.i18n.DataNodeMiscMessages; import org.apache.iotdb.rpc.RpcUtils; import org.apache.iotdb.rpc.TSStatusCode; @@ -77,6 +78,8 @@ public class AsyncTSStatusRPCHandler extends DataNodeAsyncRequestRPCHandler<TSSt @Override public void onError(Exception e) { + DNAuditLogger.getInstance() + .recordTrustedChannelFailureAuditLogIfNecessary(e, targetNode.getInternalEndPoint()); String errorMsg = "Failed to " + requestType diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/MPPDataExchangeService.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/MPPDataExchangeService.java index a4cef2e4b71..6afcd2a3a91 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/MPPDataExchangeService.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/MPPDataExchangeService.java @@ -32,7 +32,9 @@ import org.apache.iotdb.commons.exception.runtime.RPCServiceException; import org.apache.iotdb.commons.service.ServiceType; import org.apache.iotdb.commons.service.ThriftService; import org.apache.iotdb.commons.service.ThriftServiceThread; +import org.apache.iotdb.commons.service.TrustedChannelAuditServerEventHandler; import org.apache.iotdb.commons.service.metric.MetricService; +import org.apache.iotdb.db.audit.DNAuditLogger; import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.i18n.DataNodeQueryMessages; @@ -103,7 +105,10 @@ public class MPPDataExchangeService extends ThriftService implements MPPDataExch getBindPort(), config.getRpcMaxConcurrentClientNum(), config.getThriftServerAwaitTimeForStopService(), - new MPPDataExchangeServiceThriftHandler(), + new TrustedChannelAuditServerEventHandler( + new MPPDataExchangeServiceThriftHandler(), + new TEndPoint(getBindIP(), getBindPort()), + DNAuditLogger.getInstance()::recordTrustedChannelFailureAuditLogIfNecessary), config.isRpcThriftCompressionEnable(), commonConfig.getKeyStorePath(), commonConfig.getKeyStorePwd(), diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/sink/SinkChannel.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/sink/SinkChannel.java index ec6c9c680a1..dc460466fc4 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/sink/SinkChannel.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/sink/SinkChannel.java @@ -23,6 +23,7 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.commons.client.IClientManager; import org.apache.iotdb.commons.client.sync.SyncDataNodeMPPDataExchangeServiceClient; import org.apache.iotdb.commons.utils.TestOnly; +import org.apache.iotdb.db.audit.DNAuditLogger; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.i18n.DataNodeQueryMessages; import org.apache.iotdb.db.queryengine.common.FragmentInstanceId; @@ -579,6 +580,8 @@ public class SinkChannel implements ISinkChannel { client.onNewDataBlockEvent(newDataBlockEvent); break; } catch (Exception e) { + DNAuditLogger.getInstance() + .recordTrustedChannelFailureAuditLogIfNecessary(e, remoteEndpoint); LOGGER.warn( DataNodeQueryMessages.FAILED_TO_SEND_NEW_DATA_BLOCK_EVENT_ATTEMPT, attempt, e); if (attempt == MAX_ATTEMPT_TIMES) { @@ -627,6 +630,8 @@ public class SinkChannel implements ISinkChannel { client.onEndOfDataBlockEvent(endOfDataBlockEvent); break; } catch (Exception e) { + DNAuditLogger.getInstance() + .recordTrustedChannelFailureAuditLogIfNecessary(e, remoteEndpoint); LOGGER.warn(DataNodeQueryMessages.FAILED_TO_SEND_END_OF_DATA_BLOCK_EVENT, attempt, e); if (attempt == MAX_ATTEMPT_TIMES) { LOGGER.warn(DataNodeQueryMessages.FAILED_TO_SEND_END_OF_DATA_BLOCK_EVENT_2, e); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/source/SourceHandle.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/source/SourceHandle.java index 4f90db0caed..66a2c70e551 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/source/SourceHandle.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/source/SourceHandle.java @@ -23,6 +23,7 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.commons.client.IClientManager; import org.apache.iotdb.commons.client.sync.SyncDataNodeMPPDataExchangeServiceClient; import org.apache.iotdb.commons.utils.TestOnly; +import org.apache.iotdb.db.audit.DNAuditLogger; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.i18n.DataNodeQueryMessages; import org.apache.iotdb.db.queryengine.common.FragmentInstanceId; @@ -663,6 +664,8 @@ public class SourceHandle implements ISourceHandle { } break; } catch (Throwable e) { + DNAuditLogger.getInstance() + .recordTrustedChannelFailureAuditLogIfNecessary(e, remoteEndpoint); LOGGER.warn( DataNodeQueryMessages.FAILED_TO_GET_DATA_BLOCK, @@ -744,6 +747,8 @@ public class SourceHandle implements ISourceHandle { client.onAcknowledgeDataBlockEvent(acknowledgeDataBlockEvent); break; } catch (Throwable e) { + DNAuditLogger.getInstance() + .recordTrustedChannelFailureAuditLogIfNecessary(e, remoteEndpoint); LOGGER.warn( DataNodeQueryMessages.FAILED_TO_SEND_ACK_DATA_BLOCK_EVENT, startSequenceId, @@ -795,6 +800,8 @@ public class SourceHandle implements ISourceHandle { client.onCloseSinkChannelEvent(closeSinkChannelEvent); break; } catch (Throwable e) { + DNAuditLogger.getInstance() + .recordTrustedChannelFailureAuditLogIfNecessary(e, remoteEndpoint); LOGGER.warn( DataNodeQueryMessages.SEND_CLOSE_SINK_CHANNEL_EVENT_FAILED, remoteFragmentInstanceId, diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/AsyncPlanNodeSender.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/AsyncPlanNodeSender.java index da5a2dfc889..00611993e12 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/AsyncPlanNodeSender.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/AsyncPlanNodeSender.java @@ -91,7 +91,8 @@ public class AsyncPlanNodeSender { pendingNumber, instanceId2RespMap, needRetryInstanceIndex, - startSendTime); + startSendTime, + entry.getKey()); try { AsyncDataNodeInternalServiceClient client = asyncInternalServiceClientManager.borrowClient(entry.getKey()); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/AsyncSendPlanNodeHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/AsyncSendPlanNodeHandler.java index 23f99da3484..e73e4f6e1cd 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/AsyncSendPlanNodeHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/AsyncSendPlanNodeHandler.java @@ -19,8 +19,10 @@ package org.apache.iotdb.db.queryengine.plan.scheduler; +import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.commons.service.metric.PerformanceOverviewMetrics; import org.apache.iotdb.commons.utils.StatusUtils; +import org.apache.iotdb.db.audit.DNAuditLogger; import org.apache.iotdb.mpp.rpc.thrift.TSendBatchPlanNodeResp; import org.apache.iotdb.mpp.rpc.thrift.TSendSinglePlanNodeResp; import org.apache.iotdb.rpc.RpcUtils; @@ -43,6 +45,7 @@ public class AsyncSendPlanNodeHandler implements AsyncMethodCallback<TSendBatchP private final Map<Integer, TSendSinglePlanNodeResp> instanceId2RespMap; private final List<Integer> needRetryInstanceIndex; private final long sendTime; + private final TEndPoint targetEndPoint; private static final PerformanceOverviewMetrics PERFORMANCE_OVERVIEW_METRICS = PerformanceOverviewMetrics.getInstance(); @@ -51,12 +54,14 @@ public class AsyncSendPlanNodeHandler implements AsyncMethodCallback<TSendBatchP AtomicLong pendingNumber, Map<Integer, TSendSinglePlanNodeResp> instanceId2RespMap, List<Integer> needRetryInstanceIndex, - long sendTime) { + long sendTime, + TEndPoint targetEndPoint) { this.instanceIds = instanceIds; this.pendingNumber = pendingNumber; this.instanceId2RespMap = instanceId2RespMap; this.needRetryInstanceIndex = needRetryInstanceIndex; this.sendTime = sendTime; + this.targetEndPoint = targetEndPoint; } @Override @@ -78,6 +83,7 @@ public class AsyncSendPlanNodeHandler implements AsyncMethodCallback<TSendBatchP @Override public void onError(Exception e) { + DNAuditLogger.getInstance().recordTrustedChannelFailureAuditLogIfNecessary(e, targetEndPoint); if (needRetry(e)) { needRetryInstanceIndex.addAll(instanceIds); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/FragmentInstanceDispatcherImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/FragmentInstanceDispatcherImpl.java index df719a724fe..c76912ced24 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/FragmentInstanceDispatcherImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/FragmentInstanceDispatcherImpl.java @@ -35,6 +35,7 @@ import org.apache.iotdb.commons.service.metric.PerformanceOverviewMetrics; import org.apache.iotdb.commons.utils.RetryUtils; import org.apache.iotdb.consensus.exception.ConsensusGroupNotExistException; import org.apache.iotdb.consensus.exception.RatisReadUnavailableException; +import org.apache.iotdb.db.audit.DNAuditLogger; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.exception.mpp.FragmentInstanceDispatchException; import org.apache.iotdb.db.exception.query.QueryTimeoutRuntimeException; @@ -587,6 +588,7 @@ public class FragmentInstanceDispatcherImpl implements IFragInstanceDispatcher { try { dispatchRemoteHelper(instance, endPoint); } catch (ClientManagerException | TException | RatisReadUnavailableException e) { + DNAuditLogger.getInstance().recordTrustedChannelFailureAuditLogIfNecessary(e, endPoint); LOGGER.warn( DataNodeQueryMessages .CAN_T_EXECUTE_REQUEST_ON_NODE_ARG_ERROR_MSG_IS_ARG_AND_WE_TRY_TO_RECONNECT_THIS_NODE, @@ -613,6 +615,7 @@ public class FragmentInstanceDispatcherImpl implements IFragInstanceDispatcher { | TException | RatisReadUnavailableException | ConsensusGroupNotExistException e1) { + DNAuditLogger.getInstance().recordTrustedChannelFailureAuditLogIfNecessary(e1, endPoint); dispatchRemoteFailed(endPoint, e1); } } catch (ConsensusGroupNotExistException e) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeInternalRPCService.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeInternalRPCService.java index b58d647fcd7..f9b9c6e0101 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeInternalRPCService.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeInternalRPCService.java @@ -19,6 +19,7 @@ package org.apache.iotdb.db.service; +import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.commons.concurrent.ThreadName; import org.apache.iotdb.commons.conf.CommonConfig; import org.apache.iotdb.commons.conf.CommonDescriptor; @@ -26,7 +27,9 @@ import org.apache.iotdb.commons.exception.runtime.RPCServiceException; import org.apache.iotdb.commons.service.ServiceType; import org.apache.iotdb.commons.service.ThriftService; import org.apache.iotdb.commons.service.ThriftServiceThread; +import org.apache.iotdb.commons.service.TrustedChannelAuditServerEventHandler; import org.apache.iotdb.commons.service.metric.MetricService; +import org.apache.iotdb.db.audit.DNAuditLogger; import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.protocol.thrift.handler.InternalServiceThriftHandler; @@ -74,7 +77,10 @@ public class DataNodeInternalRPCService extends ThriftService getBindPort(), config.getRpcMaxConcurrentClientNum(), config.getThriftServerAwaitTimeForStopService(), - new InternalServiceThriftHandler(), + new TrustedChannelAuditServerEventHandler( + new InternalServiceThriftHandler(), + new TEndPoint(getBindIP(), getBindPort()), + DNAuditLogger.getInstance()::recordTrustedChannelFailureAuditLogIfNecessary), config.isRpcThriftCompressionEnable(), commonConfig.getKeyStorePath(), commonConfig.getKeyStorePwd(), diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/ExternalRPCService.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/ExternalRPCService.java index f9b34dd9e58..782788547bf 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/ExternalRPCService.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/ExternalRPCService.java @@ -18,6 +18,7 @@ */ package org.apache.iotdb.db.service; +import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.commons.concurrent.ThreadName; import org.apache.iotdb.commons.conf.CommonConfig; import org.apache.iotdb.commons.conf.CommonDescriptor; @@ -25,7 +26,9 @@ import org.apache.iotdb.commons.exception.runtime.RPCServiceException; import org.apache.iotdb.commons.service.ServiceType; import org.apache.iotdb.commons.service.ThriftService; import org.apache.iotdb.commons.service.ThriftServiceThread; +import org.apache.iotdb.commons.service.TrustedChannelAuditServerEventHandler; import org.apache.iotdb.commons.service.metric.MetricService; +import org.apache.iotdb.db.audit.DNAuditLogger; import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.i18n.DataNodeMiscMessages; @@ -94,7 +97,7 @@ public class ExternalRPCService extends ThriftService implements ExternalRPCServ getBindPort(), config.getRpcMaxConcurrentClientNum(), config.getThriftServerAwaitTimeForStopService(), - new RPCServiceThriftHandler(impl), + newTrustedChannelAuditHandler(), config.isRpcThriftCompressionEnable(), commonConfig.getKeyStorePath(), commonConfig.getKeyStorePwd(), @@ -111,7 +114,7 @@ public class ExternalRPCService extends ThriftService implements ExternalRPCServ getBindPort(), config.getRpcMaxConcurrentClientNum(), config.getThriftServerAwaitTimeForStopService(), - new RPCServiceThriftHandler(impl), + newTrustedChannelAuditHandler(), config.isRpcThriftCompressionEnable(), commonConfig.getKeyStorePath(), commonConfig.getKeyStorePwd(), @@ -150,6 +153,13 @@ public class ExternalRPCService extends ThriftService implements ExternalRPCServ return value != null && !value.trim().isEmpty(); } + private TrustedChannelAuditServerEventHandler newTrustedChannelAuditHandler() { + return new TrustedChannelAuditServerEventHandler( + new RPCServiceThriftHandler(impl), + new TEndPoint(getBindIP(), getBindPort()), + DNAuditLogger.getInstance()::recordTrustedChannelFailureAuditLogIfNecessary); + } + private static class RPCServiceHolder { private static final ExternalRPCService INSTANCE = new ExternalRPCService(); diff --git a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java index ae97d09ddc2..94194a086b4 100644 --- a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java @@ -222,5 +222,7 @@ public final class CommonMessages { public static final String EXCEPTION_THE_ORDER_BY_CLAUSE_OF_THE_DATA_ARGUMENT_MUST_CONTAIN_EXACTLY_THE_TIME_COLUMN_SPECIFIED_BY_THE_TIMECOL_ARGUMENT_4375BAE9 = "The ORDER BY clause of the DATA argument must contain exactly the time column specified by the TIMECOL argument."; public static final String EXCEPTION_UNSUPPORTED_M4_VALUE_TYPE_AF0EF286 = "Unsupported M4 value type: "; public static final String EXCEPTION_DISK_SPACE_WARNING_THRESHOLD_MUST_BE_IN_0_1_BUT_WAS_7B345766 = "disk_space_warning_threshold must be in [0, 1), but was "; + public static final String LOG_TRUSTED_CHANNEL_FUNCTION_FAILED_INITIATOR_ARG_TARGET_ARG_E4C28443 = + "Trusted channel function failed: initiator=%s, target=%s"; } diff --git a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java index 71c45ccaf3f..f0b84bbc279 100644 --- a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java @@ -215,5 +215,7 @@ public final class CommonMessages { public static final String EXCEPTION_THE_ORDER_BY_CLAUSE_OF_THE_DATA_ARGUMENT_MUST_CONTAIN_EXACTLY_THE_TIME_COLUMN_SPECIFIED_BY_THE_TIMECOL_ARGUMENT_4375BAE9 = "DATA 参数的 ORDER BY 子句必须仅包含 TIMECOL 参数指定的时间列。"; public static final String EXCEPTION_UNSUPPORTED_M4_VALUE_TYPE_AF0EF286 = "不支持的 M4 值类型："; public static final String EXCEPTION_DISK_SPACE_WARNING_THRESHOLD_MUST_BE_IN_0_1_BUT_WAS_7B345766 = "disk_space_warning_threshold 必须在 [0, 1) 范围内，但实际为 "; + public static final String LOG_TRUSTED_CHANNEL_FUNCTION_FAILED_INITIATOR_ARG_TARGET_ARG_E4C28443 = + "可信信道功能失效：initiator=%s，target=%s"; } diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/audit/AbstractAuditLogger.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/audit/AbstractAuditLogger.java index d7c00c9128b..bdc17c49686 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/audit/AbstractAuditLogger.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/audit/AbstractAuditLogger.java @@ -19,12 +19,23 @@ package org.apache.iotdb.commons.audit; +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.commons.auth.entity.PrivilegeType; +import org.apache.iotdb.commons.auth.entity.User; import org.apache.iotdb.commons.conf.CommonConfig; import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.commons.i18n.CommonMessages; +import org.apache.iotdb.commons.utils.NodeUrlUtils; + +import javax.net.ssl.SSLException; import java.util.function.Supplier; public abstract class AbstractAuditLogger { + private static final long INTERNAL_AUDIT_LOG_USER_ID = 4; + private static final ThreadLocal<Boolean> RECORDING_TRUSTED_CHANNEL_FAILURE = + ThreadLocal.withInitial(() -> false); + public static final String OBJECT_AUTHENTICATION_AUDIT_STR = "User %s (ID=%d) requests authority on object %s with result %s"; public static final String AUDIT_LOG_NODE_ID = "node_id"; @@ -61,4 +72,64 @@ public abstract class AbstractAuditLogger { auditObject.get(), auditEntity.getResult())); } + + /** + * Records a failure of the trusted-channel function. + * + * <p>The caller determines the channel direction and supplies the actual initiator and target + * identifiers. This keeps the audit hook independent of any concrete SSL/TLS implementation. + */ + public void recordTrustedChannelFailureAuditLog( + final IAuditEntity auditEntity, + final Supplier<String> initiator, + final Supplier<String> target) { + log( + auditEntity + .setAuditEventType(AuditEventType.TRUSTED_CHANNEL_FUNCTION_FAILURE) + .setAuditLogOperation(AuditLogOperation.CONTROL) + .setResult(false), + () -> + String.format( + CommonMessages + .LOG_TRUSTED_CHANNEL_FUNCTION_FAILED_INITIATOR_ARG_TARGET_ARG_E4C28443, + initiator.get(), + target.get())); + } + + public static boolean isSslFailure(Throwable failure) { + Throwable cause = failure; + while (cause != null) { + if (cause instanceof SSLException) { + return true; + } + cause = cause.getCause(); + } + return false; + } + + public void recordTrustedChannelFailureAuditLogIfNecessary( + Throwable failure, TEndPoint initiator, TEndPoint target) { + if (RECORDING_TRUSTED_CHANNEL_FAILURE.get() + || !isSslFailure(failure) + || initiator == null + || target == null) { + return; + } + + final String initiatorIdentifier = NodeUrlUtils.convertTEndPointUrl(initiator); + final String targetIdentifier = NodeUrlUtils.convertTEndPointUrl(target); + RECORDING_TRUSTED_CHANNEL_FAILURE.set(true); + try { + recordTrustedChannelFailureAuditLog( + new UserEntity( + INTERNAL_AUDIT_LOG_USER_ID, + User.BUILTIN_INTERNAL_AUDIT_LOG_USERNAME, + initiatorIdentifier) + .setPrivilegeType(PrivilegeType.AUDIT), + () -> initiatorIdentifier, + () -> targetIdentifier); + } finally { + RECORDING_TRUSTED_CHANNEL_FAILURE.remove(); + } + } } diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/audit/AuditEventType.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/audit/AuditEventType.java index 510912a302f..2c99e436676 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/audit/AuditEventType.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/audit/AuditEventType.java @@ -44,7 +44,7 @@ public enum AuditEventType { LOGIN_EXCEED_LIMIT, SESSION_TIME_EXCEEDED, LOGIN_REJECT_IP, - SESSION_ENCRYPT_FAILED, + TRUSTED_CHANNEL_FUNCTION_FAILURE, SYSTEM_OPERATION, DN_SHUTDOWN; diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/audit/AuditEventType.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/audit/TrustedChannelFailureHandler.java similarity index 56% copy from iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/audit/AuditEventType.java copy to iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/audit/TrustedChannelFailureHandler.java index 510912a302f..e342cc8d231 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/audit/AuditEventType.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/audit/TrustedChannelFailureHandler.java @@ -19,38 +19,12 @@ package org.apache.iotdb.commons.audit; -public enum AuditEventType { - CHANGE_AUDIT_OPTION, - AUDIT_STORAGE_FULL, - GENERATE_KEY, - DESTROY_KEY, - EXECUTE_ENCRYPT, - OBJECT_AUTHENTICATION, - LBAC_AUTHENTICATION, - EXPORT_DATA_WITH_LABEL, - IMPORT_DATA_WITH_LABEL, - INTEGRITY_CHECK, - LOGIN_FAIL_MAX_TIMES, - MODIFY_PASSWD, - LOGIN, - LOGOUT, - LOGIN_FINAL, - MODIFY_SECURITY_OPTIONS, - MODIFY_DEFAULT_SECURITY_VALUES, - REVOKE_FAILED, - GRANT_ROLE_FAILED, - LOGIN_RESOURCE_RESTRICT, - LOGIN_FAILED_TRIES, - LOGIN_EXCEED_LIMIT, - SESSION_TIME_EXCEEDED, - LOGIN_REJECT_IP, - SESSION_ENCRYPT_FAILED, - SYSTEM_OPERATION, +import org.apache.iotdb.common.rpc.thrift.TEndPoint; - DN_SHUTDOWN; +@FunctionalInterface +public interface TrustedChannelFailureHandler { - @Override - public String toString() { - return name(); - } + TrustedChannelFailureHandler NO_OP = (failure, initiator, target) -> {}; + + void onFailure(Throwable failure, TEndPoint initiator, TEndPoint target); } diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/request/AsyncRequestManager.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/request/AsyncRequestManager.java index b13d1ab0637..4332ebfa65f 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/request/AsyncRequestManager.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/request/AsyncRequestManager.java @@ -210,6 +210,7 @@ public abstract class AsyncRequestManager<RequestType, NodeLocation, Client> { endPoint, e.getMessage(), retryCount); + onRequestFailure(e, endPoint); if (handler != null) { try { handler.onError(e); @@ -237,6 +238,10 @@ public abstract class AsyncRequestManager<RequestType, NodeLocation, Client> { // In default, no need to do this } + protected void onRequestFailure(Exception failure, TEndPoint targetEndPoint) { + // In default, no need to do this + } + protected abstract TEndPoint nodeLocationToEndPoint(NodeLocation location); protected abstract AsyncRequestRPCHandler<?, RequestType, NodeLocation> buildHandler( diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/TrustedChannelAuditServerEventHandler.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/TrustedChannelAuditServerEventHandler.java new file mode 100644 index 00000000000..4c9679ee5eb --- /dev/null +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/TrustedChannelAuditServerEventHandler.java @@ -0,0 +1,135 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.commons.service; + +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.commons.audit.TrustedChannelFailureHandler; +import org.apache.iotdb.rpc.TElasticFramedTransport; + +import org.apache.thrift.protocol.TProtocol; +import org.apache.thrift.server.ServerContext; +import org.apache.thrift.server.TServerEventHandler; +import org.apache.thrift.transport.TSocket; +import org.apache.thrift.transport.TTransport; + +import javax.net.ssl.SSLException; +import javax.net.ssl.SSLSocket; + +import java.io.IOException; +import java.io.UncheckedIOException; +import java.net.InetSocketAddress; +import java.net.Socket; +import java.net.SocketAddress; +import java.util.Objects; + +/** + * Performs the server-side TLS handshake before the first Thrift request is processed and reports + * handshake failures together with the raw socket peer and the local service endpoint. + */ +public class TrustedChannelAuditServerEventHandler implements TServerEventHandler { + + private final TServerEventHandler delegate; + private final TEndPoint target; + private final TrustedChannelFailureHandler failureHandler; + + public TrustedChannelAuditServerEventHandler( + TServerEventHandler delegate, TEndPoint target, TrustedChannelFailureHandler failureHandler) { + this.delegate = Objects.requireNonNull(delegate); + this.target = Objects.requireNonNull(target); + this.failureHandler = Objects.requireNonNull(failureHandler); + } + + @Override + public void preServe() { + delegate.preServe(); + } + + @Override + public ServerContext createContext(TProtocol input, TProtocol output) { + startHandshakeIfNecessary(output); + return delegate.createContext(input, output); + } + + @Override + public void deleteContext(ServerContext serverContext, TProtocol input, TProtocol output) { + if (serverContext != null) { + delegate.deleteContext(serverContext, input, output); + } + } + + @Override + public void processContext( + ServerContext serverContext, TTransport inputTransport, TTransport outputTransport) { + delegate.processContext(serverContext, inputTransport, outputTransport); + } + + private void startHandshakeIfNecessary(TProtocol output) { + Socket socket = getSocket(output); + if (!(socket instanceof SSLSocket)) { + return; + } + + try { + ((SSLSocket) socket).startHandshake(); + } catch (IOException e) { + if (e instanceof SSLException) { + notifyFailure(e, socket.getRemoteSocketAddress()); + } + try { + socket.close(); + } catch (IOException closeFailure) { + e.addSuppressed(closeFailure); + } + throw new UncheckedIOException(e); + } + } + + private void notifyFailure(Throwable failure, SocketAddress remoteAddress) { + TEndPoint initiator = toEndPoint(remoteAddress); + if (initiator == null) { + return; + } + try { + failureHandler.onFailure(failure, initiator, target); + } catch (RuntimeException auditFailure) { + failure.addSuppressed(auditFailure); + } + } + + private static Socket getSocket(TProtocol protocol) { + if (protocol == null || !(protocol.getTransport() instanceof TElasticFramedTransport)) { + return null; + } + TTransport socketTransport = ((TElasticFramedTransport) protocol.getTransport()).getSocket(); + return socketTransport instanceof TSocket ? ((TSocket) socketTransport).getSocket() : null; + } + + private static TEndPoint toEndPoint(SocketAddress socketAddress) { + if (!(socketAddress instanceof InetSocketAddress)) { + return null; + } + InetSocketAddress inetSocketAddress = (InetSocketAddress) socketAddress; + String host = + inetSocketAddress.getAddress() == null + ? inetSocketAddress.getHostString() + : inetSocketAddress.getAddress().getHostAddress(); + return new TEndPoint(host, inetSocketAddress.getPort()); + } +} diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/audit/AbstractAuditLoggerTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/audit/AbstractAuditLoggerTest.java new file mode 100644 index 00000000000..ccb8c937720 --- /dev/null +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/audit/AbstractAuditLoggerTest.java @@ -0,0 +1,114 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.commons.audit; + +import org.apache.iotdb.common.rpc.thrift.TEndPoint; + +import org.apache.thrift.TException; +import org.junit.Test; + +import javax.net.ssl.SSLException; +import javax.net.ssl.SSLHandshakeException; + +import java.io.IOException; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Supplier; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; + +public class AbstractAuditLoggerTest { + + @Test + public void testRecordTrustedChannelFailureAuditLog() { + TestAuditLogger auditLogger = new TestAuditLogger(); + UserEntity auditEntity = new UserEntity(0, "system", "127.0.0.1"); + AtomicInteger identifierEvaluationCount = new AtomicInteger(); + + auditLogger.recordTrustedChannelFailureAuditLog( + auditEntity, + () -> { + identifierEvaluationCount.incrementAndGet(); + return "[email protected]:10730"; + }, + () -> { + identifierEvaluationCount.incrementAndGet(); + return "[email protected]:10730"; + }); + + assertSame(auditEntity, auditLogger.auditEntity); + assertEquals( + AuditEventType.TRUSTED_CHANNEL_FUNCTION_FAILURE, + auditLogger.auditEntity.getAuditEventType()); + assertEquals(AuditLogOperation.CONTROL, auditLogger.auditEntity.getAuditLogOperation()); + assertFalse(auditLogger.auditEntity.getResult()); + assertEquals(0, identifierEvaluationCount.get()); + assertEquals( + "Trusted channel function failed: [email protected]:10730, [email protected]:10730", + auditLogger.auditLog.get()); + assertEquals(2, identifierEvaluationCount.get()); + } + + @Test + public void testIsSslFailure() { + assertTrue(AbstractAuditLogger.isSslFailure(new SSLException("ssl failure"))); + assertTrue( + AbstractAuditLogger.isSslFailure( + new TException(new IOException(new SSLHandshakeException("handshake failure"))))); + assertFalse( + AbstractAuditLogger.isSslFailure(new TException(new IOException("network failure")))); + assertFalse(AbstractAuditLogger.isSslFailure(null)); + } + + @Test + public void testRecordTrustedChannelFailureAuditLogIfNecessary() { + TestAuditLogger auditLogger = new TestAuditLogger(); + TEndPoint initiator = new TEndPoint("10.0.0.1", 10730); + TEndPoint target = new TEndPoint("10.0.0.2", 10730); + + auditLogger.recordTrustedChannelFailureAuditLogIfNecessary( + new IOException("network failure"), initiator, target); + assertNull(auditLogger.auditEntity); + + auditLogger.recordTrustedChannelFailureAuditLogIfNecessary( + new SSLHandshakeException("handshake failure"), initiator, target); + assertEquals( + AuditEventType.TRUSTED_CHANNEL_FUNCTION_FAILURE, + auditLogger.auditEntity.getAuditEventType()); + assertEquals( + "Trusted channel function failed: initiator=10.0.0.1:10730, target=10.0.0.2:10730", + auditLogger.auditLog.get()); + } + + private static class TestAuditLogger extends AbstractAuditLogger { + + private IAuditEntity auditEntity; + private Supplier<String> auditLog; + + @Override + public void log(IAuditEntity auditLogFields, Supplier<String> log) { + auditEntity = auditLogFields; + auditLog = log; + } + } +} diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/service/TrustedChannelAuditServerEventHandlerTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/service/TrustedChannelAuditServerEventHandlerTest.java new file mode 100644 index 00000000000..091f4b0d123 --- /dev/null +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/service/TrustedChannelAuditServerEventHandlerTest.java @@ -0,0 +1,121 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.commons.service; + +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.commons.audit.TrustedChannelFailureHandler; +import org.apache.iotdb.rpc.TElasticFramedTransport; + +import org.apache.thrift.protocol.TProtocol; +import org.apache.thrift.server.ServerContext; +import org.apache.thrift.server.TServerEventHandler; +import org.apache.thrift.transport.TSocket; +import org.junit.Test; +import org.mockito.InOrder; + +import javax.net.ssl.SSLHandshakeException; +import javax.net.ssl.SSLSocket; + +import java.io.UncheckedIOException; +import java.net.InetSocketAddress; +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThrows; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.isNull; +import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +public class TrustedChannelAuditServerEventHandlerTest { + + @Test + public void testHandshakeBeforeCreatingDelegateContext() throws Exception { + TServerEventHandler delegate = mock(TServerEventHandler.class); + ServerContext context = mock(ServerContext.class); + TProtocol input = mock(TProtocol.class); + SSLSocket socket = mock(SSLSocket.class); + TProtocol output = createProtocol(socket); + TEndPoint target = new TEndPoint("10.0.0.2", 10730); + when(delegate.createContext(input, output)).thenReturn(context); + + TrustedChannelAuditServerEventHandler handler = + new TrustedChannelAuditServerEventHandler( + delegate, target, TrustedChannelFailureHandler.NO_OP); + + assertSame(context, handler.createContext(input, output)); + InOrder inOrder = inOrder(socket, delegate); + inOrder.verify(socket).startHandshake(); + inOrder.verify(delegate).createContext(input, output); + } + + @Test + public void testHandshakeFailureReportsPeerAndSkipsDelegate() throws Exception { + TServerEventHandler delegate = mock(TServerEventHandler.class); + TProtocol input = mock(TProtocol.class); + SSLSocket socket = mock(SSLSocket.class); + TProtocol output = createProtocol(socket); + SSLHandshakeException failure = new SSLHandshakeException("handshake failure"); + InetSocketAddress remoteAddress = new InetSocketAddress("192.0.2.10", 45123); + TEndPoint target = new TEndPoint("10.0.0.2", 10730); + AtomicReference<Throwable> reportedFailure = new AtomicReference<>(); + AtomicReference<TEndPoint> reportedInitiator = new AtomicReference<>(); + AtomicReference<TEndPoint> reportedTarget = new AtomicReference<>(); + when(socket.getRemoteSocketAddress()).thenReturn(remoteAddress); + org.mockito.Mockito.doThrow(failure).when(socket).startHandshake(); + + TrustedChannelAuditServerEventHandler handler = + new TrustedChannelAuditServerEventHandler( + delegate, + target, + (throwable, initiator, endpoint) -> { + reportedFailure.set(throwable); + reportedInitiator.set(initiator); + reportedTarget.set(endpoint); + }); + + UncheckedIOException thrown = + assertThrows(UncheckedIOException.class, () -> handler.createContext(input, output)); + + assertSame(failure, thrown.getCause()); + assertSame(failure, reportedFailure.get()); + assertEquals(new TEndPoint("192.0.2.10", 45123), reportedInitiator.get()); + assertSame(target, reportedTarget.get()); + verify(socket).close(); + verify(delegate, never()).createContext(any(), any()); + + handler.deleteContext(null, input, output); + verify(delegate, never()).deleteContext(isNull(), any(), any()); + } + + private static TProtocol createProtocol(SSLSocket socket) { + TProtocol protocol = mock(TProtocol.class); + TElasticFramedTransport framedTransport = mock(TElasticFramedTransport.class); + TSocket socketTransport = mock(TSocket.class); + when(protocol.getTransport()).thenReturn(framedTransport); + when(framedTransport.getSocket()).thenReturn(socketTransport); + when(socketTransport.getSocket()).thenReturn(socket); + return protocol; + } +} diff --git a/pom.xml b/pom.xml index 7dc7e38602b..7da51d6e3c2 100644 --- a/pom.xml +++ b/pom.xml @@ -571,6 +571,11 @@ <artifactId>jetty-http</artifactId> <version>${jetty.version}</version> </dependency> + <dependency> + <groupId>org.eclipse.jetty</groupId> + <artifactId>jetty-io</artifactId> + <version>${jetty.version}</version> + </dependency> <dependency> <groupId>org.eclipse.jetty</groupId> <artifactId>jetty-util</artifactId>
