per-service worker monitoring infrastructure (#37)

* per-service worker monitoring infrastructure

* improve worker report style, still basic

bin/frontend.rs

@@ -110,6 +110,7 @@ struct TemplateContext {
  entries: Option<Vec<HashMap<String, String>>>,
  categories: Option<Vec<HashMap<String, String>>>,
  whats: Option<Vec<HashMap<String, String>>>,
+ workers: Option<Vec<HashMap<String, String>>>,
 }
 impl Default for TemplateContext {
  fn default() -> Self {
@@ -120,6 +121,7 @@ impl Default for TemplateContext {
  entries: None,
  categories: None,
  whats: None,
+ workers: None,
  }
  }
 }
@@ -205,6 +207,46 @@ fn admin() -> Template {
  Template::render("admin", context)
 }
 
+#[get("/workers/<service_name>")]
+fn worker_report(service_name: String) -> Result<Template, NotFound<String>> {
+ let backend = Backend::default();
+ let service_name = aux_uri_unescape(Some(&service_name)).unwrap_or_else(|| UNKNOWN.to_string());
+ if let Ok(service) = Service::find_by_name(&service_name, &backend.connection) {
+ let mut global = HashMap::new();
+ global.insert(
+ "title".to_string(),
+ format!("Worker report for service {} ", &service_name),
+ );
+ global.insert(
+ "description".to_string(),
+ format!(
+ "Worker report for service {} as registered by the CorTeX dispatcher",
+ &service_name
+ ),
+ );
+ global.insert("service_name".to_string(), service_name.to_string());
+ global.insert(
+ "service_description".to_string(),
+ service.description.clone(),
+ );
+ let mut context = TemplateContext {
+ global,
+ ..TemplateContext::default()
+ };
+
+ let workers = service
+ .select_workers(&backend.connection)
+ .unwrap()
+ .into_iter()
+ .map(|w| w.into())
+ .collect();
+ context.workers = Some(workers);
+ Ok(Template::render("workers", context))
+ } else {
+ Err(NotFound(String::from("no such service")))
+ }
+}
+
 #[get("/corpus/<corpus_name>")]
 fn corpus(corpus_name: String) -> Result<Template, NotFound<String>> {
  let backend = Backend::default();
@@ -645,6 +687,7 @@ fn rocket() -> rocket::Rocket {
  corpus,
  favicon,
  files,
+ worker_report,
  top_service_report,
  severity_service_report,
  category_service_report,

migrations/2017-10-01-204006_tasks/up.sql

@@ -1,5 +1,6 @@
 -- Your SQL goes here
-CREATE TABLE tasks (
+CREATE TABLE tasks
+(
  id BIGSERIAL PRIMARY KEY,
  service_id INTEGER NOT NULL,
  corpus_id INTEGER NOT NULL,

migrations/2018-08-16-200128_worker_metadata/down.sql

@@ -0,0 +1 @@
+DROP TABLE worker_metadata;

migrations/2018-08-16-200128_worker_metadata/up.sql

@@ -0,0 +1,16 @@
+CREATE TABLE worker_metadata
+(
+ id SERIAL PRIMARY KEY,
+ service_id INTEGER NOT NULL,
+ last_dispatched_task_id BIGINT NOT NULL,
+ last_returned_task_id BIGINT,
+ total_dispatched INTEGER NOT NULL DEFAULT 0,
+ total_returned INTEGER NOT NULL DEFAULT 0,
+ first_seen TIMESTAMP NOT NULL,
+ session_seen TIMESTAMP,
+ time_last_dispatch TIMESTAMP NOT NULL,
+ time_last_return TIMESTAMP,
+ name varchar(200) NOT NULL
+);
+create unique index worker_id_idx on worker_metadata(name, service_id);
+create index worker_service_idx on worker_metadata(service_id);

src/dispatcher/metadata.rs

@@ -0,0 +1,17 @@
+use backend;
+
+enum EventKind {
+ Send,
+ Receive,
+}
+
+pub struct Event {
+ pub kind: EventKind,
+ pub time: String,
+ pub valid: bool,
+ pub task: String,
+ pub service: String,
+ pub identity: String,
+}
+
+pub fn register_event(e: Event) {}

src/dispatcher/sink.rs

@@ -17,7 +17,7 @@ use time;
 use dispatcher::server;
 use helpers;
 use helpers::{TaskProgress, TaskReport, TaskStatus};
-use models::Service;
+use models::{Service, WorkerMetadata};
 
 /// Specifies the binding and operation parameters for a ZMQ sink component
 pub struct Sink {
@@ -166,6 +166,13 @@ impl Sink {
  },
  }
  }
+ // Also update worker metadata for transparency
+ WorkerMetadata::record_received(
+ identity.to_string(),
+ service.id,
+ taskid,
+ self.backend_address.clone(),
+ )?;
  } else {
  // Otherwise just discard the rest of the message
  println!(

src/dispatcher/ventilator.rs

@@ -11,7 +11,7 @@ use backend;
 use dispatcher::server;
 use helpers;
 use helpers::{NewTaskMessage, TaskProgress, TaskReport, TaskStatus};
-use models::Service;
+use models::{Service, WorkerMetadata};
 use std::error::Error;
 use zmq::SNDMORE;
 
@@ -122,12 +122,14 @@ impl Ventilator {
  }
  }
  }
+
  ventilator.send_msg(identity, SNDMORE)?;
+ let mut taskid = -1;
  if let Some(current_task_progress) = task_queue.pop() {
  dispatched_task_opt = Some(current_task_progress.clone());
 
  let current_task = current_task_progress.task;
- let taskid = current_task.id;
+ taskid = current_task.id;
  let serviceid = current_task.service_id;
  println!(
  "vent {}: worker {:?} received task {:?}",
@@ -168,6 +170,7 @@ impl Ventilator {
  );
  } else {
  println!("-- Failed to prepare input stream for taskid {:?}", taskid);
+ taskid = -1;
  ventilator.send(&[], 0)?;
  }
  }
@@ -179,6 +182,13 @@ impl Ventilator {
  ventilator.send_str("0", SNDMORE)?;
  ventilator.send(&[], 0)?;
  }
+ // Update this worker's metadata
+ WorkerMetadata::record_dispatched(
+ identity_str,
+ service.id,
+ taskid,
+ self.backend_address.clone(),
+ )?;
  } else {
  println!(
  "-- No such service in ventilator request: {:?}",