From 06c9127a22c23864b7982b7153d1c92b463d8db0 Mon Sep 17 00:00:00 2001 From: nofyso <1052470899@qq.com> Date: Sun, 19 Jul 2026 11:48:24 +0800 Subject: [PATCH] fix: try fix friend issue --- Cargo.lock | 10 ++-- Cargo.toml | 2 +- cuscuta-common/src/db/log/event.rs | 28 ++++++++- cuscuta-common/src/db/log/mod.rs | 11 ++-- cuscuta-common/src/db/log/status.rs | 27 ++++++++- cuscuta-worker/src/worker/pending_friend.rs | 63 +++++++++------------ cuscutactl/Cargo.toml | 2 +- cuscutactl/src/command.rs | 48 +++++++++++++++- cuscutactl/src/main.rs | 18 ++++++ cuscutactl/src/stats.rs | 58 +++++++++++++++++++ 10 files changed, 216 insertions(+), 51 deletions(-) create mode 100644 cuscutactl/src/stats.rs diff --git a/Cargo.lock b/Cargo.lock index 38e9f7c..fe13934 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -554,7 +554,7 @@ dependencies = [ [[package]] name = "cuscuta-chilo" -version = "0.1.5" +version = "0.1.6" dependencies = [ "axum", "base64", @@ -571,7 +571,7 @@ dependencies = [ [[package]] name = "cuscuta-common" -version = "0.1.5" +version = "0.1.6" dependencies = [ "base64", "chrono", @@ -589,7 +589,7 @@ dependencies = [ [[package]] name = "cuscuta-entry" -version = "0.1.5" +version = "0.1.6" dependencies = [ "anyhow", "axum", @@ -629,7 +629,7 @@ dependencies = [ [[package]] name = "cuscuta-worker" -version = "0.1.5" +version = "0.1.6" dependencies = [ "anyhow", "axum", @@ -650,7 +650,7 @@ dependencies = [ [[package]] name = "cuscutactl" -version = "0.1.0" +version = "0.1.1" dependencies = [ "anyhow", "base64", diff --git a/Cargo.toml b/Cargo.toml index f0d1531..d8fcc04 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -10,7 +10,7 @@ members = [ resolver = "2" [workspace.package] -version = "0.1.5" +version = "0.1.6" [workspace.dependencies] anyhow = "1.0.102" diff --git a/cuscuta-common/src/db/log/event.rs b/cuscuta-common/src/db/log/event.rs index 17a21fe..caebaf9 100644 --- a/cuscuta-common/src/db/log/event.rs +++ b/cuscuta-common/src/db/log/event.rs @@ -1,5 +1,5 @@ use chrono::Utc; -use redis::Client; +use redis::{Client, TypedCommands}; use crate::db::{ log::{WorkerEvent, WorkerEventType}, @@ -35,6 +35,32 @@ pub fn write_event( Ok(()) } +/// 从事件列表读取事件 +/// +/// # Errors +/// 这个函数产生的错误来自Redis的错误[`redis::RedisError`]，以及读取内容不符合预期的错误 +/// +#[allow(clippy::cast_possible_wrap)] +pub fn read_events( + redis_client: &Client, + start: usize, + limit: usize, +) -> Result, Error> { + redis_client + .get_connection() + .map_err(Error::Redis)? + .lrange( + worker_event_redis_key(), + start as isize, + (start + limit - 1) as isize, + ) + .map_err(Error::Redis)? + .into_iter() + .map(|it| serde_json::from_str::(&it)) + .collect::, _>>() + .map_err(|e| Error::BadData(format!("failed to parse json to struct: ({e})"))) +} + /// 使用`REDIS_CLIENT`尝试向事件列表写入新事件 /// /// 很不卫生对吧，但是很好用 diff --git a/cuscuta-common/src/db/log/mod.rs b/cuscuta-common/src/db/log/mod.rs index 522c373..363b005 100644 --- a/cuscuta-common/src/db/log/mod.rs +++ b/cuscuta-common/src/db/log/mod.rs @@ -43,17 +43,18 @@ pub struct WorkerEvent { } /// Worker事件种类 -#[derive(Debug, Serialize, Deserialize)] +#[derive(Debug, Serialize, Deserialize, Clone, Copy)] +#[repr(u8)] pub enum WorkerEventType { /// 最严重的错误（例如Worker离线） - Fatal, + Fatal = 9, /// Worker一般信息 - Info, + Info = 1, /// Worker痕迹信息（例如拉取到任务，任务完成——会出现很多） - Trace, + Trace = 0, /// Worker警告（例如工作循环失败） - Warn, + Warn = 2, } diff --git a/cuscuta-common/src/db/log/status.rs b/cuscuta-common/src/db/log/status.rs index adf7f4e..6a3cf5f 100644 --- a/cuscuta-common/src/db/log/status.rs +++ b/cuscuta-common/src/db/log/status.rs @@ -1,7 +1,7 @@ use std::borrow::Cow; use chrono::Utc; -use redis::Client; +use redis::{Client, TypedCommands}; use crate::db::{ job::{Job, SubQueue}, @@ -38,3 +38,28 @@ pub fn update_worker_status( pipe.exec(&mut connection).map_err(Error::Redis)?; Ok(()) } + +/// 搜索Worker状态 +/// +/// # Errors +/// 这个函数产生的错误来自Redis的错误[`redis::RedisError`]，以及读取内容不符合预期的错误 +/// +pub fn search_worker_status( + redis_client: &Client, +) -> Result)>, Error> { + let mut connection = redis_client.get_connection().map_err(Error::Redis)?; + connection + .scan_match::<_, String>(worker_status_redis_key("*")) + .map_err(Error::Redis)? + .collect::, _>>() + .map_err(Error::Redis)? + .into_iter() + .map(|it| connection.get(&it).map(|r| (it, r))) + .collect::, _>>() + .map_err(Error::Redis)? + .into_iter() + .filter_map(|(key, option_value)| option_value.map(|it| (key, it))) + .map(|(k, v)| serde_json::from_str::(&v).map(|it| (k, it))) + .collect::, _>>() + .map_err(|e| Error::BadData(format!("failed to parse json to struct: ({e})"))) +} diff --git a/cuscuta-worker/src/worker/pending_friend.rs b/cuscuta-worker/src/worker/pending_friend.rs index f2997b7..a0264f9 100644 --- a/cuscuta-worker/src/worker/pending_friend.rs +++ b/cuscuta-worker/src/worker/pending_friend.rs @@ -1,3 +1,5 @@ +use std::collections::HashMap; + use cuscuta_common::{ api::{ self, @@ -29,7 +31,7 @@ pub async fn try_add_friends( friends: &mut Vec, ) -> Result<(), Error> { let mut connection = redis_client.get_connection().map_err(Error::Redis)?; - let ids: Vec<_> = jobs + let mut ids: HashMap<_, _> = jobs .iter() .filter_map(|it| { let (JobState::Pending { friend_info, .. } | JobState::Finished { friend_info, .. }) = @@ -44,10 +46,7 @@ pub async fn try_add_friends( let JobState::Pulled { start_timestamp } = job.state else { continue; }; - let option_existing_ids = ids - .iter() - .find(|(code, _)| code == &job.essential.friend_code); - if let Some((_, existing_friend_info)) = option_existing_ids { + if let Some(existing_friend_info) = ids.get(&job.essential.friend_code) { let friend_info = existing_friend_info.clone(); push_friend_info(&mut connection, &friend_info, job)?; job.state = JobState::Pending { @@ -59,7 +58,9 @@ pub async fn try_add_friends( continue; } let friends_new = - match resolve_friend(config, bundle_data, account_row, user_id, token, job).await { + match try_modify_remote_friend(config, bundle_data, account_row, user_id, token, job) + .await + { Ok(x) => x, Err(e) => { job.state = JobState::Failed { @@ -94,6 +95,7 @@ pub async fn try_add_friends( )); } }; + ids.insert(job.essential.friend_code.clone(), friend_add.clone()); *friends = friends_new; push_friend_info(&mut connection, &friend_add, job)?; job.essential.cursor_start = cursor.cast_signed() as i32; @@ -106,7 +108,7 @@ pub async fn try_add_friends( Ok(()) } -async fn resolve_friend( +async fn try_modify_remote_friend( config: &Config, bundle_data: &BundleData, account_row: &AccountRow, @@ -139,7 +141,24 @@ async fn resolve_friend( log::warn!( "pending_friends: friend is already exist but cache is out-of-date! trying readd" ); - delete_friend(config, bundle_data, account_row, user_id, token, job).await + xxxxxx_safe_call_ex( + config.worker_max_retry_count, + config.worker_exponential_backoff_base_millis, + config.worker_exponential_backoff_multiplier, + config.worker_exponential_backoff_max_delay_millis, + |it| it != StatusCode::TOO_MANY_REQUESTS, + || { + api::xxxxxx::api_delete_friend( + bundle_data, + &account_row.account_email, + user_id, + token, + &job.essential.friend_code, + ) + }, + ) + .await + .map(|it| it.friends) } else { log::warn!("pending_friends: failed to add friend: {e}: code: {code}"); Err(e) @@ -153,34 +172,6 @@ async fn resolve_friend( } } -async fn delete_friend( - config: &Config, - bundle_data: &BundleData, - account_row: &AccountRow, - user_id: &str, - token: &str, - job: &Job, -) -> Result, api::Error> { - xxxxxx_safe_call_ex( - config.worker_max_retry_count, - config.worker_exponential_backoff_base_millis, - config.worker_exponential_backoff_multiplier, - config.worker_exponential_backoff_max_delay_millis, - |it| it != StatusCode::TOO_MANY_REQUESTS, - || { - api::xxxxxx::api_delete_friend( - bundle_data, - &account_row.account_email, - user_id, - token, - &job.essential.friend_code, - ) - }, - ) - .await - .map(|it| it.friends) -} - fn push_friend_info( connection: &mut Connection, friend_info: &FriendInfo, diff --git a/cuscutactl/Cargo.toml b/cuscutactl/Cargo.toml index a3c0dfe..c73d461 100644 --- a/cuscutactl/Cargo.toml +++ b/cuscutactl/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "cuscutactl" -version = "0.1.0" +version = "0.1.1" edition = "2024" [dependencies] diff --git a/cuscutactl/src/command.rs b/cuscutactl/src/command.rs index 37cef0b..8a524de 100644 --- a/cuscutactl/src/command.rs +++ b/cuscutactl/src/command.rs @@ -1,4 +1,6 @@ -use clap::{Parser, Subcommand}; +use clap::{Parser, Subcommand, ValueEnum}; +use cuscuta_common::db::log::WorkerEventType; +use serde::{Deserialize, Serialize}; use crate::config::Config; @@ -31,6 +33,13 @@ pub enum SubCommands { #[command(subcommand)] command: SubCommandAccounts, }, + + /// Worker stat query + #[command(visible_alias = "stat")] + Stats { + #[command(subcommand)] + command: SubCommandStats, + }, } #[derive(Debug, Subcommand)] @@ -157,3 +166,40 @@ pub enum SubCommandAccountsRate { /// Query current rating Query, } + +#[derive(Debug, Subcommand)] +pub enum SubCommandStats { + /// Query worker state + Worker {}, + + /// Query worker event + Event { + #[arg(short, long, value_enum, default_value_t = ShowLevel::Info)] + show_level: ShowLevel, + + #[arg(long, short)] + limit: usize, + }, +} + +#[derive( + Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord, ValueEnum, Default, Serialize, Deserialize, +)] +pub enum ShowLevel { + Trace, + #[default] + Info, + Warn, + Fatal, +} + +impl From for WorkerEventType { + fn from(value: ShowLevel) -> Self { + match value { + ShowLevel::Trace => Self::Trace, + ShowLevel::Info => Self::Info, + ShowLevel::Warn => Self::Warn, + ShowLevel::Fatal => Self::Fatal, + } + } +} diff --git a/cuscutactl/src/main.rs b/cuscutactl/src/main.rs index 7cde665..f12180c 100644 --- a/cuscutactl/src/main.rs +++ b/cuscutactl/src/main.rs @@ -54,6 +54,7 @@ mod config; mod doctor; mod jobs; mod kube; +mod stats; mod url; trait Handler { @@ -140,6 +141,7 @@ async fn main() -> anyhow::Result<()> { Ok(()) } +#[allow(clippy::too_many_lines)] async fn run_command( cli: &Cli, pg_url: Result, @@ -228,6 +230,22 @@ async fn run_command( } } } + SubCommands::Stats { command } => { + let redis_url = match redis_url { + Ok(x) => x, + Err(e) => { + eprintln!("Configuration error: {e:?}"); + eprintln!("Hint: use --mode legacy --redis-url "); + bail!(e); + } + }; + match command { + command::SubCommandStats::Worker {} => stats::worker(&redis_url)?, + command::SubCommandStats::Event { show_level, limit } => { + stats::event(&redis_url, *show_level, *limit)?; + } + } + } } Ok(()) } diff --git a/cuscutactl/src/stats.rs b/cuscutactl/src/stats.rs new file mode 100644 index 0000000..f2b8385 --- /dev/null +++ b/cuscutactl/src/stats.rs @@ -0,0 +1,58 @@ +use chrono::{TimeZone, Utc}; +use cuscuta_common::db::log::{WorkerEventType, event::read_events, status::search_worker_status}; + +use crate::command::ShowLevel; + +/// Fetch worker status +pub fn worker(redis_url: &str) -> anyhow::Result<()> { + let client = redis::Client::open(redis_url)?; + let result = search_worker_status(&client)?; + for (k, stat) in result { + println!("- Worker: {k}"); + println!(" active_timestamp: {}", stat.last_active_timestamp); + println!(" cursor: {}", stat.cursor); + println!( + " sub_queue: {}", + stat.sub_queue + .as_ref() + .map_or("None", |sub_queue| &sub_queue.name) + ); + println!(" jobs:"); + for it in stat.jobs.iter() { + println!(" essential: {:?}", it.essential); + println!(" stat: {:?}", it.state); + println!(" id: {:?}", it.job_id); + } + println!(); + } + Ok(()) +} + +/// Fetch worker events +pub fn event(redis_url: &str, show_level: ShowLevel, limit: usize) -> anyhow::Result<()> { + let client = redis::Client::open(redis_url)?; + let mut out = Vec::new(); + let mut lines = 0; + loop { + let news = read_events(&client, lines, limit)?; + let len = news.len(); + if len == 0 { + break; + } + out.append( + &mut news + .into_iter() + .filter(|it| it.event_type as u8 >= WorkerEventType::from(show_level) as u8) + .collect::>(), + ); + lines += len; + } + for it in out { + let opt = Utc.timestamp_opt(it.timestamp / 1000, 0); + println!( + "{:?} {:?} {:?} {}", + opt, it.event_type, it.worker_id, it.message + ); + } + Ok(()) +}