risingwave_ctl/cmd_impl/
await_tree.rs

1// Copyright 2023 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 risingwave_common::util::StackTraceResponseExt;
16use risingwave_pb::common::WorkerType;
17use risingwave_pb::id::WorkerId;
18use risingwave_pb::monitor_service::StackTraceRequest;
19use risingwave_pb::monitor_service::stack_trace_request::ActorTracesFormat;
20use risingwave_rpc_client::MonitorClientPool;
21use rw_diagnose_tools::await_tree::AnalyzeSummary;
22use thiserror_ext::AsReport as _;
23
24use crate::CtlContext;
25
26pub async fn dump(context: &CtlContext, actor_traces_format: Option<String>) -> anyhow::Result<()> {
27    let actor_traces_format = match actor_traces_format.as_deref() {
28        Some("text") => ActorTracesFormat::Text,
29        Some("json") | None => ActorTracesFormat::Json,
30        _ => return Err(anyhow::anyhow!("Invalid actor traces format")),
31    };
32
33    // Query the meta node for the await tree of all nodes in the cluster.
34    let meta_client = context.meta_client().await?;
35    let all = meta_client
36        .get_cluster_stack_trace(actor_traces_format)
37        .await?;
38
39    if all.actor_traces.is_empty()
40        && all.rpc_traces.is_empty()
41        && all.batch_traces.is_empty()
42        && all.compaction_task_traces.is_empty()
43        && all.inflight_barrier_traces.is_empty()
44    {
45        eprintln!("No actors are running, or `--async-stack-trace` not set?");
46    }
47    println!("{}", all.output());
48
49    Ok(())
50}
51
52pub async fn bottleneck_detect(context: &CtlContext, path: Option<String>) -> anyhow::Result<()> {
53    let summary = if let Some(path) = path {
54        rw_diagnose_tools::await_tree::bottleneck_detect_from_file(&path)?
55    } else {
56        bottleneck_detect_real_time(context).await?
57    };
58    println!("{}", summary);
59    Ok(())
60}
61
62async fn bottleneck_detect_real_time(context: &CtlContext) -> anyhow::Result<AnalyzeSummary> {
63    let meta_client = context.meta_client().await?;
64
65    let compute_nodes = meta_client
66        .list_worker_nodes(Some(WorkerType::ComputeNode))
67        .await?;
68    let clients = MonitorClientPool::adhoc();
69
70    // request for json actor traces
71    let req = StackTraceRequest::default();
72
73    let mut summary = AnalyzeSummary::new();
74    let mut errors: Vec<(WorkerId, String)> = Vec::new();
75    for cn in compute_nodes {
76        let worker_id = cn.id;
77
78        let client = match clients.get(&cn).await {
79            Ok(client) => client,
80            Err(e) => {
81                errors.push((worker_id, format!("failed to connect: {}", e.as_report())));
82                continue;
83            }
84        };
85
86        let response = match client.await_tree(req).await {
87            Ok(resp) => resp,
88            Err(e) => {
89                errors.push((
90                    worker_id,
91                    format!("failed to collect stack trace: {}", e.as_report()),
92                ));
93                continue;
94            }
95        };
96
97        let partial_summary = AnalyzeSummary::from_traces(&response.actor_traces)?;
98        summary.merge_other(&partial_summary);
99    }
100    if !errors.is_empty() {
101        eprintln!("Some compute nodes failed to dump await tree:");
102        for (worker_id, err) in errors {
103            eprintln!("  - worker {worker_id}: {err}");
104        }
105    }
106    Ok(summary)
107}
risingwave_ctl/cmd_impl/await_tree.rs

risingwave_ctl/cmd_impl/
await_tree.rs
