justinwwhuang commented on code in PR #10366:
URL: https://github.com/apache/inlong/pull/10366#discussion_r1628859348
##########
inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/send/SenderManager.java:
##########
@@ -20,190 +20,201 @@
import org.apache.inlong.audit.protocol.AuditApi;
import org.apache.inlong.audit.util.AuditConfig;
import org.apache.inlong.audit.util.AuditData;
-import org.apache.inlong.audit.util.SenderResult;
-import io.netty.buffer.ByteBuf;
-import io.netty.buffer.ByteBufAllocator;
-import io.netty.channel.Channel;
-import io.netty.channel.ChannelHandlerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
+import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
-import java.security.SecureRandom;
-import java.util.ArrayList;
-import java.util.HashSet;
-import java.util.List;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.atomic.AtomicLong;
/**
* Audit sender manager
*/
public class SenderManager {
- public static final Long MAX_REQUEST_ID = 1000000000L;
- public static final int ALL_CONNECT_CHANNEL = -1;
- public static final int DEFAULT_CONNECT_CHANNEL = 2;
- public static final Logger LOG =
LoggerFactory.getLogger(SenderManager.class);
- private static final int SEND_INTERVAL_MS = 20;
- private final SecureRandom sRandom = new
SecureRandom(Long.toString(System.currentTimeMillis()).getBytes());
- private final AtomicLong requestIdSeq = new AtomicLong(0L);
- private final ConcurrentHashMap<Long, AuditData> dataMap = new
ConcurrentHashMap<>();
- private final LinkedBlockingQueue<Long> requestIdQueue = new
LinkedBlockingQueue<>();
-
- private SenderGroup sender;
- private int maxConnectChannels = ALL_CONNECT_CHANNEL;
- // IPList
- private List<String> currentIpPorts = new ArrayList<>();
+ private static final Logger LOGGER =
LoggerFactory.getLogger(SenderManager.class);
+ private static final int SEND_INTERVAL_MS = 100;
+ private final ConcurrentHashMap<Long, AuditData> failedDataMap = new
ConcurrentHashMap<>();
private AuditConfig auditConfig;
- private long lastCheckTime = System.currentTimeMillis();
+ private Socket socket = new Socket();
+ private static final int PACKAGE_HEADER_LEN = 4;
- /**
- * Constructor
- */
public SenderManager(AuditConfig config) {
- this(config, DEFAULT_CONNECT_CHANNEL);
- }
-
- /**
- * Constructor
- */
- public SenderManager(AuditConfig config, int maxConnectChannels) {
- try {
- this.auditConfig = config;
- this.maxConnectChannels = maxConnectChannels;
- this.sender = new SenderGroup(this);
- } catch (Exception ex) {
- LOG.error(ex.getMessage(), ex);
- }
+ auditConfig = config;
}
- /**
- * update config
- */
- public void setAuditProxy(HashSet<String> ipPortList) {
- if (ipPortList.equals(currentIpPorts) &&
!this.sender.isHasSendError()) {
+ public void closeSocket() {
+ if (socket.isClosed()) {
+ LOGGER.info("Audit socket is already closed");
return;
}
- this.sender.setHasSendError(false);
- List<String> newIpPorts = new ArrayList<>();
- newIpPorts.addAll(ipPortList);
- this.currentIpPorts = newIpPorts;
- int ipSize = ipPortList.size();
- int needNewSize;
- if (this.maxConnectChannels == ALL_CONNECT_CHANNEL ||
this.maxConnectChannels >= ipSize) {
- needNewSize = ipSize;
- } else {
- needNewSize = maxConnectChannels;
+ try {
+ socket.close();
+ LOGGER.info("Audit socket closed successfully");
+ } catch (IOException exception) {
+ LOGGER.error("Error closing audit socket", exception);
}
+ }
- List<String> updateConfigIpLists = new ArrayList<>();
- List<String> availableIpLists = new ArrayList<>(ipPortList);
- for (int i = 0; i < needNewSize; i++) {
- int availableIpSize = availableIpLists.size();
- int newIpPortIndex = this.sRandom.nextInt(availableIpSize);
- String ipPort = availableIpLists.remove(newIpPortIndex);
- updateConfigIpLists.add(ipPort);
- }
- LOG.info("needNewSize:{},updateConfigIpLists:{}", needNewSize,
updateConfigIpLists);
- if (updateConfigIpLists.size() > 0) {
- this.sender.updateConfig(updateConfigIpLists);
+ public boolean checkSocket() {
+ if (socket.isClosed() || !socket.isConnected()) {
+ try {
+ InetSocketAddress inetSocketAddress =
ProxyManager.getInstance().getInetSocketAddress();
+ if (inetSocketAddress == null) {
+ LOGGER.error("Audit inet socket address is null!");
+ return false;
+ }
+ reconnect(inetSocketAddress, auditConfig.getSocketTimeout());
+ } catch (IOException exception) {
+ LOGGER.error("Connect to {} has exception!",
socket.getInetAddress(), exception);
+ return false;
+ }
}
+ return socket.isConnected();
}
- /**
- * next request id
- */
- public Long nextRequestId() {
- long requestId = requestIdSeq.getAndIncrement();
- if (requestId > MAX_REQUEST_ID) {
- requestId = 0L;
- requestIdSeq.set(requestId);
- }
- return requestId;
+ private void reconnect(InetSocketAddress inetSocketAddress, int timeout)
+ throws IOException {
+ socket = new Socket();
+ socket.connect(inetSocketAddress, timeout);
+ socket.setSoTimeout(timeout);
}
/**
* Send data with command
*/
- public void send(AuditApi.BaseCommand baseCommand, AuditApi.AuditRequest
auditRequest) {
+ public boolean send(AuditApi.BaseCommand baseCommand,
AuditApi.AuditRequest auditRequest) {
AuditData data = new AuditData(baseCommand, auditRequest);
- // cache first
- Long requestId = baseCommand.getAuditRequest().getRequestId();
- this.dataMap.putIfAbsent(requestId, data);
- this.sendData(data.getDataByte());
+ for (int retry = 0; retry < auditConfig.getRetryTimes(); retry++) {
+ if (sendData(data.getDataByte())) {
+ return true;
+ }
+ LOGGER.warn("Failed to send data on attempt {}. Retrying...",
retry + 1);
+ sleep();
+ }
+
+ LOGGER.error("Failed to send data after {} attempts. Storing data for
later retry.",
+ auditConfig.getRetryTimes());
+
failedDataMap.putIfAbsent(baseCommand.getAuditRequest().getRequestId(), data);
+ return false;
+ }
+
+ private void readFully(InputStream is, byte[] buffer, int len) throws
IOException {
+ int bytesRead;
+ int totalBytesRead = 0;
+ while (totalBytesRead < len
+ && (bytesRead = is.read(buffer, totalBytesRead, len -
totalBytesRead)) != -1) {
+ totalBytesRead += bytesRead;
+ }
}
/**
* Send data byte array
*/
- private void sendData(byte[] data) {
- if (data == null || data.length <= 0) {
- LOG.warn("send data is empty!");
- return;
+ private boolean sendData(byte[] data) {
+ if (data == null || data.length == 0) {
+ LOGGER.warn("Send data is empty!");
+ return false;
}
- ByteBuf dataBuf = ByteBufAllocator.DEFAULT.buffer(data.length);
- dataBuf.writeBytes(data);
- SenderResult result = this.sender.send(dataBuf);
- if (!result.result) {
- this.sender.setHasSendError(true);
+ if (!checkSocket()) {
+ return false;
+ }
+ try {
+ OutputStream outputStream = socket.getOutputStream();
+ InputStream inputStream = socket.getInputStream();
+
+ outputStream.write(data);
+
+ byte[] header = new byte[PACKAGE_HEADER_LEN];
+ readFully(inputStream, header, PACKAGE_HEADER_LEN);
+
+ int bodyLen = ((header[0] & 0xFF) << 24) |
+ ((header[1] & 0xFF) << 16) |
+ ((header[2] & 0xFF) << 8) |
+ (header[3] & 0xFF);
+
+ byte[] body = new byte[bodyLen];
Review Comment:
Do we need to determine if bodyLen is within a reasonable range here
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
