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