tenant.openagents/omega

1use anyhow::{Context as _, Result};
2use client::{Client, telemetry::MINIDUMP_ENDPOINT};
3use feature_flags::FeatureFlagAppExt;
4use futures::{AsyncReadExt, TryStreamExt};
5use gpui::{App, AppContext, Entity, TaskExt};
6use http_client::{AsyncBody, HttpClient, Request};
7use project::{Project, worktree_store::WorktreeStoreDiagnostics};
8use proto::{CrashReport, GetCrashFilesResponse};
9use reqwest::{
10    Method,
11    multipart::{Form, Part},
12};
13use serde::Deserialize;
14use smol::stream::StreamExt;
15use std::{
16    collections::HashSet,
17    ffi::OsStr,
18    fs,
19    sync::Arc,
20    time::{Duration, Instant},
21};
22use sysinfo::{MemoryRefreshKind, ProcessRefreshKind, ProcessesToUpdate, RefreshKind, System};
23use util::ResultExt;
24use workspace::WorkspaceStore;
25
26mod hang_detection;
27
28pub fn init(client: Arc<Client>, workspace_store: Entity<WorkspaceStore>, cx: &mut App) {
29    hang_detection::start(client.clone(), cx);
30    start_memory_usage_logging(workspace_store, cx);
31
32    cx.on_flags_ready({
33        let client = client.clone();
34        move |flags_ready, cx| {
35            if flags_ready.is_staff {
36                let client = client.clone();
37                cx.background_spawn(async move {
38                    upload_build_timings(client).await.warn_on_err();
39                })
40                .detach();
41            }
42        }
43    })
44    .detach();
45
46    if client.telemetry().diagnostics_enabled() {
47        let client = client.clone();
48        cx.background_spawn(async move {
49            upload_previous_minidumps(client).await.warn_on_err();
50        })
51        .detach()
52    }
53
54    cx.observe_new(move |project: &mut Project, _, cx| {
55        let client = client.clone();
56
57        let Some(remote_client) = project.remote_client() else {
58            return;
59        };
60        remote_client.update(cx, |remote_client, cx| {
61            if !client.telemetry().diagnostics_enabled() {
62                return;
63            }
64            let request = remote_client
65                .proto_client()
66                .request(proto::GetCrashFiles {});
67            cx.background_spawn(async move {
68                let GetCrashFilesResponse { crashes } = request.await?;
69
70                let Some(endpoint) = MINIDUMP_ENDPOINT.as_ref() else {
71                    return Ok(());
72                };
73                for CrashReport {
74                    metadata,
75                    minidump_contents,
76                } in crashes
77                {
78                    if let Some(metadata) = serde_json::from_str(&metadata).log_err() {
79                        upload_minidump(client.clone(), endpoint, minidump_contents, &metadata)
80                            .await
81                            .log_err();
82                    }
83                }
84
85                anyhow::Ok(())
86            })
87            .detach_and_log_err(cx);
88        })
89    })
90    .detach();
91}
92
93const MEMORY_USAGE_POLL_INTERVAL: Duration = Duration::from_secs(30);
94const MEMORY_USAGE_HEARTBEAT_INTERVAL: Duration = Duration::from_secs(10 * 60);
95const MEMORY_USAGE_MINIMUM_LOGGED_DELTA: u64 = 64 * 1024 * 1024;
96
97/// Periodically logs this process' memory usage, so that gradual memory growth can be
98///
99/// Logs on a fixed heartbeat, and additionally whenever resident memory changed
100/// significantly since the last logged value, so that bursts of growth are timestamped
101/// against the surrounding log entries.
102fn start_memory_usage_logging(workspace_store: Entity<WorkspaceStore>, cx: &App) {
103    let (diagnostics_sender, mut diagnostics_receiver) = futures::channel::mpsc::unbounded();
104    cx.spawn(async move |cx| {
105        while diagnostics_receiver.next().await.is_some() {
106            cx.update(|cx| log_worktree_diagnostics(&workspace_store, cx));
107        }
108    })
109    .detach();
110
111    let executor = cx.background_executor().clone();
112    cx.background_spawn(async move {
113        let Some(pid) = sysinfo::get_current_pid().log_err() else {
114            return;
115        };
116        let refresh_kind = ProcessRefreshKind::nothing().with_memory();
117        let mut system = System::new();
118        let mut last_logged_resident: Option<u64> = None;
119        let mut last_logged_at = Instant::now();
120        loop {
121            let refreshed = system.refresh_processes_specifics(
122                ProcessesToUpdate::Some(&[pid]),
123                false,
124                refresh_kind,
125            );
126            if refreshed == 1
127                && let Some(process) = system.process(pid)
128            {
129                let resident = process.memory();
130                let significant_change = last_logged_resident.is_none_or(|last| {
131                    resident.abs_diff(last) >= (last / 10).max(MEMORY_USAGE_MINIMUM_LOGGED_DELTA)
132                });
133                if significant_change || last_logged_at.elapsed() >= MEMORY_USAGE_HEARTBEAT_INTERVAL
134                {
135                    const MIB: u64 = 1024 * 1024;
136                    let delta = match last_logged_resident {
137                        Some(last) => {
138                            format!(" ({:+} MiB)", (resident as i64 - last as i64) / MIB as i64)
139                        }
140                        None => String::new(),
141                    };
142                    log::info!(
143                        "memory usage: resident {} MiB{delta}, virtual {} MiB",
144                        resident / MIB,
145                        process.virtual_memory() / MIB,
146                    );
147                    if diagnostics_sender.unbounded_send(()).is_err() {
148                        return;
149                    }
150                    last_logged_resident = Some(resident);
151                    last_logged_at = Instant::now();
152                }
153            }
154            executor.timer(MEMORY_USAGE_POLL_INTERVAL).await;
155        }
156    })
157    .detach();
158}
159
160fn log_worktree_diagnostics(workspace_store: &Entity<WorkspaceStore>, cx: &App) {
161    let workspaces = workspace_store
162        .read(cx)
163        .workspaces()
164        .filter_map(|workspace| workspace.upgrade())
165        .collect::<Vec<_>>();
166    let mut worktree_store_ids = HashSet::new();
167    let mut store_count = 0;
168    let mut aggregate = WorktreeStoreDiagnostics::default();
169
170    for workspace in workspaces {
171        let project = workspace.read(cx).project().clone();
172        let worktree_store = project.read(cx).worktree_store();
173        if !worktree_store_ids.insert(worktree_store.entity_id()) {
174            continue;
175        }
176        store_count += 1;
177
178        let WorktreeStoreDiagnostics {
179            worktree_slots,
180            live_worktrees,
181            visible_worktrees,
182            strong_handles,
183            dead_weak_handles,
184            loading_worktrees,
185            total_entries,
186            visible_entries,
187            largest_worktree,
188        } = worktree_store.read(cx).diagnostics(cx);
189        aggregate.worktree_slots += worktree_slots;
190        aggregate.live_worktrees += live_worktrees;
191        aggregate.visible_worktrees += visible_worktrees;
192        aggregate.strong_handles += strong_handles;
193        aggregate.dead_weak_handles += dead_weak_handles;
194        aggregate.loading_worktrees += loading_worktrees;
195        aggregate.total_entries += total_entries;
196        aggregate.visible_entries += visible_entries;
197
198        if let Some(largest_worktree) = largest_worktree
199            && aggregate
200                .largest_worktree
201                .as_ref()
202                .is_none_or(|largest| largest_worktree.entries > largest.entries)
203        {
204            aggregate.largest_worktree = Some(largest_worktree);
205        }
206    }
207
208    let WorktreeStoreDiagnostics {
209        worktree_slots,
210        live_worktrees,
211        visible_worktrees,
212        strong_handles,
213        dead_weak_handles,
214        loading_worktrees,
215        total_entries,
216        visible_entries,
217        largest_worktree,
218    } = aggregate;
219    match largest_worktree {
220        Some(largest_worktree) => log::info!(
221            "worktree diagnostics: stores {store_count}, slots {worktree_slots}, live {live_worktrees}, visible {visible_worktrees}, strong {strong_handles}, dead weak {dead_weak_handles}, loading {loading_worktrees}, entries {total_entries}, visible entries {visible_entries}, largest {} ({} entries, {} visible)",
222            largest_worktree.path.display(),
223            largest_worktree.entries,
224            largest_worktree.visible_entries,
225        ),
226        None => log::info!(
227            "worktree diagnostics: stores {store_count}, slots {worktree_slots}, live {live_worktrees}, visible {visible_worktrees}, strong {strong_handles}, dead weak {dead_weak_handles}, loading {loading_worktrees}, entries {total_entries}, visible entries {visible_entries}, largest none",
228        ),
229    }
230}
231
232pub async fn upload_previous_minidumps(client: Arc<Client>) -> anyhow::Result<()> {
233    let Some(minidump_endpoint) = MINIDUMP_ENDPOINT.as_ref() else {
234        log::warn!("Minidump endpoint not set");
235        return Ok(());
236    };
237
238    let mut children = smol::fs::read_dir(paths::logs_dir()).await?;
239    while let Some(child) = children.next().await {
240        let child = child?;
241        let child_path = child.path();
242        if child_path.extension() != Some(OsStr::new("dmp")) {
243            continue;
244        }
245        let mut json_path = child_path.clone();
246        json_path.set_extension("json");
247        let Ok(metadata) = smol::fs::read(&json_path)
248            .await
249            .map_err(|e| anyhow::anyhow!(e))
250            .and_then(|data| serde_json::from_slice(&data).map_err(|e| anyhow::anyhow!(e)))
251        else {
252            continue;
253        };
254        if upload_minidump(
255            client.clone(),
256            minidump_endpoint,
257            smol::fs::read(&child_path)
258                .await
259                .context("Failed to read minidump")?,
260            &metadata,
261        )
262        .await
263        .log_err()
264        .is_some()
265        {
266            fs::remove_file(child_path).ok();
267            fs::remove_file(json_path).ok();
268        }
269    }
270    Ok(())
271}
272
273async fn upload_minidump(
274    client: Arc<Client>,
275    endpoint: &str,
276    minidump: Vec<u8>,
277    metadata: &crashes::CrashInfo,
278) -> Result<()> {
279    if metadata.init.commit_sha == "no sha" {
280        log::warn!("No commit sha set, skipping minidump upload");
281        return Ok(());
282    }
283    let mut form = Form::new()
284        .part(
285            "upload_file_minidump",
286            Part::bytes(minidump)
287                .file_name("minidump.dmp")
288                .mime_str("application/octet-stream")?,
289        )
290        .text(
291            "sentry[tags][channel]",
292            metadata.init.release_channel.clone(),
293        )
294        .text("sentry[tags][version]", metadata.init.zed_version.clone())
295        .text("sentry[tags][binary]", metadata.init.binary.clone())
296        .text("sentry[release]", metadata.init.commit_sha.clone())
297        .text("platform", "rust");
298    let mut panic_message = "".to_owned();
299    if let Some(panic_info) = metadata.panic.as_ref() {
300        panic_message = panic_info.message.clone();
301        form = form
302            .text("sentry[logentry][formatted]", panic_info.message.clone())
303            .text("span", panic_info.span.clone());
304    }
305    if let Some(minidump_error) = metadata.minidump_error.clone() {
306        form = form.text("minidump_error", minidump_error);
307    }
308    if let Some(abort_message) = metadata.abort_message.as_ref() {
309        // Sentry tag values are limited to 200 characters on a single line, so
310        // put a searchable prefix in the tag (which grouping rules also match
311        // on) and the full message in a context.
312        let tag: String = abort_message
313            .lines()
314            .next()
315            .unwrap_or_default()
316            .chars()
317            .take(200)
318            .collect();
319        form = form
320            .text("sentry[tags][abort_message]", tag)
321            .text("sentry[contexts][abort][message]", abort_message.clone());
322    }
323
324    if let Some(is_staff) = &metadata
325        .user_info
326        .as_ref()
327        .and_then(|user_info| user_info.is_staff)
328    {
329        form = form.text(
330            "sentry[user][is_staff]",
331            if *is_staff { "true" } else { "false" },
332        );
333    }
334
335    if let Some(metrics_id) = metadata
336        .user_info
337        .as_ref()
338        .and_then(|user_info| user_info.metrics_id.as_ref())
339    {
340        form = form.text("sentry[user][id]", metrics_id.clone());
341    } else if let Some(id) = client.telemetry().installation_id() {
342        form = form.text("sentry[user][id]", format!("installation-{}", id))
343    }
344
345    ::telemetry::event!(
346        "Minidump Uploaded",
347        panic_message = panic_message,
348        crashed_version = metadata.init.zed_version.clone(),
349        commit_sha = metadata.init.commit_sha.clone(),
350    );
351
352    let gpu_count = metadata.gpus.len();
353    for (index, gpu) in metadata.gpus.iter().cloned().enumerate() {
354        let system_specs::GpuInfo {
355            device_name,
356            device_pci_id,
357            vendor_name,
358            vendor_pci_id,
359            driver_version,
360            driver_name,
361        } = gpu;
362        let num = if gpu_count == 1 && metadata.active_gpu.is_none() {
363            String::new()
364        } else {
365            index.to_string()
366        };
367        let name = format!("gpu{num}");
368        let root = format!("sentry[contexts][{name}]");
369        form = form
370            .text(
371                format!("{root}[Description]"),
372                "A GPU found on the users system. May or may not be the GPU Omega is running on",
373            )
374            .text(format!("{root}[type]"), "gpu")
375            .text(format!("{root}[name]"), device_name.unwrap_or(name))
376            .text(format!("{root}[id]"), format!("{:#06x}", device_pci_id))
377            .text(
378                format!("{root}[vendor_id]"),
379                format!("{:#06x}", vendor_pci_id),
380            )
381            .text_if_some(format!("{root}[vendor_name]"), vendor_name)
382            .text_if_some(format!("{root}[driver_version]"), driver_version)
383            .text_if_some(format!("{root}[driver_name]"), driver_name);
384    }
385    if let Some(active_gpu) = metadata.active_gpu.clone() {
386        form = form
387            .text(
388                "sentry[contexts][Active_GPU][Description]",
389                "The GPU Omega is running on",
390            )
391            .text("sentry[contexts][Active_GPU][type]", "gpu")
392            .text("sentry[contexts][Active_GPU][name]", active_gpu.device_name)
393            .text(
394                "sentry[contexts][Active_GPU][driver_version]",
395                active_gpu.driver_info,
396            )
397            .text(
398                "sentry[contexts][Active_GPU][driver_name]",
399                active_gpu.driver_name,
400            )
401            .text(
402                "sentry[contexts][Active_GPU][is_software_emulated]",
403                active_gpu.is_software_emulated.to_string(),
404            );
405    }
406
407    // TODO: feature-flag-context, and more of device-context like screen resolution, available ram, device model, etc
408
409    let content_type = format!("multipart/form-data; boundary={}", form.boundary());
410    let mut body_bytes = Vec::new();
411    let mut stream = form
412        .into_stream()
413        .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))
414        .into_async_read();
415    stream.read_to_end(&mut body_bytes).await?;
416    let req = Request::builder()
417        .method(Method::POST)
418        .uri(endpoint)
419        .header("Content-Type", content_type)
420        .body(AsyncBody::from(body_bytes))?;
421    let mut response_text = String::new();
422    let mut response = client.http_client().send(req).await?;
423    response
424        .body_mut()
425        .read_to_string(&mut response_text)
426        .await?;
427    if !response.status().is_success() {
428        anyhow::bail!("failed to upload minidump: {response_text}");
429    }
430    log::info!("Uploaded minidump. event id: {response_text}");
431    Ok(())
432}
433
434#[derive(Debug, Deserialize)]
435struct BuildTiming {
436    started_at: chrono::DateTime<chrono::Utc>,
437    duration_ms: f32,
438    first_crate: String,
439    target: String,
440    blocked_ms: f32,
441    command: String,
442}
443
444// NOTE: this is a bit of a hack. We want to be able to have internal
445// metrics around build times, but we don't have an easy way to authenticate
446// users - except - we know internal users use Zed.
447// So, we have it upload the timings on their behalf, it'd be better to do
448// this more directly in ./script/cargo-timing-info.js.
449async fn upload_build_timings(_client: Arc<Client>) -> Result<()> {
450    let build_timings_dir = paths::data_dir().join("build_timings");
451
452    if !build_timings_dir.exists() {
453        return Ok(());
454    }
455
456    let cpu_count = std::thread::available_parallelism()
457        .map(|n| n.get())
458        .unwrap_or(1);
459    let system = System::new_with_specifics(
460        RefreshKind::nothing().with_memory(MemoryRefreshKind::everything()),
461    );
462    let ram_size_gb = (system.total_memory() as f64) / (1024.0 * 1024.0 * 1024.0);
463
464    let mut entries = smol::fs::read_dir(&build_timings_dir).await?;
465    while let Some(entry) = entries.next().await {
466        let entry = entry?;
467        let path = entry.path();
468
469        if path.extension() != Some(OsStr::new("json")) {
470            continue;
471        }
472
473        let contents = match smol::fs::read_to_string(&path).await {
474            Ok(contents) => contents,
475            Err(err) => {
476                log::warn!("Failed to read build timing file {:?}: {}", path, err);
477                continue;
478            }
479        };
480
481        let timing: BuildTiming = match serde_json::from_str(&contents) {
482            Ok(timing) => timing,
483            Err(err) => {
484                log::warn!("Failed to parse build timing file {:?}: {}", path, err);
485                continue;
486            }
487        };
488
489        telemetry::event!(
490            "Build Timing: Cargo Build",
491            started_at = timing.started_at.to_rfc3339(),
492            duration_ms = timing.duration_ms,
493            first_crate = timing.first_crate,
494            target = timing.target,
495            blocked_ms = timing.blocked_ms,
496            command = timing.command,
497            cpu_count = cpu_count,
498            ram_size_gb = ram_size_gb
499        );
500
501        if let Err(err) = smol::fs::remove_file(&path).await {
502            log::warn!("Failed to delete build timing file {:?}: {}", path, err);
503        }
504    }
505
506    Ok(())
507}
508
509trait FormExt {
510    fn text_if_some(
511        self,
512        label: impl Into<std::borrow::Cow<'static, str>>,
513        value: Option<impl Into<std::borrow::Cow<'static, str>>>,
514    ) -> Self;
515}
516
517impl FormExt for Form {
518    fn text_if_some(
519        self,
520        label: impl Into<std::borrow::Cow<'static, str>>,
521        value: Option<impl Into<std::borrow::Cow<'static, str>>>,
522    ) -> Self {
523        match value {
524            Some(value) => self.text(label.into(), value.into()),
525            None => self,
526        }
527    }
528}
529