diff --git a/pageserver/src/metrics.rs b/pageserver/src/metrics.rs index 67f6fcc5ec..0c3e9ff0a4 100644 --- a/pageserver/src/metrics.rs +++ b/pageserver/src/metrics.rs @@ -1,12 +1,12 @@ -use std::collections::HashMap; + use std::num::NonZeroUsize; -use std::sync::{Arc, Mutex}; +use std::sync::Arc; use std::time::{Duration, Instant}; use enum_map::{Enum as _, EnumMap}; use futures::Future; use metrics::{ - CounterVec, GaugeVec, Histogram, HistogramVec, IntCounter, IntCounterPair, + CounterVec, GaugeVec, Histogram, HistogramVec, IntCounter, IntCounterPairVec, IntCounterVec, IntGauge, IntGaugeVec, UIntGauge, UIntGaugeVec, register_counter_vec, register_gauge_vec, register_histogram, register_histogram_vec, register_int_counter, register_int_counter_pair_vec, register_int_counter_vec, @@ -258,15 +258,6 @@ static DISK_CONSISTENT_LSN: Lazy = Lazy::new(|| { .expect("failed to define a metric") }); -pub(crate) static PROJECTED_REMOTE_CONSISTENT_LSN: Lazy = Lazy::new(|| { - register_uint_gauge_vec!( - "pageserver_projected_remote_consistent_lsn", - "Projected remote consistent LSN grouped by timeline", - &["tenant_id", "shard_id", "timeline_id"] - ) - .expect("failed to define a metric") -}); - static PITR_HISTORY_SIZE: Lazy = Lazy::new(|| { register_uint_gauge_vec!( "pageserver_pitr_history_size", @@ -357,15 +348,6 @@ static VISIBLE_PHYSICAL_SIZE: Lazy = Lazy::new(|| { .expect("failed to define a metric") }); -static REMOTE_PHYSICAL_SIZE: Lazy = Lazy::new(|| { - register_uint_gauge_vec!( - "pageserver_remote_physical_size", - "The size of the layer files present in the remote storage that are listed in the remote index_part.json.", - // Corollary: If any files are missing from the index part, they won't be included here. - &["tenant_id", "shard_id", "timeline_id"] - ) - .expect("failed to define a metric") -}); pub(crate) static REMOTE_ONDEMAND_DOWNLOADED_LAYERS: Lazy = Lazy::new(|| { register_int_counter!( @@ -1130,45 +1112,8 @@ pub(crate) enum ComputeCommandKind { } -// remote storage metrics -static REMOTE_TIMELINE_CLIENT_CALLS: Lazy = Lazy::new(|| { - register_int_counter_pair_vec!( - "pageserver_remote_timeline_client_calls_started", - "Number of started calls to remote timeline client.", - "pageserver_remote_timeline_client_calls_finished", - "Number of finshed calls to remote timeline client.", - &[ - "tenant_id", - "shard_id", - "timeline_id", - "file_kind", - "op_kind" - ], - ) - .unwrap() -}); -static REMOTE_TIMELINE_CLIENT_BYTES_STARTED_COUNTER: Lazy = - Lazy::new(|| { - register_int_counter_vec!( - "pageserver_remote_timeline_client_bytes_started", - "Incremented by the number of bytes associated with a remote timeline client operation. \ - The increment happens when the operation is scheduled.", - &["tenant_id", "shard_id", "timeline_id", "file_kind", "op_kind"], - ) - .expect("failed to define a metric") - }); - -static REMOTE_TIMELINE_CLIENT_BYTES_FINISHED_COUNTER: Lazy = Lazy::new(|| { - register_int_counter_vec!( - "pageserver_remote_timeline_client_bytes_finished", - "Incremented by the number of bytes associated with a remote timeline client operation. \ - The increment happens when the operation finishes (regardless of success/failure/shutdown).", - &["tenant_id", "shard_id", "timeline_id", "file_kind", "op_kind"], - ) - .expect("failed to define a metric") -}); pub(crate) struct TenantManagerMetrics { pub(crate) tenant_slot_writes: IntCounter, @@ -1356,22 +1301,6 @@ impl RemoteOpFileKind { } } -pub(crate) static REMOTE_TIMELINE_CLIENT_COMPLETION_LATENCY: Lazy = Lazy::new(|| { - register_histogram_vec!( - "pageserver_remote_timeline_client_seconds_global", - "Time spent on remote timeline client operations. \ - Grouped by task_kind, file_kind, operation_kind and status. \ - The task_kind is \ - - for layer downloads, populated from RequestContext (primary objective of having the label) \ - - for index downloads, set to 'unknown' \ - - for any upload operation, set to 'RemoteUploadTask' \ - This keeps dimensionality at bay. \ - Does not account for time spent waiting in remote timeline client's queues.", - &["task_kind", "file_kind", "op_kind", "status"] - ) - .expect("failed to define a metric") -}); - pub(crate) static TENANT_TASK_EVENTS: Lazy = Lazy::new(|| { register_int_counter_vec!( "pageserver_tenant_task_events", @@ -2019,301 +1948,19 @@ pub(crate) fn remove_tenant_metrics(tenant_shard_id: &TenantShardId) { // we leave the BROKEN_TENANTS_SET entry if any } -/// Maintain a per timeline gauge in addition to the global gauge. -pub(crate) struct PerTimelineRemotePhysicalSizeGauge { -} - -impl PerTimelineRemotePhysicalSizeGauge { - fn new(_per_timeline_gauge: UIntGauge) -> Self { - Self { - } - } - pub(crate) fn set(&self,_sz: u64) { - - } - pub(crate) fn get(&self) -> u64 { - - 0 - } -} - -impl Drop for PerTimelineRemotePhysicalSizeGauge { - fn drop(&mut self) { - - } -} - -pub(crate) struct RemoteTimelineClientMetrics { - tenant_id: String, - shard_id: String, - timeline_id: String, - pub(crate) remote_physical_size_gauge: PerTimelineRemotePhysicalSizeGauge, - calls: Mutex>, - bytes_started_counter: Mutex>, - bytes_finished_counter: Mutex>, - pub(crate) projected_remote_consistent_lsn_gauge: UIntGauge, -} - -impl RemoteTimelineClientMetrics { - pub fn new(tenant_shard_id: &TenantShardId, timeline_id: &TimelineId) -> Self { - let tenant_id_str = tenant_shard_id.tenant_id.to_string(); - let shard_id_str = format!("{}", tenant_shard_id.shard_slug()); - let timeline_id_str = timeline_id.to_string(); - - let remote_physical_size_gauge = PerTimelineRemotePhysicalSizeGauge::new( - REMOTE_PHYSICAL_SIZE - .get_metric_with_label_values(&[&tenant_id_str, &shard_id_str, &timeline_id_str]) - .unwrap(), - ); - - let projected_remote_consistent_lsn_gauge = PROJECTED_REMOTE_CONSISTENT_LSN - .get_metric_with_label_values(&[&tenant_id_str, &shard_id_str, &timeline_id_str]) - .unwrap(); - - RemoteTimelineClientMetrics { - tenant_id: tenant_id_str, - shard_id: shard_id_str, - timeline_id: timeline_id_str, - calls: Mutex::new(HashMap::default()), - bytes_started_counter: Mutex::new(HashMap::default()), - bytes_finished_counter: Mutex::new(HashMap::default()), - remote_physical_size_gauge, - projected_remote_consistent_lsn_gauge, - } - } - - pub fn remote_operation_time( - &self, - task_kind: Option, - file_kind: &RemoteOpFileKind, - op_kind: &RemoteOpKind, - status: &'static str, - ) -> Histogram { - REMOTE_TIMELINE_CLIENT_COMPLETION_LATENCY - .get_metric_with_label_values(&[ - task_kind.as_ref().map(|tk| tk.into()).unwrap_or("unknown"), - file_kind.as_str(), - op_kind.as_str(), - status, - ]) - .unwrap() - } - - fn calls_counter_pair( - &self, - file_kind: &RemoteOpFileKind, - op_kind: &RemoteOpKind, - ) -> IntCounterPair { - let mut guard = self.calls.lock().unwrap(); - let key = (file_kind.as_str(), op_kind.as_str()); - let metric = guard.entry(key).or_insert_with(move || { - REMOTE_TIMELINE_CLIENT_CALLS - .get_metric_with_label_values(&[ - &self.tenant_id, - &self.shard_id, - &self.timeline_id, - key.0, - key.1, - ]) - .unwrap() - }); - metric.clone() - } - - fn bytes_finished_counter( - &self, - file_kind: &RemoteOpFileKind, - op_kind: &RemoteOpKind, - ) -> IntCounter { - let mut guard = self.bytes_finished_counter.lock().unwrap(); - let key = (file_kind.as_str(), op_kind.as_str()); - let metric = guard.entry(key).or_insert_with(move || { - REMOTE_TIMELINE_CLIENT_BYTES_FINISHED_COUNTER - .get_metric_with_label_values(&[ - &self.tenant_id, - &self.shard_id, - &self.timeline_id, - key.0, - key.1, - ]) - .unwrap() - }); - metric.clone() - } -} - -#[cfg(test)] -impl RemoteTimelineClientMetrics { - pub fn get_bytes_started_counter_value( - &self, - _file_kind: &RemoteOpFileKind, - _op_kind: &RemoteOpKind, - ) -> Option { - None // FIXME: Return dummy value as counter access is commented out - } - - pub fn get_bytes_finished_counter_value( - &self, - _file_kind: &RemoteOpFileKind, - _op_kind: &RemoteOpKind, - ) -> Option { - None // FIXME: Return dummy value as counter access is commented out - } -} - -/// See [`RemoteTimelineClientMetrics::call_begin`]. -#[must_use] -pub(crate) struct RemoteTimelineClientCallMetricGuard { - /// Decremented on drop. - calls_counter_pair: Option, - /// If Some(), this references the bytes_finished metric, and we increment it by the given `u64` on drop. - bytes_finished: Option<(IntCounter, u64)>, -} - -impl RemoteTimelineClientCallMetricGuard { - /// Consume this guard object without performing the metric updates it would do on `drop()`. - /// The caller vouches to do the metric updates manually. - pub fn will_decrement_manually(mut self) { - let RemoteTimelineClientCallMetricGuard { - calls_counter_pair, - bytes_finished, - } = &mut self; - calls_counter_pair.take(); - bytes_finished.take(); - } -} - -impl Drop for RemoteTimelineClientCallMetricGuard { - fn drop(&mut self) { - } -} - -/// The enum variants communicate to the [`RemoteTimelineClientMetrics`] whether to -/// track the byte size of this call in applicable metric(s). -pub(crate) enum RemoteTimelineClientMetricsCallTrackSize { - /// Do not account for this call's byte size in any metrics. - /// The `reason` field is there to make the call sites self-documenting - /// about why they don't need the metric. - DontTrackSize { reason: &'static str }, - /// Track the byte size of the call in applicable metric(s). - Bytes(u64), -} - -impl RemoteTimelineClientMetrics { - /// Update the metrics that change when a call to the remote timeline client instance starts. - /// - /// Drop the returned guard object once the operation is finished to updates corresponding metrics that track completions. - /// Or, use [`RemoteTimelineClientCallMetricGuard::will_decrement_manually`] and [`call_end`](Self::call_end) if that - /// is more suitable. - /// Never do both. - pub(crate) fn call_begin( - &self, - file_kind: &RemoteOpFileKind, - op_kind: &RemoteOpKind, - size: RemoteTimelineClientMetricsCallTrackSize, - ) -> RemoteTimelineClientCallMetricGuard { - let calls_counter_pair = self.calls_counter_pair(file_kind, op_kind); - // calls_counter_pair.inc(); - - let bytes_finished = match size { - RemoteTimelineClientMetricsCallTrackSize::DontTrackSize { reason: _reason } => { - // nothing to do - None - } - RemoteTimelineClientMetricsCallTrackSize::Bytes(size) => { - // self.bytes_started_counter(file_kind, op_kind).inc_by(size); - let finished_counter = self.bytes_finished_counter(file_kind, op_kind); - Some((finished_counter, size)) - } - }; - RemoteTimelineClientCallMetricGuard { - calls_counter_pair: Some(calls_counter_pair), - bytes_finished, - } - } - - /// Manually udpate the metrics that track completions, instead of using the guard object. - /// Using the guard object is generally preferable. - /// See [`call_begin`](Self::call_begin) for more context. - pub(crate) fn call_end( - &self, - _file_kind: &RemoteOpFileKind, - _op_kind: &RemoteOpKind, - _size: RemoteTimelineClientMetricsCallTrackSize, - ) { - - } -} - -impl Drop for RemoteTimelineClientMetrics { - fn drop(&mut self) { - let RemoteTimelineClientMetrics { - tenant_id, - shard_id, - timeline_id, - remote_physical_size_gauge, - calls, - bytes_started_counter, - bytes_finished_counter, - projected_remote_consistent_lsn_gauge, - } = self; - for ((a, b), _) in calls.get_mut().unwrap().drain() { - let mut res = [Ok(()), Ok(())]; - REMOTE_TIMELINE_CLIENT_CALLS - .remove_label_values(&mut res, &[tenant_id, shard_id, timeline_id, a, b]); - // don't care about results - } - for ((a, b), _) in bytes_started_counter.get_mut().unwrap().drain() { - let _ = REMOTE_TIMELINE_CLIENT_BYTES_STARTED_COUNTER.remove_label_values(&[ - tenant_id, - shard_id, - timeline_id, - a, - b, - ]); - } - for ((a, b), _) in bytes_finished_counter.get_mut().unwrap().drain() { - let _ = REMOTE_TIMELINE_CLIENT_BYTES_FINISHED_COUNTER.remove_label_values(&[ - tenant_id, - shard_id, - timeline_id, - a, - b, - ]); - } - { - let _ = remote_physical_size_gauge; // use to avoid 'unused' warning in desctructuring above - let _ = REMOTE_PHYSICAL_SIZE.remove_label_values(&[tenant_id, shard_id, timeline_id]); - } - { - let _ = projected_remote_consistent_lsn_gauge; - let _ = PROJECTED_REMOTE_CONSISTENT_LSN.remove_label_values(&[ - tenant_id, - shard_id, - timeline_id, - ]); - } - } -} /// Wrapper future that measures the time spent by a remote storage operation, /// and records the time and success/failure as a prometheus metric. pub(crate) trait MeasureRemoteOp: Sized + Future> { async fn measure_remote_op( self, - task_kind: Option, // not all caller contexts have a RequestContext / TaskKind handy - file_kind: RemoteOpFileKind, - op: RemoteOpKind, - metrics: Arc, + _task_kind: Option, // not all caller contexts have a RequestContext / TaskKind handy + _file_kind: RemoteOpFileKind, + _op: RemoteOpKind, ) -> Result { - let start = Instant::now(); - let res = self.await; - let duration = start.elapsed(); - let status = if res.is_ok() { &"success" } else { &"failure" }; - metrics - .remote_operation_time(task_kind, &file_kind, &op, status) - .observe(duration.as_secs_f64()); - res + + self.await + } } @@ -2645,24 +2292,8 @@ pub(crate) mod disk_usage_based_eviction { pub(crate) static METRICS: Lazy = Lazy::new(Metrics::default); } -static TOKIO_EXECUTOR_THREAD_COUNT: Lazy = Lazy::new(|| { - register_uint_gauge_vec!( - "pageserver_tokio_executor_thread_configured_count", - "Total number of configued tokio executor threads in the process. - The `setup` label denotes whether we're running with multiple runtimes or a single runtime.", - &["setup"], - ) - .unwrap() -}); -pub(crate) fn set_tokio_runtime_setup(setup: &str, num_threads: NonZeroUsize) { - static SERIALIZE: std::sync::Mutex<()> = std::sync::Mutex::new(()); - let _guard = SERIALIZE.lock().unwrap(); - TOKIO_EXECUTOR_THREAD_COUNT.reset(); - TOKIO_EXECUTOR_THREAD_COUNT - .get_metric_with_label_values(&[setup]) - .unwrap() - .set(u64::try_from(num_threads.get()).unwrap()); +pub(crate) fn set_tokio_runtime_setup(_setup: &str, _num_threads: NonZeroUsize) { } pub fn preinitialize_metrics( diff --git a/pageserver/src/tenant/remote_timeline_client.rs b/pageserver/src/tenant/remote_timeline_client.rs index 10a13ef1a2..5fac8689d8 100644 --- a/pageserver/src/tenant/remote_timeline_client.rs +++ b/pageserver/src/tenant/remote_timeline_client.rs @@ -224,8 +224,7 @@ use crate::context::RequestContext; use crate::deletion_queue::{DeletionQueueClient, DeletionQueueError}; use crate::metrics::{ MeasureRemoteOp, REMOTE_ONDEMAND_DOWNLOADED_BYTES, REMOTE_ONDEMAND_DOWNLOADED_LAYERS, - RemoteOpFileKind, RemoteOpKind, RemoteTimelineClientMetrics, - RemoteTimelineClientMetricsCallTrackSize, + RemoteOpFileKind, RemoteOpKind, }; use crate::task_mgr::{BACKGROUND_RUNTIME, TaskKind, shutdown_token}; use crate::tenant::metadata::TimelineMetadata; @@ -357,8 +356,6 @@ pub(crate) struct RemoteTimelineClient { upload_queue: Mutex, - pub(crate) metrics: Arc, - storage_impl: GenericRemoteStorage, deletion_queue_client: DeletionQueueClient, @@ -405,10 +402,6 @@ impl RemoteTimelineClient { storage_impl: remote_storage, deletion_queue_client, upload_queue: Mutex::new(UploadQueue::Uninitialized), - metrics: Arc::new(RemoteTimelineClientMetrics::new( - &tenant_shard_id, - &timeline_id, - )), config: std::sync::RwLock::new(RemoteTimelineClientConfig::from(location_conf)), cancel: CancellationToken::new(), } @@ -597,21 +590,13 @@ impl RemoteTimelineClient { .map_err(|_| UploadQueueNotReadyError) } - fn update_remote_physical_size_gauge(&self, current_remote_index_part: Option<&IndexPart>) { - let size: u64 = if let Some(current_remote_index_part) = current_remote_index_part { - current_remote_index_part - .layer_metadata - .values() - .map(|ilmd| ilmd.file_size) - .sum() - } else { - 0 - }; - self.metrics.remote_physical_size_gauge.set(size); + fn update_remote_physical_size_gauge(&self, _current_remote_index_part: Option<&IndexPart>) { + + } pub fn get_remote_physical_size(&self) -> u64 { - self.metrics.remote_physical_size_gauge.get() + 0 } // @@ -626,13 +611,6 @@ impl RemoteTimelineClient { &self, cancel: &CancellationToken, ) -> Result { - let _unfinished_gauge_guard = self.metrics.call_begin( - &RemoteOpFileKind::Index, - &RemoteOpKind::Download, - crate::metrics::RemoteTimelineClientMetricsCallTrackSize::DontTrackSize { - reason: "no need for a downloads gauge", - }, - ); let (index_part, index_generation, index_last_modified) = download::download_index_part( &self.storage_impl, @@ -645,7 +623,7 @@ impl RemoteTimelineClient { Option::::None, RemoteOpFileKind::Index, RemoteOpKind::Download, - Arc::clone(&self.metrics), + ) .await?; @@ -720,13 +698,7 @@ impl RemoteTimelineClient { ctx: &RequestContext, ) -> Result { let downloaded_size = { - let _unfinished_gauge_guard = self.metrics.call_begin( - &RemoteOpFileKind::Layer, - &RemoteOpKind::Download, - crate::metrics::RemoteTimelineClientMetricsCallTrackSize::DontTrackSize { - reason: "no need for a downloads gauge", - }, - ); + download::download_layer_file( self.conf, &self.storage_impl, @@ -743,7 +715,7 @@ impl RemoteTimelineClient { Some(ctx.task_kind()), RemoteOpFileKind::Layer, RemoteOpKind::Download, - Arc::clone(&self.metrics), + ) .await? }; @@ -1027,7 +999,6 @@ impl RemoteTimelineClient { let op = UploadOp::UploadMetadata { uploaded: Box::new(index_part.clone()), }; - self.metric_begin(&op); upload_queue.queued_operations.push_back(op); upload_queue.latest_files_changes_since_metadata_upload_scheduled = 0; @@ -1265,7 +1236,6 @@ impl RemoteTimelineClient { ); let op = UploadOp::UploadLayer(layer, metadata, None); - self.metric_begin(&op); upload_queue.queued_operations.push_back(op); } @@ -1442,7 +1412,6 @@ impl RemoteTimelineClient { let op = UploadOp::Delete(Delete { layers: with_metadata, }); - self.metric_begin(&op); upload_queue.queued_operations.push_back(op); } @@ -2180,7 +2149,7 @@ impl RemoteTimelineClient { Some(TaskKind::RemoteUploadTask), RemoteOpFileKind::Layer, RemoteOpKind::Upload, - Arc::clone(&self.metrics), + ) .await } @@ -2197,7 +2166,7 @@ impl RemoteTimelineClient { Some(TaskKind::RemoteUploadTask), RemoteOpFileKind::Index, RemoteOpKind::Upload, - Arc::clone(&self.metrics), + ) .await; if res.is_ok() { @@ -2343,10 +2312,7 @@ impl RemoteTimelineClient { upload_queue.clean.1 = Some(task.task_id); let lsn = upload_queue.clean.0.metadata.disk_consistent_lsn(); - self.metrics - .projected_remote_consistent_lsn_gauge - .set(lsn.0); - + if self.generation.is_none() { // Legacy mode: skip validating generation upload_queue.visible_remote_consistent_lsn.store(lsn); @@ -2387,64 +2353,6 @@ impl RemoteTimelineClient { .await; } - self.metric_end(&task.op); - for coalesced_op in &task.coalesced_ops { - self.metric_end(coalesced_op); - } - } - - fn metric_impl( - &self, - op: &UploadOp, - ) -> Option<( - RemoteOpFileKind, - RemoteOpKind, - RemoteTimelineClientMetricsCallTrackSize, - )> { - use RemoteTimelineClientMetricsCallTrackSize::DontTrackSize; - let res = match op { - UploadOp::UploadLayer(_, m, _) => ( - RemoteOpFileKind::Layer, - RemoteOpKind::Upload, - RemoteTimelineClientMetricsCallTrackSize::Bytes(m.file_size), - ), - UploadOp::UploadMetadata { .. } => ( - RemoteOpFileKind::Index, - RemoteOpKind::Upload, - DontTrackSize { - reason: "metadata uploads are tiny", - }, - ), - UploadOp::Delete(_delete) => ( - RemoteOpFileKind::Layer, - RemoteOpKind::Delete, - DontTrackSize { - reason: "should we track deletes? positive or negative sign?", - }, - ), - UploadOp::Barrier(..) | UploadOp::Shutdown => { - // we do not account these - return None; - } - }; - Some(res) - } - - fn metric_begin(&self, op: &UploadOp) { - let (file_kind, op_kind, track_bytes) = match self.metric_impl(op) { - Some(x) => x, - None => return, - }; - let guard = self.metrics.call_begin(&file_kind, &op_kind, track_bytes); - guard.will_decrement_manually(); // in metric_end(), see right below - } - - fn metric_end(&self, op: &UploadOp) { - let (file_kind, op_kind, track_bytes) = match self.metric_impl(op) { - Some(x) => x, - None => return, - }; - self.metrics.call_end(&file_kind, &op_kind, track_bytes); } /// Close the upload queue for new operations and cancel queued operations. @@ -2524,7 +2432,6 @@ impl RemoteTimelineClient { // Tear down queued ops for op in qi.queued_operations.into_iter() { - self.metric_end(&op); // Dropping UploadOp::Barrier() here will make wait_completion() return with an Err() // which is exactly what we want to happen. drop(op); @@ -2834,10 +2741,6 @@ mod tests { storage_impl: self.harness.remote_storage.clone(), deletion_queue_client: self.harness.deletion_queue.new_client(), upload_queue: Mutex::new(UploadQueue::Uninitialized), - metrics: Arc::new(RemoteTimelineClientMetrics::new( - &self.harness.tenant_shard_id, - &TIMELINE_ID, - )), config: std::sync::RwLock::new(RemoteTimelineClientConfig::from(&location_conf)), cancel: CancellationToken::new(), }) @@ -3064,99 +2967,7 @@ mod tests { ); } - #[tokio::test] - async fn bytes_unfinished_gauge_for_layer_file_uploads() { - // Setup - - let TestSetup { - harness, - tenant: _tenant, - timeline, - .. - } = TestSetup::new("metrics").await.unwrap(); - let client = &timeline.remote_client; - - let layer_file_name_1: LayerName = "000000000000000000000000000000000000-FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF__00000000016B59D8-00000000016B5A51".parse().unwrap(); - let local_path = local_layer_path( - harness.conf, - &timeline.tenant_shard_id, - &timeline.timeline_id, - &layer_file_name_1, - &harness.generation, - ); - let content_1 = dummy_contents("foo"); - std::fs::write(&local_path, &content_1).unwrap(); - - let layer_file_1 = Layer::for_resident( - harness.conf, - &timeline, - local_path, - layer_file_name_1.clone(), - LayerFileMetadata::new(content_1.len() as u64, harness.generation, harness.shard), - ); - - #[derive(Debug, PartialEq, Clone, Copy)] - struct BytesStartedFinished { - started: Option, - finished: Option, - } - impl std::ops::Add for BytesStartedFinished { - type Output = Self; - fn add(self, rhs: Self) -> Self::Output { - Self { - started: self.started.map(|v| v + rhs.started.unwrap_or(0)), - finished: self.finished.map(|v| v + rhs.finished.unwrap_or(0)), - } - } - } - let get_bytes_started_stopped = || { - let started = client - .metrics - .get_bytes_started_counter_value(&RemoteOpFileKind::Layer, &RemoteOpKind::Upload) - .map(|v| v.try_into().unwrap()); - let stopped = client - .metrics - .get_bytes_finished_counter_value(&RemoteOpFileKind::Layer, &RemoteOpKind::Upload) - .map(|v| v.try_into().unwrap()); - BytesStartedFinished { - started, - finished: stopped, - } - }; - - // Test - tracing::info!("now doing actual test"); - - let actual_a = get_bytes_started_stopped(); - - client - .schedule_layer_file_upload(layer_file_1.clone()) - .unwrap(); - - let actual_b = get_bytes_started_stopped(); - - client.wait_completion().await.unwrap(); - - let actual_c = get_bytes_started_stopped(); - - // Validate - - let expected_b = actual_a - + BytesStartedFinished { - started: Some(content_1.len()), - // assert that the _finished metric is created eagerly so that subtractions work on first sample - finished: Some(0), - }; - assert_eq!(actual_b, expected_b); - - let expected_c = actual_a - + BytesStartedFinished { - started: Some(content_1.len()), - finished: Some(content_1.len()), - }; - assert_eq!(actual_c, expected_c); - } - + async fn inject_index_part(test_state: &TestSetup, generation: Generation) -> IndexPart { // An empty IndexPart, just sufficient to ensure deserialization will succeed let example_index_part = IndexPart::example();