Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 29 additions & 8 deletions opsqueue/app/main.rs
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ use opentelemetry_resource_detectors::{OsResourceDetector, ProcessResourceDetect
use opentelemetry_sdk::trace::{RandomIdGenerator, Sampler, SdkTracerProvider};
use opsqueue::{common::submission::db::periodically_cleanup_old, config::Config, prometheus};
use std::{
error::Error,
sync::{atomic::AtomicBool, Arc},
time::Duration,
};
Comment thread
ReinierMaas marked this conversation as resolved.
Expand Down Expand Up @@ -47,9 +48,9 @@ pub async fn async_main() {
.expect("Timed out while initiating the database");

moro_local::async_scope!(|scope| {
scope.spawn(db_pool.periodically_checkpoint_wal());
let checkpoint_handle = scope.spawn(db_pool.periodically_checkpoint_wal());

scope.spawn(opsqueue::server::serve_producer_and_consumer(
let server_handle = scope.spawn(opsqueue::server::serve_producer_and_consumer(
config,
&server_addr,
&db_pool,
Expand All @@ -60,15 +61,15 @@ pub async fn async_main() {
));

let max_age = config.max_submission_age.into();
scope.spawn(periodically_cleanup_old(db_pool.writer_pool(), max_age));
let cleanup_handle = scope.spawn(periodically_cleanup_old(db_pool.writer_pool(), max_age));

scope.spawn(prometheus::periodically_calculate_scaling_metrics(
let prometheus_handle = scope.spawn(prometheus::periodically_calculate_scaling_metrics(
&db_pool,
&cancellation_token,
));

// Set up complete. Start up watchdog, which will mark app healthy when appropriate
scope.spawn(opsqueue::server::app_watchdog(
let watchdog_handle = scope.spawn(opsqueue::server::app_watchdog(
app_healthy_flag.clone(),
&db_pool,
cancellation_token.clone(),
Expand All @@ -79,9 +80,29 @@ pub async fn async_main() {
opsqueue::version_info()
);

tokio::signal::ctrl_c()
.await
.expect("Failed to set up Ctrl+C signal handler");
tokio::select! {
_ = checkpoint_handle => {
tracing::error!("Checkpointing task exited unexpectedly");
},
_ = server_handle => {
tracing::error!("Server task exited unexpectedly");
},
_ = cleanup_handle => {
tracing::error!("Cleanup task exited unexpectedly");
},
_ = prometheus_handle => {
tracing::error!("Prometheus metrics calculation task exited unexpectedly");
},
_ = watchdog_handle => {
tracing::error!("Watchdog task exited unexpectedly");
},
res = tokio::signal::ctrl_c() => {
match res {
Ok(_) => tracing::warn!("Received Ctrl-C signal"),
Err(ref err) => tracing::error!(error = err as &dyn Error, "Error while waiting for Ctrl-C signal"),
};
},
};
Comment on lines +83 to +105

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

These are all Future<Output = ()>, nothing to see here...


tracing::warn!("Opsqueue is shutting down");

Expand Down
Loading