datum_agent/dcp/
proto.rs

1use std::collections::HashMap;
2
3use prost::Message as ProstMessage;
4
5/// DCP protocol version implemented by this crate.
6pub const DCP_PROTOCOL_VERSION: &str = "0.9.1";
7
8/// DCP major version accepted by this crate.
9pub const DCP_PROTOCOL_MAJOR: u32 = 0;
10
11/// Client role advertised in the initial DCP hello.
12#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, prost::Enumeration)]
13#[repr(i32)]
14pub enum ClientKind {
15    Unspecified = 0,
16    Cli = 1,
17    Tui = 2,
18    Daemon = 3,
19    ClusterNode = 4,
20}
21
22/// Response status for unary DCP requests and hello negotiation.
23#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, prost::Enumeration)]
24#[repr(i32)]
25pub enum ResponseStatus {
26    Ok = 0,
27    BadRequest = 1,
28    Unauthorized = 2,
29    NotFound = 3,
30    Conflict = 4,
31    Failed = 5,
32    ProtocolMismatch = 6,
33    DeadlineExceeded = 7,
34}
35
36/// Authentication data carried in the hello. Remote deployments are expected to
37/// rely on QUIC mTLS first; this is an optional operator token hook.
38#[derive(Clone, PartialEq, ProstMessage)]
39pub struct Auth {
40    #[prost(string, tag = "1")]
41    pub bearer_token: String,
42}
43
44/// First client-to-server frame on every DCP stream.
45#[derive(Clone, PartialEq, ProstMessage)]
46pub struct Hello {
47    #[prost(string, tag = "1")]
48    pub protocol_version: String,
49    #[prost(string, tag = "2")]
50    pub node_id: String,
51    #[prost(enumeration = "ClientKind", tag = "3")]
52    pub client_kind: i32,
53    #[prost(message, optional, tag = "4")]
54    pub auth: Option<Auth>,
55    #[prost(string, repeated, tag = "5")]
56    pub capabilities: Vec<String>,
57}
58
59impl Hello {
60    #[must_use]
61    pub fn new(node_id: impl Into<String>, client_kind: ClientKind) -> Self {
62        Self {
63            protocol_version: DCP_PROTOCOL_VERSION.to_owned(),
64            node_id: node_id.into(),
65            client_kind: client_kind as i32,
66            auth: None,
67            capabilities: Vec::new(),
68        }
69    }
70}
71
72/// Unary or subscription request.
73#[derive(Clone, PartialEq, ProstMessage)]
74pub struct Request {
75    #[prost(uint64, tag = "1")]
76    pub request_id: u64,
77    #[prost(uint64, tag = "2")]
78    pub deadline_ms: u64,
79    #[prost(
80        oneof = "request::Command",
81        tags = "10, 11, 12, 13, 14, 15, 16, 17, 18, 19"
82    )]
83    pub command: Option<request::Command>,
84}
85
86pub mod request {
87    #[derive(Clone, PartialEq, prost::Oneof)]
88    pub enum Command {
89        #[prost(message, tag = "10")]
90        ListJobs(super::ListJobs),
91        #[prost(message, tag = "11")]
92        StartJob(super::StartJob),
93        #[prost(message, tag = "12")]
94        DrainJob(super::DrainJob),
95        #[prost(message, tag = "13")]
96        StopJob(super::StopJob),
97        #[prost(message, tag = "14")]
98        RestartJob(super::RestartJob),
99        #[prost(message, tag = "15")]
100        JobStatus(super::JobStatusRequest),
101        #[prost(message, tag = "16")]
102        SubscribeEvents(super::SubscribeEvents),
103        #[prost(message, tag = "17")]
104        SubscribeMetrics(super::SubscribeMetrics),
105        #[prost(message, tag = "18")]
106        GetConfig(super::GetConfig),
107        #[prost(message, tag = "19")]
108        PutConfig(super::PutConfig),
109    }
110}
111
112#[derive(Clone, PartialEq, ProstMessage)]
113pub struct ListJobs {}
114
115/// StartJob uses the v0.9.1 registered-factory model.
116///
117/// Dynamic blueprint upload is intentionally out of scope until cluster
118/// placement work in v0.10.
119#[derive(Clone, PartialEq, ProstMessage)]
120pub struct StartJob {
121    #[prost(string, tag = "1")]
122    pub factory_name: String,
123    #[prost(string, tag = "2")]
124    pub instance_name: String,
125    #[prost(map = "string, string", tag = "3")]
126    pub params: HashMap<String, String>,
127}
128
129#[derive(Clone, PartialEq, ProstMessage)]
130pub struct DrainJob {
131    #[prost(string, tag = "1")]
132    pub name: String,
133}
134
135#[derive(Clone, PartialEq, ProstMessage)]
136pub struct StopJob {
137    #[prost(string, tag = "1")]
138    pub name: String,
139}
140
141#[derive(Clone, PartialEq, ProstMessage)]
142pub struct RestartJob {
143    #[prost(string, tag = "1")]
144    pub name: String,
145}
146
147#[derive(Clone, PartialEq, ProstMessage)]
148pub struct JobStatusRequest {
149    #[prost(string, tag = "1")]
150    pub name: String,
151}
152
153#[derive(Clone, PartialEq, ProstMessage)]
154pub struct SubscribeEvents {
155    #[prost(uint64, tag = "1")]
156    pub buffer: u64,
157}
158
159#[derive(Clone, PartialEq, ProstMessage)]
160pub struct SubscribeMetrics {
161    #[prost(uint64, tag = "1")]
162    pub interval_ms: u64,
163}
164
165#[derive(Clone, PartialEq, ProstMessage)]
166pub struct GetConfig {
167    #[prost(string, tag = "1")]
168    pub key: String,
169}
170
171#[derive(Clone, PartialEq, ProstMessage)]
172pub struct PutConfig {
173    #[prost(string, tag = "1")]
174    pub key: String,
175    #[prost(string, tag = "2")]
176    pub value: String,
177}
178
179#[derive(Clone, PartialEq, ProstMessage)]
180pub struct Response {
181    #[prost(uint64, tag = "1")]
182    pub request_id: u64,
183    #[prost(enumeration = "ResponseStatus", tag = "2")]
184    pub status: i32,
185    #[prost(string, tag = "3")]
186    pub message: String,
187    #[prost(bytes = "vec", tag = "4")]
188    pub payload: Vec<u8>,
189}
190
191impl Response {
192    #[must_use]
193    pub fn ok(request_id: u64, payload: Vec<u8>) -> Self {
194        Self {
195            request_id,
196            status: ResponseStatus::Ok as i32,
197            message: String::new(),
198            payload,
199        }
200    }
201
202    #[must_use]
203    pub fn error(request_id: u64, status: ResponseStatus, message: impl Into<String>) -> Self {
204        Self {
205            request_id,
206            status: status as i32,
207            message: message.into(),
208            payload: Vec::new(),
209        }
210    }
211
212    #[must_use]
213    pub fn response_status(&self) -> ResponseStatus {
214        ResponseStatus::try_from(self.status).unwrap_or(ResponseStatus::Failed)
215    }
216}
217
218#[derive(Clone, PartialEq, ProstMessage)]
219pub struct JobList {
220    #[prost(message, repeated, tag = "1")]
221    pub jobs: Vec<JobStatus>,
222}
223
224#[derive(Clone, PartialEq, ProstMessage)]
225pub struct ConfigValue {
226    #[prost(string, tag = "1")]
227    pub key: String,
228    #[prost(string, tag = "2")]
229    pub value: String,
230    #[prost(bool, tag = "3")]
231    pub existed: bool,
232}
233
234#[derive(Clone, PartialEq, ProstMessage)]
235pub struct JobStatus {
236    #[prost(string, tag = "1")]
237    pub name: String,
238    #[prost(uint64, tag = "2")]
239    pub job_id: u64,
240    #[prost(string, tag = "3")]
241    pub state: String,
242    #[prost(string, tag = "4")]
243    pub desired_state: String,
244    #[prost(uint64, tag = "5")]
245    pub generation: u64,
246    #[prost(uint64, tag = "6")]
247    pub starts_total: u64,
248    #[prost(uint64, tag = "7")]
249    pub restarts_total: u64,
250    #[prost(uint64, optional, tag = "8")]
251    pub last_start_at_ms: Option<u64>,
252    #[prost(uint64, optional, tag = "9")]
253    pub last_exit_at_ms: Option<u64>,
254    #[prost(string, tag = "10")]
255    pub last_exit_reason: String,
256    #[prost(uint64, optional, tag = "11")]
257    pub backoff_remaining_ms: Option<u64>,
258    #[prost(uint64, optional, tag = "12")]
259    pub drain_remaining_ms: Option<u64>,
260    #[prost(bool, tag = "13")]
261    pub drain_supported: bool,
262    #[prost(uint64, optional, tag = "14")]
263    pub active_streams: Option<u64>,
264}
265
266#[derive(Clone, PartialEq, ProstMessage)]
267pub struct EventFrame {
268    #[prost(uint64, tag = "1")]
269    pub subscription_id: u64,
270    #[prost(message, optional, tag = "2")]
271    pub event: Option<Event>,
272}
273
274#[derive(Clone, PartialEq, ProstMessage)]
275pub struct Event {
276    #[prost(uint64, tag = "1")]
277    pub sequence: u64,
278    #[prost(uint64, tag = "2")]
279    pub timestamp_ms: u64,
280    #[prost(string, tag = "3")]
281    pub name: String,
282    #[prost(uint64, tag = "4")]
283    pub job_id: u64,
284    #[prost(uint64, tag = "5")]
285    pub generation: u64,
286    #[prost(string, tag = "6")]
287    pub kind: String,
288    #[prost(string, tag = "7")]
289    pub detail: String,
290}
291
292#[derive(Clone, PartialEq, ProstMessage)]
293pub struct MetricFrame {
294    #[prost(uint64, tag = "1")]
295    pub subscription_id: u64,
296    #[prost(message, optional, tag = "2")]
297    pub sample: Option<MetricSample>,
298}
299
300#[derive(Clone, PartialEq, ProstMessage)]
301pub struct MetricSample {
302    #[prost(uint64, tag = "1")]
303    pub timestamp_ms: u64,
304    #[prost(message, repeated, tag = "2")]
305    pub streams: Vec<StreamMetric>,
306}
307
308#[derive(Clone, PartialEq, ProstMessage)]
309pub struct StreamMetric {
310    #[prost(uint64, tag = "1")]
311    pub id: u64,
312    #[prost(string, tag = "2")]
313    pub name: String,
314    #[prost(uint64, tag = "3")]
315    pub elements_through: u64,
316    #[prost(uint64, tag = "4")]
317    pub restarts: u64,
318    #[prost(string, tag = "5")]
319    pub state: String,
320    #[prost(uint64, tag = "6")]
321    pub started_at_ms: u64,
322    #[prost(uint64, tag = "7")]
323    pub state_changed_at_ms: u64,
324    #[prost(uint64, optional, tag = "8")]
325    pub finished_at_ms: Option<u64>,
326    #[prost(uint64, tag = "9")]
327    pub uptime_ms: u64,
328}
329
330#[derive(Clone, PartialEq, ProstMessage)]
331pub struct DcpFrame {
332    #[prost(oneof = "dcp_frame::Frame", tags = "1, 2, 3, 4, 5")]
333    pub frame: Option<dcp_frame::Frame>,
334}
335
336impl DcpFrame {
337    #[must_use]
338    pub fn hello(hello: Hello) -> Self {
339        Self {
340            frame: Some(dcp_frame::Frame::Hello(hello)),
341        }
342    }
343
344    #[must_use]
345    pub fn request(request: Request) -> Self {
346        Self {
347            frame: Some(dcp_frame::Frame::Request(request)),
348        }
349    }
350
351    #[must_use]
352    pub fn response(response: Response) -> Self {
353        Self {
354            frame: Some(dcp_frame::Frame::Response(response)),
355        }
356    }
357
358    #[must_use]
359    pub fn event(subscription_id: u64, event: Event) -> Self {
360        Self {
361            frame: Some(dcp_frame::Frame::Event(EventFrame {
362                subscription_id,
363                event: Some(event),
364            })),
365        }
366    }
367
368    #[must_use]
369    pub fn metric(subscription_id: u64, sample: MetricSample) -> Self {
370        Self {
371            frame: Some(dcp_frame::Frame::Metric(MetricFrame {
372                subscription_id,
373                sample: Some(sample),
374            })),
375        }
376    }
377}
378
379pub mod dcp_frame {
380    #[derive(Clone, PartialEq, prost::Oneof)]
381    pub enum Frame {
382        #[prost(message, tag = "1")]
383        Hello(super::Hello),
384        #[prost(message, tag = "2")]
385        Request(super::Request),
386        #[prost(message, tag = "3")]
387        Response(super::Response),
388        #[prost(message, tag = "4")]
389        Event(super::EventFrame),
390        #[prost(message, tag = "5")]
391        Metric(super::MetricFrame),
392    }
393}
datum_agent/dcp/proto.rs

datum_agent/dcp/
proto.rs
