wuchong commented on code in PR #2282:
URL: https://github.com/apache/fluss/pull/2282#discussion_r2655254074
##########
fluss-server/src/main/java/org/apache/fluss/server/metrics/group/TableMetricGroup.java:
##########
@@ -236,17 +243,159 @@ public BucketMetricGroup addBucketMetricGroup(
public void removeBucketMetricGroup(TableBucket tableBucket) {
BucketMetricGroup metricGroup = buckets.remove(tableBucket);
- metricGroup.close();
+ if (metricGroup != null) {
+ metricGroup.close();
+ }
+ // Also remove RocksDB metrics if exists
+ RocksDBMetrics rocksDBMetrics = rocksDBMetricsMap.remove(tableBucket);
+ if (rocksDBMetrics != null) {
+ try {
+ rocksDBMetrics.close();
+ } catch (Exception e) {
+ // Ignore close errors
+ }
+ }
+ }
+
+ /**
+ * Register RocksDB metrics for a bucket. This allows table-level
aggregation without
+ * registering bucket-level metrics.
+ *
+ * @param tableBucket the table bucket
+ * @param rocksDBMetrics the RocksDB metrics accessor
+ */
+ public void registerRocksDBMetrics(TableBucket tableBucket, RocksDBMetrics
rocksDBMetrics) {
Review Comment:
This method is confusing in relation to `registerRocksDBMetrics()`.
Currently:
- `registerRocksDBMetrics()` registers actual Fluss `Metric` instances
(i.e., exported to the metrics system),
- while this `registerRocksDBMetrics(TableBucket tableBucket, RocksDBMetrics
rocksDBMetrics)` merely adds an internal `RocksDBMetrics` object to this class.
In Fluss, other `XxxMetrics` classes (e.g., `RequestsMetrics`,
`UserMetrics`) directly manage and expose true Fluss `Metric`s. In contrast,
`RocksDBMetrics` only collects low-level RocksDB statistics and does not
register them as metrics by itself.
To clarify the distinction and align naming with responsibility, I propose:
- Rename `RocksDBMetrics` → `RocksDBStatistics`
- Rename this method → `registerRocksDBStatistics()`
This change would improve consistency, reduce confusion, and better reflect
the actual role of the class.
##########
fluss-server/src/main/java/org/apache/fluss/server/metrics/group/TableMetricGroup.java:
##########
@@ -43,6 +44,10 @@ public class TableMetricGroup extends AbstractMetricGroup {
private final Map<TableBucket, BucketMetricGroup> buckets = new
HashMap<>();
+ // Directly manage RocksDB metrics for aggregation
+ // This is cleaner than passing through BucketMetricGroup
+ private final Map<TableBucket, RocksDBMetrics> rocksDBMetricsMap = new
HashMap<>();
Review Comment:
Both this `rocksDBMetricsMap` and `buckets` maps should be
`ConcurrentHashMap`, because they may be invoked/iterated in other threads
(metric reporter).
##########
fluss-server/src/main/java/org/apache/fluss/server/kv/KvManager.java:
##########
@@ -304,6 +312,9 @@ public void dropKv(TableBucket tableBucket) {
if (dropKvTablet != null) {
TablePath tablePath = dropKvTablet.getTablePath();
try {
+ // Unregister RocksDB metrics from TableMetricGroup
+ serverMetricGroup.unregisterRocksDBMetrics(tablePath,
tableBucket);
Review Comment:
We already unregister RocksDB metrics in `ReplicaManager#stopReplica`, which
invokes `TableMetricGroup#removeBucketMetricGroup(...)` by
`TabletServerMetricGroup#removeTableBucketMetricGroup`. That method, in turn,
handles the unregistration of the RocksDB metrics.
##########
fluss-server/src/main/java/org/apache/fluss/server/metrics/group/TableMetricGroup.java:
##########
@@ -236,17 +243,159 @@ public BucketMetricGroup addBucketMetricGroup(
public void removeBucketMetricGroup(TableBucket tableBucket) {
BucketMetricGroup metricGroup = buckets.remove(tableBucket);
- metricGroup.close();
+ if (metricGroup != null) {
+ metricGroup.close();
+ }
+ // Also remove RocksDB metrics if exists
+ RocksDBMetrics rocksDBMetrics = rocksDBMetricsMap.remove(tableBucket);
+ if (rocksDBMetrics != null) {
+ try {
+ rocksDBMetrics.close();
+ } catch (Exception e) {
+ // Ignore close errors
+ }
+ }
+ }
+
+ /**
+ * Register RocksDB metrics for a bucket. This allows table-level
aggregation without
+ * registering bucket-level metrics.
+ *
+ * @param tableBucket the table bucket
+ * @param rocksDBMetrics the RocksDB metrics accessor
+ */
+ public void registerRocksDBMetrics(TableBucket tableBucket, RocksDBMetrics
rocksDBMetrics) {
+ rocksDBMetricsMap.put(tableBucket, rocksDBMetrics);
+ }
+
+ /**
+ * Unregister RocksDB metrics for a bucket.
+ *
+ * @param tableBucket the table bucket
+ */
+ public void unregisterRocksDBMetrics(TableBucket tableBucket) {
+ RocksDBMetrics rocksDBMetrics = rocksDBMetricsMap.remove(tableBucket);
+ if (rocksDBMetrics != null) {
+ try {
+ rocksDBMetrics.close();
+ } catch (Exception e) {
+ // Ignore close errors
+ }
+ }
}
public int bucketGroupsCount() {
return buckets.size();
}
+ public java.util.Collection<BucketMetricGroup> getBucketMetricGroups() {
Review Comment:
remove, unused
##########
fluss-server/src/main/java/org/apache/fluss/server/metrics/group/TabletServerMetricGroup.java:
##########
@@ -30,13 +30,19 @@
import org.apache.fluss.metrics.ThreadSafeSimpleCounter;
import org.apache.fluss.metrics.groups.AbstractMetricGroup;
import org.apache.fluss.metrics.registry.MetricRegistry;
+import org.apache.fluss.server.kv.rocksdb.RocksDBMetrics;
import org.apache.fluss.utils.MapUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.util.Map;
/** The metric group for tablet server. */
public class TabletServerMetricGroup extends AbstractMetricGroup {
+ private static final Logger LOG =
LoggerFactory.getLogger(TabletServerMetricGroup.class);
Review Comment:
remove, not used
--
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]
