This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new e7010eec5 [core] Introduce
compactionCompletedCount/compactionQueuedCount metrics. (#4391)
e7010eec5 is described below
commit e7010eec51b502fc7ccf53846ad25b9759aedf7a
Author: Fantasy-Jay <[email protected]>
AuthorDate: Wed Oct 30 11:14:00 2024 +0800
[core] Introduce compactionCompletedCount/compactionQueuedCount metrics.
(#4391)
---
docs/content/maintenance/metrics.md | 10 ++++++++
.../append/BucketedAppendCompactManager.java | 8 ++++++
.../org/apache/paimon/compact/CompactTask.java | 8 ++++++
.../mergetree/compact/MergeTreeCompactManager.java | 3 +++
.../operation/metrics/CompactionMetrics.java | 29 ++++++++++++++++++++++
.../operation/metrics/CompactionMetricsTest.java | 27 +++++++++++++++++++-
.../flink/compact/UnawareBucketCompactor.java | 17 +++++++++++++
.../flink/compact/UnawareBucketCompactorTest.java | 26 +++++++++++++++----
8 files changed, 122 insertions(+), 6 deletions(-)
diff --git a/docs/content/maintenance/metrics.md
b/docs/content/maintenance/metrics.md
index c21be8ec0..e8a160498 100644
--- a/docs/content/maintenance/metrics.md
+++ b/docs/content/maintenance/metrics.md
@@ -249,6 +249,16 @@ Below is lists of Paimon built-in metrics. They are
summarized into types of sca
<td>Gauge</td>
<td>The average runtime of compaction threads, calculated based on
recorded compaction time data in milliseconds. The value represents the average
duration of compaction operations. Higher values indicate longer average
compaction times, which may suggest the need for performance optimization.</td>
</tr>
+ <tr>
+ <td>compactionCompletedCount</td>
+ <td>Counter</td>
+ <td>The total number of compactions that have completed.</td>
+ </tr>
+ <tr>
+ <td>compactionQueuedCount</td>
+ <td>Counter</td>
+ <td>The total number of compactions that are queued/running.</td>
+ </tr>
</tbody>
</table>
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/BucketedAppendCompactManager.java
b/paimon-core/src/main/java/org/apache/paimon/append/BucketedAppendCompactManager.java
index 3f9114069..02f469624 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/BucketedAppendCompactManager.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/BucketedAppendCompactManager.java
@@ -115,10 +115,17 @@ public class BucketedAppendCompactManager extends
CompactFutureManager {
targetFileSize,
rewriter,
metricsReporter));
+ recordCompactionsQueuedRequest();
compacting = new ArrayList<>(toCompact);
toCompact.clear();
}
+ private void recordCompactionsQueuedRequest() {
+ if (metricsReporter != null) {
+ metricsReporter.increaseCompactionsQueuedCount();
+ }
+ }
+
private void triggerCompactionWithBestEffort() {
if (taskFuture != null) {
return;
@@ -130,6 +137,7 @@ public class BucketedAppendCompactManager extends
CompactFutureManager {
executor.submit(
new AutoCompactTask(
dvMaintainer, compacting, rewriter,
metricsReporter));
+ recordCompactionsQueuedRequest();
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/compact/CompactTask.java
b/paimon-core/src/main/java/org/apache/paimon/compact/CompactTask.java
index 3e5079c03..7da79aa51 100644
--- a/paimon-core/src/main/java/org/apache/paimon/compact/CompactTask.java
+++ b/paimon-core/src/main/java/org/apache/paimon/compact/CompactTask.java
@@ -53,6 +53,7 @@ public abstract class CompactTask implements
Callable<CompactResult> {
if (metricsReporter != null) {
metricsReporter.reportCompactionTime(
System.currentTimeMillis() - startMillis);
+
metricsReporter.increaseCompactionsCompletedCount();
}
},
LOG);
@@ -63,6 +64,13 @@ public abstract class CompactTask implements
Callable<CompactResult> {
return result;
} finally {
MetricUtils.safeCall(this::stopTimer, LOG);
+ MetricUtils.safeCall(this::decreaseCompactionsQueuedCount, LOG);
+ }
+ }
+
+ private void decreaseCompactionsQueuedCount() {
+ if (metricsReporter != null) {
+ metricsReporter.decreaseCompactionsQueuedCount();
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManager.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManager.java
index 28853a123..8ae292013 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManager.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManager.java
@@ -211,6 +211,9 @@ public class MergeTreeCompactManager extends
CompactFutureManager {
.collect(Collectors.joining(", ")));
}
taskFuture = executor.submit(task);
+ if (metricsReporter != null) {
+ metricsReporter.increaseCompactionsQueuedCount();
+ }
}
/** Finish current task, and update result files to {@link Levels}. */
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/metrics/CompactionMetrics.java
b/paimon-core/src/main/java/org/apache/paimon/operation/metrics/CompactionMetrics.java
index dd899e9bf..b0940a939 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/metrics/CompactionMetrics.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/metrics/CompactionMetrics.java
@@ -20,6 +20,7 @@ package org.apache.paimon.operation.metrics;
import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.metrics.Counter;
import org.apache.paimon.metrics.MetricGroup;
import org.apache.paimon.metrics.MetricRegistry;
@@ -41,6 +42,8 @@ public class CompactionMetrics {
public static final String AVG_LEVEL0_FILE_COUNT = "avgLevel0FileCount";
public static final String COMPACTION_THREAD_BUSY = "compactionThreadBusy";
public static final String AVG_COMPACTION_TIME = "avgCompactionTime";
+ public static final String COMPACTION_COMPLETED_COUNT =
"compactionCompletedCount";
+ public static final String COMPACTION_QUEUED_COUNT =
"compactionQueuedCount";
private static final long BUSY_MEASURE_MILLIS = 60_000;
private static final int COMPACTION_TIME_WINDOW = 100;
@@ -48,6 +51,8 @@ public class CompactionMetrics {
private final Map<PartitionAndBucket, ReporterImpl> reporters;
private final Map<Long, CompactTimer> compactTimers;
private final Queue<Long> compactionTimes;
+ private Counter compactionsCompletedCounter;
+ private Counter compactionsQueuedCounter;
public CompactionMetrics(MetricRegistry registry, String tableName) {
this.metricGroup = registry.tableMetricGroup(GROUP_NAME, tableName);
@@ -70,6 +75,9 @@ public class CompactionMetrics {
metricGroup.gauge(
AVG_COMPACTION_TIME, () ->
getCompactionTimeStream().average().orElse(0.0));
metricGroup.gauge(COMPACTION_THREAD_BUSY, () ->
getCompactBusyStream().sum());
+
+ compactionsCompletedCounter =
metricGroup.counter(COMPACTION_COMPLETED_COUNT);
+ compactionsQueuedCounter =
metricGroup.counter(COMPACTION_QUEUED_COUNT);
}
private LongStream getLevel0FileCountStream() {
@@ -98,6 +106,12 @@ public class CompactionMetrics {
void reportCompactionTime(long time);
+ void increaseCompactionsCompletedCount();
+
+ void increaseCompactionsQueuedCount();
+
+ void decreaseCompactionsQueuedCount();
+
void unregister();
}
@@ -133,6 +147,21 @@ public class CompactionMetrics {
this.level0FileCount = count;
}
+ @Override
+ public void increaseCompactionsCompletedCount() {
+ compactionsCompletedCounter.inc();
+ }
+
+ @Override
+ public void increaseCompactionsQueuedCount() {
+ compactionsQueuedCounter.inc();
+ }
+
+ @Override
+ public void decreaseCompactionsQueuedCount() {
+ compactionsQueuedCounter.dec();
+ }
+
@Override
public void unregister() {
reporters.remove(key);
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/metrics/CompactionMetricsTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/metrics/CompactionMetricsTest.java
index 2ce1cc4ea..222b99769 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/metrics/CompactionMetricsTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/metrics/CompactionMetricsTest.java
@@ -19,7 +19,9 @@
package org.apache.paimon.operation.metrics;
import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.metrics.Counter;
import org.apache.paimon.metrics.Gauge;
+import org.apache.paimon.metrics.Metric;
import org.apache.paimon.metrics.MetricRegistryImpl;
import org.junit.jupiter.api.Test;
@@ -36,6 +38,8 @@ public class CompactionMetricsTest {
assertThat(getMetric(metrics,
CompactionMetrics.AVG_LEVEL0_FILE_COUNT)).isEqualTo(-1.0);
assertThat(getMetric(metrics,
CompactionMetrics.COMPACTION_THREAD_BUSY)).isEqualTo(0.0);
assertThat(getMetric(metrics,
CompactionMetrics.AVG_COMPACTION_TIME)).isEqualTo(0.0);
+ assertThat(getMetric(metrics,
CompactionMetrics.COMPACTION_COMPLETED_COUNT)).isEqualTo(0L);
+ assertThat(getMetric(metrics,
CompactionMetrics.COMPACTION_QUEUED_COUNT)).isEqualTo(0L);
CompactionMetrics.Reporter[] reporters = new
CompactionMetrics.Reporter[3];
for (int i = 0; i < reporters.length; i++) {
reporters[i] = metrics.createReporter(BinaryRow.EMPTY_ROW, i);
@@ -44,6 +48,8 @@ public class CompactionMetricsTest {
assertThat(getMetric(metrics,
CompactionMetrics.MAX_LEVEL0_FILE_COUNT)).isEqualTo(0L);
assertThat(getMetric(metrics,
CompactionMetrics.AVG_LEVEL0_FILE_COUNT)).isEqualTo(0.0);
assertThat(getMetric(metrics,
CompactionMetrics.COMPACTION_THREAD_BUSY)).isEqualTo(0.0);
+ assertThat(getMetric(metrics,
CompactionMetrics.COMPACTION_COMPLETED_COUNT)).isEqualTo(0L);
+ assertThat(getMetric(metrics,
CompactionMetrics.COMPACTION_QUEUED_COUNT)).isEqualTo(0L);
reporters[0].reportLevel0FileCount(5);
reporters[1].reportLevel0FileCount(3);
@@ -60,9 +66,28 @@ public class CompactionMetricsTest {
reporters[0].reportCompactionTime(270000);
assertThat(getMetric(metrics, CompactionMetrics.AVG_COMPACTION_TIME))
.isEqualTo(273333.3333333333);
+
+ // enqueue compaction request
+ reporters[0].increaseCompactionsQueuedCount();
+ reporters[1].increaseCompactionsQueuedCount();
+ reporters[2].increaseCompactionsQueuedCount();
+ assertThat(getMetric(metrics,
CompactionMetrics.COMPACTION_COMPLETED_COUNT)).isEqualTo(0L);
+ assertThat(getMetric(metrics,
CompactionMetrics.COMPACTION_QUEUED_COUNT)).isEqualTo(3L);
+
+ // completed compactions and remove them from queue
+ reporters[0].increaseCompactionsCompletedCount();
+ reporters[0].decreaseCompactionsQueuedCount();
+ reporters[1].decreaseCompactionsQueuedCount();
+ assertThat(getMetric(metrics,
CompactionMetrics.COMPACTION_COMPLETED_COUNT)).isEqualTo(1L);
+ assertThat(getMetric(metrics,
CompactionMetrics.COMPACTION_QUEUED_COUNT)).isEqualTo(1L);
}
private Object getMetric(CompactionMetrics metrics, String metricName) {
- return ((Gauge<?>)
metrics.getMetricGroup().getMetrics().get(metricName)).getValue();
+ Metric metric = metrics.getMetricGroup().getMetrics().get(metricName);
+ if (metric instanceof Gauge) {
+ return ((Gauge<?>) metric).getValue();
+ } else {
+ return ((Counter) metric).getCount();
+ }
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/UnawareBucketCompactor.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/UnawareBucketCompactor.java
index a34f00907..e8021329c 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/UnawareBucketCompactor.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/UnawareBucketCompactor.java
@@ -100,14 +100,31 @@ public class UnawareBucketCompactor {
metricsReporter.reportCompactionTime(
System.currentTimeMillis()
-
startMillis);
+ metricsReporter
+
.increaseCompactionsCompletedCount();
}
},
LOG);
return commitMessage;
} finally {
MetricUtils.safeCall(this::stopTimer,
LOG);
+ MetricUtils.safeCall(
+
this::decreaseCompactionsQueuedCount, LOG);
}
}));
+ recordCompactionsQueuedRequest();
+ }
+
+ private void recordCompactionsQueuedRequest() {
+ if (metricsReporter != null) {
+ metricsReporter.increaseCompactionsQueuedCount();
+ }
+ }
+
+ private void decreaseCompactionsQueuedCount() {
+ if (metricsReporter != null) {
+ metricsReporter.decreaseCompactionsQueuedCount();
+ }
}
private void startTimer() {
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/compact/UnawareBucketCompactorTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/compact/UnawareBucketCompactorTest.java
index 69229ddce..1ec26afab 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/compact/UnawareBucketCompactorTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/compact/UnawareBucketCompactorTest.java
@@ -34,8 +34,10 @@ import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.utils.ExecutorThreadFactory;
+import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.Gauge;
import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.SimpleCounter;
import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
@@ -48,6 +50,8 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import static
org.apache.paimon.operation.metrics.CompactionMetrics.AVG_COMPACTION_TIME;
+import static
org.apache.paimon.operation.metrics.CompactionMetrics.COMPACTION_COMPLETED_COUNT;
+import static
org.apache.paimon.operation.metrics.CompactionMetrics.COMPACTION_QUEUED_COUNT;
import static
org.apache.paimon.operation.metrics.CompactionMetrics.COMPACTION_THREAD_BUSY;
/** Test for {@link UnawareBucketCompactor}. */
@@ -65,7 +69,8 @@ public class UnawareBucketCompactorTest {
Executors.newSingleThreadScheduledExecutor(
new ExecutorThreadFactory(
Thread.currentThread().getName() +
"-append-only-compact-worker"));
- Map<String, Gauge> map = new HashMap<>();
+ Map<String, Gauge> gaugeMap = new HashMap<>();
+ Map<String, Counter> counterMap = new HashMap<>();
UnawareBucketCompactor unawareBucketCompactor =
new UnawareBucketCompactor(
(FileStoreTable) catalog.getTable(identifier()),
@@ -74,7 +79,7 @@ public class UnawareBucketCompactorTest {
new FileStoreSourceReaderTest.DummyMetricGroup() {
@Override
public <T, G extends Gauge<T>> G gauge(String
name, G gauge) {
- map.put(name, gauge);
+ gaugeMap.put(name, gauge);
return null;
}
@@ -87,6 +92,13 @@ public class UnawareBucketCompactorTest {
public MetricGroup addGroup(String key, String
value) {
return this;
}
+
+ @Override
+ public Counter counter(String name) {
+ SimpleCounter counter = new SimpleCounter();
+ counterMap.put(name, counter);
+ return counter;
+ }
});
for (int i = 0; i < 320; i++) {
@@ -94,11 +106,15 @@ public class UnawareBucketCompactorTest {
Thread.sleep(250);
}
- double compactionThreadBusy = (double)
map.get(COMPACTION_THREAD_BUSY).getValue();
- double compactionAvrgTime = (double)
map.get(AVG_COMPACTION_TIME).getValue();
+ double compactionThreadBusy = (double)
gaugeMap.get(COMPACTION_THREAD_BUSY).getValue();
+ double compactionAvgTime = (double)
gaugeMap.get(AVG_COMPACTION_TIME).getValue();
+ long compactionsCompletedCount =
counterMap.get(COMPACTION_COMPLETED_COUNT).getCount();
+ long compactionsQueuedCount =
counterMap.get(COMPACTION_QUEUED_COUNT).getCount();
Assertions.assertThat(compactionThreadBusy).isGreaterThan(45).isLessThan(55);
-
Assertions.assertThat(compactionAvrgTime).isGreaterThan(120).isLessThan(140);
+
Assertions.assertThat(compactionAvgTime).isGreaterThan(120).isLessThan(140);
+ Assertions.assertThat(compactionsCompletedCount).isEqualTo(320L);
+ Assertions.assertThat(compactionsQueuedCount).isEqualTo(0L);
}
protected Catalog getCatalog() {
