dora_core/descriptor/
validate.rs

1use crate::{
2    adjust_shared_library_path,
3    descriptor::{self, source_is_url},
4    get_python_path,
5};
6
7use dora_message::{
8    config::{Input, InputMapping, UserInputMapping},
9    descriptor::{CoreNodeKind, OperatorSource, ResolvedNode, DYNAMIC_SOURCE, SHELL_SOURCE},
10    id::{DataId, NodeId, OperatorId},
11};
12use eyre::{bail, eyre, Context};
13use std::{collections::BTreeMap, path::Path, process::Command};
14use tracing::info;
15
16use super::{resolve_path, Descriptor, DescriptorExt};
17const VERSION: &str = env!("CARGO_PKG_VERSION");
18
19pub fn check_dataflow(
20    dataflow: &Descriptor,
21    working_dir: &Path,
22    remote_daemon_id: Option<&[&str]>,
23    coordinator_is_remote: bool,
24) -> eyre::Result<()> {
25    let nodes = dataflow.resolve_aliases_and_set_defaults()?;
26    let mut has_python_operator = false;
27
28    // check that nodes and operators exist
29    for node in nodes.values() {
30        match &node.kind {
31            descriptor::CoreNodeKind::Custom(custom) => match &custom.source {
32                dora_message::descriptor::NodeSource::Local => match custom.path.as_str() {
33                    SHELL_SOURCE => (),
34                    DYNAMIC_SOURCE => (),
35                    source => {
36                        if source_is_url(source) {
37                            info!("{source} is a URL."); // TODO: Implement url check.
38                        } else if let Some(remote_daemon_id) = remote_daemon_id {
39                            if let Some(deploy) = &node.deploy {
40                                if let Some(machine) = &deploy.machine {
41                                    if remote_daemon_id.contains(&machine.as_str())
42                                        || coordinator_is_remote
43                                    {
44                                        info!("skipping path check for remote node `{}`", node.id);
45                                    }
46                                }
47                            }
48                        } else if custom.build.is_some() {
49                            info!("skipping path check for node with build command");
50                        } else {
51                            resolve_path(source, working_dir).wrap_err_with(|| {
52                                format!("Could not find source path `{}`", source)
53                            })?;
54                        };
55                    }
56                },
57                dora_message::descriptor::NodeSource::GitBranch { repo, rev } => {
58                    info!("skipping check for node with git source");
59                }
60            },
61            descriptor::CoreNodeKind::Runtime(node) => {
62                for operator_definition in &node.operators {
63                    match &operator_definition.config.source {
64                        OperatorSource::SharedLibrary(path) => {
65                            if source_is_url(path) {
66                                info!("{path} is a URL."); // TODO: Implement url check.
67                            } else if operator_definition.config.build.is_some() {
68                                info!("skipping path check for operator with build command");
69                            } else {
70                                let path = adjust_shared_library_path(Path::new(&path))?;
71                                if !working_dir.join(&path).exists() {
72                                    bail!("no shared library at `{}`", path.display());
73                                }
74                            }
75                        }
76                        OperatorSource::Python(python_source) => {
77                            has_python_operator = true;
78                            let path = &python_source.source;
79                            if source_is_url(path) {
80                                info!("{path} is a URL."); // TODO: Implement url check.
81                            } else if !working_dir.join(path).exists() {
82                                bail!("no Python library at `{path}`");
83                            }
84                        }
85                        OperatorSource::Wasm(path) => {
86                            if source_is_url(path) {
87                                info!("{path} is a URL."); // TODO: Implement url check.
88                            } else if !working_dir.join(path).exists() {
89                                bail!("no WASM library at `{path}`");
90                            }
91                        }
92                    }
93                }
94            }
95        }
96    }
97
98    // check that all inputs mappings point to an existing output
99    for node in nodes.values() {
100        match &node.kind {
101            descriptor::CoreNodeKind::Custom(custom_node) => {
102                for (input_id, input) in &custom_node.run_config.inputs {
103                    check_input(input, &nodes, &format!("{}/{input_id}", node.id))?;
104                }
105            }
106            descriptor::CoreNodeKind::Runtime(runtime_node) => {
107                for operator_definition in &runtime_node.operators {
108                    for (input_id, input) in &operator_definition.config.inputs {
109                        check_input(
110                            input,
111                            &nodes,
112                            &format!("{}/{}/{input_id}", operator_definition.id, node.id),
113                        )?;
114                    }
115                }
116            }
117        };
118    }
119
120    // Check that nodes can resolve `send_stdout_as`
121    for node in nodes.values() {
122        node.send_stdout_as()
123            .context("Could not resolve `send_stdout_as` configuration")?;
124    }
125
126    if has_python_operator {
127        check_python_runtime()?;
128    }
129
130    Ok(())
131}
132
133pub trait ResolvedNodeExt {
134    fn send_stdout_as(&self) -> eyre::Result<Option<String>>;
135}
136
137impl ResolvedNodeExt for ResolvedNode {
138    fn send_stdout_as(&self) -> eyre::Result<Option<String>> {
139        match &self.kind {
140            // TODO: Split stdout between operators
141            CoreNodeKind::Runtime(n) => {
142                let count = n
143                    .operators
144                    .iter()
145                    .filter(|op| op.config.send_stdout_as.is_some())
146                    .count();
147                if count == 1 && n.operators.len() > 1 {
148                    tracing::warn!("All stdout from all operators of a runtime are going to be sent in the selected `send_stdout_as` operator.")
149                } else if count > 1 {
150                    return Err(eyre!("More than one `send_stdout_as` entries for a runtime node. Please only use one `send_stdout_as` per runtime."));
151                }
152                Ok(n.operators.iter().find_map(|op| {
153                    op.config
154                        .send_stdout_as
155                        .clone()
156                        .map(|stdout| format!("{}/{}", op.id, stdout))
157                }))
158            }
159            CoreNodeKind::Custom(n) => Ok(n.send_stdout_as.clone()),
160        }
161    }
162}
163
164fn check_input(
165    input: &Input,
166    nodes: &BTreeMap<NodeId, super::ResolvedNode>,
167    input_id_str: &str,
168) -> Result<(), eyre::ErrReport> {
169    match &input.mapping {
170        InputMapping::Timer { interval: _ } => {}
171        InputMapping::User(UserInputMapping { source, output }) => {
172            let source_node = nodes.values().find(|n| &n.id == source).ok_or_else(|| {
173                eyre!("source node `{source}` mapped to input `{input_id_str}` does not exist",)
174            })?;
175            match &source_node.kind {
176                CoreNodeKind::Custom(custom_node) => {
177                    if !custom_node.run_config.outputs.contains(output) {
178                        bail!(
179                            "output `{source}/{output}` mapped to \
180                            input `{input_id_str}` does not exist",
181                        );
182                    }
183                }
184                CoreNodeKind::Runtime(runtime) => {
185                    let (operator_id, output) = output.split_once('/').unwrap_or_default();
186                    let operator_id = OperatorId::from(operator_id.to_owned());
187                    let output = DataId::from(output.to_owned());
188
189                    let operator = runtime
190                        .operators
191                        .iter()
192                        .find(|o| o.id == operator_id)
193                        .ok_or_else(|| {
194                            eyre!(
195                                "source operator `{source}/{operator_id}` used \
196                                for input `{input_id_str}` does not exist",
197                            )
198                        })?;
199
200                    if !operator.config.outputs.contains(&output) {
201                        bail!(
202                            "output `{source}/{operator_id}/{output}` mapped to \
203                            input `{input_id_str}` does not exist",
204                        );
205                    }
206                }
207            }
208        }
209    };
210    Ok(())
211}
212
213fn check_python_runtime() -> eyre::Result<()> {
214    // Check if python dora-rs is installed and match cli version
215    let reinstall_command =
216        format!("Please reinstall it with: `pip install dora-rs=={VERSION} --force`");
217    let mut command = Command::new(get_python_path().context("Could not get python binary")?);
218    command.args([
219        "-c",
220        &format!(
221            "
222import dora;
223assert dora.__version__=='{VERSION}',  'Python dora-rs should be {VERSION}, but current version is %s. {reinstall_command}' % (dora.__version__)
224        "
225        ),
226    ]);
227    let mut result = command
228        .spawn()
229        .wrap_err("Could not spawn python dora-rs command.")?;
230    let status = result
231        .wait()
232        .wrap_err("Could not get exit status when checking python dora-rs")?;
233
234    if !status.success() {
235        bail!("Something went wrong with Python dora-rs. {reinstall_command}")
236    }
237
238    Ok(())
239}
dora_core/descriptor/validate.rs

dora_core/descriptor/
validate.rs
