TLA Line data Source code
1 : //
2 : // Main entry point for the safekeeper executable
3 : //
4 : use anyhow::{bail, Context, Result};
5 : use camino::{Utf8Path, Utf8PathBuf};
6 : use clap::{ArgAction, Parser};
7 : use futures::future::BoxFuture;
8 : use futures::stream::FuturesUnordered;
9 : use futures::{FutureExt, StreamExt};
10 : use remote_storage::RemoteStorageConfig;
11 : use sd_notify::NotifyState;
12 : use tokio::runtime::Handle;
13 : use tokio::signal::unix::{signal, SignalKind};
14 : use tokio::task::JoinError;
15 : use toml_edit::Document;
16 :
17 : use std::fs::{self, File};
18 : use std::io::{ErrorKind, Write};
19 : use std::str::FromStr;
20 : use std::sync::Arc;
21 : use std::time::Duration;
22 : use storage_broker::Uri;
23 : use tokio::sync::mpsc;
24 :
25 : use tracing::*;
26 : use utils::pid_file;
27 :
28 : use metrics::set_build_info_metric;
29 : use safekeeper::defaults::{
30 : DEFAULT_HEARTBEAT_TIMEOUT, DEFAULT_HTTP_LISTEN_ADDR, DEFAULT_MAX_OFFLOADER_LAG_BYTES,
31 : DEFAULT_PG_LISTEN_ADDR,
32 : };
33 : use safekeeper::wal_service;
34 : use safekeeper::GlobalTimelines;
35 : use safekeeper::SafeKeeperConf;
36 : use safekeeper::{broker, WAL_SERVICE_RUNTIME};
37 : use safekeeper::{control_file, BROKER_RUNTIME};
38 : use safekeeper::{http, WAL_REMOVER_RUNTIME};
39 : use safekeeper::{remove_wal, WAL_BACKUP_RUNTIME};
40 : use safekeeper::{wal_backup, HTTP_RUNTIME};
41 : use storage_broker::DEFAULT_ENDPOINT;
42 : use utils::auth::{JwtAuth, Scope, SwappableJwtAuth};
43 : use utils::{
44 : id::NodeId,
45 : logging::{self, LogFormat},
46 : project_build_tag, project_git_version,
47 : sentry_init::init_sentry,
48 : tcp_listener,
49 : };
50 :
51 : const PID_FILE_NAME: &str = "safekeeper.pid";
52 : const ID_FILE_NAME: &str = "safekeeper.id";
53 :
54 : project_git_version!(GIT_VERSION);
55 : project_build_tag!(BUILD_TAG);
56 :
57 : const FEATURES: &[&str] = &[
58 : #[cfg(feature = "testing")]
59 : "testing",
60 : ];
61 :
62 CBC 1239 : fn version() -> String {
63 1239 : format!(
64 1239 : "{GIT_VERSION} failpoints: {}, features: {:?}",
65 1239 : fail::has_failpoints(),
66 1239 : FEATURES,
67 1239 : )
68 1239 : }
69 :
70 : const ABOUT: &str = r#"
71 : A fleet of safekeepers is responsible for reliably storing WAL received from
72 : compute, passing it through consensus (mitigating potential computes brain
73 : split), and serving the hardened part further downstream to pageserver(s).
74 : "#;
75 :
76 1240 : #[derive(Parser)]
77 : #[command(name = "Neon safekeeper", version = GIT_VERSION, about = ABOUT, long_about = None)]
78 : struct Args {
79 : /// Path to the safekeeper data directory.
80 : #[arg(short = 'D', long, default_value = "./")]
81 UBC 0 : datadir: Utf8PathBuf,
82 : /// Safekeeper node id.
83 : #[arg(long)]
84 : id: Option<u64>,
85 : /// Initialize safekeeper with given id and exit.
86 : #[arg(long)]
87 0 : init: bool,
88 : /// Listen endpoint for receiving/sending WAL in the form host:port.
89 : #[arg(short, long, default_value = DEFAULT_PG_LISTEN_ADDR)]
90 0 : listen_pg: String,
91 : /// Listen endpoint for receiving/sending WAL in the form host:port allowing
92 : /// only tenant scoped auth tokens. Pointless if auth is disabled.
93 : #[arg(long, default_value = None, verbatim_doc_comment)]
94 : listen_pg_tenant_only: Option<String>,
95 : /// Listen http endpoint for management and metrics in the form host:port.
96 : #[arg(long, default_value = DEFAULT_HTTP_LISTEN_ADDR)]
97 0 : listen_http: String,
98 : /// Advertised endpoint for receiving/sending WAL in the form host:port. If not
99 : /// specified, listen_pg is used to advertise instead.
100 : #[arg(long, default_value = None)]
101 : advertise_pg: Option<String>,
102 : /// Availability zone of the safekeeper.
103 : #[arg(long)]
104 : availability_zone: Option<String>,
105 : /// Do not wait for changes to be written safely to disk. Unsafe.
106 : #[arg(short, long)]
107 0 : no_sync: bool,
108 : /// Dump control file at path specified by this argument and exit.
109 : #[arg(long)]
110 : dump_control_file: Option<Utf8PathBuf>,
111 : /// Broker endpoint for storage nodes coordination in the form
112 : /// http[s]://host:port. In case of https schema TLS is connection is
113 : /// established; plaintext otherwise.
114 : #[arg(long, default_value = DEFAULT_ENDPOINT, verbatim_doc_comment)]
115 0 : broker_endpoint: Uri,
116 : /// Broker keepalive interval.
117 : #[arg(long, value_parser= humantime::parse_duration, default_value = storage_broker::DEFAULT_KEEPALIVE_INTERVAL)]
118 0 : broker_keepalive_interval: Duration,
119 : /// Peer safekeeper is considered dead after not receiving heartbeats from
120 : /// it during this period passed as a human readable duration.
121 : #[arg(long, value_parser= humantime::parse_duration, default_value = DEFAULT_HEARTBEAT_TIMEOUT, verbatim_doc_comment)]
122 0 : heartbeat_timeout: Duration,
123 : /// Enable/disable peer recovery.
124 : #[arg(long, default_value = "false", action=ArgAction::Set)]
125 0 : peer_recovery: bool,
126 : /// Remote storage configuration for WAL backup (offloading to s3) as TOML
127 : /// inline table, e.g.
128 : /// {"max_concurrent_syncs" = 17, "max_sync_errors": 13, "bucket_name": "<BUCKETNAME>", "bucket_region":"<REGION>", "concurrency_limit": 119}
129 : /// Safekeeper offloads WAL to
130 : /// [prefix_in_bucket/]<tenant_id>/<timeline_id>/<segment_file>, mirroring
131 : /// structure on the file system.
132 : #[arg(long, value_parser = parse_remote_storage, verbatim_doc_comment)]
133 : remote_storage: Option<RemoteStorageConfig>,
134 : /// Safekeeper won't be elected for WAL offloading if it is lagging for more than this value in bytes
135 CBC 1240 : #[arg(long, default_value_t = DEFAULT_MAX_OFFLOADER_LAG_BYTES)]
136 UBC 0 : max_offloader_lag: u64,
137 : /// Number of max parallel WAL segments to be offloaded to remote storage.
138 : #[arg(long, default_value = "5")]
139 0 : wal_backup_parallel_jobs: usize,
140 : /// Disable WAL backup to s3. When disabled, safekeeper removes WAL ignoring
141 : /// WAL backup horizon.
142 : #[arg(long)]
143 0 : disable_wal_backup: bool,
144 : /// If given, enables auth on incoming connections to WAL service endpoint
145 : /// (--listen-pg). Value specifies path to a .pem public key used for
146 : /// validations of JWT tokens. Empty string is allowed and means disabling
147 : /// auth.
148 : #[arg(long, verbatim_doc_comment, value_parser = opt_pathbuf_parser)]
149 : pg_auth_public_key_path: Option<Utf8PathBuf>,
150 : /// If given, enables auth on incoming connections to tenant only WAL
151 : /// service endpoint (--listen-pg-tenant-only). Value specifies path to a
152 : /// .pem public key used for validations of JWT tokens. Empty string is
153 : /// allowed and means disabling auth.
154 : #[arg(long, verbatim_doc_comment, value_parser = opt_pathbuf_parser)]
155 : pg_tenant_only_auth_public_key_path: Option<Utf8PathBuf>,
156 : /// If given, enables auth on incoming connections to http management
157 : /// service endpoint (--listen-http). Value specifies path to a .pem public
158 : /// key used for validations of JWT tokens. Empty string is allowed and
159 : /// means disabling auth.
160 : #[arg(long, verbatim_doc_comment, value_parser = opt_pathbuf_parser)]
161 : http_auth_public_key_path: Option<Utf8PathBuf>,
162 : /// Format for logging, either 'plain' or 'json'.
163 : #[arg(long, default_value = "plain")]
164 0 : log_format: String,
165 : /// Run everything in single threaded current thread runtime, might be
166 : /// useful for debugging.
167 : #[arg(long)]
168 0 : current_thread_runtime: bool,
169 : }
170 :
171 : // Like PathBufValueParser, but allows empty string.
172 CBC 69 : fn opt_pathbuf_parser(s: &str) -> Result<Utf8PathBuf, String> {
173 69 : Ok(Utf8PathBuf::from_str(s).unwrap())
174 69 : }
175 :
176 : #[tokio::main(flavor = "current_thread")]
177 1239 : async fn main() -> anyhow::Result<()> {
178 1239 : // We want to allow multiple occurences of the same arg (taking the last) so
179 1239 : // that neon_local could generate command with defaults + overrides without
180 1239 : // getting 'argument cannot be used multiple times' error. This seems to be
181 1239 : // impossible with pure Derive API, so convert struct to Command, modify it,
182 1239 : // parse arguments, and then fill the struct back.
183 1239 : let cmd = <Args as clap::CommandFactory>::command()
184 1239 : .args_override_self(true)
185 1239 : .version(version());
186 1239 : let mut matches = cmd.get_matches();
187 1239 : let mut args = <Args as clap::FromArgMatches>::from_arg_matches_mut(&mut matches)?;
188 :
189 : // I failed to modify opt_pathbuf_parser to return Option<PathBuf> in
190 : // reasonable time, so turn empty string into option post factum.
191 1239 : if let Some(pb) = &args.pg_auth_public_key_path {
192 776 : if pb.as_os_str().is_empty() {
193 1 : args.pg_auth_public_key_path = None;
194 775 : }
195 463 : }
196 1239 : if let Some(pb) = &args.pg_tenant_only_auth_public_key_path {
197 776 : if pb.as_os_str().is_empty() {
198 UBC 0 : args.pg_tenant_only_auth_public_key_path = None;
199 CBC 776 : }
200 463 : }
201 1239 : if let Some(pb) = &args.http_auth_public_key_path {
202 776 : if pb.as_os_str().is_empty() {
203 2 : args.http_auth_public_key_path = None;
204 774 : }
205 463 : }
206 :
207 1239 : if let Some(addr) = args.dump_control_file {
208 754 : let state = control_file::FileStorage::load_control_file(addr)?;
209 754 : let json = serde_json::to_string(&state)?;
210 754 : print!("{json}");
211 754 : return Ok(());
212 485 : }
213 485 :
214 485 : // important to keep the order of:
215 485 : // 1. init logging
216 485 : // 2. tracing panic hook
217 485 : // 3. sentry
218 485 : logging::init(
219 485 : LogFormat::from_config(&args.log_format)?,
220 485 : logging::TracingErrorLayerEnablement::Disabled,
221 485 : logging::Output::Stdout,
222 UBC 0 : )?;
223 CBC 485 : logging::replace_panic_hook_with_tracing_panic_hook().forget();
224 485 : info!("version: {GIT_VERSION}");
225 485 : info!("buld_tag: {BUILD_TAG}");
226 :
227 485 : let args_workdir = &args.datadir;
228 485 : let workdir = args_workdir.canonicalize_utf8().with_context(|| {
229 UBC 0 : format!("Failed to get the absolute path for input workdir {args_workdir:?}")
230 CBC 485 : })?;
231 :
232 : // Change into the data directory.
233 485 : std::env::set_current_dir(&workdir)?;
234 :
235 : // Set or read our ID.
236 485 : let id = set_id(&workdir, args.id.map(NodeId))?;
237 485 : if args.init {
238 UBC 0 : return Ok(());
239 CBC 485 : }
240 :
241 485 : let pg_auth = match args.pg_auth_public_key_path.as_ref() {
242 : None => {
243 464 : info!("pg auth is disabled");
244 464 : None
245 : }
246 21 : Some(path) => {
247 21 : info!("loading pg auth JWT key from {path}");
248 : Some(Arc::new(
249 21 : JwtAuth::from_key_path(path).context("failed to load the auth key")?,
250 : ))
251 : }
252 : };
253 485 : let pg_tenant_only_auth = match args.pg_tenant_only_auth_public_key_path.as_ref() {
254 : None => {
255 463 : info!("pg tenant only auth is disabled");
256 463 : None
257 : }
258 22 : Some(path) => {
259 22 : info!("loading pg tenant only auth JWT key from {path}");
260 : Some(Arc::new(
261 22 : JwtAuth::from_key_path(path).context("failed to load the auth key")?,
262 : ))
263 : }
264 : };
265 485 : let http_auth = match args.http_auth_public_key_path.as_ref() {
266 : None => {
267 465 : info!("http auth is disabled");
268 465 : None
269 : }
270 20 : Some(path) => {
271 20 : info!("loading http auth JWT key(s) from {path}");
272 20 : let jwt_auth = JwtAuth::from_key_path(path).context("failed to load the auth key")?;
273 20 : Some(Arc::new(SwappableJwtAuth::new(jwt_auth)))
274 : }
275 : };
276 :
277 485 : let conf = SafeKeeperConf {
278 485 : workdir,
279 485 : my_id: id,
280 485 : listen_pg_addr: args.listen_pg,
281 485 : listen_pg_addr_tenant_only: args.listen_pg_tenant_only,
282 485 : listen_http_addr: args.listen_http,
283 485 : advertise_pg_addr: args.advertise_pg,
284 485 : availability_zone: args.availability_zone,
285 485 : no_sync: args.no_sync,
286 485 : broker_endpoint: args.broker_endpoint,
287 485 : broker_keepalive_interval: args.broker_keepalive_interval,
288 485 : heartbeat_timeout: args.heartbeat_timeout,
289 485 : peer_recovery_enabled: args.peer_recovery,
290 485 : remote_storage: args.remote_storage,
291 485 : max_offloader_lag_bytes: args.max_offloader_lag,
292 485 : wal_backup_enabled: !args.disable_wal_backup,
293 485 : backup_parallel_jobs: args.wal_backup_parallel_jobs,
294 485 : pg_auth,
295 485 : pg_tenant_only_auth,
296 485 : http_auth,
297 485 : current_thread_runtime: args.current_thread_runtime,
298 485 : };
299 485 :
300 485 : // initialize sentry if SENTRY_DSN is provided
301 485 : let _sentry_guard = init_sentry(
302 485 : Some(GIT_VERSION.into()),
303 485 : &[("node_id", &conf.my_id.to_string())],
304 485 : );
305 970 : start_safekeeper(conf).await
306 : }
307 :
308 : /// Result of joining any of main tasks: upper error means task failed to
309 : /// complete, e.g. panicked, inner is error produced by task itself.
310 : type JoinTaskRes = Result<anyhow::Result<()>, JoinError>;
311 :
312 485 : async fn start_safekeeper(conf: SafeKeeperConf) -> Result<()> {
313 485 : // Prevent running multiple safekeepers on the same directory
314 485 : let lock_file_path = conf.workdir.join(PID_FILE_NAME);
315 485 : let lock_file =
316 485 : pid_file::claim_for_current_process(&lock_file_path).context("claim pid file")?;
317 485 : info!("claimed pid file at {lock_file_path:?}");
318 :
319 : // ensure that the lock file is held even if the main thread of the process is panics
320 : // we need to release the lock file only when the current process is gone
321 485 : std::mem::forget(lock_file);
322 :
323 485 : info!("starting safekeeper WAL service on {}", conf.listen_pg_addr);
324 485 : let pg_listener = tcp_listener::bind(conf.listen_pg_addr.clone()).map_err(|e| {
325 UBC 0 : error!("failed to bind to address {}: {}", conf.listen_pg_addr, e);
326 0 : e
327 CBC 485 : })?;
328 :
329 485 : let pg_listener_tenant_only =
330 485 : if let Some(listen_pg_addr_tenant_only) = &conf.listen_pg_addr_tenant_only {
331 485 : info!(
332 485 : "starting safekeeper tenant scoped WAL service on {}",
333 485 : listen_pg_addr_tenant_only
334 485 : );
335 485 : let listener = tcp_listener::bind(listen_pg_addr_tenant_only.clone()).map_err(|e| {
336 UBC 0 : error!(
337 0 : "failed to bind to address {}: {}",
338 0 : listen_pg_addr_tenant_only, e
339 0 : );
340 0 : e
341 CBC 485 : })?;
342 485 : Some(listener)
343 : } else {
344 UBC 0 : None
345 : };
346 :
347 CBC 485 : info!(
348 485 : "starting safekeeper HTTP service on {}",
349 485 : conf.listen_http_addr
350 485 : );
351 485 : let http_listener = tcp_listener::bind(conf.listen_http_addr.clone()).map_err(|e| {
352 UBC 0 : error!("failed to bind to address {}: {}", conf.listen_http_addr, e);
353 0 : e
354 CBC 485 : })?;
355 :
356 : // Register metrics collector for active timelines. It's important to do this
357 : // after daemonizing, otherwise process collector will be upset.
358 485 : let timeline_collector = safekeeper::metrics::TimelineCollector::new();
359 485 : metrics::register_internal(Box::new(timeline_collector))?;
360 :
361 485 : let (wal_backup_launcher_tx, wal_backup_launcher_rx) = mpsc::channel(100);
362 485 :
363 485 : // Keep handles to main tasks to die if any of them disappears.
364 485 : let mut tasks_handles: FuturesUnordered<BoxFuture<(String, JoinTaskRes)>> =
365 485 : FuturesUnordered::new();
366 485 :
367 485 : // Start wal backup launcher before loading timelines as we'll notify it
368 485 : // through the channel about timelines which need offloading, not draining
369 485 : // the channel would cause deadlock.
370 485 : let current_thread_rt = conf
371 485 : .current_thread_runtime
372 485 : .then(|| Handle::try_current().expect("no runtime in main"));
373 485 : let conf_ = conf.clone();
374 485 : let wal_backup_handle = current_thread_rt
375 485 : .as_ref()
376 485 : .unwrap_or_else(|| WAL_BACKUP_RUNTIME.handle())
377 485 : .spawn(wal_backup::wal_backup_launcher_task_main(
378 485 : conf_,
379 485 : wal_backup_launcher_rx,
380 485 : ))
381 485 : .map(|res| ("WAL backup launcher".to_owned(), res));
382 485 : tasks_handles.push(Box::pin(wal_backup_handle));
383 485 :
384 485 : // Load all timelines from disk to memory.
385 485 : GlobalTimelines::init(conf.clone(), wal_backup_launcher_tx).await?;
386 :
387 485 : let conf_ = conf.clone();
388 485 : // Run everything in current thread rt, if asked.
389 485 : if conf.current_thread_runtime {
390 UBC 0 : info!("running in current thread runtime");
391 CBC 485 : }
392 :
393 485 : let wal_service_handle = current_thread_rt
394 485 : .as_ref()
395 485 : .unwrap_or_else(|| WAL_SERVICE_RUNTIME.handle())
396 485 : .spawn(wal_service::task_main(
397 485 : conf_,
398 485 : pg_listener,
399 485 : Scope::SafekeeperData,
400 485 : ))
401 485 : // wrap with task name for error reporting
402 485 : .map(|res| ("WAL service main".to_owned(), res));
403 485 : tasks_handles.push(Box::pin(wal_service_handle));
404 :
405 485 : if let Some(pg_listener_tenant_only) = pg_listener_tenant_only {
406 485 : let conf_ = conf.clone();
407 485 : let wal_service_handle = current_thread_rt
408 485 : .as_ref()
409 485 : .unwrap_or_else(|| WAL_SERVICE_RUNTIME.handle())
410 485 : .spawn(wal_service::task_main(
411 485 : conf_,
412 485 : pg_listener_tenant_only,
413 485 : Scope::Tenant,
414 485 : ))
415 485 : // wrap with task name for error reporting
416 485 : .map(|res| ("WAL service tenant only main".to_owned(), res));
417 485 : tasks_handles.push(Box::pin(wal_service_handle));
418 485 : }
419 :
420 485 : let conf_ = conf.clone();
421 485 : let http_handle = current_thread_rt
422 485 : .as_ref()
423 485 : .unwrap_or_else(|| HTTP_RUNTIME.handle())
424 485 : .spawn(http::task_main(conf_, http_listener))
425 485 : .map(|res| ("HTTP service main".to_owned(), res));
426 485 : tasks_handles.push(Box::pin(http_handle));
427 485 :
428 485 : let conf_ = conf.clone();
429 485 : let broker_task_handle = current_thread_rt
430 485 : .as_ref()
431 485 : .unwrap_or_else(|| BROKER_RUNTIME.handle())
432 485 : .spawn(broker::task_main(conf_).instrument(info_span!("broker")))
433 485 : .map(|res| ("broker main".to_owned(), res));
434 485 : tasks_handles.push(Box::pin(broker_task_handle));
435 485 :
436 485 : let conf_ = conf.clone();
437 485 : let wal_remover_handle = current_thread_rt
438 485 : .as_ref()
439 485 : .unwrap_or_else(|| WAL_REMOVER_RUNTIME.handle())
440 485 : .spawn(remove_wal::task_main(conf_))
441 485 : .map(|res| ("WAL remover".to_owned(), res));
442 485 : tasks_handles.push(Box::pin(wal_remover_handle));
443 485 :
444 485 : set_build_info_metric(GIT_VERSION, BUILD_TAG);
445 :
446 : // TODO: update tokio-stream, convert to real async Stream with
447 : // SignalStream, map it to obtain missing signal name, combine streams into
448 : // single stream we can easily sit on.
449 485 : let mut sigquit_stream = signal(SignalKind::quit())?;
450 485 : let mut sigint_stream = signal(SignalKind::interrupt())?;
451 485 : let mut sigterm_stream = signal(SignalKind::terminate())?;
452 :
453 : // Notify systemd that we are ready. This is important as currently loading
454 : // timelines takes significant time (~30s in busy regions).
455 485 : if let Err(e) = sd_notify::notify(true, &[NotifyState::Ready]) {
456 UBC 0 : warn!("systemd notify failed: {:?}", e);
457 CBC 485 : }
458 :
459 485 : tokio::select! {
460 UBC 0 : Some((task_name, res)) = tasks_handles.next()=> {
461 0 : error!("{} task failed: {:?}, exiting", task_name, res);
462 : std::process::exit(1);
463 : }
464 : // On any shutdown signal, log receival and exit. Additionally, handling
465 : // SIGQUIT prevents coredump.
466 CBC 374 : _ = sigquit_stream.recv() => info!("received SIGQUIT, terminating"),
467 UBC 0 : _ = sigint_stream.recv() => info!("received SIGINT, terminating"),
468 CBC 111 : _ = sigterm_stream.recv() => info!("received SIGTERM, terminating")
469 :
470 : };
471 485 : std::process::exit(0);
472 UBC 0 : }
473 :
474 : /// Determine safekeeper id.
475 CBC 485 : fn set_id(workdir: &Utf8Path, given_id: Option<NodeId>) -> Result<NodeId> {
476 485 : let id_file_path = workdir.join(ID_FILE_NAME);
477 485 :
478 485 : let my_id: NodeId;
479 485 : // If file with ID exists, read it in; otherwise set one passed.
480 485 : match fs::read(&id_file_path) {
481 89 : Ok(id_serialized) => {
482 89 : my_id = NodeId(
483 89 : std::str::from_utf8(&id_serialized)
484 89 : .context("failed to parse safekeeper id")?
485 89 : .parse()
486 89 : .context("failed to parse safekeeper id")?,
487 : );
488 89 : if let Some(given_id) = given_id {
489 89 : if given_id != my_id {
490 UBC 0 : bail!(
491 0 : "safekeeper already initialized with id {}, can't set {}",
492 0 : my_id,
493 0 : given_id
494 0 : );
495 CBC 89 : }
496 UBC 0 : }
497 CBC 89 : info!("safekeeper ID {}", my_id);
498 : }
499 396 : Err(error) => match error.kind() {
500 : ErrorKind::NotFound => {
501 396 : my_id = if let Some(given_id) = given_id {
502 396 : given_id
503 : } else {
504 UBC 0 : bail!("safekeeper id is not specified");
505 : };
506 CBC 396 : let mut f = File::create(&id_file_path)
507 396 : .with_context(|| format!("Failed to create id file at {id_file_path:?}"))?;
508 396 : f.write_all(my_id.to_string().as_bytes())?;
509 396 : f.sync_all()?;
510 396 : info!("initialized safekeeper id {}", my_id);
511 : }
512 : _ => {
513 UBC 0 : return Err(error.into());
514 : }
515 : },
516 : }
517 CBC 485 : Ok(my_id)
518 485 : }
519 :
520 : // Parse RemoteStorage from TOML table.
521 24 : fn parse_remote_storage(storage_conf: &str) -> anyhow::Result<RemoteStorageConfig> {
522 24 : // funny toml doesn't consider plain inline table as valid document, so wrap in a key to parse
523 24 : let storage_conf_toml = format!("remote_storage = {storage_conf}");
524 24 : let parsed_toml = storage_conf_toml.parse::<Document>()?; // parse
525 24 : let (_, storage_conf_parsed_toml) = parsed_toml.iter().next().unwrap(); // and strip key off again
526 24 : RemoteStorageConfig::from_toml(storage_conf_parsed_toml).and_then(|parsed_config| {
527 24 : // XXX: Don't print the original toml here, there might be some sensitive data
528 24 : parsed_config.context("Incorrectly parsed remote storage toml as no remote storage config")
529 24 : })
530 24 : }
531 :
532 1 : #[test]
533 1 : fn verify_cli() {
534 1 : use clap::CommandFactory;
535 1 : Args::command().debug_assert()
536 1 : }
|
