WIP: Generic activation program #9

Draft
ecoppens wants to merge 6 commits from ecoppens/colmena:main into main
11 changed files with 175 additions and 110 deletions
Showing only changes of commit 71200e5c17 - Show all commits

View file

@ -55,6 +55,9 @@ pub enum ColmenaError {
#[snafu(display("Unexpected active profile: {:?}", profile))] #[snafu(display("Unexpected active profile: {:?}", profile))]
ActiveProfileUnexpected { profile: Profile }, ActiveProfileUnexpected { profile: Profile },
#[snafu(display("Invalid JSON from activation program: {}", error))]
InvalidActivationProgramJson { error: serde_json::Error },

UnexpectedActivationProgramReply IMHO

`UnexpectedActivationProgramReply` IMHO
#[snafu(display("Could not determine current profile"))] #[snafu(display("Could not determine current profile"))]
FailedToGetCurrentProfile, FailedToGetCurrentProfile,

View file

@ -28,7 +28,7 @@ use super::{
host::Local as LocalHost, host::Local as LocalHost,
key::{Key, UploadAt as UploadKeyAt}, key::{Key, UploadAt as UploadKeyAt},
ColmenaError, ColmenaResult, CopyDirection, CopyOptions, Hive, Host, NodeConfig, NodeName, ColmenaError, ColmenaResult, CopyDirection, CopyOptions, Hive, Host, NodeConfig, NodeName,
Profile, ProfileDerivation, RebootOptions, StorePath, Profile, ProfileDerivation, RebootOptions,
}; };
/// A deployment. /// A deployment.
@ -637,18 +637,7 @@ impl Deployment {
} }
} }
let activation_program: &StorePath = host.activate(&profile_r, arc_self.goal).await?;
&arc_self.hive
.get_registry_config()
.await?
.systems
.get(target.config.system_type.as_ref().unwrap_or(&"nixos".to_owned()))
.expect(&format!("System type {} does not exists",
target.config.system_type.as_ref().unwrap_or(&"nixos".to_owned())
))
.activation_program;
host.activate(&profile_r, arc_self.goal, activation_program).await?;
job.success_with_message(arc_self.goal.success_str().to_string())?; job.success_with_message(arc_self.goal.success_str().to_string())?;

View file

@ -197,7 +197,7 @@ let
]; ];
serializableSystemTypeConfigKeys = [ serializableSystemTypeConfigKeys = [
"supportsDeployment" "supportsDeployment" "activationProgram" "protocol"
]; ];
in rec { in rec {

View file

@ -274,24 +274,37 @@ impl Hive {
for node in selected_nodes.into_iter() { for node in selected_nodes.into_iter() {
let config = node_configs.remove(&node).unwrap(); let config = node_configs.remove(&node).unwrap();
let host = config.to_ssh_host().map(|mut host| { if let Some(system_type) = config.system_type.as_ref() {
Review

Note: do not forget to check also if there's an activationProgram in the registry and if not, fallback to the legacy SSH mode.

Note: do not forget to check also if there's an activationProgram in the registry and if not, fallback to the legacy SSH mode.
n_ssh += 1; log::debug!(
"Using generic host (system_type: {}) for node {}",
if let Some(ssh_config) = &ssh_config { system_type,
host.set_ssh_config(ssh_config.clone()); node.0
} );
let system_config = registry.systems.get(system_type).unwrap();
if self.is_flake() { let generic_host = config.to_generic_host(system_config);
host.set_use_nix3_copy(true); let target = TargetNode::new(node.clone(), Some(Box::new(generic_host)), config);
}
host.upcast()
});
let ssh_host = host.is_some();
let target = TargetNode::new(node.clone(), host, config);
if !ssh_only || ssh_host {
targets.insert(node, target); targets.insert(node, target);
} else {
log::debug!("Using SSH host for node {}", node.0);
let host = config.to_ssh_host().map(|mut host| {
n_ssh += 1;
if let Some(ssh_config) = &ssh_config {
host.set_ssh_config(ssh_config.clone());
}
if self.is_flake() {
host.set_use_nix3_copy(true);
}
host.upcast()
});
let ssh_host = host.is_some();
let target = TargetNode::new(node.clone(), host, config);
if !ssh_only || ssh_host {
targets.insert(node, target);
}
} }
} }

View file

@ -264,6 +264,15 @@ 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

View file

@ -6,35 +6,31 @@ use serde::{Deserialize, Serialize};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, BufWriter}; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, BufWriter};
use tokio::process::Command; use tokio::process::Command;
use super::{CopyDirection, CopyOptions, Host}; use super::{CopyDirection, CopyOptions, Host, RebootOptions};
use crate::error::{ColmenaError, ColmenaResult}; use crate::error::{ColmenaError, ColmenaResult};
use crate::job::JobHandle; use crate::job::JobHandle;
use crate::nix::{self, Profile, StorePath}; use crate::nix::{self, Key, Profile, StorePath, SystemTypeConfig};
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub type TransportId = String;
pub struct TransportId(String);
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct Transport { pub struct Transport {
id: TransportId, id: TransportId,
name: String,
long_name: String, long_name: String,
description: String, description: String,
} }
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub type GoalId = String;
pub struct GoalId(String);
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct Goal { pub struct Goal {
id: GoalId, id: GoalId,
name: String,
long_name: String, long_name: String,
description: String, description: String,
} }
/// A request to the activation program /// A request to the activation program
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] #[derive(Clone, Debug, Serialize, Deserialize)]
pub enum Request { pub enum Request {
/// Asks for activation program's capabilities /// Asks for activation program's capabilities
Capabilities, Capabilities,
@ -44,7 +40,7 @@ pub enum Request {
transport: TransportId, transport: TransportId,
path: StorePath, path: StorePath,
to_host: bool, to_host: bool,
options: HashMap<String, String>, options: CopyOptions,
}, },
/// Deploys the profile to host /// Deploys the profile to host
@ -52,18 +48,32 @@ pub enum Request {
transport: TransportId, transport: TransportId,
goal: GoalId, goal: GoalId,
toplevel: StorePath, toplevel: StorePath,
options: HashMap<String, String>, options: CopyOptions,
}, },
/// Realizes the derivation /// Realizes the derivation
Realize { Realize {
transport: TransportId, transport: TransportId,
path: StorePath, path: StorePath,
options: HashMap<String, String>,
}, },
/// Uploads keys to host /// Uploads keys to host
UploadKeys { transport: TransportId }, UploadKeys {
transport: TransportId,
keys: HashMap<String, Key>,
require_ownership: bool,
},
Activate {
transport: TransportId,
profile: StorePath,
goal: GoalId,
},
Reboot {
transport: TransportId,
wait_for_boot: bool,
},
} }
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
@ -96,47 +106,61 @@ impl Host for GenericHost {
direction: CopyDirection, direction: CopyDirection,
options: CopyOptions, options: CopyOptions,
) -> ColmenaResult<()> { ) -> ColmenaResult<()> {
self.call(Request::CopyClosure { self.call_default_handler(Request::CopyClosure {
transport: self.transport.clone(), transport: self.transport.clone(),
path: closure.clone(), path: closure.clone(),
to_host: direction == CopyDirection::ToRemote, to_host: direction == CopyDirection::ToRemote,
options: HashMap::from([( options,
"include_outputs".to_owned(),
if options.include_outputs {
"true".to_owned()
} else {
"false".to_owned()
},
)]),
}) })
.await .await
} }
async fn realize_remote(&mut self, derivation: &StorePath) -> ColmenaResult<Vec<StorePath>> { async fn realize_remote(&mut self, derivation: &StorePath) -> ColmenaResult<Vec<StorePath>> {
self.call(Request::Realize { Ok(self
transport: self.transport.clone(), .call(
path: derivation.clone(), Request::Realize {
options: HashMap::new(), transport: self.transport.clone(),
}) path: derivation.clone(),
.await?; },
move |response, mut store_paths| {
Ok(Vec::new()) match response {
Response::Progress { phase } => println!("{}", phase),
Response::NewStorePath { store_path } => store_paths.push(store_path),
_ => (),
};
store_paths
},
Vec::new(),
)
.await?)
} }
fn set_job(&mut self, bar: Option<JobHandle>) {} fn set_job(&mut self, _: Option<JobHandle>) {}
async fn deploy( async fn deploy(
&mut self, &mut self,
profile: &Profile, profile: &Profile,
goal: nix::Goal, goal: nix::Goal,
copy_options: CopyOptions, copy_options: CopyOptions,
activation_program: Option<&StorePath>,
) -> ColmenaResult<()> { ) -> ColmenaResult<()> {
self.call(Request::Deploy { self.call_default_handler(Request::Deploy {
transport: self.transport.clone(), transport: self.transport.clone(),
goal: GoalId(goal.to_string()), goal: goal.to_string(),
toplevel: profile.as_store_path().clone(), toplevel: profile.as_store_path().clone(),
options: HashMap::new(), options: copy_options,
})
.await
}
async fn upload_keys(
&mut self,
keys: &HashMap<String, Key>,
require_ownership: bool,
) -> ColmenaResult<()> {
self.call_default_handler(Request::UploadKeys {
transport: self.transport.clone(),
keys: keys.clone(),
require_ownership,
}) })
.await .await
} }
@ -148,17 +172,50 @@ impl Host for GenericHost {
async fn get_main_system_profile(&mut self) -> ColmenaResult<Profile> { async fn get_main_system_profile(&mut self) -> ColmenaResult<Profile> {
Err(ColmenaError::Unsupported) Err(ColmenaError::Unsupported)
} }
async fn activate(&mut self, profile: &Profile, goal: nix::Goal) -> ColmenaResult<()> {
self.call_default_handler(Request::Activate {
transport: self.transport.clone(),
profile: profile.as_store_path().clone(),
goal: goal.to_string(),
})
.await
}
async fn reboot(&mut self, options: RebootOptions) -> ColmenaResult<()> {
self.call_default_handler(Request::Reboot {
transport: self.transport.clone(),
wait_for_boot: options.wait_for_boot,
})
.await
}
} }
impl GenericHost { impl GenericHost {
async fn call(&mut self, request: Request) -> ColmenaResult<()> { pub fn new(system: &SystemTypeConfig) -> GenericHost {
Self {
activation_program: system.activation_program.clone(),
transport: system.protocol.clone(),
}
}
async fn call<F, T>(
&mut self,
request: Request,
handler: F,
initial_value: T,
) -> ColmenaResult<T>
where
F: Fn(Response, T) -> T,
{
let mut command = Command::new(self.activation_program.as_path()) let mut command = Command::new(self.activation_program.as_path())
.stdin(Stdio::piped()) .stdin(Stdio::piped())
.stdout(Stdio::piped()) .stdout(Stdio::piped())
.stderr(Stdio::piped()) .stderr(Stdio::piped())
.spawn()?; .spawn()?;
let json = serde_json::to_string(&request).unwrap(); let json = serde_json::to_string(&request)
.map_err(|error| ColmenaError::InvalidActivationProgramJson { error })?;
let mut stdin = BufWriter::new(command.stdin.take().unwrap()); let mut stdin = BufWriter::new(command.stdin.take().unwrap());
let mut stdout = BufReader::new(command.stdout.take().unwrap()).lines(); let mut stdout = BufReader::new(command.stdout.take().unwrap()).lines();
@ -172,11 +229,26 @@ impl GenericHost {
.expect("child process encountered an error"); .expect("child process encountered an error");
}); });
let mut value = initial_value;
while let Some(line) = stdout.next_line().await? { while let Some(line) = stdout.next_line().await? {
let response: Response = serde_json::from_str(line.as_str()).unwrap(); let response: Response = serde_json::from_str(line.as_str())
println!("{:?}", response); .map_err(|error| ColmenaError::InvalidActivationProgramJson { error })?;
value = handler(response, value);
} }
Ok(()) Ok(value)
}
async fn call_default_handler(&mut self, request: Request) -> ColmenaResult<()> {
self.call(
request,
|response, _| match response {
Response::Progress { phase } => println!("{phase}"),
_ => (),
},
(),
)
.await
} }
} }

View file

@ -78,12 +78,7 @@ impl Host for Local {
Ok(()) Ok(())
} }
async fn activate( async fn activate(&mut self, profile: &Profile, goal: Goal) -> ColmenaResult<()> {
&mut self,
profile: &Profile,
goal: Goal,
activation_program: &StorePath,
) -> ColmenaResult<()> {
if !goal.requires_activation() { if !goal.requires_activation() {
return Err(ColmenaError::Unsupported); return Err(ColmenaError::Unsupported);
} }
@ -96,9 +91,7 @@ impl Host for Local {
} }
let command = { let command = {
let activation_command = profile let activation_command = profile.activation_command(goal).unwrap();
.activation_command(goal, activation_program)
.unwrap();
self.make_privileged_command(&activation_command) self.make_privileged_command(&activation_command)
}; };

View file

@ -1,6 +1,7 @@
use std::collections::HashMap; use std::collections::HashMap;
use async_trait::async_trait; use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use super::{Goal, Key, Profile, StorePath}; use super::{Goal, Key, Profile, StorePath};
use crate::error::{ColmenaError, ColmenaResult}; use crate::error::{ColmenaError, ColmenaResult};
@ -15,14 +16,15 @@ pub use local::Local;
mod key_uploader; mod key_uploader;
mod generic; mod generic;
pub use generic::GenericHost;
#[derive(Copy, Clone, Debug, PartialEq, Eq)] #[derive(Copy, Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub enum CopyDirection { pub enum CopyDirection {
ToRemote, ToRemote,
FromRemote, FromRemote,
} }
#[derive(Copy, Clone, Debug)] #[derive(Copy, Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct CopyOptions { pub struct CopyOptions {
include_outputs: bool, include_outputs: bool,
use_substitutes: bool, use_substitutes: bool,
@ -130,7 +132,6 @@ pub trait Host: Send + Sync + std::fmt::Debug {
profile: &Profile, profile: &Profile,
goal: Goal, goal: Goal,
copy_options: CopyOptions, copy_options: CopyOptions,
activation_program: Option<&StorePath>,
) -> ColmenaResult<()> { ) -> ColmenaResult<()> {
self.copy_closure( self.copy_closure(
profile.as_store_path(), profile.as_store_path(),
@ -140,12 +141,7 @@ pub trait Host: Send + Sync + std::fmt::Debug {
.await?; .await?;
if goal.requires_activation() { if goal.requires_activation() {
self.activate( self.activate(profile, goal).await?;
profile,
goal,
activation_program.expect("Unknown activation program"),
)
.await?;
} }
Ok(()) Ok(())
@ -179,12 +175,7 @@ pub trait Host: Send + Sync + std::fmt::Debug {
/// ///
/// The profile must already exist on the host. You should probably use deploy instead. /// The profile must already exist on the host. You should probably use deploy instead.
#[allow(unused_variables)] #[allow(unused_variables)]
async fn activate( async fn activate(&mut self, profile: &Profile, goal: Goal) -> ColmenaResult<()> {
&mut self,
profile: &Profile,
goal: Goal,
activation_program: &StorePath,
) -> ColmenaResult<()> {
Err(ColmenaError::Unsupported) Err(ColmenaError::Unsupported)
} }

View file

@ -90,12 +90,7 @@ impl Host for Ssh {
Ok(()) Ok(())
} }
async fn activate( async fn activate(&mut self, profile: &Profile, goal: Goal) -> ColmenaResult<()> {
&mut self,
profile: &Profile,
goal: Goal,
activation_program: &StorePath,
) -> ColmenaResult<()> {
if !goal.requires_activation() { if !goal.requires_activation() {
return Err(ColmenaError::Unsupported); return Err(ColmenaError::Unsupported);
} }
@ -106,9 +101,7 @@ impl Host for Ssh {
self.run_command(set_profile).await?; self.run_command(set_profile).await?;
} }
let activation_command = profile let activation_command = profile.activation_command(goal).unwrap();
.activation_command(goal, activation_program)
.unwrap();
let v: Vec<&str> = activation_command.iter().map(|s| &**s).collect(); let v: Vec<&str> = activation_command.iter().map(|s| &**s).collect();
let command = self.ssh(&v); let command = self.ssh(&v);
self.run_command(command).await self.run_command(command).await

View file

@ -10,8 +10,8 @@ use validator::{Validate, ValidationError as ValidationErrorType};
use crate::error::{ColmenaError, ColmenaResult}; use crate::error::{ColmenaError, ColmenaResult};
pub mod host; pub mod host;
use host::Ssh;
pub use host::{CopyDirection, CopyOptions, Host, RebootOptions}; pub use host::{CopyDirection, CopyOptions, Host, RebootOptions};
use host::{GenericHost, Ssh};
pub mod hive; pub mod hive;
pub use hive::{Hive, HivePath}; pub use hive::{Hive, HivePath};
@ -104,6 +104,8 @@ 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)]
@ -211,6 +213,10 @@ impl NodeConfig {
host host
}) })
} }
pub fn to_generic_host(&self, system_config: &SystemTypeConfig) -> GenericHost {
GenericHost::new(system_config)
}
} }
impl NixFlags { impl NixFlags {

View file

@ -26,11 +26,7 @@ impl Profile {
} }
/// Returns the command to activate this profile. /// Returns the command to activate this profile.
pub fn activation_command( pub fn activation_command(&self, goal: Goal) -> Option<Vec<String>> {
&self,
goal: Goal,
activation_program: &StorePath,
) -> Option<Vec<String>> {
if let Some(goal) = goal.as_str() { if let Some(goal) = goal.as_str() {
let path = self.as_path().join("bin/switch-to-configuration"); let path = self.as_path().join("bin/switch-to-configuration");
let switch_to_configuration = path let switch_to_configuration = path