LCOV - code coverage report
Current view: top level - pageserver/src/tenant/remote_timeline_client - upload.rs (source / functions) Coverage Total Hit
Test: 6df3fc19ec669bcfbbf9aba41d1338898d24eaa0.info Lines: 45.0 % 191 86
Test Date: 2025-03-12 18:28:53 Functions: 18.2 % 33 6

            Line data    Source code
       1              : //! Helper functions to upload files to remote storage with a RemoteStorage
       2              : 
       3              : use std::io::{ErrorKind, SeekFrom};
       4              : use std::time::SystemTime;
       5              : 
       6              : use anyhow::{Context, bail};
       7              : use bytes::Bytes;
       8              : use camino::Utf8Path;
       9              : use fail::fail_point;
      10              : use pageserver_api::shard::TenantShardId;
      11              : use remote_storage::{GenericRemoteStorage, RemotePath, TimeTravelError};
      12              : use tokio::fs::{self, File};
      13              : use tokio::io::AsyncSeekExt;
      14              : use tokio_util::sync::CancellationToken;
      15              : use tracing::info;
      16              : use utils::id::{TenantId, TimelineId};
      17              : use utils::{backoff, pausable_failpoint};
      18              : 
      19              : use super::Generation;
      20              : use super::index::IndexPart;
      21              : use super::manifest::TenantManifest;
      22              : use crate::tenant::remote_timeline_client::{
      23              :     remote_index_path, remote_initdb_archive_path, remote_initdb_preserved_archive_path,
      24              :     remote_tenant_manifest_path,
      25              : };
      26              : 
      27              : /// Serializes and uploads the given index part data to the remote storage.
      28         2999 : pub(crate) async fn upload_index_part(
      29         2999 :     storage: &GenericRemoteStorage,
      30         2999 :     tenant_shard_id: &TenantShardId,
      31         2999 :     timeline_id: &TimelineId,
      32         2999 :     generation: Generation,
      33         2999 :     index_part: &IndexPart,
      34         2999 :     cancel: &CancellationToken,
      35         2999 : ) -> anyhow::Result<()> {
      36         2999 :     tracing::trace!("uploading new index part");
      37              : 
      38         2999 :     fail_point!("before-upload-index", |_| {
      39            0 :         bail!("failpoint before-upload-index")
      40         2999 :     });
      41         2999 :     pausable_failpoint!("before-upload-index-pausable");
      42              : 
      43              :     // Safety: refuse to persist invalid index metadata, to mitigate the impact of any bug that produces this
      44              :     // (this should never happen)
      45         2997 :     index_part.validate().map_err(|e| anyhow::anyhow!(e))?;
      46              : 
      47              :     // FIXME: this error comes too late
      48         2997 :     let serialized = index_part.to_json_bytes()?;
      49         2997 :     let serialized = Bytes::from(serialized);
      50         2997 : 
      51         2997 :     let index_part_size = serialized.len();
      52         2997 : 
      53         2997 :     let remote_path = remote_index_path(tenant_shard_id, timeline_id, generation);
      54         2997 :     storage
      55         2997 :         .upload_storage_object(
      56         2997 :             futures::stream::once(futures::future::ready(Ok(serialized))),
      57         2997 :             index_part_size,
      58         2997 :             &remote_path,
      59         2997 :             cancel,
      60         2997 :         )
      61         2997 :         .await
      62         2981 :         .with_context(|| format!("upload index part for '{tenant_shard_id} / {timeline_id}'"))
      63         2981 : }
      64              : /// Serializes and uploads the given tenant manifest data to the remote storage.
      65            4 : pub(crate) async fn upload_tenant_manifest(
      66            4 :     storage: &GenericRemoteStorage,
      67            4 :     tenant_shard_id: &TenantShardId,
      68            4 :     generation: Generation,
      69            4 :     tenant_manifest: &TenantManifest,
      70            4 :     cancel: &CancellationToken,
      71            4 : ) -> anyhow::Result<()> {
      72            4 :     tracing::trace!("uploading new tenant manifest");
      73              : 
      74            4 :     fail_point!("before-upload-manifest", |_| {
      75            0 :         bail!("failpoint before-upload-manifest")
      76            4 :     });
      77            4 :     pausable_failpoint!("before-upload-manifest-pausable");
      78              : 
      79            4 :     let serialized = tenant_manifest.to_json_bytes()?;
      80            4 :     let serialized = Bytes::from(serialized);
      81            4 : 
      82            4 :     let tenant_manifest_site = serialized.len();
      83            4 : 
      84            4 :     let remote_path = remote_tenant_manifest_path(tenant_shard_id, generation);
      85            4 :     storage
      86            4 :         .upload_storage_object(
      87            4 :             futures::stream::once(futures::future::ready(Ok(serialized))),
      88            4 :             tenant_manifest_site,
      89            4 :             &remote_path,
      90            4 :             cancel,
      91            4 :         )
      92            4 :         .await
      93            4 :         .with_context(|| format!("upload tenant manifest for '{tenant_shard_id}'"))
      94            4 : }
      95              : 
      96              : /// Attempts to upload given layer files.
      97              : /// No extra checks for overlapping files is made and any files that are already present remotely will be overwritten, if submitted during the upload.
      98              : ///
      99              : /// On an error, bumps the retries count and reschedules the entire task.
     100         3366 : pub(super) async fn upload_timeline_layer<'a>(
     101         3366 :     storage: &'a GenericRemoteStorage,
     102         3366 :     local_path: &'a Utf8Path,
     103         3366 :     remote_path: &'a RemotePath,
     104         3366 :     metadata_size: u64,
     105         3366 :     cancel: &CancellationToken,
     106         3366 : ) -> anyhow::Result<()> {
     107         3366 :     fail_point!("before-upload-layer", |_| {
     108            0 :         bail!("failpoint before-upload-layer")
     109         3366 :     });
     110              : 
     111         3366 :     pausable_failpoint!("before-upload-layer-pausable");
     112              : 
     113         3196 :     let source_file_res = fs::File::open(&local_path).await;
     114         3116 :     let source_file = match source_file_res {
     115         3116 :         Ok(source_file) => source_file,
     116            0 :         Err(e) if e.kind() == ErrorKind::NotFound => {
     117            0 :             // If we encounter this arm, it wasn't intended, but it's also not
     118            0 :             // a big problem, if it's because the file was deleted before an
     119            0 :             // upload. However, a nonexistent file can also be indicative of
     120            0 :             // something worse, like when a file is scheduled for upload before
     121            0 :             // it has been written to disk yet.
     122            0 :             //
     123            0 :             // This is tested against `test_compaction_delete_before_upload`
     124            0 :             info!(path = %local_path, "File to upload doesn't exist. Likely the file has been deleted and an upload is not required any more.");
     125            0 :             return Ok(());
     126              :         }
     127            0 :         Err(e) => Err(e).with_context(|| format!("open a source file for layer {local_path:?}"))?,
     128              :     };
     129              : 
     130         3116 :     let fs_size = source_file
     131         3116 :         .metadata()
     132         3116 :         .await
     133         3107 :         .with_context(|| format!("get the source file metadata for layer {local_path:?}"))?
     134         3107 :         .len();
     135         3107 : 
     136         3107 :     if metadata_size != fs_size {
     137            0 :         bail!(
     138            0 :             "File {local_path:?} has its current FS size {fs_size} diferent from initially determined {metadata_size}"
     139            0 :         );
     140         3107 :     }
     141              : 
     142         3107 :     let fs_size = usize::try_from(fs_size)
     143         3107 :         .with_context(|| format!("convert {local_path:?} size {fs_size} usize"))?;
     144              : 
     145         3107 :     let reader = tokio_util::io::ReaderStream::with_capacity(source_file, super::BUFFER_SIZE);
     146         3107 : 
     147         3107 :     storage
     148         3107 :         .upload(reader, fs_size, remote_path, None, cancel)
     149         3107 :         .await
     150         2992 :         .with_context(|| format!("upload layer from local path '{local_path}'"))
     151         2992 : }
     152              : 
     153            0 : pub(super) async fn copy_timeline_layer(
     154            0 :     storage: &GenericRemoteStorage,
     155            0 :     source_path: &RemotePath,
     156            0 :     target_path: &RemotePath,
     157            0 :     cancel: &CancellationToken,
     158            0 : ) -> anyhow::Result<()> {
     159            0 :     fail_point!("before-copy-layer", |_| {
     160            0 :         bail!("failpoint before-copy-layer")
     161            0 :     });
     162              : 
     163            0 :     pausable_failpoint!("before-copy-layer-pausable");
     164              : 
     165            0 :     storage
     166            0 :         .copy_object(source_path, target_path, cancel)
     167            0 :         .await
     168            0 :         .with_context(|| format!("copy layer {source_path} to {target_path}"))
     169            0 : }
     170              : 
     171              : /// Uploads the given `initdb` data to the remote storage.
     172            0 : pub(crate) async fn upload_initdb_dir(
     173            0 :     storage: &GenericRemoteStorage,
     174            0 :     tenant_id: &TenantId,
     175            0 :     timeline_id: &TimelineId,
     176            0 :     mut initdb_tar_zst: File,
     177            0 :     size: u64,
     178            0 :     cancel: &CancellationToken,
     179            0 : ) -> anyhow::Result<()> {
     180            0 :     tracing::trace!("uploading initdb dir");
     181              : 
     182              :     // We might have read somewhat into the file already in the prior retry attempt
     183            0 :     initdb_tar_zst.seek(SeekFrom::Start(0)).await?;
     184              : 
     185            0 :     let file = tokio_util::io::ReaderStream::with_capacity(initdb_tar_zst, super::BUFFER_SIZE);
     186            0 : 
     187            0 :     let remote_path = remote_initdb_archive_path(tenant_id, timeline_id);
     188            0 :     storage
     189            0 :         .upload_storage_object(file, size as usize, &remote_path, cancel)
     190            0 :         .await
     191            0 :         .with_context(|| format!("upload initdb dir for '{tenant_id} / {timeline_id}'"))
     192            0 : }
     193              : 
     194            0 : pub(crate) async fn preserve_initdb_archive(
     195            0 :     storage: &GenericRemoteStorage,
     196            0 :     tenant_id: &TenantId,
     197            0 :     timeline_id: &TimelineId,
     198            0 :     cancel: &CancellationToken,
     199            0 : ) -> anyhow::Result<()> {
     200            0 :     let source_path = remote_initdb_archive_path(tenant_id, timeline_id);
     201            0 :     let dest_path = remote_initdb_preserved_archive_path(tenant_id, timeline_id);
     202            0 :     storage
     203            0 :         .copy_object(&source_path, &dest_path, cancel)
     204            0 :         .await
     205            0 :         .with_context(|| format!("backing up initdb archive for '{tenant_id} / {timeline_id}'"))
     206            0 : }
     207              : 
     208            0 : pub(crate) async fn time_travel_recover_tenant(
     209            0 :     storage: &GenericRemoteStorage,
     210            0 :     tenant_shard_id: &TenantShardId,
     211            0 :     timestamp: SystemTime,
     212            0 :     done_if_after: SystemTime,
     213            0 :     cancel: &CancellationToken,
     214            0 : ) -> Result<(), TimeTravelError> {
     215            0 :     let warn_after = 3;
     216            0 :     let max_attempts = 10;
     217            0 :     let mut prefixes = Vec::with_capacity(2);
     218            0 :     if tenant_shard_id.is_shard_zero() {
     219            0 :         // Also recover the unsharded prefix for a shard of zero:
     220            0 :         // - if the tenant is totally unsharded, the unsharded prefix contains all the data
     221            0 :         // - if the tenant is sharded, we still want to recover the initdb data, but we only
     222            0 :         //   want to do it once, so let's do it on the 0 shard
     223            0 :         let timelines_path_unsharded =
     224            0 :             super::remote_timelines_path_unsharded(&tenant_shard_id.tenant_id);
     225            0 :         prefixes.push(timelines_path_unsharded);
     226            0 :     }
     227            0 :     if !tenant_shard_id.is_unsharded() {
     228            0 :         // If the tenant is sharded, we need to recover the sharded prefix
     229            0 :         let timelines_path = super::remote_timelines_path(tenant_shard_id);
     230            0 :         prefixes.push(timelines_path);
     231            0 :     }
     232            0 :     for prefix in &prefixes {
     233            0 :         backoff::retry(
     234            0 :             || async {
     235            0 :                 storage
     236            0 :                     .time_travel_recover(Some(prefix), timestamp, done_if_after, cancel)
     237            0 :                     .await
     238            0 :             },
     239            0 :             |e| !matches!(e, TimeTravelError::Other(_)),
     240            0 :             warn_after,
     241            0 :             max_attempts,
     242            0 :             "time travel recovery of tenant prefix",
     243            0 :             cancel,
     244            0 :         )
     245            0 :         .await
     246            0 :         .ok_or_else(|| TimeTravelError::Cancelled)
     247            0 :         .and_then(|x| x)?;
     248              :     }
     249            0 :     Ok(())
     250            0 : }
        

Generated by: LCOV version 2.1-beta