This is an automated email from the ASF dual-hosted git repository.
gosonzhang 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 3621195 [INLONG-1826][Feature][InLong-Agent] Use jmx metric defined
in inlong-common (#1827)
3621195 is described below
commit 36211957f038ab93ba4d791e76cfaf02462eb408
Author: ziruipeng <[email protected]>
AuthorDate: Mon Nov 22 20:20:23 2021 +0800
[INLONG-1826][Feature][InLong-Agent] Use jmx metric defined in
inlong-common (#1827)
---
.../apache/inlong/agent/metrics/TestMetrics.java | 2 +-
inlong-agent/agent-core/pom.xml | 6 ++++
.../apache/inlong/agent/core/job/JobManager.java | 8 ++---
.../apache/inlong/agent/core/job/JobMetrics.java | 35 ++++++++++--------
.../apache/inlong/agent/core/task/TaskManager.java | 8 ++---
.../apache/inlong/agent/core/task/TaskMetrics.java | 41 ++++++++++++----------
.../apache/inlong/agent/core/task/TaskWrapper.java | 2 +-
.../apache/inlong/agent/core/TestTaskMetrics.java} | 39 +++++++++-----------
.../inlong/agent/plugin/channel/MemoryChannel.java | 25 ++++++-------
.../inlong/agent/plugin/metrics/PluginMetric.java | 38 ++++++++++----------
.../inlong/agent/plugin/sinks/PulsarSink.java | 10 +++---
.../inlong/agent/plugin/sinks/SenderManager.java | 4 +--
.../plugin/sources/reader/TextFileReader.java | 7 ++--
13 files changed, 119 insertions(+), 106 deletions(-)
diff --git
a/inlong-agent/agent-common/src/test/java/org/apache/inlong/agent/metrics/TestMetrics.java
b/inlong-agent/agent-common/src/test/java/org/apache/inlong/agent/metrics/TestMetrics.java
index 9f6a712..ad0a9a1 100755
---
a/inlong-agent/agent-common/src/test/java/org/apache/inlong/agent/metrics/TestMetrics.java
+++
b/inlong-agent/agent-common/src/test/java/org/apache/inlong/agent/metrics/TestMetrics.java
@@ -42,7 +42,7 @@ public class TestMetrics {
}
@Test
- public void testMetric() throws Exception {
+ public void testMetric() {
MetricTest metricTest = MetricTest.getMetrics();
Assert.assertNotNull(metricTest.counterInt);
Assert.assertNotNull(metricTest.counterLong);
diff --git a/inlong-agent/agent-core/pom.xml b/inlong-agent/agent-core/pom.xml
index 15f0a31..38a3dfb 100755
--- a/inlong-agent/agent-core/pom.xml
+++ b/inlong-agent/agent-core/pom.xml
@@ -60,5 +60,11 @@
<artifactId>agent-common</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.inlong</groupId>
+ <artifactId>inlong-common</artifactId>
+ <version>${project.version}</version>
+ <scope>compile</scope>
+ </dependency>
</dependencies>
</project>
\ No newline at end of file
diff --git
a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/job/JobManager.java
b/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/job/JobManager.java
index a44dd9f..c357b8b 100644
---
a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/job/JobManager.java
+++
b/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/job/JobManager.java
@@ -106,7 +106,7 @@ public class JobManager extends AbstractDaemon {
LOGGER.warn("{} has been added to running pool, "
+ "cannot be added repeatedly", job.getJobInstanceId());
} else {
- jobMetrics.runningJobs.incr();
+ jobMetrics.runningJobs.incrementAndGet();
}
} catch (Exception rje) {
LOGGER.debug("reject job {}", job.getJobInstanceId(), rje);
@@ -199,7 +199,7 @@ public class JobManager extends AbstractDaemon {
public void markJobAsSuccess(String jobId) {
JobWrapper wrapper = jobs.remove(jobId);
if (wrapper != null) {
- jobMetrics.runningJobs.decr();
+ jobMetrics.runningJobs.decrementAndGet();
LOGGER.info("job instance {} is success", jobId);
// mark job as success.
jobConfDB.updateJobState(jobId, StateSearchKey.SUCCESS);
@@ -210,8 +210,8 @@ public class JobManager extends AbstractDaemon {
JobWrapper wrapper = jobs.remove(jobId);
if (wrapper != null) {
LOGGER.info("job instance {} is failed", jobId);
- jobMetrics.runningJobs.decr();
- jobMetrics.fatalJobs.incr();
+ jobMetrics.runningJobs.decrementAndGet();
+ jobMetrics.fatalJobs.incrementAndGet();
// mark job as success.
jobConfDB.updateJobState(jobId, StateSearchKey.FAILED);
}
diff --git
a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/job/JobMetrics.java
b/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/job/JobMetrics.java
index 28b0e60..28fd6c2 100644
---
a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/job/JobMetrics.java
+++
b/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/job/JobMetrics.java
@@ -18,29 +18,36 @@
package org.apache.inlong.agent.core.job;
import java.util.concurrent.atomic.AtomicBoolean;
-import org.apache.inlong.agent.metrics.Metric;
-import org.apache.inlong.agent.metrics.Metrics;
-import org.apache.inlong.agent.metrics.MetricsRegister;
-import org.apache.inlong.agent.metrics.gauge.GaugeInt;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.inlong.commons.config.metrics.CountMetric;
+import org.apache.inlong.commons.config.metrics.Dimension;
+import org.apache.inlong.commons.config.metrics.MetricDomain;
+import org.apache.inlong.commons.config.metrics.MetricItem;
+import org.apache.inlong.commons.config.metrics.MetricRegister;
+
+@MetricDomain(name = "AgentJob")
+public class JobMetrics extends MetricItem {
-@Metrics
-public class JobMetrics {
private static final JobMetrics JOB_METRICS = new JobMetrics();
+
private static final AtomicBoolean REGISTER_ONCE = new
AtomicBoolean(false);
+ private static final String AGENT_JOB_METRIC = "AgentJobMetric";
- @Metric
- GaugeInt runningJobs;
+ @Dimension
+ public String tagName;
- @Metric
- GaugeInt fatalJobs;
+ @CountMetric
+ public AtomicLong runningJobs = new AtomicLong(0);
- private JobMetrics() {
- }
+ @CountMetric
+ public AtomicLong fatalJobs = new AtomicLong(0);
- static JobMetrics create() {
+ public static JobMetrics create() {
if (REGISTER_ONCE.compareAndSet(false, true)) {
- MetricsRegister.register("Job", "STateSummary", null, JOB_METRICS);
+ JOB_METRICS.tagName = AGENT_JOB_METRIC;
+ MetricRegister.register(JOB_METRICS);
}
return JOB_METRICS;
}
}
+
diff --git
a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/task/TaskManager.java
b/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/task/TaskManager.java
index 6f895d5..487f487 100755
---
a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/task/TaskManager.java
+++
b/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/task/TaskManager.java
@@ -120,7 +120,7 @@ public class TaskManager extends AbstractDaemon {
LOGGER.warn("reject task {}",
wrapper.getTask().getTaskId(), ex);
}
}
- taskMetrics.runningTasks.incr();
+ taskMetrics.runningTasks.incrementAndGet();
}
}
@@ -137,7 +137,7 @@ public class TaskManager extends AbstractDaemon {
LOGGER.error("cannot submit to retry queue, max {}, current
{}", taskMaxCapacity,
retryTasks.size());
} else {
- taskMetrics.retryingTasks.incr();
+ taskMetrics.retryingTasks.incrementAndGet();
}
return success;
} catch (Exception ex) {
@@ -180,7 +180,7 @@ public class TaskManager extends AbstractDaemon {
* @param taskId - task id
*/
public void removeTask(String taskId) {
- taskMetrics.runningTasks.decr();
+ taskMetrics.runningTasks.decrementAndGet();
TaskWrapper taskWrapper = tasks.remove(taskId);
if (taskWrapper != null) {
taskWrapper.waitForFinish();
@@ -224,7 +224,7 @@ public class TaskManager extends AbstractDaemon {
while (!retryTasks.isEmpty()) {
TaskWrapper taskWrapper = retryTasks.poll();
if (taskWrapper != null) {
- taskMetrics.retryingTasks.decr();
+ taskMetrics.retryingTasks.decrementAndGet();
submitTask(taskWrapper);
}
}
diff --git
a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/task/TaskMetrics.java
b/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/task/TaskMetrics.java
old mode 100755
new mode 100644
index 5863e88..0d91fe7
---
a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/task/TaskMetrics.java
+++
b/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/task/TaskMetrics.java
@@ -18,38 +18,41 @@
package org.apache.inlong.agent.core.task;
import java.util.concurrent.atomic.AtomicBoolean;
-import org.apache.inlong.agent.metrics.Metric;
-import org.apache.inlong.agent.metrics.Metrics;
-import org.apache.inlong.agent.metrics.MetricsRegister;
-import org.apache.inlong.agent.metrics.counter.CounterLong;
-import org.apache.inlong.agent.metrics.gauge.GaugeInt;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.inlong.commons.config.metrics.CountMetric;
+import org.apache.inlong.commons.config.metrics.Dimension;
+import org.apache.inlong.commons.config.metrics.MetricDomain;
+import org.apache.inlong.commons.config.metrics.MetricItem;
+import org.apache.inlong.commons.config.metrics.MetricRegister;
/**
- * Metric collector for task level.
+ * metrics for agent task
*/
-@Metrics
-public class TaskMetrics {
+@MetricDomain(name = "AgentTask")
+public class TaskMetrics extends MetricItem {
- private static final TaskMetrics TASK_METRICS = new TaskMetrics();
+ private static final TaskMetrics JOB_METRICS = new TaskMetrics();
private static final AtomicBoolean REGISTER_ONCE = new
AtomicBoolean(false);
+ public static final String AGENT_TASK = "AgentTaskMetric";
- @Metric
- GaugeInt runningTasks;
+ @Dimension
+ public String module;
- @Metric
- GaugeInt retryingTasks;
+ @CountMetric
+ public AtomicLong runningTasks = new AtomicLong(0);
- @Metric
- CounterLong fatalTasks;
+ @CountMetric
+ public AtomicLong retryingTasks = new AtomicLong(0);
- private TaskMetrics() {
- }
+ @CountMetric
+ public AtomicLong fatalTasks = new AtomicLong(0);
public static TaskMetrics create() {
// register one time.
if (REGISTER_ONCE.compareAndSet(false, true)) {
- MetricsRegister.register("Task", "StateSummary", null,
TASK_METRICS);
+ JOB_METRICS.module = AGENT_TASK;
+ MetricRegister.register(JOB_METRICS);
}
- return TASK_METRICS;
+ return JOB_METRICS;
}
}
diff --git
a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/task/TaskWrapper.java
b/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/task/TaskWrapper.java
index 11ab945..5f8fd47 100755
---
a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/task/TaskWrapper.java
+++
b/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/task/TaskWrapper.java
@@ -163,7 +163,7 @@ public class TaskWrapper extends AbstractStateWrapper {
retryTime.incrementAndGet();
if (!shouldRetry()) {
doChangeState(State.FATAL);
- taskManager.getTaskMetrics().fatalTasks.incr();
+ taskManager.getTaskMetrics().fatalTasks.incrementAndGet();
}
}).addCallback(State.FAILED, State.FATAL, (before, after) -> {
diff --git
a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/job/JobMetrics.java
b/inlong-agent/agent-core/src/test/java/org/apache/inlong/agent/core/TestTaskMetrics.java
similarity index 51%
copy from
inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/job/JobMetrics.java
copy to
inlong-agent/agent-core/src/test/java/org/apache/inlong/agent/core/TestTaskMetrics.java
index 28b0e60..5f8d69c 100644
---
a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/job/JobMetrics.java
+++
b/inlong-agent/agent-core/src/test/java/org/apache/inlong/agent/core/TestTaskMetrics.java
@@ -15,32 +15,27 @@
* limitations under the License.
*/
-package org.apache.inlong.agent.core.job;
+package org.apache.inlong.agent.core;
-import java.util.concurrent.atomic.AtomicBoolean;
-import org.apache.inlong.agent.metrics.Metric;
-import org.apache.inlong.agent.metrics.Metrics;
-import org.apache.inlong.agent.metrics.MetricsRegister;
-import org.apache.inlong.agent.metrics.gauge.GaugeInt;
+import org.apache.inlong.agent.core.task.TaskMetrics;
+import org.junit.Assert;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
-@Metrics
-public class JobMetrics {
- private static final JobMetrics JOB_METRICS = new JobMetrics();
- private static final AtomicBoolean REGISTER_ONCE = new
AtomicBoolean(false);
+public class TestTaskMetrics {
- @Metric
- GaugeInt runningJobs;
+ private static final Logger LOGGER =
LoggerFactory.getLogger(AgentBaseTestsHelper.class);
- @Metric
- GaugeInt fatalJobs;
-
- private JobMetrics() {
- }
-
- static JobMetrics create() {
- if (REGISTER_ONCE.compareAndSet(false, true)) {
- MetricsRegister.register("Job", "STateSummary", null, JOB_METRICS);
+ @Test
+ public void testAgentMetrics() {
+ try {
+ TaskMetrics taskMetrics = TaskMetrics.create();
+ taskMetrics.retryingTasks.addAndGet(1);
+ Assert.assertEquals(taskMetrics.module, "AgentTaskMetric");
+ } catch (Exception ex) {
+ LOGGER.error("error happens" + ex);
}
- return JOB_METRICS;
}
+
}
diff --git
a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/channel/MemoryChannel.java
b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/channel/MemoryChannel.java
index e49cac1..c998ea4 100644
---
a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/channel/MemoryChannel.java
+++
b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/channel/MemoryChannel.java
@@ -33,7 +33,7 @@ public class MemoryChannel implements Channel {
private LinkedBlockingQueue<Message> queue;
- private final PluginMetric metric = new PluginMetric();
+ private final PluginMetric pluginMetricNew = new
PluginMetric("AgentMemoryPlugin");
/**
* {@inheritDoc}
@@ -42,12 +42,12 @@ public class MemoryChannel implements Channel {
public void push(Message message) {
try {
if (message != null) {
- metric.readNum.incr();
+ pluginMetricNew.readNum.incrementAndGet();
queue.put(message);
- metric.readSuccessNum.incr();
+ pluginMetricNew.readSuccessNum.incrementAndGet();
}
} catch (InterruptedException ex) {
- metric.readFailedNum.incr();
+ pluginMetricNew.readFailedNum.incrementAndGet();
Thread.currentThread().interrupt();
}
}
@@ -56,17 +56,17 @@ public class MemoryChannel implements Channel {
public boolean push(Message message, long timeout, TimeUnit unit) {
try {
if (message != null) {
- metric.readNum.incr();
+ pluginMetricNew.readNum.incrementAndGet();
boolean result = queue.offer(message, timeout, unit);
if (result) {
- metric.readSuccessNum.incr();
+ pluginMetricNew.readSuccessNum.incrementAndGet();
} else {
- metric.readFailedNum.incr();
+ pluginMetricNew.readFailedNum.incrementAndGet();
}
return result;
}
} catch (InterruptedException ex) {
- metric.readFailedNum.incr();
+ pluginMetricNew.readFailedNum.incrementAndGet();
Thread.currentThread().interrupt();
}
return false;
@@ -80,11 +80,11 @@ public class MemoryChannel implements Channel {
try {
Message message = queue.poll(timeout, unit);
if (message != null) {
- metric.sendSuccessNum.incr();
+ pluginMetricNew.sendSuccessNum.incrementAndGet();
}
return message;
} catch (InterruptedException ex) {
- metric.sendFailedNum.incr();
+ pluginMetricNew.sendFailedNum.incrementAndGet();
Thread.currentThread().interrupt();
throw new IllegalStateException(ex);
}
@@ -104,7 +104,8 @@ public class MemoryChannel implements Channel {
}
LOGGER.info("destroy channel, memory channel metric, readNum: {},
readSuccessNum: {}, "
+ "readFailedNum: {}, sendSuccessNum: {}, sendFailedNum: {}",
- metric.readNum.snapshot(), metric.readSuccessNum.snapshot(),
metric.readFailedNum.snapshot(),
- metric.sendSuccessNum.snapshot(), metric.sendFailedNum.snapshot());
+ pluginMetricNew.readNum.get(),
pluginMetricNew.readSuccessNum.get(),
+ pluginMetricNew.readFailedNum.get(),
pluginMetricNew.sendSuccessNum.get(),
+ pluginMetricNew.sendFailedNum.get());
}
}
diff --git
a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/metrics/PluginMetric.java
b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/metrics/PluginMetric.java
old mode 100755
new mode 100644
index 4d692b4..6e53b38
---
a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/metrics/PluginMetric.java
+++
b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/metrics/PluginMetric.java
@@ -17,41 +17,43 @@
package org.apache.inlong.agent.plugin.metrics;
+import java.util.concurrent.atomic.AtomicLong;
import org.apache.inlong.agent.metrics.Metric;
-import org.apache.inlong.agent.metrics.Metrics;
-import org.apache.inlong.agent.metrics.MetricsRegister;
-import org.apache.inlong.agent.metrics.Tag;
-import org.apache.inlong.agent.metrics.counter.CounterLong;
+import org.apache.inlong.commons.config.metrics.Dimension;
+import org.apache.inlong.commons.config.metrics.MetricDomain;
+import org.apache.inlong.commons.config.metrics.MetricItem;
+import org.apache.inlong.commons.config.metrics.MetricRegister;
/**
- * Common plugin metrics
+ * metrics for agent plugin
*/
-@Metrics
-public class PluginMetric {
+@MetricDomain(name = "PluginMetric")
+public class PluginMetric extends MetricItem {
- @Metric
- public Tag tagName;
+ @Dimension
+ public String tagName;
@Metric
- public CounterLong readNum;
+ public AtomicLong readNum = new AtomicLong(0);
@Metric
- public CounterLong sendNum;
+ public AtomicLong sendNum = new AtomicLong(0);
@Metric
- public CounterLong sendFailedNum;
+ public AtomicLong sendFailedNum = new AtomicLong(0);
@Metric
- public CounterLong readFailedNum;
+ public AtomicLong readFailedNum = new AtomicLong(0);
@Metric
- public CounterLong readSuccessNum;
+ public AtomicLong readSuccessNum = new AtomicLong(0);
@Metric
- public CounterLong sendSuccessNum;
+ public AtomicLong sendSuccessNum = new AtomicLong(0);
- public PluginMetric() {
- // every metric should register, otherwise not working.
- MetricsRegister.register("Plugin", "PluginSummary", null, this);
+ public PluginMetric(String tagName) {
+ this.tagName = tagName;
+ MetricRegister.register(this);
}
+
}
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 ed101cf..55b01ca 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
@@ -72,7 +72,7 @@ public class PulsarSink extends AbstractDaemon implements
Sink {
private LinkedBlockingQueue<byte[]> cache;
private final List<Producer<byte[]>> producerList = new ArrayList<>();
- private final PluginMetric metric = new PluginMetric();
+ private final PluginMetric pluginMetricNew = new
PluginMetric("AgentPulsarMetric");
private PulsarClient client;
@Override
@@ -81,7 +81,7 @@ public class PulsarSink extends AbstractDaemon implements
Sink {
// if message is not null
try {
// put message to cache, wait until cache is not full.
- metric.sendNum.incr();
+ pluginMetricNew.sendNum.incrementAndGet();
cache.put(message.getBody());
} catch (Exception ignored) {
// ignore it
@@ -116,7 +116,7 @@ public class PulsarSink extends AbstractDaemon implements
Sink {
try {
stop();
LOGGER.info("send success num is {}, failed num is {}",
- metric.sendSuccessNum.snapshot(),
metric.sendFailedNum.snapshot());
+ pluginMetricNew.sendSuccessNum.get(),
pluginMetricNew.sendFailedNum.get());
} catch (Exception ex) {
LOGGER.error("exception caught", ex);
}
@@ -136,12 +136,12 @@ public class PulsarSink extends AbstractDaemon implements
Sink {
// exception is not null, that means not success.
// TODO: add metric or retry sending message.
if (t != null) {
- metric.sendFailedNum.incr();
+ pluginMetricNew.sendFailedNum.incrementAndGet();
if (!cache.offer(item)) {
LOGGER.warn("message {} not add back to retry", m);
}
} else {
- metric.sendSuccessNum.incr();
+ pluginMetricNew.sendSuccessNum.incrementAndGet();
}
});
} else {
diff --git
a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/SenderManager.java
b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/SenderManager.java
index ad652b1..548901a 100755
---
a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/SenderManager.java
+++
b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/SenderManager.java
@@ -83,7 +83,7 @@ public class SenderManager {
private TaskPositionManager taskPositionManager;
private final int maxSenderPerGroup;
private final String sourceFilePath;
- private final PluginMetric metric = new PluginMetric();
+ private final PluginMetric metric = new PluginMetric("AgentSenderManager");
public SenderManager(JobProfile jobConf, String inlongGroupId, String
sourceFilePath) {
AgentConfiguration conf = AgentConfiguration.getAgentConf();
@@ -194,7 +194,7 @@ public class SenderManager {
sendBatch(jobId, groupId, streamId, bodyList, retry + 1,
dataTime);
return;
}
- metric.sendSuccessNum.incr(bodyList.size());
+ metric.sendSuccessNum.addAndGet(bodyList.size());
taskPositionManager.updateFileSinkPosition(jobId, sourceFilePath,
bodyList.size());
}
diff --git
a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/reader/TextFileReader.java
b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/reader/TextFileReader.java
index 385292e..bbf7c6c 100755
---
a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/reader/TextFileReader.java
+++
b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/reader/TextFileReader.java
@@ -65,8 +65,7 @@ public class TextFileReader implements Reader {
this.file = file;
this.position = position;
this.md5 = md5;
- textFileMetric = new PluginMetric();
- textFileMetric.tagName.setName(file.getAbsolutePath());
+ textFileMetric = new PluginMetric("AgentTextMetric");
}
public TextFileReader(File file) {
@@ -78,7 +77,7 @@ public class TextFileReader implements Reader {
if (iterator != null && iterator.hasNext()) {
String message = iterator.next();
if (validateMessage(message)) {
- textFileMetric.readNum.incr();
+ textFileMetric.readNum.incrementAndGet();
return new
DefaultMessage(message.getBytes(StandardCharsets.UTF_8));
}
}
@@ -164,6 +163,6 @@ public class TextFileReader implements Reader {
public void destroy() {
AgentUtils.finallyClose(stream);
LOGGER.info("destroy reader with read {} num {}",
- textFileMetric.tagName.getName(),
textFileMetric.readNum.snapshot());
+ textFileMetric.tagName, textFileMetric.readNum.get());
}
}
