This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 15750b2a27 [INLONG-8315][Audit] Filter out illegal data that is more
than 7 days old (configurable) (#8319)
15750b2a27 is described below
commit 15750b2a27e18ec785d1f6f58b0f7665c7c31615
Author: megru <[email protected]>
AuthorDate: Mon Jun 26 18:21:31 2023 +0800
[INLONG-8315][Audit] Filter out illegal data that is more than 7 days old
(configurable) (#8319)
Co-authored-by: megruzhao <[email protected]>
Co-authored-by: healchow <[email protected]>
---
.../apache/inlong/audit/consts/ConfigConstants.java | 2 ++
.../inlong/audit/source/ServerMessageFactory.java | 8 +++++---
.../inlong/audit/source/ServerMessageHandler.java | 18 +++++++++++++++++-
.../apache/inlong/audit/source/SimpleTcpSource.java | 17 +++++++++++++----
inlong-audit/conf/application.properties | 3 +++
5 files changed, 40 insertions(+), 8 deletions(-)
diff --git
a/inlong-audit/audit-common/src/main/java/org/apache/inlong/audit/consts/ConfigConstants.java
b/inlong-audit/audit-common/src/main/java/org/apache/inlong/audit/consts/ConfigConstants.java
index 987a83a0d3..89db74c6a7 100644
---
a/inlong-audit/audit-common/src/main/java/org/apache/inlong/audit/consts/ConfigConstants.java
+++
b/inlong-audit/audit-common/src/main/java/org/apache/inlong/audit/consts/ConfigConstants.java
@@ -70,4 +70,6 @@ public class ConfigConstants {
public static final String MANAGER_PATH = "/inlong/manager/openapi";
public static final String MANAGER_GET_CONFIG_PATH = "/audit/getConfig";
+
+ public static final String MSG_VALID_THRESHOLD_DAYS =
"msg.valid.threshold.days";
}
diff --git
a/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/source/ServerMessageFactory.java
b/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/source/ServerMessageFactory.java
index d964d0d663..af07f6e2e1 100644
---
a/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/source/ServerMessageFactory.java
+++
b/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/source/ServerMessageFactory.java
@@ -41,6 +41,7 @@ public class ServerMessageFactory extends
ChannelInitializer<SocketChannel> {
private ServiceDecoder serviceDecoder;
private String messageHandlerName;
private int maxConnections = Integer.MAX_VALUE;
+ private final long msgValidThresholdDays;
private int maxMsgLength;
private String name;
@@ -57,7 +58,7 @@ public class ServerMessageFactory extends
ChannelInitializer<SocketChannel> {
*/
public ServerMessageFactory(AbstractSource source,
ChannelGroup allChannels, ServiceDecoder serviceDecoder,
- String messageHandlerName, Integer maxMsgLength, Integer maxCons,
String name) {
+ String messageHandlerName, Integer maxMsgLength, Integer maxCons,
Long msgValidThresholdDays, String name) {
this.source = source;
this.processor = source.getChannelProcessor();
this.allChannels = allChannels;
@@ -66,6 +67,7 @@ public class ServerMessageFactory extends
ChannelInitializer<SocketChannel> {
this.name = name;
this.maxConnections = maxCons;
this.maxMsgLength = maxMsgLength;
+ this.msgValidThresholdDays = msgValidThresholdDays;
}
@Override
@@ -83,10 +85,10 @@ public class ServerMessageFactory extends
ChannelInitializer<SocketChannel> {
Constructor<?> ctor =
clazz.getConstructor(AbstractSource.class,
ServiceDecoder.class,
- ChannelGroup.class, Integer.class);
+ ChannelGroup.class, Integer.class, Long.class);
ChannelInboundHandlerAdapter messageHandler =
(ChannelInboundHandlerAdapter) ctor
- .newInstance(source, serviceDecoder, allChannels,
maxConnections);
+ .newInstance(source, serviceDecoder, allChannels,
maxConnections, msgValidThresholdDays);
ch.pipeline().addLast("messageHandler", messageHandler);
} catch (Exception e) {
diff --git
a/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/source/ServerMessageHandler.java
b/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/source/ServerMessageHandler.java
index ef5e42de79..cc684ca35a 100644
---
a/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/source/ServerMessageHandler.java
+++
b/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/source/ServerMessageHandler.java
@@ -47,6 +47,7 @@ import static
com.google.common.base.Preconditions.checkArgument;
/**
* Server message handler
*/
+
public class ServerMessageHandler extends ChannelInboundHandlerAdapter {
private static final Logger LOGGER =
LoggerFactory.getLogger(ServerMessageHandler.class);
@@ -56,13 +57,16 @@ public class ServerMessageHandler extends
ChannelInboundHandlerAdapter {
private final ChannelProcessor processor;
private final ServiceDecoder serviceDecoder;
private final int maxConnections;
+ private final long msgValidThresholdDays;
+ private final long ONE_DAY_MS = 24 * 60 * 60 * 1000;
public ServerMessageHandler(AbstractSource source, ServiceDecoder
serviceDecoder,
- ChannelGroup allChannels, Integer maxCons) {
+ ChannelGroup allChannels, Integer maxCons, Long
msgValidThresholdDays) {
this.processor = source.getChannelProcessor();
this.serviceDecoder = serviceDecoder;
this.allChannels = allChannels;
this.maxConnections = maxCons;
+ this.msgValidThresholdDays = msgValidThresholdDays;
}
@Override
@@ -147,6 +151,12 @@ public class ServerMessageHandler extends
ChannelInboundHandlerAdapter {
List<AuditMessageBody> bodyList = auditRequest.getMsgBodyList();
int errorMsgBody = 0;
for (AuditMessageBody auditMessageBody : bodyList) {
+ long msgDays = messageDays(auditMessageBody.getLogTs());
+ if (msgDays >= this.msgValidThresholdDays) {
+ LOGGER.warn("Discard the data as it is from {} days ago, only
the data with a log timestamp"
+ + " less than {} days is valid", msgDays,
this.msgValidThresholdDays);
+ continue;
+ }
AuditData auditData = new AuditData();
auditData.setIp(auditRequest.getMsgHeader().getIp());
auditData.setThreadId(auditRequest.getMsgHeader().getThreadId());
@@ -182,6 +192,12 @@ public class ServerMessageHandler extends
ChannelInboundHandlerAdapter {
return reply;
}
+ public long messageDays(long logTs) {
+ long currentTime = System.currentTimeMillis();
+ long timeDiff = currentTime - logTs;
+ return timeDiff / ONE_DAY_MS;
+ }
+
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
LOGGER.error("exception caught", cause);
diff --git
a/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/source/SimpleTcpSource.java
b/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/source/SimpleTcpSource.java
index 44177b2053..d3d8208d19 100644
---
a/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/source/SimpleTcpSource.java
+++
b/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/source/SimpleTcpSource.java
@@ -49,14 +49,13 @@ import java.net.InetSocketAddress;
/**
* Simple tcp source
- *
*/
public class SimpleTcpSource extends AbstractSource implements Configurable,
EventDrivenSource {
private static final Logger logger =
LoggerFactory.getLogger(SimpleTcpSource.class);
private static final String CONNECTIONS = "connections";
-
protected int maxConnections = Integer.MAX_VALUE;
+ protected long msgValidThresholdDays;
protected Context context;
private ServerBootstrap bootstrap = null;
@@ -87,6 +86,8 @@ public class SimpleTcpSource extends AbstractSource
implements Configurable, Eve
private static int DEFAULT_MAX_CONNECTIONS = 5000;
+ private static long DEFAULT_MSG_VALID_THRESHOLD_DAYS = 7L;
+
private static int MIN_MSG_LENGTH = 4;
private static int MAX_MSG_LENGTH = 1024 * 64;
@@ -160,11 +161,11 @@ public class SimpleTcpSource extends AbstractSource
implements Configurable, Eve
Constructor ctor =
clazz.getConstructor(AbstractSource.class,
ChannelGroup.class,
ServiceDecoder.class, String.class,
- Integer.class, Integer.class, String.class);
+ Integer.class, Integer.class, Long.class,
String.class);
logger.info("Using channel processor:{}",
this.getClass().getName());
fac = (ChannelInitializer) ctor
.newInstance(this, allChannels, serviceDecoder,
- messageHandlerName, maxMsgLength, maxConnections,
this.getName());
+ messageHandlerName, maxMsgLength, maxConnections,
msgValidThresholdDays, this.getName());
} catch (Exception e) {
logger.error(
@@ -247,6 +248,14 @@ public class SimpleTcpSource extends AbstractSource
implements Configurable, Eve
context.getString(CONNECTIONS));
}
+ try {
+ msgValidThresholdDays =
+ context.getLong(ConfigConstants.MSG_VALID_THRESHOLD_DAYS,
DEFAULT_MSG_VALID_THRESHOLD_DAYS);
+ } catch (NumberFormatException e) {
+ logger.warn("BaseSource\'s \"msg.valid.threshold.days\" property
must specify a long value.",
+
context.getString(ConfigConstants.MSG_VALID_THRESHOLD_DAYS));
+ }
+
msgFactoryName = context.getString(ConfigConstants.MSG_FACTORY_NAME,
"org.apache.inlong.audit.source.ServerMessageFactory");
msgFactoryName = msgFactoryName.trim();
diff --git a/inlong-audit/conf/application.properties
b/inlong-audit/conf/application.properties
index de305bd119..41fd4e1e28 100644
--- a/inlong-audit/conf/application.properties
+++ b/inlong-audit/conf/application.properties
@@ -67,6 +67,9 @@ audit.kafka.topic=inlong-audit
audit.kafka.consumer.name=inlong-audit-consumer
audit.kafka.group.id=audit-consumer-group
+# Invalid data will be discarded if exceed the threshold days
+msg.valid.threshold.days=7
+
# es config
elasticsearch.host=127.0.0.1
elasticsearch.port=9200
