dora_message/
coordinator_to_cli.rs

1use std::{collections::BTreeMap, net::IpAddr};
2
3use uuid::Uuid;
4
5pub use crate::common::{LogLevel, LogMessage, NodeError, NodeErrorCause, NodeExitStatus};
6use crate::{
7    BuildId,
8    common::DaemonId,
9    descriptor::Descriptor,
10    id::{DataId, NodeId},
11};
12
13#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
14pub enum ControlRequestReply {
15    Error(String),
16    CoordinatorStopped,
17    /// Response to [`ControlRequest::Hello`][crate::cli_to_coordinator::ControlRequest::Hello].
18    /// Carries the coordinator's own dora crate version so the CLI can
19    /// display it on version mismatches and in debug output
20    /// (dora-rs/adora#151).
21    HelloOk {
22        dora_version: semver::Version,
23    },
24    DataflowBuildTriggered {
25        build_id: BuildId,
26    },
27    DataflowBuildFinished {
28        build_id: BuildId,
29        result: Result<(), String>,
30    },
31    DataflowStartTriggered {
32        uuid: Uuid,
33    },
34    DataflowSpawned {
35        uuid: Uuid,
36    },
37    DataflowReloaded {
38        uuid: Uuid,
39    },
40    DataflowStopped {
41        uuid: Uuid,
42        result: DataflowResult,
43    },
44    DataflowRestarted {
45        old_uuid: Uuid,
46        new_uuid: Uuid,
47    },
48    DataflowList(DataflowList),
49    /// Response to [`ControlRequest::Clean`][crate::cli_to_coordinator::ControlRequest::Clean].
50    ///
51    /// `cleaned` lists dataflows the coordinator successfully removed
52    /// from both in-memory state and the persisted store (with the
53    /// cascade-delete of associated `dora param` rows). `failed` lists
54    /// dataflows that were eligible but whose persisted-store delete
55    /// errored: their in-memory entries are preserved so a later
56    /// `dora clean` can retry. The CLI needs both lists to tell
57    /// "nothing eligible" apart from "all candidates failed to clean".
58    CleanResult {
59        cleaned: DataflowList,
60        failed: Vec<CleanFailure>,
61    },
62    DataflowInfo {
63        uuid: Uuid,
64        name: Option<String>,
65        descriptor: Descriptor,
66    },
67    DestroyOk,
68    DaemonConnected(bool),
69    ConnectedDaemons(Vec<DaemonInfo>),
70    Logs(Vec<u8>),
71    CliAndDefaultDaemonIps {
72        default_daemon: Option<IpAddr>,
73        cli: Option<IpAddr>,
74    },
75    NodeInfoList(Vec<NodeInfo>),
76    TopicSubscribed {
77        subscription_id: Uuid,
78        /// Binary-frame encoding the coordinator will send, see
79        /// [`TOPIC_DATA_PROTOCOL_VERSION`](crate::TOPIC_DATA_PROTOCOL_VERSION).
80        ///
81        /// `None` means the coordinator predates the handshake and therefore
82        /// sends bincode frames; the client rejects the subscription rather
83        /// than misparse them as postcard.
84        #[serde(default)]
85        protocol_version: Option<u16>,
86    },
87    TraceList(Vec<TraceSummary>),
88    TraceSpans(Vec<TraceSpan>),
89    NodeRestarted {
90        dataflow_id: Uuid,
91        node_id: NodeId,
92    },
93    NodeStopped {
94        dataflow_id: Uuid,
95        node_id: NodeId,
96    },
97    TopicPublished,
98    // --- Dynamic Topology ---
99    NodeAdded {
100        dataflow_id: Uuid,
101        node_id: NodeId,
102    },
103    NodeRemoved {
104        dataflow_id: Uuid,
105        node_id: NodeId,
106    },
107    NodeReplaced {
108        dataflow_id: Uuid,
109        node_id: NodeId,
110    },
111    MappingAdded {
112        dataflow_id: Uuid,
113        source_node: NodeId,
114        source_output: DataId,
115        target_node: NodeId,
116        target_input: DataId,
117    },
118    MappingRemoved {
119        dataflow_id: Uuid,
120        source_node: NodeId,
121        source_output: DataId,
122        target_node: NodeId,
123        target_input: DataId,
124    },
125    ParamList {
126        params: Vec<(String, serde_json::Value)>,
127    },
128    ParamValue {
129        key: String,
130        value: serde_json::Value,
131    },
132    ParamSet,
133    ParamDeleted,
134}
135
136#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
137pub struct NodeInfo {
138    pub dataflow_id: Uuid,
139    pub dataflow_name: Option<String>,
140    pub node_id: NodeId,
141    pub daemon_id: DaemonId,
142    pub metrics: Option<NodeMetricsInfo>,
143    /// Per-dataflow cross-daemon network I/O counters (shared across nodes in same dataflow)
144    #[serde(default)]
145    pub network: Option<crate::daemon_to_coordinator::NetworkMetrics>,
146}
147
148/// Resource metrics for a node (from daemon)
149#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
150pub struct NodeMetricsInfo {
151    /// Process ID
152    pub pid: u32,
153    /// CPU usage percentage (0-100 per core)
154    pub cpu_usage: f32,
155    /// Memory usage in megabytes
156    pub memory_mb: f64,
157    /// Disk read MB/s (if available)
158    pub disk_read_mb_s: Option<f64>,
159    /// Disk write MB/s (if available)
160    pub disk_write_mb_s: Option<f64>,
161    /// Number of times this node has been restarted
162    #[serde(default)]
163    pub restart_count: u32,
164    /// Input IDs that have timed out (circuit breaker open)
165    #[serde(default)]
166    pub broken_inputs: Vec<String>,
167    /// Current health status
168    #[serde(default)]
169    pub status: crate::daemon_to_coordinator::NodeStatus,
170    /// Number of pending messages in the node's input queue
171    #[serde(default)]
172    pub pending_messages: u64,
173}
174
175/// Health information about a connected daemon.
176#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
177pub struct DaemonInfo {
178    pub daemon_id: DaemonId,
179    pub last_heartbeat_ago_ms: u64,
180    /// Fault tolerance stats from the daemon (if available).
181    #[serde(default)]
182    pub ft_stats: Option<crate::daemon_to_coordinator::FaultToleranceSnapshot>,
183}
184
185#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
186pub struct TraceSummary {
187    pub trace_id: String,
188    pub root_span_name: String,
189    pub span_count: usize,
190    pub start_time: u64,
191    pub total_duration_us: u64,
192}
193
194#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
195pub struct TraceSpan {
196    pub trace_id: String,
197    pub span_id: u64,
198    pub parent_span_id: Option<u64>,
199    pub name: String,
200    pub target: String,
201    pub level: String,
202    pub start_time: u64,
203    pub duration_us: u64,
204    pub fields: Vec<(String, String)>,
205}
206
207#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
208pub struct DataflowResult {
209    pub uuid: Uuid,
210    pub timestamp: uhlc::Timestamp,
211    pub node_results: BTreeMap<NodeId, Result<(), NodeError>>,
212}
213
214impl DataflowResult {
215    pub fn ok_empty(uuid: Uuid, timestamp: uhlc::Timestamp) -> Self {
216        Self {
217            uuid,
218            timestamp,
219            node_results: Default::default(),
220        }
221    }
222
223    pub fn is_ok(&self) -> bool {
224        self.node_results.values().all(|r| r.is_ok())
225    }
226}
227
228#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
229pub struct DataflowList(pub Vec<DataflowListEntry>);
230
231impl DataflowList {
232    pub fn get_active(&self) -> Vec<DataflowIdAndName> {
233        self.0
234            .iter()
235            .filter(|d| d.status == DataflowStatus::Running)
236            .map(|d| d.id.clone())
237            .collect()
238    }
239}
240
241#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
242pub struct DataflowListEntry {
243    pub id: DataflowIdAndName,
244    pub status: DataflowStatus,
245}
246
247/// A dataflow that `dora clean` failed to remove from the persisted
248/// store. Reported alongside the successful list so the CLI can show
249/// the user what didn't get cleaned and exit non-zero.
250#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
251pub struct CleanFailure {
252    pub id: DataflowIdAndName,
253    /// Human-readable error from the persisted store (e.g. an
254    /// underlying redb / I/O failure). The in-memory entry is
255    /// preserved so a later `dora clean` can retry.
256    pub error: String,
257}
258
259#[derive(Debug, Clone, Copy, serde::Deserialize, serde::Serialize, PartialEq, Eq)]
260pub enum DataflowStatus {
261    Running,
262    Finished,
263    Failed,
264}
265
266#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
267pub struct DataflowIdAndName {
268    pub uuid: Uuid,
269    pub name: Option<String>,
270}
271
272impl std::fmt::Display for DataflowIdAndName {
273    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
274        if let Some(name) = &self.name {
275            write!(f, "[{name}] {}", self.uuid)
276        } else {
277            write!(f, "[<unnamed>] {}", self.uuid)
278        }
279    }
280}
dora_message/coordinator_to_cli.rs

dora_message/
coordinator_to_cli.rs
