Tag S3 objects

safekeeper/src/wal_backup.rs

@@ -500,7 +500,27 @@ fn get_segments(start: Lsn, end: Lsn, seg_size: usize) -> Vec<Segment> {
     res
 }
 
-pub async fn backup_object(
+async fn backup_object(
+    source_file: &Utf8Path,
+    target_file: &RemotePath,
+    size: usize,
+) -> Result<()> {
+    let storage = get_configured_remote_storage();
+
+    let file = File::open(&source_file)
+        .await
+        .with_context(|| format!("Failed to open file {source_file:?} for wal backup"))?;
+
+    let file = tokio_util::io::ReaderStream::with_capacity(file, BUFFER_SIZE);
+
+    let cancel = CancellationToken::new();
+
+    storage
+        .upload_storage_object(file, size, target_file, &cancel)
+        .await
+}
+
+pub(crate) async fn backup_partial_segment(
     source_file: &Utf8Path,
     target_file: &RemotePath,
     size: usize,
@@ -519,7 +539,13 @@ pub async fn backup_object(
     let cancel = CancellationToken::new();
 
     storage
-        .upload_storage_object(file, size, target_file, &cancel)
+        .upload(
+            file,
+            size,
+            target_file,
+            Some(HashMap::from([("sk_type", "partial_segment")])),
+            &cancel,
+        )
         .await
 }
 

safekeeper/src/wal_backup_partial.rs

@@ -136,7 +136,7 @@ impl PartialBackup {
         let remote_path = RemotePath::new(self.remote_prefix.join(&prepared.name).as_ref())?;
 
         // Upload first `backup_bytes` bytes of the segment to the remote storage.
-        wal_backup::backup_object(&local_path, &remote_path, backup_bytes).await?;
+        wal_backup::backup_partial_segment(&local_path, &remote_path, backup_bytes).await?;
         PARTIAL_BACKUP_UPLOADED_BYTES.inc_by(backup_bytes as u64);
 
         // We uploaded the segment, now let's verify that the data is still actual.