rocketmq_runtime/
shutdown_report.rs

1// Copyright 2023 The RocketMQ Rust Authors
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 serde::Serialize;
18
19use crate::blocking::BlockingTaskSnapshot;
20use crate::task_group::TaskGroupId;
21use crate::task_group::TaskId;
22use crate::task_group::TaskKind;
23use crate::task_group::TaskState;
24
25/// Represents shutdown report.
26#[derive(Debug, Serialize)]
27pub struct ShutdownReport {
28    /// The name value.
29    pub name: String,
30    /// The elapsed value.
31    #[serde(with = "duration_millis")]
32    pub elapsed: Duration,
33    /// The completed value.
34    pub completed: usize,
35    /// The cancelled value.
36    pub cancelled: usize,
37    /// The aborted value.
38    pub aborted: usize,
39    /// Lifecycle components that completed with an explicit failure.
40    pub failed: usize,
41    /// The panicked value.
42    pub panicked: usize,
43    /// Tasks this shutdown aborted after its deadline expired, plus tasks
44    /// still registered when it finished.
45    pub timed_out: usize,
46    /// Tasks still registered when the shutdown finished.
47    pub leaked: usize,
48    /// The blocking still running value.
49    pub blocking_still_running: usize,
50    /// The children value.
51    pub children: Vec<ShutdownReport>,
52    /// Tasks still registered when the shutdown finished, at most
53    /// [`Self::REMAINING_TASKS_LIMIT`] of them.
54    pub remaining_tasks: Vec<TaskSnapshot>,
55    /// Remaining tasks left out of `remaining_tasks` because of its limit.
56    /// They are still counted in `leaked`.
57    pub remaining_tasks_omitted: usize,
58    /// The blocking tasks value.
59    pub blocking_tasks: Vec<BlockingTaskSnapshot>,
60    /// The annotations value.
61    pub annotations: Vec<ShutdownAnnotation>,
62}
63
64impl Clone for ShutdownReport {
65    fn clone(&self) -> Self {
66        let mut copy = self.clone_local();
67        let mut pending = vec![(self, &mut copy)];
68        while let Some((source, target)) = pending.pop() {
69            target.children = source.children.iter().map(Self::clone_local).collect();
70            pending.extend(source.children.iter().zip(target.children.iter_mut()));
71        }
72        copy
73    }
74}
75
76impl ShutdownReport {
77    fn clone_local(&self) -> Self {
78        Self {
79            name: self.name.clone(),
80            elapsed: self.elapsed,
81            completed: self.completed,
82            cancelled: self.cancelled,
83            aborted: self.aborted,
84            failed: self.failed,
85            panicked: self.panicked,
86            timed_out: self.timed_out,
87            leaked: self.leaked,
88            blocking_still_running: self.blocking_still_running,
89            children: Vec::new(),
90            remaining_tasks: self.remaining_tasks.clone(),
91            remaining_tasks_omitted: self.remaining_tasks_omitted,
92            blocking_tasks: self.blocking_tasks.clone(),
93            annotations: self.annotations.clone(),
94        }
95    }
96
97    /// Creates a new `ShutdownReport`.
98    pub fn new(name: impl Into<String>, elapsed: Duration) -> Self {
99        Self {
100            name: name.into(),
101            elapsed,
102            completed: 0,
103            cancelled: 0,
104            aborted: 0,
105            failed: 0,
106            panicked: 0,
107            timed_out: 0,
108            leaked: 0,
109            blocking_still_running: 0,
110            children: Vec::new(),
111            remaining_tasks: Vec::new(),
112            remaining_tasks_omitted: 0,
113            blocking_tasks: Vec::new(),
114            annotations: Vec::new(),
115        }
116    }
117
118    /// Maximum number of task snapshots kept in `remaining_tasks`.
119    ///
120    /// The limit bounds the size of a report for a group that leaks many
121    /// tasks; the counters stay exact.
122    pub const REMAINING_TASKS_LIMIT: usize = 64;
123
124    pub(crate) fn push_remaining_task(&mut self, task: TaskSnapshot) {
125        if self.remaining_tasks.len() < Self::REMAINING_TASKS_LIMIT {
126            self.remaining_tasks.push(task);
127        } else {
128            self.remaining_tasks_omitted += 1;
129        }
130    }
131
132    /// Returns whether healthy.
133    pub fn is_healthy(&self) -> bool {
134        let mut pending = vec![self];
135        while let Some(report) = pending.pop() {
136            if report.leaked != 0
137                || report.failed != 0
138                || report.panicked != 0
139                || report.timed_out != 0
140                || report.blocking_still_running != 0
141            {
142                return false;
143            }
144            pending.extend(&report.children);
145        }
146        true
147    }
148
149    /// Returns the assert no task leak.
150    pub fn assert_no_task_leak(&self) -> Result<(), String> {
151        if self.is_healthy() {
152            Ok(())
153        } else {
154            Err(format!("shutdown report is unhealthy: {}", self.to_json()))
155        }
156    }
157
158    /// Executes log if unhealthy.
159    pub fn log_if_unhealthy(&self) {
160        if !self.is_healthy() {
161            tracing::warn!(report = %self.to_json(), "runtime shutdown report is unhealthy");
162        }
163    }
164
165    /// Converts this value to json.
166    pub fn to_json(&self) -> String {
167        serde_json::to_string_pretty(self).unwrap_or_else(|error| format!("{{\"serialization_error\":\"{}\"}}", error))
168    }
169
170    /// Executes merge blocking.
171    pub fn merge_blocking(&mut self, snapshot: crate::blocking::BlockingExecutorSnapshot) {
172        self.merge_blocking_tasks(snapshot.blocking_still_running, snapshot.tasks);
173    }
174
175    /// Adds blocking work observed through an executor's task table.
176    ///
177    /// The still-running annotation is added once, however many executors
178    /// contribute running work.
179    pub(crate) fn merge_blocking_tasks(&mut self, blocking_still_running: usize, tasks: Vec<BlockingTaskSnapshot>) {
180        const BLOCKING_STILL_RUNNING: &str =
181            "spawn_blocking tasks may continue after timeout; see blocking_still_running";
182
183        self.blocking_still_running += blocking_still_running;
184        self.blocking_tasks.extend(tasks);
185        if self.blocking_still_running > 0
186            && !self
187                .annotations
188                .iter()
189                .any(|annotation| annotation.message == BLOCKING_STILL_RUNNING)
190        {
191            self.annotations.push(ShutdownAnnotation::new(BLOCKING_STILL_RUNNING));
192        }
193    }
194}
195
196/// Represents task snapshot.
197#[derive(Debug, Clone, Serialize)]
198pub struct TaskSnapshot {
199    /// The id identifier.
200    pub id: TaskId,
201    /// The name value.
202    pub name: String,
203    /// The group identifier.
204    pub group_id: TaskGroupId,
205    /// The group name value.
206    pub group_name: String,
207    /// The kind value.
208    pub kind: TaskKind,
209    /// The state value.
210    pub state: TaskState,
211    /// The elapsed value.
212    #[serde(with = "duration_millis")]
213    pub elapsed: Duration,
214}
215
216/// Represents shutdown annotation.
217#[derive(Debug, Clone, Serialize)]
218pub struct ShutdownAnnotation {
219    /// The message value.
220    pub message: String,
221}
222
223impl ShutdownAnnotation {
224    /// Creates a new `ShutdownAnnotation`.
225    pub fn new(message: impl Into<String>) -> Self {
226        Self {
227            message: message.into(),
228        }
229    }
230}
231
232mod duration_millis {
233    use std::time::Duration;
234
235    use serde::Serializer;
236
237    pub fn serialize<S>(duration: &Duration, serializer: S) -> Result<S::Ok, S::Error>
238    where
239        S: Serializer,
240    {
241        serializer.serialize_u64(duration.as_millis() as u64)
242    }
243}
244
245#[cfg(test)]
246mod tests {
247    use super::*;
248
249    #[test]
250    fn explicit_component_failure_makes_the_report_unhealthy() {
251        let mut report = ShutdownReport::new("failed-component", Duration::ZERO);
252        report.failed = 1;
253
254        assert!(!report.is_healthy());
255        assert!(report.to_json().contains("\"failed\": 1"));
256    }
257
258    #[tokio::test]
259    async fn remaining_tasks_are_capped_and_the_omission_is_counted() {
260        let context = crate::RuntimeContext::from_current("remaining-task-limit");
261        let group_id = context.root_group().id();
262        let mut report = ShutdownReport::new("many-leaks", Duration::ZERO);
263        for id in 0..10_000 {
264            report.leaked += 1;
265            report.push_remaining_task(TaskSnapshot {
266                id: TaskId::new(group_id, id),
267                name: "leaked".to_owned(),
268                group_id,
269                group_name: "many-leaks".to_owned(),
270                kind: TaskKind::Worker,
271                state: TaskState::Leaked,
272                elapsed: Duration::ZERO,
273            });
274        }
275
276        assert_eq!(report.remaining_tasks.len(), ShutdownReport::REMAINING_TASKS_LIMIT);
277        assert_eq!(
278            report.remaining_tasks_omitted,
279            10_000 - ShutdownReport::REMAINING_TASKS_LIMIT
280        );
281        assert_eq!(report.leaked, 10_000);
282        let json = report.to_json();
283        assert!(json.contains("\"remaining_tasks_omitted\": 9936"), "{json}");
284        assert!(json.len() < 64 * 1024, "report JSON is {} bytes", json.len());
285    }
286}
rocketmq_runtime/shutdown_report.rs

rocketmq_runtime/
shutdown_report.rs
