libdd_crashtracker/crash_info/
mod.rs

1// Copyright 2024-Present Datadog, Inc. https://www.datadoghq.com/
2// SPDX-License-Identifier: Apache-2.0
3
4mod builder;
5mod error_data;
6mod errors_intake;
7mod experimental;
8mod metadata;
9mod os_info;
10mod proc_info;
11mod sig_info;
12mod spans;
13mod stacktrace;
14mod telemetry;
15mod test_utils;
16mod unknown_value;
17
18pub use builder::*;
19pub use error_data::*;
20pub use errors_intake::*;
21pub use experimental::*;
22use libdd_common::Endpoint;
23pub use metadata::Metadata;
24pub use os_info::*;
25pub use proc_info::*;
26pub use sig_info::*;
27pub use spans::*;
28pub use stacktrace::*;
29pub use telemetry::*;
30
31use anyhow::Context;
32use schemars::JsonSchema;
33use serde::{Deserialize, Serialize};
34use std::{collections::HashMap, fs::File, path::Path};
35
36pub fn build_crash_ping_message(sig_info: &SigInfo) -> String {
37    format!(
38        "Crashtracker crash ping: crash processing started - Process terminated by signal {:?}",
39        sig_info.si_signo_human_readable
40    )
41}
42
43#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
44pub struct CrashInfo {
45    #[serde(default, skip_serializing_if = "HashMap::is_empty")]
46    pub counters: HashMap<String, i64>,
47    pub data_schema_version: String,
48    pub error: ErrorData,
49    #[serde(default, skip_serializing_if = "Option::is_none")]
50    pub experimental: Option<Experimental>,
51    #[serde(default, skip_serializing_if = "HashMap::is_empty")]
52    pub files: HashMap<String, Vec<String>>,
53    #[serde(default, skip_serializing_if = "Option::is_none")]
54    pub fingerprint: Option<String>,
55    pub incomplete: bool,
56    #[serde(default, skip_serializing_if = "Vec::is_empty")]
57    pub log_messages: Vec<String>,
58    pub metadata: Metadata,
59    pub os_info: OsInfo,
60    #[serde(default, skip_serializing_if = "Option::is_none")]
61    pub proc_info: Option<ProcInfo>, //TODO, update the schema
62    #[serde(default, skip_serializing_if = "Option::is_none")]
63    pub sig_info: Option<SigInfo>, //TODO, update the schema
64    #[serde(default, skip_serializing_if = "Vec::is_empty")]
65    pub span_ids: Vec<Span>,
66    pub timestamp: String,
67    #[serde(default, skip_serializing_if = "Vec::is_empty")]
68    pub trace_ids: Vec<Span>,
69    pub uuid: String,
70}
71
72impl CrashInfo {
73    pub fn current_schema_version() -> String {
74        "1.4".to_string()
75    }
76
77    pub fn demangle_names(&mut self) -> anyhow::Result<()> {
78        self.error.demangle_names()
79    }
80}
81
82#[cfg(unix)]
83impl CrashInfo {
84    pub fn normalize_ips(&mut self, pid: u32) -> anyhow::Result<()> {
85        self.error.normalize_ips(pid)
86    }
87
88    pub fn resolve_names(&mut self, pid: u32) -> anyhow::Result<()> {
89        self.error.resolve_names(pid)
90    }
91
92    pub fn enrich_callstacks(&mut self, pid: u32) -> anyhow::Result<()> {
93        let src = ErrorData::create_symbolizer_source(pid);
94        let normalizer = ErrorData::create_normalizer();
95        let mut symbolizer = blazesym::symbolize::Symbolizer::new();
96        let mut elf_resolvers = CachedElfResolvers::new(&mut symbolizer);
97
98        // We must call ips normalization first.
99        // This will allow us to feed the symbolizer with the ELF Resolvers
100        let rval1 = self
101            .error
102            .normalize_ips_impl(pid, &normalizer, &mut elf_resolvers);
103        let rval2 = self.error.resolve_names_impl(&symbolizer, &src);
104        anyhow::ensure!(
105            rval1.is_ok() && rval2.is_ok(),
106            "normalize_ips: {rval1:?}\tresolve_names: {rval2:?}"
107        );
108        Ok(())
109    }
110}
111
112impl CrashInfo {
113    /// Emit the CrashInfo as structured json in file `path`.
114    pub fn to_file(&self, path: &Path) -> anyhow::Result<()> {
115        let file = File::options()
116            .create(true)
117            .append(true)
118            .open(path)
119            .with_context(|| format!("Failed to create {}", path.display()))?;
120        serde_json::to_writer_pretty(file, self)
121            .with_context(|| format!("Failed to write json to {}", path.display()))?;
122        Ok(())
123    }
124
125    pub fn upload_to_endpoint(&self, endpoint: &Option<Endpoint>) -> anyhow::Result<()> {
126        let rt = tokio::runtime::Builder::new_current_thread()
127            .enable_all()
128            .build()?;
129
130        rt.block_on(async { self.async_upload_to_endpoint(endpoint).await })
131    }
132
133    pub async fn async_upload_to_endpoint(
134        &self,
135        endpoint: &Option<Endpoint>,
136    ) -> anyhow::Result<()> {
137        // If we're debugging to a file, dump the actual crashinfo into a json
138        if let Some(endpoint) = endpoint {
139            if Some("file") == endpoint.url.scheme_str() {
140                let path = libdd_common::decode_uri_path_in_authority(&endpoint.url)
141                    .context("crash output file path was not correctly formatted")?;
142                self.to_file(&path)?;
143            }
144        }
145
146        let telemetry_future = self.upload_to_telemetry(endpoint);
147        let errors_intake_future = self.upload_to_errors_intake(endpoint);
148        let (_telemetry_result, _errors_intake_result) =
149            tokio::join!(telemetry_future, errors_intake_future);
150        Ok(())
151    }
152
153    async fn upload_to_telemetry(&self, endpoint: &Option<Endpoint>) -> anyhow::Result<()> {
154        let uploader = TelemetryCrashUploader::new(&self.metadata, endpoint)?;
155        uploader.upload_to_telemetry(self).await?;
156        Ok(())
157    }
158
159    async fn upload_to_errors_intake(&self, endpoint: &Option<Endpoint>) -> anyhow::Result<()> {
160        let uploader = ErrorsIntakeUploader::new(endpoint)?;
161        if uploader.is_enabled() {
162            uploader.upload_to_errors_intake(self).await?;
163        }
164        Ok(())
165    }
166}
167
168#[cfg(test)]
169mod tests {
170    use schemars::schema::RootSchema;
171    use std::fs;
172
173    use super::*;
174    #[test]
175    fn test_schema_matches_rfc() {
176        let rfc_schema_filename = concat!(
177            env!("CARGO_MANIFEST_DIR"),
178            "/../docs/RFCs/artifacts/0011-crashtracker-unified-runtime-stack-schema.json"
179        );
180        let schema = schemars::schema_for!(CrashInfo);
181        let schema_json = serde_json::to_string_pretty(&schema).expect("Schema to serialize");
182
183        // Try to load the existing RFC schema
184        let path = Path::new(rfc_schema_filename);
185        let existing_schema_json = fs::read_to_string(path);
186
187        match existing_schema_json {
188            Ok(rfc_schema_json) => {
189                let rfc_schema: RootSchema =
190                    serde_json::from_str(&rfc_schema_json).expect("RFC schema to be valid JSON");
191                if rfc_schema != schema {
192                    eprintln!(
193                        "Schema mismatch — updating file at {} with the latest schema.",
194                        rfc_schema_filename
195                    );
196                    fs::write(path, &schema_json).expect("Failed to write updated schema");
197                    panic!("Schema updated. Please commit the new file.");
198                }
199            }
200            Err(_) => {
201                eprintln!(
202                    "RFC schema file not found — creating new schema file at {}",
203                    rfc_schema_filename
204                );
205                fs::create_dir_all(path.parent().unwrap())
206                    .expect("Failed to create parent directories");
207                fs::write(path, &schema_json).expect("Failed to write schema file");
208                panic!("New schema file created. Please commit it.");
209            }
210        }
211    }
212
213    impl test_utils::TestInstance for CrashInfo {
214        fn test_instance(seed: u64) -> Self {
215            let mut counters = HashMap::new();
216            counters.insert("collecting_sample".to_owned(), 1);
217            counters.insert("not_profiling".to_owned(), 0);
218
219            let span_ids = vec![
220                Span {
221                    id: "42".to_string(),
222                    thread_name: Some("thread1".to_string()),
223                },
224                Span {
225                    id: "24".to_string(),
226                    thread_name: Some("thread2".to_string()),
227                },
228            ];
229
230            let trace_ids = vec![
231                Span {
232                    id: "345".to_string(),
233                    thread_name: Some("thread111".to_string()),
234                },
235                Span {
236                    id: "666".to_string(),
237                    thread_name: Some("thread222".to_string()),
238                },
239            ];
240
241            Self {
242                counters,
243                data_schema_version: CrashInfo::current_schema_version(),
244                error: ErrorData::test_instance(seed),
245                experimental: None,
246                files: HashMap::new(),
247                fingerprint: None,
248                incomplete: true,
249                log_messages: vec![],
250                metadata: Metadata::test_instance(seed),
251                os_info: ::os_info::Info::unknown().into(),
252                proc_info: Some(ProcInfo::test_instance(seed)),
253                sig_info: Some(SigInfo::test_instance(seed)),
254                span_ids,
255                timestamp: chrono::DateTime::from_timestamp(1568898000 /* Datadog IPO */, 0)
256                    .unwrap()
257                    .to_string(),
258                trace_ids,
259                uuid: uuid::uuid!("1d6b97cb-968c-40c9-af6e-e4b4d71e8781").to_string(),
260            }
261        }
262    }
263}
libdd_crashtracker/crash_info/mod.rs

libdd_crashtracker/crash_info/
mod.rs
