Line data Source code
1 : //! This module contains functions to serve per-tenant background processes,
2 : //! such as compaction and GC
3 :
4 : use std::ops::ControlFlow;
5 : use std::str::FromStr;
6 : use std::sync::Arc;
7 : use std::time::{Duration, Instant};
8 :
9 : use crate::context::{DownloadBehavior, RequestContext};
10 : use crate::metrics::TENANT_TASK_EVENTS;
11 : use crate::task_mgr;
12 : use crate::task_mgr::{TaskKind, BACKGROUND_RUNTIME};
13 : use crate::tenant::config::defaults::DEFAULT_COMPACTION_PERIOD;
14 : use crate::tenant::throttle::Stats;
15 : use crate::tenant::timeline::CompactionError;
16 : use crate::tenant::{Tenant, TenantState};
17 : use rand::Rng;
18 : use tokio_util::sync::CancellationToken;
19 : use tracing::*;
20 : use utils::{backoff, completion};
21 :
22 : static CONCURRENT_BACKGROUND_TASKS: once_cell::sync::Lazy<tokio::sync::Semaphore> =
23 12 : once_cell::sync::Lazy::new(|| {
24 12 : let total_threads = task_mgr::TOKIO_WORKER_THREADS.get();
25 12 : let permits = usize::max(
26 12 : 1,
27 12 : // while a lot of the work is done on spawn_blocking, we still do
28 12 : // repartitioning in the async context. this should give leave us some workers
29 12 : // unblocked to be blocked on other work, hopefully easing any outside visible
30 12 : // effects of restarts.
31 12 : //
32 12 : // 6/8 is a guess; previously we ran with unlimited 8 and more from
33 12 : // spawn_blocking.
34 12 : (total_threads * 3).checked_div(4).unwrap_or(0),
35 12 : );
36 12 : assert_ne!(permits, 0, "we will not be adding in permits later");
37 12 : assert!(
38 12 : permits < total_threads,
39 0 : "need threads avail for shorter work"
40 : );
41 12 : tokio::sync::Semaphore::new(permits)
42 12 : });
43 :
44 330 : #[derive(Debug, PartialEq, Eq, Clone, Copy, strum_macros::IntoStaticStr)]
45 : #[strum(serialize_all = "snake_case")]
46 : pub(crate) enum BackgroundLoopKind {
47 : Compaction,
48 : Gc,
49 : Eviction,
50 : IngestHouseKeeping,
51 : ConsumptionMetricsCollectMetrics,
52 : ConsumptionMetricsSyntheticSizeWorker,
53 : InitialLogicalSizeCalculation,
54 : HeatmapUpload,
55 : SecondaryDownload,
56 : }
57 :
58 : impl BackgroundLoopKind {
59 330 : fn as_static_str(&self) -> &'static str {
60 330 : let s: &'static str = self.into();
61 330 : s
62 330 : }
63 : }
64 :
65 : /// Cancellation safe.
66 330 : pub(crate) async fn concurrent_background_tasks_rate_limit_permit(
67 330 : loop_kind: BackgroundLoopKind,
68 330 : _ctx: &RequestContext,
69 330 : ) -> tokio::sync::SemaphorePermit<'static> {
70 330 : let _guard = crate::metrics::BACKGROUND_LOOP_SEMAPHORE_WAIT_GAUGE
71 330 : .with_label_values(&[loop_kind.as_static_str()])
72 330 : .guard();
73 :
74 : pausable_failpoint!(
75 : "initial-size-calculation-permit-pause",
76 : loop_kind == BackgroundLoopKind::InitialLogicalSizeCalculation
77 : );
78 :
79 : // TODO: assert that we run on BACKGROUND_RUNTIME; requires tokio_unstable Handle::id();
80 330 : match CONCURRENT_BACKGROUND_TASKS.acquire().await {
81 330 : Ok(permit) => permit,
82 0 : Err(_closed) => unreachable!("we never close the semaphore"),
83 : }
84 330 : }
85 :
86 : /// Start per tenant background loops: compaction and gc.
87 0 : pub fn start_background_loops(
88 0 : tenant: &Arc<Tenant>,
89 0 : background_jobs_can_start: Option<&completion::Barrier>,
90 0 : ) {
91 0 : let tenant_shard_id = tenant.tenant_shard_id;
92 0 : task_mgr::spawn(
93 0 : BACKGROUND_RUNTIME.handle(),
94 0 : TaskKind::Compaction,
95 0 : Some(tenant_shard_id),
96 0 : None,
97 0 : &format!("compactor for tenant {tenant_shard_id}"),
98 0 : false,
99 0 : {
100 0 : let tenant = Arc::clone(tenant);
101 0 : let background_jobs_can_start = background_jobs_can_start.cloned();
102 0 : async move {
103 0 : let cancel = task_mgr::shutdown_token();
104 : tokio::select! {
105 : _ = cancel.cancelled() => { return Ok(()) },
106 : _ = completion::Barrier::maybe_wait(background_jobs_can_start) => {}
107 : };
108 0 : compaction_loop(tenant, cancel)
109 0 : // If you rename this span, change the RUST_LOG env variable in test_runner/performance/test_branch_creation.py
110 0 : .instrument(info_span!("compaction_loop", tenant_id = %tenant_shard_id.tenant_id, shard_id = %tenant_shard_id.shard_slug()))
111 0 : .await;
112 0 : Ok(())
113 0 : }
114 0 : },
115 0 : );
116 0 : task_mgr::spawn(
117 0 : BACKGROUND_RUNTIME.handle(),
118 0 : TaskKind::GarbageCollector,
119 0 : Some(tenant_shard_id),
120 0 : None,
121 0 : &format!("garbage collector for tenant {tenant_shard_id}"),
122 0 : false,
123 0 : {
124 0 : let tenant = Arc::clone(tenant);
125 0 : let background_jobs_can_start = background_jobs_can_start.cloned();
126 0 : async move {
127 0 : let cancel = task_mgr::shutdown_token();
128 : tokio::select! {
129 : _ = cancel.cancelled() => { return Ok(()) },
130 : _ = completion::Barrier::maybe_wait(background_jobs_can_start) => {}
131 : };
132 0 : gc_loop(tenant, cancel)
133 0 : .instrument(info_span!("gc_loop", tenant_id = %tenant_shard_id.tenant_id, shard_id = %tenant_shard_id.shard_slug()))
134 0 : .await;
135 0 : Ok(())
136 0 : }
137 0 : },
138 0 : );
139 0 :
140 0 : task_mgr::spawn(
141 0 : BACKGROUND_RUNTIME.handle(),
142 0 : TaskKind::IngestHousekeeping,
143 0 : Some(tenant_shard_id),
144 0 : None,
145 0 : &format!("ingest housekeeping for tenant {tenant_shard_id}"),
146 0 : false,
147 0 : {
148 0 : let tenant = Arc::clone(tenant);
149 0 : let background_jobs_can_start = background_jobs_can_start.cloned();
150 0 : async move {
151 0 : let cancel = task_mgr::shutdown_token();
152 : tokio::select! {
153 : _ = cancel.cancelled() => { return Ok(()) },
154 : _ = completion::Barrier::maybe_wait(background_jobs_can_start) => {}
155 : };
156 0 : ingest_housekeeping_loop(tenant, cancel)
157 0 : .instrument(info_span!("ingest_housekeeping_loop", tenant_id = %tenant_shard_id.tenant_id, shard_id = %tenant_shard_id.shard_slug()))
158 0 : .await;
159 0 : Ok(())
160 0 : }
161 0 : },
162 0 : );
163 0 : }
164 :
165 : ///
166 : /// Compaction task's main loop
167 : ///
168 0 : async fn compaction_loop(tenant: Arc<Tenant>, cancel: CancellationToken) {
169 0 : const MAX_BACKOFF_SECS: f64 = 300.0;
170 0 : // How many errors we have seen consequtively
171 0 : let mut error_run_count = 0;
172 0 :
173 0 : let mut last_throttle_flag_reset_at = Instant::now();
174 0 :
175 0 : TENANT_TASK_EVENTS.with_label_values(&["start"]).inc();
176 0 : async {
177 0 : let ctx = RequestContext::todo_child(TaskKind::Compaction, DownloadBehavior::Download);
178 0 : let mut first = true;
179 0 : loop {
180 0 : tokio::select! {
181 : _ = cancel.cancelled() => {
182 : return;
183 : },
184 : tenant_wait_result = wait_for_active_tenant(&tenant) => match tenant_wait_result {
185 : ControlFlow::Break(()) => return,
186 : ControlFlow::Continue(()) => (),
187 : },
188 : }
189 :
190 0 : let period = tenant.get_compaction_period();
191 0 :
192 0 : // TODO: we shouldn't need to await to find tenant and this could be moved outside of
193 0 : // loop, #3501. There are also additional "allowed_errors" in tests.
194 0 : if first {
195 0 : first = false;
196 0 : if random_init_delay(period, &cancel).await.is_err() {
197 0 : break;
198 0 : }
199 0 : }
200 :
201 0 : let started_at = Instant::now();
202 :
203 0 : let sleep_duration = if period == Duration::ZERO {
204 : #[cfg(not(feature = "testing"))]
205 : info!("automatic compaction is disabled");
206 : // check again in 10 seconds, in case it's been enabled again.
207 0 : Duration::from_secs(10)
208 : } else {
209 : // Run compaction
210 0 : if let Err(e) = tenant.compaction_iteration(&cancel, &ctx).await {
211 0 : let wait_duration = backoff::exponential_backoff_duration_seconds(
212 0 : error_run_count + 1,
213 0 : 1.0,
214 0 : MAX_BACKOFF_SECS,
215 0 : );
216 0 : error_run_count += 1;
217 0 : let wait_duration = Duration::from_secs_f64(wait_duration);
218 0 : log_compaction_error(
219 0 : &e,
220 0 : error_run_count,
221 0 : &wait_duration,
222 0 : cancel.is_cancelled(),
223 0 : );
224 0 : wait_duration
225 : } else {
226 0 : error_run_count = 0;
227 0 : period
228 : }
229 : };
230 :
231 0 : let elapsed = started_at.elapsed();
232 0 : warn_when_period_overrun(elapsed, period, BackgroundLoopKind::Compaction);
233 0 :
234 0 : // the duration is recorded by performance tests by enabling debug in this function
235 0 : tracing::debug!(elapsed_ms=elapsed.as_millis(), "compaction iteration complete");
236 :
237 : // Perhaps we did no work and the walredo process has been idle for some time:
238 : // give it a chance to shut down to avoid leaving walredo process running indefinitely.
239 0 : if let Some(walredo_mgr) = &tenant.walredo_mgr {
240 0 : walredo_mgr.maybe_quiesce(period * 10);
241 0 : }
242 :
243 : // TODO: move this (and walredo quiesce) to a separate task that isn't affected by the back-off,
244 : // so we get some upper bound guarantee on when walredo quiesce / this throttling reporting here happens.
245 0 : info_span!(parent: None, "timeline_get_throttle", tenant_id=%tenant.tenant_shard_id, shard_id=%tenant.tenant_shard_id.shard_slug()).in_scope(|| {
246 0 : let now = Instant::now();
247 0 : let prev = std::mem::replace(&mut last_throttle_flag_reset_at, now);
248 0 : let Stats { count_accounted, count_throttled, sum_throttled_usecs } = tenant.timeline_get_throttle.reset_stats();
249 0 : if count_throttled == 0 {
250 0 : return;
251 0 : }
252 0 : let allowed_rps = tenant.timeline_get_throttle.steady_rps();
253 0 : let delta = now - prev;
254 0 : info!(
255 0 : n_seconds=%format_args!("{:.3}",
256 0 : delta.as_secs_f64()),
257 : count_accounted,
258 : count_throttled,
259 : sum_throttled_usecs,
260 0 : allowed_rps=%format_args!("{allowed_rps:.0}"),
261 0 : "shard was throttled in the last n_seconds")
262 0 : });
263 0 :
264 0 : // Sleep
265 0 : if tokio::time::timeout(sleep_duration, cancel.cancelled())
266 0 : .await
267 0 : .is_ok()
268 : {
269 0 : break;
270 0 : }
271 : }
272 0 : }
273 0 : .await;
274 0 : TENANT_TASK_EVENTS.with_label_values(&["stop"]).inc();
275 0 : }
276 :
277 0 : fn log_compaction_error(
278 0 : e: &CompactionError,
279 0 : error_run_count: u32,
280 0 : sleep_duration: &std::time::Duration,
281 0 : task_cancelled: bool,
282 0 : ) {
283 : use crate::tenant::upload_queue::NotInitialized;
284 : use crate::tenant::PageReconstructError;
285 : use CompactionError::*;
286 :
287 : enum LooksLike {
288 : Info,
289 : Error,
290 : }
291 :
292 0 : let decision = match e {
293 0 : ShuttingDown => None,
294 0 : _ if task_cancelled => Some(LooksLike::Info),
295 0 : Other(e) => {
296 0 : let root_cause = e.root_cause();
297 :
298 0 : let is_stopping = {
299 0 : let upload_queue = root_cause
300 0 : .downcast_ref::<NotInitialized>()
301 0 : .is_some_and(|e| e.is_stopping());
302 0 :
303 0 : let timeline = root_cause
304 0 : .downcast_ref::<PageReconstructError>()
305 0 : .is_some_and(|e| e.is_stopping());
306 0 :
307 0 : upload_queue || timeline
308 : };
309 :
310 0 : if is_stopping {
311 0 : Some(LooksLike::Info)
312 : } else {
313 0 : Some(LooksLike::Error)
314 : }
315 : }
316 : };
317 :
318 0 : match decision {
319 0 : Some(LooksLike::Info) => info!(
320 0 : "Compaction failed {error_run_count} times, retrying in {sleep_duration:?}: {e:#}",
321 : ),
322 0 : Some(LooksLike::Error) => error!(
323 0 : "Compaction failed {error_run_count} times, retrying in {sleep_duration:?}: {e:?}",
324 : ),
325 0 : None => {}
326 : }
327 0 : }
328 :
329 : ///
330 : /// GC task's main loop
331 : ///
332 0 : async fn gc_loop(tenant: Arc<Tenant>, cancel: CancellationToken) {
333 0 : const MAX_BACKOFF_SECS: f64 = 300.0;
334 0 : // How many errors we have seen consequtively
335 0 : let mut error_run_count = 0;
336 0 :
337 0 : TENANT_TASK_EVENTS.with_label_values(&["start"]).inc();
338 0 : async {
339 0 : // GC might require downloading, to find the cutoff LSN that corresponds to the
340 0 : // cutoff specified as time.
341 0 : let ctx =
342 0 : RequestContext::todo_child(TaskKind::GarbageCollector, DownloadBehavior::Download);
343 0 : let mut first = true;
344 0 : loop {
345 0 : tokio::select! {
346 : _ = cancel.cancelled() => {
347 : return;
348 : },
349 : tenant_wait_result = wait_for_active_tenant(&tenant) => match tenant_wait_result {
350 : ControlFlow::Break(()) => return,
351 : ControlFlow::Continue(()) => (),
352 : },
353 : }
354 :
355 0 : let period = tenant.get_gc_period();
356 0 :
357 0 : if first {
358 0 : first = false;
359 0 : if random_init_delay(period, &cancel).await.is_err() {
360 0 : break;
361 0 : }
362 0 : }
363 :
364 0 : let started_at = Instant::now();
365 0 :
366 0 : let gc_horizon = tenant.get_gc_horizon();
367 0 : let sleep_duration = if period == Duration::ZERO || gc_horizon == 0 {
368 : #[cfg(not(feature = "testing"))]
369 : info!("automatic GC is disabled");
370 : // check again in 10 seconds, in case it's been enabled again.
371 0 : Duration::from_secs(10)
372 : } else {
373 : // Run gc
374 0 : let res = tenant
375 0 : .gc_iteration(None, gc_horizon, tenant.get_pitr_interval(), &cancel, &ctx)
376 0 : .await;
377 0 : if let Err(e) = res {
378 0 : let wait_duration = backoff::exponential_backoff_duration_seconds(
379 0 : error_run_count + 1,
380 0 : 1.0,
381 0 : MAX_BACKOFF_SECS,
382 0 : );
383 0 : error_run_count += 1;
384 0 : let wait_duration = Duration::from_secs_f64(wait_duration);
385 0 : error!(
386 0 : "Gc failed {error_run_count} times, retrying in {wait_duration:?}: {e:?}",
387 : );
388 0 : wait_duration
389 : } else {
390 0 : error_run_count = 0;
391 0 : period
392 : }
393 : };
394 :
395 0 : warn_when_period_overrun(started_at.elapsed(), period, BackgroundLoopKind::Gc);
396 0 :
397 0 : // Sleep
398 0 : if tokio::time::timeout(sleep_duration, cancel.cancelled())
399 0 : .await
400 0 : .is_ok()
401 : {
402 0 : break;
403 0 : }
404 : }
405 0 : }
406 0 : .await;
407 0 : TENANT_TASK_EVENTS.with_label_values(&["stop"]).inc();
408 0 : }
409 :
410 0 : async fn ingest_housekeeping_loop(tenant: Arc<Tenant>, cancel: CancellationToken) {
411 0 : TENANT_TASK_EVENTS.with_label_values(&["start"]).inc();
412 0 : async {
413 0 : loop {
414 0 : tokio::select! {
415 : _ = cancel.cancelled() => {
416 : return;
417 : },
418 : tenant_wait_result = wait_for_active_tenant(&tenant) => match tenant_wait_result {
419 : ControlFlow::Break(()) => return,
420 : ControlFlow::Continue(()) => (),
421 : },
422 : }
423 :
424 : // We run ingest housekeeping with the same frequency as compaction: it is not worth
425 : // having a distinct setting. But we don't run it in the same task, because compaction
426 : // blocks on acquiring the background job semaphore.
427 0 : let period = tenant.get_compaction_period();
428 :
429 : // If compaction period is set to zero (to disable it), then we will use a reasonable default
430 0 : let period = if period == Duration::ZERO {
431 0 : humantime::Duration::from_str(DEFAULT_COMPACTION_PERIOD)
432 0 : .unwrap()
433 0 : .into()
434 : } else {
435 0 : period
436 : };
437 :
438 : // Jitter the period by +/- 5%
439 0 : let period =
440 0 : rand::thread_rng().gen_range((period * (95)) / 100..(period * (105)) / 100);
441 0 :
442 0 : // Always sleep first: we do not need to do ingest housekeeping early in the lifetime of
443 0 : // a tenant, since it won't have started writing any ephemeral files yet.
444 0 : if tokio::time::timeout(period, cancel.cancelled())
445 0 : .await
446 0 : .is_ok()
447 : {
448 0 : break;
449 0 : }
450 0 :
451 0 : let started_at = Instant::now();
452 0 : tenant.ingest_housekeeping().await;
453 :
454 0 : warn_when_period_overrun(
455 0 : started_at.elapsed(),
456 0 : period,
457 0 : BackgroundLoopKind::IngestHouseKeeping,
458 0 : );
459 : }
460 0 : }
461 0 : .await;
462 0 : TENANT_TASK_EVENTS.with_label_values(&["stop"]).inc();
463 0 : }
464 :
465 0 : async fn wait_for_active_tenant(tenant: &Arc<Tenant>) -> ControlFlow<()> {
466 0 : // if the tenant has a proper status already, no need to wait for anything
467 0 : if tenant.current_state() == TenantState::Active {
468 0 : ControlFlow::Continue(())
469 : } else {
470 0 : let mut tenant_state_updates = tenant.subscribe_for_state_updates();
471 : loop {
472 0 : match tenant_state_updates.changed().await {
473 : Ok(()) => {
474 0 : let new_state = &*tenant_state_updates.borrow();
475 0 : match new_state {
476 : TenantState::Active => {
477 0 : debug!("Tenant state changed to active, continuing the task loop");
478 0 : return ControlFlow::Continue(());
479 : }
480 0 : state => {
481 0 : debug!("Not running the task loop, tenant is not active: {state:?}");
482 0 : continue;
483 : }
484 : }
485 : }
486 0 : Err(_sender_dropped_error) => {
487 0 : return ControlFlow::Break(());
488 : }
489 : }
490 : }
491 : }
492 0 : }
493 :
494 0 : #[derive(thiserror::Error, Debug)]
495 : #[error("cancelled")]
496 : pub(crate) struct Cancelled;
497 :
498 : /// Provide a random delay for background task initialization.
499 : ///
500 : /// This delay prevents a thundering herd of background tasks and will likely keep them running on
501 : /// different periods for more stable load.
502 0 : pub(crate) async fn random_init_delay(
503 0 : period: Duration,
504 0 : cancel: &CancellationToken,
505 0 : ) -> Result<(), Cancelled> {
506 0 : if period == Duration::ZERO {
507 0 : return Ok(());
508 0 : }
509 0 :
510 0 : let d = {
511 0 : let mut rng = rand::thread_rng();
512 0 : rng.gen_range(Duration::ZERO..=period)
513 0 : };
514 0 :
515 0 : match tokio::time::timeout(d, cancel.cancelled()).await {
516 0 : Ok(_) => Err(Cancelled),
517 0 : Err(_) => Ok(()),
518 : }
519 0 : }
520 :
521 : /// Attention: the `task` and `period` beocme labels of a pageserver-wide prometheus metric.
522 0 : pub(crate) fn warn_when_period_overrun(
523 0 : elapsed: Duration,
524 0 : period: Duration,
525 0 : task: BackgroundLoopKind,
526 0 : ) {
527 0 : // Duration::ZERO will happen because it's the "disable [bgtask]" value.
528 0 : if elapsed >= period && period != Duration::ZERO {
529 : // humantime does no significant digits clamping whereas Duration's debug is a bit more
530 : // intelligent. however it makes sense to keep the "configuration format" for period, even
531 : // though there's no way to output the actual config value.
532 0 : info!(
533 : ?elapsed,
534 0 : period = %humantime::format_duration(period),
535 0 : ?task,
536 0 : "task iteration took longer than the configured period"
537 : );
538 0 : crate::metrics::BACKGROUND_LOOP_PERIOD_OVERRUN_COUNT
539 0 : .with_label_values(&[task.as_static_str(), &format!("{}", period.as_secs())])
540 0 : .inc();
541 0 : }
542 0 : }
|
