swale 0.2.0

A scheduled asset-graph orchestrator on object storage.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82

//! `swale daemon`.

use std::collections::BTreeMap;
use std::process::ExitCode;
use std::sync::Arc;
use std::time::Duration;

use swale::{Daemon, DaemonOptions, OperatorSet};

use super::CommandResult;
use super::store::{Runtime, StoreArg, create_store};

pub(crate) async fn daemon(
    store: StoreArg,
    concurrency: usize,
    pools: Vec<(String, usize)>,
    sync_interval: Duration,
    retention: Duration,
) -> CommandResult {
    tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::try_from_env("SWALE_LOG")
                .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("warn,swale=info")),
        )
        .with_ansi(std::io::IsTerminal::is_terminal(&std::io::stderr()))
        .with_writer(std::io::stderr)
        .init();

    let store = create_store(store)?;
    let operators = Arc::new(OperatorSet::builtin());
    let definitions = store.definitions(operators.clone());
    let mut pool_sizes = BTreeMap::from([("default".to_string(), concurrency)]);
    pool_sizes.extend(pools);
    let requests = store.requests();
    let runtime = Runtime::open(store, operators, definitions, &pool_sizes).await?;
    let daemon = Daemon::new(
        runtime.queue.clone(),
        runtime.scheduler.clone(),
        runtime.pools.clone(),
        requests,
    );
    let result = daemon
        .run(
            DaemonOptions {
                sync_interval,
                retention,
                ..DaemonOptions::default()
            },
            async {
                let _ = tokio::signal::ctrl_c().await;
            },
        )
        .await;
    drop(daemon);
    runtime.close().await?;
    result?;
    Ok(ExitCode::SUCCESS)
}

/// Parses a `--pool` argument of the form `name=steps`.
pub(crate) fn parse_pool_arg(raw: &str) -> Result<(String, usize), String> {
    let invalid = || format!("`{raw}` is not `name=steps`, such as `warehouse=2`");
    let (name, steps) = raw.split_once('=').ok_or_else(invalid)?;
    let steps: usize = steps.parse().map_err(|_| invalid())?;
    if !swale::graph::is_name(name) || steps == 0 {
        return Err(invalid());
    }
    Ok((name.to_string(), steps))
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn a_pool_argument_is_a_name_and_a_positive_step_count() {
        assert_eq!(parse_pool_arg("warehouse=2"), Ok(("warehouse".into(), 2)));
        for raw in ["warehouse", "warehouse=0", "warehouse=x", "Bad=1", "=1"] {
            assert!(parse_pool_arg(raw).is_err(), "{raw}");
        }
    }
}