Line data Source code
1 : //! The deleter is the final stage in the deletion queue. It accumulates remote
2 : //! paths to delete, and periodically executes them in batches of up to 1000
3 : //! using the DeleteObjects request.
4 : //!
5 : //! Its purpose is to increase efficiency of remote storage I/O by issuing a smaller
6 : //! number of full-sized DeleteObjects requests, rather than a larger number of
7 : //! smaller requests.
8 :
9 : use remote_storage::GenericRemoteStorage;
10 : use remote_storage::RemotePath;
11 : use remote_storage::TimeoutOrCancel;
12 : use std::time::Duration;
13 : use tokio_util::sync::CancellationToken;
14 : use tracing::info;
15 : use tracing::warn;
16 : use utils::backoff;
17 : use utils::pausable_failpoint;
18 :
19 : use crate::metrics;
20 :
21 : use super::DeletionQueueError;
22 : use super::FlushOp;
23 :
24 : const AUTOFLUSH_INTERVAL: Duration = Duration::from_secs(10);
25 :
26 : pub(super) enum DeleterMessage {
27 : Delete(Vec<RemotePath>),
28 : Flush(FlushOp),
29 : }
30 :
31 : /// Non-persistent deletion queue, for coalescing multiple object deletes into
32 : /// larger DeleteObjects requests.
33 : pub(super) struct Deleter {
34 : // Accumulate up to 1000 keys for the next deletion operation
35 : accumulator: Vec<RemotePath>,
36 :
37 : rx: tokio::sync::mpsc::Receiver<DeleterMessage>,
38 :
39 : cancel: CancellationToken,
40 : remote_storage: GenericRemoteStorage,
41 : }
42 :
43 : impl Deleter {
44 16 : pub(super) fn new(
45 16 : remote_storage: GenericRemoteStorage,
46 16 : rx: tokio::sync::mpsc::Receiver<DeleterMessage>,
47 16 : cancel: CancellationToken,
48 16 : ) -> Self {
49 16 : Self {
50 16 : remote_storage,
51 16 : rx,
52 16 : cancel,
53 16 : accumulator: Vec::new(),
54 16 : }
55 16 : }
56 :
57 : /// Wrap the remote `delete_objects` with a failpoint
58 12 : async fn remote_delete(&self) -> Result<(), anyhow::Error> {
59 12 : // A backoff::retry is used here for two reasons:
60 12 : // - To provide a backoff rather than busy-polling the API on errors
61 12 : // - To absorb transient 429/503 conditions without hitting our error
62 12 : // logging path for issues deleting objects.
63 12 : backoff::retry(
64 12 : || async {
65 12 : fail::fail_point!("deletion-queue-before-execute", |_| {
66 0 : info!("Skipping execution, failpoint set");
67 :
68 0 : metrics::DELETION_QUEUE
69 0 : .remote_errors
70 0 : .with_label_values(&["failpoint"])
71 0 : .inc();
72 0 : Err(anyhow::anyhow!("failpoint: deletion-queue-before-execute"))
73 12 : });
74 :
75 12 : self.remote_storage
76 12 : .delete_objects(&self.accumulator, &self.cancel)
77 12 : .await
78 24 : },
79 12 : TimeoutOrCancel::caused_by_cancel,
80 12 : 3,
81 12 : 10,
82 12 : "executing deletion batch",
83 12 : &self.cancel,
84 12 : )
85 12 : .await
86 12 : .ok_or_else(|| anyhow::anyhow!("Shutting down"))
87 12 : .and_then(|x| x)
88 12 : }
89 :
90 : /// Block until everything in accumulator has been executed
91 44 : async fn flush(&mut self) -> Result<(), DeletionQueueError> {
92 56 : while !self.accumulator.is_empty() && !self.cancel.is_cancelled() {
93 12 : pausable_failpoint!("deletion-queue-before-execute-pause");
94 12 : match self.remote_delete().await {
95 : Ok(()) => {
96 : // Note: we assume that the remote storage layer returns Ok(()) if some
97 : // or all of the deleted objects were already gone.
98 12 : metrics::DELETION_QUEUE
99 12 : .keys_executed
100 12 : .inc_by(self.accumulator.len() as u64);
101 12 : info!(
102 0 : "Executed deletion batch {}..{}",
103 0 : self.accumulator
104 0 : .first()
105 0 : .expect("accumulator should be non-empty"),
106 0 : self.accumulator
107 0 : .last()
108 0 : .expect("accumulator should be non-empty"),
109 : );
110 12 : self.accumulator.clear();
111 : }
112 0 : Err(e) => {
113 0 : if self.cancel.is_cancelled() {
114 0 : return Err(DeletionQueueError::ShuttingDown);
115 0 : }
116 0 : warn!("DeleteObjects request failed: {e:#}, will continue trying");
117 0 : metrics::DELETION_QUEUE
118 0 : .remote_errors
119 0 : .with_label_values(&["execute"])
120 0 : .inc();
121 : }
122 : };
123 : }
124 44 : if self.cancel.is_cancelled() {
125 : // Expose an error because we may not have actually flushed everything
126 4 : Err(DeletionQueueError::ShuttingDown)
127 : } else {
128 40 : Ok(())
129 : }
130 44 : }
131 :
132 16 : pub(super) async fn background(&mut self) -> Result<(), DeletionQueueError> {
133 16 : let max_keys_per_delete = self.remote_storage.max_keys_per_delete();
134 16 : self.accumulator.reserve(max_keys_per_delete);
135 :
136 : loop {
137 76 : if self.cancel.is_cancelled() {
138 0 : return Err(DeletionQueueError::ShuttingDown);
139 76 : }
140 :
141 76 : let msg = match tokio::time::timeout(AUTOFLUSH_INTERVAL, self.rx.recv()).await {
142 56 : Ok(Some(m)) => m,
143 : Ok(None) => {
144 : // All queue senders closed
145 0 : info!("Shutting down");
146 0 : return Err(DeletionQueueError::ShuttingDown);
147 : }
148 : Err(_) => {
149 : // Timeout, we hit deadline to execute whatever we have in hand. These functions will
150 : // return immediately if no work is pending
151 8 : self.flush().await?;
152 :
153 4 : continue;
154 : }
155 : };
156 :
157 56 : match msg {
158 20 : DeleterMessage::Delete(mut list) => {
159 32 : while !list.is_empty() || self.accumulator.len() == max_keys_per_delete {
160 12 : if self.accumulator.len() == max_keys_per_delete {
161 0 : self.flush().await?;
162 : // If we have received this number of keys, proceed with attempting to execute
163 0 : assert_eq!(self.accumulator.len(), 0);
164 12 : }
165 :
166 12 : let available_slots = max_keys_per_delete - self.accumulator.len();
167 12 : let take_count = std::cmp::min(available_slots, list.len());
168 12 : for path in list.drain(list.len() - take_count..) {
169 12 : self.accumulator.push(path);
170 12 : }
171 : }
172 : }
173 36 : DeleterMessage::Flush(flush_op) => {
174 36 : // If flush() errors, we drop the flush_op and the caller will get
175 36 : // an error recv()'ing their oneshot channel.
176 36 : self.flush().await?;
177 36 : flush_op.notify();
178 : }
179 : }
180 : }
181 4 : }
182 : }
|
