Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
72 changes: 36 additions & 36 deletions crates/client/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -26,12 +26,12 @@
//! // Get queue statistics for all namespaces
//! let all_stats = admin.all_stats().await?;
//! for stats in all_stats {
//! println!("{}: queue={}, dead={}", stats.namespace, stats.queue, stats.dead);
//! println!("{}: queue={}, dead={}", stats.namespace, stats.queue_len, stats.dead_len);
//! }
//!
//! // Get stats for a specific namespace
//! let stats = admin.stats("myapp").await?;
//! println!("Queue: {}, Dead: {}", stats.queue, stats.dead);
//! println!("Queue: {}, Dead: {}", stats.queue_len, stats.dead_len);
//!
//! // List worker pools
//! let workers = admin.list_workers().await?;
Expand All @@ -53,13 +53,13 @@ pub use wg_core::{Job, JobId, JobOptions, JobStatus, Result, WgError};
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct QueueStats {
/// Number of jobs in the immediate queue.
pub queue: usize,
pub queue_len: usize,
/// Number of jobs scheduled for future execution.
pub scheduled: usize,
pub schedule_len: usize,
/// Number of jobs waiting for retry.
pub retry: usize,
pub retry_len: usize,
/// Number of dead (failed) jobs.
pub dead: usize,
pub dead_len: usize,
}

/// Queue statistics with namespace info.
Expand All @@ -68,13 +68,13 @@ pub struct NamespaceQueueStats {
/// The namespace name.
pub namespace: String,
/// Number of jobs in the immediate queue.
pub queue: usize,
pub queue_len: usize,
/// Number of jobs scheduled for future execution.
pub scheduled: usize,
pub schedule_len: usize,
/// Number of jobs waiting for retry.
pub retry: usize,
pub retry_len: usize,
/// Number of dead (failed) jobs.
pub dead: usize,
pub dead_len: usize,
}

/// A parsed dead job with its metadata.
Expand Down Expand Up @@ -145,7 +145,7 @@ impl<B: Backend + Clone> AdminClient<B> {
let mut all_stats = Vec::with_capacity(namespaces.len());

for ns in namespaces {
let (queue, scheduled, retry, dead) = tokio::try_join!(
let (queue_len, schedule_len, retry_len, dead_len) = tokio::try_join!(
self.backend.queue_len(&ns),
self.backend.schedule_len(&ns),
self.backend.retry_len(&ns),
Expand All @@ -154,10 +154,10 @@ impl<B: Backend + Clone> AdminClient<B> {

all_stats.push(NamespaceQueueStats {
namespace: ns,
queue,
scheduled,
retry,
dead,
queue_len,
schedule_len,
retry_len,
dead_len,
});
}

Expand All @@ -168,18 +168,18 @@ impl<B: Backend + Clone> AdminClient<B> {

/// Get all queue statistics for a namespace.
pub async fn stats(&self, ns: &str) -> Result<QueueStats> {
let (queue, scheduled, retry, dead) = tokio::try_join!(
let (queue_len, schedule_len, retry_len, dead_len) = tokio::try_join!(
self.backend.queue_len(ns),
self.backend.schedule_len(ns),
self.backend.retry_len(ns),
self.backend.dead_len(ns),
)?;

Ok(QueueStats {
queue,
scheduled,
retry,
dead,
queue_len,
schedule_len,
retry_len,
dead_len,
})
}

Expand Down Expand Up @@ -344,34 +344,34 @@ mod tests {
#[test]
fn test_queue_stats_default() {
let stats = QueueStats {
queue: 10,
scheduled: 5,
retry: 2,
dead: 1,
queue_len: 10,
schedule_len: 5,
retry_len: 2,
dead_len: 1,
};

assert_eq!(stats.queue, 10);
assert_eq!(stats.scheduled, 5);
assert_eq!(stats.retry, 2);
assert_eq!(stats.dead, 1);
assert_eq!(stats.queue_len, 10);
assert_eq!(stats.schedule_len, 5);
assert_eq!(stats.retry_len, 2);
assert_eq!(stats.dead_len, 1);
}

#[test]
fn test_queue_stats_serialization() {
let stats = QueueStats {
queue: 10,
scheduled: 5,
retry: 2,
dead: 1,
queue_len: 10,
schedule_len: 5,
retry_len: 2,
dead_len: 1,
};

let json = serde_json::to_string(&stats).unwrap();
let parsed: QueueStats = serde_json::from_str(&json).unwrap();

assert_eq!(parsed.queue, stats.queue);
assert_eq!(parsed.scheduled, stats.scheduled);
assert_eq!(parsed.retry, stats.retry);
assert_eq!(parsed.dead, stats.dead);
assert_eq!(parsed.queue_len, stats.queue_len);
assert_eq!(parsed.schedule_len, stats.schedule_len);
assert_eq!(parsed.retry_len, stats.retry_len);
assert_eq!(parsed.dead_len, stats.dead_len);
}

#[test]
Expand Down
54 changes: 0 additions & 54 deletions crates/core/client.rs
Original file line number Diff line number Diff line change
Expand Up @@ -156,62 +156,8 @@ impl<B: Backend + Clone> Client<B> {
self.backend.dead_len(&self.namespace).await
}

/// List all namespaces that have any data in the backend.
///
/// This is useful for admin/monitoring tools to discover all active namespaces.
pub async fn list_namespaces(&self) -> Result<Vec<String>> {
self.backend.list_namespaces().await
}

/// Get statistics for the current namespace.
pub async fn stats(&self) -> Result<NamespaceStats> {
Ok(NamespaceStats {
namespace: self.namespace.clone(),
queue_len: self.queue_len().await?,
schedule_len: self.schedule_len().await?,
retry_len: self.retry_len().await?,
dead_len: self.dead_len().await?,
})
}

/// Get statistics for all namespaces.
///
/// Returns stats for every namespace that has data in the backend.
pub async fn all_stats(&self) -> Result<Vec<NamespaceStats>> {
let namespaces = self.list_namespaces().await?;
let mut all_stats = Vec::with_capacity(namespaces.len());

for ns in namespaces {
let stats = NamespaceStats {
namespace: ns.clone(),
queue_len: self.backend.queue_len(&ns).await?,
schedule_len: self.backend.schedule_len(&ns).await?,
retry_len: self.backend.retry_len(&ns).await?,
dead_len: self.backend.dead_len(&ns).await?,
};
all_stats.push(stats);
}

Ok(all_stats)
}

/// Get a reference to the underlying backend.
pub fn backend(&self) -> &B {
&self.backend
}
}

/// Statistics for a namespace.
#[derive(Debug, Clone)]
pub struct NamespaceStats {
/// The namespace name.
pub namespace: String,
/// Number of jobs in the immediate queue.
pub queue_len: usize,
/// Number of jobs in the schedule queue.
pub schedule_len: usize,
/// Number of jobs in the retry queue.
pub retry_len: usize,
/// Number of jobs in the dead letter queue.
pub dead_len: usize,
}
2 changes: 1 addition & 1 deletion crates/core/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ mod worker;

// Re-export main types
pub use backend::{Backend, DynBackend, SharedBackend, WorkerPoolInfo};
pub use client::{Client, NamespaceStats};
pub use client::Client;
pub use config::{ClientConfig, WorkerConfig, WorkerConfigBuilder};
pub use error::{Result, WgError};
pub use heartbeat::{generate_pool_id, Heartbeater};
Expand Down