refactor(log): in-place writer lifecycle and span-based core observability (#7889)

* refactor(log): migrate logging to tracing-estuary pipeline

Swap flexi_logger + clash-verge-logger wiring for the tracing-estuary
crate: buffered file sink with explicit panic/exit drains, live level
switching, and tokio-trace/chrome-trace behind crate features.

AsyncLogger moves in-house (VecDeque ring); file format and rotation
stay unchanged.

* fix(log): reset sidecar writer in place and re-pin service-ipc v2.6.7

Rebuilding the sidecar FileLogWriter on config changes leaked flexi's
unstoppable flusher thread plus an fd each time; the writer is now flushed
and reset in place, matching what the service crate does.
get_clash_logs_by_service drops the CompactString conversion layer now that
service-ipc v2.6.7 returns plain Vec<String>, and AsyncLogger is renamed
LogRing to match.

* refactor(log): span-based observability and log noise cleanup

Instrument spans carry the structure flat lines never could: the core-start
spine (mode, readiness/owner generations, pid, config file), proxy apply and
clear routes with guard generations, service IPC sessions with refusal codes
and outcomes, profile-switch terminal outcomes, subscription fetch strategies,
core upgrades with per-proxy version probes, backup flows, timer tasks, config
generation uids, and the startup setup flow; running-mode changes emit one
guarded transition event each.
~90 hand-written noise lines are gone or demoted (entry+success pairs, polled
getter echoes, phase brackets, window sub-steps, platform install chatter);
errors keep their anyhow chain, operation name, and item id, and swallowed
retry results surface per attempt.
This commit is contained in:
Tunglies
2026-09-08 05:51:04 +08:00
committed by GitHub
parent f8b69b5d62
commit 2f370d6490
43 changed files with 807 additions and 797 deletions
+20 -8
View File
@@ -90,7 +90,7 @@ async fn ask_to_stage(path: &std::path::Path) -> StageAttempt {
match crate::core::service::stage_runtime_by_service(path).await {
Ok(StageRequest::Answered(outcome)) => StageAttempt::Answered(outcome),
Ok(StageRequest::Refused { code, message }) => {
let message = message.to_string().into();
let message = message.into();
if StageRequest::is_about_the_bundle(code) {
StageAttempt::RefusedTheBundle(message)
} else {
@@ -103,6 +103,7 @@ async fn ask_to_stage(path: &std::path::Path) -> StageAttempt {
/// Confirms one silent request because the service may commit before its reply is lost.
/// Re-staging an already committed bundle is idempotent and bounded by `confirm_within`.
#[tracing::instrument(skip_all, level = "debug", fields(confirm_within = ?confirm_within, asks = 1))]
async fn stage_with_confirmation<Ask, Fut>(confirm_within: Duration, ask: Ask) -> StageAttempt
where
Ask: Fn() -> Fut,
@@ -117,6 +118,7 @@ where
Type::Core,
"Staging did not answer ({first}); asking once more before replacing the core"
);
tracing::Span::current().record("asks", 2);
match tokio::time::timeout(confirm_within, ask()).await {
Ok(StageAttempt::Unanswered(again)) => {
StageAttempt::Unanswered(format!("{first}; asked again: {again}").into())
@@ -150,6 +152,7 @@ impl CoreManager {
self.update_config_with_force(true).await
}
#[tracing::instrument(skip_all, level = "info", fields(force))]
pub async fn update_config_with_force(&self, force: bool) -> Result<ValidationOutcome> {
if handle::Handle::global().is_exiting() {
return Ok(ValidationOutcome::Skipped {
@@ -158,7 +161,7 @@ impl CoreManager {
}
if !self.try_start_config_update() {
logging!(info, Type::Core, "Configuration update is already running");
logging!(debug, Type::Core, "Configuration update is already running");
return Ok(ValidationOutcome::Busy);
}
defer! {
@@ -179,6 +182,7 @@ impl CoreManager {
self.perform_config_update(None).await
}
#[tracing::instrument(skip_all, level = "info", fields(profile = ?candidate.current, outcome = tracing::field::Empty))]
pub(crate) async fn update_config_forced_with_profiles(
&self,
candidate: &IProfiles,
@@ -202,17 +206,23 @@ impl CoreManager {
{
Ok(outcome) => outcome,
Err(error) => {
tracing::Span::current().record("outcome", "rolled_back");
self.restore_profile_config(rollback).await?;
return Err(error);
}
};
if !outcome.is_valid() {
tracing::Span::current().record("outcome", "invalid");
crate::config::profiles::restore_selected_nodes().await;
return Ok(Err(outcome));
}
match candidate.save_file().await {
Ok(()) => Ok(Ok(guard)),
Ok(()) => {
tracing::Span::current().record("outcome", "committed");
Ok(Ok(guard))
}
Err(error) => {
tracing::Span::current().record("outcome", "save_rolled_back");
self.restore_profile_config(rollback).await?;
Err(error)
}
@@ -273,7 +283,7 @@ impl CoreManager {
F: FnOnce(&mut IRuntime),
{
if !self.try_start_config_update() {
logging!(info, Type::Core, "Configuration update is already running");
logging!(debug, Type::Core, "Configuration update is already running");
return Ok(ValidationOutcome::Busy);
}
defer! {
@@ -315,6 +325,7 @@ impl CoreManager {
/// Applies through the service, replacing the core when in-place staging is unavailable.
/// Caller must hold `lifecycle_lock`.
#[tracing::instrument(skip_all, level = "info", fields(path = %path.display(), outcome = tracing::field::Empty))]
async fn apply_config_by_service(&self, path: &std::path::Path) -> Result<()> {
match plan_config_application(&self.attempt_staging(path).await) {
ConfigApplication::Fail(message) => {
@@ -328,7 +339,7 @@ impl CoreManager {
}
ConfigApplication::ReloadFrom(staged) => match self.reload_config(&staged).await {
Ok(()) => {
logging!(info, Type::Core, "Configuration staged and applied by service");
tracing::Span::current().record("outcome", "staged");
return Ok(());
}
Err(err) => logging!(
@@ -341,7 +352,7 @@ impl CoreManager {
}
self.replace_service_core_with_config(path).await?;
logging!(info, Type::Core, "Configuration materialized and applied by service");
tracing::Span::current().record("outcome", "replaced");
Ok(())
}
@@ -353,9 +364,10 @@ impl CoreManager {
}
/// Reload the Core from `path`, and replace the Core if it will not take it.
#[tracing::instrument(skip_all, level = "info", fields(outcome = tracing::field::Empty))]
async fn reload_or_restart(&self, path: &str) -> Result<()> {
let Err(err) = self.reload_config(path).await else {
logging!(info, Type::Core, "Configuration applied");
tracing::Span::current().record("outcome", "reloaded");
return Ok(());
};
@@ -366,7 +378,7 @@ impl CoreManager {
);
match self.restart_core_during_config_update().await {
Ok(_) => {
logging!(info, Type::Core, "Configuration applied after restart");
tracing::Span::current().record("outcome", "restarted");
Ok(())
}
Err(err) => {
+15 -3
View File
@@ -317,6 +317,7 @@ impl CoreManager {
.await
}
#[tracing::instrument(skip_all, level = "info", fields(status = tracing::field::Empty, tun_disabled = false, readiness_generation = tracing::field::Empty))]
pub async fn continue_with_sidecar(&self) -> Result<()> {
if !self.try_start_config_update() {
anyhow::bail!("configuration update is already running");
@@ -329,6 +330,7 @@ impl CoreManager {
let config_write = Config::lock_config_write().await;
let _life = self.lifecycle_lock.lock().await;
let status = SERVICE_MANAGER.current().await;
tracing::Span::current().record("status", tracing::field::debug(&status));
let mode = self.get_running_mode();
if !can_allow_sidecar_for_session(&mode, &status) {
anyhow::bail!("Sidecar continuation is not allowed from {mode:?} / {status:?}");
@@ -342,6 +344,7 @@ impl CoreManager {
.state()
.tun_should_be_disabled(tun_enabled)
{
tracing::Span::current().record("tun_disabled", true);
Config::disable_tun_and_persist().await?;
}
Config::generate().await
@@ -361,6 +364,7 @@ impl CoreManager {
{
anyhow::bail!("Sidecar did not become ready");
}
tracing::Span::current().record("readiness_generation", self.current_core_readiness_generation());
self.apply_proxy_after_start().await
}
.await;
@@ -376,6 +380,7 @@ impl CoreManager {
Ok(())
}
#[tracing::instrument(skip_all, level = "info", fields(readiness_generation = tracing::field::Empty))]
pub async fn uninstall_service_and_start_sidecar(&self) -> Result<()> {
if !self.try_start_config_update() {
anyhow::bail!("configuration update is already running");
@@ -413,6 +418,7 @@ impl CoreManager {
{
anyhow::bail!("Sidecar did not become ready after service uninstall");
}
tracing::Span::current().record("readiness_generation", self.current_core_readiness_generation());
self.apply_proxy_after_start().await
},
)
@@ -456,6 +462,7 @@ impl CoreManager {
.await
}
#[tracing::instrument(skip_all, level = "info", fields(mode = ?*self.get_running_mode(), readiness_generation = self.current_core_readiness_generation(), owner_generation = crate::core::service::owner_monitor_generation()))]
pub(crate) async fn apply_proxy_after_start(&self) -> Result<()> {
let expectation = ProxyRestoreExpectation::capture(
*self.get_running_mode(),
@@ -519,6 +526,7 @@ impl CoreManager {
Ok(())
}
#[tracing::instrument(skip_all, level = "info", fields(decision = tracing::field::Empty))]
async fn start_core_inner(&self) -> Result<()> {
if Handle::global().is_exiting() {
return Ok(());
@@ -526,7 +534,7 @@ impl CoreManager {
if !matches!(*self.get_running_mode(), RunningMode::NotRunning) {
logging!(
info,
debug,
Type::Core,
"start_core called while a core is running; treated as no-op"
);
@@ -534,6 +542,7 @@ impl CoreManager {
}
let startup = self.prepare_startup().await;
tracing::Span::current().record("decision", tracing::field::debug(&startup));
if matches!(startup, StartupDecision::Wait) {
self.rollback_failed_start().await;
return Ok(());
@@ -563,6 +572,7 @@ impl CoreManager {
result
}
#[tracing::instrument(skip_all, level = "info", fields(mode = ?*self.get_running_mode()))]
pub async fn stop_core(&self) -> Result<()> {
let _life = self.lifecycle_lock.lock().await;
self.controlled_stop_core_inner().await
@@ -594,7 +604,7 @@ impl CoreManager {
}
async fn stop_core_unprepared_inner(&self) -> Result<()> {
CLASH_LOGGER.clear_logs().await;
CLASH_LOGGER.clear_logs();
match *self.get_running_mode() {
RunningMode::Service => self.stop_core_by_service().await,
RunningMode::Sidecar => {
@@ -605,6 +615,7 @@ impl CoreManager {
}
}
#[tracing::instrument(skip_all, level = "info")]
pub async fn restart_core(&self) -> Result<()> {
if !self.try_start_config_update() {
anyhow::bail!("configuration update is already running");
@@ -617,7 +628,6 @@ impl CoreManager {
pub(crate) async fn restart_core_during_config_update(&self) -> Result<()> {
let _life = self.lifecycle_lock.lock().await;
logging!(info, Type::Core, "Restarting core");
let proxy_intent = self.proxy_stop_intent().await;
run_core_replacement_transition(
|| self.controlled_stop_core_with_intent(proxy_intent),
@@ -635,6 +645,7 @@ impl CoreManager {
.await
}
#[tracing::instrument(skip_all, level = "info", fields(core = %clash_core))]
pub async fn change_core(&self, clash_core: &String) -> Result<()> {
if !IVerge::VALID_CLASH_CORES.contains(&clash_core.as_str()) {
anyhow::bail!("invalid clash core: {clash_core}");
@@ -757,6 +768,7 @@ impl CoreManager {
}
#[cfg(target_os = "windows")]
#[tracing::instrument(skip_all, level = "debug", fields(outcome = tracing::field::Empty))]
async fn try_handoff_sidecar_to_service(&self) -> HandoffOutcome {
// Before probing: a probe reply records an observation, which clears sidecar_allowed.
if crate::core::runstate::RUN_STATE.state().sidecar_allowed {
+13 -4
View File
@@ -4,8 +4,7 @@ mod state;
use anyhow::Result;
use arc_swap::{ArcSwap, ArcSwapOption};
use clash_verge_logger::AsyncLogger;
use clash_verge_logging::{Type, logging};
use clash_verge_logging::{LogRing, Type, logging};
use once_cell::sync::Lazy;
use std::{
fmt,
@@ -22,7 +21,7 @@ use crate::singleton;
#[cfg(target_os = "windows")]
use std::os::windows::io::OwnedHandle;
pub(crate) static CLASH_LOGGER: Lazy<Arc<AsyncLogger>> = Lazy::new(|| Arc::new(AsyncLogger::new()));
pub(crate) static CLASH_LOGGER: Lazy<Arc<LogRing>> = Lazy::new(|| Arc::new(LogRing::new()));
tokio::task_local! {
static PROFILE_SELECTIONS_PENDING_COMMIT: bool;
@@ -161,6 +160,10 @@ impl CoreManager {
/// Run State derives PAC availability and the outward mode mirror from this; callers must
/// not set those alongside.
pub fn core_started(&self, mode: RunningMode) {
let previous = *self.get_running_mode();
if previous != mode {
logging!(info, Type::Core, "Core running mode changed: {previous} -> {mode}");
}
self.run_state.core_started(mode);
}
@@ -169,6 +172,10 @@ impl CoreManager {
/// Core readiness is invalidated here rather than inside Run State because readiness
/// belongs to the process this manager supervises.
pub fn core_stopped(&self) {
let previous = *self.get_running_mode();
if !matches!(previous, RunningMode::NotRunning) {
logging!(info, Type::Core, "Core running mode changed: {previous} -> NotRunning");
}
self.invalidate_core_readiness();
self.run_state.core_stopped();
}
@@ -244,6 +251,7 @@ impl CoreManager {
self.config_update_in_progress.store(false, Ordering::Release);
}
#[tracing::instrument(skip_all, level = "info", fields(retries = 0))]
pub async fn init(&self) -> Result<bool> {
const MAX_PORT_FALLBACK_RETRIES: usize = 3;
@@ -266,10 +274,11 @@ impl CoreManager {
match crate::config::Config::resolve_startup_mixed_port().await {
Ok(true) => {
retries += 1;
tracing::Span::current().record("retries", retries);
logging!(
warn,
Type::Core,
"Retrying core startup after mixed proxy port fallback ({}/{})",
"Retrying core startup after mixed proxy port fallback ({}/{}): {start_error:#}",
retries,
MAX_PORT_FALLBACK_RETRIES
);
+27 -32
View File
@@ -10,7 +10,6 @@ use crate::{
};
use anyhow::{Context as _, Result};
use clash_verge_logging::Type;
use compact_str::CompactString;
use log::Level;
use scopeguard::defer;
use std::path::Path;
@@ -49,7 +48,16 @@ where
for attempt in 0..max_attempts {
match probe().await {
Ok(()) => return Ok(()),
Err(error) => last_error = Some(error),
Err(error) => {
logging!(
debug,
Type::Core,
"sidecar readiness probe {}/{} failed: {error:#}",
attempt + 1,
max_attempts
);
last_error = Some(error);
}
}
if attempt + 1 < max_attempts {
tokio::time::sleep(retry_delay).await;
@@ -80,16 +88,16 @@ use {
};
impl CoreManager {
pub async fn get_clash_logs(&self) -> Result<Vec<CompactString>> {
pub async fn get_clash_logs(&self) -> Result<Vec<String>> {
match *self.get_running_mode() {
RunningMode::Service => service::get_clash_logs_by_service().await,
RunningMode::Sidecar => Ok(CLASH_LOGGER.get_logs().await),
RunningMode::Sidecar => Ok(CLASH_LOGGER.get_logs()),
RunningMode::NotRunning => Ok(Vec::new()),
}
}
#[tracing::instrument(skip_all, level = "info", fields(pid = tracing::field::Empty))]
pub(super) async fn start_core_by_sidecar(&self) -> Result<()> {
logging!(info, Type::Core, "Starting core in sidecar mode");
self.core_stopped();
let sidecar_ipc = dirs::sidecar_ipc_path()?;
@@ -158,7 +166,7 @@ impl CoreManager {
};
let pid = child.pid();
logging!(trace, Type::Core, "Sidecar started with PID: {}", pid);
tracing::Span::current().record("pid", pid);
let readiness = poll_sidecar_readiness(SIDECAR_READINESS_ATTEMPTS, SIDECAR_READINESS_INTERVAL, || async {
tokio::time::timeout(SIDECAR_READINESS_PROBE_TIMEOUT, async {
@@ -192,22 +200,22 @@ impl CoreManager {
match event {
tauri_plugin_shell::process::CommandEvent::Stdout(line)
| tauri_plugin_shell::process::CommandEvent::Stderr(line) => {
let message = CompactString::from(&*String::from_utf8_lossy(&line));
let message = String::from_utf8_lossy(&line).into_owned();
Logger::global().writer_sidecar_log(Level::Error, &message);
CLASH_LOGGER.append_log(message).await;
CLASH_LOGGER.append_log(message);
}
tauri_plugin_shell::process::CommandEvent::Terminated(term) => {
let manager = Self::global();
let _ = manager.invalidate_core_readiness_if(core_readiness_generation);
let message = if let Some(code) = term.code {
CompactString::from(format!("Process terminated with code: {}", code))
format!("Process terminated with code: {}", code)
} else if let Some(signal) = term.signal {
CompactString::from(format!("Process terminated by signal: {}", signal))
format!("Process terminated by signal: {}", signal)
} else {
CompactString::from("Process terminated")
String::from("Process terminated")
};
Logger::global().writer_sidecar_log(Level::Info, &message);
CLASH_LOGGER.clear_logs().await;
CLASH_LOGGER.clear_logs();
manager.clear_terminated_sidecar(pid).await;
break;
}
@@ -222,7 +230,6 @@ impl CoreManager {
/// Terminates the sidecar after its caller has successfully cleared the
/// system proxy.
pub(super) fn stop_core_by_sidecar_unprepared(&self) {
logging!(info, Type::Core, "Stopping sidecar");
defer! {
self.core_stopped();
}
@@ -234,27 +241,16 @@ impl CoreManager {
// Setting the job handle to None clears the stored handle and
// closes the previous Windows job handle in `set_job_handle`.
self.set_job_handle(None);
logging!(
trace,
Type::Core,
"Closed job handle for sidecar process (PID: {})",
pid
);
let _ = pid;
}
let result = child.kill();
logging!(
trace,
Type::Core,
"Sidecar stopped (PID: {:?}, Result: {:?})",
pid,
result
);
if let Err(error) = child.kill() {
logging!(warn, Type::Core, "failed to terminate sidecar PID {pid}: {error:#}");
}
}
}
pub(super) async fn start_core_by_service(&self) -> Result<()> {
logging!(info, Type::Core, "Starting core in service mode");
self.core_starting();
let service_ipc = dirs::ipc_path()?;
let config_file = Config::generate_file(crate::config::ConfigType::Run).await?;
@@ -265,6 +261,7 @@ impl CoreManager {
self.start_core_by_service_with_config(&config_file).await
}
#[tracing::instrument(skip_all, level = "info", fields(config_file = %config_file.display()))]
pub(super) async fn start_core_by_service_with_config(&self, config_file: &Path) -> Result<()> {
// 交接时等待 sidecar 释放 ext-controller 通道。
#[cfg(target_os = "windows")]
@@ -283,10 +280,9 @@ impl CoreManager {
logging!(
warn,
Type::Core,
"service start attempt {}/{} failed: {}",
"service start attempt {}/{} failed: {e:#}",
attempt + 1,
timing::SERVICE_START_RETRIES,
e
timing::SERVICE_START_RETRIES
);
last_err = Some(e);
tokio::time::sleep(timing::SERVICE_START_RETRY_DELAY).await;
@@ -307,7 +303,6 @@ impl CoreManager {
}
pub(super) async fn stop_core_by_service(&self) -> Result<()> {
logging!(info, Type::Core, "Stopping service");
service::stop_core_by_service().await?;
self.core_stopped();
Ok(())