1mod event_coalescer;
2
3use crate::{ChannelId, TelemetrySettings};
4use chrono::{DateTime, Utc};
5use clock::SystemClock;
6use futures::Future;
7use gpui::{AppContext, AppMetadata, BackgroundExecutor, Task};
8use once_cell::sync::Lazy;
9use parking_lot::Mutex;
10use release_channel::ReleaseChannel;
11use settings::{Settings, SettingsStore};
12use sha2::{Digest, Sha256};
13use std::io::Write;
14use std::{env, mem, path::PathBuf, sync::Arc, time::Duration};
15use sysinfo::{CpuRefreshKind, Pid, ProcessRefreshKind, RefreshKind, System};
16use telemetry_events::{
17 ActionEvent, AppEvent, AssistantEvent, AssistantKind, CallEvent, CopilotEvent, CpuEvent,
18 EditEvent, EditorEvent, Event, EventRequestBody, EventWrapper, ExtensionEvent, MemoryEvent,
19 SettingEvent,
20};
21use tempfile::NamedTempFile;
22use util::http::{self, HttpClient, HttpClientWithUrl, Method};
23#[cfg(not(debug_assertions))]
24use util::ResultExt;
25use util::TryFutureExt;
26
27use self::event_coalescer::EventCoalescer;
28
29pub struct Telemetry {
30 clock: Arc<dyn SystemClock>,
31 http_client: Arc<HttpClientWithUrl>,
32 executor: BackgroundExecutor,
33 state: Arc<Mutex<TelemetryState>>,
34}
35
36struct TelemetryState {
37 settings: TelemetrySettings,
38 metrics_id: Option<Arc<str>>, // Per logged-in user
39 installation_id: Option<Arc<str>>, // Per app installation (different for dev, nightly, preview, and stable)
40 session_id: Option<String>, // Per app launch
41 release_channel: Option<&'static str>,
42 app_metadata: AppMetadata,
43 architecture: &'static str,
44 events_queue: Vec<EventWrapper>,
45 flush_events_task: Option<Task<()>>,
46 log_file: Option<NamedTempFile>,
47 is_staff: Option<bool>,
48 first_event_date_time: Option<DateTime<Utc>>,
49 event_coalescer: EventCoalescer,
50 max_queue_size: usize,
51}
52
53#[cfg(debug_assertions)]
54const MAX_QUEUE_LEN: usize = 5;
55
56#[cfg(not(debug_assertions))]
57const MAX_QUEUE_LEN: usize = 50;
58
59#[cfg(debug_assertions)]
60const FLUSH_INTERVAL: Duration = Duration::from_secs(1);
61
62#[cfg(not(debug_assertions))]
63const FLUSH_INTERVAL: Duration = Duration::from_secs(60 * 5);
64static ZED_CLIENT_CHECKSUM_SEED: Lazy<Option<Vec<u8>>> = Lazy::new(|| {
65 option_env!("ZED_CLIENT_CHECKSUM_SEED")
66 .map(|s| s.as_bytes().into())
67 .or_else(|| {
68 env::var("ZED_CLIENT_CHECKSUM_SEED")
69 .ok()
70 .map(|s| s.as_bytes().into())
71 })
72});
73
74impl Telemetry {
75 pub fn new(
76 clock: Arc<dyn SystemClock>,
77 client: Arc<HttpClientWithUrl>,
78 cx: &mut AppContext,
79 ) -> Arc<Self> {
80 let release_channel =
81 ReleaseChannel::try_global(cx).map(|release_channel| release_channel.display_name());
82
83 TelemetrySettings::register(cx);
84
85 let state = Arc::new(Mutex::new(TelemetryState {
86 settings: *TelemetrySettings::get_global(cx),
87 app_metadata: cx.app_metadata(),
88 architecture: env::consts::ARCH,
89 release_channel,
90 installation_id: None,
91 metrics_id: None,
92 session_id: None,
93 events_queue: Vec::new(),
94 flush_events_task: None,
95 log_file: None,
96 is_staff: None,
97 first_event_date_time: None,
98 event_coalescer: EventCoalescer::new(clock.clone()),
99 max_queue_size: MAX_QUEUE_LEN,
100 }));
101
102 #[cfg(not(debug_assertions))]
103 cx.background_executor()
104 .spawn({
105 let state = state.clone();
106 async move {
107 if let Some(tempfile) =
108 NamedTempFile::new_in(util::paths::CONFIG_DIR.as_path()).log_err()
109 {
110 state.lock().log_file = Some(tempfile);
111 }
112 }
113 })
114 .detach();
115
116 cx.observe_global::<SettingsStore>({
117 let state = state.clone();
118
119 move |cx| {
120 let mut state = state.lock();
121 state.settings = *TelemetrySettings::get_global(cx);
122 }
123 })
124 .detach();
125
126 // TODO: Replace all hardware stuff with nested SystemSpecs json
127 let this = Arc::new(Self {
128 clock,
129 http_client: client,
130 executor: cx.background_executor().clone(),
131 state,
132 });
133
134 // We should only ever have one instance of Telemetry, leak the subscription to keep it alive
135 // rather than store in TelemetryState, complicating spawn as subscriptions are not Send
136 std::mem::forget(cx.on_app_quit({
137 let this = this.clone();
138 move |_| this.shutdown_telemetry()
139 }));
140
141 this
142 }
143
144 #[cfg(any(test, feature = "test-support"))]
145 fn shutdown_telemetry(self: &Arc<Self>) -> impl Future<Output = ()> {
146 Task::ready(())
147 }
148
149 // Skip calling this function in tests.
150 // TestAppContext ends up calling this function on shutdown and it panics when trying to find the TelemetrySettings
151 #[cfg(not(any(test, feature = "test-support")))]
152 fn shutdown_telemetry(self: &Arc<Self>) -> impl Future<Output = ()> {
153 self.report_app_event("close".to_string());
154 // TODO: close final edit period and make sure it's sent
155 Task::ready(())
156 }
157
158 pub fn log_file_path(&self) -> Option<PathBuf> {
159 Some(self.state.lock().log_file.as_ref()?.path().to_path_buf())
160 }
161
162 pub fn start(
163 self: &Arc<Self>,
164 installation_id: Option<String>,
165 session_id: String,
166 cx: &mut AppContext,
167 ) {
168 let mut state = self.state.lock();
169 state.installation_id = installation_id.map(|id| id.into());
170 state.session_id = Some(session_id);
171 drop(state);
172
173 let this = self.clone();
174 cx.background_executor()
175 .spawn(async move {
176 let mut system = System::new_with_specifics(
177 RefreshKind::new().with_cpu(CpuRefreshKind::everything()),
178 );
179
180 let refresh_kind = ProcessRefreshKind::new().with_cpu().with_memory();
181 let current_process = Pid::from_u32(std::process::id());
182 system.refresh_process_specifics(current_process, refresh_kind);
183
184 // Waiting some amount of time before the first query is important to get a reasonable value
185 // https://docs.rs/sysinfo/0.29.10/sysinfo/trait.ProcessExt.html#tymethod.cpu_usage
186 const DURATION_BETWEEN_SYSTEM_EVENTS: Duration = Duration::from_secs(4 * 60);
187
188 loop {
189 smol::Timer::after(DURATION_BETWEEN_SYSTEM_EVENTS).await;
190
191 let current_process = Pid::from_u32(std::process::id());
192 system.refresh_process_specifics(current_process, refresh_kind);
193 let Some(process) = system.process(current_process) else {
194 log::error!(
195 "Failed to find own process {current_process:?} in system process table"
196 );
197 // TODO: Fire an error telemetry event
198 return;
199 };
200
201 this.report_memory_event(process.memory(), process.virtual_memory());
202 this.report_cpu_event(process.cpu_usage(), system.cpus().len() as u32);
203 }
204 })
205 .detach();
206 }
207
208 pub fn set_authenticated_user_info(
209 self: &Arc<Self>,
210 metrics_id: Option<String>,
211 is_staff: bool,
212 ) {
213 let mut state = self.state.lock();
214
215 if !state.settings.metrics {
216 return;
217 }
218
219 let metrics_id: Option<Arc<str>> = metrics_id.map(|id| id.into());
220 state.metrics_id.clone_from(&metrics_id);
221 state.is_staff = Some(is_staff);
222 drop(state);
223 }
224
225 pub fn report_editor_event(
226 self: &Arc<Self>,
227 file_extension: Option<String>,
228 vim_mode: bool,
229 operation: &'static str,
230 copilot_enabled: bool,
231 copilot_enabled_for_language: bool,
232 ) {
233 let event = Event::Editor(EditorEvent {
234 file_extension,
235 vim_mode,
236 operation: operation.into(),
237 copilot_enabled,
238 copilot_enabled_for_language,
239 });
240
241 self.report_event(event)
242 }
243
244 pub fn report_copilot_event(
245 self: &Arc<Self>,
246 suggestion_id: Option<String>,
247 suggestion_accepted: bool,
248 file_extension: Option<String>,
249 ) {
250 let event = Event::Copilot(CopilotEvent {
251 suggestion_id,
252 suggestion_accepted,
253 file_extension,
254 });
255
256 self.report_event(event)
257 }
258
259 pub fn report_assistant_event(
260 self: &Arc<Self>,
261 conversation_id: Option<String>,
262 kind: AssistantKind,
263 model: String,
264 ) {
265 let event = Event::Assistant(AssistantEvent {
266 conversation_id,
267 kind,
268 model: model.to_string(),
269 });
270
271 self.report_event(event)
272 }
273
274 pub fn report_call_event(
275 self: &Arc<Self>,
276 operation: &'static str,
277 room_id: Option<u64>,
278 channel_id: Option<ChannelId>,
279 ) {
280 let event = Event::Call(CallEvent {
281 operation: operation.to_string(),
282 room_id,
283 channel_id: channel_id.map(|cid| cid.0),
284 });
285
286 self.report_event(event)
287 }
288
289 pub fn report_cpu_event(self: &Arc<Self>, usage_as_percentage: f32, core_count: u32) {
290 let event = Event::Cpu(CpuEvent {
291 usage_as_percentage,
292 core_count,
293 });
294
295 self.report_event(event)
296 }
297
298 pub fn report_memory_event(
299 self: &Arc<Self>,
300 memory_in_bytes: u64,
301 virtual_memory_in_bytes: u64,
302 ) {
303 let event = Event::Memory(MemoryEvent {
304 memory_in_bytes,
305 virtual_memory_in_bytes,
306 });
307
308 self.report_event(event)
309 }
310
311 pub fn report_app_event(self: &Arc<Self>, operation: String) -> Event {
312 let event = Event::App(AppEvent { operation });
313
314 self.report_event(event.clone());
315
316 event
317 }
318
319 pub fn report_setting_event(self: &Arc<Self>, setting: &'static str, value: String) {
320 let event = Event::Setting(SettingEvent {
321 setting: setting.to_string(),
322 value,
323 });
324
325 self.report_event(event)
326 }
327
328 pub fn report_extension_event(self: &Arc<Self>, extension_id: Arc<str>, version: Arc<str>) {
329 self.report_event(Event::Extension(ExtensionEvent {
330 extension_id,
331 version,
332 }))
333 }
334
335 pub fn log_edit_event(self: &Arc<Self>, environment: &'static str) {
336 let mut state = self.state.lock();
337 let period_data = state.event_coalescer.log_event(environment);
338 drop(state);
339
340 if let Some((start, end, environment)) = period_data {
341 let event = Event::Edit(EditEvent {
342 duration: end.timestamp_millis() - start.timestamp_millis(),
343 environment: environment.to_string(),
344 });
345
346 self.report_event(event);
347 }
348 }
349
350 pub fn report_action_event(self: &Arc<Self>, source: &'static str, action: String) {
351 let event = Event::Action(ActionEvent {
352 source: source.to_string(),
353 action,
354 });
355
356 self.report_event(event)
357 }
358
359 fn report_event(self: &Arc<Self>, event: Event) {
360 let mut state = self.state.lock();
361
362 if !state.settings.metrics {
363 return;
364 }
365
366 if state.flush_events_task.is_none() {
367 let this = self.clone();
368 let executor = self.executor.clone();
369 state.flush_events_task = Some(self.executor.spawn(async move {
370 executor.timer(FLUSH_INTERVAL).await;
371 this.flush_events();
372 }));
373 }
374
375 let date_time = self.clock.utc_now();
376
377 let milliseconds_since_first_event = match state.first_event_date_time {
378 Some(first_event_date_time) => {
379 date_time.timestamp_millis() - first_event_date_time.timestamp_millis()
380 }
381 None => {
382 state.first_event_date_time = Some(date_time);
383 0
384 }
385 };
386
387 let signed_in = state.metrics_id.is_some();
388 state.events_queue.push(EventWrapper {
389 signed_in,
390 milliseconds_since_first_event,
391 event,
392 });
393
394 if state.installation_id.is_some() && state.events_queue.len() >= state.max_queue_size {
395 drop(state);
396 self.flush_events();
397 }
398 }
399
400 pub fn metrics_id(self: &Arc<Self>) -> Option<Arc<str>> {
401 self.state.lock().metrics_id.clone()
402 }
403
404 pub fn installation_id(self: &Arc<Self>) -> Option<Arc<str>> {
405 self.state.lock().installation_id.clone()
406 }
407
408 pub fn is_staff(self: &Arc<Self>) -> Option<bool> {
409 self.state.lock().is_staff
410 }
411
412 pub fn flush_events(self: &Arc<Self>) {
413 let mut state = self.state.lock();
414 state.first_event_date_time = None;
415 let mut events = mem::take(&mut state.events_queue);
416 state.flush_events_task.take();
417 drop(state);
418 if events.is_empty() {
419 return;
420 }
421
422 if ZED_CLIENT_CHECKSUM_SEED.is_none() {
423 return;
424 };
425
426 let this = self.clone();
427 self.executor
428 .spawn(
429 async move {
430 let mut json_bytes = Vec::new();
431
432 if let Some(file) = &mut this.state.lock().log_file {
433 let file = file.as_file_mut();
434 for event in &mut events {
435 json_bytes.clear();
436 serde_json::to_writer(&mut json_bytes, event)?;
437 file.write_all(&json_bytes)?;
438 file.write_all(b"\n")?;
439 }
440 }
441
442 {
443 let state = this.state.lock();
444 let request_body = EventRequestBody {
445 installation_id: state.installation_id.as_deref().map(Into::into),
446 session_id: state.session_id.clone(),
447 is_staff: state.is_staff,
448 app_version: state
449 .app_metadata
450 .app_version
451 .unwrap_or_default()
452 .to_string(),
453 os_name: state.app_metadata.os_name.to_string(),
454 os_version: state
455 .app_metadata
456 .os_version
457 .map(|version| version.to_string()),
458 architecture: state.architecture.to_string(),
459
460 release_channel: state.release_channel.map(Into::into),
461 events,
462 };
463 json_bytes.clear();
464 serde_json::to_writer(&mut json_bytes, &request_body)?;
465 }
466
467 let Some(checksum) = calculate_json_checksum(&json_bytes) else {
468 return Ok(());
469 };
470
471 let request = http::Request::builder()
472 .method(Method::POST)
473 .uri(
474 this.http_client
475 .build_zed_api_url("/telemetry/events", &[])?
476 .as_ref(),
477 )
478 .header("Content-Type", "text/plain")
479 .header("x-zed-checksum", checksum)
480 .body(json_bytes.into());
481
482 let response = this.http_client.send(request?).await?;
483 if response.status() != 200 {
484 log::error!("Failed to send events: HTTP {:?}", response.status());
485 }
486 anyhow::Ok(())
487 }
488 .log_err(),
489 )
490 .detach();
491 }
492}
493
494#[cfg(test)]
495mod tests {
496 use super::*;
497 use chrono::TimeZone;
498 use clock::FakeSystemClock;
499 use gpui::TestAppContext;
500 use util::http::FakeHttpClient;
501
502 #[gpui::test]
503 fn test_telemetry_flush_on_max_queue_size(cx: &mut TestAppContext) {
504 init_test(cx);
505 let clock = Arc::new(FakeSystemClock::new(
506 Utc.with_ymd_and_hms(1990, 4, 12, 12, 0, 0).unwrap(),
507 ));
508 let http = FakeHttpClient::with_200_response();
509 let installation_id = Some("installation_id".to_string());
510 let session_id = "session_id".to_string();
511
512 cx.update(|cx| {
513 let telemetry = Telemetry::new(clock.clone(), http, cx);
514
515 telemetry.state.lock().max_queue_size = 4;
516 telemetry.start(installation_id, session_id, cx);
517
518 assert!(is_empty_state(&telemetry));
519
520 let first_date_time = clock.utc_now();
521 let operation = "test".to_string();
522
523 let event = telemetry.report_app_event(operation.clone());
524 assert_eq!(
525 event,
526 Event::App(AppEvent {
527 operation: operation.clone(),
528 })
529 );
530 assert_eq!(telemetry.state.lock().events_queue.len(), 1);
531 assert!(telemetry.state.lock().flush_events_task.is_some());
532 assert_eq!(
533 telemetry.state.lock().first_event_date_time,
534 Some(first_date_time)
535 );
536
537 clock.advance(chrono::Duration::milliseconds(100));
538
539 let event = telemetry.report_app_event(operation.clone());
540 assert_eq!(
541 event,
542 Event::App(AppEvent {
543 operation: operation.clone(),
544 })
545 );
546 assert_eq!(telemetry.state.lock().events_queue.len(), 2);
547 assert!(telemetry.state.lock().flush_events_task.is_some());
548 assert_eq!(
549 telemetry.state.lock().first_event_date_time,
550 Some(first_date_time)
551 );
552
553 clock.advance(chrono::Duration::milliseconds(100));
554
555 let event = telemetry.report_app_event(operation.clone());
556 assert_eq!(
557 event,
558 Event::App(AppEvent {
559 operation: operation.clone(),
560 })
561 );
562 assert_eq!(telemetry.state.lock().events_queue.len(), 3);
563 assert!(telemetry.state.lock().flush_events_task.is_some());
564 assert_eq!(
565 telemetry.state.lock().first_event_date_time,
566 Some(first_date_time)
567 );
568
569 clock.advance(chrono::Duration::milliseconds(100));
570
571 // Adding a 4th event should cause a flush
572 let event = telemetry.report_app_event(operation.clone());
573 assert_eq!(
574 event,
575 Event::App(AppEvent {
576 operation: operation.clone(),
577 })
578 );
579
580 assert!(is_empty_state(&telemetry));
581 });
582 }
583
584 #[gpui::test]
585 async fn test_telemetry_flush_on_flush_interval(
586 executor: BackgroundExecutor,
587 cx: &mut TestAppContext,
588 ) {
589 init_test(cx);
590 let clock = Arc::new(FakeSystemClock::new(
591 Utc.with_ymd_and_hms(1990, 4, 12, 12, 0, 0).unwrap(),
592 ));
593 let http = FakeHttpClient::with_200_response();
594 let installation_id = Some("installation_id".to_string());
595 let session_id = "session_id".to_string();
596
597 cx.update(|cx| {
598 let telemetry = Telemetry::new(clock.clone(), http, cx);
599 telemetry.state.lock().max_queue_size = 4;
600 telemetry.start(installation_id, session_id, cx);
601
602 assert!(is_empty_state(&telemetry));
603
604 let first_date_time = clock.utc_now();
605 let operation = "test".to_string();
606
607 let event = telemetry.report_app_event(operation.clone());
608 assert_eq!(
609 event,
610 Event::App(AppEvent {
611 operation: operation.clone(),
612 })
613 );
614 assert_eq!(telemetry.state.lock().events_queue.len(), 1);
615 assert!(telemetry.state.lock().flush_events_task.is_some());
616 assert_eq!(
617 telemetry.state.lock().first_event_date_time,
618 Some(first_date_time)
619 );
620
621 let duration = Duration::from_millis(1);
622
623 // Test 1 millisecond before the flush interval limit is met
624 executor.advance_clock(FLUSH_INTERVAL - duration);
625
626 assert!(!is_empty_state(&telemetry));
627
628 // Test the exact moment the flush interval limit is met
629 executor.advance_clock(duration);
630
631 assert!(is_empty_state(&telemetry));
632 });
633 }
634
635 // TODO:
636 // Test settings
637 // Update FakeHTTPClient to keep track of the number of requests and assert on it
638
639 fn init_test(cx: &mut TestAppContext) {
640 cx.update(|cx| {
641 let settings_store = SettingsStore::test(cx);
642 cx.set_global(settings_store);
643 });
644 }
645
646 fn is_empty_state(telemetry: &Telemetry) -> bool {
647 telemetry.state.lock().events_queue.is_empty()
648 && telemetry.state.lock().flush_events_task.is_none()
649 && telemetry.state.lock().first_event_date_time.is_none()
650 }
651}
652
653pub fn calculate_json_checksum(json: &impl AsRef<[u8]>) -> Option<String> {
654 let Some(checksum_seed) = &*ZED_CLIENT_CHECKSUM_SEED else {
655 return None;
656 };
657
658 let mut summer = Sha256::new();
659 summer.update(checksum_seed);
660 summer.update(&json);
661 summer.update(checksum_seed);
662 let mut checksum = String::new();
663 for byte in summer.finalize().as_slice() {
664 use std::fmt::Write;
665 write!(&mut checksum, "{:02x}", byte).unwrap();
666 }
667
668 Some(checksum)
669}