Pipe: add information about sender's IP and port in the pipe receiver logs (#14343) (#14351)

Co-authored-by: Steve Yurong Su <rong@apache.org>

Author: XNX02

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/receiver/protocol/IoTDBConfigNodeReceiver.java

@@ -68,6 +68,8 @@
 import org.apache.iotdb.confignode.rpc.thrift.TUnsetSchemaTemplateReq;
 import org.apache.iotdb.confignode.service.ConfigNode;
 import org.apache.iotdb.consensus.exception.ConsensusException;
+import org.apache.iotdb.db.protocol.session.IClientSession;
+import org.apache.iotdb.db.protocol.session.SessionManager;
 import org.apache.iotdb.db.queryengine.common.header.ColumnHeaderConstant;
 import org.apache.iotdb.rpc.RpcUtils;
 import org.apache.iotdb.rpc.TSStatusCode;
@@ -91,6 +93,8 @@ public class IoTDBConfigNodeReceiver extends IoTDBFileReceiver {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBConfigNodeReceiver.class);
 
+  private static final SessionManager SESSION_MANAGER = SessionManager.getInstance();
+
   private static final AtomicInteger QUERY_ID_GENERATOR = new AtomicInteger(0);
 
   private static final PipeConfigPhysicalPlanTSStatusVisitor STATUS_VISITOR =
@@ -329,6 +333,18 @@ protected String getReceiverFileBaseDir() {
     return ConfigNodeDescriptor.getInstance().getConf().getPipeReceiverFileDir();
   }
 
+  @Override
+  protected String getSenderHost() {
+    final IClientSession session = SESSION_MANAGER.getCurrSession();
+    return session != null ? session.getClientAddress() : "unknown";
+  }
+
+  @Override
+  protected String getSenderPort() {
+    final IClientSession session = SESSION_MANAGER.getCurrSession();
+    return session != null ? String.valueOf(session.getClientPort()) : "unknown";
+  }
+
   @Override
   protected TSStatus loadFileV1(
       final PipeTransferFileSealReqV1 req, final String fileAbsolutePath) {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java

@@ -410,6 +410,18 @@ protected String getReceiverFileBaseDir() throws DiskSpaceInsufficientException
     return Objects.isNull(folderManager) ? null : folderManager.getNextFolder();
   }
 
+  @Override
+  protected String getSenderHost() {
+    final IClientSession session = SESSION_MANAGER.getCurrSession();
+    return session != null ? session.getClientAddress() : "unknown";
+  }
+
+  @Override
+  protected String getSenderPort() {
+    final IClientSession session = SESSION_MANAGER.getCurrSession();
+    return session != null ? String.valueOf(session.getClientPort()) : "unknown";
+  }
+
   @Override
   protected TSStatus loadFileV1(final PipeTransferFileSealReqV1 req, final String fileAbsolutePath)
       throws IOException {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/session/IClientSession.java

@@ -45,7 +45,7 @@ public abstract class IClientSession {
 
   public abstract String getClientAddress();
 
-  abstract int getClientPort();
+  public abstract int getClientPort();
 
   abstract TSConnectionType getConnectionType();
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/session/RestClientSession.java

@@ -38,7 +38,7 @@ public String getClientAddress() {
   }
 
   @Override
-  int getClientPort() {
+  public int getClientPort() {
     return 0;
   }
 

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java

@@ -174,14 +174,20 @@ protected TPipeTransferResp handleTransferHandshakeV1(final PipeTransferHandshak
     receiverFileDirWithIdSuffix.set(newReceiverDir);
 
     LOGGER.info(
-        "Receiver id = {}: Handshake successfully, receiver file dir = {}.",
+        "Receiver id = {}: Handshake successfully! Sender's host = {}, port = {}. Receiver's file dir = {}.",
         receiverId.get(),
+        getSenderHost(),
+        getSenderPort(),
         newReceiverDir.getPath());
     return new TPipeTransferResp(RpcUtils.SUCCESS_STATUS);
   }
 
   protected abstract String getReceiverFileBaseDir() throws Exception;
 
+  protected abstract String getSenderHost();
+
+  protected abstract String getSenderPort();
+
   protected TPipeTransferResp handleTransferHandshakeV2(final PipeTransferHandshakeV2Req req)
       throws IOException {
     // Reject to handshake if the receiver can not take clusterId from config node.