fix: pageserver_ondisk_layers metric

pageserver/src/tenant.rs

@@ -81,6 +81,7 @@ pub mod block_io;
 pub mod disk_btree;
 pub(crate) mod ephemeral_file;
 pub mod layer_map;
+mod measured_layer_map;
 
 pub mod metadata;
 mod par_fsync;

pageserver/src/tenant/layer_map.rs

@@ -48,7 +48,6 @@ mod layer_coverage;
 
 use crate::context::RequestContext;
 use crate::keyspace::KeyPartitioning;
-use crate::metrics::NUM_ONDISK_LAYERS;
 use crate::repository::Key;
 use crate::tenant::storage_layer::InMemoryLayer;
 use crate::tenant::storage_layer::Layer;
@@ -154,11 +153,7 @@ where
         expected: &Arc<L>,
         new: Arc<L>,
     ) -> anyhow::Result<Replacement<Arc<L>>> {
-        fail::fail_point!("layermap-replace-notfound", |_| Ok(
-            // this is not what happens if an L0 layer was not found a anyhow error but perhaps
-            // that should be changed. this is good enough to show a replacement failure.
-            Replacement::NotFound
-        ));
+        fail::fail_point!("layermap-replace-notfound", |_| Ok(Replacement::NotFound));
 
         self.layer_map.replace_historic_noflush(expected, new)
     }
@@ -270,6 +265,8 @@ where
 
     /// Start a batch of updates, applied on drop
     pub fn batch_update(&mut self) -> BatchedUpdates<'_, L> {
+        // Note: MeasuredLayerMap::batch_update wraps this, when extending or changing batched
+        // updates, make changes there as well
         BatchedUpdates { layer_map: self }
     }
 
@@ -287,8 +284,6 @@ where
         if Self::is_l0(&layer) {
             self.l0_delta_layers.push(layer);
         }
-
-        NUM_ONDISK_LAYERS.inc();
     }
 
     ///
@@ -313,8 +308,6 @@ where
                 "failed to locate removed historic layer from l0_delta_layers"
             );
         }
-
-        NUM_ONDISK_LAYERS.dec();
     }
 
     pub(self) fn replace_historic_noflush(
@@ -340,12 +333,15 @@ where
 
         let l0_index = if expected_l0 {
             // find the index in case replace worked, we need to replace that as well
-            Some(
-                self.l0_delta_layers
-                    .iter()
-                    .position(|slot| Self::compare_arced_layers(slot, expected))
-                    .ok_or_else(|| anyhow::anyhow!("existing l0 delta layer was not found"))?,
-            )
+            let pos = self
+                .l0_delta_layers
+                .iter()
+                .position(|slot| Self::compare_arced_layers(slot, expected));
+
+            if pos.is_none() {
+                return Ok(Replacement::NotFound);
+            }
+            pos
         } else {
             None
         };

pageserver/src/tenant/measured_layer_map.rs

@@ -0,0 +1,105 @@
+use std::sync::Arc;
+
+use crate::metrics::NUM_ONDISK_LAYERS;
+use crate::tenant::layer_map::{self, Replacement};
+use crate::tenant::storage_layer::PersistentLayer;
+
+/// Metrics updating wrapper around the real layermap.
+///
+/// The real layermap is needed to be generic over all layers to allow benchmarking and testing,
+/// but for real pageserver use we need only `dyn PersistentLayer`.
+///
+/// `Deref` and `DerefMut` delegate to original layer map, this type only implements the needed
+/// methods for metrics.
+///
+/// See [layer_map::LayerMap].
+#[derive(Default)]
+pub struct MeasuredLayerMap(layer_map::LayerMap<dyn PersistentLayer>);
+
+impl std::ops::Deref for MeasuredLayerMap {
+    type Target = layer_map::LayerMap<dyn PersistentLayer>;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl std::ops::DerefMut for MeasuredLayerMap {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.0
+    }
+}
+
+impl MeasuredLayerMap {
+    /// See [layer_map::LayerMap::batch_update].
+    pub fn batch_update(&mut self) -> BatchedUpdates<'_> {
+        BatchedUpdates {
+            inner: self.0.batch_update(),
+        }
+    }
+}
+
+/// See [layer_map::BatchedUpdates].
+///
+/// Wrapper for handling metric updates.
+#[must_use]
+pub struct BatchedUpdates<'a> {
+    inner: layer_map::BatchedUpdates<'a, dyn PersistentLayer>,
+}
+
+impl BatchedUpdates<'_> {
+    /// See [layer_map::BatchedUpdates::insert_historic].
+    pub fn insert_historic(&mut self, layer: Arc<dyn PersistentLayer>) {
+        let is_remote = layer.is_remote_layer();
+
+        self.inner.insert_historic(layer);
+
+        if !is_remote {
+            NUM_ONDISK_LAYERS.inc();
+        }
+    }
+
+    /// See [layer_map::BatchedUpdates::remove_historic].
+    pub fn remove_historic(&mut self, layer: Arc<dyn PersistentLayer>) {
+        let is_remote = layer.is_remote_layer();
+
+        self.inner.remove_historic(layer);
+
+        if !is_remote {
+            NUM_ONDISK_LAYERS.dec();
+        }
+    }
+
+    /// See [layer_map::BatchedUpdates::replace_historic].
+    pub fn replace_historic(
+        &mut self,
+        expected: &Arc<dyn PersistentLayer>,
+        new: Arc<dyn PersistentLayer>,
+    ) -> anyhow::Result<Replacement<Arc<dyn PersistentLayer>>> {
+        use Replacement::*;
+
+        let is_remote = new.is_remote_layer();
+
+        let res = self.inner.replace_historic(expected, new)?;
+
+        match &res {
+            Replaced { .. } => {
+                if is_remote != expected.is_remote_layer() {
+                    if is_remote {
+                        NUM_ONDISK_LAYERS.dec();
+                    } else {
+                        NUM_ONDISK_LAYERS.inc();
+                    }
+                }
+            }
+            NotFound | RemovalBuffered | Unexpected(_) => {}
+        }
+
+        Ok(res)
+    }
+
+    /// See [layer_map::BatchedUpdates::flush].
+    pub fn flush(self) {
+        self.inner.flush()
+    }
+}

pageserver/src/tenant/timeline.rs

@@ -71,7 +71,7 @@ use crate::ZERO_PAGE;
 use crate::{is_temporary, task_mgr};
 use walreceiver::spawn_connection_manager_task;
 
-use super::layer_map::BatchedUpdates;
+use super::measured_layer_map::{BatchedUpdates, MeasuredLayerMap};
 use super::remote_timeline_client::index::IndexPart;
 use super::remote_timeline_client::RemoteTimelineClient;
 use super::storage_layer::{DeltaLayer, ImageLayer, Layer, LayerAccessStatsReset};
@@ -113,7 +113,7 @@ pub struct Timeline {
 
     pub pg_version: u32,
 
-    pub(super) layers: RwLock<LayerMap<dyn PersistentLayer>>,
+    pub(super) layers: RwLock<MeasuredLayerMap>,
 
     last_freeze_at: AtomicLsn,
     // Atomic would be more appropriate here.
@@ -1039,7 +1039,7 @@ impl Timeline {
         &self,
         _layer_removal_cs: &tokio::sync::MutexGuard<'_, ()>,
         local_layer: &Arc<dyn PersistentLayer>,
-        batch_updates: &mut BatchedUpdates<'_, dyn PersistentLayer>,
+        batch_updates: &mut BatchedUpdates<'_>,
     ) -> anyhow::Result<bool> {
         use super::layer_map::Replacement;
 
@@ -1189,7 +1189,7 @@ impl Timeline {
                 timeline_id,
                 tenant_id,
                 pg_version,
-                layers: RwLock::new(LayerMap::default()),
+                layers: RwLock::new(MeasuredLayerMap::default()),
 
                 walredo_mgr,
 
@@ -1948,7 +1948,7 @@ impl Timeline {
         // we cannot remove layers otherwise, since gc and compaction will race
         _layer_removal_cs: &tokio::sync::MutexGuard<'_, ()>,
         layer: Arc<dyn PersistentLayer>,
-        updates: &mut BatchedUpdates<'_, dyn PersistentLayer>,
+        updates: &mut BatchedUpdates<'_>,
     ) -> anyhow::Result<()> {
         if !layer.is_remote_layer() {
             layer.delete_resident_layer_file()?;

test_runner/regress/test_ondemand_download.py

@@ -544,7 +544,7 @@ def test_compaction_downloads_on_demand_with_image_creation(
     """
     neon_env_builder.enable_remote_storage(
         remote_storage_kind=remote_storage_kind,
-        test_name="test_compaction_downloads_on_demand",
+        test_name="test_compaction_downloads_on_demand_with_image_creation",
     )
 
     env = neon_env_builder.init_start()
@@ -611,13 +611,24 @@ def test_compaction_downloads_on_demand_with_image_creation(
     layers = pageserver_http.layer_map_info(tenant_id, timeline_id)
     assert not layers.in_memory_layers, "no inmemory layers expected after post-commit checkpoint"
 
+    total_populated_layers = (
+        pageserver_http.get_metrics().query_one("pageserver_ondisk_layers").value
+    )
     kinds_before: DefaultDict[str, int] = defaultdict(int)
+    remotes_before = 0
 
     for layer in layers.historic_layers:
         kinds_before[layer.kind] += 1
+        if layer.remote:
+            remotes_before += 1
         pageserver_http.evict_layer(tenant_id, timeline_id, layer.layer_file_name)
 
     assert dict(kinds_before) == {"Delta": 4}
+    assert remotes_before == 0
+
+    post_eviction_total_layers = (
+        pageserver_http.get_metrics().query_one("pageserver_ondisk_layers").value
+    )
 
     # now having evicted all layers, reconfigure to have lower image creation
     # threshold to expose image creation to downloading all of the needed
@@ -628,11 +639,26 @@ def test_compaction_downloads_on_demand_with_image_creation(
     pageserver_http.timeline_compact(tenant_id, timeline_id)
     layers = pageserver_http.layer_map_info(tenant_id, timeline_id)
     kinds_after: DefaultDict[str, int] = defaultdict(int)
+    remotes_after = 0
     for layer in layers.historic_layers:
         kinds_after[layer.kind] += 1
+        if layer.remote:
+            remotes_after += 1
+        log.info(layer)
 
     assert dict(kinds_after) == {"Delta": 4, "Image": 1}
 
+    post_compaction_total_layers = (
+        pageserver_http.get_metrics().query_one("pageserver_ondisk_layers").value
+    )
+
+    # assumption: floats here are small enough to compare with integers safely
+    # writing a separate test to check these seems overkill
+    assert total_populated_layers == post_eviction_total_layers + 4
+    # corrected with remotes_after because only 3 out of 4 seem to be usually
+    # required for layer creation
+    assert post_compaction_total_layers == total_populated_layers + 1 - remotes_after
+
 
 def stringify(conf: Dict[str, Any]) -> Dict[str, str]:
     return dict(map(lambda x: (x[0], str(x[1])), conf.items()))

test_runner/regress/test_tenants.py

@@ -55,10 +55,10 @@ def test_tenant_creation_fails(neon_simple_env: NeonEnv):
 
 
 def test_tenants_normal_work(neon_env_builder: NeonEnvBuilder):
+    """Tests tenants with and without wal acceptors"""
     neon_env_builder.num_safekeepers = 3
 
     env = neon_env_builder.init_start()
-    """Tests tenants with and without wal acceptors"""
     tenant_1, _ = env.neon_cli.create_tenant()
     tenant_2, _ = env.neon_cli.create_tenant()