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