diff --git a/external-service-impl/rest/pom.xml b/external-service-impl/rest/pom.xml index f6cedd0079684..fc064d28254ec 100644 --- a/external-service-impl/rest/pom.xml +++ b/external-service-impl/rest/pom.xml @@ -98,6 +98,10 @@ org.eclipse.jetty jetty-http + + org.eclipse.jetty + jetty-io + org.apache.iotdb node-commons 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 506facdb4ce25..6ecba9aece8c6 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 @@ private void startSSL( 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 0000000000000..dcf84fd1cd298 --- /dev/null +++ b/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/TrustedChannelAuditHandshakeListener.java @@ -0,0 +1,75 @@ +/* + * 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) { + if (auditFailure != failure) { + 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 0000000000000..fbfdfca1c9d5d --- /dev/null +++ b/external-service-impl/rest/src/test/java/org/apache/iotdb/rest/TrustedChannelAuditHandshakeListenerTest.java @@ -0,0 +1,96 @@ +/* + * 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 actualFailure = new AtomicReference<>(); + AtomicReference actualInitiator = new AtomicReference<>(); + AtomicReference 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()); + } + + @Test + public void testReportingFailureDoesNotSelfSuppressHandshakeFailure() { + RuntimeException failure = new RuntimeException("test"); + TrustedChannelAuditHandshakeListener listener = + new TrustedChannelAuditHandshakeListener( + (ignoredFailure, initiator, target) -> { + throw failure; + }); + EndPoint endPoint = + newEndPoint( + new InetSocketAddress("192.0.2.10", 45123), new InetSocketAddress("192.0.2.20", 18080)); + + listener.recordHandshakeFailure(endPoint, failure); + + assertEquals(0, failure.getSuppressed().length); + } + + 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 ccc1008eec311..2b9661d373c63 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,6 +19,7 @@ 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.confignode.conf.ConfigNodeConfig; @@ -43,4 +44,9 @@ public CNAuditLogger(ConfigManager configManager) { @Override public void log(IAuditEntity auditLogFields, Supplier log) {} + + public void recordTrustedChannelFailureAuditLogIfNecessary(Throwable failure, TEndPoint target) { + 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 516596a61b8a9..0c13761598671 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.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 void getDataNodeHeartBeat( client.getDataNodeHeartBeat(req, handler); dispatched = true; } catch (Exception e) { + recordTrustedChannelFailureAuditLogIfNecessary(e, endPoint); handleError(handler, e); } finally { returnClientIfNotDispatched(endPoint, client, dispatched); @@ -105,6 +107,17 @@ private void handleError(final DataNodeWriteAuditLogHandler handler, final Excep } } + 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 9c90c469137de..a651fd39b5840 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.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 @@ protected void adjustClientTimeoutIfNecessary( } } + @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 f35b6d906a525..4611636f51f5a 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.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 Object sendSyncRequestToDataNodeWithRetry( 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 Object sendSyncRequestToDataNodeWithGivenRetry( 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 TRegionLeaderChangeResp changeRegionLeader( 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/manager/pipe/sink/client/IoTDBConfigNodeSyncClientManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/sink/client/IoTDBConfigNodeSyncClientManager.java index 25c0bc58b6142..ad937673cc24c 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/sink/client/IoTDBConfigNodeSyncClientManager.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/sink/client/IoTDBConfigNodeSyncClientManager.java @@ -85,4 +85,16 @@ protected PipeTransferHandshakeV2Req buildHandshakeV2Req(Map par protected String getClusterId() { return ConfigNode.getInstance().getConfigManager().getClusterManager().getClusterId(); } + + @Override + protected void onClientConnectionFailure( + final Exception failure, final TEndPoint targetEndPoint) { + if (ConfigNode.getInstance() == null || ConfigNode.getInstance().getConfigManager() == null) { + return; + } + ConfigNode.getInstance() + .getConfigManager() + .getAuditLogger() + .recordTrustedChannelFailureAuditLogIfNecessary(failure, targetEndPoint); + } } 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 15be2e90d43f1..f325da85ab083 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.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 void initThriftServiceThread() throws IllegalAccessException { 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 d54299992fe68..114a8aed4fcfb 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 @@ 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 @@ private 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 @@ private ConsensusConfig( this.iotConsensusConfig = iotConsensusConfig; this.iotConsensusV2Config = iotConsensusV2Config; this.directoryStrategyType = directoryStrategyType; + this.trustedChannelFailureHandler = trustedChannelFailureHandler; } public TEndPoint getThisNodeEndPoint() { @@ -95,6 +99,10 @@ public DirectoryStrategyType getDirectoryStrategyType() { return directoryStrategyType; } + public TrustedChannelFailureHandler getTrustedChannelFailureHandler() { + return trustedChannelFailureHandler; + } + public static ConsensusConfig.Builder newBuilder() { return new ConsensusConfig.Builder(); } @@ -111,6 +119,8 @@ public static class Builder { 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 ConsensusConfig build() { .orElseGet(() -> IoTConsensusConfig.newBuilder().build()), Optional.ofNullable(iotConsensusV2Config) .orElseGet(() -> IoTConsensusV2Config.newBuilder().build()), - directoryStrategyType); + directoryStrategyType, + trustedChannelFailureHandler); } public Builder setThisNode(TEndPoint thisNode) { @@ -171,5 +182,13 @@ public Builder setDirectoryStrategyType(DirectoryStrategyType directoryStrategyT 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 1e247ad599e1c..ce21fa0c8dbc8 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 IoTConsensus(ConsensusConfig config, Registry registry) { 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() .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 cfa175b2de585..15590dcd2bac4 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 void initThriftServiceThread() 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 cf1d405639838..8d2803093a3a2 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,11 @@ public IoTConsensusV2(ConsensusConfig config, IStateMachine.Registry registry) { this.storageDir = new File(config.getStorageDir()); this.config = config.getIoTConsensusV2Config(); this.registry = registry; - this.rpcService = new IoTConsensusV2RPCService(thisNode, config.getIoTConsensusV2Config()); + IoTV2GlobalComponentContainer.getInstance() + .configureTrustedChannelFailureHandler(thisNode, config.getTrustedChannelFailureHandler()); + 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 c218c8051e800..2266781307645 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 void initThriftServiceThread() throws IllegalAccessException { 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 90026e37ba0fe..41d371f4facb6 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,13 @@ 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.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 +60,20 @@ public void createViewIfNecessary() {} @Override public synchronized void log(IAuditEntity auditLogFields, Supplier log) {} + /** + * Records an outbound trusted-channel failure using the DataNode internal endpoint as the stable + * initiator identifier. Call the three-argument overload when the transport has a distinct local + * service endpoint. + */ + public void recordTrustedChannelFailureAuditLogIfNecessary(Throwable failure, TEndPoint target) { + 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 a4c8e00f1ad06..fbda403b320d7 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.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 @@ private static ConsensusConfig buildConsensusConfig() { 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/pipe/sink/client/IoTDBDataNodeSyncClientManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeSyncClientManager.java index e8f081cff0035..c56706bab5a4e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeSyncClientManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeSyncClientManager.java @@ -25,6 +25,7 @@ import org.apache.iotdb.commons.pipe.sink.client.IoTDBSyncClient; import org.apache.iotdb.commons.pipe.sink.client.IoTDBSyncClientManager; import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeTransferHandshakeV2Req; +import org.apache.iotdb.db.audit.DNAuditLogger; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferDataNodeHandshakeV1Req; @@ -98,6 +99,13 @@ protected String getClusterId() { return IoTDBDescriptor.getInstance().getConfig().getClusterId(); } + @Override + protected void onClientConnectionFailure( + final Exception failure, final TEndPoint targetEndPoint) { + DNAuditLogger.getInstance() + .recordTrustedChannelFailureAuditLogIfNecessary(failure, targetEndPoint); + } + public Pair getClient(final String deviceId) { final TEndPoint endPoint = LEADER_CACHE_MANAGER.getLeaderEndPoint(deviceId); return useLeaderCache 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 a9fe5a60cd4cb..572cd65ca53f9 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.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 @@ private void tryToConnect(int timeoutMs) throws TException { 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 @@ private void tryToConnect(int timeoutMs) throws TException { return; } catch (TException e) { logger.warn(DataNodeMiscMessages.NODE_MAY_DOWN_TRY_NEXT, tryEndpoint); + recordTrustedChannelFailureAuditLogIfNecessary(e, tryEndpoint); exception = e; } } @@ -461,6 +464,7 @@ private T executeRemoteCallWithRetry(final Operation call, final Predicat 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 @@ private T executeRemoteCallWithRetry(final Operation call, final Predicat 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 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 8815acc036e0a..72f1886310b38 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.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 @@ private R executeRemoteCallWithRetry(RemoteCall call) throws TException { 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 @@ private void tryToConnect(int timeoutMs) { 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 void connect(TEndPoint endpoint, int timeoutMs) throws TException { 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 c69f6ea4e40dc..fefb4983dd9d1 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 @@ 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 void onComplete(TSStatus response) { @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 a4cef2e4b713b..6afcd2a3a918f 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.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 void initThriftServiceThread() throws IllegalAccessException { 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 ec6c9c680a1b5..f55f07fe59c9d 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,8 +23,10 @@ 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.DataNodeEndPoints; import org.apache.iotdb.db.queryengine.common.FragmentInstanceId; import org.apache.iotdb.db.queryengine.exception.exchange.GetTsBlockFromClosedOrAbortedChannelException; import org.apache.iotdb.db.queryengine.execution.exchange.MPPDataExchangeManager.SinkListener; @@ -579,6 +581,9 @@ public void run() { client.onNewDataBlockEvent(newDataBlockEvent); break; } catch (Exception e) { + DNAuditLogger.getInstance() + .recordTrustedChannelFailureAuditLogIfNecessary( + e, DataNodeEndPoints.LOCAL_HOST_DATA_BLOCK_ENDPOINT, remoteEndpoint); LOGGER.warn( DataNodeQueryMessages.FAILED_TO_SEND_NEW_DATA_BLOCK_EVENT_ATTEMPT, attempt, e); if (attempt == MAX_ATTEMPT_TIMES) { @@ -627,6 +632,9 @@ public void run() { client.onEndOfDataBlockEvent(endOfDataBlockEvent); break; } catch (Exception e) { + DNAuditLogger.getInstance() + .recordTrustedChannelFailureAuditLogIfNecessary( + e, DataNodeEndPoints.LOCAL_HOST_DATA_BLOCK_ENDPOINT, 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 4f90db0caed41..243e947108c27 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,8 +23,10 @@ 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.DataNodeEndPoints; import org.apache.iotdb.db.queryengine.common.FragmentInstanceId; import org.apache.iotdb.db.queryengine.execution.exchange.MPPDataExchangeManager.SourceHandleListener; import org.apache.iotdb.db.queryengine.execution.memory.LocalMemoryManager; @@ -663,6 +665,9 @@ public void run() { } break; } catch (Throwable e) { + DNAuditLogger.getInstance() + .recordTrustedChannelFailureAuditLogIfNecessary( + e, DataNodeEndPoints.LOCAL_HOST_DATA_BLOCK_ENDPOINT, remoteEndpoint); LOGGER.warn( DataNodeQueryMessages.FAILED_TO_GET_DATA_BLOCK, @@ -744,6 +749,9 @@ public void run() { client.onAcknowledgeDataBlockEvent(acknowledgeDataBlockEvent); break; } catch (Throwable e) { + DNAuditLogger.getInstance() + .recordTrustedChannelFailureAuditLogIfNecessary( + e, DataNodeEndPoints.LOCAL_HOST_DATA_BLOCK_ENDPOINT, remoteEndpoint); LOGGER.warn( DataNodeQueryMessages.FAILED_TO_SEND_ACK_DATA_BLOCK_EVENT, startSequenceId, @@ -795,6 +803,9 @@ public void run() { client.onCloseSinkChannelEvent(closeSinkChannelEvent); break; } catch (Throwable e) { + DNAuditLogger.getInstance() + .recordTrustedChannelFailureAuditLogIfNecessary( + e, DataNodeEndPoints.LOCAL_HOST_DATA_BLOCK_ENDPOINT, 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 da5a2dfc889b1..00611993e126b 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 void sendAll() { 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 23f99da3484c2..e73e4f6e1cdb0 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 instanceId2RespMap; private final List needRetryInstanceIndex; private final long sendTime; + private final TEndPoint targetEndPoint; private static final PerformanceOverviewMetrics PERFORMANCE_OVERVIEW_METRICS = PerformanceOverviewMetrics.getInstance(); @@ -51,12 +54,14 @@ public AsyncSendPlanNodeHandler( AtomicLong pendingNumber, Map instanceId2RespMap, List 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 void onComplete(TSendBatchPlanNodeResp sendBatchPlanNodeResp) { @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 df719a724fe4b..c76912ced2409 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.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 @@ private void dispatchRemote(FragmentInstance instance, TEndPoint endPoint) 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 @@ private void dispatchRemote(FragmentInstance instance, TEndPoint endPoint) | 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 b58d647fcd7dc..f9b9c6e010110 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.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 void initThriftServiceThread() throws IllegalAccessException { 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 f9b34dd9e58b1..782788547bf0e 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.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 void initThriftServiceThread() throws IllegalAccessException { getBindPort(), config.getRpcMaxConcurrentClientNum(), config.getThriftServerAwaitTimeForStopService(), - new RPCServiceThriftHandler(impl), + newTrustedChannelAuditHandler(), config.isRpcThriftCompressionEnable(), commonConfig.getKeyStorePath(), commonConfig.getKeyStorePwd(), @@ -111,7 +114,7 @@ public void initThriftServiceThread() throws IllegalAccessException { getBindPort(), config.getRpcMaxConcurrentClientNum(), config.getThriftServerAwaitTimeForStopService(), - new RPCServiceThriftHandler(impl), + newTrustedChannelAuditHandler(), config.isRpcThriftCompressionEnable(), commonConfig.getKeyStorePath(), commonConfig.getKeyStorePwd(), @@ -150,6 +153,13 @@ private boolean hasText(String value) { 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 ae97d09ddc252..94194a086b47c 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 @@ private 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 71c45ccaf3fc0..f0b84bbc27901 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 @@ private 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 d7c00c9128b8d..0089bf9eddb6e 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 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,68 @@ public void recordObjectAuthenticationAuditLog( auditObject.get(), auditEntity.getResult())); } + + /** + * Records a failure of the trusted-channel function. + * + *

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 initiator, + final Supplier 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); + } catch (RuntimeException auditFailure) { + if (auditFailure != failure) { + failure.addSuppressed(auditFailure); + } + } 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 510912a302fdc..2c99e436676b8 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/TrustedChannelFailureHandler.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/audit/TrustedChannelFailureHandler.java new file mode 100644 index 0000000000000..e342cc8d231b2 --- /dev/null +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/audit/TrustedChannelFailureHandler.java @@ -0,0 +1,30 @@ +/* + * 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; + +@FunctionalInterface +public interface TrustedChannelFailureHandler { + + 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/async/AsyncIoTConsensusV2ServiceClient.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/async/AsyncIoTConsensusV2ServiceClient.java index cc8b6f6e1bfbf..0214467dbdf47 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/async/AsyncIoTConsensusV2ServiceClient.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/async/AsyncIoTConsensusV2ServiceClient.java @@ -26,6 +26,7 @@ import org.apache.iotdb.commons.client.property.ThriftClientProperty; import org.apache.iotdb.commons.conf.CommonConfig; import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.commons.consensus.iotv2.container.IoTV2GlobalComponentContainer; import org.apache.iotdb.commons.i18n.ClientMessages; import org.apache.iotdb.consensus.iotconsensusv2.thrift.IoTConsensusV2IService; import org.apache.iotdb.rpc.TNonblockingTransportWrapper; @@ -89,6 +90,7 @@ public void onComplete() { @Override public void onError(Exception e) { + IoTV2GlobalComponentContainer.getInstance().reportTrustedChannelFailure(e, endpoint); super.onError(e); ThriftClient.resolveException(e, this); returnSelf(); @@ -167,12 +169,17 @@ public void destroyObject( @Override public PooledObject makeObject(TEndPoint endPoint) throws Exception { - return new DefaultPooledObject<>( - new AsyncIoTConsensusV2ServiceClient( - thriftClientProperty, - endPoint, - tManagers[Math.floorMod(clientCnt.incrementAndGet(), tManagers.length)], - clientManager)); + try { + return new DefaultPooledObject<>( + new AsyncIoTConsensusV2ServiceClient( + thriftClientProperty, + endPoint, + tManagers[Math.floorMod(clientCnt.incrementAndGet(), tManagers.length)], + clientManager)); + } catch (final Exception e) { + IoTV2GlobalComponentContainer.getInstance().reportTrustedChannelFailure(e, endPoint); + throw e; + } } @Override 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 b13d1ab06373d..e8a47db01f640 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,13 @@ protected void sendAsyncRequest( endPoint, e.getMessage(), retryCount); + try { + onRequestFailure(e, endPoint); + } catch (final RuntimeException reportingFailure) { + if (reportingFailure != e) { + e.addSuppressed(reportingFailure); + } + } if (handler != null) { try { handler.onError(e); @@ -237,6 +244,10 @@ protected void adjustClientTimeoutIfNecessary(RequestType type, Client client, L // 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 buildHandler( diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/sync/SyncIoTConsensusV2ServiceClient.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/sync/SyncIoTConsensusV2ServiceClient.java index 0b74424277593..caae534840052 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/sync/SyncIoTConsensusV2ServiceClient.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/sync/SyncIoTConsensusV2ServiceClient.java @@ -26,6 +26,7 @@ import org.apache.iotdb.commons.client.property.ThriftClientProperty; import org.apache.iotdb.commons.conf.CommonConfig; import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.commons.consensus.iotv2.container.IoTV2GlobalComponentContainer; import org.apache.iotdb.consensus.iotconsensusv2.thrift.IoTConsensusV2IService; import org.apache.iotdb.rpc.DeepCopyRpcTransportFactory; import org.apache.iotdb.rpc.TConfigurationConst; @@ -135,14 +136,22 @@ public void destroyObject( @Override public PooledObject makeObject(TEndPoint endpoint) throws Exception { - return new DefaultPooledObject<>( - SyncThriftClientWithErrorHandler.newErrorHandler( - SyncIoTConsensusV2ServiceClient.class, - SyncIoTConsensusV2ServiceClient.class.getConstructor( - thriftClientProperty.getClass(), endpoint.getClass(), clientManager.getClass()), - thriftClientProperty, - endpoint, - clientManager)); + try { + return new DefaultPooledObject<>( + SyncThriftClientWithErrorHandler.newErrorHandlerWithFailureHandler( + SyncIoTConsensusV2ServiceClient.class, + SyncIoTConsensusV2ServiceClient.class.getConstructor( + thriftClientProperty.getClass(), endpoint.getClass(), clientManager.getClass()), + (failure, client) -> + IoTV2GlobalComponentContainer.getInstance() + .reportTrustedChannelFailure(failure, client.getTEndpoint()), + thriftClientProperty, + endpoint, + clientManager)); + } catch (final Exception e) { + IoTV2GlobalComponentContainer.getInstance().reportTrustedChannelFailure(e, endpoint); + throw e; + } } @Override diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/sync/SyncThriftClientWithErrorHandler.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/sync/SyncThriftClientWithErrorHandler.java index f85d1421547f5..77b25895cb2f6 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/sync/SyncThriftClientWithErrorHandler.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/sync/SyncThriftClientWithErrorHandler.java @@ -29,9 +29,20 @@ import java.lang.reflect.Constructor; import java.lang.reflect.Method; +import java.util.function.BiConsumer; public class SyncThriftClientWithErrorHandler implements MethodInterceptor { + private static final BiConsumer NO_OP_FAILURE_HANDLER = + (failure, client) -> {}; + + private final BiConsumer failureHandler; + + private SyncThriftClientWithErrorHandler( + final BiConsumer failureHandler) { + this.failureHandler = failureHandler; + } + /** * Note: The caller needs to ensure that the constructor corresponds to the class, or the cast * might fail. @@ -39,9 +50,31 @@ public class SyncThriftClientWithErrorHandler implements MethodInterceptor { @SuppressWarnings("unchecked") public static V newErrorHandler( Class targetClass, Constructor constructor, Object... args) { + return newErrorHandler(targetClass, constructor, NO_OP_FAILURE_HANDLER, args); + } + + @SuppressWarnings("unchecked") + public static V newErrorHandlerWithFailureHandler( + Class targetClass, + Constructor constructor, + BiConsumer failureHandler, + Object... args) { + return newErrorHandler( + targetClass, + constructor, + (failure, client) -> failureHandler.accept(failure, (V) client), + args); + } + + @SuppressWarnings("unchecked") + private static V newErrorHandler( + Class targetClass, + Constructor constructor, + BiConsumer failureHandler, + Object... args) { Enhancer enhancer = new Enhancer(); enhancer.setSuperclass(targetClass); - enhancer.setCallback(new SyncThriftClientWithErrorHandler()); + enhancer.setCallback(new SyncThriftClientWithErrorHandler(failureHandler)); if (constructor == null) { return (V) enhancer.create(); } @@ -54,6 +87,13 @@ public Object intercept(Object o, Method method, Object[] objects, MethodProxy m try { return methodProxy.invokeSuper(o, objects); } catch (Throwable t) { + try { + failureHandler.accept(t, (ThriftClient) o); + } catch (final RuntimeException reportingFailure) { + if (reportingFailure != t) { + t.addSuppressed(reportingFailure); + } + } ThriftClient.resolveException(t, (ThriftClient) o); throw new TException( ClientMessages.EXCEPTION_ERROR_CALLING_METHOD_C04E5A63 diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/iotv2/container/IoTV2GlobalComponentContainer.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/iotv2/container/IoTV2GlobalComponentContainer.java index fdf34e434cc85..bb63be960fc6d 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/iotv2/container/IoTV2GlobalComponentContainer.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/iotv2/container/IoTV2GlobalComponentContainer.java @@ -20,6 +20,7 @@ package org.apache.iotdb.commons.consensus.iotv2.container; import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.commons.audit.TrustedChannelFailureHandler; import org.apache.iotdb.commons.client.ClientPoolFactory.AsyncIoTConsensusV2ServiceClientPoolFactory; import org.apache.iotdb.commons.client.ClientPoolFactory.SyncIoTConsensusV2ServiceClientPoolFactory; import org.apache.iotdb.commons.client.IClientManager; @@ -36,6 +37,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.util.Objects; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; @@ -54,6 +56,9 @@ public class IoTV2GlobalComponentContainer { private final IClientManager asyncClientManager; private final IClientManager syncClientManager; private final ScheduledExecutorService backgroundTaskService; + private volatile TEndPoint thisNode; + private volatile TrustedChannelFailureHandler trustedChannelFailureHandler = + TrustedChannelFailureHandler.NO_OP; private PipeSubtaskExecutor consensusExecutor; private IoTV2GlobalComponentContainer() { @@ -87,6 +92,26 @@ public ScheduledExecutorService getBackgroundTaskService() { return this.backgroundTaskService; } + public void configureTrustedChannelFailureHandler( + final TEndPoint thisNode, final TrustedChannelFailureHandler trustedChannelFailureHandler) { + this.thisNode = Objects.requireNonNull(thisNode); + this.trustedChannelFailureHandler = Objects.requireNonNull(trustedChannelFailureHandler); + } + + public void reportTrustedChannelFailure(final Throwable failure, final TEndPoint targetEndPoint) { + final TEndPoint initiatorEndPoint = thisNode; + if (failure == null || initiatorEndPoint == null || targetEndPoint == null) { + return; + } + try { + trustedChannelFailureHandler.onFailure(failure, initiatorEndPoint, targetEndPoint); + } catch (final RuntimeException reportingFailure) { + if (reportingFailure != failure) { + failure.addSuppressed(reportingFailure); + } + } + } + public void stopBackgroundTaskService() { backgroundTaskService.shutdownNow(); try { diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/client/IoTDBSyncClientManager.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/client/IoTDBSyncClientManager.java index 83b85a6d58303..d34f84fe9d685 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/client/IoTDBSyncClientManager.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/client/IoTDBSyncClientManager.java @@ -33,6 +33,7 @@ import org.apache.iotdb.rpc.TSStatusCode; import org.apache.iotdb.service.rpc.thrift.TPipeTransferResp; +import org.apache.thrift.transport.TTransportException; import org.apache.tsfile.utils.Pair; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -200,22 +201,10 @@ protected void reconstructClient(TEndPoint endPoint) { private boolean initClientAndStatus( final Pair clientAndStatus, final TEndPoint endPoint) { try { - clientAndStatus.setLeft( - new IoTDBSyncClient( - new ThriftClientProperty.Builder() - .setConnectionTimeoutMs(PIPE_CONFIG.getPipeSinkHandshakeTimeoutMs()) - .setRpcThriftCompressionEnabled( - PIPE_CONFIG.isPipeSinkRPCThriftCompressionEnabled()) - .build(), - endPoint.getIp(), - endPoint.getPort(), - useSSL, - trustStorePath, - trustStorePwd, - keyStorePath, - keyStorePwd)); + clientAndStatus.setLeft(createClient(endPoint)); return true; } catch (Exception e) { + notifyClientConnectionFailure(e, endPoint); endPoint2HandshakeErrorMessage.put(endPoint, e.getMessage()); PipeLogger.log( LOGGER::warn, @@ -228,6 +217,21 @@ private boolean initClientAndStatus( } } + protected IoTDBSyncClient createClient(final TEndPoint endPoint) throws TTransportException { + return new IoTDBSyncClient( + new ThriftClientProperty.Builder() + .setConnectionTimeoutMs(PIPE_CONFIG.getPipeSinkHandshakeTimeoutMs()) + .setRpcThriftCompressionEnabled(PIPE_CONFIG.isPipeSinkRPCThriftCompressionEnabled()) + .build(), + endPoint.getIp(), + endPoint.getPort(), + useSSL, + trustStorePath, + trustStorePwd, + keyStorePath, + keyStorePwd); + } + public void sendHandshakeReq(final Pair clientAndStatus) { final IoTDBSyncClient client = clientAndStatus.getLeft(); try { @@ -289,6 +293,7 @@ public void sendHandshakeReq(final Pair clientAndStatu LOGGER.info(PipeMessages.HANDSHAKE_SUCCESS_TARGET, client.getIpAddress(), client.getPort()); } } catch (Exception e) { + notifyClientConnectionFailure(e, client.getEndPoint()); LOGGER.warn( PipeMessages.HANDSHAKE_ERROR_WITH_TARGET_SERVER, client.getIpAddress(), @@ -299,6 +304,22 @@ public void sendHandshakeReq(final Pair clientAndStatu } } + private void notifyClientConnectionFailure( + final Exception failure, final TEndPoint targetEndPoint) { + try { + onClientConnectionFailure(failure, targetEndPoint); + } catch (final RuntimeException reportingFailure) { + if (reportingFailure != failure) { + failure.addSuppressed(reportingFailure); + } + } + } + + protected void onClientConnectionFailure( + final Exception failure, final TEndPoint targetEndPoint) { + // In default, no need to do this + } + protected abstract PipeTransferHandshakeV1Req buildHandshakeV1Req() throws IOException; protected abstract PipeTransferHandshakeV2Req buildHandshakeV2Req(Map params) 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 0000000000000..c1063728c8f7d --- /dev/null +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/TrustedChannelAuditServerEventHandler.java @@ -0,0 +1,139 @@ +/* + * 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) { + if (closeFailure != e) { + 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) { + if (auditFailure != failure) { + 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 0000000000000..73d0fb5fda130 --- /dev/null +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/audit/AbstractAuditLoggerTest.java @@ -0,0 +1,145 @@ +/* + * 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 "DataNode-1@10.0.0.1:10730"; + }, + () -> { + identifierEvaluationCount.incrementAndGet(); + return "DataNode-2@10.0.0.2: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: initiator=DataNode-1@10.0.0.1:10730, target=DataNode-2@10.0.0.2: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()); + } + + @Test + public void testAuditFailureShouldNotReplaceTrustedChannelFailure() { + final RuntimeException auditFailure = new RuntimeException("audit failure"); + final ThrowingAuditLogger auditLogger = new ThrowingAuditLogger(auditFailure); + final SSLHandshakeException channelFailure = new SSLHandshakeException("handshake failure"); + final TEndPoint initiator = new TEndPoint("10.0.0.1", 10730); + final TEndPoint target = new TEndPoint("10.0.0.2", 10730); + + auditLogger.recordTrustedChannelFailureAuditLogIfNecessary(channelFailure, initiator, target); + + assertEquals(1, auditLogger.invocationCount.get()); + assertEquals(1, channelFailure.getSuppressed().length); + assertSame(auditFailure, channelFailure.getSuppressed()[0]); + } + + private static class TestAuditLogger extends AbstractAuditLogger { + + private IAuditEntity auditEntity; + private Supplier auditLog; + + @Override + public void log(IAuditEntity auditLogFields, Supplier log) { + auditEntity = auditLogFields; + auditLog = log; + } + } + + private static class ThrowingAuditLogger extends AbstractAuditLogger { + + private final RuntimeException auditFailure; + private final AtomicInteger invocationCount = new AtomicInteger(); + + private ThrowingAuditLogger(final RuntimeException auditFailure) { + this.auditFailure = auditFailure; + } + + @Override + public void log(final IAuditEntity auditLogFields, final Supplier log) { + invocationCount.incrementAndGet(); + throw auditFailure; + } + } +} diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/client/request/AsyncRequestManagerTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/client/request/AsyncRequestManagerTest.java index 40b2bf9730b7e..28c076b8964b6 100644 --- a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/client/request/AsyncRequestManagerTest.java +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/client/request/AsyncRequestManagerTest.java @@ -34,6 +34,7 @@ import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; public class AsyncRequestManagerTest { @@ -83,6 +84,23 @@ public void handlerErrorFailureShouldNotEscapeDispatchFailure() throws Exception Assert.assertEquals(Collections.singletonList(1), context.getRequestIndices()); } + @Test + public void reportingFailureShouldNotReplaceDispatchFailure() { + final TestAsyncRequestManager manager = new TestAsyncRequestManager(true, false, false, true); + final AsyncRequestContext context = + new AsyncRequestContext<>( + TestRequestType.TEST, + "request", + Collections.singletonMap(1, new TestNodeLocation(new TEndPoint("localhost", 6667)))); + + manager.sendAsyncRequest(context, 1, null, true); + + Assert.assertEquals("borrow failed", context.getResponseMap().get(1)); + Assert.assertEquals(1, manager.getReportedFailure().getSuppressed().length); + Assert.assertEquals( + "reporting failed", manager.getReportedFailure().getSuppressed()[0].getMessage()); + } + private enum TestRequestType { TEST } @@ -103,23 +121,38 @@ private static class TestAsyncRequestManager private boolean failOnBorrow; private boolean failOnDispatch; private boolean failOnError; + private boolean failOnReporting; + private final AtomicReference reportedFailure = new AtomicReference<>(); private TestAsyncRequestManager() { - this(true, false, false); + this(true, false, false, false); } private TestAsyncRequestManager( final boolean failOnBorrow, final boolean failOnDispatch, final boolean failOnError) { + this(failOnBorrow, failOnDispatch, failOnError, false); + } + + private TestAsyncRequestManager( + final boolean failOnBorrow, + final boolean failOnDispatch, + final boolean failOnError, + final boolean failOnReporting) { super(1); this.failOnBorrow = failOnBorrow; this.failOnDispatch = failOnDispatch; this.failOnError = failOnError; + this.failOnReporting = failOnReporting; } private int getBorrowAttempts() { return borrowAttempts.get(); } + private Exception getReportedFailure() { + return reportedFailure.get(); + } + @Override protected void initClientManager(final int selectorNumOfAsyncClientManager) { clientManager = @@ -168,6 +201,14 @@ protected TEndPoint nodeLocationToEndPoint(final TestNodeLocation location) { return location.endPoint; } + @Override + protected void onRequestFailure(final Exception failure, final TEndPoint targetEndPoint) { + reportedFailure.set(failure); + if (failOnReporting) { + throw new RuntimeException("reporting failed"); + } + } + @SuppressWarnings("unchecked") @Override protected AsyncRequestRPCHandler buildHandler( diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/sink/client/IoTDBSyncClientManagerTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/sink/client/IoTDBSyncClientManagerTest.java new file mode 100644 index 0000000000000..ea2e4cad25894 --- /dev/null +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/sink/client/IoTDBSyncClientManagerTest.java @@ -0,0 +1,113 @@ +/* + * 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.pipe.sink.client; + +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.commons.audit.UserEntity; +import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeTransferHandshakeV1Req; +import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeTransferHandshakeV2Req; + +import org.apache.thrift.transport.TTransportException; +import org.junit.Test; + +import javax.net.ssl.SSLHandshakeException; + +import java.io.IOException; +import java.util.Collections; +import java.util.Map; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; + +public class IoTDBSyncClientManagerTest { + + @Test + public void sslConnectionFailureShouldBeReportedWithoutReplacingPipeFailure() { + final TEndPoint target = new TEndPoint("127.0.0.1", 6667); + final TestIoTDBSyncClientManager clientManager = new TestIoTDBSyncClientManager(target); + + clientManager.reconstruct(target); + + assertSame(target, clientManager.reportedTarget); + assertTrue(clientManager.reportedFailure.getCause() instanceof SSLHandshakeException); + assertEquals(1, clientManager.reportedFailure.getSuppressed().length); + assertSame(clientManager.reportingFailure, clientManager.reportedFailure.getSuppressed()[0]); + } + + private static class TestIoTDBSyncClientManager extends IoTDBSyncClientManager { + + private final RuntimeException reportingFailure = new RuntimeException("reporting failed"); + private Exception reportedFailure; + private TEndPoint reportedTarget; + + private TestIoTDBSyncClientManager(final TEndPoint target) { + super( + Collections.singletonList(target), + true, + "trust-store", + "trust-store-password", + null, + null, + false, + "round-robin", + new UserEntity(0, "system", "127.0.0.1"), + "password", + false, + "sync", + false, + true, + false); + } + + private void reconstruct(final TEndPoint target) { + reconstructClient(target); + } + + @Override + protected IoTDBSyncClient createClient(final TEndPoint endPoint) throws TTransportException { + throw new TTransportException(new SSLHandshakeException("handshake failed")); + } + + @Override + protected void onClientConnectionFailure( + final Exception failure, final TEndPoint targetEndPoint) { + reportedFailure = failure; + reportedTarget = targetEndPoint; + throw reportingFailure; + } + + @Override + protected PipeTransferHandshakeV1Req buildHandshakeV1Req() throws IOException { + return null; + } + + @Override + protected PipeTransferHandshakeV2Req buildHandshakeV2Req(final Map params) + throws IOException { + return null; + } + + @Override + protected String getClusterId() { + return "test"; + } + } +} 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 0000000000000..ede6d64cbc440 --- /dev/null +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/service/TrustedChannelAuditServerEventHandlerTest.java @@ -0,0 +1,144 @@ +/* + * 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 reportedFailure = new AtomicReference<>(); + AtomicReference reportedInitiator = new AtomicReference<>(); + AtomicReference 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()); + } + + @Test + public void testCloseFailureDoesNotSelfSuppressHandshakeFailure() 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"); + when(socket.getRemoteSocketAddress()).thenReturn(new InetSocketAddress("192.0.2.10", 45123)); + org.mockito.Mockito.doThrow(failure).when(socket).startHandshake(); + org.mockito.Mockito.doThrow(failure).when(socket).close(); + + TrustedChannelAuditServerEventHandler handler = + new TrustedChannelAuditServerEventHandler( + delegate, new TEndPoint("10.0.0.2", 10730), TrustedChannelFailureHandler.NO_OP); + + UncheckedIOException thrown = + assertThrows(UncheckedIOException.class, () -> handler.createContext(input, output)); + + assertSame(failure, thrown.getCause()); + assertEquals(0, failure.getSuppressed().length); + verify(delegate, never()).createContext(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 7dc7e38602b84..7da51d6e3c2f3 100644 --- a/pom.xml +++ b/pom.xml @@ -571,6 +571,11 @@ jetty-http ${jetty.version} + + org.eclipse.jetty + jetty-io + ${jetty.version} + org.eclipse.jetty jetty-util