diff --git a/Cargo.lock b/Cargo.lock index 136f2ee..db459bb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -297,6 +297,31 @@ dependencies = [ "libc", ] +[[package]] +name = "crossbeam-deque" +version = "0.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9dd111b7b7f7d55b72c0a6ae361660ee5853c9af73f70c3c2ef6858b950e2e51" +dependencies = [ + "crossbeam-epoch", + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-epoch" +version = "0.9.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5b82ac4a3c2ca9c3460964f020e1402edd5753411d7737aa39c3714ad1b5420e" +dependencies = [ + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-utils" +version = "0.8.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d0a5c400df2834b80a4c3327b3aad3a4c4cd4de0629063962b03235697506a28" + [[package]] name = "crypto-common" version = "0.1.6" @@ -858,6 +883,7 @@ dependencies = [ "bitmaps", "rand_core 0.6.4", "rand_xoshiro", + "rayon", "serde", "sized-chunks", "typenum", @@ -1501,6 +1527,26 @@ dependencies = [ "rand_core 0.6.4", ] +[[package]] +name = "rayon" +version = "1.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "368f01d005bf8fd9b1206fb6fa653e6c4a81ceb1466406b81792d87c5677a58f" +dependencies = [ + "either", + "rayon-core", +] + +[[package]] +name = "rayon-core" +version = "1.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "22e18b0f0062d30d4230b2e85ff77fdfe4326feb054b9783a3460d8435c8ab91" +dependencies = [ + "crossbeam-deque", + "crossbeam-utils", +] + [[package]] name = "rdrand" version = "0.4.0" diff --git a/Cargo.toml b/Cargo.toml index 03c28f6..6c96925 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -24,7 +24,7 @@ tokio = { version = "1.48.0", features = ["full"] } tracing = { version = "0.1.41", features = ["release_max_level_debug"] } tracing-log = "0.2.0" tracing-subscriber = "0.3.20" -im = { version = "15.1.0", features = ["serde"] } +im = { version = "15.1.0", features = ["serde", "rayon"] } anyhow = "1.0.100" prost = "0.14.1" nix = { version = "0.30.1", features = ["user", "poll", "term"] } diff --git a/runtime/evaluate.nix b/runtime/evaluate.nix index 183dd26..eec4224 100644 --- a/runtime/evaluate.nix +++ b/runtime/evaluate.nix @@ -8,19 +8,20 @@ let module = import ./module; - mergedHive = { - meta = { }; + mergedHive = + { + meta = { }; - defaults = { }; - } - // hive - # Map nixosConfigurations into nodes - // (builtins.mapAttrs (name: value: { - imports = - value._module.args.modules - # Include any custom stuff within `colmena` - ++ [ hive.${name} or { } ]; - }) nixosConfigurations); + defaults = { }; + } + // hive + # Map nixosConfigurations into nodes + // (builtins.mapAttrs (name: value: { + imports = + value._module.args.modules + # Include any custom stuff within `colmena` + ++ [ hive.${name} or { } ]; + }) nixosConfigurations); nodeNames = builtins.filter ( name: @@ -55,20 +56,20 @@ let evalConfig = import (resolvedNixpkgs.path + "/nixos/lib/eval-config.nix"); in evalConfig { - modules = [ - module + modules = + [ + module - mergedHive.defaults - mergedHive.${name} - ] - ++ (resolvedNixpkgs.lib.optional isFlake { - config.nixpkgs.flake.source = resolvedNixpkgs.lib.mkDefault resolvedNixpkgs.path; - }); + mergedHive.defaults + mergedHive.${name} + ] + ++ (resolvedNixpkgs.lib.optional isFlake { + config.nixpkgs.flake.source = resolvedNixpkgs.lib.mkDefault resolvedNixpkgs.path; + }); system = null; specialArgs = { inherit name nodes; - } - // mergedHive.meta.specialArgs or { }; + } // mergedHive.meta.specialArgs or { }; }; nodes = builtins.listToAttrs ( map (name: { @@ -84,8 +85,10 @@ rec { topLevels = builtins.mapAttrs (name: _: getTopLevel name) nodes; inspect = { - _schema = 1; + _schema = 2; nodes = builtins.mapAttrs (_: v: v.config.deployment) nodes; }; + + shallow = builtins.mapAttrs (_: v: v.config.deployment.tags) nodes; } diff --git a/wire/cli/src/apply.rs b/wire/cli/src/apply.rs index 90477a2..d3c9df5 100644 --- a/wire/cli/src/apply.rs +++ b/wire/cli/src/apply.rs @@ -1,10 +1,10 @@ // SPDX-License-Identifier: AGPL-3.0-or-later // Copyright 2024-2025 wire Contributors -use futures::{FutureExt, StreamExt}; +use futures::{FutureExt, StreamExt, TryFutureExt}; use itertools::{Either, Itertools}; use lib::hive::node::{Context, GoalExecutor, Name, StepState, should_apply_locally}; -use lib::hive::{Hive, HiveLocation}; +use lib::hive::{Hive, HiveLocation, ShallowHive}; use lib::{SubCommandModifiers, errors::HiveLibError}; use miette::{Diagnostic, IntoDiagnostic, Result}; use std::collections::HashSet; @@ -49,7 +49,7 @@ fn read_apply_targets_from_stdin() -> Result<(Vec, Vec)> { // #[instrument(skip_all, fields(goal = %args.goal, on = %args.on.iter().join(", ")))] pub async fn apply( - hive: &mut Hive, + shallow: ShallowHive, location: HiveLocation, args: ApplyArgs, mut modifiers: SubCommandModifiers, @@ -58,7 +58,8 @@ pub async fn apply( let location = Arc::new(location); // Respect user's --always-build-local arg - hive.force_always_local(args.always_build_local)?; + // TODO + // hive.force_always_local(args.always_build_local)?; let header_span_enter = header_span.enter(); @@ -85,34 +86,39 @@ pub async fn apply( }, ); - let mut set = hive - .nodes - .iter_mut() - .filter(|(name, node)| { - args.on.is_empty() - || names.contains(name) - || node.tags.iter().any(|tag| tags.contains(tag)) + // Filters nodes by name & tag, then creates a pipeline that evaluates the node + // and executes it. + let mut set = shallow + .iter() + .filter(|(name, node_tags)| { + args.on.is_empty() || names.contains(name) || node_tags.iter().any(|tag| tags.contains(tag)) }) + .map(|(name, _)| (name, Hive::node_from_path(name, &location, modifiers))) .map(|(name, node)| { - info!("Resolved {:?} to include {}", args.on, name); - - let should_apply_locally = should_apply_locally(node.allow_local_deployment, &name.0); - - let context = Context { - node, - name, - goal: args.goal.clone().try_into().unwrap(), - state: StepState::default(), - no_keys: args.no_keys, - hive_location: location.clone(), - modifiers, - reboot: args.reboot, - should_apply_locally, - }; - - GoalExecutor::new(context) - .execute() - .map(move |result| (name, result)) + node.and_then(async |mut node| { + let name = name.clone(); + info!("Resolved {:?} to include {}", args.on, name); + + let should_apply_locally = + should_apply_locally(node.allow_local_deployment, &name.0); + + let context = Context { + node: &mut node, + name: &name, + goal: args.goal.clone().try_into().unwrap(), + state: StepState::default(), + no_keys: args.no_keys, + hive_location: location.clone(), + modifiers, + reboot: args.reboot, + should_apply_locally, + }; + + GoalExecutor::new(context) + .execute() + .await + }) + .map(move |result| (name, result)) }) .peekable(); diff --git a/wire/cli/src/main.rs b/wire/cli/src/main.rs index 7b51590..4a7d2ec 100644 --- a/wire/cli/src/main.rs +++ b/wire/cli/src/main.rs @@ -51,8 +51,8 @@ async fn main() -> Result<()> { match args.command { cli::Commands::Apply(apply_args) => { - let mut hive = Hive::new_from_path(&location, modifiers).await?; - apply::apply(&mut hive, location, apply_args, modifiers).await?; + let shallow = Hive::shallow_from_path(&location, modifiers).await?; + apply::apply(shallow, location, apply_args, modifiers).await?; } cli::Commands::Inspect { online: _, json } => println!("{}", { let hive = Hive::new_from_path(&location, modifiers).await?; diff --git a/wire/lib/src/commands/common.rs b/wire/lib/src/commands/common.rs index 1c6dc71..f22e724 100644 --- a/wire/lib/src/commands/common.rs +++ b/wire/lib/src/commands/common.rs @@ -66,7 +66,9 @@ pub async fn evaluate_hive_attribute( "{uri}#wire --apply \"hive: {}\"", match goal { EvalGoal::Inspect => "hive.inspect".to_string(), + EvalGoal::Shallow => "hive.shallow".to_string(), EvalGoal::GetTopLevel(node) => format!("hive.topLevels.{node}"), + EvalGoal::InspectNode(node) => format!("hive.inspect.nodes.{node}"), } ) } @@ -76,7 +78,9 @@ pub async fn evaluate_hive_attribute( &path.to_string_lossy(), match goal { EvalGoal::Inspect => "inspect".to_string(), + EvalGoal::Shallow => "topLevels --apply \"builtins.attrNames\"".to_string(), EvalGoal::GetTopLevel(node) => format!("topLevels.{node}"), + EvalGoal::InspectNode(node) => format!("hive.inspect.nodes.{node}"), } ) } diff --git a/wire/lib/src/hive/mod.rs b/wire/lib/src/hive/mod.rs index ae639cc..e0544e0 100644 --- a/wire/lib/src/hive/mod.rs +++ b/wire/lib/src/hive/mod.rs @@ -32,6 +32,8 @@ pub struct Hive { pub schema: u32, } +pub type ShallowHive = HashMap>; + pub enum Action<'a> { Inspect, EvaluateNode(OccupiedEntry<'a, String, Node>), @@ -48,7 +50,7 @@ fn check_schema_version<'de, D: Deserializer<'de>>(d: D) -> Result Result { + info!("evaluating hive {location:?} names"); + + let output = evaluate_hive_attribute(location, &EvalGoal::Shallow, modifiers).await?; + + let names: ShallowHive = serde_json::from_str(&output).map_err(|err| { + HiveLibError::HiveInitializationError(HiveInitializationError::ParseEvaluateError(err)) + })?; + + Ok(names) + } + + #[instrument(skip_all, name = "eval_names")] + pub async fn node_from_path<'a>( + name: &'a Name, + location: &'a HiveLocation, + modifiers: SubCommandModifiers, + ) -> Result { + info!("evaluating {name:?} from {location:?}"); + + let output = + evaluate_hive_attribute(location, &EvalGoal::InspectNode(name), modifiers).await?; + + let names: Node = serde_json::from_str(&output).map_err(|err| { + HiveLibError::HiveInitializationError(HiveInitializationError::ParseEvaluateError(err)) + })?; + + Ok(names) + } + /// # Errors /// /// Returns an error if a node in nodes does not exist in the hive. diff --git a/wire/lib/src/lib.rs b/wire/lib/src/lib.rs index 9e2e6e0..9ee5836 100644 --- a/wire/lib/src/lib.rs +++ b/wire/lib/src/lib.rs @@ -50,6 +50,8 @@ impl Default for SubCommandModifiers { pub enum EvalGoal<'a> { Inspect, + Shallow, + InspectNode(&'a Name), GetTopLevel(&'a Name), }