Line data Source code
1 : #![recursion_limit = "300"]
2 : #![deny(clippy::undocumented_unsafe_blocks)]
3 :
4 : mod auth;
5 : pub mod basebackup;
6 : pub mod config;
7 : pub mod consumption_metrics;
8 : pub mod context;
9 : pub mod controller_upcall_client;
10 : pub mod deletion_queue;
11 : pub mod disk_usage_eviction_task;
12 : pub mod http;
13 : pub mod import_datadir;
14 : pub mod l0_flush;
15 :
16 : extern crate hyper0 as hyper;
17 :
18 : use futures::StreamExt;
19 : use futures::stream::FuturesUnordered;
20 : pub use pageserver_api::keyspace;
21 : use tokio_util::sync::CancellationToken;
22 : mod assert_u64_eq_usize;
23 : pub mod aux_file;
24 : pub mod metrics;
25 : pub mod page_cache;
26 : pub mod page_service;
27 : pub mod pgdatadir_mapping;
28 : pub mod span;
29 : pub(crate) mod statvfs;
30 : pub mod task_mgr;
31 : pub mod tenant;
32 : pub mod utilization;
33 : pub mod virtual_file;
34 : pub mod walingest;
35 : pub mod walredo;
36 :
37 : use camino::Utf8Path;
38 : use deletion_queue::DeletionQueue;
39 : use tenant::mgr::{BackgroundPurges, TenantManager};
40 : use tenant::secondary;
41 : use tracing::{info, info_span};
42 :
43 : /// Current storage format version
44 : ///
45 : /// This is embedded in the header of all the layer files.
46 : /// If you make any backwards-incompatible changes to the storage
47 : /// format, bump this!
48 : /// Note that TimelineMetadata uses its own version number to track
49 : /// backwards-compatible changes to the metadata format.
50 : pub const STORAGE_FORMAT_VERSION: u16 = 3;
51 :
52 : pub const DEFAULT_PG_VERSION: u32 = 16;
53 :
54 : // Magic constants used to identify different kinds of files
55 : pub const IMAGE_FILE_MAGIC: u16 = 0x5A60;
56 : pub const DELTA_FILE_MAGIC: u16 = 0x5A61;
57 :
58 : static ZERO_PAGE: bytes::Bytes = bytes::Bytes::from_static(&[0u8; 8192]);
59 :
60 : pub use crate::metrics::preinitialize_metrics;
61 :
62 : pub struct CancellableTask {
63 : pub task: tokio::task::JoinHandle<()>,
64 : pub cancel: CancellationToken,
65 : }
66 : pub struct HttpEndpointListener(pub CancellableTask);
67 : pub struct ConsumptionMetricsTasks(pub CancellableTask);
68 : pub struct DiskUsageEvictionTask(pub CancellableTask);
69 : impl CancellableTask {
70 0 : pub async fn shutdown(self) {
71 0 : self.cancel.cancel();
72 0 : self.task.await.unwrap();
73 0 : }
74 : }
75 :
76 : #[tracing::instrument(skip_all, fields(%exit_code))]
77 : #[allow(clippy::too_many_arguments)]
78 : pub async fn shutdown_pageserver(
79 : http_listener: HttpEndpointListener,
80 : page_service: page_service::Listener,
81 : consumption_metrics_worker: ConsumptionMetricsTasks,
82 : disk_usage_eviction_task: Option<DiskUsageEvictionTask>,
83 : tenant_manager: &TenantManager,
84 : background_purges: BackgroundPurges,
85 : mut deletion_queue: DeletionQueue,
86 : secondary_controller_tasks: secondary::GlobalTasks,
87 : exit_code: i32,
88 : ) {
89 : use std::time::Duration;
90 :
91 : let started_at = std::time::Instant::now();
92 :
93 : // If the orderly shutdown below takes too long, we still want to make
94 : // sure that all walredo processes are killed and wait()ed on by us, not systemd.
95 : //
96 : // (Leftover walredo processes are the hypothesized trigger for the systemd freezes
97 : // that we keep seeing in prod => https://github.com/neondatabase/cloud/issues/11387.
98 : //
99 : // We use a thread instead of a tokio task because the background runtime is likely busy
100 : // with the final flushing / uploads. This activity here has priority, and due to lack
101 : // of scheduling priority feature sin the tokio scheduler, using a separate thread is
102 : // an effective priority booster.
103 : let walredo_extraordinary_shutdown_thread_span = {
104 : let span = info_span!(parent: None, "walredo_extraordinary_shutdown_thread");
105 : span.follows_from(tracing::Span::current());
106 : span
107 : };
108 : let walredo_extraordinary_shutdown_thread_cancel = CancellationToken::new();
109 : let walredo_extraordinary_shutdown_thread = std::thread::spawn({
110 : let walredo_extraordinary_shutdown_thread_cancel =
111 : walredo_extraordinary_shutdown_thread_cancel.clone();
112 0 : move || {
113 0 : let rt = tokio::runtime::Builder::new_current_thread()
114 0 : .enable_all()
115 0 : .build()
116 0 : .unwrap();
117 0 : let _entered = rt.enter();
118 0 : let _entered = walredo_extraordinary_shutdown_thread_span.enter();
119 0 : if let Ok(()) = rt.block_on(tokio::time::timeout(
120 0 : Duration::from_secs(8),
121 0 : walredo_extraordinary_shutdown_thread_cancel.cancelled(),
122 0 : )) {
123 0 : info!("cancellation requested");
124 0 : return;
125 0 : }
126 0 : let managers = tenant::WALREDO_MANAGERS
127 0 : .lock()
128 0 : .unwrap()
129 0 : // prevents new walredo managers from being inserted
130 0 : .take()
131 0 : .expect("only we take()");
132 0 : // Use FuturesUnordered to get in queue early for each manager's
133 0 : // heavier_once_cell semaphore wait list.
134 0 : // Also, for idle tenants that for some reason haven't
135 0 : // shut down yet, it's quite likely that we're not going
136 0 : // to get Poll::Pending once.
137 0 : let mut futs: FuturesUnordered<_> = managers
138 0 : .into_iter()
139 0 : .filter_map(|(_, mgr)| mgr.upgrade())
140 0 : .map(|mgr| async move { tokio::task::unconstrained(mgr.shutdown()).await })
141 0 : .collect();
142 0 : info!(count=%futs.len(), "built FuturesUnordered");
143 0 : let mut last_log_at = std::time::Instant::now();
144 : #[derive(Debug, Default)]
145 : struct Results {
146 : initiated: u64,
147 : already: u64,
148 : }
149 0 : let mut results = Results::default();
150 0 : while let Some(we_initiated) = rt.block_on(futs.next()) {
151 0 : if we_initiated {
152 0 : results.initiated += 1;
153 0 : } else {
154 0 : results.already += 1;
155 0 : }
156 0 : if last_log_at.elapsed() > Duration::from_millis(100) {
157 0 : info!(remaining=%futs.len(), ?results, "progress");
158 0 : last_log_at = std::time::Instant::now();
159 0 : }
160 : }
161 0 : info!(?results, "done");
162 0 : }
163 : });
164 :
165 : // Shut down the libpq endpoint task. This prevents new connections from
166 : // being accepted.
167 : let remaining_connections = timed(
168 : page_service.stop_accepting(),
169 : "shutdown LibpqEndpointListener",
170 : Duration::from_secs(1),
171 : )
172 : .await;
173 :
174 : // Shut down all the tenants. This flushes everything to disk and kills
175 : // the checkpoint and GC tasks.
176 : timed(
177 : tenant_manager.shutdown(),
178 : "shutdown all tenants",
179 : Duration::from_secs(5),
180 : )
181 : .await;
182 :
183 : // Shut down any page service tasks: any in-progress work for particular timelines or tenants
184 : // should already have been canclled via mgr::shutdown_all_tenants
185 : timed(
186 : remaining_connections.shutdown(),
187 : "shutdown PageRequestHandlers",
188 : Duration::from_secs(1),
189 : )
190 : .await;
191 :
192 : // Best effort to persist any outstanding deletions, to avoid leaking objects
193 : deletion_queue.shutdown(Duration::from_secs(5)).await;
194 :
195 : timed(
196 : consumption_metrics_worker.0.shutdown(),
197 : "shutdown consumption metrics",
198 : Duration::from_secs(1),
199 : )
200 : .await;
201 :
202 : timed(
203 0 : futures::future::OptionFuture::from(disk_usage_eviction_task.map(|t| t.0.shutdown())),
204 : "shutdown disk usage eviction",
205 : Duration::from_secs(1),
206 : )
207 : .await;
208 :
209 : timed(
210 : background_purges.shutdown(),
211 : "shutdown background purges",
212 : Duration::from_secs(1),
213 : )
214 : .await;
215 :
216 : // Shut down the HTTP endpoint last, so that you can still check the server's
217 : // status while it's shutting down.
218 : // FIXME: We should probably stop accepting commands like attach/detach earlier.
219 : timed(
220 : http_listener.0.shutdown(),
221 : "shutdown http",
222 : Duration::from_secs(1),
223 : )
224 : .await;
225 :
226 : timed(
227 : secondary_controller_tasks.wait(), // cancellation happened in caller
228 : "secondary controller wait",
229 : Duration::from_secs(1),
230 : )
231 : .await;
232 :
233 : // There should be nothing left, but let's be sure
234 : timed(
235 : task_mgr::shutdown_tasks(None, None, None),
236 : "shutdown leftovers",
237 : Duration::from_secs(1),
238 : )
239 : .await;
240 :
241 : info!("cancel & join walredo_extraordinary_shutdown_thread");
242 : walredo_extraordinary_shutdown_thread_cancel.cancel();
243 : walredo_extraordinary_shutdown_thread.join().unwrap();
244 : info!("walredo_extraordinary_shutdown_thread done");
245 :
246 : info!(
247 : elapsed_ms = started_at.elapsed().as_millis(),
248 : "Shut down successfully completed"
249 : );
250 : std::process::exit(exit_code);
251 : }
252 :
253 : /// Per-tenant configuration file.
254 : /// Full path: `tenants/<tenant_id>/config-v1`.
255 : pub(crate) const TENANT_LOCATION_CONFIG_NAME: &str = "config-v1";
256 :
257 : /// Per-tenant copy of their remote heatmap, downloaded into the local
258 : /// tenant path while in secondary mode.
259 : pub(crate) const TENANT_HEATMAP_BASENAME: &str = "heatmap-v1.json";
260 :
261 : /// A suffix used for various temporary files. Any temporary files found in the
262 : /// data directory at pageserver startup can be automatically removed.
263 : pub(crate) const TEMP_FILE_SUFFIX: &str = "___temp";
264 :
265 16 : pub fn is_temporary(path: &Utf8Path) -> bool {
266 16 : match path.file_name() {
267 16 : Some(name) => name.ends_with(TEMP_FILE_SUFFIX),
268 0 : None => false,
269 : }
270 16 : }
271 :
272 : /// During pageserver startup, we need to order operations not to exhaust tokio worker threads by
273 : /// blocking.
274 : ///
275 : /// The instances of this value exist only during startup, otherwise `None` is provided, meaning no
276 : /// delaying is needed.
277 : #[derive(Clone)]
278 : pub struct InitializationOrder {
279 : /// Each initial tenant load task carries this until it is done loading timelines from remote storage
280 : pub initial_tenant_load_remote: Option<utils::completion::Completion>,
281 :
282 : /// Each initial tenant load task carries this until completion.
283 : pub initial_tenant_load: Option<utils::completion::Completion>,
284 :
285 : /// Barrier for when we can start any background jobs.
286 : ///
287 : /// This can be broken up later on, but right now there is just one class of a background job.
288 : pub background_jobs_can_start: utils::completion::Barrier,
289 : }
290 :
291 : /// Time the future with a warning when it exceeds a threshold.
292 124 : async fn timed<Fut: std::future::Future>(
293 124 : fut: Fut,
294 124 : name: &str,
295 124 : warn_at: std::time::Duration,
296 124 : ) -> <Fut as std::future::Future>::Output {
297 124 : let started = std::time::Instant::now();
298 124 :
299 124 : let mut fut = std::pin::pin!(fut);
300 124 :
301 124 : match tokio::time::timeout(warn_at, &mut fut).await {
302 120 : Ok(ret) => {
303 120 : tracing::info!(
304 : stage = name,
305 0 : elapsed_ms = started.elapsed().as_millis(),
306 0 : "completed"
307 : );
308 120 : ret
309 : }
310 : Err(_) => {
311 4 : tracing::info!(
312 : stage = name,
313 0 : elapsed_ms = started.elapsed().as_millis(),
314 0 : "still waiting, taking longer than expected..."
315 : );
316 :
317 4 : let ret = fut.await;
318 :
319 : // this has a global allowed_errors
320 4 : tracing::warn!(
321 : stage = name,
322 0 : elapsed_ms = started.elapsed().as_millis(),
323 0 : "completed, took longer than expected"
324 : );
325 :
326 4 : ret
327 : }
328 : }
329 124 : }
330 :
331 : /// Like [`timed`], but the warning timeout only starts after `cancel` has been cancelled.
332 0 : async fn timed_after_cancellation<Fut: std::future::Future>(
333 0 : fut: Fut,
334 0 : name: &str,
335 0 : warn_at: std::time::Duration,
336 0 : cancel: &CancellationToken,
337 0 : ) -> <Fut as std::future::Future>::Output {
338 0 : let mut fut = std::pin::pin!(fut);
339 0 :
340 0 : tokio::select! {
341 0 : _ = cancel.cancelled() => {
342 0 : timed(fut, name, warn_at).await
343 : }
344 0 : ret = &mut fut => {
345 0 : ret
346 : }
347 : }
348 0 : }
349 :
350 : #[cfg(test)]
351 : mod timed_tests {
352 : use std::time::Duration;
353 :
354 : use super::timed;
355 :
356 : #[tokio::test]
357 4 : async fn timed_completes_when_inner_future_completes() {
358 4 : // A future that completes on time should have its result returned
359 4 : let r1 = timed(
360 4 : async move {
361 4 : tokio::time::sleep(Duration::from_millis(10)).await;
362 4 : 123
363 4 : },
364 4 : "test 1",
365 4 : Duration::from_millis(50),
366 4 : )
367 4 : .await;
368 4 : assert_eq!(r1, 123);
369 4 :
370 4 : // A future that completes too slowly should also have its result returned
371 4 : let r1 = timed(
372 4 : async move {
373 4 : tokio::time::sleep(Duration::from_millis(50)).await;
374 4 : 456
375 4 : },
376 4 : "test 1",
377 4 : Duration::from_millis(10),
378 4 : )
379 4 : .await;
380 4 : assert_eq!(r1, 456);
381 4 : }
382 : }
|
