ant_core/node/daemon/forward/
document.rs

1//! Building the document that actually goes to Elasticsearch.
2//!
3//! Field names here are not ours to choose: they are the beta index's mapping (V2-1016), and a
4//! mismatch means a field lands as dynamically-mapped text instead of the keyword the dashboards
5//! aggregate on. Two in particular read wrong at a glance and are right:
6//!
7//! - the time field is `@timestamp`, not `timestamp`;
8//! - the node's build is `binary_version`, while `version` and `commit` carry whatever ant-node
9//!   said about *itself* on its startup line. Keeping them separate avoids two half-populated
10//!   fields meaning the same thing with no rule for which wins.
11//!
12//! Two mapped fields are deliberately never sent. `host` is stripped by the ingest pipeline —
13//! machine hostnames routinely contain someone's name — and `beta_user` is stamped server-side from
14//! the authenticated API key, so anything we sent would be discarded and replaced anyway.
15
16use serde::Serialize;
17
18use super::tail::TailedEvent;
19use crate::node::types::NodeConfig;
20
21/// Value used for `channel` when a node has never been given an explicit upgrade channel.
22///
23/// Not folded into `"stable"`: the beta cohort is counted by aggregating this field, and claiming a
24/// node is on stable when nobody ever said so would quietly distort that count.
25const CHANNEL_UNSET: &str = "unset";
26
27/// The identity fields every event from a given node carries.
28///
29/// Resolved once when the forwarder picks the node up, rather than per event.
30#[derive(Debug, Clone, PartialEq, Eq)]
31pub struct NodeTags {
32    pub node_id: u32,
33    pub service: String,
34    pub binary_version: String,
35    pub channel: String,
36}
37
38impl NodeTags {
39    #[must_use]
40    pub fn from_config(config: &NodeConfig) -> Self {
41        Self {
42            node_id: config.id,
43            service: config.service_name.clone(),
44            binary_version: config.version.clone(),
45            channel: config
46                .upgrade_channel
47                .map_or_else(|| CHANNEL_UNSET.to_string(), |channel| channel.to_string()),
48        }
49    }
50}
51
52/// A document ready to be framed into a bulk request.
53#[derive(Debug, Clone, PartialEq, Eq)]
54pub struct ForwardDocument {
55    /// Deterministic `_id`, so replaying a batch is idempotent.
56    pub id: String,
57    /// Daily index, derived from this event's own timestamp.
58    pub index: String,
59    pub source: DocumentSource,
60}
61
62/// The `_source` body of a forwarded document.
63#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
64pub struct DocumentSource {
65    #[serde(rename = "@timestamp")]
66    pub timestamp: String,
67    pub level: String,
68    #[serde(skip_serializing_if = "Option::is_none")]
69    pub target: Option<String>,
70    pub message: String,
71
72    /// Keyword in the mapping, so it is sent as a string rather than a number.
73    pub node_id: String,
74    pub service: String,
75    pub binary_version: String,
76    pub channel: String,
77    pub os: String,
78    pub arch: String,
79
80    #[serde(skip_serializing_if = "Option::is_none")]
81    pub peer_id: Option<String>,
82    #[serde(skip_serializing_if = "Option::is_none")]
83    pub version: Option<String>,
84    #[serde(skip_serializing_if = "Option::is_none")]
85    pub commit: Option<String>,
86}
87
88impl ForwardDocument {
89    /// Build a document, or `None` if the event's timestamp cannot name an index.
90    ///
91    /// Parsing already rejects unusable timestamps in both layouts, so `None` here is a
92    /// belt-and-braces case rather than an expected one.
93    ///
94    /// `installation_id` namespaces the document id; see [`TailedEvent::document_id`] for why the
95    /// local position alone is not unique across the beta cohort.
96    #[must_use]
97    pub fn build(
98        tailed: &TailedEvent,
99        tags: &NodeTags,
100        index_prefix: &str,
101        installation_id: &str,
102    ) -> Option<Self> {
103        let index = format!("{index_prefix}-{}", tailed.event.index_date()?);
104
105        Some(Self {
106            id: tailed.document_id(installation_id),
107            index,
108            source: DocumentSource {
109                timestamp: tailed.event.timestamp.clone(),
110                level: tailed.event.level.as_str().to_string(),
111                target: tailed.event.target.clone(),
112                message: tailed.event.message.clone(),
113                node_id: tags.node_id.to_string(),
114                service: tags.service.clone(),
115                binary_version: tags.binary_version.clone(),
116                channel: tags.channel.clone(),
117                os: std::env::consts::OS.to_string(),
118                arch: std::env::consts::ARCH.to_string(),
119                peer_id: tailed.event.peer_id.clone(),
120                version: tailed.event.version.clone(),
121                commit: tailed.event.commit.clone(),
122            },
123        })
124    }
125
126    /// Approximate serialized size, used to keep a batch well under the endpoint's body cap.
127    #[must_use]
128    pub fn approx_bytes(&self) -> usize {
129        // The action line plus the source line, give or take the JSON punctuation.
130        self.id.len() + self.index.len() + self.source.message.len() + 256
131    }
132}
133
134#[cfg(test)]
135mod tests {
136    use super::*;
137    use crate::node::daemon::forward::parse::parse_line;
138    use crate::node::types::{EvmNetwork, UpgradeChannel};
139    use std::collections::HashMap;
140    use std::path::PathBuf;
141
142    fn node_config(channel: Option<UpgradeChannel>) -> NodeConfig {
143        NodeConfig {
144            id: 7,
145            service_name: "node7".to_string(),
146            rewards_address: "0xabc".to_string(),
147            data_dir: PathBuf::from("/data/node-7"),
148            log_dir: Some(PathBuf::from("/logs/node-7")),
149            node_port: None,
150            binary_path: PathBuf::from("/bin/antnode"),
151            version: "0.17.2-beta.1".to_string(),
152            env_variables: HashMap::new(),
153            bootstrap_peers: Vec::new(),
154            upgrade_channel: channel,
155            evm_network: EvmNetwork::default(),
156            eviction: None,
157        }
158    }
159
160    fn tailed(line: &str) -> TailedEvent {
161        TailedEvent {
162            node_id: 7,
163            file_name: "ant-node.2026-08-19.log".to_string(),
164            byte_offset: 4096,
165            event: parse_line(line).unwrap(),
166        }
167    }
168
169    const INSTALL: &str = "0123456789abcdef";
170
171    const LINE: &str =
172        "2026-08-19T20:50:00.123456Z  INFO ant_node::node: connected peer_id=12D3KooWabc";
173
174    #[test]
175    fn builds_a_document_with_the_mapped_field_names() {
176        let tags = NodeTags::from_config(&node_config(Some(UpgradeChannel::Beta)));
177        let document = ForwardDocument::build(&tailed(LINE), &tags, "beta-nodes", INSTALL).unwrap();
178
179        assert_eq!(
180            document.id,
181            "0123456789abcdef-7-ant-node.2026-08-19.log-4096"
182        );
183        assert_eq!(document.index, "beta-nodes-2026.08.19");
184
185        let json: serde_json::Value = serde_json::to_value(&document.source).unwrap();
186        assert_eq!(json["@timestamp"], "2026-08-19T20:50:00.123456Z");
187        assert_eq!(json["level"], "INFO");
188        assert_eq!(json["target"], "ant_node::node");
189        assert_eq!(json["node_id"], "7");
190        assert_eq!(json["service"], "node7");
191        assert_eq!(json["binary_version"], "0.17.2-beta.1");
192        assert_eq!(json["channel"], "beta");
193        assert_eq!(json["peer_id"], "12D3KooWabc");
194        assert_eq!(json["os"], std::env::consts::OS);
195        assert_eq!(json["arch"], std::env::consts::ARCH);
196    }
197
198    /// The time field is `@timestamp`; a document using `timestamp` would not be searchable by time.
199    #[test]
200    fn the_time_field_is_at_timestamp_and_nothing_else() {
201        let tags = NodeTags::from_config(&node_config(None));
202        let document = ForwardDocument::build(&tailed(LINE), &tags, "beta-nodes", INSTALL).unwrap();
203        let json = serde_json::to_value(&document.source).unwrap();
204
205        assert!(json.get("@timestamp").is_some());
206        assert!(json.get("timestamp").is_none());
207    }
208
209    /// Both are stamped or stripped server-side; sending them is at best pointless.
210    #[test]
211    fn host_and_beta_user_are_never_sent() {
212        let tags = NodeTags::from_config(&node_config(Some(UpgradeChannel::Beta)));
213        let document = ForwardDocument::build(&tailed(LINE), &tags, "beta-nodes", INSTALL).unwrap();
214        let json = serde_json::to_value(&document.source).unwrap();
215
216        assert!(json.get("host").is_none());
217        assert!(json.get("beta_user").is_none());
218    }
219
220    #[test]
221    fn an_unspecified_channel_is_not_reported_as_stable() {
222        let tags = NodeTags::from_config(&node_config(None));
223        assert_eq!(tags.channel, "unset");
224
225        let stable = NodeTags::from_config(&node_config(Some(UpgradeChannel::Stable)));
226        assert_eq!(stable.channel, "stable");
227    }
228
229    #[test]
230    fn the_index_comes_from_the_events_timestamp_not_the_wall_clock() {
231        let tags = NodeTags::from_config(&node_config(None));
232
233        let yesterday = ForwardDocument::build(
234            &tailed("2026-08-19T23:59:59.000000Z  INFO ant_node: late"),
235            &tags,
236            "beta-nodes",
237            INSTALL,
238        )
239        .unwrap();
240        let today = ForwardDocument::build(
241            &tailed("2026-08-20T00:00:01.000000Z  INFO ant_node: early"),
242            &tags,
243            "beta-nodes",
244            INSTALL,
245        )
246        .unwrap();
247
248        assert_eq!(yesterday.index, "beta-nodes-2026.08.19");
249        assert_eq!(today.index, "beta-nodes-2026.08.20");
250    }
251
252    /// The property that makes a replayed batch idempotent: same event in, same `_id` and index out,
253    /// no matter when the replay happens.
254    #[test]
255    fn rebuilding_the_same_event_yields_the_same_id_and_index() {
256        let tags = NodeTags::from_config(&node_config(Some(UpgradeChannel::Beta)));
257        let first = ForwardDocument::build(&tailed(LINE), &tags, "beta-nodes", INSTALL).unwrap();
258        let second = ForwardDocument::build(&tailed(LINE), &tags, "beta-nodes", INSTALL).unwrap();
259
260        assert_eq!(first.id, second.id);
261        assert_eq!(first.index, second.index);
262    }
263
264    #[test]
265    fn absent_optional_fields_are_omitted_rather_than_sent_as_null() {
266        let tags = NodeTags::from_config(&node_config(None));
267        let document = ForwardDocument::build(
268            &tailed("2026-08-19T20:50:00.123456Z  INFO plain message with no fields"),
269            &tags,
270            "beta-nodes",
271            INSTALL,
272        )
273        .unwrap();
274        let json = serde_json::to_value(&document.source).unwrap();
275
276        assert!(json.get("peer_id").is_none());
277        assert!(json.get("version").is_none());
278        assert!(json.get("commit").is_none());
279        assert!(json.get("target").is_none());
280    }
281
282    #[test]
283    fn a_custom_index_prefix_is_honoured() {
284        let tags = NodeTags::from_config(&node_config(None));
285        let document =
286            ForwardDocument::build(&tailed(LINE), &tags, "my-test-index", INSTALL).unwrap();
287        assert_eq!(document.index, "my-test-index-2026.08.19");
288    }
289}
ant_core/node/daemon/forward/document.rs

ant_core/node/daemon/forward/
document.rs
