alemi/dashboard
1
0
Fork 0
mirror of https://git.alemi.dev/dashboard.git synced 2024-11-14 11:59:18 +01:00
Code Issues Projects Releases Packages Wiki Activity

feat: track workers by index

Browse source
This commit is contained in:
əlemi 2022-11-05 19:05:31 +01:00
parent 372e17caad
commit a2e353d18f
Signed by: alemi
GPG key ID: A4895B84D311642C
2 changed files with 11 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
src/main.rs
View file

@ -100,11 +100,11 @@ fn main() {
.block_on(async {
let mut jobs = vec![];
for db_uri in db_uris {
for (i, db_uri) in db_uris.iter().enumerate() {
let db = match Database::connect(db_uri.clone()).await {
Ok(v) => v,
Err(e) => {
error!(target: "worker", "Could not connect to db: {:?}", e);
error!(target: "worker", "Could not connect to db #{}: {:?}", i, e);
return;
}
};
@ -118,6 +118,7 @@ fn main() {
args.interval as i64,
args.cache_time as i64,
run_rx.clone(),
i,
)
)
);

15
src/worker/surveyor.rs
View file

@ -3,7 +3,7 @@ use std::sync::Arc;
use chrono::Utc;
use sea_orm::{DatabaseConnection, ActiveValue::NotSet, Set, EntityTrait};
use tokio::sync::watch;
use tracing::{error, info};
use tracing::error;
use crate::data::{entities, FetchError};
@ -16,6 +16,7 @@ pub async fn surveyor_loop(
interval:i64,
cache_time:i64,
run: watch::Receiver<bool>,
index: usize,
) {
let mut last_activation = Utc::now().timestamp();
let mut last_fetch = 0;
@ -35,14 +36,14 @@ pub async fn surveyor_loop(
match entities::sources::Entity::find().all(&db).await {
Ok(srcs) => sources = srcs,
Err(e) => {
error!(target: "surveyor", "Could not fetch sources: {:?}", e);
error!(target: "surveyor", "[{}] Could not fetch sources: {:?}", index, e);
continue;
}
}
match entities::metrics::Entity::find().all(&db).await {
Ok(mtrcs) => metrics = Arc::new(mtrcs),
Err(e) => {
error!(target: "surveyor", "Could not fetch metrics: {:?}", e);
error!(target: "surveyor", "[{}] Could not fetch metrics: {:?}", index, e);
continue;
}
}
@ -70,7 +71,7 @@ pub async fn surveyor_loop(
if let Err(e) = entities::sources::Entity::update(
entities::sources::ActiveModel{id: Set(source_clone.id), last_update: Set(now), ..Default::default()}
).exec(&db_clone).await {
error!(target: "surveyor", "Failed setting last_update ({:?}) for source {:?} but successfully fetched '{}', aborting", e, source_clone, res);
error!(target: "surveyor", "[{}] Failed setting last_update ({:?}) for source {:?} but successfully fetched '{}', aborting", index, e, source_clone, res);
return;
}
for metric in metrics_snapshot.iter().filter(|x| source_clone.id == x.source_id) {
@ -80,14 +81,14 @@ pub async fn surveyor_loop(
entities::points::ActiveModel {
id: NotSet, metric_id: Set(metric.id), x: Set(v.x), y: Set(v.y),
}).exec(&db_clone).await {
error!(target: "surveyor", "Could not insert record {:?} : {:?}", v, e);
error!(target: "surveyor", "[{}] Could not insert record {:?} : {:?}", index, v, e);
}
},
Err(e) => error!(target: "surveyor", "Failed extracting '{}' from {}: {:?}", metric.name, source_clone.name, e),
Err(e) => error!(target: "surveyor", "[{}] Failed extracting '{}' from {}: {:?}", index, metric.name, source_clone.name, e),
}
}
},
Err(e) => error!(target: "surveyor", "Failed fetching {}: {:?}", source_clone.name, e),
Err(e) => error!(target: "surveyor", "[{}] Failed fetching {}: {:?}", index, source_clone.name, e),
}
});
}