risingwave_common/util/
prost.rs

1// Copyright 2022 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::collections::btree_map::Entry;
16use std::fmt::{Display, Formatter};
17use std::ops::Deref;
18
19use risingwave_pb::batch_plan;
20use risingwave_pb::monitor_service::StackTraceResponse;
21use tracing::warn;
22
23pub trait TypeUrl {
24    fn type_url() -> &'static str;
25}
26
27impl TypeUrl for batch_plan::ExchangeNode {
28    fn type_url() -> &'static str {
29        "type.googleapis.com/plan.ExchangeNode"
30    }
31}
32
33pub struct StackTraceResponseOutput<'a>(&'a StackTraceResponse);
34
35impl Deref for StackTraceResponseOutput<'_> {
36    type Target = StackTraceResponse;
37
38    fn deref(&self) -> &Self::Target {
39        self.0
40    }
41}
42
43impl Display for StackTraceResponseOutput<'_> {
44    fn fmt(&self, s: &mut Formatter<'_>) -> std::fmt::Result {
45        if !self.node_errors.is_empty() {
46            writeln!(s, "--- Stack Trace Errors ---")?;
47            for (worker_id, err) in &self.node_errors {
48                writeln!(s, ">> Worker {worker_id}")?;
49                writeln!(s, "{err}\n")?;
50            }
51        }
52        if !self.actor_traces.is_empty() {
53            writeln!(s, "--- Actor Traces ---")?;
54            for (actor_id, trace) in &self.actor_traces {
55                writeln!(s, ">> Actor {}", *actor_id)?;
56                writeln!(s, "{trace}")?;
57            }
58        }
59        if !self.rpc_traces.is_empty() {
60            let _ = writeln!(s, "--- RPC Traces ---");
61            for (name, trace) in &self.rpc_traces {
62                writeln!(s, ">> RPC {name}")?;
63                writeln!(s, "{trace}")?;
64            }
65        }
66        if !self.batch_traces.is_empty() {
67            writeln!(s, "--- Batch Traces ---")?;
68            for (name, trace) in &self.batch_traces {
69                writeln!(s, ">> Batch Task {name}")?;
70                writeln!(s, "{trace}")?;
71            }
72        }
73        if !self.compaction_task_traces.is_empty() {
74            writeln!(s, "--- Compactor Traces ---")?;
75            for (name, trace) in &self.compaction_task_traces {
76                writeln!(s, ">> Compaction Task {name}")?;
77                writeln!(s, "{trace}")?;
78            }
79        }
80
81        if !self.inflight_barrier_traces.is_empty() {
82            writeln!(s, "--- Inflight Barrier Traces ---")?;
83            for (name, trace) in &self.inflight_barrier_traces {
84                writeln!(s, ">> Barrier {name}")?;
85                writeln!(s, "{trace}")?;
86            }
87        }
88
89        writeln!(s, "\n\n--- Barrier Worker States ---")?;
90        for (worker_id, state) in &self.barrier_worker_state {
91            writeln!(s, ">> Worker {worker_id}")?;
92            writeln!(s, "{state}\n")?;
93        }
94
95        if !self.jvm_stack_traces.is_empty() {
96            writeln!(s, "\n\n--- JVM Stack Traces ---")?;
97            for (worker_id, state) in &self.jvm_stack_traces {
98                writeln!(s, ">> Worker {worker_id}")?;
99                writeln!(s, "{state}\n")?;
100            }
101        }
102
103        if !self.meta_traces.is_empty() {
104            writeln!(s, "\n\n--- Meta Traces ---")?;
105            for (key, value) in &self.meta_traces {
106                writeln!(s, ">> {key}")?;
107                writeln!(s, "{value}\n")?;
108            }
109        }
110
111        Ok(())
112    }
113}
114
115#[easy_ext::ext(StackTraceResponseExt)]
116impl StackTraceResponse {
117    pub fn merge_other(&mut self, b: StackTraceResponse) {
118        self.actor_traces.extend(b.actor_traces);
119        self.rpc_traces.extend(b.rpc_traces);
120        self.batch_traces.extend(b.batch_traces);
121        self.compaction_task_traces.extend(b.compaction_task_traces);
122        self.inflight_barrier_traces
123            .extend(b.inflight_barrier_traces);
124        for (worker_id, err) in b.node_errors {
125            if self.node_errors.contains_key(&worker_id) {
126                warn!(
127                    worker_id = %worker_id,
128                    error = %err,
129                    "duplicate node error. skipped"
130                );
131                continue;
132            }
133            self.node_errors.insert(worker_id, err);
134        }
135        for (worker_id, worker_state) in b.barrier_worker_state {
136            match self.barrier_worker_state.entry(worker_id) {
137                Entry::Occupied(_entry) => {
138                    warn!(
139                        %worker_id,
140                        worker_state, "duplicate barrier worker state. skipped"
141                    );
142                }
143                Entry::Vacant(entry) => {
144                    entry.insert(worker_state);
145                }
146            }
147        }
148        for (worker_id, worker_state) in b.jvm_stack_traces {
149            match self.jvm_stack_traces.entry(worker_id) {
150                Entry::Occupied(_entry) => {
151                    warn!(
152                        %worker_id,
153                        worker_state, "duplicate jvm stack trace. skipped"
154                    );
155                }
156                Entry::Vacant(entry) => {
157                    entry.insert(worker_state);
158                }
159            }
160        }
161        self.meta_traces.extend(b.meta_traces);
162    }
163
164    pub fn output(&self) -> StackTraceResponseOutput<'_> {
165        StackTraceResponseOutput(self)
166    }
167}
risingwave_common/util/prost.rs

risingwave_common/util/
prost.rs
