datum_agent/dcp/
proto.rs

1use std::collections::HashMap;
2
3use prost::Message as ProstMessage;
4
5/// DCP protocol version implemented by this crate.
6///
7/// Minor-version bumps are additive within one DCP major: new request variants
8/// and payload fields may be added without changing the major. Servers keep
9/// rejecting unknown major versions during `Hello` negotiation.
10pub const DCP_PROTOCOL_VERSION: &str = "0.11.0";
11
12/// DCP major version accepted by this crate.
13pub const DCP_PROTOCOL_MAJOR: u32 = 0;
14
15/// Client role advertised in the initial DCP hello.
16#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, prost::Enumeration)]
17#[repr(i32)]
18pub enum ClientKind {
19    Unspecified = 0,
20    Cli = 1,
21    Tui = 2,
22    Daemon = 3,
23    ClusterNode = 4,
24}
25
26/// Response status for unary DCP requests and hello negotiation.
27#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, prost::Enumeration)]
28#[repr(i32)]
29pub enum ResponseStatus {
30    Ok = 0,
31    BadRequest = 1,
32    Unauthorized = 2,
33    NotFound = 3,
34    Conflict = 4,
35    Failed = 5,
36    ProtocolMismatch = 6,
37    DeadlineExceeded = 7,
38}
39
40/// Authentication data carried in the hello. Remote deployments are expected to
41/// rely on QUIC mTLS first; this is an optional operator token hook.
42#[derive(Clone, PartialEq, ProstMessage)]
43pub struct Auth {
44    #[prost(string, tag = "1")]
45    pub bearer_token: String,
46}
47
48/// First client-to-server frame on every DCP stream.
49#[derive(Clone, PartialEq, ProstMessage)]
50pub struct Hello {
51    #[prost(string, tag = "1")]
52    pub protocol_version: String,
53    #[prost(string, tag = "2")]
54    pub node_id: String,
55    #[prost(enumeration = "ClientKind", tag = "3")]
56    pub client_kind: i32,
57    #[prost(message, optional, tag = "4")]
58    pub auth: Option<Auth>,
59    #[prost(string, repeated, tag = "5")]
60    pub capabilities: Vec<String>,
61}
62
63impl Hello {
64    #[must_use]
65    pub fn new(node_id: impl Into<String>, client_kind: ClientKind) -> Self {
66        Self {
67            protocol_version: DCP_PROTOCOL_VERSION.to_owned(),
68            node_id: node_id.into(),
69            client_kind: client_kind as i32,
70            auth: None,
71            capabilities: Vec::new(),
72        }
73    }
74}
75
76/// Unary or subscription request.
77#[derive(Clone, PartialEq, ProstMessage)]
78pub struct Request {
79    #[prost(uint64, tag = "1")]
80    pub request_id: u64,
81    #[prost(uint64, tag = "2")]
82    pub deadline_ms: u64,
83    #[prost(
84        oneof = "request::Command",
85        tags = "10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31"
86    )]
87    pub command: Option<request::Command>,
88}
89
90pub mod request {
91    #[derive(Clone, PartialEq, prost::Oneof)]
92    pub enum Command {
93        #[prost(message, tag = "10")]
94        ListJobs(super::ListJobs),
95        #[prost(message, tag = "11")]
96        StartJob(super::StartJob),
97        #[prost(message, tag = "12")]
98        DrainJob(super::DrainJob),
99        #[prost(message, tag = "13")]
100        StopJob(super::StopJob),
101        #[prost(message, tag = "14")]
102        RestartJob(super::RestartJob),
103        #[prost(message, tag = "15")]
104        JobStatus(super::JobStatusRequest),
105        #[prost(message, tag = "16")]
106        SubscribeEvents(super::SubscribeEvents),
107        #[prost(message, tag = "17")]
108        SubscribeMetrics(super::SubscribeMetrics),
109        #[prost(message, tag = "18")]
110        GetConfig(super::GetConfig),
111        #[prost(message, tag = "19")]
112        PutConfig(super::PutConfig),
113        #[prost(message, tag = "20")]
114        ListClusterJobs(super::ListClusterJobs),
115        #[prost(message, tag = "21")]
116        ClusterNodeInfo(super::ClusterNodeInfo),
117        #[prost(message, tag = "22")]
118        SubmitClusterJob(super::SubmitClusterJob),
119        #[prost(message, tag = "23")]
120        RememberClusterAssignment(super::RememberClusterAssignment),
121        #[prost(message, tag = "24")]
122        AllocateShard(super::ShardAllocationRequest),
123        #[prost(message, tag = "25")]
124        RememberShardAllocations(super::RememberShardAllocations),
125        #[prost(message, tag = "26")]
126        GetShardAllocations(super::GetShardAllocations),
127        #[prost(message, tag = "27")]
128        ForwardShardEnvelopes(super::ForwardShardEnvelopes),
129        #[prost(message, tag = "28")]
130        CompleteShardingAsk(super::CompleteShardingAsk),
131        #[prost(message, tag = "29")]
132        OpenShardPipe(super::OpenShardPipe),
133        #[prost(message, tag = "30")]
134        UnsubscribeMetrics(super::UnsubscribeMetrics),
135        #[prost(message, tag = "31")]
136        SubscribeClusterEvents(super::SubscribeClusterEvents),
137    }
138}
139
140#[derive(Clone, PartialEq, ProstMessage)]
141pub struct ListJobs {}
142
143/// StartJob uses the v0.9.1 registered-factory model.
144///
145/// Dynamic blueprint upload is intentionally out of scope until cluster
146/// placement work in v0.10.
147#[derive(Clone, PartialEq, ProstMessage)]
148pub struct StartJob {
149    #[prost(string, tag = "1")]
150    pub factory_name: String,
151    #[prost(string, tag = "2")]
152    pub instance_name: String,
153    #[prost(map = "string, string", tag = "3")]
154    pub params: HashMap<String, String>,
155    #[prost(message, optional, tag = "4")]
156    pub cluster: Option<ClusterJobStart>,
157}
158
159#[derive(Clone, PartialEq, ProstMessage)]
160pub struct DrainJob {
161    #[prost(string, tag = "1")]
162    pub name: String,
163    #[prost(bool, tag = "2")]
164    pub cluster: bool,
165}
166
167#[derive(Clone, PartialEq, ProstMessage)]
168pub struct StopJob {
169    #[prost(string, tag = "1")]
170    pub name: String,
171    #[prost(bool, tag = "2")]
172    pub cluster: bool,
173}
174
175#[derive(Clone, PartialEq, ProstMessage)]
176pub struct RestartJob {
177    #[prost(string, tag = "1")]
178    pub name: String,
179    #[prost(bool, tag = "2")]
180    pub cluster: bool,
181}
182
183#[derive(Clone, PartialEq, ProstMessage)]
184pub struct JobStatusRequest {
185    #[prost(string, tag = "1")]
186    pub name: String,
187    #[prost(bool, tag = "2")]
188    pub cluster: bool,
189}
190
191#[derive(Clone, PartialEq, ProstMessage)]
192pub struct SubscribeEvents {
193    #[prost(uint64, tag = "1")]
194    pub buffer: u64,
195}
196
197#[derive(Clone, PartialEq, ProstMessage)]
198pub struct SubscribeClusterEvents {}
199
200#[derive(Clone, PartialEq, ProstMessage)]
201pub struct SubscribeMetrics {
202    #[prost(uint64, tag = "1")]
203    pub interval_ms: u64,
204    /// Restrict samples to instrumentation names owned by these jobs. Empty
205    /// means every local stream.
206    #[prost(string, repeated, tag = "2")]
207    pub job_names: Vec<String>,
208    /// Sample only this server's instrumentation registry. Cluster node
209    /// sessions set this to prevent recursive proxying.
210    #[prost(bool, tag = "3")]
211    pub local_only: bool,
212}
213
214#[derive(Clone, PartialEq, ProstMessage)]
215pub struct UnsubscribeMetrics {
216    #[prost(uint64, tag = "1")]
217    pub subscription_id: u64,
218}
219
220#[derive(Clone, PartialEq, ProstMessage)]
221pub struct GetConfig {
222    #[prost(string, tag = "1")]
223    pub key: String,
224}
225
226#[derive(Clone, PartialEq, ProstMessage)]
227pub struct PutConfig {
228    #[prost(string, tag = "1")]
229    pub key: String,
230    #[prost(string, tag = "2")]
231    pub value: String,
232}
233
234/// Aggregate job snapshots across the cluster.
235///
236/// The server fans out to currently known peer node sessions and returns
237/// partial results with per-node errors when a peer is unreachable or the
238/// bounded wait expires.
239#[derive(Clone, PartialEq, ProstMessage)]
240pub struct ListClusterJobs {
241    /// Per-peer fan-out timeout in milliseconds. A zero value uses the server
242    /// default.
243    #[prost(uint64, tag = "1")]
244    pub timeout_ms: u64,
245}
246
247/// Return the local node's cluster membership and DCP session view.
248#[derive(Clone, PartialEq, ProstMessage)]
249pub struct ClusterNodeInfo {
250    /// Snapshot collection timeout in milliseconds. A zero value uses the
251    /// server default.
252    #[prost(uint64, tag = "1")]
253    pub timeout_ms: u64,
254}
255
256/// Submit a registered-factory job to the placement coordinator.
257#[derive(Clone, PartialEq, ProstMessage)]
258pub struct SubmitClusterJob {
259    #[prost(string, tag = "1")]
260    pub factory_name: String,
261    #[prost(string, tag = "2")]
262    pub instance_name: String,
263    #[prost(map = "string, string", tag = "3")]
264    pub params: HashMap<String, String>,
265    #[prost(message, optional, tag = "4")]
266    pub placement: Option<PlacementSpec>,
267    /// Coordinator/placement timeout in milliseconds. A zero value uses the
268    /// server default.
269    #[prost(uint64, tag = "5")]
270    pub timeout_ms: u64,
271}
272
273/// Internal best-effort replication of coordinator assignment state to peers.
274/// A present assignment upserts intent; an absent assignment removes it.
275#[derive(Clone, PartialEq, ProstMessage)]
276pub struct RememberClusterAssignment {
277    #[prost(string, tag = "1")]
278    pub instance_name: String,
279    #[prost(message, optional, tag = "2")]
280    pub assignment: Option<ClusterJobStart>,
281    /// Placement generation being removed when `assignment` is absent.
282    #[prost(uint64, tag = "3")]
283    pub tombstone_placement_generation: u64,
284    /// Coordinator that issued the removal when `assignment` is absent.
285    #[prost(string, tag = "4")]
286    pub tombstone_coordinator_node_id: String,
287}
288
289/// Request the coordinator allocation for one shard.
290#[derive(Clone, PartialEq, ProstMessage)]
291pub struct ShardAllocationRequest {
292    #[prost(string, tag = "1")]
293    pub type_name: String,
294    #[prost(string, tag = "2")]
295    pub shard_id: String,
296    /// Coordinator timeout in milliseconds. A zero value uses the server
297    /// default.
298    #[prost(uint64, tag = "3")]
299    pub timeout_ms: u64,
300}
301
302/// Coordinator answer for one shard home.
303#[derive(Clone, PartialEq, ProstMessage)]
304pub struct ShardAllocation {
305    #[prost(string, tag = "1")]
306    pub type_name: String,
307    #[prost(string, tag = "2")]
308    pub shard_id: String,
309    #[prost(string, tag = "3")]
310    pub node_id: String,
311    #[prost(uint64, tag = "4")]
312    pub generation: u64,
313}
314
315/// Allocation-table row replicated to peer regions.
316#[derive(Clone, PartialEq, ProstMessage)]
317pub struct ShardAllocationEntry {
318    #[prost(string, tag = "1")]
319    pub type_name: String,
320    #[prost(string, tag = "2")]
321    pub shard_id: String,
322    #[prost(string, tag = "3")]
323    pub node_id: String,
324    #[prost(uint64, tag = "4")]
325    pub generation: u64,
326}
327
328/// Allocation-table snapshot. v0.10 keeps this in memory; persistence and
329/// remember-entities belong to later sharding work packages.
330#[derive(Clone, PartialEq, ProstMessage)]
331pub struct ShardAllocationTable {
332    #[prost(message, repeated, tag = "1")]
333    pub entries: Vec<ShardAllocationEntry>,
334}
335
336/// Best-effort allocation-table replication from the active coordinator.
337#[derive(Clone, PartialEq, ProstMessage)]
338pub struct RememberShardAllocations {
339    #[prost(message, optional, tag = "1")]
340    pub table: Option<ShardAllocationTable>,
341}
342
343/// Ask a peer region for its cached allocation table.
344#[derive(Clone, PartialEq, ProstMessage)]
345pub struct GetShardAllocations {
346    #[prost(string, tag = "1")]
347    pub type_name: String,
348}
349
350/// One serialized entity envelope carried between peer regions.
351#[derive(Clone, PartialEq, ProstMessage)]
352pub struct ShardEnvelopeWire {
353    #[prost(string, tag = "1")]
354    pub entity_id: String,
355    #[prost(string, tag = "2")]
356    pub shard_id: String,
357    #[prost(bytes = "vec", tag = "3")]
358    pub payload: Vec<u8>,
359}
360
361/// Batched peer-region forwarding on the existing C2 session.
362#[derive(Clone, PartialEq, ProstMessage)]
363pub struct ForwardShardEnvelopes {
364    #[prost(string, tag = "1")]
365    pub type_name: String,
366    #[prost(message, repeated, tag = "2")]
367    pub envelopes: Vec<ShardEnvelopeWire>,
368}
369
370/// Per-envelope enqueue result for a forwarded batch.
371#[derive(Clone, PartialEq, ProstMessage)]
372pub struct ShardEnvelopeAck {
373    #[prost(string, tag = "1")]
374    pub entity_id: String,
375    #[prost(string, tag = "2")]
376    pub shard_id: String,
377    #[prost(bool, tag = "3")]
378    pub ok: bool,
379    #[prost(string, tag = "4")]
380    pub message: String,
381}
382
383/// Result for a peer-region forwarding batch.
384#[derive(Clone, PartialEq, ProstMessage)]
385pub struct ShardEnvelopeBatchResult {
386    #[prost(message, repeated, tag = "1")]
387    pub acks: Vec<ShardEnvelopeAck>,
388}
389
390/// Complete an ask whose reply port originated on another region.
391#[derive(Clone, PartialEq, ProstMessage)]
392pub struct CompleteShardingAsk {
393    #[prost(uint64, tag = "1")]
394    pub request_id: u64,
395    #[prost(bool, tag = "2")]
396    pub ok: bool,
397    #[prost(bytes = "vec", tag = "3")]
398    pub payload: Vec<u8>,
399    #[prost(string, tag = "4")]
400    pub message: String,
401}
402
403/// Switch this DCP connection to the sharding hot-path pipe after hello.
404#[derive(Clone, PartialEq, ProstMessage)]
405pub struct OpenShardPipe {}
406
407/// One sharding hot-path pipe frame.
408///
409/// Frames are one-way and ordered per peer connection. The sender batches only
410/// when its local pipe queue already has backlog; idle sends are flushed
411/// immediately so p50 does not pay a linger tax.
412#[derive(Clone, PartialEq, ProstMessage)]
413pub struct ShardPipeFrame {
414    #[prost(message, repeated, tag = "1")]
415    pub forwards: Vec<ForwardShardEnvelopes>,
416    #[prost(message, repeated, tag = "2")]
417    pub replies: Vec<CompleteShardingAsk>,
418}
419
420#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, prost::Enumeration)]
421#[repr(i32)]
422pub enum PlacementStrategy {
423    LeastJobs = 0,
424    Pinned = 1,
425}
426
427#[derive(Clone, PartialEq, ProstMessage)]
428pub struct PlacementSpec {
429    #[prost(string, tag = "1")]
430    pub role_constraint: String,
431    #[prost(enumeration = "PlacementStrategy", tag = "2")]
432    pub strategy: i32,
433    #[prost(string, tag = "3")]
434    pub pinned_node_id: String,
435}
436
437#[derive(Clone, PartialEq, ProstMessage)]
438pub struct ClusterPlacementHistory {
439    #[prost(uint64, tag = "1")]
440    pub generation: u64,
441    #[prost(string, tag = "2")]
442    pub from_node_id: String,
443    #[prost(string, tag = "3")]
444    pub to_node_id: String,
445    #[prost(string, tag = "4")]
446    pub reason: String,
447    #[prost(uint64, tag = "5")]
448    pub timestamp_ms: u64,
449}
450
451/// Cluster metadata carried on coordinator-issued StartJob.
452#[derive(Clone, PartialEq, ProstMessage)]
453pub struct ClusterJobStart {
454    #[prost(string, tag = "1")]
455    pub factory_name: String,
456    #[prost(map = "string, string", tag = "2")]
457    pub params: HashMap<String, String>,
458    #[prost(message, optional, tag = "3")]
459    pub placement: Option<PlacementSpec>,
460    #[prost(string, tag = "4")]
461    pub coordinator_node_id: String,
462    #[prost(string, tag = "5")]
463    pub assigned_node_id: String,
464    #[prost(uint64, tag = "6")]
465    pub placement_generation: u64,
466    #[prost(message, repeated, tag = "7")]
467    pub history: Vec<ClusterPlacementHistory>,
468}
469
470#[derive(Clone, PartialEq, ProstMessage)]
471pub struct Response {
472    #[prost(uint64, tag = "1")]
473    pub request_id: u64,
474    #[prost(enumeration = "ResponseStatus", tag = "2")]
475    pub status: i32,
476    #[prost(string, tag = "3")]
477    pub message: String,
478    #[prost(bytes = "vec", tag = "4")]
479    pub payload: Vec<u8>,
480}
481
482impl Response {
483    #[must_use]
484    pub fn ok(request_id: u64, payload: Vec<u8>) -> Self {
485        Self {
486            request_id,
487            status: ResponseStatus::Ok as i32,
488            message: String::new(),
489            payload,
490        }
491    }
492
493    #[must_use]
494    pub fn error(request_id: u64, status: ResponseStatus, message: impl Into<String>) -> Self {
495        Self {
496            request_id,
497            status: status as i32,
498            message: message.into(),
499            payload: Vec::new(),
500        }
501    }
502
503    #[must_use]
504    pub fn response_status(&self) -> ResponseStatus {
505        ResponseStatus::try_from(self.status).unwrap_or(ResponseStatus::Failed)
506    }
507}
508
509#[derive(Clone, PartialEq, ProstMessage)]
510pub struct JobList {
511    #[prost(message, repeated, tag = "1")]
512    pub jobs: Vec<JobStatus>,
513}
514
515/// Cluster-wide job list response.
516#[derive(Clone, PartialEq, ProstMessage)]
517pub struct ClusterJobList {
518    #[prost(message, repeated, tag = "1")]
519    pub nodes: Vec<ClusterJobNode>,
520    #[prost(message, repeated, tag = "2")]
521    pub errors: Vec<ClusterNodeError>,
522    #[prost(bool, tag = "3")]
523    pub partial: bool,
524}
525
526/// Jobs reported by one cluster node.
527#[derive(Clone, PartialEq, ProstMessage)]
528pub struct ClusterJobNode {
529    #[prost(string, tag = "1")]
530    pub node_id: String,
531    #[prost(string, tag = "2")]
532    pub address: String,
533    #[prost(bool, tag = "3")]
534    pub local: bool,
535    #[prost(message, repeated, tag = "4")]
536    pub jobs: Vec<JobStatus>,
537}
538
539/// Cluster node information response.
540#[derive(Clone, PartialEq, ProstMessage)]
541pub struct ClusterNodeList {
542    #[prost(message, repeated, tag = "1")]
543    pub nodes: Vec<ClusterNodeStatus>,
544    #[prost(message, repeated, tag = "2")]
545    pub errors: Vec<ClusterNodeError>,
546    #[prost(bool, tag = "3")]
547    pub partial: bool,
548    #[prost(string, tag = "4")]
549    pub coordinator_node_id: String,
550}
551
552/// Membership and DCP session state for one node.
553#[derive(Clone, PartialEq, ProstMessage)]
554pub struct ClusterNodeStatus {
555    #[prost(string, tag = "1")]
556    pub node_id: String,
557    #[prost(string, tag = "2")]
558    pub member_state: String,
559    #[prost(string, tag = "3")]
560    pub address: String,
561    #[prost(string, tag = "4")]
562    pub agent_addr: String,
563    #[prost(string, repeated, tag = "5")]
564    pub roles: Vec<String>,
565    #[prost(bool, tag = "6")]
566    pub unreachable: bool,
567    #[prost(bool, tag = "7")]
568    pub local: bool,
569    #[prost(string, tag = "8")]
570    pub session_state: String,
571}
572
573/// Per-node partial-result error for cluster fan-out requests.
574#[derive(Clone, PartialEq, ProstMessage)]
575pub struct ClusterNodeError {
576    #[prost(string, tag = "1")]
577    pub node_id: String,
578    #[prost(string, tag = "2")]
579    pub message: String,
580}
581
582#[derive(Clone, PartialEq, ProstMessage)]
583pub struct ConfigValue {
584    #[prost(string, tag = "1")]
585    pub key: String,
586    #[prost(string, tag = "2")]
587    pub value: String,
588    #[prost(bool, tag = "3")]
589    pub existed: bool,
590}
591
592#[derive(Clone, PartialEq, ProstMessage)]
593pub struct JobStatus {
594    #[prost(string, tag = "1")]
595    pub name: String,
596    #[prost(uint64, tag = "2")]
597    pub job_id: u64,
598    #[prost(string, tag = "3")]
599    pub state: String,
600    #[prost(string, tag = "4")]
601    pub desired_state: String,
602    #[prost(uint64, tag = "5")]
603    pub generation: u64,
604    #[prost(uint64, tag = "6")]
605    pub starts_total: u64,
606    #[prost(uint64, tag = "7")]
607    pub restarts_total: u64,
608    #[prost(uint64, optional, tag = "8")]
609    pub last_start_at_ms: Option<u64>,
610    #[prost(uint64, optional, tag = "9")]
611    pub last_exit_at_ms: Option<u64>,
612    #[prost(string, tag = "10")]
613    pub last_exit_reason: String,
614    #[prost(uint64, optional, tag = "11")]
615    pub backoff_remaining_ms: Option<u64>,
616    #[prost(uint64, optional, tag = "12")]
617    pub drain_remaining_ms: Option<u64>,
618    #[prost(bool, tag = "13")]
619    pub drain_supported: bool,
620    #[prost(uint64, optional, tag = "14")]
621    pub active_streams: Option<u64>,
622    #[prost(bool, tag = "15")]
623    pub cluster_job: bool,
624    #[prost(string, tag = "16")]
625    pub factory_name: String,
626    #[prost(message, optional, tag = "17")]
627    pub placement: Option<PlacementSpec>,
628    #[prost(string, tag = "18")]
629    pub coordinator_node_id: String,
630    #[prost(string, tag = "19")]
631    pub placement_node_id: String,
632    #[prost(uint64, tag = "20")]
633    pub placement_generation: u64,
634    #[prost(message, repeated, tag = "21")]
635    pub placement_history: Vec<ClusterPlacementHistory>,
636    #[prost(map = "string, string", tag = "22")]
637    pub params: HashMap<String, String>,
638}
639
640#[derive(Clone, PartialEq, ProstMessage)]
641pub struct EventFrame {
642    #[prost(uint64, tag = "1")]
643    pub subscription_id: u64,
644    #[prost(message, optional, tag = "2")]
645    pub event: Option<Event>,
646}
647
648#[derive(Clone, PartialEq, ProstMessage)]
649pub struct Event {
650    #[prost(uint64, tag = "1")]
651    pub sequence: u64,
652    #[prost(uint64, tag = "2")]
653    pub timestamp_ms: u64,
654    #[prost(string, tag = "3")]
655    pub name: String,
656    #[prost(uint64, tag = "4")]
657    pub job_id: u64,
658    #[prost(uint64, tag = "5")]
659    pub generation: u64,
660    #[prost(string, tag = "6")]
661    pub kind: String,
662    #[prost(string, tag = "7")]
663    pub detail: String,
664}
665
666#[derive(Clone, PartialEq, ProstMessage)]
667pub struct ClusterEventFrame {
668    #[prost(uint64, tag = "1")]
669    pub subscription_id: u64,
670    #[prost(message, optional, tag = "2")]
671    pub event: Option<ClusterEvent>,
672}
673
674#[derive(Clone, PartialEq, ProstMessage)]
675pub struct ClusterEvent {
676    #[prost(uint64, tag = "1")]
677    pub sequence: u64,
678    #[prost(uint64, tag = "2")]
679    pub timestamp_ms: u64,
680    #[prost(string, tag = "3")]
681    pub kind: String,
682    #[prost(string, tag = "4")]
683    pub node_id: String,
684    #[prost(string, tag = "5")]
685    pub detail: String,
686    #[prost(uint64, optional, tag = "6")]
687    pub generation: Option<u64>,
688}
689
690#[derive(Clone, PartialEq, ProstMessage)]
691pub struct MetricFrame {
692    #[prost(uint64, tag = "1")]
693    pub subscription_id: u64,
694    #[prost(message, optional, tag = "2")]
695    pub sample: Option<MetricSample>,
696}
697
698#[derive(Clone, PartialEq, ProstMessage)]
699pub struct MetricSample {
700    #[prost(uint64, tag = "1")]
701    pub timestamp_ms: u64,
702    #[prost(message, repeated, tag = "2")]
703    pub streams: Vec<StreamMetric>,
704    #[prost(message, repeated, tag = "3")]
705    pub nodes: Vec<NodeMetric>,
706}
707
708/// Process resource telemetry for one Datum agent node.
709#[derive(Clone, PartialEq, ProstMessage)]
710pub struct NodeMetric {
711    #[prost(string, tag = "1")]
712    pub node_id: String,
713    #[prost(double, tag = "2")]
714    pub cpu_percent: f64,
715    #[prost(uint64, tag = "3")]
716    pub cpu_ms: u64,
717    #[prost(uint64, tag = "4")]
718    pub rss_bytes: u64,
719    #[prost(uint64, optional, tag = "5")]
720    pub mem_total_bytes: Option<u64>,
721    #[prost(uint32, tag = "6")]
722    pub threads: u32,
723    #[prost(uint64, tag = "7")]
724    pub sample_ts_ms: u64,
725}
726
727#[derive(Clone, PartialEq, ProstMessage)]
728pub struct StreamMetric {
729    #[prost(uint64, tag = "1")]
730    pub id: u64,
731    #[prost(string, tag = "2")]
732    pub name: String,
733    #[prost(uint64, tag = "3")]
734    pub elements_through: u64,
735    #[prost(uint64, tag = "4")]
736    pub restarts: u64,
737    #[prost(string, tag = "5")]
738    pub state: String,
739    #[prost(uint64, tag = "6")]
740    pub started_at_ms: u64,
741    #[prost(uint64, tag = "7")]
742    pub state_changed_at_ms: u64,
743    #[prost(uint64, optional, tag = "8")]
744    pub finished_at_ms: Option<u64>,
745    #[prost(uint64, tag = "9")]
746    pub uptime_ms: u64,
747}
748
749#[derive(Clone, PartialEq, ProstMessage)]
750pub struct DcpFrame {
751    #[prost(oneof = "dcp_frame::Frame", tags = "1, 2, 3, 4, 5, 6, 7")]
752    pub frame: Option<dcp_frame::Frame>,
753}
754
755impl DcpFrame {
756    #[must_use]
757    pub fn hello(hello: Hello) -> Self {
758        Self {
759            frame: Some(dcp_frame::Frame::Hello(hello)),
760        }
761    }
762
763    #[must_use]
764    pub fn request(request: Request) -> Self {
765        Self {
766            frame: Some(dcp_frame::Frame::Request(request)),
767        }
768    }
769
770    #[must_use]
771    pub fn response(response: Response) -> Self {
772        Self {
773            frame: Some(dcp_frame::Frame::Response(response)),
774        }
775    }
776
777    #[must_use]
778    pub fn event(subscription_id: u64, event: Event) -> Self {
779        Self {
780            frame: Some(dcp_frame::Frame::Event(EventFrame {
781                subscription_id,
782                event: Some(event),
783            })),
784        }
785    }
786
787    #[must_use]
788    pub fn cluster_event(subscription_id: u64, event: ClusterEvent) -> Self {
789        Self {
790            frame: Some(dcp_frame::Frame::ClusterEvent(ClusterEventFrame {
791                subscription_id,
792                event: Some(event),
793            })),
794        }
795    }
796
797    #[must_use]
798    pub fn metric(subscription_id: u64, sample: MetricSample) -> Self {
799        Self {
800            frame: Some(dcp_frame::Frame::Metric(MetricFrame {
801                subscription_id,
802                sample: Some(sample),
803            })),
804        }
805    }
806
807    #[must_use]
808    pub fn shard_pipe(pipe: ShardPipeFrame) -> Self {
809        Self {
810            frame: Some(dcp_frame::Frame::ShardPipe(pipe)),
811        }
812    }
813}
814
815pub mod dcp_frame {
816    #[allow(clippy::large_enum_variant)]
817    #[derive(Clone, PartialEq, prost::Oneof)]
818    pub enum Frame {
819        #[prost(message, tag = "1")]
820        Hello(super::Hello),
821        #[prost(message, tag = "2")]
822        Request(super::Request),
823        #[prost(message, tag = "3")]
824        Response(super::Response),
825        #[prost(message, tag = "4")]
826        Event(super::EventFrame),
827        #[prost(message, tag = "5")]
828        Metric(super::MetricFrame),
829        #[prost(message, tag = "6")]
830        ShardPipe(super::ShardPipeFrame),
831        #[prost(message, tag = "7")]
832        ClusterEvent(super::ClusterEventFrame),
833    }
834}
835
836#[cfg(test)]
837mod tests {
838    use super::*;
839
840    #[test]
841    fn node_metric_round_trips_on_the_wire() {
842        let sample = MetricSample {
843            timestamp_ms: 1_720_000_000_000,
844            streams: Vec::new(),
845            nodes: vec![NodeMetric {
846                node_id: "node-1".to_owned(),
847                cpu_percent: 37.5,
848                cpu_ms: 12_345,
849                rss_bytes: 512 * 1024 * 1024,
850                mem_total_bytes: Some(16 * 1024 * 1024 * 1024),
851                threads: 9,
852                sample_ts_ms: 1_720_000_000_000,
853            }],
854        };
855
856        let decoded = MetricSample::decode(sample.encode_to_vec().as_slice())
857            .expect("metric sample should decode");
858        assert_eq!(decoded, sample);
859    }
860
861    #[test]
862    fn cluster_event_round_trips_on_the_wire() {
863        let frame = DcpFrame::cluster_event(
864            42,
865            ClusterEvent {
866                sequence: 7,
867                timestamp_ms: 1_720_000_000_000,
868                kind: "JobReplaced".to_owned(),
869                node_id: "node-2".to_owned(),
870                detail: "job ingest moved node-1 -> node-2".to_owned(),
871                generation: Some(3),
872            },
873        );
874
875        let decoded = DcpFrame::decode(frame.encode_to_vec().as_slice())
876            .expect("cluster event frame should decode");
877        assert_eq!(decoded, frame);
878    }
879}
datum_agent/dcp/proto.rs

datum_agent/dcp/
proto.rs
