From 01eab39fd3232f0f086100d126214c6cfac3e7c5 Mon Sep 17 00:00:00 2001 From: Karthik Nadig Date: Wed, 29 Jul 2026 08:40:57 -0700 Subject: [PATCH] feat: report refresh phase progress telemetry (Fixes #479) Expose a refresh operation ID in results and emit privacy-safe phase and per-locator timing notifications for timeout diagnosis. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- crates/pet-core/src/telemetry/mod.rs | 5 + .../src/telemetry/refresh_progress.rs | 35 +++ crates/pet-reporter/src/jsonrpc.rs | 37 +++ crates/pet/src/find.rs | 277 ++++++++++++++++-- crates/pet/src/jsonrpc.rs | 11 +- crates/pet/src/lib.rs | 5 +- crates/pet/tests/ci_homebrew_container.rs | 1 + crates/pet/tests/ci_jupyter_container.rs | 1 + crates/pet/tests/ci_poetry.rs | 2 + crates/pet/tests/ci_test.rs | 5 + crates/pet/tests/jsonrpc_client.rs | 36 +++ crates/pet/tests/jsonrpc_server_test.rs | 54 +++- docs/JSONRPC.md | 34 +++ 13 files changed, 463 insertions(+), 40 deletions(-) create mode 100644 crates/pet-core/src/telemetry/refresh_progress.rs diff --git a/crates/pet-core/src/telemetry/mod.rs b/crates/pet-core/src/telemetry/mod.rs index b158c4eb..ffabe6df 100644 --- a/crates/pet-core/src/telemetry/mod.rs +++ b/crates/pet-core/src/telemetry/mod.rs @@ -5,12 +5,14 @@ use inaccurate_python_info::InaccuratePythonEnvironmentInfo; use missing_conda_info::MissingCondaEnvironments; use missing_poetry_info::MissingPoetryEnvironments; use refresh_performance::RefreshPerformance; +use refresh_progress::RefreshProgress; use serde::{Deserialize, Serialize}; pub mod inaccurate_python_info; pub mod missing_conda_info; pub mod missing_poetry_info; pub mod refresh_performance; +pub mod refresh_progress; pub type NumberOfCustomSearchPaths = u32; @@ -38,6 +40,8 @@ pub enum TelemetryEvent { MissingPoetryEnvironments(MissingPoetryEnvironments), /// Telemetry with metrics for finding all environments as a result of refresh. RefreshPerformance(RefreshPerformance), + /// Progress through a refresh operation, including per-locator timing. + RefreshProgress(RefreshProgress), } pub fn get_telemetry_event_name(event: &TelemetryEvent) -> &'static str { @@ -57,5 +61,6 @@ pub fn get_telemetry_event_name(event: &TelemetryEvent) -> &'static str { TelemetryEvent::MissingCondaEnvironments(_) => "MissingCondaEnvironments", TelemetryEvent::MissingPoetryEnvironments(_) => "MissingPoetryEnvironments", TelemetryEvent::RefreshPerformance(_) => "RefreshPerformance", + TelemetryEvent::RefreshProgress(_) => "RefreshProgress", } } diff --git a/crates/pet-core/src/telemetry/refresh_progress.rs b/crates/pet-core/src/telemetry/refresh_progress.rs new file mode 100644 index 00000000..14c02b20 --- /dev/null +++ b/crates/pet-core/src/telemetry/refresh_progress.rs @@ -0,0 +1,35 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub enum RefreshProgressPhase { + Locators, + Path, + GlobalVirtualEnvs, + Workspaces, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub enum RefreshProgressStatus { + Started, + Completed, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct RefreshProgress { + pub refresh_id: u64, + pub phase: RefreshProgressPhase, + pub status: RefreshProgressStatus, + pub elapsed_ms: u128, + #[serde(skip_serializing_if = "Option::is_none")] + pub phase_elapsed_ms: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub locator_name: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub locator_elapsed_ms: Option, +} diff --git a/crates/pet-reporter/src/jsonrpc.rs b/crates/pet-reporter/src/jsonrpc.rs index fd121d4e..7f03ca8c 100644 --- a/crates/pet-reporter/src/jsonrpc.rs +++ b/crates/pet-reporter/src/jsonrpc.rs @@ -173,6 +173,43 @@ mod tests { assert_eq!(value["data"]["refreshPerformance"]["total"], json!(10)); } + #[test] + fn refresh_progress_serializes_privacy_safe_fields() { + use pet_core::telemetry::refresh_progress::{ + RefreshProgress, RefreshProgressPhase, RefreshProgressStatus, + }; + + let event = TelemetryEvent::RefreshProgress(RefreshProgress { + refresh_id: 42, + phase: RefreshProgressPhase::Locators, + status: RefreshProgressStatus::Completed, + elapsed_ms: 15, + phase_elapsed_ms: None, + locator_name: Some("Conda".to_string()), + locator_elapsed_ms: Some(10), + }); + let payload = TelemetryData { + event: get_telemetry_event_name(&event).to_string(), + data: event, + }; + + assert_eq!( + serde_json::to_value(payload).unwrap(), + json!({ + "event": "RefreshProgress", + "data": { + "refreshProgress": { + "refreshId": 42, + "phase": "locators", + "status": "completed", + "elapsedMs": 15, + "locatorName": "Conda", + "locatorElapsedMs": 10 + } + } + }) + ); + } #[test] fn log_payload_uses_camel_case_fields_and_level_renames() { let payload = Log { diff --git a/crates/pet/src/find.rs b/crates/pet/src/find.rs index 30d9a631..40722b20 100644 --- a/crates/pet/src/find.rs +++ b/crates/pet/src/find.rs @@ -7,6 +7,10 @@ use pet_core::env::PythonEnv; use pet_core::os_environment::Environment; use pet_core::python_environment::PythonEnvironmentKind; use pet_core::reporter::Reporter; +use pet_core::telemetry::refresh_progress::{ + RefreshProgress, RefreshProgressPhase, RefreshProgressStatus, +}; +use pet_core::telemetry::TelemetryEvent; use pet_core::{Configuration, Locator, LocatorKind}; use pet_env_var_path::get_search_paths_from_env_variables; use pet_global_virtualenvs::list_global_virtual_envs_paths; @@ -21,7 +25,7 @@ use std::collections::BTreeMap; use std::fs; use std::path::PathBuf; use std::sync::Mutex; -use std::time::Duration; +use std::time::{Duration, Instant}; use std::{sync::Arc, thread}; use tracing::{info_span, instrument}; @@ -42,6 +46,32 @@ pub enum SearchScope { Workspace, } +fn report_refresh_progress( + reporter: &dyn Reporter, + refresh_id: Option, + refresh_start: Instant, + phase: RefreshProgressPhase, + status: RefreshProgressStatus, + phase_elapsed: Option, + locator: Option<(String, Option)>, +) { + let Some(refresh_id) = refresh_id else { + return; + }; + + let (locator_name, locator_elapsed_ms) = locator.map_or((None, None), |(name, elapsed)| { + (Some(name), elapsed.map(|duration| duration.as_millis())) + }); + reporter.report_telemetry(&TelemetryEvent::RefreshProgress(RefreshProgress { + refresh_id, + phase, + status, + elapsed_ms: refresh_start.elapsed().as_millis(), + phase_elapsed_ms: phase_elapsed.map(|duration| duration.as_millis()), + locator_name, + locator_elapsed_ms, + })); +} #[instrument(skip(reporter, configuration, locators, environment), fields(search_scope = ?search_scope))] pub fn find_and_report_envs( reporter: &dyn Reporter, @@ -49,13 +79,14 @@ pub fn find_and_report_envs( locators: &Arc>>, environment: &dyn Environment, search_scope: Option, + refresh_id: Option, ) -> Arc> { let summary = Arc::new(Mutex::new(Summary { total: Duration::from_secs(0), locators: BTreeMap::new(), breakdown: BTreeMap::new(), })); - let start = std::time::Instant::now(); + let refresh_start = Instant::now(); // From settings let environment_directories = configuration.environment_directories.unwrap_or_default(); @@ -76,7 +107,16 @@ pub fn find_and_report_envs( s.spawn(|| { // Find in all the finders let _span = info_span!("locators_phase").entered(); - let start = std::time::Instant::now(); + let start = Instant::now(); + report_refresh_progress( + reporter, + refresh_id, + refresh_start, + RefreshProgressPhase::Locators, + RefreshProgressStatus::Started, + None, + None, + ); if search_global { thread::scope(|s| { for locator in locators.iter() { @@ -96,33 +136,71 @@ pub fn find_and_report_envs( s.spawn(move || { let locator_name = format!("{:?}", locator.get_kind()); let _span = info_span!("locator_find", locator = %locator_name).entered(); - let start = std::time::Instant::now(); + let start = Instant::now(); + report_refresh_progress( + reporter, + refresh_id, + refresh_start, + RefreshProgressPhase::Locators, + RefreshProgressStatus::Started, + None, + Some((locator_name.clone(), None)), + ); trace!("Searching using locator: {:?}", locator.get_kind()); locator.find(reporter); + let elapsed = start.elapsed(); trace!( "Completed searching using locator: {:?} in {:?}", locator.get_kind(), - start.elapsed() + elapsed ); summary .lock() .unwrap() .locators - .insert(locator.get_kind(), start.elapsed()); + .insert(locator.get_kind(), elapsed); + report_refresh_progress( + reporter, + refresh_id, + refresh_start, + RefreshProgressPhase::Locators, + RefreshProgressStatus::Completed, + None, + Some((locator_name, Some(elapsed))), + ); }); } }); } + let elapsed = start.elapsed(); summary .lock() .unwrap() .breakdown - .insert("Locators", start.elapsed()); + .insert("Locators", elapsed); + report_refresh_progress( + reporter, + refresh_id, + refresh_start, + RefreshProgressPhase::Locators, + RefreshProgressStatus::Completed, + Some(elapsed), + None, + ); }); // Step 2: Search in PATH variable s.spawn(|| { let _span = info_span!("path_search_phase").entered(); - let start = std::time::Instant::now(); + let start = Instant::now(); + report_refresh_progress( + reporter, + refresh_id, + refresh_start, + RefreshProgressPhase::Path, + RefreshProgressStatus::Started, + None, + None, + ); if search_global { let global_env_search_paths: Vec = get_search_paths_from_env_variables(environment); @@ -139,11 +217,17 @@ pub fn find_and_report_envs( &global_env_search_paths, ); } - summary - .lock() - .unwrap() - .breakdown - .insert("Path", start.elapsed()); + let elapsed = start.elapsed(); + summary.lock().unwrap().breakdown.insert("Path", elapsed); + report_refresh_progress( + reporter, + refresh_id, + refresh_start, + RefreshProgressPhase::Path, + RefreshProgressStatus::Completed, + Some(elapsed), + None, + ); }); // Step 3: Search in some global locations for virtual envs. // Convert to Arc<[PathBuf]> for O(1) cloning in thread spawns @@ -152,7 +236,16 @@ pub fn find_and_report_envs( let summary_for_step3 = summary.clone(); s.spawn(move || { let _span = info_span!("global_virtualenvs_phase").entered(); - let start = std::time::Instant::now(); + let start = Instant::now(); + report_refresh_progress( + reporter, + refresh_id, + refresh_start, + RefreshProgressPhase::GlobalVirtualEnvs, + RefreshProgressStatus::Started, + None, + None, + ); if search_global { let mut possible_environments = vec![]; @@ -197,11 +290,21 @@ pub fn find_and_report_envs( &global_env_search_paths, ); } + let elapsed = start.elapsed(); summary_for_step3 .lock() .unwrap() .breakdown - .insert("GlobalVirtualEnvs", start.elapsed()); + .insert("GlobalVirtualEnvs", elapsed); + report_refresh_progress( + reporter, + refresh_id, + refresh_start, + RefreshProgressPhase::GlobalVirtualEnvs, + RefreshProgressStatus::Completed, + Some(elapsed), + None, + ); }); // Step 4: Find in workspace folders too. // This can be merged with step 2 as well, as we're only look for environments @@ -213,7 +316,16 @@ pub fn find_and_report_envs( let summary_for_step4 = summary.clone(); s.spawn(move || { let _span = info_span!("workspace_search_phase").entered(); - let start = std::time::Instant::now(); + let start = Instant::now(); + report_refresh_progress( + reporter, + refresh_id, + refresh_start, + RefreshProgressPhase::Workspaces, + RefreshProgressStatus::Started, + None, + None, + ); thread::scope(|s| { // Find environments in the workspace folders. if !workspace_directories.is_empty() { @@ -252,14 +364,24 @@ pub fn find_and_report_envs( } }); + let elapsed = start.elapsed(); summary_for_step4 .lock() .unwrap() .breakdown - .insert("Workspaces", start.elapsed()); + .insert("Workspaces", elapsed); + report_refresh_progress( + reporter, + refresh_id, + refresh_start, + RefreshProgressPhase::Workspaces, + RefreshProgressStatus::Completed, + Some(elapsed), + None, + ); }); }); - summary.lock().expect("summary mutex poisoned").total = start.elapsed(); + summary.lock().expect("summary mutex poisoned").total = refresh_start.elapsed(); summary } @@ -438,9 +560,11 @@ pub fn identify_python_executables_using_locators( #[cfg(test)] mod tests { + use super::*; + use pet_core::{manager::EnvManager, python_environment::PythonEnvironment}; use std::fs; - #[cfg(unix)] use std::path::PathBuf; + use std::sync::Mutex as StdMutex; use tempfile::TempDir; /// Test that `path().is_dir()` properly follows symlinks to directories. @@ -653,4 +777,119 @@ mod tests { "canonicalize() would resolve to target, but path() does not" ); } + struct EmptyEnvironment; + + impl Environment for EmptyEnvironment { + fn get_user_home(&self) -> Option { + None + } + fn get_root(&self) -> Option { + None + } + fn get_env_var(&self, _key: String) -> Option { + None + } + fn get_know_global_search_locations(&self) -> Vec { + Vec::new() + } + } + + struct NoopCondaLocator; + + impl Locator for NoopCondaLocator { + fn get_kind(&self) -> LocatorKind { + LocatorKind::Conda + } + fn supported_categories(&self) -> Vec { + vec![PythonEnvironmentKind::Conda] + } + fn try_from(&self, _env: &PythonEnv) -> Option { + None + } + fn find(&self, _reporter: &dyn Reporter) {} + } + + #[derive(Default)] + struct ProgressReporter { + events: StdMutex>, + } + + impl Reporter for ProgressReporter { + fn report_manager(&self, _manager: &EnvManager) {} + fn report_environment(&self, _env: &PythonEnvironment) {} + fn report_telemetry(&self, event: &TelemetryEvent) { + self.events.lock().unwrap().push(event.clone()); + } + } + + #[test] + fn refresh_progress_reports_phases_and_locator_timing() { + let reporter = ProgressReporter::default(); + let locators: Arc>> = Arc::new(vec![Arc::new(NoopCondaLocator)]); + + find_and_report_envs( + &reporter, + Configuration::default(), + &locators, + &EmptyEnvironment, + Some(SearchScope::Global(PythonEnvironmentKind::Conda)), + Some(42), + ); + + let events = reporter.events.lock().unwrap(); + let progress = events + .iter() + .filter_map(|event| match event { + TelemetryEvent::RefreshProgress(progress) => Some(progress), + _ => None, + }) + .collect::>(); + + assert_eq!(progress.len(), 10); + assert!(progress.iter().all(|progress| progress.refresh_id == 42)); + for phase in [ + RefreshProgressPhase::Locators, + RefreshProgressPhase::Path, + RefreshProgressPhase::GlobalVirtualEnvs, + RefreshProgressPhase::Workspaces, + ] { + assert_eq!( + progress + .iter() + .filter(|progress| { + progress.phase == phase + && progress.status == RefreshProgressStatus::Started + && progress.locator_name.is_none() + }) + .count(), + 1 + ); + assert_eq!( + progress + .iter() + .filter(|progress| { + progress.phase == phase + && progress.status == RefreshProgressStatus::Completed + && progress.locator_name.is_none() + && progress.phase_elapsed_ms.is_some() + }) + .count(), + 1 + ); + } + + let locator_progress = progress + .iter() + .filter(|progress| progress.locator_name.as_deref() == Some("Conda")) + .collect::>(); + assert_eq!(locator_progress.len(), 2); + assert!(locator_progress.iter().any(|progress| { + progress.status == RefreshProgressStatus::Started + && progress.locator_elapsed_ms.is_none() + })); + assert!(locator_progress.iter().any(|progress| { + progress.status == RefreshProgressStatus::Completed + && progress.locator_elapsed_ms.is_some() + })); + } } diff --git a/crates/pet/src/jsonrpc.rs b/crates/pet/src/jsonrpc.rs index ee93fa46..489062ce 100644 --- a/crates/pet/src/jsonrpc.rs +++ b/crates/pet/src/jsonrpc.rs @@ -486,6 +486,7 @@ const MISSING_ENVS_AVAILABLE: u64 = u64::MAX; const MISSING_ENVS_COMPLETED: u64 = u64::MAX - 1; static MISSING_ENVS_REPORTING_STATE: AtomicU64 = AtomicU64::new(MISSING_ENVS_AVAILABLE); +static NEXT_REFRESH_ID: AtomicU64 = AtomicU64::new(1); pub fn start_jsonrpc_server() { // Initialize tracing for performance profiling (controlled by RUST_LOG env var) @@ -656,18 +657,20 @@ pub struct RefreshOptions { } #[derive(Debug, Clone, Deserialize, Serialize)] +#[serde(rename_all = "camelCase")] pub struct RefreshResult { duration: u128, + refresh_id: u64, } impl RefreshResult { - pub fn new(duration: Duration) -> RefreshResult { + pub fn new(duration: Duration, refresh_id: u64) -> RefreshResult { RefreshResult { duration: duration.as_millis(), + refresh_id, } } } - fn normalize_refresh_params(params: Value) -> Value { match params { Value::Null => json!({}), @@ -965,6 +968,7 @@ fn execute_refresh( refresh_options: &RefreshOptions, configuration_state: &ConfigurationState, ) -> RefreshExecution { + let refresh_id = NEXT_REFRESH_ID.fetch_add(1, Ordering::Relaxed); let refresh_locators = create_refresh_locators( context.os_environment.deref(), context.conda_locator.as_ref(), @@ -1004,6 +1008,7 @@ fn execute_refresh( &refresh_locators.locators, context.os_environment.deref(), search_scope.clone(), + Some(refresh_id), ); let summary = summary.lock().expect("summary mutex poisoned"); for locator in summary.locators.iter() { @@ -1051,7 +1056,7 @@ fn execute_refresh( }; RefreshExecution { - result: RefreshResult::new(summary.total), + result: RefreshResult::new(summary.total, refresh_id), perf, reporter, configuration: context.configuration.clone(), diff --git a/crates/pet/src/lib.rs b/crates/pet/src/lib.rs index a98374f6..532de84f 100644 --- a/crates/pet/src/lib.rs +++ b/crates/pet/src/lib.rs @@ -195,7 +195,8 @@ fn find_envs( let stdio_reporter = Arc::new(stdio::create_reporter(options.print_list, kind)); let reporter = CacheReporter::new(stdio_reporter.clone()); - let summary = find_and_report_envs(&reporter, config, locators, environment, search_scope); + let summary = + find_and_report_envs(&reporter, config, locators, environment, search_scope, None); if options.report_missing { // By now all conda envs have been found // Spawn conda @@ -307,7 +308,7 @@ fn find_envs_json( let collect_reporter = Arc::new(collect::create_reporter()); let reporter = CacheReporter::new(collect_reporter.clone()); - find_and_report_envs(&reporter, config, locators, environment, search_scope); + find_and_report_envs(&reporter, config, locators, environment, search_scope, None); if options.report_missing { let _ = conda_locator.find_and_report_missing_envs(&reporter, options.conda_executable.clone()); diff --git a/crates/pet/tests/ci_homebrew_container.rs b/crates/pet/tests/ci_homebrew_container.rs index 52e9bf2a..db453cb7 100644 --- a/crates/pet/tests/ci_homebrew_container.rs +++ b/crates/pet/tests/ci_homebrew_container.rs @@ -44,6 +44,7 @@ fn verify_python_in_homebrew_contaner() { &create_locators(conda_locator.clone(), poetry_locator.clone(), &environment), &environment, None, + None, ); let environments = reporter.environments.lock().unwrap().clone(); diff --git a/crates/pet/tests/ci_jupyter_container.rs b/crates/pet/tests/ci_jupyter_container.rs index 710cfd6e..64b20cf3 100644 --- a/crates/pet/tests/ci_jupyter_container.rs +++ b/crates/pet/tests/ci_jupyter_container.rs @@ -47,6 +47,7 @@ fn verify_python_in_jupyter_contaner() { &create_locators(conda_locator.clone(), poetry_locator.clone(), &environment), &environment, None, + None, ); let environments = reporter.environments.lock().unwrap().clone(); diff --git a/crates/pet/tests/ci_poetry.rs b/crates/pet/tests/ci_poetry.rs index 277c07a0..ccb1ee5b 100644 --- a/crates/pet/tests/ci_poetry.rs +++ b/crates/pet/tests/ci_poetry.rs @@ -55,6 +55,7 @@ fn verify_ci_poetry_global() { &locators, &environment, None, + None, ); let environments = reporter.environments.lock().unwrap().clone(); @@ -127,6 +128,7 @@ fn verify_ci_poetry_project() { &locators, &environment, None, + None, ); let environments = reporter.environments.lock().unwrap().clone(); diff --git a/crates/pet/tests/ci_test.rs b/crates/pet/tests/ci_test.rs index 218737cd..187df616 100644 --- a/crates/pet/tests/ci_test.rs +++ b/crates/pet/tests/ci_test.rs @@ -91,6 +91,7 @@ fn verify_validity_of_discovered_envs() { &locators, &environment, None, + None, ); let environments = reporter.environments.lock().unwrap().clone(); @@ -154,6 +155,7 @@ fn check_if_virtualenvwrapper_exists() { &create_locators(conda_locator.clone(), poetry_locator.clone(), &environment), &environment, None, + None, ); let environments = reporter.environments.lock().unwrap().clone(); @@ -196,6 +198,7 @@ fn check_if_pipenv_exists() { &create_locators(conda_locator.clone(), poetry_locator.clone(), &environment), &environment, None, + None, ); let environments = reporter.environments.lock().unwrap().clone(); @@ -234,6 +237,7 @@ fn check_if_pyenv_virtualenv_exists() { &create_locators(conda_locator.clone(), poetry_locator.clone(), &environment), &environment, None, + None, ); let environments = reporter.environments.lock().unwrap().clone(); @@ -712,6 +716,7 @@ fn verify_bin_usr_bin_user_local_are_separate_python_envs() { &create_locators(conda_locator.clone(), poetry_locator.clone(), &environment), &environment, None, + None, ); let environments = reporter.environments.lock().unwrap().clone(); diff --git a/crates/pet/tests/jsonrpc_client.rs b/crates/pet/tests/jsonrpc_client.rs index 9612bce1..1496ffcc 100644 --- a/crates/pet/tests/jsonrpc_client.rs +++ b/crates/pet/tests/jsonrpc_client.rs @@ -17,8 +17,10 @@ static REQUEST_ID: AtomicU32 = AtomicU32::new(1); const DEFAULT_REQUEST_TIMEOUT: Duration = Duration::from_secs(30); #[derive(Debug, Clone, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] pub struct RefreshResult { pub duration: u128, + pub refresh_id: u64, } #[derive(Debug, Clone, Deserialize, PartialEq, Eq)] @@ -192,6 +194,40 @@ impl PetJsonRpcClient { .count() } + pub fn telemetry_events(&self, event: &str) -> Vec { + self.notifications() + .into_iter() + .filter(|notification| { + notification.method == "telemetry" + && notification.params["event"].as_str() == Some(event) + }) + .map(|notification| notification.params) + .collect() + } + + pub fn telemetry_event_count(&self, event: &str) -> usize { + self.telemetry_events(event).len() + } + + pub fn wait_for_telemetry_event_count( + &self, + event: &str, + expected_count: usize, + timeout: Duration, + ) -> Result<(), String> { + let deadline = Instant::now() + timeout; + while Instant::now() <= deadline { + if self.telemetry_event_count(event) >= expected_count { + return Ok(()); + } + thread::sleep(Duration::from_millis(10)); + } + Err(format!( + "Timed out waiting for {expected_count} '{event}' telemetry events; saw {}. stderr: {}", + self.telemetry_event_count(event), + self.stderr_output() + )) + } pub fn wait_for_notification_count( &self, method: &str, diff --git a/crates/pet/tests/jsonrpc_server_test.rs b/crates/pet/tests/jsonrpc_server_test.rs index d6fe8d3b..4b9b5dda 100644 --- a/crates/pet/tests/jsonrpc_server_test.rs +++ b/crates/pet/tests/jsonrpc_server_test.rs @@ -142,13 +142,28 @@ fn configure_and_workspace_refresh_report_fake_venv() { .expect("configure request failed"); client.clear_notifications(); - client + let refresh = client .refresh(Some(json!({ "searchPaths": [workspace.clone()] }))) .expect("refresh request failed"); client - .wait_for_notification_count("telemetry", 1, Duration::from_secs(5)) - .expect("timed out waiting for refresh telemetry"); + .wait_for_telemetry_event_count("RefreshPerformance", 1, Duration::from_secs(5)) + .expect("timed out waiting for refresh performance telemetry"); + let progress = client.telemetry_events("RefreshProgress"); + assert_eq!( + progress.len(), + 8, + "expected started/completed for four phases" + ); + assert!(progress.iter().all(|event| { + event["data"]["refreshProgress"]["refreshId"].as_u64() == Some(refresh.refresh_id) + })); + assert!(progress.iter().all(|event| { + let data = &event["data"]["refreshProgress"]; + data.get("executable").is_none() + && data.get("prefix").is_none() + && data.get("path").is_none() + })); let environments = client.environment_notifications(); assert_single_environment( &environments, @@ -162,7 +177,7 @@ fn configure_and_workspace_refresh_report_fake_venv() { 0, "fake venv refresh should not report any managers" ); - assert_eq!(client.notification_count("telemetry"), 1); + assert_eq!(client.telemetry_event_count("RefreshPerformance"), 1); } #[test] @@ -198,8 +213,8 @@ fn concurrent_identical_refresh_requests_share_one_notification_stream() { assert_eq!(refresh_results.len(), 3); for result in refresh_results.windows(2) { assert_eq!( - result[0].duration, result[1].duration, - "joined refreshes should reuse the same refresh result" + result[0], result[1], + "joined refreshes should reuse the same refresh result and ID" ); } @@ -211,8 +226,14 @@ fn concurrent_identical_refresh_requests_share_one_notification_stream() { ) .expect("timed out waiting for environment notifications"); client - .wait_for_notification_count("telemetry", 1, Duration::from_secs(5)) - .expect("timed out waiting for refresh telemetry"); + .wait_for_telemetry_event_count("RefreshPerformance", 1, Duration::from_secs(5)) + .expect("timed out waiting for refresh performance telemetry"); + let progress = client.telemetry_events("RefreshProgress"); + assert_eq!(progress.len(), 8); + assert!(progress.iter().all(|event| { + event["data"]["refreshProgress"]["refreshId"].as_u64() + == Some(refresh_results[0].refresh_id) + })); let environments = client.environment_notifications(); assert_eq!( @@ -254,9 +275,9 @@ fn concurrent_identical_refresh_requests_share_one_notification_stream() { "identical refresh requests should emit one environment notification stream" ); assert_eq!( - client.notification_count("telemetry"), + client.telemetry_event_count("RefreshPerformance"), 1, - "identical refresh requests should emit one telemetry notification" + "identical refresh requests should emit one performance event" ); } @@ -283,21 +304,22 @@ fn concurrent_distinct_refresh_requests_run_separately() { let handle_b = thread::spawn(move || client_b.refresh(Some(json!({ "searchPaths": [workspace_b] })))); - handle_a + let result_a = handle_a .join() .expect("first refresh thread panicked") .expect("first refresh failed"); - handle_b + let result_b = handle_b .join() .expect("second refresh thread panicked") .expect("second refresh failed"); + assert_ne!(result_a.refresh_id, result_b.refresh_id); client .wait_for_notification_count("environment", 2, Duration::from_secs(5)) .expect("timed out waiting for environment notifications"); client - .wait_for_notification_count("telemetry", 2, Duration::from_secs(5)) - .expect("timed out waiting for telemetry notifications"); + .wait_for_telemetry_event_count("RefreshPerformance", 2, Duration::from_secs(5)) + .expect("timed out waiting for refresh performance telemetry"); let mut environments = client.environment_notifications(); environments.sort_by(|left, right| left.name.cmp(&right.name)); @@ -335,8 +357,8 @@ fn concurrent_distinct_refresh_requests_run_separately() { ) ); assert_eq!( - client.notification_count("telemetry"), + client.telemetry_event_count("RefreshPerformance"), 2, - "distinct refresh requests should emit separate telemetry notifications" + "distinct refresh requests should emit separate performance events" ); } diff --git a/docs/JSONRPC.md b/docs/JSONRPC.md index 85292326..fca5a403 100644 --- a/docs/JSONRPC.md +++ b/docs/JSONRPC.md @@ -166,9 +166,43 @@ interface RefreshResult { * Duration is in milliseconds. */ duration: number; + /** + * Identifier shared by this result and all RefreshProgress telemetry emitted + * for the refresh operation. Concurrent identical requests that join the same + * operation receive the same identifier. + */ + refreshId: number; } ``` +## Refresh Progress Telemetry + +During a refresh, the server emits `telemetry` notifications when each major phase +and locator starts and completes. These notifications contain timing and enum values +only; they never contain paths, environment names, executable paths, usernames, or +command lines. + +```typescript +interface RefreshProgressTelemetry { + event: "RefreshProgress"; + data: { + refreshProgress: { + refreshId: number; + phase: "locators" | "path" | "globalVirtualEnvs" | "workspaces"; + status: "started" | "completed"; + elapsedMs: number; + phaseElapsedMs?: number; + locatorName?: string; + locatorElapsedMs?: number; + }; + }; +} +``` + +`phaseElapsedMs` is present for completed phases. Locator events use the `locators` +phase and include `locatorName`; completed locator events also include +`locatorElapsedMs`. + # Resolve Request Use this request to resolve a Python environment from a given Python path.