WIP: Generic activation program #9
3 changed files with 30 additions and 51 deletions
|
@ -134,12 +134,12 @@ with builtins; rec {
|
||||||
type = types.nullOr types.str;
|
type = types.nullOr types.str;
|
||||||
default = "root";
|
default = "root";
|
||||||
};
|
};
|
||||||
connectionOptions = lib.mkOption {
|
connectionUri = lib.mkOption {
|
||||||
description = mdDoc ''
|
description = mdDoc ''
|
||||||
Connection options given to the activation program.
|
Connection options given to the activation program.
|
||||||
'';
|
'';
|
||||||
type = types.attrsOf types.str;
|
type = types.str;
|
||||||
default = { };
|
default = "ssh://localhost";
|
||||||
};
|
};
|
||||||
allowLocalDeployment = lib.mkOption {
|
allowLocalDeployment = lib.mkOption {
|
||||||
description = mdDoc ''
|
description = mdDoc ''
|
||||||
|
@ -271,15 +271,6 @@ with builtins; rec {
|
||||||
'';
|
'';
|
||||||
type = types.path;
|
type = types.path;
|
||||||
};
|
};
|
||||||
protocol = lib.mkOption {
|
|
||||||
type = types.str;
|
|
||||||
default = "local";
|
|
||||||
description = mdDoc ''
|
|
||||||
Which protocol to use to deploy.
|
|
||||||
|
|
||||||
It can be ssh, local, etc. but strongly depends on `activationProgram`
|
|
||||||
'';
|
|
||||||
};
|
|
||||||
};
|
};
|
||||||
};
|
};
|
||||||
registryOptions = { lib, ... }: let
|
registryOptions = { lib, ... }: let
|
||||||
|
|
|
@ -3,7 +3,7 @@ use std::process::Stdio;
|
||||||
|
|
||||||
use async_trait::async_trait;
|
use async_trait::async_trait;
|
||||||
use serde::{Deserialize, Serialize};
|
use serde::{Deserialize, Serialize};
|
||||||
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, BufWriter};
|
use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader, BufWriter};
|
||||||
use tokio::process::{Child, ChildStdin, ChildStdout, Command};
|
use tokio::process::{Child, ChildStdin, ChildStdout, Command};
|
||||||
|
|
||||||
use super::{CopyDirection, CopyOptions, Host, RebootOptions};
|
use super::{CopyDirection, CopyOptions, Host, RebootOptions};
|
||||||
|
@ -32,11 +32,8 @@ pub struct Goal {
|
||||||
/// A request to the activation program
|
/// A request to the activation program
|
||||||
#[derive(Clone, Debug, Serialize, Deserialize)]
|
#[derive(Clone, Debug, Serialize, Deserialize)]
|
||||||
pub enum Request {
|
pub enum Request {
|
||||||
/// Asks for activation program's capabilities
|
|
||||||
Capabilities,
|
|
||||||
|
|
||||||
Connection {
|
Connection {
|
||||||
connection_options: HashMap<String, String>,
|
connection_uri: String,
|
||||||
},
|
},
|
||||||
|
|
||||||
/// Copy closure to/from host
|
/// Copy closure to/from host
|
||||||
|
@ -74,25 +71,19 @@ pub enum Request {
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
|
||||||
pub enum CapabilityResponse {
|
|
||||||
V1 {
|
|
||||||
supported_transports: Vec<Transport>,
|
|
||||||
supported_goals: Vec<Goal>,
|
|
||||||
},
|
|
||||||
}
|
|
||||||
|
|
||||||
/// A response from the activation program
|
/// A response from the activation program
|
||||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||||
pub enum Response {
|
pub enum Response {
|
||||||
Capabilities(CapabilityResponse),
|
ConnectionFailed { error: String },
|
||||||
|
ConnectionSucceded { supported_goals: Vec<String> },
|
||||||
Progress { phase: String },
|
Progress { phase: String },
|
||||||
NewStorePath { store_path: StorePath },
|
NewStorePath { store_path: StorePath },
|
||||||
Failed { error: String },
|
Failed { error: String },
|
||||||
|
Unsupported,
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Debug)]
|
#[derive(Debug)]
|
||||||
pub struct OpenedCommand {
|
pub struct ActivationCommand {
|
||||||
command: Child,
|
command: Child,
|
||||||
stdin: BufWriter<ChildStdin>,
|
stdin: BufWriter<ChildStdin>,
|
||||||
stdout: BufReader<ChildStdout>,
|
stdout: BufReader<ChildStdout>,
|
||||||
|
@ -100,8 +91,8 @@ pub struct OpenedCommand {
|
||||||
|
|
||||||
#[derive(Debug)]
|
#[derive(Debug)]
|
||||||
pub struct GenericHost {
|
pub struct GenericHost {
|
||||||
activation_program: OpenedCommand,
|
activation_program: ActivationCommand,
|
||||||
connection_options: HashMap<String, String>,
|
connection_uri: String,
|
||||||
}
|
}
|
||||||
|
|
||||||
#[async_trait]
|
#[async_trait]
|
||||||
|
@ -192,10 +183,7 @@ impl Host for GenericHost {
|
||||||
}
|
}
|
||||||
|
|
||||||
impl GenericHost {
|
impl GenericHost {
|
||||||
pub fn new(
|
pub fn new(system: &SystemTypeConfig, connection_uri: String) -> ColmenaResult<GenericHost> {
|
||||||
system: &SystemTypeConfig,
|
|
||||||
connection_options: &HashMap<String, String>,
|
|
||||||
) -> ColmenaResult<GenericHost> {
|
|
||||||
let mut command = Command::new(system.activation_program.as_path())
|
let mut command = Command::new(system.activation_program.as_path())
|
||||||
.stdin(Stdio::piped())
|
.stdin(Stdio::piped())
|
||||||
.stdout(Stdio::piped())
|
.stdout(Stdio::piped())
|
||||||
|
@ -205,7 +193,7 @@ impl GenericHost {
|
||||||
let stdin = BufWriter::new(command.stdin.take().unwrap());
|
let stdin = BufWriter::new(command.stdin.take().unwrap());
|
||||||
let stdout = BufReader::new(command.stdout.take().unwrap());
|
let stdout = BufReader::new(command.stdout.take().unwrap());
|
||||||
|
|
||||||
let activation_program = OpenedCommand {
|
let activation_program = ActivationCommand {
|
||||||
command,
|
command,
|
||||||
stdin,
|
stdin,
|
||||||
stdout,
|
stdout,
|
||||||
|
@ -213,13 +201,13 @@ impl GenericHost {
|
||||||
|
|
||||||
Ok(Self {
|
Ok(Self {
|
||||||
activation_program,
|
activation_program,
|
||||||
connection_options: connection_options.clone(),
|
connection_uri,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
pub async fn connect(&mut self) -> ColmenaResult<()> {
|
pub async fn connect(&mut self) -> ColmenaResult<()> {
|
||||||
self.call_default_handler(Request::Connection {
|
self.call_default_handler(Request::Connection {
|
||||||
connection_options: self.connection_options.clone(),
|
connection_uri: self.connection_uri.clone(),
|
||||||
})
|
})
|
||||||
.await
|
.await
|
||||||
}
|
}
|
||||||
|
@ -248,28 +236,30 @@ impl GenericHost {
|
||||||
let mut line = String::new();
|
let mut line = String::new();
|
||||||
let mut value = initial_value;
|
let mut value = initial_value;
|
||||||
|
|
||||||
|
// We're reading JSONL, so we can read all the line and parse it
|
||||||
while stdout.read_line(&mut line).await.is_ok_and(|x| x != 0) {
|
while stdout.read_line(&mut line).await.is_ok_and(|x| x != 0) {
|
||||||
if line == "\n" {
|
if line == "\n" {
|
||||||
|
log::trace!("finished receiving responses from activation program");
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
log::trace!("receiving from activation program:\n{}", line);
|
log::trace!("receiving from activation program:\n{}", line);
|
||||||
let response: Response = serde_json::from_str(line.as_str())
|
let response: Response = serde_json::from_str(line.as_str())
|
||||||
.map_err(|error| ColmenaError::InvalidActivationProgramJson { error })?;
|
.map_err(|error| ColmenaError::InvalidActivationProgramJson { error })?;
|
||||||
value = handler(response, value);
|
|
||||||
|
match response {
|
||||||
|
Response::Progress { phase } => {
|
||||||
|
log::info!("{phase}");
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
Response::Unsupported => return Err(ColmenaError::Unsupported),
|
||||||
|
_ => value = handler(response, value),
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
Ok(value)
|
Ok(value)
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn call_default_handler(&mut self, request: Request) -> ColmenaResult<()> {
|
async fn call_default_handler(&mut self, request: Request) -> ColmenaResult<()> {
|
||||||
self.call(
|
self.call(request, |_, _| {}, ()).await
|
||||||
request,
|
|
||||||
|response, _| match response {
|
|
||||||
Response::Progress { phase } => println!("{phase}"),
|
|
||||||
_ => (),
|
|
||||||
},
|
|
||||||
(),
|
|
||||||
)
|
|
||||||
.await
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -67,8 +67,8 @@ pub struct NodeConfig {
|
||||||
#[serde(rename = "targetPort")]
|
#[serde(rename = "targetPort")]
|
||||||
target_port: Option<u16>,
|
target_port: Option<u16>,
|
||||||
|
|
||||||
#[serde(rename = "connectionOptions")]
|
#[serde(rename = "connectionUri")]
|
||||||
connection_options: HashMap<String, String>,
|
connection_uri: String,
|
||||||
|
|
||||||
#[serde(rename = "allowLocalDeployment")]
|
#[serde(rename = "allowLocalDeployment")]
|
||||||
allow_local_deployment: bool,
|
allow_local_deployment: bool,
|
||||||
|
@ -107,8 +107,6 @@ pub struct SystemTypeConfig {
|
||||||
|
|
||||||
#[serde(rename = "activationProgram")]
|
#[serde(rename = "activationProgram")]
|
||||||
pub activation_program: StorePath,
|
pub activation_program: StorePath,
|
||||||
|
|
||||||
pub protocol: String,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Debug, Clone, Validate, Deserialize)]
|
#[derive(Debug, Clone, Validate, Deserialize)]
|
||||||
|
@ -218,7 +216,7 @@ impl NodeConfig {
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn to_generic_host(&self, system_config: &SystemTypeConfig) -> ColmenaResult<GenericHost> {
|
pub fn to_generic_host(&self, system_config: &SystemTypeConfig) -> ColmenaResult<GenericHost> {
|
||||||
GenericHost::new(system_config, &self.connection_options)
|
GenericHost::new(system_config, self.connection_uri.clone())
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
Loading…
Reference in a new issue