gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-10-13 08:02:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(obs): export bucket gauges from live snapshots (#8399)

Browse Source 
* fix(obs): export bucket gauges from live snapshots

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* fix(obs): collapse snapshot update condition

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

---------

Co-authored-by: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
Hauser
2026-10-08 02:12:48 +08:00
committed by GitHub
parent 04bc5dffd6
commit df988cfd86
4 changed files with 149 additions and 247 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/obs/src/metrics/scheduler.rs
+35 -201
View File
@@ -110,9 +110,7 @@ use crate::metrics::schema::bucket_replication::{
use crate::metrics::schema::cluster_usage::{
BUCKET_LABEL as USAGE_BUCKET_LABEL, RANGE_LABEL as USAGE_RANGE_LABEL, USAGE_BUCKET_DELETE_MARKERS_COUNT_MD,
USAGE_BUCKET_OBJECT_SIZE_DISTRIBUTION_MD, USAGE_BUCKET_OBJECT_VERSION_COUNT_DISTRIBUTION_MD, USAGE_BUCKET_OBJECTS_TOTAL_MD,
USAGE_BUCKET_QUOTA_TOTAL_BYTES_MD, USAGE_BUCKET_TOTAL_BYTES_MD, USAGE_BUCKET_VERSIONS_COUNT_MD, USAGE_BUCKETS_COUNT_MD,
USAGE_DELETE_MARKERS_COUNT_MD, USAGE_OBJECTS_COUNT_MD, USAGE_OBJECTS_DISTRIBUTION_MD, USAGE_SINCE_LAST_UPDATE_SECONDS_MD,
USAGE_TOTAL_BYTES_MD, USAGE_VERSIONS_COUNT_MD, USAGE_VERSIONS_DISTRIBUTION_MD,
USAGE_BUCKET_QUOTA_TOTAL_BYTES_MD, USAGE_BUCKET_TOTAL_BYTES_MD, USAGE_BUCKET_VERSIONS_COUNT_MD,
};
use crate::metrics::schema::node_bucket::{BUCKET_OBJECTS_TOTAL_MD, BUCKET_QUOTA_BYTES_MD, BUCKET_USAGE_BYTES_MD};
use crate::metrics::schema::notification_target::{
@@ -150,7 +148,7 @@ use crate::metrics::stats_collector::{
};
use crate::metrics::storage_snapshot::StorageSnapshotMetrics;
use crate::node_identity::{SERVER_LABEL, current_local_node_identity};
use crate::telemetry::retire_metric_series;
use crate::telemetry::{process_global_meter, retire_metric_series};
use futures_util::FutureExt;
use rustfs_audit::audit_target_metrics;
use rustfs_config::METER_INTERVAL;
@@ -731,6 +729,17 @@ fn storage_snapshot_metrics(scope: &'static str, collection_interval: Duration)
)
}
fn bucket_metrics_snapshot(collection_interval: Duration) -> StorageSnapshotMetrics {
let export_interval = Duration::from_secs(
get_env_opt_u64(ENV_OBS_METER_INTERVAL)
.filter(|value| *value > 0)
.unwrap_or(METER_INTERVAL),
);
let max_age = collection_interval.max(export_interval).saturating_mul(3);
let meter = process_global_meter().unwrap_or_else(|| opentelemetry::global::meter("rustfs.storage"));
StorageSnapshotMetrics::new_unscoped(meter, max_age)
}
fn metrics_interval(period: Duration, initial_delay: Duration) -> Interval {
let mut interval = tokio::time::interval_at(Instant::now() + initial_delay, period);
interval.set_missed_tick_behavior(MissedTickBehavior::Delay);
@@ -850,26 +859,6 @@ fn bucket_live_keys(stats: &[crate::metrics::collectors::BucketStats]) -> HashSe
stats.iter().map(|stat| stat.name.clone()).collect()
}
fn bucket_observation_live_keys(stats: &[crate::metrics::collectors::BucketStats]) -> HashSet<BucketKey> {
stats
.iter()
.filter(|stat| stat.size_bytes.is_some() || stat.objects_count.is_some())
.map(|stat| stat.name.clone())
.collect()
}
fn bucket_observation_retire_keys(
previous_observations: &HashSet<BucketKey>,
current_buckets: &HashSet<BucketKey>,
current_observations: &HashSet<BucketKey>,
) -> Vec<BucketKey> {
previous_observations
.difference(current_observations)
.filter(|bucket| current_buckets.contains(*bucket))
.cloned()
.collect()
}
fn collect_bucket_zero_tombstone_metrics(zero_tombstones: &HashMap<BucketKey, u8>) -> Vec<PrometheusMetric> {
if zero_tombstones.is_empty() {
return Vec::new();
@@ -892,14 +881,11 @@ fn collect_bucket_zero_tombstone_metrics(zero_tombstones: &HashMap<BucketKey, u8
struct BucketSeriesState {
has_seen_snapshot: bool,
live_keys: HashSet<BucketKey>,
observation_keys: HashSet<BucketKey>,
zero_tombstones: HashMap<BucketKey, u8>,
}
struct BucketSeriesUpdate {
metrics: Vec<PrometheusMetric>,
retire_observations: Vec<BucketKey>,
retire_buckets: Vec<BucketKey>,
}
impl BucketSeriesState {
@@ -910,10 +896,6 @@ impl BucketSeriesState {
) -> Option<BucketSeriesUpdate> {
let stats = stats?;
let current_bucket_keys = bucket_live_keys(stats);
let current_observation_keys = bucket_observation_live_keys(stats);
let retire_observations =
bucket_observation_retire_keys(&self.observation_keys, &current_bucket_keys, &current_observation_keys);
self.observation_keys = current_observation_keys;
update_series_zero_tombstones(
&mut self.has_seen_snapshot,
&mut self.live_keys,
@@ -923,49 +905,11 @@ impl BucketSeriesState {
);
let mut metrics = collect_bucket_metrics(stats);
metrics.extend(collect_bucket_zero_tombstone_metrics(&self.zero_tombstones));
let retire_buckets = expire_series_zero_tombstones(&mut self.zero_tombstones);
Some(BucketSeriesUpdate {
metrics,
retire_observations,
retire_buckets,
})
let _ = expire_series_zero_tombstones(&mut self.zero_tombstones);
Some(BucketSeriesUpdate { metrics })
}
}
fn retire_bucket_metric_series(bucket: &str) -> usize {
let bucket_label: Cow<'static, str> = Cow::Owned(bucket.to_string());
let labels = [("bucket", bucket_label.clone())];
retire_metric_series(&BUCKET_USAGE_BYTES_MD.get_full_metric_name(), &labels)
+ retire_metric_series(&BUCKET_OBJECTS_TOTAL_MD.get_full_metric_name(), &labels)
+ retire_metric_series(&BUCKET_QUOTA_BYTES_MD.get_full_metric_name(), &labels)
}
fn retire_bucket_observation_metric_series(bucket: &str) -> usize {
let labels = [("bucket", Cow::Owned(bucket.to_string()))];
retire_metric_series(&BUCKET_USAGE_BYTES_MD.get_full_metric_name(), &labels)
+ retire_metric_series(&BUCKET_OBJECTS_TOTAL_MD.get_full_metric_name(), &labels)
}
fn retire_cluster_usage_metric_series() -> usize {
let labels: [(&'static str, Cow<'static, str>); 0] = [];
[
USAGE_SINCE_LAST_UPDATE_SECONDS_MD.get_full_metric_name(),
USAGE_TOTAL_BYTES_MD.get_full_metric_name(),
USAGE_OBJECTS_COUNT_MD.get_full_metric_name(),
USAGE_VERSIONS_COUNT_MD.get_full_metric_name(),
USAGE_DELETE_MARKERS_COUNT_MD.get_full_metric_name(),
USAGE_BUCKETS_COUNT_MD.get_full_metric_name(),
]
.iter()
.map(|name| retire_metric_series(name, &labels))
.sum()
}
fn retire_cluster_usage_distribution_series(metric_name: String, range: &str) -> usize {
let labels = [(USAGE_RANGE_LABEL, Cow::Owned(range.to_string()))];
retire_metric_series(&metric_name, &labels)
}
fn bucket_usage_live_keys(stats: &[crate::metrics::collectors::BucketUsageStats]) -> HashSet<BucketKey> {
stats.iter().map(|stat| stat.bucket.clone()).collect()
}
@@ -1043,24 +987,6 @@ fn collect_bucket_usage_zero_tombstone_metrics(
zero_metrics
}
fn retire_bucket_usage_metric_series(bucket: &str) -> usize {
let bucket_label: Cow<'static, str> = Cow::Owned(bucket.to_string());
let labels = [(USAGE_BUCKET_LABEL, bucket_label.clone())];
retire_metric_series(&USAGE_BUCKET_TOTAL_BYTES_MD.get_full_metric_name(), &labels)
+ retire_metric_series(&USAGE_BUCKET_OBJECTS_TOTAL_MD.get_full_metric_name(), &labels)
+ retire_metric_series(&USAGE_BUCKET_VERSIONS_COUNT_MD.get_full_metric_name(), &labels)
+ retire_metric_series(&USAGE_BUCKET_DELETE_MARKERS_COUNT_MD.get_full_metric_name(), &labels)
+ retire_metric_series(&USAGE_BUCKET_QUOTA_TOTAL_BYTES_MD.get_full_metric_name(), &labels)
}
fn retire_bucket_usage_distribution_series(metric_name: String, bucket: &str, range: &str) -> usize {
let labels = [
(USAGE_RANGE_LABEL, Cow::Owned(range.to_string())),
(USAGE_BUCKET_LABEL, Cow::Owned(bucket.to_string())),
];
retire_metric_series(&metric_name, &labels)
}
fn audit_target_live_keys(stats: &[AuditTargetRuntimeStats]) -> HashSet<AuditTargetKey> {
stats
.iter()
@@ -1687,14 +1613,13 @@ pub fn init_metrics_runtime(token: CancellationToken) {
let mut interval = metrics_interval(cluster_interval, stagger_duration(cluster_interval, 1, 3));
let tombstone_cycles = config.replication_bandwidth_zero_tombstone_cycles;
let mut has_seen_bucket_usage_snapshot = false;
let mut bucket_usage_snapshot = bucket_metrics_snapshot(cluster_interval);
let mut prev_bucket_usage_keys: HashSet<BucketKey> = HashSet::new();
let mut bucket_usage_zero_tombstones: HashMap<BucketKey, u8> = HashMap::new();
let mut prev_bucket_usage_object_size_keys: HashSet<BucketRangeKey> = HashSet::new();
let mut bucket_usage_object_size_zero_tombstones: HashMap<BucketRangeKey, u8> = HashMap::new();
let mut prev_bucket_usage_version_keys: HashSet<BucketRangeKey> = HashSet::new();
let mut bucket_usage_version_zero_tombstones: HashMap<BucketRangeKey, u8> = HashMap::new();
let mut prev_cluster_usage_object_size_keys: HashSet<String> = HashSet::new();
let mut prev_cluster_usage_version_keys: HashSet<String> = HashSet::new();
loop {
tokio::select! {
_ = interval.tick() => {
@@ -1704,6 +1629,7 @@ pub fn init_metrics_runtime(token: CancellationToken) {
"supplementary_cluster_stats",
async {
let mut metrics = Vec::new();
let mut usage_metrics_update = None;
if let Some(stats) = collect_cluster_config_stats().await {
metrics.extend(collect_cluster_config_metrics(&stats));
@@ -1713,32 +1639,10 @@ pub fn init_metrics_runtime(token: CancellationToken) {
metrics.extend(collect_iam_metrics(&stats));
}
let usage_collection_started = std::time::Instant::now();
if let Some((cluster_usage, bucket_usage)) = collect_cluster_usage_metric_stats().await {
let current_cluster_usage_object_size_keys = cluster_usage
.object_size_distribution
.iter()
.map(|(range, _)| range.clone())
.collect::<HashSet<_>>();
for range in prev_cluster_usage_object_size_keys.difference(&current_cluster_usage_object_size_keys) {
let _ = retire_cluster_usage_distribution_series(
USAGE_OBJECTS_DISTRIBUTION_MD.get_full_metric_name(),
range,
);
}
prev_cluster_usage_object_size_keys = current_cluster_usage_object_size_keys;
let current_cluster_usage_version_keys = cluster_usage
.versions_distribution
.iter()
.map(|(range, _)| range.clone())
.collect::<HashSet<_>>();
for range in prev_cluster_usage_version_keys.difference(&current_cluster_usage_version_keys) {
let _ = retire_cluster_usage_distribution_series(
USAGE_VERSIONS_DISTRIBUTION_MD.get_full_metric_name(),
range,
);
}
prev_cluster_usage_version_keys = current_cluster_usage_version_keys;
metrics.extend(collect_cluster_usage_metrics(&cluster_usage));
let mut usage_metrics = collect_cluster_usage_metrics(&cluster_usage);
usage_metrics.extend(collect_bucket_usage_metrics(&bucket_usage));
update_series_zero_tombstones(
&mut has_seen_bucket_usage_snapshot,
&mut prev_bucket_usage_keys,
@@ -1760,64 +1664,21 @@ pub fn init_metrics_runtime(token: CancellationToken) {
bucket_usage_version_live_keys(&bucket_usage),
tombstone_cycles,
);
metrics.extend(collect_bucket_usage_metrics(&bucket_usage));
metrics.extend(collect_bucket_usage_zero_tombstone_metrics(
usage_metrics.extend(collect_bucket_usage_zero_tombstone_metrics(
&bucket_usage_zero_tombstones,
&bucket_usage_object_size_zero_tombstones,
&bucket_usage_version_zero_tombstones,
));
for bucket in expire_series_zero_tombstones(&mut bucket_usage_zero_tombstones) {
let _ = retire_bucket_usage_metric_series(&bucket);
}
for (bucket, range) in expire_series_zero_tombstones(&mut bucket_usage_object_size_zero_tombstones) {
let _ = retire_bucket_usage_distribution_series(
USAGE_BUCKET_OBJECT_SIZE_DISTRIBUTION_MD.get_full_metric_name(),
&bucket,
&range,
);
}
for (bucket, range) in expire_series_zero_tombstones(&mut bucket_usage_version_zero_tombstones) {
let _ = retire_bucket_usage_distribution_series(
USAGE_BUCKET_OBJECT_VERSION_COUNT_DISTRIBUTION_MD.get_full_metric_name(),
&bucket,
&range,
);
}
} else if has_seen_bucket_usage_snapshot {
let _ = retire_cluster_usage_metric_series();
for range in prev_cluster_usage_object_size_keys.drain() {
let _ = retire_cluster_usage_distribution_series(
USAGE_OBJECTS_DISTRIBUTION_MD.get_full_metric_name(),
&range,
);
}
for range in prev_cluster_usage_version_keys.drain() {
let _ = retire_cluster_usage_distribution_series(
USAGE_VERSIONS_DISTRIBUTION_MD.get_full_metric_name(),
&range,
);
}
for bucket in prev_bucket_usage_keys.drain() {
let _ = retire_bucket_usage_metric_series(&bucket);
}
for (bucket, range) in prev_bucket_usage_object_size_keys.drain() {
let _ = retire_bucket_usage_distribution_series(
USAGE_BUCKET_OBJECT_SIZE_DISTRIBUTION_MD.get_full_metric_name(),
&bucket,
&range,
);
}
for (bucket, range) in prev_bucket_usage_version_keys.drain() {
let _ = retire_bucket_usage_distribution_series(
USAGE_BUCKET_OBJECT_VERSION_COUNT_DISTRIBUTION_MD.get_full_metric_name(),
&bucket,
&range,
);
}
bucket_usage_zero_tombstones.clear();
bucket_usage_object_size_zero_tombstones.clear();
bucket_usage_version_zero_tombstones.clear();
has_seen_bucket_usage_snapshot = false;
let _ = expire_series_zero_tombstones(&mut bucket_usage_zero_tombstones);
let _ = expire_series_zero_tombstones(&mut bucket_usage_object_size_zero_tombstones);
let _ = expire_series_zero_tombstones(&mut bucket_usage_version_zero_tombstones);
usage_metrics_update = Some(usage_metrics);
}
if let Some(usage_metrics) = usage_metrics_update
&& let Err(error) = bucket_usage_snapshot.replace_collected(usage_metrics, usage_collection_started)
{
error!(event = EVENT_METRICS_RUNTIME_STATE, component = LOG_COMPONENT_OBS, subsystem = LOG_SUBSYSTEM_METRICS_RUNTIME, collector = "supplementary_cluster_stats", result = "invalid_bucket_snapshot", error = %error, "bucket usage metrics snapshot rejected");
}
if !metrics.is_empty() {
@@ -1840,20 +1701,18 @@ pub fn init_metrics_runtime(token: CancellationToken) {
let mut interval = metrics_interval(bucket_interval, Duration::ZERO);
let tombstone_cycles = config.replication_bandwidth_zero_tombstone_cycles;
let mut series_state = BucketSeriesState::default();
let mut snapshot = bucket_metrics_snapshot(bucket_interval);
loop {
tokio::select! {
_ = interval.tick() => {
run_metrics_collector_tick(health, MetricsCollectorTaskId::BucketStats, "bucket_stats", async {
let collection_started = std::time::Instant::now();
let stats = collect_bucket_stats().await;
let Some(update) = series_state.observe(stats.as_deref(), tombstone_cycles) else {
return;
};
for bucket in update.retire_observations {
let _ = retire_bucket_observation_metric_series(&bucket);
}
report_metrics(&update.metrics);
for bucket in update.retire_buckets {
let _ = retire_bucket_metric_series(&bucket);
if let Err(error) = snapshot.replace_collected(update.metrics, collection_started) {
error!(event = EVENT_METRICS_RUNTIME_STATE, component = LOG_COMPONENT_OBS, subsystem = LOG_SUBSYSTEM_METRICS_RUNTIME, collector = "bucket_stats", result = "invalid_snapshot", error = %error, "bucket metrics snapshot rejected");
}
}).await;
}
@@ -3485,29 +3344,6 @@ mod tests {
assert!(zero_tombstones.is_empty());
}
#[test]
fn bucket_observation_retirement_distinguishes_unknown_usage_from_deletion() {
let previous = HashSet::from(["bucket".to_string()]);
let unknown_stats = vec![crate::metrics::collectors::BucketStats {
name: "bucket".to_string(),
size_bytes: None,
objects_count: None,
quota_bytes: 1024,
}];
let current_buckets = bucket_live_keys(&unknown_stats);
let current_observations = bucket_observation_live_keys(&unknown_stats);
assert_eq!(
bucket_observation_retire_keys(&previous, &current_buckets, &current_observations),
vec!["bucket".to_string()],
"an existing bucket with unknown usage must retire its previous usage observations"
);
assert!(
bucket_observation_retire_keys(&previous, &HashSet::new(), &HashSet::new()).is_empty(),
"a deleted bucket remains governed by the zero-tombstone lifecycle"
);
}
#[test]
fn unavailable_bucket_snapshot_preserves_metric_series_state() {
let mut state = BucketSeriesState::default();
@@ -3520,12 +3356,10 @@ mod tests {
assert!(state.observe(Some(&initial), 2).is_some());
let live_keys = state.live_keys.clone();
let observation_keys = state.observation_keys.clone();
let zero_tombstones = state.zero_tombstones.clone();
assert!(state.observe(None, 2).is_none());
assert_eq!(state.live_keys, live_keys);
assert_eq!(state.observation_keys, observation_keys);
assert_eq!(state.zero_tombstones, zero_tombstones);
}
crates/obs/src/metrics/storage_snapshot.rs
+109 -46
View File
@@ -60,34 +60,46 @@ pub(crate) struct StorageSnapshotMetrics {
impl StorageSnapshotMetrics {
pub(crate) fn new(meter: Meter, scope: &'static str, observer: String, max_age: Duration) -> Self {
let snapshot = Arc::new(RwLock::new(Snapshot::default()));
let labels = vec![KeyValue::new(COLLECTION_SCOPE, scope), KeyValue::new(OBSERVER, observer)];
let weak = Arc::downgrade(&snapshot);
let timestamp_labels = labels.clone();
meter
.f64_observable_gauge(LAST_SUCCESS)
.with_description("Unix timestamp of the last successful storage snapshot collection")
.with_callback(move |observer| {
let Some(snapshot) = weak.upgrade() else { return };
let Ok(snapshot) = snapshot.read() else { return };
if snapshot.collected_at.is_some() {
observer.observe(snapshot.timestamp_seconds, &timestamp_labels);
}
})
.build();
let weak = Arc::downgrade(&snapshot);
let age_labels = labels.clone();
meter
.f64_observable_gauge(MAX_AGE)
.with_description("Validity budget in seconds remaining after storage snapshot collection completed")
.with_callback(move |observer| {
let Some(snapshot) = weak.upgrade() else { return };
let Ok(snapshot) = snapshot.read() else { return };
if snapshot.collected_at.is_some() {
observer.observe(snapshot.remaining_age.as_secs_f64(), &age_labels);
}
})
.build();
Self::new_with_labels(meter, labels, max_age, true)
}
/// Creates an unscoped snapshot for dynamic metric series whose membership
/// must match the latest authoritative collection.
pub(crate) fn new_unscoped(meter: Meter, max_age: Duration) -> Self {
Self::new_with_labels(meter, Vec::new(), max_age, false)
}
fn new_with_labels(meter: Meter, labels: Vec<KeyValue>, max_age: Duration, publish_freshness: bool) -> Self {
let snapshot = Arc::new(RwLock::new(Snapshot::default()));
if publish_freshness {
let weak = Arc::downgrade(&snapshot);
let timestamp_labels = labels.clone();
meter
.f64_observable_gauge(LAST_SUCCESS)
.with_description("Unix timestamp of the last successful storage snapshot collection")
.with_callback(move |observer| {
let Some(snapshot) = weak.upgrade() else { return };
let Ok(snapshot) = snapshot.read() else { return };
if snapshot.collected_at.is_some() {
observer.observe(snapshot.timestamp_seconds, &timestamp_labels);
}
})
.build();
let weak = Arc::downgrade(&snapshot);
let age_labels = labels.clone();
meter
.f64_observable_gauge(MAX_AGE)
.with_description("Validity budget in seconds remaining after storage snapshot collection completed")
.with_callback(move |observer| {
let Some(snapshot) = weak.upgrade() else { return };
let Ok(snapshot) = snapshot.read() else { return };
if snapshot.collected_at.is_some() {
observer.observe(snapshot.remaining_age.as_secs_f64(), &age_labels);
}
})
.build();
}
Self {
meter,
snapshot,
@@ -221,6 +233,7 @@ mod tests {
#[derive(Clone, Debug)]
struct ExportedPoint {
scope: String,
name: String,
value: f64,
attributes: Vec<KeyValue>,
@@ -233,28 +246,33 @@ mod tests {
async fn export(&self, metrics: &ResourceMetrics) -> OTelSdkResult {
let mut exported = self.0.lock().unwrap();
exported.clear();
for metric in metrics.scope_metrics().flat_map(|scope| scope.metrics()) {
match metric.data() {
AggregatedMetrics::F64(MetricData::Gauge(gauge)) => {
for point in gauge.data_points() {
exported.push(ExportedPoint {
name: metric.name().to_string(),
value: point.value(),
attributes: point.attributes().cloned().collect(),
});
for scope in metrics.scope_metrics() {
let scope_name = scope.scope().name().to_string();
for metric in scope.metrics() {
match metric.data() {
AggregatedMetrics::F64(MetricData::Gauge(gauge)) => {
for point in gauge.data_points() {
exported.push(ExportedPoint {
scope: scope_name.clone(),
name: metric.name().to_string(),
value: point.value(),
attributes: point.attributes().cloned().collect(),
});
}
}
}
AggregatedMetrics::U64(MetricData::Sum(sum)) => {
assert!(sum.is_monotonic());
for point in sum.data_points() {
exported.push(ExportedPoint {
name: metric.name().to_string(),
value: point.value() as f64,
attributes: point.attributes().cloned().collect(),
});
AggregatedMetrics::U64(MetricData::Sum(sum)) => {
assert!(sum.is_monotonic());
for point in sum.data_points() {
exported.push(ExportedPoint {
scope: scope_name.clone(),
name: metric.name().to_string(),
value: point.value() as f64,
attributes: point.attributes().cloned().collect(),
});
}
}
data => panic!("unexpected storage aggregation: {data:?}"),
}
data => panic!("unexpected storage aggregation: {data:?}"),
}
}
Ok(())
@@ -282,6 +300,16 @@ mod tests {
(provider, exporter, snapshot)
}
fn setup_unscoped() -> (SdkMeterProvider, Exporter, StorageSnapshotMetrics) {
let exporter = Exporter::default();
let reader = PeriodicReader::builder(exporter.clone())
.with_interval(Duration::from_secs(3600))
.build();
let provider = SdkMeterProvider::builder().with_reader(reader).build();
let snapshot = StorageSnapshotMetrics::new_unscoped(provider.meter("service-test"), Duration::from_secs(30));
(provider, exporter, snapshot)
}
fn drive(name: &'static str, kind: MetricType, id: &'static str, value: f64) -> PrometheusMetric {
PrometheusMetric::new(name, kind, "test", value)
.with_label("server", "node1:9000")
@@ -326,6 +354,41 @@ mod tests {
provider.shutdown().unwrap();
}
#[test]
fn actual_sdk_stops_exporting_deleted_bucket_quota_series_after_tombstone() {
let (provider, exporter, mut snapshot) = setup_unscoped();
let bucket = "deleted-bucket";
let quota_metrics = |value| {
vec![
PrometheusMetric::from_descriptor(&super::super::schema::node_bucket::BUCKET_QUOTA_BYTES_MD, value)
.with_label("bucket", bucket),
PrometheusMetric::from_descriptor(&super::super::schema::cluster_usage::USAGE_BUCKET_QUOTA_TOTAL_BYTES_MD, value)
.with_label("bucket", bucket),
]
};
snapshot.replace(quota_metrics(1024.0)).unwrap();
assert_eq!(
flush(&provider, &exporter)
.iter()
.filter(|point| point.value == 1024.0)
.count(),
2
);
assert!(flush(&provider, &exporter).iter().all(|point| point.scope == "service-test"));
snapshot.replace(quota_metrics(0.0)).unwrap();
let tombstones = flush(&provider, &exporter);
assert_eq!(tombstones.iter().filter(|point| point.value == 0.0).count(), 2);
snapshot.replace(Vec::new()).unwrap();
for _ in 0..2 {
let points = flush(&provider, &exporter);
assert!(points.is_empty(), "deleted bucket quota points must not remain in cumulative exports");
}
provider.shutdown().unwrap();
}
#[test]
fn actual_sdk_preserves_counter_resets_instead_of_accumulating_old_absolute_values() {
let (provider, exporter, mut snapshot) = setup();
crates/obs/src/telemetry/mod.rs
+1
View File
@@ -53,6 +53,7 @@ use crate::TelemetryError;
use crate::config::OtelConfig;
pub use filter::HTTP_SERVER_LOG_TARGET;
pub use guard::OtelGuard;
pub(crate) use recorder::process_global_meter;
pub use recorder::{Recorder, retire_metric_series};
use rustfs_config::observability::ENV_OBS_LOG_DIRECTORY;
use rustfs_config::{DEFAULT_LOG_LEVEL, ENVIRONMENT, observability::DEFAULT_OBS_ENVIRONMENT_PRODUCTION};
crates/obs/src/telemetry/recorder.rs
+4
View File
@@ -226,6 +226,10 @@ fn remember_global_recorder(recorder: &Recorder) {
let _ = GLOBAL_RECORDER.set(recorder.clone());
}
pub(crate) fn process_global_meter() -> Option<Meter> {
GLOBAL_RECORDER.get().map(|recorder| recorder.meter.clone())
}
pub(crate) fn install_process_global_recorder(recorder: Recorder) -> Result<(), metrics::SetRecorderError<Recorder>> {
metrics::set_global_recorder(recorder.clone())?;
remember_global_recorder(&recorder);