13617540422

Committed 02 Mar 2025 05:04PM UTC coverage: 32.874% (-1.8%) from 34.692%

Build # 13617540422

Build Type

push

github

Committed by

dariusbakunas

Commit Message

feat: add version checks and ability to link plugins statically

Coverage Stats

0 of 144 new or added lines in 13 files covered. (0.0%)

5 existing lines in 3 files now uncovered.

715 of 2175 relevant lines covered (32.87%)

1.09 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/cogrs-core/src/executor/task_executor.rs

use crate::executor::worker_message::WorkerMessage;
use crate::inventory::host::Host;
use crate::playbook::task::Task;
use crate::vars::variable::Variable;
use anyhow::Result;
use cogrs_modules::action_handler::ActionHandler;
use cogrs_modules::task_result::TaskResult;
use cogrs_plugins::connection::ConnectionPlugin;
use cogrs_plugins::shell::ShellPlugin;
use indexmap::IndexMap;
use log::debug;
use tokio::sync::mpsc;

pub struct TaskExecutor;

#[cfg(feature = "static-plugins")]
async fn load_plugins(
    task: &Task,
    mut task_vars: IndexMap<String, Variable>,
) -> Result<(Box<dyn ConnectionPlugin>, Box<dyn ShellPlugin>)> {
    use sh_lib::Sh;
    use ssh_lib::Ssh;

    let mut connection_plugin = Box::new(Ssh::default());

    let parameters = serde_json::to_string(&task_vars)?;
    connection_plugin.initialize(&parameters)?;

    let shell_plugin = Box::new(Sh::default());
    Ok((connection_plugin, shell_plugin))
}

#[cfg(not(feature = "static-plugins"))]
async fn load_plugins(
    task: &Task,
    mut task_vars: IndexMap<String, Variable>,
) -> Result<(Box<dyn ConnectionPlugin>, Box<dyn ShellPlugin>)> {
    let plugin_loader = cogrs_plugins::plugin_loader::PluginLoader::instance();
    let mut loader = plugin_loader.lock().await;

    let shell_plugin = loader.get_shell_plugin("sh").await?;
    let mut connection_plugin = loader.get_connection_plugin(task.connection()).await?;
    let parameters = serde_json::to_string(&task_vars)?;
    connection_plugin.initialize(&parameters)?;
    Ok((connection_plugin, shell_plugin))
}

impl TaskExecutor {
    pub fn new() -> Self {
        Self {}
    }

    pub async fn run(
        &self,
        host: &Host,
        task: &Task,
        mut task_vars: IndexMap<String, Variable>,
        sender: &mpsc::Sender<WorkerMessage>,
    ) -> Result<TaskResult> {
        debug!(
            "executor run() - task {}, host: {}",
            task.uuid(),
            host.name()
        );

        task_vars.insert(
            String::from("host"),
            Variable::String(host.address().to_string()),
        );

        task_vars.insert(
            String::from("remote_user"),
            Variable::String("test-user".to_string()),
        );

        // TODO: handle conditionals

        let result = TaskResult::new(host.name(), task.uuid());
        // TODO: handle with_*
        // TODO: get connection plugin
        let (connection_plugin, shell_plugin) = load_plugins(task, task_vars).await?;
        let action_handler = ActionHandler::new(connection_plugin, shell_plugin);
        action_handler.run().await?;

        Ok(result)
    }

    fn get_connection(current_connection: &str) {}
}

1	use crate::executor::worker_message::WorkerMessage;
2	use crate::inventory::host::Host;
3	use crate::playbook::task::Task;
4	use crate::vars::variable::Variable;
5	use anyhow::Result;
6	use cogrs_modules::action_handler::ActionHandler;
7	use cogrs_modules::task_result::TaskResult;
8	use cogrs_plugins::connection::ConnectionPlugin;
9	use cogrs_plugins::shell::ShellPlugin;
10	use indexmap::IndexMap;
11	use log::debug;
12	use tokio::sync::mpsc;
13
14	pub struct TaskExecutor;
15
16	#[cfg(feature = "static-plugins")]
NEW 17	async fn load_plugins(	×
18	task: &Task,
19	mut task_vars: IndexMap<String, Variable>,
20	) -> Result<(Box<dyn ConnectionPlugin>, Box<dyn ShellPlugin>)> {
21	use sh_lib::Sh;
22	use ssh_lib::Ssh;
23
NEW 24	let mut connection_plugin = Box::new(Ssh::default());	×
25
NEW 26	let parameters = serde_json::to_string(&task_vars)?;	×
NEW 27	connection_plugin.initialize(&parameters)?;	×
28
NEW 29	let shell_plugin = Box::new(Sh::default());	×
NEW 30	Ok((connection_plugin, shell_plugin))	×
31	}
32
33	#[cfg(not(feature = "static-plugins"))]
34	async fn load_plugins(
35	task: &Task,
36	mut task_vars: IndexMap<String, Variable>,
37	) -> Result<(Box<dyn ConnectionPlugin>, Box<dyn ShellPlugin>)> {
38	let plugin_loader = cogrs_plugins::plugin_loader::PluginLoader::instance();
39	let mut loader = plugin_loader.lock().await;
40
41	let shell_plugin = loader.get_shell_plugin("sh").await?;
42	let mut connection_plugin = loader.get_connection_plugin(task.connection()).await?;
43	let parameters = serde_json::to_string(&task_vars)?;
44	connection_plugin.initialize(&parameters)?;
45	Ok((connection_plugin, shell_plugin))
46	}
47
48	impl TaskExecutor {
49	pub fn new() -> Self {
50	Self {}
51	}
52
53	pub async fn run(	×
54	&self,
55	host: &Host,
56	task: &Task,
57	mut task_vars: IndexMap<String, Variable>,
58	sender: &mpsc::Sender<WorkerMessage>,
59	) -> Result<TaskResult> {
60	debug!(	×
61	"executor run() - task {}, host: {}",
62	task.uuid(),
63	host.name()
64	);
65
66	task_vars.insert(	×
67	String::from("host"),	×
68	Variable::String(host.address().to_string()),	×
69	);
70
NEW 71	task_vars.insert(	×
NEW 72	String::from("remote_user"),	×
NEW 73	Variable::String("test-user".to_string()),	×
74	);
75
76	// TODO: handle conditionals
77
78	let result = TaskResult::new(host.name(), task.uuid());	×
79	// TODO: handle with_*
80	// TODO: get connection plugin
NEW 81	let (connection_plugin, shell_plugin) = load_plugins(task, task_vars).await?;	×
NEW 82	let action_handler = ActionHandler::new(connection_plugin, shell_plugin);	×
NEW 83	action_handler.run().await?;	×
84
85	Ok(result)	×
86	}
87
88	fn get_connection(current_connection: &str) {}	×
89	}

dariusbakunas / cogrs / 13617540422

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous