diff --git a/fluss-client/src/main/java/org/apache/fluss/client/admin/Admin.java b/fluss-client/src/main/java/org/apache/fluss/client/admin/Admin.java index 5d749b6d432..9053c9d6bbc 100644 --- a/fluss-client/src/main/java/org/apache/fluss/client/admin/Admin.java +++ b/fluss-client/src/main/java/org/apache/fluss/client/admin/Admin.java @@ -89,7 +89,13 @@ @PublicEvolving public interface Admin extends AutoCloseable { - /** Get the current server node information. asynchronously. */ + /** + * Gets the current server node information and a machine resource snapshot asynchronously. + * + *

The returned {@link ServerNode} contains {@link ServerNode#resourceInfo()} for each + * currently available node. The resource snapshot includes CPU, memory and data disk + * information and is collected when this method is called. + */ CompletableFuture> getServerNodes(); /** @@ -289,6 +295,17 @@ CompletableFuture createTable( */ CompletableFuture> listTables(String databaseName); + /** + * Lists all tables in a database together with their current statistics asynchronously. + * + *

The returned table metadata is static metadata, while the statistics contain a snapshot of + * the data size and row count collected from the tablet servers. + * + * @param databaseName the name of the database + * @return a future containing the table metadata and statistics + */ + CompletableFuture> listTableDetails(String databaseName); + /** * Alter a table with the given {@code tableChanges}. * @@ -558,6 +575,9 @@ ListOffsetsResult listOffsets( /** * Asynchronously gets the statistics of this table. * + *

The returned statistics include the row count and the current local physical data size + * reported by tablet leaders. The data size is nullable for servers that do not provide it. + * * @return A future TableStats */ CompletableFuture getTableStats(TablePath tablePath); diff --git a/fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java b/fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java index a0909ceec73..8d43a056891 100644 --- a/fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java +++ b/fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java @@ -26,7 +26,9 @@ import org.apache.fluss.client.metadata.RemoteLogManifestInfo; import org.apache.fluss.client.utils.ClientRpcMessageUtils; import org.apache.fluss.cluster.Cluster; +import org.apache.fluss.cluster.NodeResourceInfo; import org.apache.fluss.cluster.ServerNode; +import org.apache.fluss.cluster.ServerType; import org.apache.fluss.cluster.rebalance.GoalType; import org.apache.fluss.cluster.rebalance.RebalanceProgress; import org.apache.fluss.cluster.rebalance.ServerTag; @@ -76,6 +78,8 @@ import org.apache.fluss.rpc.messages.GetLakeSnapshotRequest; import org.apache.fluss.rpc.messages.GetLatestKvSnapshotsRequest; import org.apache.fluss.rpc.messages.GetProducerOffsetsRequest; +import org.apache.fluss.rpc.messages.GetServerInfoRequest; +import org.apache.fluss.rpc.messages.GetServerInfoResponse; import org.apache.fluss.rpc.messages.GetTableInfoRequest; import org.apache.fluss.rpc.messages.GetTableSchemaRequest; import org.apache.fluss.rpc.messages.GetTableStatsRequest; @@ -180,6 +184,10 @@ public FlussAdmin(RpcClient client, MetadataUpdater metadataUpdater) { @Override public CompletableFuture> getServerNodes() { + return getServerNodesWithoutResourceInfo().thenCompose(this::attachResourceInfo); + } + + private CompletableFuture> getServerNodesWithoutResourceInfo() { CompletableFuture> future = new CompletableFuture<>(); CompletableFuture.runAsync( () -> { @@ -203,6 +211,57 @@ public CompletableFuture> getServerNodes() { return future; } + private CompletableFuture> attachResourceInfo(List serverNodes) { + List> futures = new ArrayList<>(serverNodes.size()); + for (ServerNode serverNode : serverNodes) { + futures.add( + getNodeResourceInfo(serverNode) + .thenApply(resourceInfo -> serverNode.withResourceInfo(resourceInfo))); + } + return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])) + .thenApply( + ignored -> { + List nodes = new ArrayList<>(futures.size()); + for (CompletableFuture future : futures) { + nodes.add(future.join()); + } + return nodes; + }); + } + + private CompletableFuture getNodeResourceInfo(ServerNode serverNode) { + CompletableFuture responseFuture; + if (serverNode.serverType() == ServerType.COORDINATOR) { + responseFuture = + metadataUpdater + .newCoordinatorServerClient() + .getServerInfo(new GetServerInfoRequest()); + } else { + TabletServerGateway tabletGateway = + metadataUpdater.newTabletServerClientForNode(serverNode.id()); + if (tabletGateway == null) { + CompletableFuture unavailable = new CompletableFuture<>(); + unavailable.completeExceptionally( + new FlussRuntimeException( + "Tablet server is no longer available: " + serverNode)); + return unavailable; + } + responseFuture = tabletGateway.getServerInfo(new GetServerInfoRequest()); + } + return responseFuture.thenApply(FlussAdmin::toNodeResourceInfo); + } + + private static NodeResourceInfo toNodeResourceInfo(GetServerInfoResponse response) { + return new NodeResourceInfo( + response.getCpuCores(), + response.getMemoryTotalBytes(), + response.getCpuUsageRatio(), + response.getMemoryUsedBytes(), + response.hasDataDiskTotalBytes() ? response.getDataDiskTotalBytes() : null, + response.hasDataDiskUsedBytes() ? response.getDataDiskUsedBytes() : null, + response.getCollectedAtMs()); + } + @Override public CompletableFuture getTableSchema(TablePath tablePath) { GetTableSchemaRequest request = new GetTableSchemaRequest(); @@ -373,6 +432,39 @@ public CompletableFuture> listTables(String databaseName) { return readOnlyGateway.listTables(request).thenApply(ListTablesResponse::getTableNamesList); } + @Override + public CompletableFuture> listTableDetails(String databaseName) { + return listTables(databaseName) + .thenCompose( + tableNames -> { + List> detailFutures = + new ArrayList<>(tableNames.size()); + for (String tableName : tableNames) { + TablePath tablePath = TablePath.of(databaseName, tableName); + CompletableFuture tableInfoFuture = + getTableInfo(tablePath); + CompletableFuture tableStatsFuture = + getTableStats(tablePath); + detailFutures.add( + tableInfoFuture.thenCombine( + tableStatsFuture, TableInfoWithStats::new)); + } + + return CompletableFuture.allOf( + detailFutures.toArray(new CompletableFuture[0])) + .thenApply( + ignored -> { + List details = + new ArrayList<>(detailFutures.size()); + for (CompletableFuture future : + detailFutures) { + details.add(future.join()); + } + return details; + }); + }); + } + @Override public CompletableFuture> listPartitionInfos(TablePath tablePath) { return listPartitionInfos(tablePath, null); @@ -557,7 +649,7 @@ public CompletableFuture getTableStats(TablePath tablePath) { partitionInfos = Collections.singletonList(null); } // create all TableBuckets for each partition and bucket combination - Map> bucketToRowCountMap = new HashMap<>(); + Map> bucketToStatsMap = new HashMap<>(); for (PartitionInfo partitionInfo : partitionInfos) { for (int bucket = 0; bucket < bucketCount; bucket++) { TableBucket tb = @@ -565,19 +657,34 @@ public CompletableFuture getTableStats(TablePath tablePath) { tableInfo.getTableId(), partitionInfo == null ? null : partitionInfo.getPartitionId(), bucket); - bucketToRowCountMap.put(tb, new CompletableFuture<>()); + bucketToStatsMap.put(tb, new CompletableFuture<>()); } } Map requestMap = prepareTableStatsRequests( - metadataUpdater, bucketToRowCountMap.keySet(), tablePath); + metadataUpdater, bucketToStatsMap.keySet(), tablePath); sendTableStatsRequest( - metadataUpdater, tableInfo.getTableId(), requestMap, bucketToRowCountMap); - return FutureUtils.combineAll(bucketToRowCountMap.values()) + metadataUpdater, tableInfo.getTableId(), requestMap, bucketToStatsMap); + return FutureUtils.combineAll(bucketToStatsMap.values()) .thenApply( - counts -> { - long totalRowCount = counts.stream().reduce(0L, Long::sum); - return new TableStats(totalRowCount); + stats -> { + long totalRowCount = 0L; + long totalDataSizeBytes = 0L; + long collectedAtMs = -1L; + boolean dataSizeAvailable = true; + for (BucketStats stat : stats) { + totalRowCount += stat.rowCount; + if (stat.dataSizeBytes == null) { + dataSizeAvailable = false; + } else { + totalDataSizeBytes += stat.dataSizeBytes; + } + collectedAtMs = Math.max(collectedAtMs, stat.collectedAtMs); + } + return new TableStats( + totalRowCount, + dataSizeAvailable ? totalDataSizeBytes : null, + collectedAtMs); }); } catch (Exception e) { throw new FlussRuntimeException( @@ -826,7 +933,7 @@ private static void sendTableStatsRequest( MetadataUpdater metadataUpdater, long tableId, Map leaderToRequestMap, - Map> bucketToRowCountMap) { + Map> bucketToStatsMap) { leaderToRequestMap.forEach( (leader, request) -> { TabletServerGateway gateway = @@ -839,7 +946,7 @@ private static void sendTableStatsRequest( .whenComplete( (response, t) -> handleTableStatsResponse( - response, t, tableId, bucketToRowCountMap)); + response, t, tableId, bucketToStatsMap)); } }); } @@ -848,12 +955,13 @@ private static void handleTableStatsResponse( GetTableStatsResponse response, Throwable t, long tableId, - Map> bucketToRowCountMap) { + Map> bucketToStatsMap) { if (t != null) { // fail all futures to fail fast - bucketToRowCountMap.values().forEach(f -> f.completeExceptionally(t)); + bucketToStatsMap.values().forEach(f -> f.completeExceptionally(t)); return; } + long collectedAtMs = response.hasCollectedAtMs() ? response.getCollectedAtMs() : -1L; for (PbTableStatsRespForBucket resp : response.getBucketsRespsList()) { TableBucket tb = new TableBucket( @@ -861,15 +969,33 @@ private static void handleTableStatsResponse( resp.hasPartitionId() ? resp.getPartitionId() : null, resp.getBucketId()); if (resp.hasErrorCode()) { - bucketToRowCountMap + bucketToStatsMap .get(tb) .completeExceptionally(ApiError.fromErrorMessage(resp).exception()); } else { - bucketToRowCountMap.get(tb).complete(resp.getRowCount()); + bucketToStatsMap + .get(tb) + .complete( + new BucketStats( + resp.getRowCount(), + resp.hasDataSizeBytes() ? resp.getDataSizeBytes() : null, + collectedAtMs)); } } } + private static final class BucketStats { + private final long rowCount; + private final @Nullable Long dataSizeBytes; + private final long collectedAtMs; + + private BucketStats(long rowCount, @Nullable Long dataSizeBytes, long collectedAtMs) { + this.rowCount = rowCount; + this.dataSizeBytes = dataSizeBytes; + this.collectedAtMs = collectedAtMs; + } + } + private static Map prepareListOffsetsRequests( MetadataUpdater metadataUpdater, long tableId, diff --git a/fluss-client/src/main/java/org/apache/fluss/client/admin/TableInfoWithStats.java b/fluss-client/src/main/java/org/apache/fluss/client/admin/TableInfoWithStats.java new file mode 100644 index 00000000000..5de711d4b5c --- /dev/null +++ b/fluss-client/src/main/java/org/apache/fluss/client/admin/TableInfoWithStats.java @@ -0,0 +1,81 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.client.admin; + +import org.apache.fluss.annotation.PublicEvolving; +import org.apache.fluss.metadata.TableInfo; +import org.apache.fluss.metadata.TableStats; + +import java.util.Objects; + +/** + * A table's static metadata together with its current statistics. + * + * @since 0.9 + */ +@PublicEvolving +public final class TableInfoWithStats { + + private final TableInfo tableInfo; + private final TableStats tableStats; + + /** Creates a table information and statistics result. */ + public TableInfoWithStats(TableInfo tableInfo, TableStats tableStats) { + this.tableInfo = tableInfo; + this.tableStats = tableStats; + } + + /** Returns the static table metadata. */ + public TableInfo getTableInfo() { + return tableInfo; + } + + /** Returns the current table statistics. */ + public TableStats getTableStats() { + return tableStats; + } + + @Override + public boolean equals(Object other) { + if (this == other) { + return true; + } + if (other == null || getClass() != other.getClass()) { + return false; + } + TableInfoWithStats that = (TableInfoWithStats) other; + return Objects.equals(tableInfo, that.tableInfo) + && Objects.equals(tableStats, that.tableStats); + } + + @Override + public int hashCode() { + return Objects.hash(tableInfo, tableStats); + } + + @Override + public String toString() { + return "TableInfoWithStats{" + + "tableInfo=" + + tableInfo + + ", tableStats=" + + tableStats + + '}'; + } +} diff --git a/fluss-client/src/test/java/org/apache/fluss/client/ServerInfoClientExample.java b/fluss-client/src/test/java/org/apache/fluss/client/ServerInfoClientExample.java new file mode 100644 index 00000000000..f4728e48fd2 --- /dev/null +++ b/fluss-client/src/test/java/org/apache/fluss/client/ServerInfoClientExample.java @@ -0,0 +1,59 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.client; + +import org.apache.fluss.client.admin.Admin; +import org.apache.fluss.cluster.ServerNode; +import org.apache.fluss.config.Configuration; + +import java.util.List; + +/** A small executable example for querying server node information through the Java client. */ +public final class ServerInfoClientExample { + + private ServerInfoClientExample() {} + + /** + * Connects to Fluss and prints node information and the latest machine resource snapshot + * reported by all coordinator and tablet servers. + * + * @param args optionally contains the bootstrap server list, for example {@code localhost:9123} + * @throws Exception if the client cannot connect or the request fails + */ + public static void main(String[] args) throws Exception { + String bootstrapServers = args.length == 0 ? "localhost:9123" : args[0]; + + Configuration configuration = new Configuration(); + configuration.setString("bootstrap.servers", bootstrapServers); + + try (Connection connection = ConnectionFactory.createConnection(configuration); + Admin admin = connection.getAdmin()) { + List serverNodes = admin.getServerNodes().get(); + if (serverNodes.isEmpty()) { + System.out.println("No Fluss server nodes were returned."); + return; + } + + for (ServerNode serverNode : serverNodes) { + System.out.println(serverNode); + System.out.println(" resourceInfo=" + serverNode.resourceInfo()); + } + } + } +} diff --git a/fluss-common/src/main/java/org/apache/fluss/cluster/NodeResourceInfo.java b/fluss-common/src/main/java/org/apache/fluss/cluster/NodeResourceInfo.java new file mode 100644 index 00000000000..74c9ecfd96f --- /dev/null +++ b/fluss-common/src/main/java/org/apache/fluss/cluster/NodeResourceInfo.java @@ -0,0 +1,140 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.cluster; + +import org.apache.fluss.annotation.PublicEvolving; + +import javax.annotation.Nullable; + +import java.util.Objects; + +/** A snapshot of the machine resources available to a Fluss server node. */ +@PublicEvolving +public final class NodeResourceInfo { + + private final double cpuCores; + private final long memoryTotalBytes; + private final double cpuUsageRatio; + private final long memoryUsedBytes; + private final @Nullable Long dataDiskTotalBytes; + private final @Nullable Long dataDiskUsedBytes; + private final long collectedAtMs; + + /** Creates a node resource information snapshot. */ + public NodeResourceInfo( + double cpuCores, + long memoryTotalBytes, + double cpuUsageRatio, + long memoryUsedBytes, + @Nullable Long dataDiskTotalBytes, + @Nullable Long dataDiskUsedBytes, + long collectedAtMs) { + this.cpuCores = cpuCores; + this.memoryTotalBytes = memoryTotalBytes; + this.cpuUsageRatio = cpuUsageRatio; + this.memoryUsedBytes = memoryUsedBytes; + this.dataDiskTotalBytes = dataDiskTotalBytes; + this.dataDiskUsedBytes = dataDiskUsedBytes; + this.collectedAtMs = collectedAtMs; + } + + /** Returns the effective CPU capacity visible to the Fluss process. */ + public double cpuCores() { + return cpuCores; + } + + /** Returns the effective memory capacity visible to the Fluss process, in bytes. */ + public long memoryTotalBytes() { + return memoryTotalBytes; + } + + /** Returns the current machine CPU usage ratio in the range [0, 1]. */ + public double cpuUsageRatio() { + return cpuUsageRatio; + } + + /** Returns the current machine memory usage, in bytes. */ + public long memoryUsedBytes() { + return memoryUsedBytes; + } + + /** Returns the total capacity of the Fluss data disks, or null when unavailable. */ + public @Nullable Long dataDiskTotalBytes() { + return dataDiskTotalBytes; + } + + /** Returns the used capacity of the Fluss data disks, or null when unavailable. */ + public @Nullable Long dataDiskUsedBytes() { + return dataDiskUsedBytes; + } + + /** Returns the collection timestamp of this snapshot in epoch milliseconds. */ + public long collectedAtMs() { + return collectedAtMs; + } + + @Override + public boolean equals(Object other) { + if (this == other) { + return true; + } + if (other == null || getClass() != other.getClass()) { + return false; + } + NodeResourceInfo that = (NodeResourceInfo) other; + return Double.compare(that.cpuCores, cpuCores) == 0 + && memoryTotalBytes == that.memoryTotalBytes + && Double.compare(that.cpuUsageRatio, cpuUsageRatio) == 0 + && memoryUsedBytes == that.memoryUsedBytes + && collectedAtMs == that.collectedAtMs + && Objects.equals(dataDiskTotalBytes, that.dataDiskTotalBytes) + && Objects.equals(dataDiskUsedBytes, that.dataDiskUsedBytes); + } + + @Override + public int hashCode() { + return Objects.hash( + cpuCores, + memoryTotalBytes, + cpuUsageRatio, + memoryUsedBytes, + dataDiskTotalBytes, + dataDiskUsedBytes, + collectedAtMs); + } + + @Override + public String toString() { + return "NodeResourceInfo{" + + "cpuCores=" + + cpuCores + + ", memoryTotalBytes=" + + memoryTotalBytes + + ", cpuUsageRatio=" + + cpuUsageRatio + + ", memoryUsedBytes=" + + memoryUsedBytes + + ", dataDiskTotalBytes=" + + dataDiskTotalBytes + + ", dataDiskUsedBytes=" + + dataDiskUsedBytes + + ", collectedAtMs=" + + collectedAtMs + + '}'; + } +} diff --git a/fluss-common/src/main/java/org/apache/fluss/cluster/ServerNode.java b/fluss-common/src/main/java/org/apache/fluss/cluster/ServerNode.java index a41cce4016e..ec19504e540 100644 --- a/fluss-common/src/main/java/org/apache/fluss/cluster/ServerNode.java +++ b/fluss-common/src/main/java/org/apache/fluss/cluster/ServerNode.java @@ -39,20 +39,34 @@ public class ServerNode { /** rack info for ServerNode. Currently, only tabletServer has rack info. */ private final @Nullable String rack; + /** The latest machine resource snapshot for this node, when available. */ + private final @Nullable NodeResourceInfo resourceInfo; + // Cache hashCode as it is called in performance sensitive parts of the code (e.g. // RecordAccumulator.ready) private Integer hash; public ServerNode(int id, String host, int port, ServerType serverType) { - this(id, host, port, serverType, null); + this(id, host, port, serverType, null, null); } public ServerNode(int id, String host, int port, ServerType serverType, @Nullable String rack) { + this(id, host, port, serverType, rack, null); + } + + public ServerNode( + int id, + String host, + int port, + ServerType serverType, + @Nullable String rack, + @Nullable NodeResourceInfo resourceInfo) { this.id = id; this.host = host; this.port = port; this.serverType = serverType; this.rack = rack; + this.resourceInfo = resourceInfo; if (serverType == ServerType.COORDINATOR) { this.uid = "cs-" + id; } else { @@ -96,6 +110,16 @@ public ServerType serverType() { return rack; } + /** Returns the latest machine resource snapshot for this node, or null when unavailable. */ + public @Nullable NodeResourceInfo resourceInfo() { + return resourceInfo; + } + + /** Returns a copy of this node with the given machine resource snapshot. */ + public ServerNode withResourceInfo(@Nullable NodeResourceInfo resourceInfo) { + return new ServerNode(id, host, port, serverType, rack, resourceInfo); + } + /** * Check whether this node is empty, which may be the case if noNode() is used as a placeholder * in a response payload with an error. diff --git a/fluss-common/src/main/java/org/apache/fluss/metadata/TableStats.java b/fluss-common/src/main/java/org/apache/fluss/metadata/TableStats.java index 4f4022b55a1..ea4657a0a8f 100644 --- a/fluss-common/src/main/java/org/apache/fluss/metadata/TableStats.java +++ b/fluss-common/src/main/java/org/apache/fluss/metadata/TableStats.java @@ -21,20 +21,33 @@ import org.apache.fluss.annotation.PublicEvolving; +import javax.annotation.Nullable; + import java.util.Objects; /** * Statistics of a table. * + *

The data size is the current local physical size reported by the tablet leaders. It may be + * unavailable when the server does not support this field. + * * @since 0.9 */ @PublicEvolving public class TableStats { private final long rowCount; + private final @Nullable Long dataSizeBytes; + private final long collectedAtMs; public TableStats(long rowCount) { + this(rowCount, null, -1L); + } + + public TableStats(long rowCount, @Nullable Long dataSizeBytes, long collectedAtMs) { this.rowCount = rowCount; + this.dataSizeBytes = dataSizeBytes; + this.collectedAtMs = collectedAtMs; } /** Returns the current total row count of the table. */ @@ -42,22 +55,41 @@ public long getRowCount() { return rowCount; } + /** Returns the current local data size of the table, or null if it is unavailable. */ + public @Nullable Long getDataSizeBytes() { + return dataSizeBytes; + } + + /** Returns the time at which the table statistics response was collected. */ + public long getCollectedAtMs() { + return collectedAtMs; + } + @Override public boolean equals(Object o) { if (o == null || getClass() != o.getClass()) { return false; } TableStats that = (TableStats) o; - return rowCount == that.rowCount; + return rowCount == that.rowCount + && collectedAtMs == that.collectedAtMs + && Objects.equals(dataSizeBytes, that.dataSizeBytes); } @Override public int hashCode() { - return Objects.hashCode(rowCount); + return Objects.hash(rowCount, dataSizeBytes, collectedAtMs); } @Override public String toString() { - return "TableStats{" + "rowCount=" + rowCount + '}'; + return "TableStats{" + + "rowCount=" + + rowCount + + ", dataSizeBytes=" + + dataSizeBytes + + ", collectedAtMs=" + + collectedAtMs + + '}'; } } diff --git a/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java b/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java index 644b11fd7e7..a23c0223146 100644 --- a/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java +++ b/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java @@ -109,6 +109,13 @@ public class MetricNames { public static final String HISTORICAL_LOOKUP_CACHE_CAPACITY_EVICTIONS = "lookupCacheCapacityEvictions"; + public static final String NODE_CPU_CORES = "machineCpuCores"; + public static final String NODE_MEMORY_TOTAL_BYTES = "machineMemoryTotalBytes"; + public static final String NODE_CPU_USAGE_RATIO = "machineCpuUsageRatio"; + public static final String NODE_MEMORY_USED_BYTES = "machineMemoryUsedBytes"; + public static final String DATA_DISK_TOTAL_BYTES = "dataDiskTotalBytes"; + public static final String DATA_DISK_USED_BYTES = "dataDiskUsedBytes"; + // -------------------------------------------------------------------------------------------- // metrics for user // -------------------------------------------------------------------------------------------- diff --git a/fluss-common/src/test/java/org/apache/fluss/cluster/ServerNodeTest.java b/fluss-common/src/test/java/org/apache/fluss/cluster/ServerNodeTest.java index 92914b53ec4..c8ebe3e0906 100644 --- a/fluss-common/src/test/java/org/apache/fluss/cluster/ServerNodeTest.java +++ b/fluss-common/src/test/java/org/apache/fluss/cluster/ServerNodeTest.java @@ -45,6 +45,13 @@ void testServerNode() { assertThat(serverNode.hashCode()).isNotEqualTo(serverNode2.hashCode()); assertThat(serverNode).isEqualTo(new ServerNode(0, "HOST1", 9023, ServerType.COORDINATOR)); + NodeResourceInfo resourceInfo = + new NodeResourceInfo(4.0, 1024L, 0.5, 512L, null, null, 1000L); + ServerNode enrichedServerNode = serverNode.withResourceInfo(resourceInfo); + assertThat(enrichedServerNode.resourceInfo()).isEqualTo(resourceInfo); + assertThat(enrichedServerNode).isEqualTo(serverNode); + assertThat(enrichedServerNode.hashCode()).isEqualTo(serverNode.hashCode()); + assertThat(serverNode.toString()).isEqualTo("HOST1:9023 (id: cs-0, rack: null)"); assertThat(serverNode2.toString()).isEqualTo("HOST2:9123 (id: ts-1, rack: null)"); } diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/testutils/TestAdminAdapter.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/testutils/TestAdminAdapter.java index e8120c83d26..dc79767257f 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/testutils/TestAdminAdapter.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/testutils/TestAdminAdapter.java @@ -26,6 +26,7 @@ import org.apache.fluss.client.admin.OffsetSpec; import org.apache.fluss.client.admin.ProducerOffsetsResult; import org.apache.fluss.client.admin.RegisterResult; +import org.apache.fluss.client.admin.TableInfoWithStats; import org.apache.fluss.client.metadata.ActiveKvSnapshots; import org.apache.fluss.client.metadata.KvSnapshotMetadata; import org.apache.fluss.client.metadata.KvSnapshots; @@ -162,6 +163,11 @@ public CompletableFuture> listTables(String databaseName) { throw new UnsupportedOperationException("Not implemented in TestAdminAdapter"); } + @Override + public CompletableFuture> listTableDetails(String databaseName) { + throw new UnsupportedOperationException("Not implemented in TestAdminAdapter"); + } + @Override public CompletableFuture alterTable( TablePath tablePath, List tableChanges, boolean ignoreIfNotExists) { diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/entity/TableStatsResultForBucket.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/entity/TableStatsResultForBucket.java index cdea0109ee6..f032b4c3340 100644 --- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/entity/TableStatsResultForBucket.java +++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/entity/TableStatsResultForBucket.java @@ -22,23 +22,38 @@ import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.rpc.protocol.ApiError; +import javax.annotation.Nullable; + /** Result of {@link org.apache.fluss.rpc.messages.GetTableStatsResponse} for each table bucket. */ public class TableStatsResultForBucket extends ResultForBucket { private final long rowCount; + private final @Nullable Long dataSizeBytes; public TableStatsResultForBucket(TableBucket tableBucket, long rowCount) { + this(tableBucket, rowCount, null); + } + + public TableStatsResultForBucket( + TableBucket tableBucket, long rowCount, @Nullable Long dataSizeBytes) { super(tableBucket); this.rowCount = rowCount; + this.dataSizeBytes = dataSizeBytes; } public TableStatsResultForBucket(TableBucket tableBucket, ApiError error) { super(tableBucket, error); this.rowCount = -1; + this.dataSizeBytes = null; } /** Returns the row count of the table bucket. If the request is failed, it will return -1. */ public long getRowCount() { return rowCount; } + + /** Returns the data size of the table bucket. If unavailable, it will return null. */ + public @Nullable Long getDataSizeBytes() { + return dataSizeBytes; + } } diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/gateway/AdminReadOnlyGateway.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/gateway/AdminReadOnlyGateway.java index 574e1a510dd..49055c03ac9 100644 --- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/gateway/AdminReadOnlyGateway.java +++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/gateway/AdminReadOnlyGateway.java @@ -34,6 +34,8 @@ import org.apache.fluss.rpc.messages.GetLakeSnapshotResponse; import org.apache.fluss.rpc.messages.GetLatestKvSnapshotsRequest; import org.apache.fluss.rpc.messages.GetLatestKvSnapshotsResponse; +import org.apache.fluss.rpc.messages.GetServerInfoRequest; +import org.apache.fluss.rpc.messages.GetServerInfoResponse; import org.apache.fluss.rpc.messages.GetTableInfoRequest; import org.apache.fluss.rpc.messages.GetTableInfoResponse; import org.apache.fluss.rpc.messages.GetTableSchemaRequest; @@ -202,4 +204,18 @@ CompletableFuture describeClusterConfigs( */ @RPC(api = ApiKeys.GET_CLUSTER_HEALTH) CompletableFuture getClusterHealth(GetClusterHealthRequest request); + + /** + * Get the latest server information collected by the server that receives the request. + * + * @param request get server information request + * @return a future returns the local server information snapshot + */ + @RPC(api = ApiKeys.GET_SERVER_INFO) + default CompletableFuture getServerInfo(GetServerInfoRequest request) { + CompletableFuture future = new CompletableFuture<>(); + future.completeExceptionally( + new UnsupportedOperationException("Server information unavailable")); + return future; + } } diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java index baf4256650e..84eb9ba9df4 100644 --- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java +++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java @@ -110,7 +110,8 @@ public enum ApiKeys { SCAN_KV(1061, 0, 0, PUBLIC), GET_CLUSTER_HEALTH(1062, 0, 0, PUBLIC), LIST_REMOTE_LOG_MANIFESTS(1063, 0, 0, PUBLIC), - LIST_KV_SNAPSHOTS(1064, 0, 0, PUBLIC); + LIST_KV_SNAPSHOTS(1064, 0, 0, PUBLIC), + GET_SERVER_INFO(1065, 0, 0, PUBLIC); private static final Map ID_TO_TYPE = Arrays.stream(ApiKeys.values()) diff --git a/fluss-rpc/src/main/proto/FlussApi.proto b/fluss-rpc/src/main/proto/FlussApi.proto index 2a05018d73e..a779f3c6d16 100644 --- a/fluss-rpc/src/main/proto/FlussApi.proto +++ b/fluss-rpc/src/main/proto/FlussApi.proto @@ -357,6 +357,7 @@ message GetTableStatsRequest { message GetTableStatsResponse { repeated PbTableStatsRespForBucket buckets_resp = 1; + optional int64 collected_at_ms = 2; } // notify bucket leader and isr request @@ -816,6 +817,20 @@ message GetClusterHealthResponse { required int32 status = 5; // PbClusterHealthStatus: GREEN=0, YELLOW=1, RED=2, UNKNOWN=3 } +message GetServerInfoRequest { } + +message GetServerInfoResponse { + required string server_id = 1; + required int32 server_type = 2; + required double cpu_cores = 3; + required int64 memory_total_bytes = 4; + required double cpu_usage_ratio = 5; + required int64 memory_used_bytes = 6; + optional int64 data_disk_total_bytes = 7; + optional int64 data_disk_used_bytes = 8; + required int64 collected_at_ms = 9; +} + // --------------- Inner classes ---------------- message PbApiVersion { @@ -1357,12 +1372,11 @@ message PbTableStatsRespForBucket { // Absent if row count is not available (e.g., WAL changelog mode or legacy tables). optional int64 row_count = 5; - // --- data size stats (future) --- + // --- data size stats --- // The data size in bytes of this bucket. // For KV tables: the size of the KV store. // For Log tables: the size of the log segments. - // Reserved for future use. - // optional int64 data_size_bytes = 6; + optional int64 data_size_bytes = 6; // --- Column-level stats (future) --- // Per-column statistics, keyed by column index. diff --git a/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java b/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java index e2ef1094ee1..c23b66dd206 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java @@ -58,6 +58,8 @@ import org.apache.fluss.rpc.messages.GetLakeSnapshotResponse; import org.apache.fluss.rpc.messages.GetLatestKvSnapshotsRequest; import org.apache.fluss.rpc.messages.GetLatestKvSnapshotsResponse; +import org.apache.fluss.rpc.messages.GetServerInfoRequest; +import org.apache.fluss.rpc.messages.GetServerInfoResponse; import org.apache.fluss.rpc.messages.GetTableInfoRequest; import org.apache.fluss.rpc.messages.GetTableInfoResponse; import org.apache.fluss.rpc.messages.GetTableSchemaRequest; @@ -92,6 +94,7 @@ import org.apache.fluss.server.metadata.PartitionNegativeCache; import org.apache.fluss.server.metadata.ServerMetadataCache; import org.apache.fluss.server.metadata.TableMetadata; +import org.apache.fluss.server.metrics.ServerNodeMetrics; import org.apache.fluss.server.tablet.TabletService; import org.apache.fluss.server.utils.ServerRpcMessageUtils; import org.apache.fluss.server.zk.ZooKeeperClient; @@ -149,6 +152,7 @@ public abstract class RpcServiceBase extends RpcGatewayService implements AdminR protected final @Nullable Authorizer authorizer; protected final DynamicConfigManager dynamicConfigManager; protected final PartitionNegativeCache partitionNegativeCache; + private final ServerNodeMetrics serverNodeMetrics; private long tokenLastUpdateTimeMs = 0; private ObtainedSecurityToken securityToken = null; @@ -162,7 +166,8 @@ public RpcServiceBase( MetadataManager metadataManager, @Nullable Authorizer authorizer, DynamicConfigManager dynamicConfigManager, - ExecutorService ioExecutor) { + ExecutorService ioExecutor, + ServerNodeMetrics serverNodeMetrics) { this.remoteFileSystem = remoteFileSystem; this.provider = provider; this.apiManager = new ApiManager(provider); @@ -172,6 +177,7 @@ public RpcServiceBase( this.dynamicConfigManager = dynamicConfigManager; this.partitionNegativeCache = new PartitionNegativeCache(); this.ioExecutor = ioExecutor; + this.serverNodeMetrics = serverNodeMetrics; } @VisibleForTesting @@ -604,6 +610,14 @@ public CompletableFuture describeClusterConfigs( new DescribeClusterConfigsResponse().addAllConfigs(toPbConfigEntries(configs))); } + @Override + public CompletableFuture getServerInfo(GetServerInfoRequest request) { + if (authorizer != null) { + authorizer.authorize(currentSession(), OperationType.DESCRIBE, Resource.cluster()); + } + return CompletableFuture.completedFuture(serverNodeMetrics.toResponse()); + } + protected MetadataResponse processMetadataRequest( MetadataRequest request, String listenerName, diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java index a2f9b4fad97..29171438557 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java @@ -38,6 +38,7 @@ import org.apache.fluss.server.metadata.CoordinatorMetadataCache; import org.apache.fluss.server.metadata.ServerMetadataCache; import org.apache.fluss.server.metrics.ServerMetricUtils; +import org.apache.fluss.server.metrics.ServerNodeMetrics; import org.apache.fluss.server.metrics.group.CoordinatorMetricGroup; import org.apache.fluss.server.metrics.group.LakeTieringMetricGroup; import org.apache.fluss.server.storage.DiskWriteLimitConfigValidator; @@ -106,6 +107,9 @@ public class CoordinatorServer extends ServerBase { @GuardedBy("lock") private CoordinatorMetricGroup serverMetricGroup; + @GuardedBy("lock") + private ServerNodeMetrics serverNodeMetrics; + @GuardedBy("lock") private RpcServer rpcServer; @@ -235,6 +239,16 @@ protected void initCoordinatorStandby() throws Exception { ServerMetricUtils.validateAndGetClusterId(conf), endpoints.get(0).getHost(), serverId); + this.serverNodeMetrics = + new ServerNodeMetrics( + conf, + metricRegistry, + ServerMetricUtils.validateAndGetClusterId(conf), + endpoints.get(0).getHost(), + serverId, + ServerType.COORDINATOR, + null, + scheduler); this.zkClient = ZooKeeperUtils.startZookeeperClient(conf, this); @@ -289,6 +303,7 @@ protected void initCoordinatorStandby() throws Exception { remoteDirDynamicLoader, dynamicConfigManager, ioExecutor, + serverNodeMetrics, kvSnapshotLeaseManager, coordinatorLeaderElection, replicaCapacityController); @@ -590,6 +605,15 @@ CompletableFuture stopServices() { synchronized (lock) { Throwable exception = leaderElectionException; + try { + if (serverNodeMetrics != null) { + serverNodeMetrics.close(); + serverNodeMetrics = null; + } + } catch (Throwable t) { + exception = ExceptionUtils.firstOrSuppressed(t, exception); + } + try { // We must shut down the scheduler early because otherwise, the scheduler could // touch other resources that might have been shutdown and cause exceptions. diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java index 344839513fd..e9507732dae 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java @@ -174,6 +174,7 @@ import org.apache.fluss.server.kv.snapshot.CompletedSnapshotJsonSerde; import org.apache.fluss.server.metadata.CoordinatorMetadataCache; import org.apache.fluss.server.metadata.CoordinatorMetadataProvider; +import org.apache.fluss.server.metrics.ServerNodeMetrics; import org.apache.fluss.server.utils.ServerRpcMessageUtils; import org.apache.fluss.server.zk.ZooKeeperClient; import org.apache.fluss.server.zk.ZooKeeperClient.TableBucketAndManifest; @@ -276,6 +277,7 @@ public CoordinatorService( RemoteDirDynamicLoader remoteDirDynamicLoader, DynamicConfigManager dynamicConfigManager, ExecutorService ioExecutor, + ServerNodeMetrics serverNodeMetrics, KvSnapshotLeaseManager kvSnapshotLeaseManager, CoordinatorLeaderElection coordinatorLeaderElection, ReplicaCapacityController replicaCapacityController) { @@ -286,7 +288,8 @@ public CoordinatorService( metadataManager, authorizer, dynamicConfigManager, - ioExecutor); + ioExecutor, + serverNodeMetrics); this.defaultBucketNumber = conf.getInt(ConfigOptions.DEFAULT_BUCKET_NUMBER); this.defaultReplicationFactor = conf.getInt(ConfigOptions.DEFAULT_REPLICATION_FACTOR); this.logTableAllowCreation = conf.getBoolean(ConfigOptions.LOG_TABLE_ALLOW_CREATION); diff --git a/fluss-server/src/main/java/org/apache/fluss/server/metrics/ServerNodeMetricGroup.java b/fluss-server/src/main/java/org/apache/fluss/server/metrics/ServerNodeMetricGroup.java new file mode 100644 index 00000000000..ca016d37a98 --- /dev/null +++ b/fluss-server/src/main/java/org/apache/fluss/server/metrics/ServerNodeMetricGroup.java @@ -0,0 +1,62 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.server.metrics; + +import org.apache.fluss.cluster.ServerType; +import org.apache.fluss.metrics.CharacterFilter; +import org.apache.fluss.metrics.groups.AbstractMetricGroup; +import org.apache.fluss.metrics.registry.MetricRegistry; + +import java.util.Map; + +/** Metric group containing node metrics shared by coordinator and tablet servers. */ +final class ServerNodeMetricGroup extends AbstractMetricGroup { + + private static final String NAME = "server"; + + private final String clusterId; + private final String hostname; + private final String serverId; + private final ServerType serverType; + + ServerNodeMetricGroup( + MetricRegistry registry, + String clusterId, + String hostname, + String serverId, + ServerType serverType) { + super(registry, new String[] {clusterId, hostname, NAME}, null); + this.clusterId = clusterId; + this.hostname = hostname; + this.serverId = serverId; + this.serverType = serverType; + } + + @Override + protected String getGroupName(CharacterFilter filter) { + return NAME; + } + + @Override + protected void putVariables(Map variables) { + variables.put("cluster_id", clusterId); + variables.put("host", hostname); + variables.put("server_id", String.valueOf(serverId)); + variables.put("server_type", serverType.name().toLowerCase()); + } +} diff --git a/fluss-server/src/main/java/org/apache/fluss/server/metrics/ServerNodeMetrics.java b/fluss-server/src/main/java/org/apache/fluss/server/metrics/ServerNodeMetrics.java new file mode 100644 index 00000000000..98ee6966ee3 --- /dev/null +++ b/fluss-server/src/main/java/org/apache/fluss/server/metrics/ServerNodeMetrics.java @@ -0,0 +1,256 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.server.metrics; + +import org.apache.fluss.cluster.ServerType; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.metrics.Gauge; +import org.apache.fluss.metrics.MetricNames; +import org.apache.fluss.metrics.groups.MetricGroup; +import org.apache.fluss.metrics.registry.MetricRegistry; +import org.apache.fluss.rpc.messages.GetServerInfoResponse; +import org.apache.fluss.server.metadata.TabletServerResource; +import org.apache.fluss.server.storage.DiskUsageCollector; +import org.apache.fluss.server.tablet.TabletServerResourceProbe; +import org.apache.fluss.utils.concurrent.Scheduler; + +import javax.annotation.Nullable; + +import java.io.File; +import java.lang.management.ManagementFactory; +import java.util.List; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.atomic.AtomicReference; + +/** Collects and exposes one cached node metrics snapshot for a Fluss server process. */ +public final class ServerNodeMetrics implements AutoCloseable { + + private static final long SAMPLE_INTERVAL_MS = 10_000L; + + private final TabletServerResourceProbe resourceProbe; + private final String serverId; + private final ServerType serverType; + private final DiskUsageCollector diskUsageCollector; + private final AtomicReference latestSnapshot; + private final ScheduledFuture samplingTask; + private final ServerNodeMetricGroup metricGroup; + + /** Creates and starts a periodic node metrics collector. */ + public ServerNodeMetrics( + Configuration conf, + MetricRegistry registry, + String clusterId, + String hostname, + String serverId, + ServerType serverType, + @Nullable List dataDirs, + Scheduler scheduler) { + this.resourceProbe = new TabletServerResourceProbe(conf); + this.serverId = serverId; + this.serverType = serverType; + this.diskUsageCollector = dataDirs == null ? null : new DiskUsageCollector(dataDirs); + this.latestSnapshot = new AtomicReference<>(collectSnapshot()); + this.metricGroup = + new ServerNodeMetricGroup(registry, clusterId, hostname, serverId, serverType); + registerMetrics(metricGroup); + this.samplingTask = + scheduler.schedule("server-node-metrics", this::refresh, 0L, SAMPLE_INTERVAL_MS); + } + + /** Returns the latest cached node metrics snapshot. */ + public Snapshot snapshot() { + return latestSnapshot.get(); + } + + /** Converts the latest cached snapshot to the public RPC response. */ + public GetServerInfoResponse toResponse() { + Snapshot snapshot = snapshot(); + GetServerInfoResponse response = new GetServerInfoResponse(); + response.setServerId(serverId) + .setServerType(serverType.toTypeId()) + .setCpuCores(snapshot.getCpuCores()) + .setMemoryTotalBytes(snapshot.getMemoryTotalBytes()) + .setCpuUsageRatio(snapshot.getCpuUsageRatio()) + .setMemoryUsedBytes(snapshot.getMemoryUsedBytes()) + .setCollectedAtMs(snapshot.getCollectedAtMs()); + if (snapshot.hasDataDisk()) { + response.setDataDiskTotalBytes(snapshot.getDataDiskTotalBytes()); + response.setDataDiskUsedBytes(snapshot.getDataDiskUsedBytes()); + } + return response; + } + + /** Stops sampling and unregisters all node metrics. */ + @Override + public void close() { + samplingTask.cancel(false); + metricGroup.close(); + } + + private void refresh() { + latestSnapshot.set(collectSnapshot()); + } + + private Snapshot collectSnapshot() { + TabletServerResource resource = resourceProbe.probe(); + double cpuCores = + resource.getCpuCores() == null + ? Runtime.getRuntime().availableProcessors() + : resource.getCpuCores(); + long memoryTotalBytes = + resource.getMemoryBytes() == null + ? getOperatingSystemBean().getTotalPhysicalMemorySize() + : resource.getMemoryBytes(); + com.sun.management.OperatingSystemMXBean operatingSystemBean = getOperatingSystemBean(); + double cpuUsageRatio = normalizeCpuUsage(operatingSystemBean.getSystemCpuLoad()); + long memoryUsedBytes = + resourceProbe + .probeMemoryUsedBytes() + .orElse( + Math.max( + 0L, + operatingSystemBean.getTotalPhysicalMemorySize() + - operatingSystemBean.getFreePhysicalMemorySize())); + + Long dataDiskTotalBytes = null; + Long dataDiskUsedBytes = null; + if (diskUsageCollector != null) { + try { + DiskUsageCollector.DiskUsage diskUsage = diskUsageCollector.collectUsage(); + dataDiskTotalBytes = diskUsage.getTotalBytes(); + dataDiskUsedBytes = diskUsage.getUsedBytes(); + } catch (Exception ignored) { + // The previous disk values remain available through the previous snapshot. + Snapshot previous = latestSnapshot == null ? null : latestSnapshot.get(); + if (previous != null && previous.hasDataDisk()) { + dataDiskTotalBytes = previous.getDataDiskTotalBytes(); + dataDiskUsedBytes = previous.getDataDiskUsedBytes(); + } + } + } + return new Snapshot( + cpuCores, + memoryTotalBytes, + cpuUsageRatio, + memoryUsedBytes, + dataDiskTotalBytes, + dataDiskUsedBytes, + System.currentTimeMillis()); + } + + private com.sun.management.OperatingSystemMXBean getOperatingSystemBean() { + return (com.sun.management.OperatingSystemMXBean) + ManagementFactory.getOperatingSystemMXBean(); + } + + private double normalizeCpuUsage(double cpuUsageRatio) { + if (Double.isNaN(cpuUsageRatio) || cpuUsageRatio < 0.0) { + return 0.0; + } + return Math.min(cpuUsageRatio, 1.0); + } + + private void registerMetrics(MetricGroup group) { + group.>gauge( + MetricNames.NODE_CPU_CORES, () -> latestSnapshot.get().getCpuCores()); + group.>gauge( + MetricNames.NODE_MEMORY_TOTAL_BYTES, + () -> latestSnapshot.get().getMemoryTotalBytes()); + group.>gauge( + MetricNames.NODE_CPU_USAGE_RATIO, () -> latestSnapshot.get().getCpuUsageRatio()); + group.>gauge( + MetricNames.NODE_MEMORY_USED_BYTES, + () -> latestSnapshot.get().getMemoryUsedBytes()); + if (diskUsageCollector != null) { + group.>gauge( + MetricNames.DATA_DISK_TOTAL_BYTES, + () -> latestSnapshot.get().getDataDiskTotalBytes()); + group.>gauge( + MetricNames.DATA_DISK_USED_BYTES, + () -> latestSnapshot.get().getDataDiskUsedBytes()); + } + } + + /** Immutable node metrics snapshot. */ + public static final class Snapshot { + private final double cpuCores; + private final long memoryTotalBytes; + private final double cpuUsageRatio; + private final long memoryUsedBytes; + private final @Nullable Long dataDiskTotalBytes; + private final @Nullable Long dataDiskUsedBytes; + private final long collectedAtMs; + + private Snapshot( + double cpuCores, + long memoryTotalBytes, + double cpuUsageRatio, + long memoryUsedBytes, + @Nullable Long dataDiskTotalBytes, + @Nullable Long dataDiskUsedBytes, + long collectedAtMs) { + this.cpuCores = cpuCores; + this.memoryTotalBytes = memoryTotalBytes; + this.cpuUsageRatio = cpuUsageRatio; + this.memoryUsedBytes = memoryUsedBytes; + this.dataDiskTotalBytes = dataDiskTotalBytes; + this.dataDiskUsedBytes = dataDiskUsedBytes; + this.collectedAtMs = collectedAtMs; + } + + /** Returns the effective CPU capacity. */ + public double getCpuCores() { + return cpuCores; + } + + /** Returns the effective memory capacity in bytes. */ + public long getMemoryTotalBytes() { + return memoryTotalBytes; + } + + /** Returns the machine CPU usage ratio. */ + public double getCpuUsageRatio() { + return cpuUsageRatio; + } + + /** Returns the machine memory usage in bytes. */ + public long getMemoryUsedBytes() { + return memoryUsedBytes; + } + + /** Returns whether this snapshot contains Fluss data disk metrics. */ + public boolean hasDataDisk() { + return dataDiskTotalBytes != null && dataDiskUsedBytes != null; + } + + /** Returns the total Fluss data disk capacity in bytes. */ + public long getDataDiskTotalBytes() { + return dataDiskTotalBytes == null ? 0L : dataDiskTotalBytes; + } + + /** Returns the used Fluss data disk capacity in bytes. */ + public long getDataDiskUsedBytes() { + return dataDiskUsedBytes == null ? 0L : dataDiskUsedBytes; + } + + /** Returns the collection timestamp in epoch milliseconds. */ + public long getCollectedAtMs() { + return collectedAtMs; + } + } +} diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java index b3357019230..840c59b2882 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java @@ -1679,7 +1679,11 @@ public void getTableStats( try { Replica replica = getReplicaOrException(tb); long rowCount = replica.getRowCount(); - results.add(new TableStatsResultForBucket(tb, rowCount)); + long dataSizeBytes = replica.getLogTablet().logSize(); + if (replica.isKvTable()) { + dataSizeBytes += replica.getLatestKvSnapshotSize(); + } + results.add(new TableStatsResultForBucket(tb, rowCount, dataSizeBytes)); } catch (Exception e) { LOG.error("Error getting table stats on replica {}", tableBucket, e); results.add(new TableStatsResultForBucket(tb, ApiError.fromThrowable(e))); diff --git a/fluss-server/src/main/java/org/apache/fluss/server/storage/DiskUsageCollector.java b/fluss-server/src/main/java/org/apache/fluss/server/storage/DiskUsageCollector.java index 332d29ea90f..99a75f4ceb1 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/storage/DiskUsageCollector.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/storage/DiskUsageCollector.java @@ -65,6 +65,16 @@ public DiskUsageCollector(List dataDirs) { * IOException} is thrown only when all directories fail. */ public double collect() throws IOException { + DiskUsage diskUsage = collectUsage(); + return diskUsage.getUsageRatio(); + } + + /** + * Collects the total and used space of the distinct file stores backing the data directories. + */ + public DiskUsage collectUsage() throws IOException { + long totalSpace = 0L; + long usableSpace = 0L; double maxRatio = 0.0; Set counted = new HashSet<>(); int failures = 0; @@ -76,7 +86,10 @@ public double collect() throws IOException { if (total <= 0L) { continue; } - double ratio = (double) (total - fs.getUsableSpace()) / total; + long usable = fs.getUsableSpace(); + totalSpace += total; + usableSpace += usable; + double ratio = (double) (total - usable) / total; if (ratio > maxRatio) { maxRatio = ratio; } @@ -89,6 +102,34 @@ public double collect() throws IOException { if (failures > 0 && failures == dataDirs.size()) { throw new IOException("All " + failures + " data directories failed FileStore lookup."); } - return maxRatio; + return new DiskUsage(totalSpace, totalSpace - usableSpace, maxRatio); + } + + /** A snapshot of the local data disk capacity and usage. */ + public static final class DiskUsage { + private final long totalBytes; + private final long usedBytes; + private final double usageRatio; + + private DiskUsage(long totalBytes, long usedBytes, double usageRatio) { + this.totalBytes = totalBytes; + this.usedBytes = usedBytes; + this.usageRatio = usageRatio; + } + + /** Returns the total capacity in bytes. */ + public long getTotalBytes() { + return totalBytes; + } + + /** Returns the used capacity in bytes. */ + public long getUsedBytes() { + return usedBytes; + } + + /** Returns the maximum usage ratio across the collected file stores. */ + public double getUsageRatio() { + return usageRatio; + } } } diff --git a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java index 45dfacd09ff..f6c2e4806d5 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java @@ -47,6 +47,7 @@ import org.apache.fluss.server.metadata.TabletServerMetadataCache; import org.apache.fluss.server.metadata.TabletServerResource; import org.apache.fluss.server.metrics.ServerMetricUtils; +import org.apache.fluss.server.metrics.ServerNodeMetrics; import org.apache.fluss.server.metrics.UserMetrics; import org.apache.fluss.server.metrics.group.TabletServerMetricGroup; import org.apache.fluss.server.replica.ReplicaManager; @@ -138,6 +139,9 @@ public class TabletServer extends ServerBase { @GuardedBy("lock") private TabletServerMetricGroup tabletServerMetricGroup; + @GuardedBy("lock") + private ServerNodeMetrics serverNodeMetrics; + @GuardedBy("lock") private TabletServerMetadataCache metadataCache; @@ -242,6 +246,16 @@ protected void startServices() throws Exception { this.metadataCache = new TabletServerMetadataCache(metadataManager); this.localDiskManager = LocalDiskManager.create(conf); + this.serverNodeMetrics = + new ServerNodeMetrics( + conf, + metricRegistry, + ServerMetricUtils.validateAndGetClusterId(conf), + endpoints.get(0).getHost(), + String.valueOf(serverId), + ServerType.TABLET_SERVER, + localDiskManager.dataDirs(), + scheduler); this.logManager = LogManager.create( conf, @@ -322,6 +336,7 @@ protected void startServices() throws Exception { authorizer, dynamicConfigManager, ioExecutor, + serverNodeMetrics, replicaStateChangeExecutor, scannerManager, coordinatorGateway, @@ -433,6 +448,15 @@ CompletableFuture stopServices() { synchronized (lock) { Throwable exception = null; + try { + if (serverNodeMetrics != null) { + serverNodeMetrics.close(); + serverNodeMetrics = null; + } + } catch (Throwable t) { + exception = ExceptionUtils.firstOrSuppressed(t, exception); + } + try { if (tabletServerMetricGroup != null) { tabletServerMetricGroup.close(); diff --git a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServerResourceProbe.java b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServerResourceProbe.java index 1ea19d419bd..f7c68a316c2 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServerResourceProbe.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServerResourceProbe.java @@ -57,6 +57,15 @@ public TabletServerResource probe() { probeCpuCores().orElse(null), probeMemoryBytes().orElse(null)); } + /** Returns the current memory usage from the cgroup, when available. */ + public Optional probeMemoryUsedBytes() { + Optional cgroupV2Memory = readMemoryValue(cgroupRoot.resolve("memory.current")); + if (cgroupV2Memory.isPresent()) { + return cgroupV2Memory; + } + return readMemoryValue(cgroupRoot.resolve("memory").resolve("memory.usage_in_bytes")); + } + private Optional probeCpuCores() { Optional configuredCpuCores = conf.getOptional(ConfigOptions.TABLET_SERVER_ADVERTISED_RESOURCE_CPU_CORES); diff --git a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java index bd3ef49b35a..9e45bbb7441 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java @@ -105,6 +105,7 @@ import org.apache.fluss.server.metadata.ClusterMetadata; import org.apache.fluss.server.metadata.TabletServerMetadataCache; import org.apache.fluss.server.metadata.TabletServerMetadataProvider; +import org.apache.fluss.server.metrics.ServerNodeMetrics; import org.apache.fluss.server.replica.Replica; import org.apache.fluss.server.replica.ReplicaManager; import org.apache.fluss.server.utils.ServerRpcMessageUtils; @@ -181,6 +182,7 @@ public TabletService( @Nullable Authorizer authorizer, DynamicConfigManager dynamicConfigManager, ExecutorService ioExecutor, + ServerNodeMetrics serverNodeMetrics, ExecutorService replicaStateChangeExecutor, ScannerManager scannerManager, CoordinatorGateway coordinatorGateway, @@ -192,7 +194,8 @@ public TabletService( metadataManager, authorizer, dynamicConfigManager, - ioExecutor); + ioExecutor, + serverNodeMetrics); this.serviceName = "server-" + serverId; this.replicaManager = replicaManager; this.metadataCache = metadataCache; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java b/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java index 9952d485711..171bb98157f 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java @@ -2339,7 +2339,8 @@ public static List getTableStatsRequestData(GetTableStatsRequest re public static GetTableStatsResponse makeGetTableStatsResponse( List stats) { - GetTableStatsResponse response = new GetTableStatsResponse(); + GetTableStatsResponse response = + new GetTableStatsResponse().setCollectedAtMs(System.currentTimeMillis()); for (TableStatsResultForBucket statForBucket : stats) { TableBucket tb = statForBucket.getTableBucket(); PbTableStatsRespForBucket respForBucket = @@ -2352,6 +2353,9 @@ public static GetTableStatsResponse makeGetTableStatsResponse( statForBucket.getErrorCode(), statForBucket.getErrorMessage()); } else { respForBucket.setRowCount(statForBucket.getRowCount()); + if (statForBucket.getDataSizeBytes() != null) { + respForBucket.setDataSizeBytes(statForBucket.getDataSizeBytes()); + } } } return response; diff --git a/website/docs/apis/java/index.md b/website/docs/apis/java/index.md index 74c2bfa44c2..ee7bb59608c 100644 --- a/website/docs/apis/java/index.md +++ b/website/docs/apis/java/index.md @@ -158,6 +158,35 @@ System.out.println("Row count: " + stats.getRowCount()); `getTableStats` for Primary Key Tables requires the table to use the default changelog mode (`'table.changelog.image' = 'FULL'`). Tables configured with `'table.changelog.image' = 'WAL'` do not support this feature. ::: +## Server Information + +The `getServerNodes` method returns the current Coordinator and TabletServer nodes. Each returned +`ServerNode` includes a `NodeResourceInfo` snapshot containing the CPU, memory, and data disk +information collected from that node. This information is queried directly from the Fluss nodes +and does not require querying Prometheus. + +```java +List serverNodes = admin.getServerNodes().get(); +for (ServerNode serverNode : serverNodes) { + NodeResourceInfo resourceInfo = serverNode.resourceInfo(); + + System.out.println("Node: " + serverNode); + if (resourceInfo != null) { + System.out.println("CPU cores: " + resourceInfo.cpuCores()); + System.out.println("CPU usage ratio: " + resourceInfo.cpuUsageRatio()); + System.out.println("Memory total (bytes): " + resourceInfo.memoryTotalBytes()); + System.out.println("Memory used (bytes): " + resourceInfo.memoryUsedBytes()); + System.out.println("Data disk total (bytes): " + resourceInfo.dataDiskTotalBytes()); + System.out.println("Data disk used (bytes): " + resourceInfo.dataDiskUsedBytes()); + System.out.println("Collected at (epoch ms): " + resourceInfo.collectedAtMs()); + } +} +``` + +The data disk fields are `null` for Coordinator nodes because Coordinators do not store Fluss +table data locally. The snapshot timestamp is available through `collectedAtMs()` when the +freshness of the resource information needs to be checked. + ## Table API ### Writers In order to write data to Fluss tables, first you need to create a Table instance.