LCOV - code coverage report
Current view: top level - storage_scrubber/src - pageserver_physical_gc.rs (source / functions) Coverage Total Hit
Test: 5fe7fa8d483b39476409aee736d6d5e32728bfac.info Lines: 0.0 % 559 0
Test Date: 2025-03-12 16:10:49 Functions: 0.0 % 38 0

            Line data    Source code
       1              : use std::collections::{BTreeMap, BTreeSet, HashMap};
       2              : use std::sync::Arc;
       3              : use std::time::Duration;
       4              : 
       5              : use async_stream::try_stream;
       6              : use futures::future::Either;
       7              : use futures_util::{StreamExt, TryStreamExt};
       8              : use pageserver::tenant::IndexPart;
       9              : use pageserver::tenant::remote_timeline_client::index::LayerFileMetadata;
      10              : use pageserver::tenant::remote_timeline_client::manifest::OffloadedTimelineManifest;
      11              : use pageserver::tenant::remote_timeline_client::{
      12              :     parse_remote_index_path, parse_remote_tenant_manifest_path, remote_layer_path,
      13              : };
      14              : use pageserver::tenant::storage_layer::LayerName;
      15              : use pageserver_api::controller_api::TenantDescribeResponse;
      16              : use pageserver_api::shard::{ShardIndex, TenantShardId};
      17              : use remote_storage::{GenericRemoteStorage, ListingObject, RemotePath};
      18              : use reqwest::Method;
      19              : use serde::Serialize;
      20              : use storage_controller_client::control_api;
      21              : use tokio_util::sync::CancellationToken;
      22              : use tracing::{Instrument, info_span};
      23              : use utils::backoff;
      24              : use utils::generation::Generation;
      25              : use utils::id::{TenantId, TenantTimelineId};
      26              : 
      27              : use crate::checks::{
      28              :     BlobDataParseResult, ListTenantManifestResult, RemoteTenantManifestInfo, list_tenant_manifests,
      29              :     list_timeline_blobs,
      30              : };
      31              : use crate::metadata_stream::{stream_tenant_timelines, stream_tenants};
      32              : use crate::{BucketConfig, MAX_RETRIES, NodeKind, RootTarget, TenantShardTimelineId, init_remote};
      33              : 
      34              : #[derive(Serialize, Default)]
      35              : pub struct GcSummary {
      36              :     indices_deleted: usize,
      37              :     tenant_manifests_deleted: usize,
      38              :     remote_storage_errors: usize,
      39              :     controller_api_errors: usize,
      40              :     ancestor_layers_deleted: usize,
      41              : }
      42              : 
      43              : impl GcSummary {
      44            0 :     fn merge(&mut self, other: Self) {
      45            0 :         let Self {
      46            0 :             indices_deleted,
      47            0 :             tenant_manifests_deleted,
      48            0 :             remote_storage_errors,
      49            0 :             ancestor_layers_deleted,
      50            0 :             controller_api_errors,
      51            0 :         } = other;
      52            0 : 
      53            0 :         self.indices_deleted += indices_deleted;
      54            0 :         self.tenant_manifests_deleted += tenant_manifests_deleted;
      55            0 :         self.remote_storage_errors += remote_storage_errors;
      56            0 :         self.ancestor_layers_deleted += ancestor_layers_deleted;
      57            0 :         self.controller_api_errors += controller_api_errors;
      58            0 :     }
      59              : }
      60              : 
      61              : #[derive(clap::ValueEnum, Debug, Clone, Copy)]
      62              : pub enum GcMode {
      63              :     // Delete nothing
      64              :     DryRun,
      65              : 
      66              :     // Enable only removing old-generation indices
      67              :     IndicesOnly,
      68              : 
      69              :     // Enable all forms of GC
      70              :     Full,
      71              : }
      72              : 
      73              : impl std::fmt::Display for GcMode {
      74            0 :     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
      75            0 :         match self {
      76            0 :             GcMode::DryRun => write!(f, "dry-run"),
      77            0 :             GcMode::IndicesOnly => write!(f, "indices-only"),
      78            0 :             GcMode::Full => write!(f, "full"),
      79              :         }
      80            0 :     }
      81              : }
      82              : 
      83              : mod refs {
      84              :     use super::*;
      85              :     // Map of cross-shard layer references, giving a refcount for each layer in each shard that is referenced by some other
      86              :     // shard in the same tenant.  This is sparse!  The vast majority of timelines will have no cross-shard refs, and those that
      87              :     // do have cross shard refs should eventually drop most of them via compaction.
      88              :     //
      89              :     // In our inner map type, the TTID in the key is shard-agnostic, and the ShardIndex in the value refers to the _ancestor
      90              :     // which is is referenced_.
      91              :     #[derive(Default)]
      92              :     pub(super) struct AncestorRefs(
      93              :         BTreeMap<TenantTimelineId, HashMap<(ShardIndex, LayerName), usize>>,
      94              :     );
      95              : 
      96              :     impl AncestorRefs {
      97              :         /// Insert references for layers discovered in a particular shard-timeline that refer to an ancestral shard-timeline.
      98            0 :         pub(super) fn update(
      99            0 :             &mut self,
     100            0 :             ttid: TenantShardTimelineId,
     101            0 :             layers: Vec<(LayerName, LayerFileMetadata)>,
     102            0 :         ) {
     103            0 :             let ttid_refs = self.0.entry(ttid.as_tenant_timeline_id()).or_default();
     104            0 :             for (layer_name, layer_metadata) in layers {
     105            0 :                 // Increment refcount of this layer in the ancestor shard
     106            0 :                 *(ttid_refs
     107            0 :                     .entry((layer_metadata.shard, layer_name))
     108            0 :                     .or_default()) += 1;
     109            0 :             }
     110            0 :         }
     111              : 
     112              :         /// For a particular TTID, return the map of all ancestor layers referenced by a descendent to their refcount
     113              :         ///
     114              :         /// The `ShardIndex` in the result's key is the index of the _ancestor_, not the descendent.
     115            0 :         pub(super) fn get_ttid_refcounts(
     116            0 :             &self,
     117            0 :             ttid: &TenantTimelineId,
     118            0 :         ) -> Option<&HashMap<(ShardIndex, LayerName), usize>> {
     119            0 :             self.0.get(ttid)
     120            0 :         }
     121              :     }
     122              : }
     123              : 
     124              : use refs::AncestorRefs;
     125              : 
     126              : // As we see shards for a tenant, acccumulate knowledge needed for cross-shard GC:
     127              : // - Are there any ancestor shards?
     128              : // - Are there any refs to ancestor shards' layers?
     129              : #[derive(Default)]
     130              : struct TenantRefAccumulator {
     131              :     shards_seen: HashMap<TenantId, BTreeSet<ShardIndex>>,
     132              : 
     133              :     // For each shard that has refs to an ancestor's layers, the set of ancestor layers referred to
     134              :     ancestor_ref_shards: AncestorRefs,
     135              : }
     136              : 
     137              : impl TenantRefAccumulator {
     138            0 :     fn update(&mut self, ttid: TenantShardTimelineId, index_part: &IndexPart) {
     139            0 :         let this_shard_idx = ttid.tenant_shard_id.to_index();
     140            0 :         (*self
     141            0 :             .shards_seen
     142            0 :             .entry(ttid.tenant_shard_id.tenant_id)
     143            0 :             .or_default())
     144            0 :         .insert(this_shard_idx);
     145            0 : 
     146            0 :         let mut ancestor_refs = Vec::new();
     147            0 :         for (layer_name, layer_metadata) in &index_part.layer_metadata {
     148            0 :             if layer_metadata.shard != this_shard_idx {
     149            0 :                 // This is a reference from this shard to a layer in an ancestor shard: we must track this
     150            0 :                 // as a marker to not GC this layer from the parent.
     151            0 :                 ancestor_refs.push((layer_name.clone(), layer_metadata.clone()));
     152            0 :             }
     153              :         }
     154              : 
     155            0 :         tracing::info!(%ttid, "Found {} ancestor refs", ancestor_refs.len());
     156            0 :         self.ancestor_ref_shards.update(ttid, ancestor_refs);
     157            0 :     }
     158              : 
     159              :     /// Consume Self and return a vector of ancestor tenant shards that should be GC'd, and map of referenced ancestor layers to preserve
     160            0 :     async fn into_gc_ancestors(
     161            0 :         self,
     162            0 :         controller_client: &control_api::Client,
     163            0 :         summary: &mut GcSummary,
     164            0 :     ) -> (Vec<TenantShardId>, AncestorRefs) {
     165            0 :         let mut ancestors_to_gc = Vec::new();
     166            0 :         for (tenant_id, shard_indices) in self.shards_seen {
     167              :             // Find the highest shard count
     168            0 :             let latest_count = shard_indices
     169            0 :                 .iter()
     170            0 :                 .map(|i| i.shard_count)
     171            0 :                 .max()
     172            0 :                 .expect("Always at least one shard");
     173            0 : 
     174            0 :             let mut shard_indices = shard_indices.iter().collect::<Vec<_>>();
     175            0 :             let (mut latest_shards, ancestor_shards) = {
     176            0 :                 let at =
     177            0 :                     itertools::partition(&mut shard_indices, |i| i.shard_count == latest_count);
     178            0 :                 (shard_indices[0..at].to_owned(), &shard_indices[at..])
     179            0 :             };
     180            0 :             // Sort shards, as we will later compare them with a sorted list from the controller
     181            0 :             latest_shards.sort();
     182            0 : 
     183            0 :             // Check that we have a complete view of the latest shard count: this should always be the case unless we happened
     184            0 :             // to scan the S3 bucket halfway through a shard split.
     185            0 :             if latest_shards.len() != latest_count.count() as usize {
     186              :                 // This should be extremely rare, so we warn on it.
     187            0 :                 tracing::warn!(%tenant_id, "Missed some shards at count {:?}: {latest_shards:?}", latest_count);
     188            0 :                 continue;
     189            0 :             }
     190            0 : 
     191            0 :             // Check if we have any non-latest-count shards
     192            0 :             if ancestor_shards.is_empty() {
     193            0 :                 tracing::debug!(%tenant_id, "No ancestor shards to clean up");
     194            0 :                 continue;
     195            0 :             }
     196            0 : 
     197            0 :             // Based on S3 view, this tenant looks like it might have some ancestor shard work to do.  We
     198            0 :             // must only do this work if the tenant is not currently being split: otherwise, it is not safe
     199            0 :             // to GC ancestors, because if the split fails then the controller will try to attach ancestor
     200            0 :             // shards again.
     201            0 :             match controller_client
     202            0 :                 .dispatch::<(), TenantDescribeResponse>(
     203            0 :                     Method::GET,
     204            0 :                     format!("control/v1/tenant/{tenant_id}"),
     205            0 :                     None,
     206            0 :                 )
     207            0 :                 .await
     208              :             {
     209            0 :                 Err(e) => {
     210            0 :                     // We were not able to learn the latest shard split state from the controller, so we will not
     211            0 :                     // do ancestor GC on this tenant.
     212            0 :                     tracing::warn!(%tenant_id, "Failed to query storage controller, will not do ancestor GC: {e}");
     213            0 :                     summary.controller_api_errors += 1;
     214            0 :                     continue;
     215              :                 }
     216            0 :                 Ok(desc) => {
     217            0 :                     // We expect to see that the latest shard count matches the one we saw in S3, and that none
     218            0 :                     // of the shards indicate splitting in progress.
     219            0 : 
     220            0 :                     let controller_indices: Vec<ShardIndex> = desc
     221            0 :                         .shards
     222            0 :                         .iter()
     223            0 :                         .map(|s| s.tenant_shard_id.to_index())
     224            0 :                         .collect();
     225            0 :                     if !controller_indices.iter().eq(latest_shards.iter().copied()) {
     226            0 :                         tracing::info!(%tenant_id, "Latest shards seen in S3 ({latest_shards:?}) don't match controller state ({controller_indices:?})");
     227            0 :                         continue;
     228            0 :                     }
     229            0 : 
     230            0 :                     if desc.shards.iter().any(|s| s.is_splitting) {
     231            0 :                         tracing::info!(%tenant_id, "One or more shards is currently splitting");
     232            0 :                         continue;
     233            0 :                     }
     234            0 : 
     235            0 :                     // This shouldn't be too noisy, because we only log this for tenants that have some ancestral refs.
     236            0 :                     tracing::info!(%tenant_id, "Validated state with controller: {desc:?}");
     237              :                 }
     238              :             }
     239              : 
     240              :             // GC ancestor shards
     241            0 :             for ancestor_shard in ancestor_shards.iter().map(|idx| TenantShardId {
     242            0 :                 tenant_id,
     243            0 :                 shard_count: idx.shard_count,
     244            0 :                 shard_number: idx.shard_number,
     245            0 :             }) {
     246            0 :                 ancestors_to_gc.push(ancestor_shard);
     247            0 :             }
     248              :         }
     249              : 
     250            0 :         (ancestors_to_gc, self.ancestor_ref_shards)
     251            0 :     }
     252              : }
     253              : 
     254            0 : fn is_old_enough(min_age: &Duration, key: &ListingObject, summary: &mut GcSummary) -> bool {
     255              :     // Validation: we will only GC indices & layers after a time threshold (e.g. one week) so that during an incident
     256              :     // it is easier to read old data for analysis, and easier to roll back shard splits without having to un-delete any objects.
     257            0 :     let age = match key.last_modified.elapsed() {
     258            0 :         Ok(e) => e,
     259              :         Err(_) => {
     260            0 :             tracing::warn!("Bad last_modified time: {:?}", key.last_modified);
     261            0 :             summary.remote_storage_errors += 1;
     262            0 :             return false;
     263              :         }
     264              :     };
     265            0 :     let old_enough = &age > min_age;
     266            0 : 
     267            0 :     if !old_enough {
     268            0 :         tracing::info!(
     269            0 :             "Skipping young object {} < {}",
     270            0 :             humantime::format_duration(age),
     271            0 :             humantime::format_duration(*min_age)
     272              :         );
     273            0 :     }
     274              : 
     275            0 :     old_enough
     276            0 : }
     277              : 
     278              : /// Same as [`is_old_enough`], but doesn't require a [`ListingObject`] passed to it.
     279            0 : async fn check_is_old_enough(
     280            0 :     remote_client: &GenericRemoteStorage,
     281            0 :     key: &RemotePath,
     282            0 :     min_age: &Duration,
     283            0 :     summary: &mut GcSummary,
     284            0 : ) -> Option<bool> {
     285            0 :     let listing_object = remote_client
     286            0 :         .head_object(key, &CancellationToken::new())
     287            0 :         .await
     288            0 :         .ok()?;
     289            0 :     Some(is_old_enough(min_age, &listing_object, summary))
     290            0 : }
     291              : 
     292            0 : async fn maybe_delete_index(
     293            0 :     remote_client: &GenericRemoteStorage,
     294            0 :     min_age: &Duration,
     295            0 :     latest_gen: Generation,
     296            0 :     obj: &ListingObject,
     297            0 :     mode: GcMode,
     298            0 :     summary: &mut GcSummary,
     299            0 : ) {
     300            0 :     // Validation: we will only delete things that parse cleanly
     301            0 :     let basename = obj.key.get_path().file_name().unwrap();
     302            0 :     let candidate_generation =
     303            0 :         match parse_remote_index_path(RemotePath::from_string(basename).unwrap()) {
     304            0 :             Some(g) => g,
     305              :             None => {
     306            0 :                 if basename == IndexPart::FILE_NAME {
     307              :                     // A legacy pre-generation index
     308            0 :                     Generation::none()
     309              :                 } else {
     310              :                     // A strange key: we will not delete this because we don't understand it.
     311            0 :                     tracing::warn!("Bad index key");
     312            0 :                     return;
     313              :                 }
     314              :             }
     315              :         };
     316              : 
     317              :     // Validation: we will only delete indices more than one generation old, to avoid interfering
     318              :     // in typical migrations, even if they are very long running.
     319            0 :     if candidate_generation >= latest_gen {
     320              :         // This shouldn't happen: when we loaded metadata, it should have selected the latest
     321              :         // generation already, and only populated [`S3TimelineBlobData::unused_index_keys`]
     322              :         // with older generations.
     323            0 :         tracing::warn!("Deletion candidate is >= latest generation, this is a bug!");
     324            0 :         return;
     325            0 :     } else if candidate_generation.next() == latest_gen {
     326              :         // Skip deleting the latest-1th generation's index.
     327            0 :         return;
     328            0 :     }
     329            0 : 
     330            0 :     if !is_old_enough(min_age, obj, summary) {
     331            0 :         return;
     332            0 :     }
     333              : 
     334            0 :     if matches!(mode, GcMode::DryRun) {
     335            0 :         tracing::info!("Dry run: would delete this key");
     336            0 :         return;
     337            0 :     }
     338            0 : 
     339            0 :     // All validations passed: erase the object
     340            0 :     let cancel = CancellationToken::new();
     341            0 :     match backoff::retry(
     342            0 :         || remote_client.delete(&obj.key, &cancel),
     343            0 :         |_| false,
     344            0 :         3,
     345            0 :         MAX_RETRIES as u32,
     346            0 :         "maybe_delete_index",
     347            0 :         &cancel,
     348            0 :     )
     349            0 :     .await
     350              :     {
     351              :         None => {
     352            0 :             unreachable!("Using a dummy cancellation token");
     353              :         }
     354              :         Some(Ok(_)) => {
     355            0 :             tracing::info!("Successfully deleted index");
     356            0 :             summary.indices_deleted += 1;
     357              :         }
     358            0 :         Some(Err(e)) => {
     359            0 :             tracing::warn!("Failed to delete index: {e}");
     360            0 :             summary.remote_storage_errors += 1;
     361              :         }
     362              :     }
     363            0 : }
     364              : 
     365            0 : async fn maybe_delete_tenant_manifest(
     366            0 :     remote_client: &GenericRemoteStorage,
     367            0 :     min_age: &Duration,
     368            0 :     latest_gen: Generation,
     369            0 :     obj: &ListingObject,
     370            0 :     mode: GcMode,
     371            0 :     summary: &mut GcSummary,
     372            0 : ) {
     373            0 :     // Validation: we will only delete things that parse cleanly
     374            0 :     let basename = obj.key.get_path().file_name().unwrap();
     375            0 :     let Some(candidate_generation) =
     376            0 :         parse_remote_tenant_manifest_path(RemotePath::from_string(basename).unwrap())
     377              :     else {
     378              :         // A strange key: we will not delete this because we don't understand it.
     379            0 :         tracing::warn!("Bad index key");
     380            0 :         return;
     381              :     };
     382              : 
     383              :     // Validation: we will only delete manifests more than one generation old, and in fact we
     384              :     // should never be called with such recent generations.
     385            0 :     if candidate_generation >= latest_gen {
     386            0 :         tracing::warn!("Deletion candidate is >= latest generation, this is a bug!");
     387            0 :         return;
     388            0 :     } else if candidate_generation.next() == latest_gen {
     389            0 :         tracing::warn!("Deletion candidate is >= latest generation - 1, this is a bug!");
     390            0 :         return;
     391            0 :     }
     392            0 : 
     393            0 :     if !is_old_enough(min_age, obj, summary) {
     394            0 :         return;
     395            0 :     }
     396              : 
     397            0 :     if matches!(mode, GcMode::DryRun) {
     398            0 :         tracing::info!("Dry run: would delete this key");
     399            0 :         return;
     400            0 :     }
     401            0 : 
     402            0 :     // All validations passed: erase the object
     403            0 :     let cancel = CancellationToken::new();
     404            0 :     match backoff::retry(
     405            0 :         || remote_client.delete(&obj.key, &cancel),
     406            0 :         |_| false,
     407            0 :         3,
     408            0 :         MAX_RETRIES as u32,
     409            0 :         "maybe_delete_tenant_manifest",
     410            0 :         &cancel,
     411            0 :     )
     412            0 :     .await
     413              :     {
     414              :         None => {
     415            0 :             unreachable!("Using a dummy cancellation token");
     416              :         }
     417              :         Some(Ok(_)) => {
     418            0 :             tracing::info!("Successfully deleted tenant manifest");
     419            0 :             summary.tenant_manifests_deleted += 1;
     420              :         }
     421            0 :         Some(Err(e)) => {
     422            0 :             tracing::warn!("Failed to delete tenant manifest: {e}");
     423            0 :             summary.remote_storage_errors += 1;
     424              :         }
     425              :     }
     426            0 : }
     427              : 
     428              : #[allow(clippy::too_many_arguments)]
     429            0 : async fn gc_ancestor(
     430            0 :     remote_client: &GenericRemoteStorage,
     431            0 :     root_target: &RootTarget,
     432            0 :     min_age: &Duration,
     433            0 :     ancestor: TenantShardId,
     434            0 :     refs: &AncestorRefs,
     435            0 :     mode: GcMode,
     436            0 :     summary: &mut GcSummary,
     437            0 : ) -> anyhow::Result<()> {
     438              :     // Scan timelines in the ancestor
     439            0 :     let timelines = stream_tenant_timelines(remote_client, root_target, ancestor).await?;
     440            0 :     let mut timelines = std::pin::pin!(timelines);
     441              : 
     442              :     // Build a list of keys to retain
     443              : 
     444            0 :     while let Some(ttid) = timelines.next().await {
     445            0 :         let ttid = ttid?;
     446              : 
     447            0 :         let data = list_timeline_blobs(remote_client, ttid, root_target).await?;
     448              : 
     449            0 :         let s3_layers = match data.blob_data {
     450              :             BlobDataParseResult::Parsed {
     451              :                 index_part: _,
     452              :                 index_part_generation: _,
     453            0 :                 s3_layers,
     454            0 :                 index_part_last_modified_time: _,
     455            0 :                 index_part_snapshot_time: _,
     456            0 :             } => s3_layers,
     457              :             BlobDataParseResult::Relic => {
     458              :                 // Post-deletion tenant location: don't try and GC it.
     459            0 :                 continue;
     460              :             }
     461              :             BlobDataParseResult::Incorrect {
     462            0 :                 errors,
     463            0 :                 s3_layers: _, // TODO(yuchen): could still check references to these s3 layers?
     464            0 :             } => {
     465            0 :                 // Our primary purpose isn't to report on bad data, but log this rather than skipping silently
     466            0 :                 tracing::warn!(
     467            0 :                     "Skipping ancestor GC for timeline {ttid}, bad metadata: {errors:?}"
     468              :                 );
     469            0 :                 continue;
     470              :             }
     471              :         };
     472              : 
     473            0 :         let ttid_refs = refs.get_ttid_refcounts(&ttid.as_tenant_timeline_id());
     474            0 :         let ancestor_shard_index = ttid.tenant_shard_id.to_index();
     475              : 
     476            0 :         for (layer_name, layer_gen) in s3_layers {
     477            0 :             let ref_count = ttid_refs
     478            0 :                 .and_then(|m| m.get(&(ancestor_shard_index, layer_name.clone())))
     479            0 :                 .copied()
     480            0 :                 .unwrap_or(0);
     481            0 : 
     482            0 :             if ref_count > 0 {
     483            0 :                 tracing::debug!(%ttid, "Ancestor layer {layer_name}  has {ref_count} refs");
     484            0 :                 continue;
     485            0 :             }
     486            0 : 
     487            0 :             tracing::info!(%ttid, "Ancestor layer {layer_name} is not referenced");
     488              : 
     489              :             // Build the key for the layer we are considering deleting
     490            0 :             let key = root_target.absolute_key(&remote_layer_path(
     491            0 :                 &ttid.tenant_shard_id.tenant_id,
     492            0 :                 &ttid.timeline_id,
     493            0 :                 ancestor_shard_index,
     494            0 :                 &layer_name,
     495            0 :                 layer_gen,
     496            0 :             ));
     497            0 : 
     498            0 :             // We apply a time threshold to GCing objects that are un-referenced: this preserves our ability
     499            0 :             // to roll back a shard split if we have to, by avoiding deleting ancestor layers right away
     500            0 :             let path = RemotePath::from_string(key.strip_prefix("/").unwrap_or(&key)).unwrap();
     501            0 :             if check_is_old_enough(remote_client, &path, min_age, summary).await != Some(true) {
     502            0 :                 continue;
     503            0 :             }
     504              : 
     505            0 :             if !matches!(mode, GcMode::Full) {
     506            0 :                 tracing::info!("Dry run: would delete key {key}");
     507            0 :                 continue;
     508            0 :             }
     509            0 : 
     510            0 :             // All validations passed: erase the object
     511            0 :             match remote_client.delete(&path, &CancellationToken::new()).await {
     512              :                 Ok(_) => {
     513            0 :                     tracing::info!("Successfully deleted unreferenced ancestor layer {key}");
     514            0 :                     summary.ancestor_layers_deleted += 1;
     515              :                 }
     516            0 :                 Err(e) => {
     517            0 :                     tracing::warn!("Failed to delete layer {key}: {e}");
     518            0 :                     summary.remote_storage_errors += 1;
     519              :                 }
     520              :             }
     521              :         }
     522              : 
     523              :         // TODO: if all the layers are gone, clean up the whole timeline dir (remove index)
     524              :     }
     525              : 
     526            0 :     Ok(())
     527            0 : }
     528              : 
     529            0 : async fn gc_tenant_manifests(
     530            0 :     remote_client: &GenericRemoteStorage,
     531            0 :     min_age: Duration,
     532            0 :     target: &RootTarget,
     533            0 :     mode: GcMode,
     534            0 :     tenant_shard_id: TenantShardId,
     535            0 : ) -> anyhow::Result<(GcSummary, Option<RemoteTenantManifestInfo>)> {
     536            0 :     let mut gc_summary = GcSummary::default();
     537            0 :     match list_tenant_manifests(remote_client, tenant_shard_id, target).await? {
     538              :         ListTenantManifestResult::WithErrors {
     539            0 :             errors,
     540              :             unknown_keys: _,
     541              :         } => {
     542            0 :             for (_key, error) in errors {
     543            0 :                 tracing::warn!(%tenant_shard_id, "list_tenant_manifests: {error}");
     544              :             }
     545            0 :             Ok((gc_summary, None))
     546              :         }
     547              :         ListTenantManifestResult::NoErrors {
     548            0 :             latest_generation,
     549            0 :             mut manifests,
     550              :         } => {
     551            0 :             let Some(latest_generation) = latest_generation else {
     552            0 :                 return Ok((gc_summary, None));
     553              :             };
     554            0 :             manifests.sort_by_key(|(generation, _obj)| *generation);
     555            0 :             // skip the two latest generations (they don't neccessarily have to be 1 apart from each other)
     556            0 :             let candidates = manifests.iter().rev().skip(2);
     557            0 :             for (_generation, key) in candidates {
     558            0 :                 maybe_delete_tenant_manifest(
     559            0 :                     remote_client,
     560            0 :                     &min_age,
     561            0 :                     latest_generation.generation,
     562            0 :                     key,
     563            0 :                     mode,
     564            0 :                     &mut gc_summary,
     565            0 :                 )
     566            0 :                 .instrument(
     567            0 :                     info_span!("maybe_delete_tenant_manifest", %tenant_shard_id, ?latest_generation.generation, %key.key),
     568              :                 )
     569            0 :                 .await;
     570              :             }
     571            0 :             Ok((gc_summary, Some(latest_generation)))
     572              :         }
     573              :     }
     574            0 : }
     575              : 
     576            0 : async fn gc_timeline(
     577            0 :     remote_client: &GenericRemoteStorage,
     578            0 :     min_age: &Duration,
     579            0 :     target: &RootTarget,
     580            0 :     mode: GcMode,
     581            0 :     ttid: TenantShardTimelineId,
     582            0 :     accumulator: &std::sync::Mutex<TenantRefAccumulator>,
     583            0 :     tenant_manifest_info: Arc<Option<RemoteTenantManifestInfo>>,
     584            0 : ) -> anyhow::Result<GcSummary> {
     585            0 :     let mut summary = GcSummary::default();
     586            0 :     let data = list_timeline_blobs(remote_client, ttid, target).await?;
     587              : 
     588            0 :     let (index_part, latest_gen, candidates) = match &data.blob_data {
     589              :         BlobDataParseResult::Parsed {
     590            0 :             index_part,
     591            0 :             index_part_generation,
     592            0 :             s3_layers: _,
     593            0 :             index_part_last_modified_time: _,
     594            0 :             index_part_snapshot_time: _,
     595            0 :         } => (index_part, *index_part_generation, data.unused_index_keys),
     596              :         BlobDataParseResult::Relic => {
     597              :             // Post-deletion tenant location: don't try and GC it.
     598            0 :             return Ok(summary);
     599              :         }
     600              :         BlobDataParseResult::Incorrect {
     601            0 :             errors,
     602            0 :             s3_layers: _,
     603            0 :         } => {
     604            0 :             // Our primary purpose isn't to report on bad data, but log this rather than skipping silently
     605            0 :             tracing::warn!("Skipping timeline {ttid}, bad metadata: {errors:?}");
     606            0 :             return Ok(summary);
     607              :         }
     608              :     };
     609              : 
     610            0 :     if let Some(tenant_manifest_info) = &*tenant_manifest_info {
     611              :         // TODO: this is O(n^2) in the number of offloaded timelines. Do a hashmap lookup instead.
     612            0 :         let maybe_offloaded = tenant_manifest_info
     613            0 :             .manifest
     614            0 :             .offloaded_timelines
     615            0 :             .iter()
     616            0 :             .find(|offloaded_timeline| offloaded_timeline.timeline_id == ttid.timeline_id);
     617            0 :         if let Some(offloaded) = maybe_offloaded {
     618            0 :             let warnings = validate_index_part_with_offloaded(index_part, offloaded);
     619            0 :             let warn = if warnings.is_empty() {
     620            0 :                 false
     621              :             } else {
     622              :                 // Verify that the manifest hasn't changed. If it has, a potential racing change could have been cause for our troubles.
     623            0 :                 match list_tenant_manifests(remote_client, ttid.tenant_shard_id, target).await? {
     624              :                     ListTenantManifestResult::WithErrors {
     625            0 :                         errors,
     626              :                         unknown_keys: _,
     627              :                     } => {
     628            0 :                         for (_key, error) in errors {
     629            0 :                             tracing::warn!(%ttid, "list_tenant_manifests in gc_timeline: {error}");
     630              :                         }
     631            0 :                         true
     632              :                     }
     633              :                     ListTenantManifestResult::NoErrors {
     634            0 :                         latest_generation,
     635              :                         manifests: _,
     636              :                     } => {
     637            0 :                         if let Some(new_latest_gen) = latest_generation {
     638            0 :                             let manifest_changed = (
     639            0 :                                 new_latest_gen.generation,
     640            0 :                                 new_latest_gen.listing_object.last_modified,
     641            0 :                             ) == (
     642            0 :                                 tenant_manifest_info.generation,
     643            0 :                                 tenant_manifest_info.listing_object.last_modified,
     644            0 :                             );
     645            0 :                             if manifest_changed {
     646            0 :                                 tracing::debug!(%ttid, "tenant manifest changed since it was loaded, suppressing {} warnings", warnings.len());
     647            0 :                             }
     648            0 :                             manifest_changed
     649              :                         } else {
     650              :                             // The latest generation is gone. This timeline is in the progress of being deleted?
     651            0 :                             false
     652              :                         }
     653              :                     }
     654              :                 }
     655              :             };
     656            0 :             if warn {
     657            0 :                 for warning in warnings {
     658            0 :                     tracing::warn!(%ttid, "{}", warning);
     659              :                 }
     660            0 :             }
     661            0 :         }
     662            0 :     }
     663              : 
     664            0 :     accumulator.lock().unwrap().update(ttid, index_part);
     665              : 
     666            0 :     for key in candidates {
     667            0 :         maybe_delete_index(remote_client, min_age, latest_gen, &key, mode, &mut summary)
     668            0 :             .instrument(info_span!("maybe_delete_index", %ttid, ?latest_gen, %key.key))
     669            0 :             .await;
     670              :     }
     671              : 
     672            0 :     Ok(summary)
     673            0 : }
     674              : 
     675            0 : fn validate_index_part_with_offloaded(
     676            0 :     index_part: &IndexPart,
     677            0 :     offloaded: &OffloadedTimelineManifest,
     678            0 : ) -> Vec<String> {
     679            0 :     let mut warnings = Vec::new();
     680            0 :     if let Some(archived_at_index_part) = index_part.archived_at {
     681            0 :         if archived_at_index_part
     682            0 :             .signed_duration_since(offloaded.archived_at)
     683            0 :             .num_seconds()
     684            0 :             != 0
     685            0 :         {
     686            0 :             warnings.push(format!(
     687            0 :                 "index-part archived_at={} differs from manifest archived_at={}",
     688            0 :                 archived_at_index_part, offloaded.archived_at
     689            0 :             ));
     690            0 :         }
     691            0 :     } else {
     692            0 :         warnings.push("Timeline offloaded in manifest but not archived in index-part".to_string());
     693            0 :     }
     694            0 :     if index_part.metadata.ancestor_timeline() != offloaded.ancestor_timeline_id {
     695            0 :         warnings.push(format!(
     696            0 :             "index-part anestor={:?} differs from manifest ancestor={:?}",
     697            0 :             index_part.metadata.ancestor_timeline(),
     698            0 :             offloaded.ancestor_timeline_id
     699            0 :         ));
     700            0 :     }
     701            0 :     warnings
     702            0 : }
     703              : 
     704              : /// Physical garbage collection: removing unused S3 objects.
     705              : ///
     706              : /// This is distinct from the garbage collection done inside the pageserver, which operates at a higher level
     707              : /// (keys, layers).  This type of garbage collection is about removing:
     708              : /// - Objects that were uploaded but never referenced in the remote index (e.g. because of a shutdown between
     709              : ///   uploading a layer and uploading an index)
     710              : /// - Index objects and tenant manifests from historic generations
     711              : ///
     712              : /// This type of GC is not necessary for correctness: rather it serves to reduce wasted storage capacity, and
     713              : /// make sure that object listings don't get slowed down by large numbers of garbage objects.
     714            0 : pub async fn pageserver_physical_gc(
     715            0 :     bucket_config: &BucketConfig,
     716            0 :     controller_client: Option<&control_api::Client>,
     717            0 :     tenant_shard_ids: Vec<TenantShardId>,
     718            0 :     min_age: Duration,
     719            0 :     mode: GcMode,
     720            0 : ) -> anyhow::Result<GcSummary> {
     721            0 :     let (remote_client, target) = init_remote(bucket_config.clone(), NodeKind::Pageserver).await?;
     722              : 
     723            0 :     let remote_client = Arc::new(remote_client);
     724            0 :     let tenants = if tenant_shard_ids.is_empty() {
     725            0 :         Either::Left(stream_tenants(&remote_client, &target))
     726              :     } else {
     727            0 :         Either::Right(futures::stream::iter(tenant_shard_ids.into_iter().map(Ok)))
     728              :     };
     729              : 
     730              :     // How many tenants to process in parallel.  We need to be mindful of pageservers
     731              :     // accessing the same per tenant prefixes, so use a lower setting than pageservers.
     732              :     const CONCURRENCY: usize = 32;
     733              : 
     734              :     // Accumulate information about each tenant for cross-shard GC step we'll do at the end
     735            0 :     let accumulator = std::sync::Mutex::new(TenantRefAccumulator::default());
     736            0 : 
     737            0 :     // Accumulate information about how many manifests we have GCd
     738            0 :     let manifest_gc_summary = std::sync::Mutex::new(GcSummary::default());
     739            0 : 
     740            0 :     // Generate a stream of TenantTimelineId
     741            0 :     let timelines = tenants.map_ok(|tenant_shard_id| {
     742            0 :         let target_ref = &target;
     743            0 :         let remote_client_ref = &remote_client;
     744            0 :         let manifest_gc_summary_ref = &manifest_gc_summary;
     745            0 :         async move {
     746            0 :             let gc_manifest_result = gc_tenant_manifests(
     747            0 :                 remote_client_ref,
     748            0 :                 min_age,
     749            0 :                 target_ref,
     750            0 :                 mode,
     751            0 :                 tenant_shard_id,
     752            0 :             )
     753            0 :             .await;
     754            0 :             let (summary_from_manifest, tenant_manifest_opt) = match gc_manifest_result {
     755            0 :                 Ok((gc_summary, tenant_manifest)) => (gc_summary, tenant_manifest),
     756            0 :                 Err(e) => {
     757            0 :                     tracing::warn!(%tenant_shard_id, "Error in gc_tenant_manifests: {e}");
     758            0 :                     (GcSummary::default(), None)
     759              :                 }
     760              :             };
     761            0 :             manifest_gc_summary_ref
     762            0 :                 .lock()
     763            0 :                 .unwrap()
     764            0 :                 .merge(summary_from_manifest);
     765            0 :             let tenant_manifest_arc = Arc::new(tenant_manifest_opt);
     766            0 :             let mut timelines = Box::pin(
     767            0 :                 stream_tenant_timelines(remote_client_ref, target_ref, tenant_shard_id).await?,
     768              :             );
     769            0 :             Ok(try_stream! {
     770            0 :                 while let Some(ttid_res) = timelines.next().await {
     771            0 :                     let ttid = ttid_res?;
     772            0 :                     yield (ttid, tenant_manifest_arc.clone());
     773            0 :                 }
     774            0 :             })
     775            0 :         }
     776            0 :     });
     777            0 : 
     778            0 :     let mut summary = GcSummary::default();
     779            0 :     {
     780            0 :         let timelines = timelines.try_buffered(CONCURRENCY);
     781            0 :         let timelines = timelines.try_flatten();
     782            0 : 
     783            0 :         let timelines = timelines.map_ok(|(ttid, tenant_manifest_arc)| {
     784            0 :             gc_timeline(
     785            0 :                 &remote_client,
     786            0 :                 &min_age,
     787            0 :                 &target,
     788            0 :                 mode,
     789            0 :                 ttid,
     790            0 :                 &accumulator,
     791            0 :                 tenant_manifest_arc,
     792            0 :             )
     793            0 :         });
     794            0 :         let timelines = timelines.try_buffered(CONCURRENCY);
     795            0 :         let mut timelines = std::pin::pin!(timelines);
     796              :         // Drain futures for per-shard GC, populating accumulator as a side effect
     797            0 :         while let Some(i) = timelines.next().await {
     798            0 :             summary.merge(i?);
     799              :         }
     800              :     }
     801              :     // Streams are lazily evaluated, so only now do we have access to the inner object
     802            0 :     summary.merge(manifest_gc_summary.into_inner().unwrap());
     803              : 
     804              :     // Execute cross-shard GC, using the accumulator's full view of all the shards built in the per-shard GC
     805            0 :     let Some(client) = controller_client else {
     806            0 :         tracing::info!("Skipping ancestor layer GC, because no `--controller-api` was specified");
     807            0 :         return Ok(summary);
     808              :     };
     809              : 
     810            0 :     let (ancestor_shards, ancestor_refs) = accumulator
     811            0 :         .into_inner()
     812            0 :         .unwrap()
     813            0 :         .into_gc_ancestors(client, &mut summary)
     814            0 :         .await;
     815              : 
     816            0 :     for ancestor_shard in ancestor_shards {
     817            0 :         gc_ancestor(
     818            0 :             &remote_client,
     819            0 :             &target,
     820            0 :             &min_age,
     821            0 :             ancestor_shard,
     822            0 :             &ancestor_refs,
     823            0 :             mode,
     824            0 :             &mut summary,
     825            0 :         )
     826            0 :         .instrument(info_span!("gc_ancestor", %ancestor_shard))
     827            0 :         .await?;
     828              :     }
     829              : 
     830            0 :     Ok(summary)
     831            0 : }
        

Generated by: LCOV version 2.1-beta