sqlite/libsql
0
0
Fork 0
mirror of https://github.com/tursodatabase/libsql.git synced 2025-08-15 16:03:04 +00:00
Code Activity

Make heartbeat_url optional

Browse Source 
Signed-off-by: Piotr Jastrzebski <piotr@chiselstrike.com>
This commit is contained in:
Piotr Jastrzebski
2023-11-03 11:47:58 +01:00
parent de02b00a7e
commit fd65d6eda6
4 changed files with 24 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
libsql-server/src/config.rs
View File

@@ -192,7 +192,7 @@ impl DbConfig {
}
pub struct HeartbeatConfig {
pub heartbeat_url: String,
pub heartbeat_url: Option<String>,
pub heartbeat_period: Duration,
pub heartbeat_auth: Option<String>,
}

30
libsql-server/src/heartbeat.rs
View File

@@ -12,7 +12,7 @@ use crate::namespace::NamespaceName;
use crate::stats::Stats;
pub async fn server_heartbeat(
url: Url,
url: Option<Url>,
auth: Option<String>,
update_period: Duration,
mut stats_subs: mpsc::Receiver<(NamespaceName, Weak<Stats>)>,
@@ -27,7 +27,7 @@ pub async fn server_heartbeat(
watched.insert(ns, stats);
}
_ = interval.tick() => {
send_stats(&mut watched, &client, &url, auth.as_deref()).await;
send_stats(&mut watched, &client, url.as_ref(), auth.as_deref()).await;
}
};
}
@@ -36,24 +36,26 @@ pub async fn server_heartbeat(
async fn send_stats(
watched: &mut HashMap<NamespaceName, Weak<Stats>>,
client: &reqwest::Client,
url: &Url,
url: Option<&Url>,
auth: Option<&str>,
) {
// first send all the stats...
for (ns, stats) in watched.iter() {
if let Some(stats) = stats.upgrade() {
let body = StatsResponse::from(stats.as_ref());
let mut url = url.clone();
url.path_segments_mut().unwrap().push(ns.as_str());
let request = client.post(url);
let request = if let Some(ref auth) = auth {
request.header("Authorization", auth.to_string())
} else {
request
};
let request = request.json(&body);
if let Err(err) = request.send().await {
tracing::warn!("Error sending heartbeat: {}", err);
if let Some(url) = url {
let mut url = url.clone();
url.path_segments_mut().unwrap().push(ns.as_str());
let request = client.post(url);
let request = if let Some(ref auth) = auth {
request.header("Authorization", auth.to_string())
} else {
request
};
let request = request.json(&body);
if let Err(err) = request.send().await {
tracing::warn!("Error sending heartbeat: {}", err);
}
}
}
}

9
libsql-server/src/lib.rs
View File

@@ -323,14 +323,17 @@ where
Some(ref config) => {
tracing::info!(
"Server sending heartbeat to URL {} every {:?}",
config.heartbeat_url,
config.heartbeat_url.as_deref().unwrap_or("<not supplied>"),
config.heartbeat_period,
);
join_set.spawn({
let heartbeat_auth = config.heartbeat_auth.clone();
let heartbeat_period = config.heartbeat_period;
let heartbeat_url =
Url::from_str(&config.heartbeat_url).context("invalid heartbeat URL")?;
let heartbeat_url = if let Some(url) = &config.heartbeat_url {
Some(Url::from_str(&url).context("invalid heartbeat URL")?)
} else {
None
};
async move {
heartbeat::server_heartbeat(
heartbeat_url,

2
libsql-server/src/main.rs
View File

@@ -439,7 +439,7 @@ async fn make_rpc_client_config(config: &Cli) -> anyhow::Result<Option<RpcClient
fn make_hearbeat_config(config: &Cli) -> Option<HeartbeatConfig> {
Some(HeartbeatConfig {
heartbeat_url: config.heartbeat_url.clone()?,
heartbeat_url: config.heartbeat_url.clone(),
heartbeat_period: Duration::from_secs(config.heartbeat_period_s),
heartbeat_auth: config.heartbeat_auth.clone(),
})