You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by do...@apache.org on 2021/10/22 02:04:47 UTC
[incubator-inlong] branch master updated: [INLONG-1642][Agent]
Agent Use Message filter to get tid from different lines in a file (#1703)
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/incubator-inlong.git
The following commit(s) were added to refs/heads/master by this push:
new a425e57 [INLONG-1642][Agent] Agent Use Message filter to get tid from different lines in a file (#1703)
a425e57 is described below
commit a425e5764d763ab49d36af78124bbc0c9cff8f6d
Author: ziruipeng <zp...@connect.ust.hk>
AuthorDate: Fri Oct 22 10:04:38 2021 +0800
[INLONG-1642][Agent] Agent Use Message filter to get tid from different lines in a file (#1703)
---
.../inlong/agent/constants/AgentConstants.java | 2 ++
.../inlong/agent/constants/CommonConstants.java | 3 ++
.../java/org/apache/inlong/agent/plugin/Sink.java | 7 ++++
.../apache/inlong/agent/task/TestTaskWrapper.java | 6 ++++
.../sinks/{ConsoleSink.java => AbstractSink.java} | 41 ++++++++++------------
.../inlong/agent/plugin/sinks/ConsoleSink.java | 6 ++++
.../inlong/agent/plugin/sinks/ProxySink.java | 29 ++++++++++++---
.../inlong/agent/plugin/sinks/PulsarSink.java | 6 ++++
.../apache/inlong/agent/plugin/TestFileAgent.java | 22 ++++++++++++
.../inlong/agent/plugin/filter/TestTidFilter.java | 41 ++++++++++++++++++++++
.../apache/inlong/agent/plugin/sinks/MockSink.java | 6 ++++
11 files changed, 142 insertions(+), 27 deletions(-)
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constants/AgentConstants.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constants/AgentConstants.java
index 9b110a1..f065006 100755
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constants/AgentConstants.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constants/AgentConstants.java
@@ -50,6 +50,8 @@ public class AgentConstants {
// default is empty.
public static final String AGENT_FETCHER_CLASSNAME = "agent.fetcher.classname";
+ public static final String AGENT_MESSAGE_FILTER_CLASSNAME = "agent.message.filter.classname";
+
public static final String AGENT_CONF_PARENT = "agent.conf.parent";
public static final String DEFAULT_AGENT_CONF_PARENT = "conf";
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constants/CommonConstants.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constants/CommonConstants.java
index f74a0e3..33a406b 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constants/CommonConstants.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constants/CommonConstants.java
@@ -77,6 +77,9 @@ public class CommonConstants {
public static final String PROXY_RETRY_SLEEP = "proxy.retry.sleep";
public static final long DEFAULT_PROXY_RETRY_SLEEP = 500;
+ public static final String FIELD_SPLITTER = "proxy.field.splitter";
+ public static final String DEFAULT_FIELD_SPLITTER = "|";
+
public static final String PROXY_KEY_BID = "bid";
public static final String PROXY_KEY_TID = "tid";
public static final String PROXY_KEY_ID = "id";
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/plugin/Sink.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/plugin/Sink.java
index 33b2ac2..50a2477 100755
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/plugin/Sink.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/plugin/Sink.java
@@ -17,6 +17,8 @@
package org.apache.inlong.agent.plugin;
+import org.apache.inlong.agent.conf.JobProfile;
+
/**
* Sink data to remote data center
*/
@@ -34,4 +36,9 @@ public interface Sink extends Stage {
* @param sourceFileName
*/
void setSourceFile(String sourceFileName);
+
+ /**
+ * every sink should include a message filter to filter out tid
+ */
+ MessageFilter initMessageFilter(JobProfile jobConf);
}
diff --git a/inlong-agent/agent-core/src/test/java/org/apache/inlong/agent/task/TestTaskWrapper.java b/inlong-agent/agent-core/src/test/java/org/apache/inlong/agent/task/TestTaskWrapper.java
index 2879376..7ba221a 100755
--- a/inlong-agent/agent-core/src/test/java/org/apache/inlong/agent/task/TestTaskWrapper.java
+++ b/inlong-agent/agent-core/src/test/java/org/apache/inlong/agent/task/TestTaskWrapper.java
@@ -30,6 +30,7 @@ import org.apache.inlong.agent.core.task.TaskWrapper;
import org.apache.inlong.agent.message.DefaultMessage;
import org.apache.inlong.agent.plugin.Channel;
import org.apache.inlong.agent.plugin.Message;
+import org.apache.inlong.agent.plugin.MessageFilter;
import org.apache.inlong.agent.plugin.Reader;
import org.apache.inlong.agent.plugin.Sink;
import org.junit.AfterClass;
@@ -173,6 +174,11 @@ public class TestTaskWrapper {
}
+ @Override
+ public MessageFilter initMessageFilter(JobProfile jobConf) {
+ return null;
+ }
+
public int getWriterCount() {
return writerCount;
}
diff --git a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/ConsoleSink.java b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/AbstractSink.java
similarity index 53%
copy from inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/ConsoleSink.java
copy to inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/AbstractSink.java
index cb79d8f..9fb3157 100644
--- a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/ConsoleSink.java
+++ b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/AbstractSink.java
@@ -17,35 +17,30 @@
package org.apache.inlong.agent.plugin.sinks;
-import java.nio.charset.StandardCharsets;
+import static org.apache.inlong.agent.constants.AgentConstants.AGENT_MESSAGE_FILTER_CLASSNAME;
+
import org.apache.inlong.agent.conf.JobProfile;
-import org.apache.inlong.agent.plugin.Message;
+import org.apache.inlong.agent.core.AgentManager;
+import org.apache.inlong.agent.plugin.MessageFilter;
import org.apache.inlong.agent.plugin.Sink;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
-/**
- * message write to console
- */
-public class ConsoleSink implements Sink {
-
- @Override
- public void write(Message message) {
- if (message != null) {
- System.out.println(new String(message.getBody(), StandardCharsets.UTF_8));
- }
- }
-
- @Override
- public void setSourceFile(String sourceFileName) {
+public abstract class AbstractSink implements Sink {
- }
+ private static final Logger LOGGER = LoggerFactory.getLogger(AgentManager.class);
@Override
- public void init(JobProfile jobConf) {
-
+ public MessageFilter initMessageFilter(JobProfile jobConf) {
+ if (jobConf.hasKey(AGENT_MESSAGE_FILTER_CLASSNAME)) {
+ try {
+ return (MessageFilter) Class.forName(jobConf.get(AGENT_MESSAGE_FILTER_CLASSNAME))
+ .getDeclaredConstructor().newInstance();
+ } catch (Exception e) {
+ LOGGER.error("init message filter error", e);
+ }
+ }
+ return null;
}
- @Override
- public void destroy() {
-
- }
}
diff --git a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/ConsoleSink.java b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/ConsoleSink.java
index cb79d8f..22f643d 100644
--- a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/ConsoleSink.java
+++ b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/ConsoleSink.java
@@ -20,6 +20,7 @@ package org.apache.inlong.agent.plugin.sinks;
import java.nio.charset.StandardCharsets;
import org.apache.inlong.agent.conf.JobProfile;
import org.apache.inlong.agent.plugin.Message;
+import org.apache.inlong.agent.plugin.MessageFilter;
import org.apache.inlong.agent.plugin.Sink;
/**
@@ -40,6 +41,11 @@ public class ConsoleSink implements Sink {
}
@Override
+ public MessageFilter initMessageFilter(JobProfile jobConf) {
+ return null;
+ }
+
+ @Override
public void init(JobProfile jobConf) {
}
diff --git a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/ProxySink.java b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/ProxySink.java
index 5ae4528..1ffe339 100755
--- a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/ProxySink.java
+++ b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/ProxySink.java
@@ -17,6 +17,7 @@
package org.apache.inlong.agent.plugin.sinks;
+import static org.apache.inlong.agent.constants.CommonConstants.DEFAULT_FIELD_SPLITTER;
import static org.apache.inlong.agent.constants.CommonConstants.PROXY_BID;
import static org.apache.inlong.agent.constants.CommonConstants.PROXY_KEY_AGENT_IP;
import static org.apache.inlong.agent.constants.CommonConstants.PROXY_KEY_ID;
@@ -39,6 +40,7 @@ import static org.apache.inlong.agent.constants.JobConstants.JOB_INSTANCE_ID;
import static org.apache.inlong.agent.constants.JobConstants.JOB_IP;
import static org.apache.inlong.agent.constants.JobConstants.JOB_RETRY;
+import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -54,16 +56,18 @@ import org.apache.inlong.agent.constants.CommonConstants;
import org.apache.inlong.agent.message.ProxyMessage;
import org.apache.inlong.agent.message.EndMessage;
import org.apache.inlong.agent.plugin.Message;
-import org.apache.inlong.agent.plugin.Sink;
+import org.apache.inlong.agent.plugin.MessageFilter;
import org.apache.inlong.agent.plugin.message.PackProxyMessage;
import org.apache.inlong.agent.utils.AgentUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class ProxySink implements Sink {
+public class ProxySink extends AbstractSink {
private static final Logger LOGGER = LoggerFactory.getLogger(ProxySink.class);
+ private MessageFilter messageFilter;
private SenderManager senderManager;
+ private byte[] fieldSplitter;
private String bid;
private String tid;
private String sourceFile;
@@ -85,7 +89,7 @@ public class ProxySink implements Sink {
public void write(Message message) {
if (message != null) {
message.getHeader().put(CommonConstants.PROXY_KEY_BID, bid);
- message.getHeader().put(CommonConstants.PROXY_KEY_TID, tid);
+ extractTidFromMessage(message, fieldSplitter);
if (!(message instanceof EndMessage)) {
ProxyMessage proxyMessage = ProxyMessage.parse(message);
// add proxy message to cache.
@@ -105,6 +109,20 @@ public class ProxySink implements Sink {
}
}
+ /**
+ * extract tid from message if message filter is presented
+ * or use the default tid
+ * @param message
+ */
+ private void extractTidFromMessage(Message message, byte[] fieldSplitter) {
+ if (messageFilter != null) {
+ message.getHeader().put(CommonConstants.PROXY_KEY_TID,
+ messageFilter.filterTid(message, fieldSplitter));
+ } else {
+ message.getHeader().put(CommonConstants.PROXY_KEY_TID, tid);
+ }
+ }
+
@Override
public void setSourceFile(String sourceFileName) {
this.sourceFile = sourceFileName;
@@ -154,7 +172,10 @@ public class ProxySink implements Sink {
dataTime = AgentUtils.timeStrConvertToMillSec(jobConf.get(JOB_DATA_TIME, ""),
jobConf.get(JOB_CYCLE_UNIT, ""));
bid = jobConf.get(PROXY_BID);
- tid = jobConf.get(PROXY_TID);
+ tid = jobConf.get(PROXY_TID, "");
+ messageFilter = initMessageFilter(jobConf);
+ fieldSplitter = jobConf.get(CommonConstants.FIELD_SPLITTER, DEFAULT_FIELD_SPLITTER).getBytes(
+ StandardCharsets.UTF_8);
executorService.execute(flushCache());
senderManager = new SenderManager(jobConf, bid, sourceFile);
try {
diff --git a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/PulsarSink.java b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/PulsarSink.java
index 2c52757..ed101cf 100755
--- a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/PulsarSink.java
+++ b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/PulsarSink.java
@@ -48,6 +48,7 @@ import java.util.concurrent.TimeUnit;
import org.apache.inlong.agent.common.AbstractDaemon;
import org.apache.inlong.agent.conf.JobProfile;
import org.apache.inlong.agent.plugin.Message;
+import org.apache.inlong.agent.plugin.MessageFilter;
import org.apache.inlong.agent.plugin.Sink;
import org.apache.inlong.agent.plugin.metrics.PluginMetric;
import org.apache.inlong.agent.plugin.utils.PluginUtils;
@@ -94,6 +95,11 @@ public class PulsarSink extends AbstractDaemon implements Sink {
}
@Override
+ public MessageFilter initMessageFilter(JobProfile jobConf) {
+ return null;
+ }
+
+ @Override
public void init(JobProfile jobConf) {
threadNum = jobConf.getInt(PULSAR_PRODUCER_THREAD_NUM, DEFAULT_PULSAR_PRODUCER_THREAD_NUM);
diff --git a/inlong-agent/agent-plugins/src/test/java/org/apache/inlong/agent/plugin/TestFileAgent.java b/inlong-agent/agent-plugins/src/test/java/org/apache/inlong/agent/plugin/TestFileAgent.java
index 301e177..a970a99 100755
--- a/inlong-agent/agent-plugins/src/test/java/org/apache/inlong/agent/plugin/TestFileAgent.java
+++ b/inlong-agent/agent-plugins/src/test/java/org/apache/inlong/agent/plugin/TestFileAgent.java
@@ -17,6 +17,7 @@
package org.apache.inlong.agent.plugin;
+import static org.apache.inlong.agent.constants.AgentConstants.AGENT_MESSAGE_FILTER_CLASSNAME;
import static org.apache.inlong.agent.constants.JobConstants.JOB_CYCLE_UNIT;
import static org.apache.inlong.agent.constants.JobConstants.JOB_DIR_FILTER_PATTERN;
import static org.apache.inlong.agent.constants.JobConstants.JOB_FILE_MAX_WAIT;
@@ -161,6 +162,27 @@ public class TestFileAgent {
}
@Test
+ public void testTidFilter() throws Exception {
+
+ String nowDate = AgentUtils.formatCurrentTimeWithoutOffset("yyyyMMdd");
+
+ try (InputStream stream = LOADER.getResourceAsStream("fileAgentJob.json")) {
+ if (stream != null) {
+ String jobJson = IOUtils.toString(stream, StandardCharsets.UTF_8);
+ JobProfile profile = JobProfile.parseJsonStr(jobJson);
+ profile.set(JOB_DIR_FILTER_PATTERN, Paths.get(testRootDir.toString(),
+ "YYYYMMDD").toString());
+ profile.set(JOB_CYCLE_UNIT, "D");
+ profile.set(AGENT_MESSAGE_FILTER_CLASSNAME,
+ "org.apache.inlong.agent.plugin.filter.DefaultMessageFilter");
+ agent.submitTriggerJob(profile);
+ }
+ }
+ createFiles(nowDate);
+ assertJobSuccess();
+ }
+
+ @Test
public void testTimeOffset() throws Exception {
String theDateBefore = AgentUtils.formatCurrentTimeWithOffset("yyyyMMdd", -1, 0, 0);
try (InputStream stream = LOADER.getResourceAsStream("fileAgentJob.json")) {
diff --git a/inlong-agent/agent-plugins/src/test/java/org/apache/inlong/agent/plugin/filter/TestTidFilter.java b/inlong-agent/agent-plugins/src/test/java/org/apache/inlong/agent/plugin/filter/TestTidFilter.java
index 663558d..e3ac7bd 100644
--- a/inlong-agent/agent-plugins/src/test/java/org/apache/inlong/agent/plugin/filter/TestTidFilter.java
+++ b/inlong-agent/agent-plugins/src/test/java/org/apache/inlong/agent/plugin/filter/TestTidFilter.java
@@ -17,11 +17,17 @@
package org.apache.inlong.agent.plugin.filter;
+import static org.apache.inlong.agent.constants.AgentConstants.AGENT_MESSAGE_FILTER_CLASSNAME;
+
import java.nio.charset.StandardCharsets;
import java.nio.file.Path;
import java.util.HashMap;
+import org.apache.inlong.agent.conf.JobProfile;
import org.apache.inlong.agent.message.ProxyMessage;
import org.apache.inlong.agent.plugin.AgentBaseTestsHelper;
+import org.apache.inlong.agent.plugin.Message;
+import org.apache.inlong.agent.plugin.MessageFilter;
+import org.apache.inlong.agent.plugin.sinks.AbstractSink;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.BeforeClass;
@@ -52,5 +58,40 @@ public class TestTidFilter {
Assert.assertEquals(s, "tid");
}
+ @Test
+ public void testAbstractSink() {
+ MockSink sinkTest = new MockSink();
+ JobProfile jobProfile = new JobProfile();
+ jobProfile.set(AGENT_MESSAGE_FILTER_CLASSNAME, "org.apache.inlong.agent.plugin.filter.DefaultMessageFilter");
+ MessageFilter messageFilter = sinkTest.initMessageFilter(jobProfile);
+ ProxyMessage proxyMessage = new ProxyMessage("tid|this is a line of file".getBytes(
+ StandardCharsets.UTF_8), new HashMap<>());
+ String s = messageFilter.filterTid(proxyMessage, "|".getBytes(StandardCharsets.UTF_8));
+ Assert.assertEquals(s, "tid");
+ }
+
+ class MockSink extends AbstractSink {
+
+ @Override
+ public void write(Message message) {
+
+ }
+
+ @Override
+ public void setSourceFile(String sourceFileName) {
+
+ }
+
+ @Override
+ public void init(JobProfile jobConf) {
+
+ }
+
+ @Override
+ public void destroy() {
+
+ }
+ }
+
}
diff --git a/inlong-agent/agent-plugins/src/test/java/org/apache/inlong/agent/plugin/sinks/MockSink.java b/inlong-agent/agent-plugins/src/test/java/org/apache/inlong/agent/plugin/sinks/MockSink.java
index bd977dd..1cea65f 100644
--- a/inlong-agent/agent-plugins/src/test/java/org/apache/inlong/agent/plugin/sinks/MockSink.java
+++ b/inlong-agent/agent-plugins/src/test/java/org/apache/inlong/agent/plugin/sinks/MockSink.java
@@ -25,6 +25,7 @@ import java.util.concurrent.atomic.AtomicLong;
import org.apache.inlong.agent.conf.JobProfile;
import org.apache.inlong.agent.core.task.TaskPositionManager;
import org.apache.inlong.agent.plugin.Message;
+import org.apache.inlong.agent.plugin.MessageFilter;
import org.apache.inlong.agent.plugin.Sink;
import org.apache.inlong.agent.utils.AgentUtils;
import org.slf4j.Logger;
@@ -53,6 +54,11 @@ public class MockSink implements Sink {
}
@Override
+ public MessageFilter initMessageFilter(JobProfile jobConf) {
+ return null;
+ }
+
+ @Override
public void init(JobProfile jobConf) {
taskPositionManager = TaskPositionManager.getTaskPositionManager();
jobInstanceId = jobConf.get(JOB_INSTANCE_ID);