Fixes #15206: Handling keept_output and asynchronous parameters in the remote-run API

relay/sources/relayd/src/configuration.rs

@@ -460,7 +460,7 @@ mod tests {
                 },
             },
             remote_run: RemoteRun {
-                command: PathBuf::from("/opt/rudder/bin/rudder"),
+                command: PathBuf::from("tests/fake_agent.sh"),
             },
         };
         assert_eq!(config.unwrap(), reference);

relay/sources/relayd/src/remote_run.rs

@@ -69,28 +69,49 @@ impl RemoteRun {
         &self,
         job_config: Arc<JobConfig>,
     ) -> Result<impl warp::reply::Reply, warp::reject::Rejection> {
+        if self.run_parameters.keep_output && self.run_parameters.asynchronous {
+            return Err(warp::reject::custom(
+                "keep_output and asynchronous cannot be true simultaneously",
+            ));
+        }
         if self.target == RemoteRunTarget::All {
-            info!("conditions OK");
             info!("remote-run triggered on all the nodes");
 
-            for node in job_config
-                .nodes
-                .read()
-                .expect("Cannot read nodes list")
-                .get_neighbors_from_target(RemoteRunTarget::All)
+            if self.run_parameters.keep_output
+                && self.target.get_connected_nodes(job_config.clone()).len() == 1
             {
-                info!("command executed :  \n on node {}", node);
+                Ok(warp::reply::html(hyper::Body::wrap_stream(
+                    self.run_parameters.execute_agent_output(
+                        job_config.clone(),
+                        self.target.get_connected_nodes(job_config.clone()),
+                    ),
+                )))
+            } else {
+                self.run_parameters.execute_agent_no_output(
+                    job_config.clone(),
+                    self.target.get_connected_nodes(job_config.clone()),
+                );
+                Ok(warp::reply::html(hyper::Body::empty()))
             }
-            Ok(warp::reply::html(hyper::Body::wrap_stream(
-                self.run_parameters.execute_agent(job_config),
-            )))
         } else {
-            info!("conditions OK");
             info!("Remote run launched on nodes: {:?}", self.target);
 
-            Ok(warp::reply::html(hyper::Body::wrap_stream(
-                self.run_parameters.execute_agent(job_config),
-            )))
+            if self.run_parameters.keep_output
+                && self.target.get_connected_nodes(job_config.clone()).len() == 1
+            {
+                Ok(warp::reply::html(hyper::Body::wrap_stream(
+                    self.run_parameters.execute_agent_output(
+                        job_config.clone(),
+                        self.target.get_connected_nodes(job_config.clone()),
+                    ),
+                )))
+            } else {
+                self.run_parameters.execute_agent_no_output(
+                    job_config.clone(),
+                    self.target.get_connected_nodes(job_config.clone()),
+                );
+                Ok(warp::reply::html(hyper::Body::empty()))
+            }
         }
     }
 }
@@ -101,6 +122,19 @@ pub enum RemoteRunTarget {
     Nodes(Vec<String>),
 }
 
+impl RemoteRunTarget {
+    pub fn get_connected_nodes(&self, job_config: Arc<JobConfig>) -> Vec<String> {
+        match &self {
+            RemoteRunTarget::All => job_config
+                .nodes
+                .read()
+                .expect("Cannot read nodes list")
+                .get_neighbors_from_target(RemoteRunTarget::All),
+            RemoteRunTarget::Nodes(nodeslist) => nodeslist.clone(),
+        }
+    }
+}
+
 #[derive(Debug)]
 pub struct Condition {
     data: String,
@@ -180,17 +214,8 @@ impl RunParameters {
         })
     }
 
-    pub fn command(
-        &self,
-        is_root: bool,
-        test_mode: bool,
-        job_config: Arc<JobConfig>,
-    ) -> (PathBuf, Vec<String>) {
-        let program = if test_mode {
-            PathBuf::from("echo")
-        } else {
-            job_config.cfg.remote_run.command.clone()
-        };
+    pub fn command(&self, is_root: bool, job_config: Arc<JobConfig>) -> (PathBuf, Vec<String>) {
+        let program = job_config.cfg.remote_run.command.clone();
 
         let mut args = vec![];
         args.push(if is_root { "agent" } else { "remote" }.to_string());
@@ -201,18 +226,20 @@ impl RunParameters {
                 self.conditions.iter().map(|c| c.data.clone()).collect();
             args.push(conditions_argument.join(","));
         }
-
         (program, args)
     }
 
-    pub fn execute_agent(
+    pub fn execute_agent_output(
         &self,
         job_config: Arc<JobConfig>,
+        nodeslist: Vec<String>,
     ) -> impl Stream<Item = hyper::Chunk, Error = Error> + Send + 'static {
-        let (program, args) = self.command(false, true, job_config);
-        let mut cmd = Command::new(program);
-        cmd.args(args);
+        let (program, args) = self.command(false, job_config);
 
+        let mut cmd = Command::new(&program);
+        cmd.args(&args);
+        cmd.arg("-H".to_string());
+        cmd.arg(&nodeslist[0]);
         cmd.stdout(Stdio::piped());
         let mut child = cmd.spawn_async().expect("failed to spawn command");
         let lines = lines_stream(&mut child);
@@ -223,6 +250,29 @@ impl RunParameters {
         tokio::spawn(child_future);
         lines
     }
+
+    pub fn execute_agent_no_output(
+        &self,
+        job_config: Arc<JobConfig>,
+        nodeslist: Vec<String>,
+    ) -> impl Stream<Item = hyper::Chunk, Error = Error> + Send + 'static {
+        let (program, args) = self.command(false, job_config);
+
+        for node in nodeslist {
+            let mut cmd = Command::new(&program);
+            cmd.args(&args);
+            cmd.arg("-H".to_string());
+            cmd.arg(node);
+            cmd.stdout(Stdio::piped());
+            let child = cmd.spawn_async().expect("failed to spawn command");
+            let child_future = child
+                .map(|_status| info!(""))
+                .map_err(|e| panic!("error while running child: {}", e));
+
+            tokio::spawn(child_future);
+        }
+        futures::stream::empty()
+    }
 }
 
 #[cfg(test)]

relay/sources/relayd/tests/fake_agent.sh

@@ -0,0 +1,2 @@
+#!/bin/bash
+echo -n "$@" > ./target/tmp/api_test.txt

relay/sources/relayd/tests/files/config/main.conf

@@ -34,4 +34,4 @@ user = "rudder"
 password = "password"
 
 [remote_run]
-command = "/opt/rudder/bin/rudder"
+command = "tests/fake_agent.sh"

relay/sources/relayd/tests/relay_api.rs

@@ -1,6 +1,6 @@
 use relayd::{configuration::CliConfiguration, init_logger, start};
 use reqwest;
-use std::{thread, time};
+use std::{fs::read_to_string, thread, time};
 
 fn start_api() -> Result<(), ()> {
     let mut retry = 10;
@@ -47,4 +47,36 @@ mod tests {
 
         assert_eq!(res.unwrap().text().unwrap(), "Unhandled rejection: Invalid agent Condition : Invalid agent Condition : Wrong condition: \'clas~1\', it should match ^[a-zA-Z0-9][a-zA-Z0-9_]*$".to_string());
     }
+
+    #[test]
+    fn it_processes_the_parameters() {
+        let cli_cfg = CliConfiguration::new("tests/test_simple/config/", false);
+
+        thread::spawn(move || {
+            start(cli_cfg, init_logger().unwrap()).unwrap();
+        });
+
+        assert!(start_api().is_ok());
+
+        let client = reqwest::Client::new();
+
+        let params = [
+            ("asynchronous", "false"),
+            ("keep_output", "true"),
+            ("classes", "class2,class3"),
+            ("nodes", "server.rudder.local"),
+        ];
+
+        client
+            .post("http://localhost:3030/rudder/relay-api/remote-run/nodes")
+            .form(&params)
+            .send().unwrap();
+
+        let data = read_to_string("target/tmp/api_test.txt").expect("Unable to read file");
+
+        assert_eq!(
+            "remote run -D class2,class3 -H server.rudder.local".to_string(),
+            data
+        );
+    }
 }

relay/sources/relayd/tests/test_simple/config/main.conf

@@ -29,4 +29,4 @@ upstream.user = "rudder"
 upstream.password = "password"
 
 [remote_run]
-command = "/opt/rudder/bin/rudder"
+command = "tests/fake_agent.sh"

relay/sources/relayd/tools/config/main.conf

@@ -52,3 +52,4 @@ password = "password"
 
 [remote_run]
 command = "/opt/rudder/bin/rudder"
+