Monitor deletions and get extra job information

src/archive/file.rs

@@ -83,7 +83,7 @@ impl FileArchive {
 impl Archive for FileArchive {
     /// Archives the files from the given SlurmJobEntry's path.
     ///
-    fn archive(&self, job_entry: &Box<dyn JobInfo>) -> Result<(), Error> {
+    fn archive_creation(&self, job_entry: &Box<dyn JobInfo>) -> Result<(), Error> {
         let archive_path = &self.archive_path;
         let target_path = determine_target_path(archive_path, &self.period);
         debug!("Target path: {:?}", target_path);
@@ -94,6 +94,10 @@ impl Archive for FileArchive {
         }
         Ok(())
     }
+
+    fn archive_removal(&self, _job_entry: &Box<dyn JobInfo>) -> Result<(), Error> {
+        Ok(())
+    }
 }
 
 /// Determines the target path for the slurm job file
@@ -196,7 +200,7 @@ mod tests {
 
         let file_archiver = FileArchive::new(&archive_dir, &Period::None);
         let jobinfo: Box<dyn JobInfo> = Box::new(slurm_job_entry);
-        file_archiver.archive(&jobinfo).unwrap();
+        file_archiver.archive_creation(&jobinfo).unwrap();
 
         assert!(Path::is_file(&archive_dir.join("job.1234_environment")));
         assert!(Path::is_file(&archive_dir.join("job.1234_script")));

src/archive/kafka.rs

@@ -140,29 +140,47 @@ impl KafkaArchive {
     }
 }
 
+#[derive(Serialize, Deserialize)]
+enum MessageKind {
+    Creation,
+    Removal,
+}
+
 #[cfg(feature = "kafka")]
 #[derive(Serialize, Deserialize)]
-struct JobMessage {
+struct JobCreationMessage {
     pub id: String,
     pub timestamp: DateTime<Utc>,
     pub cluster: String,
     pub script: String,
     pub environment: Option<HashMap<String, String>>,
+    pub kind: MessageKind,
+}
+
+#[cfg(feature = "kafka")]
+#[derive(Serialize, Deserialize)]
+struct JobRemovalMessage {
+    pub id: String,
+    pub timestamp: DateTime<Utc>,
+    pub cluster: String,
+    pub completion_info: Option<HashMap<String, String>>,
+    pub kind: MessageKind,
 }
 
 impl Archive for KafkaArchive {
-    fn archive(&self, job_entry: &Box<dyn JobInfo>) -> Result<(), Error> {
+    fn archive_creation(&self, job_entry: &Box<dyn JobInfo>) -> Result<(), Error> {
         debug!(
             "Kafka archiver, received an entry for job ID {}",
             job_entry.jobid()
         );
 
-        let doc = JobMessage {
+        let doc = JobCreationMessage {
             id: job_entry.jobid(),
             timestamp: Utc::now(),
             cluster: job_entry.cluster(),
             script: job_entry.script(),
             environment: job_entry.extra_info(),
+            kind: MessageKind::Creation,
         };
 
         if let Ok(serial) = serde_json::to_string(&doc) {
@@ -187,6 +205,42 @@ impl Archive for KafkaArchive {
             ))
         }
     }
+
+    fn archive_removal(&self, job_entry: &Box<dyn JobInfo>) -> Result<(), Error> {
+        debug!(
+            "Kafka archiver, received an removal entry for job ID {}",
+            job_entry.jobid()
+        );
+
+        let doc = JobRemovalMessage {
+            id: job_entry.jobid(),
+            timestamp: Utc::now(),
+            cluster: job_entry.cluster(),
+            completion_info: job_entry.extra_completion_info(),
+            kind: MessageKind::Removal,
+        };
+
+        if let Ok(serial) = serde_json::to_string(&doc) {
+            match self
+                .producer
+                .send::<str, str>(BaseRecord::to(&self.topic).payload(&serial))
+            {
+                Ok(_) => {
+                    debug!("Message produced correctly");
+                    Ok(())
+                }
+                Err((_e, _)) => {
+                    debug!("Could not produce job entry");
+                    Ok(())
+                }
+            }
+        } else {
+            Err(Error::new(
+                ErrorKind::InvalidData,
+                "Cannot convert job info to JSON",
+            ))
+        }
+    }
 }
 
 #[cfg(test)]

src/archive/mod.rs

@@ -49,7 +49,8 @@ pub enum Archiver {
 /// The Archive trait should be implemented by every backend.
 #[allow(clippy::borrowed_box)]
 pub trait Archive: Send {
-    fn archive(&self, slurm_job_entry: &Box<dyn JobInfo>) -> Result<(), Error>;
+    fn archive_creation(&self, job_entry: &Box<dyn JobInfo>) -> Result<(), Error>;
+    fn archive_removal(&self, job_entry: &Box<dyn JobInfo>) -> Result<(), Error>;
 }
 
 pub fn archive_builder(archiver: &Archiver) -> Result<Box<dyn Archive>, Error> {
@@ -66,29 +67,29 @@ pub fn archive_builder(archiver: &Archiver) -> Result<Box<dyn Archive>, Error> {
     }
 }
 
-/// The process function consumes job entries and call the archive function for each
+/// The process function consumes creation job entries and calls the archive function for each
 /// received entry.
 /// At the same time, it also checks if there is an incoming notification that it should
 /// stop processing. Upon receipt, it will cease operations immediately.
-pub fn process(
+pub fn process_create(
     archiver: Box<dyn Archive>,
     r: &Receiver<Box<dyn JobInfo>>,
     sigchannel: &Receiver<bool>,
     cleanup: bool,
 ) -> Result<(), Error> {
-    info!("Start processing events");
+    info!("Start processing create events");
 
     #[allow(clippy::zero_ptr, clippy::drop_copy)]
     loop {
         select! {
-            recv(sigchannel) -> b => if let Ok(true) = b  {
+            recv(sigchannel) -> b => if let Ok(true) = b {
                 if !cleanup {
                     info!("Stopped processing entries, {} skipped", r.len());
                 } else {
                     info!("Processing {} entries, then stopping", r.len());
                     for mut entry in r.iter() {
                         entry.read_job_info()?;
-                        archiver.archive(&entry)?;
+                        archiver.archive_creation(&entry)?;
                     }
                     info!("Done processing");
                 }
@@ -104,7 +105,7 @@ pub fn process(
                         sleep(dur);
                     }
                     job_entry.read_job_info()?;
-                    archiver.archive(&job_entry)?;
+                    archiver.archive_creation(&job_entry)?;
                 } else {
                     error!("Error on receiving JobEntry info");
                     break;
@@ -113,7 +114,51 @@ pub fn process(
         }
     }
 
-    debug!("Processing loop exited");
+    debug!("Processing creation loop exited");
+    Ok(())
+}
+
+/// The process_remove function consumes job removal events and gets the necessary data from the
+/// scheduler before sending the information to the archive.
+/// At the same time, it also checks if there is an incoming notification that it should
+/// stop processing. Upon receipt, it will cease operations immediately.
+pub fn process_remove(
+    archiver: Box<dyn Archive>,
+    r: &Receiver<Box<dyn JobInfo>>,
+    sigchannel: &Receiver<bool>,
+    cleanup: bool,
+) -> Result<(), Error> {
+    info!("Start processing removal events");
+
+    #[allow(clippy::zero_ptr, clippy::drop_copy)]
+    loop {
+        select! {
+            recv(sigchannel) -> b => if let Ok(true) = b {
+                if !cleanup {
+                        info!("Stopped processing entries, {} skipped", r.len());
+                } else {
+                    info!("Processing {} entries, then stopping", r.len());
+                    for mut entry in r.iter() {
+                        entry.read_job_info()?;
+                        archiver.archive_removal(&entry)?;
+                    }
+                    info!("Done processing");
+                }
+                break;
+            },
+            recv(r) -> entry => {
+                if let Ok(mut job_entry) = entry {
+                    job_entry.job_completion_info()?;
+                    archiver.archive_removal(&job_entry)?;
+                } else {
+                    error!("Error on receiving JobEntry info");
+                    break;
+                }
+            }
+        }
+    }
+
+    debug!("Processing removal loop ended");
     Ok(())
 }
 
@@ -133,8 +178,13 @@ mod tests {
     struct DummyArchiver;
 
     impl Archive for DummyArchiver {
-        fn archive(&self, _: &Box<dyn JobInfo>) -> Result<(), Error> {
-            info!("Archiving");
+        fn archive_creation(&self, _: &Box<dyn JobInfo>) -> Result<(), Error> {
+            info!("Archiving creation");
+            Ok(())
+        }
+
+        fn archive_removal(&self, job_entry: &Box<dyn JobInfo>) -> Result<(), Error> {
+            info!("Archiving removal");
             Ok(())
         }
     }
@@ -148,7 +198,7 @@ mod tests {
         scope(|s| {
             let path = PathBuf::from(current_dir().unwrap().join("tests/job.123456"));
             let slurm_job_entry = SlurmJobEntry::new(&path, "123456", "mycluster");
-            s.spawn(move |_| match process(archiver, &rx1, &rx2, false) {
+            s.spawn(move |_| match process_create(archiver, &rx1, &rx2, false) {
                 Ok(v) => assert_eq!(v, ()),
                 Err(_) => panic!("Unexpected error from process function"),
             });

src/main.rs

@@ -36,7 +36,7 @@ mod monitor;
 mod scheduler;
 mod utils;
 
-use archive::{archive_builder, process, Archive, Archiver};
+use archive::{archive_builder, process_create, process_remove, Archive, Archiver};
 
 use monitor::monitor;
 use scheduler::{create, SchedulerKind};
@@ -122,7 +122,8 @@ fn main() -> Result<(), std::io::Error> {
     }
 
     let scheduler_kind = cli.scheduler;
-    let archiver: Box<dyn Archive> = archive_builder(&cli.archiver).unwrap();
+    let create_archiver: Box<dyn Archive> = archive_builder(&cli.archiver).unwrap();
+    let remove_archiver: Box<dyn Archive> = archive_builder(&cli.archiver).unwrap();
     let cluster = cli.cluster;
 
     info!("sarchive starting. Watching spool {:?}.", &base);
@@ -137,8 +138,14 @@ fn main() -> Result<(), std::io::Error> {
     let (sig_sender, sig_receiver) = bounded(20);
     let cleanup = cli.cleanup;
 
-    // we will watch the locations provided by the scheduler
-    let (sender, receiver) = unbounded();
+    // We build two pipelines.
+    // - the create pipeline needs to process events quickly, as we need to get the information
+    //   before the job fails, is cancelled, or runs to completion
+    // - the remove pipeline processes events that can fetch their information at leisure, since
+    //   the required data can be requested at all times from the scheduler
+    let (create_sender, create_receiver) = unbounded();
+    let (remove_sender, remove_receiver) = unbounded();
+
     let sched = create(&scheduler_kind, &base, &cluster);
     if let Err(e) = scope(|s| {
         let ss = &sig_sender;
@@ -148,11 +155,12 @@ fn main() -> Result<(), std::io::Error> {
         });
 
         for loc in sched.watch_locations() {
-            let t = &sender;
+            let cs = &create_sender;
+            let rs = &remove_sender;
             let sr = &sig_receiver;
             let sl = &sched;
             let b = &base;
-            s.spawn(move |_| match monitor(sl, &loc, t, sr) {
+            s.spawn(move |_| match monitor(sl, &loc, cs, rs, sr) {
                 Ok(_) => info!("Stopped watching location {:?}", &loc),
                 Err(e) => {
                     error!("{:?}", e);
@@ -161,12 +169,21 @@ fn main() -> Result<(), std::io::Error> {
             });
         }
 
-        let r = &receiver;
+        let cr = &create_receiver;
+        let sr = &sig_receiver;
+        s.spawn(move |_| {
+            match process_create(create_archiver, cr, sr, cleanup) {
+                Ok(()) => info!("Processing creation completed succesfully"),
+                Err(e) => error!("Processing creation failed: {:?}", e),
+            };
+        });
+
+        let rr = &remove_receiver;
         let sr = &sig_receiver;
         s.spawn(move |_| {
-            match process(archiver, r, sr, cleanup) {
-                Ok(()) => info!("Processing completed succesfully"),
-                Err(e) => error!("processing failed: {:?}", e),
+            match process_remove(remove_archiver, rr, sr, cleanup) {
+                Ok(()) => info!("Processing removal completed succesfully"),
+                Err(e) => error!("Processing removal failed: {:?}", e),
             };
         });
     }) {