mirror/Easytier
1
0
Fork 0
mirror of https://github.com/EasyTier/EasyTier.git synced 2026-08-08 05:29:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

Revert "fix: eliminate unsafe code in packet construction and stats counters"

Browse Source 
This reverts commit 3464cb801a.
This commit is contained in:
fanyang
2026-06-28 14:16:47 +08:00
parent 3464cb801a
commit d4ef9decd8
12 changed files with 275 additions and 712 deletions
Download Patch File Download Diff File Expand all files Collapse all files
easytier/src/common/acl_processor.rs
+68 -100
View File
@@ -3,8 +3,7 @@ use std::{
net::{IpAddr, SocketAddr},
str::FromStr as _,
sync::Arc,
sync::atomic::{AtomicU64, Ordering::Relaxed},
time::Duration,
time::{Duration, SystemTime, UNIX_EPOCH},
};
use hotpath::instant::Instant;
@@ -77,41 +76,7 @@ pub struct FastLookupRule {
pub stateful: bool,
pub rate_limit: u32,
pub burst_limit: u32,
pub rule_stats: Arc<RuleStatsTracker>,
}
#[derive(Debug)]
pub struct RuleStatsTracker {
rule: Option<Rule>,
packets: AtomicU64,
bytes: AtomicU64,
}
impl RuleStatsTracker {
fn new(rule: Option<Rule>) -> Self {
Self {
rule,
packets: AtomicU64::new(0),
bytes: AtomicU64::new(0),
}
}
#[inline]
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "RuleStatsTracker"))]
fn increment(&self, packet_size: usize) {
self.packets.fetch_add(1, Relaxed);
self.bytes.fetch_add(packet_size as u64, Relaxed);
}
fn snapshot(&self) -> RuleStats {
RuleStats {
rule: self.rule.clone(),
stat: Some(StatItem {
packet_count: self.packets.load(Relaxed),
byte_count: self.bytes.load(Relaxed),
}),
}
}
pub rule_stats: Arc<RuleStats>,
}
// Cache key combining packet info and chain type
@@ -143,17 +108,17 @@ impl AclCacheKey {
}
// Cache entry with timestamp for LRU cleanup
#[derive(Debug)]
#[derive(Debug, Clone)]
pub(crate) struct AclCacheEntry {
pub action: Action,
pub matched_rule: RuleId,
pub last_access: AtomicU64,
pub last_access: Instant,
// New fields to track rule characteristics for proper cache behavior
pub conn_track_key: Option<String>,
pub rate_limit_keys: Vec<RateLimitKey>,
pub chain_type: ChainType,
pub acl_result: Option<AclResult>,
pub rule_stats_vec: Vec<Arc<RuleStatsTracker>>,
pub rule_stats_vec: Vec<Arc<RuleStats>>,
}
// Packet info extracted for ACL processing
@@ -246,7 +211,7 @@ pub struct AclProcessor {
default_outbound_action: Action,
default_forward_action: Action,
default_rule_stats: Arc<RuleStatsTracker>,
default_rule_stats: Arc<RuleStats>,
// Connection tracking table - shared across different processor instances if needed
conn_track: Arc<DashMap<String, ConnTrackEntry>>,
@@ -259,13 +224,6 @@ pub struct AclProcessor {
cache_max_size: usize,
cache_cleanup_interval: Duration,
// Coarse monotonic timestamp updated by the cleanup task, used to avoid
// calling Instant::now() on every cache hit.
coarse_millis: Arc<AtomicU64>,
// Hot-path counters that bypass the DashMap stats table
cache_hits: AtomicU64,
// Statistics
stats: Arc<DashMap<AclStatKey, u64>>,
@@ -301,14 +259,18 @@ impl AclProcessor {
default_outbound_action,
default_forward_action,
default_rule_stats: Arc::new(RuleStatsTracker::new(None)),
default_rule_stats: Arc::new(RuleStats {
rule: None,
stat: Some(StatItem {
packet_count: 0,
byte_count: 0,
}),
}),
conn_track: conn_track.unwrap_or_else(|| Arc::new(DashMap::new())),
rate_limiters: rate_limiters.unwrap_or_else(|| Arc::new(DashMap::new())),
rule_cache: Arc::new(DashMap::new()), // Always start with fresh cache
cache_max_size: 1024, // Limit cache to 1k entries
cache_cleanup_interval: Duration::from_secs(20), // Cleanup every 5 minutes
coarse_millis: Arc::new(AtomicU64::new(0)),
cache_hits: AtomicU64::new(0),
stats: stats.unwrap_or_else(|| Arc::new(DashMap::new())),
tasks,
};
@@ -412,14 +374,11 @@ impl AclProcessor {
let rule_cache = self.rule_cache.clone();
let cache_max_size = self.cache_max_size;
let cleanup_interval = self.cache_cleanup_interval;
let coarse_millis = self.coarse_millis.clone();
self.tasks.spawn(async move {
let mut interval = tokio::time::interval(cleanup_interval);
loop {
interval.tick().await;
let now = crate::common::stats_manager::now_monotonic_millis();
coarse_millis.store(now, Relaxed);
Self::cleanup_cache(&rule_cache, cache_max_size);
rule_cache.shrink_to_fit();
@@ -442,9 +401,10 @@ impl AclProcessor {
/// Clean up cache using LRU strategy
fn cleanup_cache(cache: &DashMap<AclCacheKey, AclCacheEntry>, max_size: usize) {
// remove cache not be used in last 15 second
let now = crate::common::stats_manager::now_monotonic_millis();
let cutoff = now.saturating_sub(15_000);
cache.retain(|_, entry| entry.last_access.load(Relaxed) > cutoff);
let expired_timepoint = Instant::now()
.checked_sub(Duration::from_secs(15))
.unwrap_or(Instant::now());
cache.retain(|_, entry| entry.last_access > expired_timepoint);
let current_size = cache.len();
if current_size <= max_size {
@@ -452,9 +412,9 @@ impl AclProcessor {
}
// Remove oldest entries (LRU cleanup)
let mut entries: Vec<(AclCacheKey, u64)> = cache
let mut entries: Vec<(AclCacheKey, Instant)> = cache
.iter()
.map(|entry| (entry.key().clone(), entry.value().last_access.load(Relaxed)))
.map(|entry| (entry.key().clone(), entry.value().last_access))
.collect();
// Sort by last_access (oldest first)
@@ -473,7 +433,6 @@ impl AclProcessor {
);
}
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "AclProcessor"))]
pub(crate) fn process_packet_with_cache_entry(
&self,
packet_info: &PacketInfo,
@@ -500,41 +459,42 @@ impl AclProcessor {
cache_entry.acl_result.clone().unwrap()
}
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "AclProcessor"))]
fn inc_cache_entry_stats(&self, cache_entry: &AclCacheEntry, packet_info: &PacketInfo) {
for rule_stats in cache_entry.rule_stats_vec.iter() {
rule_stats.increment(packet_info.packet_size);
// Use unsafe code to mutate the contents behind the Arc
let stat_ptr = rule_stats.stat.as_ref().unwrap() as *const StatItem as *mut StatItem;
unsafe {
(*stat_ptr).packet_count += 1;
(*stat_ptr).byte_count += packet_info.packet_size as u64;
}
}
}
pub fn get_rules_stats(&self) -> Vec<RuleStats> {
let mut stats: Vec<RuleStats> = Vec::new();
for rule in self.inbound_rules.iter() {
stats.push(rule.rule_stats.snapshot());
stats.push((*rule.rule_stats).clone());
}
for rule in self.outbound_rules.iter() {
stats.push(rule.rule_stats.snapshot());
stats.push((*rule.rule_stats).clone());
}
for rule in self.forward_rules.iter() {
stats.push(rule.rule_stats.snapshot());
stats.push((*rule.rule_stats).clone());
}
stats
}
/// Process a packet through ACL rules - Now lock-free!
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "AclProcessor"))]
pub fn process_packet(&self, packet_info: &PacketInfo, chain_type: ChainType) -> AclResult {
// Check cache first for performance
let cache_key = AclCacheKey::from_packet_info(packet_info, chain_type);
// If cache hit and can skip checks, return cached result.
// Use get() (read lock) instead of get_mut() (write lock) and update
// last_access via AtomicU64, avoiding expensive Instant::now().
if let Some(cached) = self.rule_cache.get(&cache_key) {
cached
.last_access
.store(self.coarse_millis.load(Relaxed), Relaxed);
self.cache_hits.fetch_add(1, Relaxed);
// If cache hit and can skip checks, return cached result
if let Some(mut cached) = self.rule_cache.get_mut(&cache_key) {
// Update last access time for LRU
cached.last_access = Instant::now();
self.increment_stat(AclStatKey::CacheHits);
return self.process_packet_with_cache_entry(packet_info, &cached);
}
@@ -556,7 +516,7 @@ impl AclProcessor {
let mut cache_entry = AclCacheEntry {
action: Action::Allow,
matched_rule: RuleId::Default,
last_access: AtomicU64::new(self.coarse_millis.load(Relaxed)),
last_access: Instant::now(),
conn_track_key: None,
rate_limit_keys: vec![],
chain_type,
@@ -621,9 +581,8 @@ impl AclProcessor {
// Cache the result with rule info
self.increment_stat(AclStatKey::RuleMatches);
self.inc_cache_entry_stats(&cache_entry, packet_info);
let result = cache_entry.acl_result.clone().unwrap();
self.cache_result(&cache_key, cache_entry);
return result;
self.cache_result(&cache_key, cache_entry.clone());
return cache_entry.acl_result.clone().unwrap();
}
let default_action = match chain_type {
@@ -659,9 +618,8 @@ impl AclProcessor {
// Cache the default result (no rule info)
self.inc_cache_entry_stats(&cache_entry, packet_info);
let result = cache_entry.acl_result.clone().unwrap();
self.cache_result(&cache_key, cache_entry);
result
self.cache_result(&cache_key, cache_entry.clone());
cache_entry.acl_result.clone().unwrap()
}
/// Get shared state for preserving across hot reloads
@@ -785,11 +743,13 @@ impl AclProcessor {
/// Check connection state for stateful rules
fn check_connection_state(&self, conn_track_key: &str, packet_info: &PacketInfo) {
let now = current_unix_secs();
self.conn_track
.entry(conn_track_key.to_string())
.and_modify(|x| {
x.last_seen = now;
x.last_seen = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_secs();
x.packet_count += 1;
x.byte_count += packet_info.packet_size as u64;
x.state = ConnState::Established as i32;
@@ -803,8 +763,14 @@ impl AclProcessor {
),
protocol: packet_info.protocol as i32,
state: ConnState::New as i32,
created_at: now,
last_seen: now,
created_at: SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_secs(),
last_seen: SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_secs(),
packet_count: 1,
byte_count: packet_info.packet_size as u64,
});
@@ -896,7 +862,13 @@ impl AclProcessor {
stateful: rule.stateful,
rate_limit: rule.rate_limit,
burst_limit: rule.burst_limit,
rule_stats: Arc::new(RuleStatsTracker::new(Some(rule.clone()))),
rule_stats: Arc::new(RuleStats {
rule: Some(rule.clone()),
stat: Some(StatItem {
packet_count: 0,
byte_count: 0,
}),
}),
}
}
@@ -922,10 +894,6 @@ impl AclProcessor {
.collect::<HashMap<_, _>>();
// Add cache statistics using enum keys
stats.insert(
AclStatKey::CacheHits.as_str(),
self.cache_hits.load(Relaxed),
);
stats.insert(AclStatKey::CacheSize.as_str(), self.rule_cache.len() as u64);
stats.insert(
AclStatKey::CacheMaxSize.as_str(),
@@ -940,11 +908,14 @@ impl AclProcessor {
conn_track: Arc<DashMap<String, ConnTrackEntry>>,
timeout_secs: u64,
) {
let current_time = current_unix_secs();
let current_time = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_secs();
let keys_to_remove: Vec<String> = conn_track
.iter()
.filter_map(|entry| {
if current_time.saturating_sub(entry.last_seen) > timeout_secs {
if current_time - entry.last_seen > timeout_secs {
Some(entry.key().clone())
} else {
None
@@ -959,7 +930,11 @@ impl AclProcessor {
/// Get cache hit rate
pub fn get_cache_hit_rate(&self) -> f64 {
let cache_hits = self.cache_hits.load(Relaxed);
let cache_hits = self
.stats
.get(&AclStatKey::CacheHits)
.map(|v| *v.value())
.unwrap_or(0);
let total_requests = cache_hits
+ self
.stats
@@ -975,13 +950,6 @@ impl AclProcessor {
}
}
fn current_unix_secs() -> u64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_secs()
}
// 新增辅助函数
fn parse_port_start(port_strs: &[String]) -> Option<u16> {
port_strs
easytier/src/common/mod.rs
-1
View File
@@ -22,7 +22,6 @@ pub mod machine_id;
pub mod netns;
pub mod network;
pub mod os_info;
pub mod sharded_counter;
pub mod stats_manager;
pub mod stun;
pub mod stun_codec_ext;
easytier/src/common/sharded_counter.rs
-110
View File
@@ -1,110 +0,0 @@
use std::sync::atomic::{AtomicU64, Ordering::Relaxed};
const PUBLISH_INTERVAL: u64 = 256;
/// Counter optimized for long-lived worker threads.
///
/// Pending values below `PUBLISH_INTERVAL` live in `ThreadLocal` shards. The
/// shards are retained until this counter is dropped, so this is intended for
/// tokio workers or similarly long-lived threads rather than high-churn threads.
pub struct ShardedCounter {
published: AtomicU64,
locals: thread_local::ThreadLocal<AtomicU64>,
}
impl ShardedCounter {
pub fn new() -> Self {
Self {
published: AtomicU64::new(0),
locals: thread_local::ThreadLocal::new(),
}
}
#[inline]
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "ShardedCounter"))]
pub fn add(&self, delta: u64) {
let local = self.locals.get_or(|| AtomicU64::new(0));
let v = local.load(Relaxed).saturating_add(delta);
local.store(v, Relaxed);
if v >= PUBLISH_INTERVAL {
let pending = local.swap(0, Relaxed);
if pending > 0 {
self.published.fetch_add(pending, Relaxed);
}
}
}
#[inline]
pub fn inc(&self) {
self.add(1);
}
pub fn get(&self) -> u64 {
self.locals
.iter()
.fold(self.published.load(Relaxed), |total, local| {
total.saturating_add(local.load(Relaxed))
})
}
pub fn reset(&self) {
self.published.store(0, Relaxed);
for local in self.locals.iter() {
local.store(0, Relaxed);
}
}
}
impl Default for ShardedCounter {
fn default() -> Self {
Self::new()
}
}
impl std::fmt::Debug for ShardedCounter {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ShardedCounter")
.field("value", &self.get())
.finish()
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::{sync::Arc, thread};
#[test]
fn sharded_counter_get_includes_other_thread_locals() {
let counter = Arc::new(ShardedCounter::new());
let thread_counter = Arc::clone(&counter);
thread::spawn(move || {
for _ in 0..10 {
thread_counter.inc();
}
})
.join()
.unwrap();
assert_eq!(counter.get(), 10);
}
#[test]
fn sharded_counter_reset_clears_other_thread_locals() {
let counter = Arc::new(ShardedCounter::new());
let thread_counter = Arc::clone(&counter);
thread::spawn(move || {
for _ in 0..10 {
thread_counter.inc();
}
})
.join()
.unwrap();
counter.reset();
assert_eq!(counter.get(), 0);
}
}
easytier/src/common/stats_manager.rs
+150 -55
View File
@@ -1,16 +1,13 @@
use crate::common::sharded_counter::ShardedCounter;
use dashmap::DashMap;
use hotpath::instant::Instant;
use serde::{Deserialize, Serialize};
use std::cell::UnsafeCell;
use std::fmt;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, LazyLock};
use std::sync::Arc;
use std::time::Duration;
use tokio::time::interval;
use tokio_util::task::AbortOnDropHandle;
static START_INSTANT: LazyLock<Instant> = LazyLock::new(Instant::now);
/// Predefined metric names for type safety
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub enum MetricName {
@@ -378,10 +375,10 @@ impl Default for LabelSet {
}
}
/// High-performance counter backed by sharded thread-local accumulation.
/// UnsafeCounter provides a high-performance counter using UnsafeCell
#[derive(Debug)]
pub struct UnsafeCounter {
inner: ShardedCounter,
value: UnsafeCell<u64>,
}
impl Default for UnsafeCounter {
@@ -393,56 +390,121 @@ impl Default for UnsafeCounter {
impl UnsafeCounter {
pub fn new() -> Self {
Self {
inner: ShardedCounter::new(),
value: UnsafeCell::new(0),
}
}
pub fn add(&self, delta: u64) {
self.inner.add(delta);
pub fn new_with_value(initial: u64) -> Self {
Self {
value: UnsafeCell::new(initial),
}
}
pub fn inc(&self) {
self.inner.inc();
/// Increment the counter by the given amount
/// # Safety
/// This method is unsafe because it uses UnsafeCell. The caller must ensure
/// that no other thread is accessing this counter simultaneously.
pub unsafe fn add(&self, delta: u64) {
let ptr = self.value.get();
unsafe {
*ptr = (*ptr).saturating_add(delta);
}
}
pub fn get(&self) -> u64 {
self.inner.get()
/// Increment the counter by 1
/// # Safety
/// This method is unsafe because it uses UnsafeCell. The caller must ensure
/// that no other thread is accessing this counter simultaneously.
pub unsafe fn inc(&self) {
unsafe {
self.add(1);
}
}
pub fn reset(&self) {
self.inner.reset();
/// Get the current value of the counter
/// # Safety
/// This method is unsafe because it uses UnsafeCell. The caller must ensure
/// that no other thread is modifying this counter simultaneously.
pub unsafe fn get(&self) -> u64 {
let ptr = self.value.get();
unsafe { *ptr }
}
/// Reset the counter to zero
/// # Safety
/// This method is unsafe because it uses UnsafeCell. The caller must ensure
/// that no other thread is accessing this counter simultaneously.
pub unsafe fn reset(&self) {
let ptr = self.value.get();
unsafe {
*ptr = 0;
}
}
/// Set the counter to a specific value
/// # Safety
/// This method is unsafe because it uses UnsafeCell. The caller must ensure
/// that no other thread is accessing this counter simultaneously.
pub unsafe fn set(&self, value: u64) {
let ptr = self.value.get();
unsafe {
*ptr = value;
}
}
}
// UnsafeCounter is Send + Sync because the safety is guaranteed by the caller
unsafe impl Send for UnsafeCounter {}
unsafe impl Sync for UnsafeCounter {}
/// MetricData contains both the counter and last update timestamp
/// Uses UnsafeCell for lock-free access
#[derive(Debug)]
struct MetricData {
counter: UnsafeCounter,
last_updated: AtomicU64,
}
pub(crate) fn now_monotonic_millis() -> u64 {
Instant::now().duration_since(*START_INSTANT).as_millis() as u64
last_updated: UnsafeCell<Instant>,
}
impl MetricData {
fn new() -> Self {
Self {
counter: UnsafeCounter::new(),
last_updated: AtomicU64::new(now_monotonic_millis()),
last_updated: UnsafeCell::new(Instant::now()),
}
}
fn touch(&self) {
self.last_updated
.store(now_monotonic_millis(), Ordering::Relaxed);
fn new_with_value(initial: u64) -> Self {
Self {
counter: UnsafeCounter::new_with_value(initial),
last_updated: UnsafeCell::new(Instant::now()),
}
}
fn get_last_updated(&self) -> u64 {
self.last_updated.load(Ordering::Relaxed)
/// Update the last_updated timestamp
/// # Safety
/// This method is unsafe because it uses UnsafeCell. The caller must ensure
/// that no other thread is accessing this timestamp simultaneously.
unsafe fn touch(&self) {
let ptr = self.last_updated.get();
unsafe {
*ptr = Instant::now();
}
}
/// Get the last updated timestamp
/// # Safety
/// This method is unsafe because it uses UnsafeCell. The caller must ensure
/// that no other thread is modifying this timestamp simultaneously.
unsafe fn get_last_updated(&self) -> Instant {
let ptr = self.last_updated.get();
unsafe { *ptr }
}
}
// MetricData is Send + Sync because the safety is guaranteed by the caller
unsafe impl Send for MetricData {}
unsafe impl Sync for MetricData {}
/// MetricKey uniquely identifies a metric with its name and labels
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
struct MetricKey {
@@ -483,23 +545,41 @@ impl CounterHandle {
}
}
/// Increment the counter by the given amount
pub fn add(&self, delta: u64) {
self.metric_data.counter.add(delta);
self.metric_data.touch();
unsafe {
self.metric_data.counter.add(delta);
self.metric_data.touch();
}
}
/// Increment the counter by 1
pub fn inc(&self) {
self.metric_data.counter.inc();
self.metric_data.touch();
unsafe {
self.metric_data.counter.inc();
self.metric_data.touch();
}
}
/// Get the current value of the counter
pub fn get(&self) -> u64 {
self.metric_data.counter.get()
unsafe { self.metric_data.counter.get() }
}
/// Reset the counter to zero
pub fn reset(&self) {
self.metric_data.counter.reset();
self.metric_data.touch();
unsafe {
self.metric_data.counter.reset();
self.metric_data.touch();
}
}
/// Set the counter to a specific value
pub fn set(&self, value: u64) {
unsafe {
self.metric_data.counter.set(value);
self.metric_data.touch();
}
}
}
@@ -535,7 +615,9 @@ impl StatsManager {
loop {
interval.tick().await;
let cutoff_millis = now_monotonic_millis().saturating_sub(180_000);
let Some(cutoff_time) = Instant::now().checked_sub(Duration::from_secs(180)) else {
continue;
};
let Some(counters) = counters_clone.upgrade() else {
break;
@@ -543,7 +625,7 @@ impl StatsManager {
counters.retain(|_, metric_data: &mut Arc<MetricData>| {
Arc::strong_count(metric_data) > 1
|| metric_data.get_last_updated() >= cutoff_millis
|| unsafe { metric_data.get_last_updated() > cutoff_time }
});
counters.shrink_to_fit();
}
@@ -581,7 +663,7 @@ impl StatsManager {
let key = entry.key();
let metric_data = entry.value();
let value = metric_data.counter.get();
let value = unsafe { metric_data.counter.get() };
metrics.push(MetricSnapshot {
name: key.name,
@@ -614,7 +696,7 @@ impl StatsManager {
let key = MetricKey::new(name, labels.clone());
if let Some(metric_data) = self.counters.get(&key) {
let value = metric_data.counter.get();
let value = unsafe { metric_data.counter.get() };
Some(MetricSnapshot {
name,
labels: labels.clone(),
@@ -715,9 +797,17 @@ mod tests {
async fn test_unsafe_counter() {
let counter = UnsafeCounter::new();
assert_eq!(counter.get(), 0);
counter.add(256);
assert_eq!(counter.get(), 256);
unsafe {
assert_eq!(counter.get(), 0);
counter.inc();
assert_eq!(counter.get(), 1);
counter.add(5);
assert_eq!(counter.get(), 6);
counter.set(10);
assert_eq!(counter.get(), 10);
counter.reset();
assert_eq!(counter.get(), 0);
}
}
#[tokio::test]
@@ -763,11 +853,11 @@ mod tests {
let stats = StatsManager::new();
let counter1 = stats.get_simple_counter(MetricName::TrafficBytesTx);
counter1.add(100);
counter1.set(100);
let labels = LabelSet::new().with_label("status", "success");
let counter2 = stats.get_counter(MetricName::PeerRpcClientTx, labels);
counter2.add(50);
counter2.set(50);
let traffic_labels = LabelSet::new()
.with_label_type(LabelType::NetworkName("default".to_string()))
@@ -775,7 +865,7 @@ mod tests {
"87ede5a2-9c3d-492d-9bbe-989b9d07e742".to_string(),
));
let counter3 = stats.get_counter(MetricName::TrafficBytesTxByInstance, traffic_labels);
counter3.add(25);
counter3.set(25);
let prometheus_output = stats.export_prometheus();
@@ -795,7 +885,7 @@ mod tests {
let labels = LabelSet::new().with_label("peer", "test");
let counter = stats.get_counter(MetricName::PeerRpcClientTx, labels.clone());
counter.add(42);
counter.set(42);
let metric = stats
.get_metric(MetricName::PeerRpcClientTx, &labels)
@@ -812,11 +902,11 @@ mod tests {
stats
.get_simple_counter(MetricName::PeerRpcClientTx)
.add(10);
stats.get_simple_counter(MetricName::PeerRpcErrors).add(2);
.set(10);
stats.get_simple_counter(MetricName::PeerRpcErrors).set(2);
stats
.get_simple_counter(MetricName::TrafficBytesTx)
.add(100);
.set(100);
let rpc_metrics = stats.get_metrics_by_prefix("peer_rpc");
assert_eq!(rpc_metrics.len(), 2);
@@ -831,15 +921,19 @@ mod tests {
// 创建一些计数器
let counter1 = stats.get_simple_counter(MetricName::PeerRpcClientTx);
counter1.add(10);
counter1.set(10);
let labels = LabelSet::new().with_label("test", "value");
let counter2 = stats.get_counter(MetricName::TrafficBytesTx, labels);
counter2.add(20);
counter2.set(20);
// 验证计数器存在
assert_eq!(stats.metric_count(), 2);
// 注意：实际的清理测试需要等待3分钟，这在单元测试中不现实
// 这里我们只验证清理机制的基本结构是否正确
// 清理逻辑在后台线程中运行，会自动删除超过3分钟未更新的条目
// 验证计数器仍然可以正常工作
counter1.inc();
assert_eq!(counter1.get(), 11);
@@ -852,14 +946,14 @@ mod tests {
async fn test_cleanup_keeps_metrics_with_live_handles() {
let stats = StatsManager::new();
let counter = stats.get_simple_counter(MetricName::TrafficBytesForwarded);
counter.add(1);
counter.set(1);
// Use a future cutoff so last_updated check always fails
let future_cutoff = now_monotonic_millis() + 1000;
let cutoff_time = Instant::now().checked_add(Duration::from_secs(1)).unwrap();
stats
.counters
.retain(|_, metric_data: &mut Arc<MetricData>| {
Arc::strong_count(metric_data) > 1 || metric_data.get_last_updated() > future_cutoff
Arc::strong_count(metric_data) > 1
|| unsafe { metric_data.get_last_updated() > cutoff_time }
});
assert_eq!(stats.metric_count(), 1);
@@ -869,7 +963,8 @@ mod tests {
stats
.counters
.retain(|_, metric_data: &mut Arc<MetricData>| {
Arc::strong_count(metric_data) > 1 || metric_data.get_last_updated() > future_cutoff
Arc::strong_count(metric_data) > 1
|| unsafe { metric_data.get_last_updated() > cutoff_time }
});
assert_eq!(stats.metric_count(), 0);
}