diff options
Diffstat (limited to 'src/commands')
| -rw-r--r-- | src/commands/build.rs | 6 | ||||
| -rw-r--r-- | src/commands/db.rs | 64 | ||||
| -rw-r--r-- | src/commands/endpoint.rs | 88 | ||||
| -rw-r--r-- | src/commands/release.rs | 14 | ||||
| -rw-r--r-- | src/commands/source.rs | 86 | ||||
| -rw-r--r-- | src/commands/tree_of.rs | 2 |
6 files changed, 177 insertions, 83 deletions
diff --git a/src/commands/build.rs b/src/commands/build.rs index ff2769d..db8d7a4 100644 --- a/src/commands/build.rs +++ b/src/commands/build.rs @@ -33,22 +33,22 @@ use tokio_stream::StreamExt; use uuid::Uuid; use crate::config::*; -use crate::filestore::path::StoreRoot; use crate::filestore::ReleaseStore; use crate::filestore::StagingStore; +use crate::filestore::path::StoreRoot; use crate::job::JobResource; use crate::log::LogItem; use crate::orchestrator::OrchestratorSetup; +use crate::package::Dag; use crate::package::PackageName; use crate::package::PackageVersion; use crate::package::Shebang; -use crate::package::Dag; use crate::repository::Repository; use crate::schema; use crate::source::SourceCache; +use crate::util::EnvironmentVariableName; use crate::util::docker::ImageName; use crate::util::progress::ProgressBars; -use crate::util::EnvironmentVariableName; /// Implementation of the "build" subcommand #[allow(clippy::too_many_arguments)] diff --git a/src/commands/db.rs b/src/commands/db.rs index e984848..672a0d6 100644 --- a/src/commands/db.rs +++ b/src/commands/db.rs @@ -33,11 +33,12 @@ use log::trace; use crate::commands::util::get_date_filter; use crate::config::Configuration; -use crate::db::models; use crate::db::DbConnectionConfig; +use crate::db::models; use crate::log::JobResult; use crate::package::Script; use crate::schema; +use crate::util::progress::ProgressBars; diesel_migrations::embed_migrations!("migrations"); @@ -46,19 +47,20 @@ pub fn db( db_connection_config: DbConnectionConfig<'_>, config: &Configuration, matches: &ArgMatches, + progressbars: ProgressBars, ) -> Result<()> { match matches.subcommand() { Some(("cli", matches)) => cli(db_connection_config, matches), - Some(("setup", _matches)) => setup(db_connection_config), - Some(("artifacts", matches)) => artifacts(db_connection_config, matches), - Some(("envvars", matches)) => envvars(db_connection_config, matches), - Some(("images", matches)) => images(db_connection_config, matches), - Some(("submit", matches)) => submit(db_connection_config, matches), - Some(("submits", matches)) => submits(db_connection_config, matches), - Some(("jobs", matches)) => jobs(db_connection_config, matches), - Some(("job", matches)) => job(db_connection_config, config, matches), - Some(("log-of", matches)) => log_of(db_connection_config, matches), - Some(("releases", matches)) => releases(db_connection_config, config, matches), + Some(("setup", _matches)) => setup(db_connection_config, progressbars), + Some(("artifacts", matches)) => artifacts(db_connection_config, matches, progressbars), + Some(("envvars", matches)) => envvars(db_connection_config, matches, progressbars), + Some(("images", matches)) => images(db_connection_config, matches, progressbars), + Some(("submit", matches)) => submit(db_connection_config, matches, progressbars), + Some(("submits", matches)) => submits(db_connection_config, matches, progressbars), + Some(("jobs", matches)) => jobs(db_connection_config, matches, progressbars), + Some(("job", matches)) => job(db_connection_config, config, matches, progressbars), + Some(("log-of", matches)) => log_of(db_connection_config, matches, progressbars), + Some(("releases", matches)) => releases(db_connection_config, config, matches, progressbars), Some((other, _)) => Err(anyhow!("Unknown subcommand: {}", other)), None => Err(anyhow!("No subcommand")), } @@ -148,18 +150,18 @@ fn cli(db_connection_config: DbConnectionConfig<'_>, matches: &ArgMatches) -> Re .run_for_uri(db_connection_config) } -fn setup(conn_cfg: DbConnectionConfig<'_>) -> Result<()> { - let conn = conn_cfg.establish_connection()?; +fn setup(conn_cfg: DbConnectionConfig<'_>, progressbars: ProgressBars) -> Result<()> { + let conn = conn_cfg.establish_connection(&progressbars)?; embedded_migrations::run_with_output(&conn, &mut std::io::stdout()).map_err(Error::from) } /// Implementation of the "db artifacts" subcommand -fn artifacts(conn_cfg: DbConnectionConfig<'_>, matches: &ArgMatches) -> Result<()> { +fn artifacts(conn_cfg: DbConnectionConfig<'_>, matches: &ArgMatches, progressbars: ProgressBars) -> Result<()> { use crate::schema::artifacts::dsl; let csv = matches.is_present("csv"); let hdrs = crate::commands::util::mk_header(vec!["Path", "Released", "Job"]); - let conn = conn_cfg.establish_connection()?; + let conn = conn_cfg.establish_connection(&progressbars)?; let data = matches .value_of("job_uuid") .map(uuid::Uuid::parse_str) @@ -203,12 +205,12 @@ fn artifacts(conn_cfg: DbConnectionConfig<'_>, matches: &ArgMatches) -> Result<( } /// Implementation of the "db envvars" subcommand -fn envvars(conn_cfg: DbConnectionConfig<'_>, matches: &ArgMatches) -> Result<()> { +fn envvars(conn_cfg: DbConnectionConfig<'_>, matches: &ArgMatches, progressbars: ProgressBars) -> Result<()> { use crate::schema::envvars::dsl; let csv = matches.is_present("csv"); let hdrs = crate::commands::util::mk_header(vec!["Name", "Value"]); - let conn = conn_cfg.establish_connection()?; + let conn = conn_cfg.establish_connection(&progressbars)?; let data = dsl::envvars .load::<models::EnvVar>(&conn)? .into_iter() @@ -225,12 +227,12 @@ fn envvars(conn_cfg: DbConnectionConfig<'_>, matches: &ArgMatches) -> Result<()> } /// Implementation of the "db images" subcommand -fn images(conn_cfg: DbConnectionConfig<'_>, matches: &ArgMatches) -> Result<()> { +fn images(conn_cfg: DbConnectionConfig<'_>, matches: &ArgMatches, progressbars: ProgressBars) -> Result<()> { use crate::schema::images::dsl; let csv = matches.is_present("csv"); let hdrs = crate::commands::util::mk_header(vec!["Name"]); - let conn = conn_cfg.establish_connection()?; + let conn = conn_cfg.establish_connection(&progressbars)?; let data = dsl::images .load::<models::Image>(&conn)? .into_iter() @@ -247,8 +249,8 @@ fn images(conn_cfg: DbConnectionConfig<'_>, matches: &ArgMatches) -> Result<()> } /// Implementation of the "db submit" subcommand -fn submit(conn_cfg: DbConnectionConfig<'_>, matches: &ArgMatches) -> Result<()> { - let conn = conn_cfg.establish_connection()?; +fn submit(conn_cfg: DbConnectionConfig<'_>, matches: &ArgMatches, progressbars: ProgressBars) -> Result<()> { + let conn = conn_cfg.establish_connection(&progressbars)?; let submit_id = matches.value_of("submit") .map(uuid::Uuid::from_str) .transpose() @@ -331,11 +333,11 @@ fn submit(conn_cfg: DbConnectionConfig<'_>, matches: &ArgMatches) -> Result<()> } /// Implementation of the "db submits" subcommand -fn submits(conn_cfg: DbConnectionConfig<'_>, matches: &ArgMatches) -> Result<()> { +fn submits(conn_cfg: DbConnectionConfig<'_>, matches: &ArgMatches, progressbars: ProgressBars) -> Result<()> { let csv = matches.is_present("csv"); let limit = matches.value_of("limit").map(i64::from_str).transpose()?; let hdrs = crate::commands::util::mk_header(vec!["Time", "UUID"]); - let conn = conn_cfg.establish_connection()?; + let conn = conn_cfg.establish_connection(&progressbars)?; let query = schema::submits::table .order_by(schema::submits::id.desc()); // required for the --limit implementation @@ -397,7 +399,7 @@ fn submits(conn_cfg: DbConnectionConfig<'_>, matches: &ArgMatches) -> Result<()> } /// Implementation of the "db jobs" subcommand -fn jobs(conn_cfg: DbConnectionConfig<'_>, matches: &ArgMatches) -> Result<()> { +fn jobs(conn_cfg: DbConnectionConfig<'_>, matches: &ArgMatches, progressbars: ProgressBars) -> Result<()> { let csv = matches.is_present("csv"); let hdrs = crate::commands::util::mk_header(vec![ "Submit", @@ -408,7 +410,7 @@ fn jobs(conn_cfg: DbConnectionConfig<'_>, matches: &ArgMatches) -> Result<()> { "Package", "Version", ]); - let conn = conn_cfg.establish_connection()?; + let conn = conn_cfg.establish_connection(&progressbars)?; let older_than_filter = get_date_filter("older_than", matches)?; let newer_than_filter = get_date_filter("newer_than", matches)?; @@ -495,14 +497,14 @@ fn jobs(conn_cfg: DbConnectionConfig<'_>, matches: &ArgMatches) -> Result<()> { } /// Implementation of the "db job" subcommand -fn job(conn_cfg: DbConnectionConfig<'_>, config: &Configuration, matches: &ArgMatches) -> Result<()> { +fn job(conn_cfg: DbConnectionConfig<'_>, config: &Configuration, matches: &ArgMatches, progressbars: ProgressBars) -> Result<()> { let script_highlight = !matches.is_present("no_script_highlight"); let script_line_numbers = !matches.is_present("no_script_line_numbers"); let configured_theme = config.script_highlight_theme(); let show_log = matches.is_present("show_log"); let show_script = matches.is_present("show_script"); let csv = matches.is_present("csv"); - let conn = conn_cfg.establish_connection()?; + let conn = conn_cfg.establish_connection(&progressbars)?; let job_uuid = matches .value_of("job_uuid") .map(uuid::Uuid::parse_str) @@ -669,8 +671,8 @@ fn job(conn_cfg: DbConnectionConfig<'_>, config: &Configuration, matches: &ArgMa } /// Implementation of the subcommand "db log-of" -fn log_of(conn_cfg: DbConnectionConfig<'_>, matches: &ArgMatches) -> Result<()> { - let conn = conn_cfg.establish_connection()?; +fn log_of(conn_cfg: DbConnectionConfig<'_>, matches: &ArgMatches, progressbars: ProgressBars) -> Result<()> { + let conn = conn_cfg.establish_connection(&progressbars)?; let job_uuid = matches .value_of("job_uuid") .map(uuid::Uuid::parse_str) @@ -692,9 +694,9 @@ fn log_of(conn_cfg: DbConnectionConfig<'_>, matches: &ArgMatches) -> Result<()> } /// Implementation of the "db releases" subcommand -fn releases(conn_cfg: DbConnectionConfig<'_>, config: &Configuration, matches: &ArgMatches) -> Result<()> { +fn releases(conn_cfg: DbConnectionConfig<'_>, config: &Configuration, matches: &ArgMatches, progressbars: ProgressBars) -> Result<()> { let csv = matches.is_present("csv"); - let conn = conn_cfg.establish_connection()?; + let conn = conn_cfg.establish_connection(&progressbars)?; let header = crate::commands::util::mk_header(["Package", "Version", "Date", "Path"].to_vec()); let mut query = schema::jobs::table .inner_join(schema::packages::table) diff --git a/src/commands/endpoint.rs b/src/commands/endpoint.rs index 6fd7d69..cd65887 100644 --- a/src/commands/endpoint.rs +++ b/src/commands/endpoint.rs @@ -11,6 +11,7 @@ //! Implementation of the 'endpoint' subcommand use std::collections::HashMap; +use std::io::Write; use std::ops::Deref; use std::str::FromStr; use std::sync::Arc; @@ -49,6 +50,7 @@ pub async fn endpoint(matches: &ArgMatches, config: &Configuration, progress_gen Some(("stats", matches)) => stats(endpoint_names, matches, config, progress_generator).await, Some(("container", matches)) => crate::commands::endpoint_container::container(endpoint_names, matches, config).await, Some(("containers", matches)) => containers(endpoint_names, matches, config).await, + Some(("images", matches)) => images(endpoint_names, matches, config).await, Some((other, _)) => Err(anyhow!("Unknown subcommand: {}", other)), None => Err(anyhow!("No subcommand")), } @@ -422,6 +424,92 @@ async fn containers_stop(endpoint_names: Vec<EndpointName>, } +async fn images(endpoint_names: Vec<EndpointName>, + matches: &ArgMatches, + config: &Configuration, +) -> Result<()> { + match matches.subcommand() { + Some(("list", matches)) => images_list(endpoint_names, matches, config).await, + Some(("verify-present", matches)) => images_present(endpoint_names, matches, config).await, + Some((other, _)) => Err(anyhow!("Unknown subcommand: {}", other)), + None => Err(anyhow!("No subcommand")), + } +} + +async fn images_list(endpoint_names: Vec<EndpointName>, + _matches: &ArgMatches, + config: &Configuration, +) -> Result<()> { + let mut iter = connect_to_endpoints(config, &endpoint_names) + .await? + .into_iter() + .map(move |ep| async move { ep.images(None).await }) + .collect::<futures::stream::FuturesUnordered<_>>() + .collect::<Result<Vec<_>>>() + .await? + .into_iter() + .flatten(); + + let out = std::io::stdout(); + let mut lock = out.lock(); + + iter.try_for_each(|img| { + writeln!(lock, "{created} {id}", created = img.created(), id = { + if let Some(tags)= img.tags() { + tags.join(", ") + } else { + img.id().clone() + } + }).map_err(Error::from) + }) +} + +async fn images_present(endpoint_names: Vec<EndpointName>, + _matches: &ArgMatches, + config: &Configuration, +) -> Result<()> { + use crate::util::docker::ImageName; + + let eps = connect_to_endpoints(config, &endpoint_names).await?; + + let ep_names_to_images = eps.iter() + .map(|ep| async move { + ep.images(None).await.map(|imgs| { + let img_tags = imgs.filter_map(|img| img.tags().clone().map(Vec::into_iter)) + .flatten() + .map(ImageName::from) + .collect(); + + (ep.name().clone(), img_tags) + }) + }) + .collect::<futures::stream::FuturesUnordered<_>>() + .collect::<Result<Vec<(EndpointName, Vec<ImageName>)>>>() + .await? + .into_iter() + .collect::<HashMap<EndpointName, Vec<ImageName>>>(); + + let out = std::io::stdout(); + let mut lock = out.lock(); + + ep_names_to_images + .iter() + .map(|(ep_name, ep_imgs)| { + config.docker() + .images() + .iter() + .map(|config_img| (ep_imgs.contains(config_img), config_img)) + .try_for_each(|(found, img_name)| { + if found { + writeln!(lock, "found {img} in {ep}", img = img_name, ep = ep_name).map_err(Error::from) + } else { + writeln!(lock, "{img} not found", img = img_name).map_err(Error::from) + } + }) + }) + .collect::<Result<()>>() +} + /// Helper function to connect to all endpoints from the configuration, that appear (by name) in /// the `endpoint_names` list pub(super) async fn connect_to_endpoints(config: &Configuration, endpoint_names: &[EndpointName]) -> Result<Vec<Arc<Endpoint>>> { diff --git a/src/commands/release.rs b/src/commands/release.rs index 6f0232b..12decff 100644 --- a/src/commands/release.rs +++ b/src/commands/release.rs @@ -22,18 +22,20 @@ use log::{debug, info, trace}; use tokio_stream::StreamExt; use crate::config::Configuration; -use crate::db::models as dbmodels; use crate::db::DbConnectionConfig; +use crate::db::models as dbmodels; +use crate::util::progress::ProgressBars; /// Implementation of the "release" subcommand pub async fn release( db_connection_config: DbConnectionConfig<'_>, config: &Configuration, matches: &ArgMatches, + progressbars: ProgressBars, ) -> Result<()> { match matches.subcommand() { - Some(("new", matches)) => new_release(db_connection_config, config, matches).await, - Some(("rm", matches)) => rm_release(db_connection_config, config, matches).await, + Some(("new", matches)) => new_release(db_connection_config, config, matches, progressbars).await, + Some(("rm", matches)) => rm_release(db_connection_config, config, matches, progressbars).await, Some((other, _matches)) => Err(anyhow!("Unknown subcommand: {}", other)), None => Err(anyhow!("Missing subcommand")), } @@ -44,6 +46,7 @@ async fn new_release( db_connection_config: DbConnectionConfig<'_>, config: &Configuration, matches: &ArgMatches, + progressbars: ProgressBars, ) -> Result<()> { let print_released_file_pathes = !matches.is_present("quiet"); let release_store_name = matches.value_of("release_store_name").unwrap(); // safe by clap @@ -60,7 +63,7 @@ async fn new_release( debug!("Release called for: {:?} {:?}", pname, pvers); - let conn = db_connection_config.establish_connection()?; + let conn = db_connection_config.establish_connection(&progressbars)?; let submit_uuid = matches .value_of("submit_uuid") .map(uuid::Uuid::parse_str) @@ -193,6 +196,7 @@ pub async fn rm_release( db_connection_config: DbConnectionConfig<'_>, config: &Configuration, matches: &ArgMatches, + progressbars: ProgressBars, ) -> Result<()> { let release_store_name = matches.value_of("release_store_name").map(String::from).unwrap(); // safe by clap if !(config.releases_directory().exists() && config.releases_directory().is_dir()) { @@ -209,7 +213,7 @@ pub async fn rm_release( let pvers = matches.value_of("package_version").map(String::from).unwrap(); // safe by clap debug!("Remove Release called for: {:?} {:?}", pname, pvers); - let conn = db_connection_config.establish_connection()?; + let conn = db_connection_config.establish_connection(&progressbars)?; let (release, artifact) = crate::schema::jobs::table .inner_join(crate::schema::packages::table) diff --git a/src/commands/source.rs b/src/commands/source.rs index 3e0167d..36cf352 100644 --- a/src/commands/source.rs +++ b/src/commands/source.rs @@ -214,6 +214,49 @@ pub async fn download( repo: Repository, progressbars: ProgressBars, ) -> Result<()> { + async fn perform_download(source: &SourceEntry, bar: &indicatif::ProgressBar) -> Result<()> { + trace!("Creating: {:?}", source); + let file = source.create().await.with_context(|| { + anyhow!( + "Creating source file destination: {}", + source.path().display() + ) + })?; + + let mut file = tokio::io::BufWriter::new(file); + let response = match reqwest::get(source.url().as_ref()).await { + Ok(resp) => resp, + Err(e) => { + bar.finish_with_message(format!("Failed: {}", source.url())); + return Err(e).with_context(|| anyhow!("Downloading '{}'", source.url())) + } + }; + + if let Some(len) = response.content_length() { + bar.set_length(len); + } + + let mut stream = reqwest::get(source.url().as_ref()).await?.bytes_stream(); + let mut bytes_written = 0; + while let Some(bytes) = stream.next().await { + let bytes = bytes?; + file.write_all(bytes.as_ref()).await?; + bytes_written += bytes.len(); + + bar.inc(bytes.len() as u64); + if let Some(len) = response.content_length() { + bar.set_message(format!("Downloading {} ({}/{} bytes)", source.url(), bytes_written, len)); + } else { + bar.set_message(format!("Downloading {} ({} bytes)", source.url(), bytes_written)); + } + } + + file.flush() + .await + .map_err(Error::from) + .map(|_| ()) + } + let force = matches.is_present("force"); let cache = PathBuf::from(config.source_cache_root()); let sc = SourceCache::new(cache); @@ -260,49 +303,6 @@ pub async fn download( if source_path_exists && !force { Err(anyhow!("Source exists: {}", source.path().display())) } else { - async fn perform_download(source: &SourceEntry, bar: &indicatif::ProgressBar) -> Result<()> { - trace!("Creating: {:?}", source); - let file = source.create().await.with_context(|| { - anyhow!( - "Creating source file destination: {}", - source.path().display() - ) - })?; - - let mut file = tokio::io::BufWriter::new(file); - let response = match reqwest::get(source.url().as_ref()).await { - Ok(resp) => resp, - Err(e) => { - bar.finish_with_message(format!("Failed: {}", source.url())); - return Err(e).with_context(|| anyhow!("Downloading '{}'", source.url())) - } - }; - - if let Some(len) = response.content_length() { - bar.set_length(len); - } - - let mut stream = reqwest::get(source.url().as_ref()).await?.bytes_stream(); - let mut bytes_written = 0; - while let Some(bytes) = stream.next().await { - let bytes = bytes?; - file.write_all(bytes.as_ref()).await?; - bytes_written += bytes.len(); - - bar.inc(bytes.len() as u64); - if let Some(len) = response.content_length() { - bar.set_message(format!("Downloading {} ({}/{} bytes)", source.url(), bytes_written, len)); - } else { - bar.set_message(format!("Downloading {} ({} bytes)", source.url(), bytes_written)); - } - } - - file.flush() - .await - .map_err(Error::from) - .map(|_| ()) - } - if source_path_exists /* && force is implied by 'if' above*/ { if let Err(e) = source.remove_file().await { bar.finish_with_message(format!("Failed to remove existing file: {}", source.path().display())); diff --git a/src/commands/tree_of.rs b/src/commands/tree_of.rs index b816d24..6b297e3 100644 --- a/src/commands/tree_of.rs +++ b/src/commands/tree_of.rs @@ -17,9 +17,9 @@ use anyhow::Result; use clap::ArgMatches; use resiter::AndThen; +use crate::package::Dag; use crate::package::PackageName; use crate::package::PackageVersionConstraint; -use crate::package::Dag; use crate::repository::Repository; /// Implementation of the "tree_of" subcommand |