Line data Source code
1 : use anyhow::{Context, Result};
2 :
3 : use camino::{Utf8Path, Utf8PathBuf};
4 : use futures::stream::FuturesOrdered;
5 : use futures::StreamExt;
6 : use tokio::task::JoinHandle;
7 : use tokio_util::sync::CancellationToken;
8 : use utils::backoff;
9 : use utils::id::NodeId;
10 :
11 : use std::cmp::min;
12 : use std::collections::{HashMap, HashSet};
13 : use std::num::NonZeroU32;
14 : use std::pin::Pin;
15 : use std::sync::Arc;
16 : use std::time::Duration;
17 :
18 : use postgres_ffi::v14::xlog_utils::XLogSegNoOffsetToRecPtr;
19 : use postgres_ffi::XLogFileName;
20 : use postgres_ffi::{XLogSegNo, PG_TLI};
21 : use remote_storage::{GenericRemoteStorage, RemotePath};
22 : use tokio::fs::File;
23 :
24 : use tokio::select;
25 : use tokio::sync::mpsc::{self, Receiver, Sender};
26 : use tokio::sync::watch;
27 : use tokio::time::sleep;
28 : use tracing::*;
29 :
30 : use utils::{id::TenantTimelineId, lsn::Lsn};
31 :
32 : use crate::metrics::{BACKED_UP_SEGMENTS, BACKUP_ERRORS};
33 : use crate::timeline::{PeerInfo, Timeline};
34 : use crate::{GlobalTimelines, SafeKeeperConf};
35 :
36 : use once_cell::sync::OnceCell;
37 :
38 : const UPLOAD_FAILURE_RETRY_MIN_MS: u64 = 10;
39 : const UPLOAD_FAILURE_RETRY_MAX_MS: u64 = 5000;
40 :
41 : /// Default buffer size when interfacing with [`tokio::fs::File`].
42 : const BUFFER_SIZE: usize = 32 * 1024;
43 :
44 : /// Check whether wal backup is required for timeline. If yes, mark that launcher is
45 : /// aware of current status and return the timeline.
46 0 : async fn is_wal_backup_required(ttid: TenantTimelineId) -> Option<Arc<Timeline>> {
47 0 : match GlobalTimelines::get(ttid).ok() {
48 0 : Some(tli) => {
49 0 : tli.wal_backup_attend().await;
50 0 : Some(tli)
51 : }
52 0 : None => None,
53 : }
54 0 : }
55 :
56 : struct WalBackupTaskHandle {
57 : shutdown_tx: Sender<()>,
58 : handle: JoinHandle<()>,
59 : }
60 :
61 : struct WalBackupTimelineEntry {
62 : timeline: Arc<Timeline>,
63 : handle: Option<WalBackupTaskHandle>,
64 : }
65 :
66 0 : async fn shut_down_task(ttid: TenantTimelineId, entry: &mut WalBackupTimelineEntry) {
67 0 : if let Some(wb_handle) = entry.handle.take() {
68 : // Tell the task to shutdown. Error means task exited earlier, that's ok.
69 0 : let _ = wb_handle.shutdown_tx.send(()).await;
70 : // Await the task itself. TODO: restart panicked tasks earlier.
71 0 : if let Err(e) = wb_handle.handle.await {
72 0 : warn!("WAL backup task for {} panicked: {}", ttid, e);
73 0 : }
74 0 : }
75 0 : }
76 :
77 : /// The goal is to ensure that normally only one safekeepers offloads. However,
78 : /// it is fine (and inevitable, as s3 doesn't provide CAS) that for some short
79 : /// time we have several ones as they PUT the same files. Also,
80 : /// - frequently changing the offloader would be bad;
81 : /// - electing seriously lagging safekeeper is undesirable;
82 : /// So we deterministically choose among the reasonably caught up candidates.
83 : /// TODO: take into account failed attempts to deal with hypothetical situation
84 : /// where s3 is unreachable only for some sks.
85 0 : fn determine_offloader(
86 0 : alive_peers: &[PeerInfo],
87 0 : wal_backup_lsn: Lsn,
88 0 : ttid: TenantTimelineId,
89 0 : conf: &SafeKeeperConf,
90 0 : ) -> (Option<NodeId>, String) {
91 0 : // TODO: remove this once we fill newly joined safekeepers since backup_lsn.
92 0 : let capable_peers = alive_peers
93 0 : .iter()
94 0 : .filter(|p| p.local_start_lsn <= wal_backup_lsn);
95 0 : match capable_peers.clone().map(|p| p.commit_lsn).max() {
96 0 : None => (None, "no connected peers to elect from".to_string()),
97 0 : Some(max_commit_lsn) => {
98 0 : let threshold = max_commit_lsn
99 0 : .checked_sub(conf.max_offloader_lag_bytes)
100 0 : .unwrap_or(Lsn(0));
101 0 : let mut caughtup_peers = capable_peers
102 0 : .clone()
103 0 : .filter(|p| p.commit_lsn >= threshold)
104 0 : .collect::<Vec<_>>();
105 0 : caughtup_peers.sort_by(|p1, p2| p1.sk_id.cmp(&p2.sk_id));
106 0 :
107 0 : // To distribute the load, shift by timeline_id.
108 0 : let offloader = caughtup_peers
109 0 : [(u128::from(ttid.timeline_id) % caughtup_peers.len() as u128) as usize]
110 0 : .sk_id;
111 0 :
112 0 : let mut capable_peers_dbg = capable_peers
113 0 : .map(|p| (p.sk_id, p.commit_lsn))
114 0 : .collect::<Vec<_>>();
115 0 : capable_peers_dbg.sort_by(|p1, p2| p1.0.cmp(&p2.0));
116 0 : (
117 0 : Some(offloader),
118 0 : format!(
119 0 : "elected {} among {:?} peers, with {} of them being caughtup",
120 0 : offloader,
121 0 : capable_peers_dbg,
122 0 : caughtup_peers.len()
123 0 : ),
124 0 : )
125 : }
126 : }
127 0 : }
128 :
129 : /// Based on peer information determine which safekeeper should offload; if it
130 : /// is me, run (per timeline) task, if not yet. OTOH, if it is not me and task
131 : /// is running, kill it.
132 0 : async fn update_task(
133 0 : conf: &SafeKeeperConf,
134 0 : ttid: TenantTimelineId,
135 0 : entry: &mut WalBackupTimelineEntry,
136 0 : ) {
137 0 : let alive_peers = entry.timeline.get_peers(conf).await;
138 0 : let wal_backup_lsn = entry.timeline.get_wal_backup_lsn().await;
139 0 : let (offloader, election_dbg_str) =
140 0 : determine_offloader(&alive_peers, wal_backup_lsn, ttid, conf);
141 0 : let elected_me = Some(conf.my_id) == offloader;
142 0 :
143 0 : if elected_me != (entry.handle.is_some()) {
144 0 : if elected_me {
145 0 : info!("elected for backup: {}", election_dbg_str);
146 :
147 0 : let (shutdown_tx, shutdown_rx) = mpsc::channel(1);
148 0 : let timeline_dir = conf.timeline_dir(&ttid);
149 0 :
150 0 : let handle = tokio::spawn(
151 0 : backup_task_main(
152 0 : ttid,
153 0 : timeline_dir,
154 0 : conf.workdir.clone(),
155 0 : conf.backup_parallel_jobs,
156 0 : shutdown_rx,
157 0 : )
158 0 : .in_current_span(),
159 0 : );
160 0 :
161 0 : entry.handle = Some(WalBackupTaskHandle {
162 0 : shutdown_tx,
163 0 : handle,
164 0 : });
165 : } else {
166 0 : info!("stepping down from backup: {}", election_dbg_str);
167 0 : shut_down_task(ttid, entry).await;
168 : }
169 0 : }
170 0 : }
171 :
172 : static REMOTE_STORAGE: OnceCell<Option<GenericRemoteStorage>> = OnceCell::new();
173 :
174 : // Storage must be configured and initialized when this is called.
175 0 : fn get_configured_remote_storage() -> &'static GenericRemoteStorage {
176 0 : REMOTE_STORAGE
177 0 : .get()
178 0 : .expect("failed to get remote storage")
179 0 : .as_ref()
180 0 : .unwrap()
181 0 : }
182 :
183 : const CHECK_TASKS_INTERVAL_MSEC: u64 = 1000;
184 :
185 : /// Sits on wal_backup_launcher_rx and starts/stops per timeline wal backup
186 : /// tasks. Having this in separate task simplifies locking, allows to reap
187 : /// panics and separate elections from offloading itself.
188 0 : pub async fn wal_backup_launcher_task_main(
189 0 : conf: SafeKeeperConf,
190 0 : mut wal_backup_launcher_rx: Receiver<TenantTimelineId>,
191 0 : ) -> anyhow::Result<()> {
192 0 : info!(
193 0 : "WAL backup launcher started, remote config {:?}",
194 0 : conf.remote_storage
195 0 : );
196 :
197 0 : let conf_ = conf.clone();
198 0 : REMOTE_STORAGE.get_or_init(|| {
199 0 : conf_
200 0 : .remote_storage
201 0 : .as_ref()
202 0 : .map(|c| GenericRemoteStorage::from_config(c).expect("failed to create remote storage"))
203 0 : });
204 0 :
205 0 : // Presence in this map means launcher is aware s3 offloading is needed for
206 0 : // the timeline, but task is started only if it makes sense for to offload
207 0 : // from this safekeeper.
208 0 : let mut tasks: HashMap<TenantTimelineId, WalBackupTimelineEntry> = HashMap::new();
209 0 :
210 0 : let mut ticker = tokio::time::interval(Duration::from_millis(CHECK_TASKS_INTERVAL_MSEC));
211 0 : loop {
212 0 : tokio::select! {
213 0 : ttid = wal_backup_launcher_rx.recv() => {
214 : // channel is never expected to get closed
215 : let ttid = ttid.unwrap();
216 : if !conf.is_wal_backup_enabled() {
217 : continue; /* just drain the channel and do nothing */
218 : }
219 0 : async {
220 0 : let timeline = is_wal_backup_required(ttid).await;
221 : // do we need to do anything at all?
222 0 : if timeline.is_some() != tasks.contains_key(&ttid) {
223 0 : if let Some(timeline) = timeline {
224 : // need to start the task
225 0 : let entry = tasks.entry(ttid).or_insert(WalBackupTimelineEntry {
226 0 : timeline,
227 0 : handle: None,
228 0 : });
229 0 : update_task(&conf, ttid, entry).await;
230 : } else {
231 : // need to stop the task
232 0 : info!("stopping WAL backup task");
233 0 : let mut entry = tasks.remove(&ttid).unwrap();
234 0 : shut_down_task(ttid, &mut entry).await;
235 : }
236 0 : }
237 0 : }.instrument(info_span!("WAL backup", ttid = %ttid)).await;
238 : }
239 : // For each timeline needing offloading, check if this safekeeper
240 : // should do the job and start/stop the task accordingly.
241 : _ = ticker.tick() => {
242 : for (ttid, entry) in tasks.iter_mut() {
243 : update_task(&conf, *ttid, entry)
244 : .instrument(info_span!("WAL backup", ttid = %ttid))
245 : .await;
246 : }
247 : }
248 0 : }
249 0 : }
250 : }
251 :
252 : struct WalBackupTask {
253 : timeline: Arc<Timeline>,
254 : timeline_dir: Utf8PathBuf,
255 : workspace_dir: Utf8PathBuf,
256 : wal_seg_size: usize,
257 : parallel_jobs: usize,
258 : commit_lsn_watch_rx: watch::Receiver<Lsn>,
259 : }
260 :
261 : /// Offload single timeline.
262 0 : async fn backup_task_main(
263 0 : ttid: TenantTimelineId,
264 0 : timeline_dir: Utf8PathBuf,
265 0 : workspace_dir: Utf8PathBuf,
266 0 : parallel_jobs: usize,
267 0 : mut shutdown_rx: Receiver<()>,
268 0 : ) {
269 0 : info!("started");
270 0 : let res = GlobalTimelines::get(ttid);
271 0 : if let Err(e) = res {
272 0 : error!("backup error: {}", e);
273 0 : return;
274 0 : }
275 0 : let tli = res.unwrap();
276 :
277 0 : let mut wb = WalBackupTask {
278 0 : wal_seg_size: tli.get_wal_seg_size().await,
279 0 : commit_lsn_watch_rx: tli.get_commit_lsn_watch_rx(),
280 0 : timeline: tli,
281 0 : timeline_dir,
282 0 : workspace_dir,
283 0 : parallel_jobs,
284 0 : };
285 0 :
286 0 : // task is spinned up only when wal_seg_size already initialized
287 0 : assert!(wb.wal_seg_size > 0);
288 :
289 0 : let mut canceled = false;
290 0 : select! {
291 0 : _ = wb.run() => {}
292 0 : _ = shutdown_rx.recv() => {
293 0 : canceled = true;
294 0 : }
295 0 : }
296 0 : info!("task {}", if canceled { "canceled" } else { "terminated" });
297 0 : }
298 :
299 : impl WalBackupTask {
300 0 : async fn run(&mut self) {
301 0 : let mut backup_lsn = Lsn(0);
302 0 :
303 0 : let mut retry_attempt = 0u32;
304 : // offload loop
305 0 : loop {
306 0 : if retry_attempt == 0 {
307 : // wait for new WAL to arrive
308 0 : if let Err(e) = self.commit_lsn_watch_rx.changed().await {
309 : // should never happen, as we hold Arc to timeline.
310 0 : error!("commit_lsn watch shut down: {:?}", e);
311 0 : return;
312 0 : }
313 : } else {
314 : // or just sleep if we errored previously
315 0 : let mut retry_delay = UPLOAD_FAILURE_RETRY_MAX_MS;
316 0 : if let Some(backoff_delay) = UPLOAD_FAILURE_RETRY_MIN_MS.checked_shl(retry_attempt)
317 0 : {
318 0 : retry_delay = min(retry_delay, backoff_delay);
319 0 : }
320 0 : sleep(Duration::from_millis(retry_delay)).await;
321 : }
322 :
323 0 : let commit_lsn = *self.commit_lsn_watch_rx.borrow();
324 0 :
325 0 : // Note that backup_lsn can be higher than commit_lsn if we
326 0 : // don't have much local WAL and others already uploaded
327 0 : // segments we don't even have.
328 0 : if backup_lsn.segment_number(self.wal_seg_size)
329 0 : >= commit_lsn.segment_number(self.wal_seg_size)
330 : {
331 0 : retry_attempt = 0;
332 0 : continue; /* nothing to do, common case as we wake up on every commit_lsn bump */
333 0 : }
334 0 : // Perhaps peers advanced the position, check shmem value.
335 0 : backup_lsn = self.timeline.get_wal_backup_lsn().await;
336 0 : if backup_lsn.segment_number(self.wal_seg_size)
337 0 : >= commit_lsn.segment_number(self.wal_seg_size)
338 : {
339 0 : retry_attempt = 0;
340 0 : continue;
341 0 : }
342 0 :
343 0 : match backup_lsn_range(
344 0 : &self.timeline,
345 0 : &mut backup_lsn,
346 0 : commit_lsn,
347 0 : self.wal_seg_size,
348 0 : &self.timeline_dir,
349 0 : &self.workspace_dir,
350 0 : self.parallel_jobs,
351 0 : )
352 0 : .await
353 : {
354 0 : Ok(()) => {
355 0 : retry_attempt = 0;
356 0 : }
357 0 : Err(e) => {
358 0 : error!(
359 0 : "failed while offloading range {}-{}: {:?}",
360 0 : backup_lsn, commit_lsn, e
361 0 : );
362 :
363 0 : retry_attempt = retry_attempt.saturating_add(1);
364 : }
365 : }
366 : }
367 0 : }
368 : }
369 :
370 0 : async fn backup_lsn_range(
371 0 : timeline: &Arc<Timeline>,
372 0 : backup_lsn: &mut Lsn,
373 0 : end_lsn: Lsn,
374 0 : wal_seg_size: usize,
375 0 : timeline_dir: &Utf8Path,
376 0 : workspace_dir: &Utf8Path,
377 0 : parallel_jobs: usize,
378 0 : ) -> Result<()> {
379 0 : if parallel_jobs < 1 {
380 0 : anyhow::bail!("parallel_jobs must be >= 1");
381 0 : }
382 0 :
383 0 : let start_lsn = *backup_lsn;
384 0 : let segments = get_segments(start_lsn, end_lsn, wal_seg_size);
385 0 :
386 0 : // Pool of concurrent upload tasks. We use `FuturesOrdered` to
387 0 : // preserve order of uploads, and update `backup_lsn` only after
388 0 : // all previous uploads are finished.
389 0 : let mut uploads = FuturesOrdered::new();
390 0 : let mut iter = segments.iter();
391 :
392 : loop {
393 0 : let added_task = match iter.next() {
394 0 : Some(s) => {
395 0 : uploads.push_back(backup_single_segment(s, timeline_dir, workspace_dir));
396 0 : true
397 : }
398 0 : None => false,
399 : };
400 :
401 : // Wait for the next segment to upload if we don't have any more segments,
402 : // or if we have too many concurrent uploads.
403 0 : if !added_task || uploads.len() >= parallel_jobs {
404 0 : let next = uploads.next().await;
405 0 : if let Some(res) = next {
406 : // next segment uploaded
407 0 : let segment = res?;
408 0 : let new_backup_lsn = segment.end_lsn;
409 0 : timeline
410 0 : .set_wal_backup_lsn(new_backup_lsn)
411 0 : .await
412 0 : .context("setting wal_backup_lsn")?;
413 0 : *backup_lsn = new_backup_lsn;
414 : } else {
415 : // no more segments to upload
416 0 : break;
417 : }
418 0 : }
419 : }
420 :
421 0 : info!(
422 0 : "offloaded segnos {:?} up to {}, previous backup_lsn {}",
423 0 : segments.iter().map(|&s| s.seg_no).collect::<Vec<_>>(),
424 0 : end_lsn,
425 0 : start_lsn,
426 0 : );
427 0 : Ok(())
428 0 : }
429 :
430 0 : async fn backup_single_segment(
431 0 : seg: &Segment,
432 0 : timeline_dir: &Utf8Path,
433 0 : workspace_dir: &Utf8Path,
434 0 : ) -> Result<Segment> {
435 0 : let segment_file_path = seg.file_path(timeline_dir)?;
436 0 : let remote_segment_path = segment_file_path
437 0 : .strip_prefix(workspace_dir)
438 0 : .context("Failed to strip workspace dir prefix")
439 0 : .and_then(RemotePath::new)
440 0 : .with_context(|| {
441 0 : format!(
442 0 : "Failed to resolve remote part of path {segment_file_path:?} for base {workspace_dir:?}",
443 0 : )
444 0 : })?;
445 :
446 0 : let res = backup_object(&segment_file_path, &remote_segment_path, seg.size()).await;
447 0 : if res.is_ok() {
448 0 : BACKED_UP_SEGMENTS.inc();
449 0 : } else {
450 0 : BACKUP_ERRORS.inc();
451 0 : }
452 0 : res?;
453 0 : debug!("Backup of {} done", segment_file_path);
454 :
455 0 : Ok(*seg)
456 0 : }
457 :
458 0 : #[derive(Debug, Copy, Clone)]
459 : pub struct Segment {
460 : seg_no: XLogSegNo,
461 : start_lsn: Lsn,
462 : end_lsn: Lsn,
463 : }
464 :
465 : impl Segment {
466 0 : pub fn new(seg_no: u64, start_lsn: Lsn, end_lsn: Lsn) -> Self {
467 0 : Self {
468 0 : seg_no,
469 0 : start_lsn,
470 0 : end_lsn,
471 0 : }
472 0 : }
473 :
474 0 : pub fn object_name(self) -> String {
475 0 : XLogFileName(PG_TLI, self.seg_no, self.size())
476 0 : }
477 :
478 0 : pub fn file_path(self, timeline_dir: &Utf8Path) -> Result<Utf8PathBuf> {
479 0 : Ok(timeline_dir.join(self.object_name()))
480 0 : }
481 :
482 0 : pub fn size(self) -> usize {
483 0 : (u64::from(self.end_lsn) - u64::from(self.start_lsn)) as usize
484 0 : }
485 : }
486 :
487 0 : fn get_segments(start: Lsn, end: Lsn, seg_size: usize) -> Vec<Segment> {
488 0 : let first_seg = start.segment_number(seg_size);
489 0 : let last_seg = end.segment_number(seg_size);
490 0 :
491 0 : let res: Vec<Segment> = (first_seg..last_seg)
492 0 : .map(|s| {
493 0 : let start_lsn = XLogSegNoOffsetToRecPtr(s, 0, seg_size);
494 0 : let end_lsn = XLogSegNoOffsetToRecPtr(s + 1, 0, seg_size);
495 0 : Segment::new(s, Lsn::from(start_lsn), Lsn::from(end_lsn))
496 0 : })
497 0 : .collect();
498 0 : res
499 0 : }
500 :
501 0 : async fn backup_object(
502 0 : source_file: &Utf8Path,
503 0 : target_file: &RemotePath,
504 0 : size: usize,
505 0 : ) -> Result<()> {
506 0 : let storage = get_configured_remote_storage();
507 :
508 0 : let file = File::open(&source_file)
509 0 : .await
510 0 : .with_context(|| format!("Failed to open file {source_file:?} for wal backup"))?;
511 :
512 0 : let file = tokio_util::io::ReaderStream::with_capacity(file, BUFFER_SIZE);
513 0 :
514 0 : let cancel = CancellationToken::new();
515 0 :
516 0 : storage
517 0 : .upload_storage_object(file, size, target_file, &cancel)
518 0 : .await
519 0 : }
520 :
521 0 : pub async fn read_object(
522 0 : file_path: &RemotePath,
523 0 : offset: u64,
524 0 : ) -> anyhow::Result<Pin<Box<dyn tokio::io::AsyncRead + Send + Sync>>> {
525 0 : let storage = REMOTE_STORAGE
526 0 : .get()
527 0 : .context("Failed to get remote storage")?
528 0 : .as_ref()
529 0 : .context("No remote storage configured")?;
530 :
531 0 : info!("segment download about to start from remote path {file_path:?} at offset {offset}");
532 :
533 0 : let cancel = CancellationToken::new();
534 :
535 0 : let download = storage
536 0 : .download_storage_object(Some((offset, None)), file_path, &cancel)
537 0 : .await
538 0 : .with_context(|| {
539 0 : format!("Failed to open WAL segment download stream for remote path {file_path:?}")
540 0 : })?;
541 :
542 0 : let reader = tokio_util::io::StreamReader::new(download.download_stream);
543 0 :
544 0 : let reader = tokio::io::BufReader::with_capacity(BUFFER_SIZE, reader);
545 0 :
546 0 : Ok(Box::pin(reader))
547 0 : }
548 :
549 : /// Delete WAL files for the given timeline. Remote storage must be configured
550 : /// when called.
551 0 : pub async fn delete_timeline(ttid: &TenantTimelineId) -> Result<()> {
552 0 : let storage = get_configured_remote_storage();
553 0 : let ttid_path = Utf8Path::new(&ttid.tenant_id.to_string()).join(ttid.timeline_id.to_string());
554 0 : let remote_path = RemotePath::new(&ttid_path)?;
555 :
556 : // see DEFAULT_MAX_KEYS_PER_LIST_RESPONSE
557 : // const Option unwrap is not stable, otherwise it would be const.
558 0 : let batch_size: NonZeroU32 = NonZeroU32::new(1000).unwrap();
559 0 :
560 0 : // A backoff::retry is used here for two reasons:
561 0 : // - To provide a backoff rather than busy-polling the API on errors
562 0 : // - To absorb transient 429/503 conditions without hitting our error
563 0 : // logging path for issues deleting objects.
564 0 : //
565 0 : // Note: listing segments might take a long time if there are many of them.
566 0 : // We don't currently have http requests timeout cancellation, but if/once
567 0 : // we have listing should get streaming interface to make progress.
568 0 :
569 0 : let cancel = CancellationToken::new(); // not really used
570 0 : backoff::retry(
571 0 : || async {
572 : // Do list-delete in batch_size batches to make progress even if there a lot of files.
573 : // Alternatively we could make list_files return iterator, but it is more complicated and
574 : // I'm not sure deleting while iterating is expected in s3.
575 : loop {
576 0 : let files = storage
577 0 : .list_files(Some(&remote_path), Some(batch_size), &cancel)
578 0 : .await?;
579 0 : if files.is_empty() {
580 0 : return Ok(()); // done
581 0 : }
582 0 : // (at least) s3 results are sorted, so can log min/max:
583 0 : // "List results are always returned in UTF-8 binary order."
584 0 : info!(
585 0 : "deleting batch of {} WAL segments [{}-{}]",
586 0 : files.len(),
587 0 : files.first().unwrap().object_name().unwrap_or(""),
588 0 : files.last().unwrap().object_name().unwrap_or("")
589 0 : );
590 0 : storage.delete_objects(&files, &cancel).await?;
591 : }
592 0 : },
593 0 : // consider TimeoutOrCancel::caused_by_cancel when using cancellation
594 0 : |_| false,
595 0 : 3,
596 0 : 10,
597 0 : "executing WAL segments deletion batch",
598 0 : &cancel,
599 0 : )
600 0 : .await
601 0 : .ok_or_else(|| anyhow::anyhow!("canceled"))
602 0 : .and_then(|x| x)?;
603 :
604 0 : Ok(())
605 0 : }
606 :
607 : /// Copy segments from one timeline to another. Used in copy_timeline.
608 0 : pub async fn copy_s3_segments(
609 0 : wal_seg_size: usize,
610 0 : src_ttid: &TenantTimelineId,
611 0 : dst_ttid: &TenantTimelineId,
612 0 : from_segment: XLogSegNo,
613 0 : to_segment: XLogSegNo,
614 0 : ) -> Result<()> {
615 0 : const SEGMENTS_PROGRESS_REPORT_INTERVAL: u64 = 1024;
616 0 :
617 0 : let storage = REMOTE_STORAGE
618 0 : .get()
619 0 : .expect("failed to get remote storage")
620 0 : .as_ref()
621 0 : .unwrap();
622 0 :
623 0 : let relative_dst_path =
624 0 : Utf8Path::new(&dst_ttid.tenant_id.to_string()).join(dst_ttid.timeline_id.to_string());
625 :
626 0 : let remote_path = RemotePath::new(&relative_dst_path)?;
627 :
628 0 : let cancel = CancellationToken::new();
629 :
630 0 : let files = storage
631 0 : .list_files(Some(&remote_path), None, &cancel)
632 0 : .await?;
633 :
634 0 : let uploaded_segments = &files
635 0 : .iter()
636 0 : .filter_map(|file| file.object_name().map(ToOwned::to_owned))
637 0 : .collect::<HashSet<_>>();
638 :
639 0 : debug!(
640 0 : "these segments have already been uploaded: {:?}",
641 0 : uploaded_segments
642 0 : );
643 :
644 0 : let relative_src_path =
645 0 : Utf8Path::new(&src_ttid.tenant_id.to_string()).join(src_ttid.timeline_id.to_string());
646 :
647 0 : for segno in from_segment..to_segment {
648 0 : if segno % SEGMENTS_PROGRESS_REPORT_INTERVAL == 0 {
649 0 : info!("copied all segments from {} until {}", from_segment, segno);
650 0 : }
651 :
652 0 : let segment_name = XLogFileName(PG_TLI, segno, wal_seg_size);
653 0 : if uploaded_segments.contains(&segment_name) {
654 0 : continue;
655 0 : }
656 0 : debug!("copying segment {}", segment_name);
657 :
658 0 : let from = RemotePath::new(&relative_src_path.join(&segment_name))?;
659 0 : let to = RemotePath::new(&relative_dst_path.join(&segment_name))?;
660 :
661 0 : storage.copy_object(&from, &to, &cancel).await?;
662 : }
663 :
664 0 : info!(
665 0 : "finished copying segments from {} until {}",
666 0 : from_segment, to_segment
667 0 : );
668 0 : Ok(())
669 0 : }
|
