diff --git a/asap-planner-rs/src/bin/optimizer_cli.rs b/asap-planner-rs/src/bin/optimizer_cli.rs index 91e083a2..e4ba6e78 100644 --- a/asap-planner-rs/src/bin/optimizer_cli.rs +++ b/asap-planner-rs/src/bin/optimizer_cli.rs @@ -3,15 +3,12 @@ use std::path::PathBuf; -use anyhow::Context; use asap_planner::optimizer::{ - build_milp_workload, load_flat_atomic_cost_table, load_workload_facts, plan_to_planner_output, - reject_avg_queries, solve_milp, MilpError, + parse_weight, plan_milp, plan_to_planner_output, reject_unwritable_queries, MilpInputs, + MilpPlan, }; use asap_planner::ControllerConfig; use clap::Parser; -use rqe_optimizer::milp::Objective; -use rqe_optimizer::saturation::SaturationCurves; #[derive(Parser, Debug)] #[command( @@ -84,51 +81,26 @@ fn main() -> anyhow::Result<()> { run_milp(&args, &config) } -/// Objective weights must be finite and non-negative: a negative weight -/// rewards cost, and NaN poisons every coefficient. -fn parse_weight(s: &str) -> Result { - match s.parse::() { - Ok(w) if w.is_finite() && w >= 0.0 => Ok(w), - Ok(w) => Err(format!("must be finite and >= 0, got {w}")), - Err(e) => Err(e.to_string()), - } -} - fn run_milp(args: &Args, config: &ControllerConfig) -> anyhow::Result<()> { - config.warn_default_slas(); - let Some(hints) = config.metrics.as_deref() else { - return Err(MilpError::MissingMetricHints.into()); - }; - let facts = load_workload_facts(&args.workload_facts, hints, args.data_ingestion_interval_ms)?; - let curves = SaturationCurves::load(&args.saturation_dir).with_context(|| { - format!( - "loading saturation curves from --saturation-dir {}", - args.saturation_dir.display() - ) - })?; - let costs = load_flat_atomic_cost_table(&args.atomic_costs)?; - let Objective::AUCCost { w_cpu, w_mem } = Objective::default(); - let (w_cpu, w_mem) = (args.w_cpu.unwrap_or(w_cpu), args.w_mem.unwrap_or(w_mem)); - // All-zero weights make every plan cost 0, so the solver's pick is arbitrary. - anyhow::ensure!( - w_cpu > 0.0 || w_mem > 0.0, - "--w-cpu and --w-mem are both 0; at least one must be positive" - ); - let objective = Objective::AUCCost { w_cpu, w_mem }; - tracing::debug!(?objective, cost_rows = costs.len(), "milp: inputs loaded"); - // Fail before solving when the plan would be written but can't be. if args.output_dir.is_some() { - reject_avg_queries(config)?; + reject_unwritable_queries(config)?; } - let workload = build_milp_workload(config, &facts, args.data_ingestion_interval_ms)?; - let solution = solve_milp( - &workload, - &facts, - &costs, + let MilpPlan { + workload, + solution, objective, - args.allow_undeployable_families, - &|raqe, deployment| curves.accuracy(raqe, deployment, &facts), + } = plan_milp( + config, + &MilpInputs { + workload_facts: &args.workload_facts, + atomic_costs: &args.atomic_costs, + saturation_dir: &args.saturation_dir, + scrape_interval_ms: args.data_ingestion_interval_ms, + w_cpu: args.w_cpu, + w_mem: args.w_mem, + allow_undeployable_families: args.allow_undeployable_families, + }, )?; println!("=== Deployments: {} ===", solution.deployments.len()); @@ -180,29 +152,8 @@ fn run_milp(args: &Args, config: &ControllerConfig) -> anyhow::Result<()> { } if let Some(dir) = &args.output_dir { - let output = plan_to_planner_output(config, &workload, &solution)?; - // Serialize both before writing either, so a failure can't leave a - // new streaming config next to a stale inference config. - let streaming = output.to_streaming_yaml_string()?; - let inference = output.to_inference_yaml_string()?; - std::fs::create_dir_all(dir)?; - std::fs::write(dir.join("streaming_config.yaml"), streaming)?; - std::fs::write(dir.join("inference_config.yaml"), inference)?; + plan_to_planner_output(config, &workload, &solution)?.write_to_dir(dir)?; println!("\nwrote configs to {}", dir.display()); } Ok(()) } - -#[cfg(test)] -mod tests { - use super::parse_weight; - - #[test] - fn weights_must_be_finite_and_non_negative() { - assert_eq!(parse_weight("0.5"), Ok(0.5)); - assert_eq!(parse_weight("0"), Ok(0.0)); - for bad in ["-1", "NaN", "inf", "x"] { - assert!(parse_weight(bad).is_err(), "{bad}"); - } - } -} diff --git a/asap-planner-rs/src/elastic_dsl/controller.rs b/asap-planner-rs/src/elastic_dsl/controller.rs index 22df6872..313fdb63 100644 --- a/asap-planner-rs/src/elastic_dsl/controller.rs +++ b/asap-planner-rs/src/elastic_dsl/controller.rs @@ -35,11 +35,7 @@ impl ElasticController { pub fn generate_to_dir(&self, dir: &Path) -> Result { let output = self.generate()?; - std::fs::create_dir_all(dir)?; - let streaming_str = serde_yaml::to_string(output.streaming_yaml())?; - let inference_str = serde_yaml::to_string(output.inference_yaml())?; - std::fs::write(dir.join("streaming_config.yaml"), streaming_str)?; - std::fs::write(dir.join("inference_config.yaml"), inference_str)?; + output.write_to_dir(dir)?; Ok(output) } } diff --git a/asap-planner-rs/src/main.rs b/asap-planner-rs/src/main.rs index cb345361..c4b2e20b 100644 --- a/asap-planner-rs/src/main.rs +++ b/asap-planner-rs/src/main.rs @@ -1,6 +1,9 @@ +use asap_planner::optimizer::{ + parse_weight, plan_milp, plan_to_planner_output, reject_unwritable_queries, MilpInputs, +}; use asap_planner::{ - Controller, ElasticController, ElasticRuntimeOptions, RuntimeOptions, SQLController, - SQLRuntimeOptions, StreamingEngine, + Controller, ControllerConfig, ElasticController, ElasticRuntimeOptions, RuntimeOptions, + SQLController, SQLRuntimeOptions, StreamingEngine, }; use asap_types::enums::QueryLanguage; use clap::Parser; @@ -53,6 +56,36 @@ struct Args { #[arg(long = "clickhouse-database", required = false)] clickhouse_database: Option, + /// `milp` plans with sketch-bench's rqe-optimizer: PromQL with + /// --input_config only, labels from its `metrics:` hints. + #[arg(long, value_enum, default_value = "legacy")] + planner: PlannerArg, + + /// MILP only. YAML workload facts: per metric, positive `value_range` and + /// `cardinality` per label set, including all labels (the series count), + /// plus the `shape` of each grouping sketches may serve. + #[arg(long = "workload-facts", required_if_eq("planner", "milp"))] + workload_facts: Option, + + /// MILP only. The flat cost table sketch-bench's + /// `study_saturation.py --phase optimizer-cost` writes + /// (`rqe_atomic_costs.json`). + #[arg(long = "atomic-costs", required_if_eq("planner", "milp"))] + atomic_costs: Option, + + /// MILP only. sketch-bench's saturation-study directory: sketch accuracy + /// is read off its error-vs-N curves at each grouping's `shape`. + #[arg(long = "saturation-dir", required_if_eq("planner", "milp"))] + saturation_dir: Option, + + /// MILP only. Objective weight on CPU-sec/sec. Default: rqe-optimizer's. + #[arg(long = "w-cpu", value_parser = parse_weight)] + w_cpu: Option, + + /// MILP only. Objective weight on memory GiB. Default: rqe-optimizer's. + #[arg(long = "w-mem", value_parser = parse_weight)] + w_mem: Option, + #[arg(short, long, action = clap::ArgAction::Count)] verbose: u8, } @@ -62,6 +95,12 @@ enum EngineArg { Precompute, } +#[derive(clap::ValueEnum, Debug, Clone, Copy, PartialEq, Eq)] +enum PlannerArg { + Legacy, + Milp, +} + fn main() -> anyhow::Result<()> { let args = Args::parse(); @@ -77,6 +116,21 @@ fn main() -> anyhow::Result<()> { EngineArg::Precompute => StreamingEngine::Precompute, }; + if args.planner == PlannerArg::Milp { + run_milp(&args)?; + println!("Generated configs in {}", args.output_dir.display()); + return Ok(()); + } + anyhow::ensure!( + args.workload_facts.is_none() + && args.atomic_costs.is_none() + && args.saturation_dir.is_none() + && args.w_cpu.is_none() + && args.w_mem.is_none(), + "--workload-facts, --atomic-costs, --saturation-dir, --w-cpu and --w-mem require \ + --planner milp" + ); + match args.query_language { QueryLanguage::promql => { let scrape_interval_ms = args.data_ingestion_interval_ms.ok_or_else(|| { @@ -158,3 +212,65 @@ fn main() -> anyhow::Result<()> { println!("Generated configs in {}", args.output_dir.display()); Ok(()) } + +fn run_milp(args: &Args) -> anyhow::Result<()> { + anyhow::ensure!( + matches!(args.query_language, QueryLanguage::promql), + "--planner milp supports only --query-language promql" + ); + anyhow::ensure!( + args.query_log.is_none(), + "--planner milp needs --input_config: query logs have no `metrics:` hints" + ); + anyhow::ensure!( + args.prometheus_url.is_none(), + "--planner milp takes labels from the config's `metrics:` hints, not --prometheus-url" + ); + anyhow::ensure!( + !args.enable_punting && args.range_duration_ms == 0 && args.step_ms == 0, + "--enable-punting, --range-duration-ms and --step-ms don't apply to --planner milp" + ); + anyhow::ensure!( + args.clickhouse_url.is_none() && args.clickhouse_database.is_none(), + "--clickhouse-url and --clickhouse-database don't apply to --planner milp" + ); + let config_path = args + .input_config + .as_deref() + .ok_or_else(|| anyhow::anyhow!("--planner milp requires --input_config"))?; + let scrape_interval_ms = args + .data_ingestion_interval_ms + .ok_or_else(|| anyhow::anyhow!("--planner milp requires --data-ingestion-interval-ms"))?; + // Per-series sample rates divide by it. + anyhow::ensure!( + scrape_interval_ms > 0, + "--data-ingestion-interval-ms must be positive" + ); + let config: ControllerConfig = serde_yaml::from_str(&std::fs::read_to_string(config_path)?)?; + // Fail before solving: the plan is always written. + reject_unwritable_queries(&config)?; + let plan = plan_milp( + &config, + &MilpInputs { + workload_facts: args + .workload_facts + .as_deref() + .expect("clap requires --workload-facts with --planner milp"), + atomic_costs: args + .atomic_costs + .as_deref() + .expect("clap requires --atomic-costs with --planner milp"), + saturation_dir: args + .saturation_dir + .as_deref() + .expect("clap requires --saturation-dir with --planner milp"), + scrape_interval_ms, + w_cpu: args.w_cpu, + w_mem: args.w_mem, + allow_undeployable_families: false, + }, + )?; + plan_to_planner_output(&config, &plan.workload, &plan.solution)? + .write_to_dir(&args.output_dir)?; + Ok(()) +} diff --git a/asap-planner-rs/src/optimizer/milp.rs b/asap-planner-rs/src/optimizer/milp.rs index c87b144a..71d8d83c 100644 --- a/asap-planner-rs/src/optimizer/milp.rs +++ b/asap-planner-rs/src/optimizer/milp.rs @@ -13,9 +13,16 @@ use thiserror::Error; use crate::config::input::ControllerConfig; +use std::path::Path; + +use anyhow::Context; +use rqe_optimizer::saturation::SaturationCurves; + use super::aqe_extractor::{extract_aqes, RQE}; +use super::atomic_costs::load_flat_atomic_cost_table; use super::error::OptimizerError; use super::solution::OptimizerItem; +use super::workload_facts::load_workload_facts; /// Slack on accuracy tolerances so `1 - sla` rounding (`1 - 0.9 = /// 0.0999...98`) doesn't reject a row measured exactly at the boundary. @@ -23,6 +30,8 @@ const SLA_EPSILON: f64 = 1e-9; #[derive(Debug, Error)] pub enum MilpError { + #[error("the MILP planner doesn't support these workload config fields: {0:?}")] + UnsupportedFields(Vec<&'static str>), #[error( "workload config has no `metrics:` hints; they are required to resolve grouping labels" )] @@ -110,6 +119,84 @@ pub fn solve_milp( Ok(solution) } +/// What a MILP planning run reads besides the workload config. +pub struct MilpInputs<'a> { + pub workload_facts: &'a Path, + pub atomic_costs: &'a Path, + /// sketch-bench's saturation-study directory; sketch accuracy is read + /// off its error-vs-N curves at each grouping's `shape`. + pub saturation_dir: &'a Path, + pub scrape_interval_ms: u64, + /// Objective weights; `None` takes rqe-optimizer's default. + pub w_cpu: Option, + pub w_mem: Option, + pub allow_undeployable_families: bool, +} + +pub struct MilpPlan { + pub workload: MilpWorkload, + pub solution: MilpSolution, + pub objective: Objective, +} + +/// Loads the facts, costs and saturation curves and solves for the cheapest +/// plan. +pub fn plan_milp(config: &ControllerConfig, inputs: &MilpInputs) -> anyhow::Result { + config.warn_default_slas(); + let Some(hints) = config.metrics.as_deref() else { + return Err(MilpError::MissingMetricHints.into()); + }; + let facts = load_workload_facts(inputs.workload_facts, hints, inputs.scrape_interval_ms)?; + let curves = SaturationCurves::load(inputs.saturation_dir).with_context(|| { + format!( + "loading saturation curves from {}", + inputs.saturation_dir.display() + ) + })?; + let costs = load_flat_atomic_cost_table(inputs.atomic_costs)?; + let objective = objective(inputs.w_cpu, inputs.w_mem)?; + tracing::debug!(?objective, cost_rows = costs.len(), "milp: inputs loaded"); + + let workload = build_milp_workload(config, &facts, inputs.scrape_interval_ms)?; + let solution = solve_milp( + &workload, + &facts, + &costs, + objective, + inputs.allow_undeployable_families, + &|raqe, deployment| curves.accuracy(raqe, deployment, &facts), + )?; + Ok(MilpPlan { + workload, + solution, + objective, + }) +} + +fn objective(w_cpu: Option, w_mem: Option) -> anyhow::Result { + let Objective::AUCCost { + w_cpu: default_cpu, + w_mem: default_mem, + } = Objective::default(); + let (w_cpu, w_mem) = (w_cpu.unwrap_or(default_cpu), w_mem.unwrap_or(default_mem)); + // All-zero weights make every plan cost 0, so the solver's pick is arbitrary. + anyhow::ensure!( + w_cpu > 0.0 || w_mem > 0.0, + "--w-cpu and --w-mem are both 0; at least one must be positive" + ); + Ok(Objective::AUCCost { w_cpu, w_mem }) +} + +/// Parses an objective weight flag. Weights must be finite and non-negative: +/// a negative weight rewards cost, and NaN poisons every coefficient. +pub fn parse_weight(s: &str) -> Result { + match s.parse::() { + Ok(w) if w.is_finite() && w >= 0.0 => Ok(w), + Ok(w) => Err(format!("must be finite and >= 0, got {w}")), + Err(e) => Err(e.to_string()), + } +} + /// A Raqe a sketch family with cost rows may serve reads its accuracy off /// the curves at its grouping's `shape`; without one it could only be /// reported unservable. Families without rows are left to that path. @@ -157,6 +244,18 @@ pub fn build_milp_workload( facts: &WorkloadFacts, scrape_interval_ms: u64, ) -> Result { + // The MILP chooses windows and sketch parameters itself. + let unsupported: Vec<&'static str> = [ + ("windowing", config.windowing.is_some()), + ("sketch_parameters", config.sketch_parameters.is_some()), + ("aggregate_cleanup", config.aggregate_cleanup.is_some()), + ] + .into_iter() + .filter_map(|(field, set)| set.then_some(field)) + .collect(); + if !unsupported.is_empty() { + return Err(MilpError::UnsupportedFields(unsupported)); + } let mut items = extract_hinted_items(config, scrape_interval_ms)?; // Stable Raqe order and ids across runs. items.sort_by(|a, b| { @@ -598,6 +697,29 @@ metrics: assert!(matches!(err, MilpError::InvalidInputs(_))); } + /// The MILP picks windows and sketch parameters itself, so these would be + /// dropped silently. + #[test] + fn unsupported_config_fields_are_rejected() { + let mut config = config(&group("sum(http_requests_total)", 0.99)); + config.windowing = + Some(serde_yaml::from_str("{type: tumbling, window_size_ms: 60000}").unwrap()); + config.sketch_parameters = Some(Default::default()); + let err = build_milp_workload(&config, &facts(&config), SCRAPE_MS).unwrap_err(); + assert!( + matches!(&err, MilpError::UnsupportedFields(f) if f == &["windowing", "sketch_parameters"]), + "{err}" + ); + config.windowing = None; + config.sketch_parameters = None; + config.aggregate_cleanup = Some(serde_yaml::from_str("{}").unwrap()); + let err = build_milp_workload(&config, &facts(&config), SCRAPE_MS).unwrap_err(); + assert!( + matches!(&err, MilpError::UnsupportedFields(f) if f == &["aggregate_cleanup"]), + "{err}" + ); + } + #[test] fn solve_shares_one_deployment_across_repeated_queries() { let query = "sum by (job) (http_requests_total)"; @@ -773,6 +895,15 @@ metrics: assert_eq!(solution.deployments.len(), 2); } + #[test] + fn weights_must_be_finite_and_non_negative() { + assert_eq!(parse_weight("0.5"), Ok(0.5)); + assert_eq!(parse_weight("0"), Ok(0.0)); + for bad in ["-1", "NaN", "inf", "x"] { + assert!(parse_weight(bad).is_err(), "{bad}"); + } + } + #[test] fn workload_requires_metric_hints() { let config: ControllerConfig = serde_yaml::from_str(&format!( diff --git a/asap-planner-rs/src/optimizer/milp_output.rs b/asap-planner-rs/src/optimizer/milp_output.rs index 715ea73b..d0c0db41 100644 --- a/asap-planner-rs/src/optimizer/milp_output.rs +++ b/asap-planner-rs/src/optimizer/milp_output.rs @@ -29,6 +29,11 @@ use super::solution::OptimizerItem; pub enum MilpOutputError { #[error("query {0:?} uses avg, which the engine can't answer from sum and count yet")] AvgQuery(String), + #[error( + "query groups set step_ms or range_duration_ms, whose retention MILP configs \ + don't set yet (#800): {0:?}" + )] + RangeQueryOverrides(Vec), #[error("sketch-bench variant {variant} ({capability:?}) has no ASAPQuery aggregation type")] UndeployableVariant { variant: String, @@ -47,17 +52,27 @@ pub enum MilpOutputError { Generator(#[from] ControllerError), } -/// Errors on the first avg query: its plan can be costed but not deployed. -pub fn reject_avg_queries(config: &ControllerConfig) -> Result<(), MilpOutputError> { - match config +/// Errors on queries whose plan can be costed but not written as configs: +/// avg queries, and range-query overrides, which only size retention. +pub fn reject_unwritable_queries(config: &ControllerConfig) -> Result<(), MilpOutputError> { + if let Some(query) = config .query_groups .iter() .flat_map(|group| &group.queries) .find(|query| contains_avg(query)) { - Some(query) => Err(MilpOutputError::AvgQuery(query.clone())), - None => Ok(()), + return Err(MilpOutputError::AvgQuery(query.clone())); } + let range_queries: Vec = config + .query_groups + .iter() + .filter(|qg| qg.step_ms.is_some() || qg.range_duration_ms.is_some()) + .flat_map(|qg| qg.queries.iter().cloned()) + .collect(); + if !range_queries.is_empty() { + return Err(MilpOutputError::RangeQueryOverrides(range_queries)); + } + Ok(()) } /// Streaming and inference YAML for `solution`. Aggregation ids follow the @@ -68,7 +83,7 @@ pub fn plan_to_planner_output( workload: &MilpWorkload, solution: &MilpSolution, ) -> Result { - reject_avg_queries(config)?; + reject_unwritable_queries(config)?; let item_of = |raqe: usize| &workload.items[workload.raqe_items[raqe]]; // Each item once per deployment, however many occurrences it has. @@ -578,6 +593,24 @@ mod tests { assert!(matches!(err, MilpOutputError::AvgQuery(q) if q == query)); } + /// Range overrides only size retention, so they block writing configs + /// but not planning. + #[test] + fn range_query_overrides_block_only_the_written_configs() { + let query = "sum(http_requests_total)"; + for field in ["step_ms: 60000", "range_duration_ms: 3600000"] { + let (config, workload, solution) = + plan(&format!("{} {field}\n", group(query, 0.99))); + let err = plan_to_planner_output(&config, &workload, &solution) + .err() + .unwrap(); + assert!( + matches!(&err, MilpOutputError::RangeQueryOverrides(q) if q == &[query]), + "{field}: {err}" + ); + } + } + #[test] fn hydra_kll_is_not_deployed() { // Its DeltaSet key tracker has no ASAPQuery pairing yet. diff --git a/asap-planner-rs/src/optimizer/mod.rs b/asap-planner-rs/src/optimizer/mod.rs index f7e1cea5..7731c14c 100644 --- a/asap-planner-rs/src/optimizer/mod.rs +++ b/asap-planner-rs/src/optimizer/mod.rs @@ -9,7 +9,10 @@ pub mod workload_facts; pub use aqe_extractor::{extract_aqes, RQE}; pub use atomic_costs::{load_flat_atomic_cost_table, AtomicCostEntry, AtomicCostTable}; pub use error::OptimizerError; -pub use milp::{build_milp_workload, solve_milp, MilpError, MilpWorkload}; -pub use milp_output::{plan_to_planner_output, reject_avg_queries, MilpOutputError}; +pub use milp::{ + build_milp_workload, parse_weight, plan_milp, solve_milp, MilpError, MilpInputs, MilpPlan, + MilpWorkload, +}; +pub use milp_output::{plan_to_planner_output, reject_unwritable_queries, MilpOutputError}; pub use solution::OptimizerItem; pub use workload_facts::{load_workload_facts, parse_workload_facts, WorkloadFactsError}; diff --git a/asap-planner-rs/src/planner_output.rs b/asap-planner-rs/src/planner_output.rs index 036a8c79..3d72be25 100644 --- a/asap-planner-rs/src/planner_output.rs +++ b/asap-planner-rs/src/planner_output.rs @@ -4,6 +4,7 @@ use asap_types::enums::QueryLanguage; use asap_types::inference_config::InferenceConfig; use asap_types::streaming_config::StreamingConfig; +use crate::error::ControllerError; use crate::generator::{ GeneratorOutput, PuntedQuery, KEY_AGGREGATIONS, KEY_AGG_SUB_TYPE, KEY_AGG_TYPE, KEY_LABELS, KEY_NUM_AGG_TO_RETAIN, KEY_PARAMETERS, KEY_QUERIES, KEY_QUERY, KEY_READ_COUNT_THRESHOLD, @@ -30,14 +31,6 @@ impl PlannerOutput { } } - pub(crate) fn streaming_yaml(&self) -> &YamlValue { - &self.streaming_yaml - } - - pub(crate) fn inference_yaml(&self) -> &YamlValue { - &self.inference_yaml - } - pub fn streaming_aggregation_count(&self) -> usize { self.aggregation_count } @@ -171,6 +164,18 @@ impl PlannerOutput { Ok(serde_yaml::to_string(&self.inference_yaml)?) } + /// Writes `streaming_config.yaml` and `inference_config.yaml`. Both are + /// serialized first, so a failure can't leave a new streaming config next + /// to a stale inference config. + pub fn write_to_dir(&self, dir: &std::path::Path) -> Result<(), ControllerError> { + let streaming = serde_yaml::to_string(&self.streaming_yaml)?; + let inference = serde_yaml::to_string(&self.inference_yaml)?; + std::fs::create_dir_all(dir)?; + std::fs::write(dir.join("streaming_config.yaml"), streaming)?; + std::fs::write(dir.join("inference_config.yaml"), inference)?; + Ok(()) + } + pub fn to_streaming_config( &self, query_language: QueryLanguage, diff --git a/asap-planner-rs/src/promql/controller.rs b/asap-planner-rs/src/promql/controller.rs index a24d6edc..354a7e84 100644 --- a/asap-planner-rs/src/promql/controller.rs +++ b/asap-planner-rs/src/promql/controller.rs @@ -162,11 +162,7 @@ impl Controller { pub fn generate_to_dir(&self, dir: &Path) -> Result { let output = self.generate()?; - std::fs::create_dir_all(dir)?; - let streaming_str = serde_yaml::to_string(output.streaming_yaml())?; - let inference_str = serde_yaml::to_string(output.inference_yaml())?; - std::fs::write(dir.join("streaming_config.yaml"), streaming_str)?; - std::fs::write(dir.join("inference_config.yaml"), inference_str)?; + output.write_to_dir(dir)?; Ok(output) } } diff --git a/asap-planner-rs/src/sql/controller.rs b/asap-planner-rs/src/sql/controller.rs index d9da13b9..5587315e 100644 --- a/asap-planner-rs/src/sql/controller.rs +++ b/asap-planner-rs/src/sql/controller.rs @@ -90,11 +90,7 @@ impl SQLController { pub fn generate_to_dir(&self, dir: &Path) -> Result { let output = self.generate()?; - std::fs::create_dir_all(dir)?; - let streaming_str = serde_yaml::to_string(output.streaming_yaml())?; - let inference_str = serde_yaml::to_string(output.inference_yaml())?; - std::fs::write(dir.join("streaming_config.yaml"), streaming_str)?; - std::fs::write(dir.join("inference_config.yaml"), inference_str)?; + output.write_to_dir(dir)?; Ok(output) } }