Line data Source code
1 : #![deny(clippy::undocumented_unsafe_blocks)]
2 :
3 : extern crate hyper0 as hyper;
4 :
5 : use camino::Utf8PathBuf;
6 : use once_cell::sync::Lazy;
7 : use remote_storage::RemoteStorageConfig;
8 : use tokio::runtime::Runtime;
9 :
10 : use std::time::Duration;
11 : use storage_broker::Uri;
12 :
13 : use utils::{auth::SwappableJwtAuth, id::NodeId, logging::SecretString};
14 :
15 : mod auth;
16 : pub mod broker;
17 : pub mod control_file;
18 : pub mod control_file_upgrade;
19 : pub mod copy_timeline;
20 : pub mod debug_dump;
21 : pub mod handler;
22 : pub mod http;
23 : pub mod json_ctrl;
24 : pub mod metrics;
25 : pub mod patch_control_file;
26 : pub mod pull_timeline;
27 : pub mod rate_limit;
28 : pub mod receive_wal;
29 : pub mod recovery;
30 : pub mod remove_wal;
31 : pub mod safekeeper;
32 : pub mod send_interpreted_wal;
33 : pub mod send_wal;
34 : pub mod state;
35 : pub mod timeline;
36 : pub mod timeline_eviction;
37 : pub mod timeline_guard;
38 : pub mod timeline_manager;
39 : pub mod timelines_set;
40 : pub mod wal_backup;
41 : pub mod wal_backup_partial;
42 : pub mod wal_reader_stream;
43 : pub mod wal_service;
44 : pub mod wal_storage;
45 :
46 : mod timelines_global_map;
47 : use std::sync::Arc;
48 : pub use timelines_global_map::GlobalTimelines;
49 : use utils::auth::JwtAuth;
50 :
51 : pub mod defaults {
52 : pub use safekeeper_api::{
53 : DEFAULT_HTTP_LISTEN_ADDR, DEFAULT_HTTP_LISTEN_PORT, DEFAULT_PG_LISTEN_ADDR,
54 : DEFAULT_PG_LISTEN_PORT,
55 : };
56 :
57 : pub const DEFAULT_HEARTBEAT_TIMEOUT: &str = "5000ms";
58 : pub const DEFAULT_MAX_OFFLOADER_LAG_BYTES: u64 = 128 * (1 << 20);
59 : pub const DEFAULT_PARTIAL_BACKUP_TIMEOUT: &str = "15m";
60 : pub const DEFAULT_CONTROL_FILE_SAVE_INTERVAL: &str = "300s";
61 : pub const DEFAULT_PARTIAL_BACKUP_CONCURRENCY: &str = "5";
62 : pub const DEFAULT_EVICTION_CONCURRENCY: usize = 2;
63 :
64 : // By default, our required residency before eviction is the same as the period that passes
65 : // before uploading a partial segment, so that in normal operation the eviction can happen
66 : // as soon as we have done the partial segment upload.
67 : pub const DEFAULT_EVICTION_MIN_RESIDENT: &str = DEFAULT_PARTIAL_BACKUP_TIMEOUT;
68 : }
69 :
70 : #[derive(Debug, Clone)]
71 : pub struct SafeKeeperConf {
72 : // Repository directory, relative to current working directory.
73 : // Normally, the safekeeper changes the current working directory
74 : // to the repository, and 'workdir' is always '.'. But we don't do
75 : // that during unit testing, because the current directory is global
76 : // to the process but different unit tests work on different
77 : // data directories to avoid clashing with each other.
78 : pub workdir: Utf8PathBuf,
79 : pub my_id: NodeId,
80 : pub listen_pg_addr: String,
81 : pub listen_pg_addr_tenant_only: Option<String>,
82 : pub listen_http_addr: String,
83 : pub advertise_pg_addr: Option<String>,
84 : pub availability_zone: Option<String>,
85 : pub no_sync: bool,
86 : pub broker_endpoint: Uri,
87 : pub broker_keepalive_interval: Duration,
88 : pub heartbeat_timeout: Duration,
89 : pub peer_recovery_enabled: bool,
90 : pub remote_storage: Option<RemoteStorageConfig>,
91 : pub max_offloader_lag_bytes: u64,
92 : pub backup_parallel_jobs: usize,
93 : pub wal_backup_enabled: bool,
94 : pub pg_auth: Option<Arc<JwtAuth>>,
95 : pub pg_tenant_only_auth: Option<Arc<JwtAuth>>,
96 : pub http_auth: Option<Arc<SwappableJwtAuth>>,
97 : /// JWT token to connect to other safekeepers with.
98 : pub sk_auth_token: Option<SecretString>,
99 : pub current_thread_runtime: bool,
100 : pub walsenders_keep_horizon: bool,
101 : pub partial_backup_timeout: Duration,
102 : pub disable_periodic_broker_push: bool,
103 : pub enable_offload: bool,
104 : pub delete_offloaded_wal: bool,
105 : pub control_file_save_interval: Duration,
106 : pub partial_backup_concurrency: usize,
107 : pub eviction_min_resident: Duration,
108 : }
109 :
110 : impl SafeKeeperConf {
111 0 : pub fn is_wal_backup_enabled(&self) -> bool {
112 0 : self.remote_storage.is_some() && self.wal_backup_enabled
113 0 : }
114 : }
115 :
116 : impl SafeKeeperConf {
117 0 : pub fn dummy() -> Self {
118 0 : SafeKeeperConf {
119 0 : workdir: Utf8PathBuf::from("./"),
120 0 : no_sync: false,
121 0 : listen_pg_addr: defaults::DEFAULT_PG_LISTEN_ADDR.to_string(),
122 0 : listen_pg_addr_tenant_only: None,
123 0 : listen_http_addr: defaults::DEFAULT_HTTP_LISTEN_ADDR.to_string(),
124 0 : advertise_pg_addr: None,
125 0 : availability_zone: None,
126 0 : remote_storage: None,
127 0 : my_id: NodeId(0),
128 0 : broker_endpoint: storage_broker::DEFAULT_ENDPOINT
129 0 : .parse()
130 0 : .expect("failed to parse default broker endpoint"),
131 0 : broker_keepalive_interval: Duration::from_secs(5),
132 0 : peer_recovery_enabled: true,
133 0 : wal_backup_enabled: true,
134 0 : backup_parallel_jobs: 1,
135 0 : pg_auth: None,
136 0 : pg_tenant_only_auth: None,
137 0 : http_auth: None,
138 0 : sk_auth_token: None,
139 0 : heartbeat_timeout: Duration::new(5, 0),
140 0 : max_offloader_lag_bytes: defaults::DEFAULT_MAX_OFFLOADER_LAG_BYTES,
141 0 : current_thread_runtime: false,
142 0 : walsenders_keep_horizon: false,
143 0 : partial_backup_timeout: Duration::from_secs(0),
144 0 : disable_periodic_broker_push: false,
145 0 : enable_offload: false,
146 0 : delete_offloaded_wal: false,
147 0 : control_file_save_interval: Duration::from_secs(1),
148 0 : partial_backup_concurrency: 1,
149 0 : eviction_min_resident: Duration::ZERO,
150 0 : }
151 0 : }
152 : }
153 :
154 : // Tokio runtimes.
155 0 : pub static WAL_SERVICE_RUNTIME: Lazy<Runtime> = Lazy::new(|| {
156 0 : tokio::runtime::Builder::new_multi_thread()
157 0 : .thread_name("WAL service worker")
158 0 : .enable_all()
159 0 : .build()
160 0 : .expect("Failed to create WAL service runtime")
161 0 : });
162 :
163 0 : pub static HTTP_RUNTIME: Lazy<Runtime> = Lazy::new(|| {
164 0 : tokio::runtime::Builder::new_multi_thread()
165 0 : .thread_name("HTTP worker")
166 0 : .enable_all()
167 0 : .build()
168 0 : .expect("Failed to create HTTP runtime")
169 0 : });
170 :
171 0 : pub static BROKER_RUNTIME: Lazy<Runtime> = Lazy::new(|| {
172 0 : tokio::runtime::Builder::new_multi_thread()
173 0 : .thread_name("broker worker")
174 0 : .worker_threads(2) // there are only 2 tasks, having more threads doesn't make sense
175 0 : .enable_all()
176 0 : .build()
177 0 : .expect("Failed to create broker runtime")
178 0 : });
179 :
180 0 : pub static WAL_BACKUP_RUNTIME: Lazy<Runtime> = Lazy::new(|| {
181 0 : tokio::runtime::Builder::new_multi_thread()
182 0 : .thread_name("WAL backup worker")
183 0 : .enable_all()
184 0 : .build()
185 0 : .expect("Failed to create WAL backup runtime")
186 0 : });
|
