risingwave_ctl/cmd_impl/hummock/
refill.rs

1// Copyright 2026 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::time::Duration;
16
17use anyhow::Context;
18use futures::future::try_join_all;
19use itertools::Itertools;
20use risingwave_common::monitor::EndpointExt;
21use risingwave_pb::common::WorkerType;
22use risingwave_pb::monitor_service::GetTableCacheRefillStatsRequest;
23use risingwave_pb::monitor_service::monitor_service_client::MonitorServiceClient;
24use serde::Serialize;
25use serde_json::Value;
26use tonic::transport::Endpoint;
27
28use crate::common::CtlContext;
29
30#[derive(Serialize)]
31struct RefillStatsEntry {
32    worker_id: u32,
33    host: String,
34    port: u32,
35    stats: Value,
36}
37
38pub async fn refill_stats(context: &CtlContext) -> anyhow::Result<()> {
39    let meta_client = context.meta_client().await?;
40    let worker_nodes = meta_client.get_cluster_info().await?.worker_nodes;
41
42    let futures = worker_nodes
43        .into_iter()
44        .filter(|worker| worker.r#type() == WorkerType::ComputeNode)
45        .map(|worker| async move {
46            let host = worker.get_host().context("compute node host is missing")?;
47            let endpoint = format!("http://{}:{}", host.host, host.port);
48            let channel = Endpoint::from_shared(endpoint)?
49                .connect_timeout(Duration::from_secs(5))
50                .monitored_connect("grpc-table-cache-refill-stats-client", Default::default())
51                .await?;
52            let mut client = MonitorServiceClient::new(channel);
53            let response = client
54                .get_table_cache_refill_stats(GetTableCacheRefillStatsRequest {})
55                .await?
56                .into_inner();
57            let stats: Value = serde_json::from_str(&response.stats)
58                .context("failed to parse table cache refill stats json")?;
59            Ok::<_, anyhow::Error>(RefillStatsEntry {
60                worker_id: worker.id.as_raw_id(),
61                host: host.host.clone(),
62                port: host.port as _,
63                stats,
64            })
65        })
66        .collect_vec();
67
68    let results = try_join_all(futures).await?;
69    println!("{}", serde_json::to_string_pretty(&results)?);
70    Ok(())
71}
risingwave_ctl/cmd_impl/hummock/refill.rs

risingwave_ctl/cmd_impl/hummock/
refill.rs
