ant_core/node/daemon/forward/
mod.rs

1//! Opt-in forwarding of managed nodes' log files to the beta-channel Elasticsearch.
2//!
3//! The daemon already knows the log directory of every node it manages, so forwarding needs no OS
4//! service, no separate install, and nothing platform-specific: `ant node logs forward enable` is
5//! the consent act, and from then on a background task tails those files and batch-ships their
6//! events until the user runs `disable`.
7//!
8//! Three properties shape the whole design:
9//!
10//! - **It must never slow a node down.** The forwarder only *reads* log files. It never touches a
11//!   node process, its stdio, or any lock on the node's path, and all of its work happens on its
12//!   own task.
13//! - **Delivery is best-effort.** This is logs-only telemetry, so a lost batch is acceptable and
14//!   nothing here is allowed to grow without bound waiting for the endpoint to come back.
15//! - **A daemon restart must not duplicate or lose events.** Tail offsets are persisted, and every
16//!   document carries a deterministic `_id` so that replaying a batch after a transport failure is
17//!   idempotent rather than duplicating whatever already landed.
18
19pub mod config;
20pub mod document;
21pub mod es;
22pub mod offsets;
23pub mod parse;
24pub mod runner;
25pub mod sink;
26pub mod tail;
27
28use serde::{Deserialize, Serialize};
29
30pub use config::{LogForwardConfig, LogLevel, DEFAULT_ENDPOINT, DEFAULT_INDEX_PREFIX};
31pub use document::{ForwardDocument, NodeTags};
32pub use es::ElasticsearchSink;
33pub use offsets::OffsetStore;
34pub use parse::{parse_line, LogEvent};
35pub use runner::{classify_nodes, spawn_log_forwarder, ForwarderHandle, DEFAULT_POLL_INTERVAL};
36pub use sink::{BatchOutcome, DocumentOutcome, DocumentQueue, LogSink, RetryPolicy};
37pub use tail::{LogTailer, TailedEvent};
38
39/// A node the forwarder is tailing.
40#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)]
41pub struct ForwardingNode {
42    pub node_id: u32,
43    pub service: String,
44    /// Log directory being tailed.
45    pub log_dir: String,
46}
47
48/// A node the forwarder cannot tail, and why.
49///
50/// The common case by far is a node added without `--log-dir-path`: node file logging is off by
51/// default, so such a node writes no log files at all and there is nothing to forward. Surfacing
52/// these explicitly is what stops `enable` looking like it succeeded while shipping nothing.
53#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)]
54pub struct SkippedNode {
55    pub node_id: u32,
56    pub service: String,
57    /// Human-readable explanation, suitable for printing directly.
58    pub reason: String,
59}
60
61impl SkippedNode {
62    /// The skip reason for a node that has no log directory configured.
63    #[must_use]
64    pub fn no_logging(node_id: u32, service: impl Into<String>) -> Self {
65        Self {
66            node_id,
67            service: service.into(),
68            reason: "logging is not enabled for this node — re-add it with --log-dir-path to \
69                     forward its logs"
70                .to_string(),
71        }
72    }
73}
74
75/// Counters describing what the forwarder has done since the daemon started.
76///
77/// Deliberately cheap to maintain and safe to lose: these are for answering "is it working?", not
78/// for accounting. They reset when the daemon restarts.
79#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)]
80pub struct ForwardStats {
81    /// Events accepted by the endpoint.
82    pub events_forwarded: u64,
83    /// Events dropped locally for being below the configured minimum level.
84    pub events_dropped_by_level: u64,
85    /// Events dropped because the in-memory queue was full — the endpoint could not keep up and
86    /// the forwarder chose to bound its memory rather than block.
87    pub events_dropped_by_overflow: u64,
88    /// Batches the endpoint accepted in full.
89    pub batches_sent: u64,
90    /// Batches abandoned after exhausting their retries.
91    pub batches_failed: u64,
92    /// Unix seconds of the last batch the endpoint accepted.
93    pub last_success_unix: Option<u64>,
94    /// Most recent delivery error, retained so `status` can explain a stalled flow.
95    pub last_error: Option<String>,
96}
97
98/// Everything `ant node logs forward status` reports.
99///
100/// Carries a token *fingerprint*, never the token: the daemon serves this over HTTP, and handing
101/// the write key back out would widen the blast radius of anything that can reach the API.
102#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)]
103pub struct LogForwardStatus {
104    pub enabled: bool,
105    pub endpoint: String,
106    pub index_prefix: String,
107    pub min_level: LogLevel,
108    /// Short, non-reversible identifier for the configured token. `None` when none is set.
109    pub token_fingerprint: Option<String>,
110    /// Whether the background task is currently running. Distinguishes "enabled but the daemon has
111    /// not been restarted yet" from "enabled and shipping".
112    pub active: bool,
113    pub nodes_forwarding: Vec<ForwardingNode>,
114    pub nodes_skipped: Vec<SkippedNode>,
115    pub stats: ForwardStats,
116}
117
118impl LogForwardStatus {
119    /// Build a status from persisted config alone, for the case where no forwarder is running —
120    /// either forwarding is disabled, or the CLI is reading the config with the daemon down.
121    #[must_use]
122    pub fn inactive(config: &LogForwardConfig) -> Self {
123        Self {
124            enabled: config.enabled,
125            endpoint: config.endpoint.clone(),
126            index_prefix: config.index_prefix.clone(),
127            min_level: config.min_level,
128            token_fingerprint: config.token_fingerprint(),
129            active: false,
130            nodes_forwarding: Vec::new(),
131            nodes_skipped: Vec::new(),
132            stats: ForwardStats::default(),
133        }
134    }
135}
136
137/// Request body for enabling forwarding.
138///
139/// Every field is optional so that re-enabling after a `disable` needs no arguments: the stored
140/// token, endpoint and level are reused unless the caller overrides them. Only the very first
141/// `enable` on a machine has to supply a token.
142#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)]
143pub struct LogForwardEnableRequest {
144    /// Write-only Elasticsearch API key. Reuses the stored one when omitted.
145    #[serde(default)]
146    pub token: Option<String>,
147    /// Ingest endpoint override, for testing against a local sink.
148    #[serde(default)]
149    pub endpoint: Option<String>,
150    /// Minimum level to forward. Defaults to INFO on first enable.
151    #[serde(default)]
152    pub min_level: Option<LogLevel>,
153}
154
155/// Merge a request onto the stored config and validate the result.
156///
157/// Kept out of both the HTTP handler and the CLI so the two paths cannot drift: `enable` means the
158/// same thing whether it arrives over the daemon's API or is written straight to disk with the
159/// daemon stopped.
160pub fn apply_enable(
161    stored: &LogForwardConfig,
162    request: &LogForwardEnableRequest,
163) -> crate::error::Result<LogForwardConfig> {
164    let mut config = stored.clone();
165    config.enabled = true;
166
167    if let Some(token) = &request.token {
168        config.token = token.trim().to_string();
169    }
170    if let Some(endpoint) = &request.endpoint {
171        config.endpoint = endpoint.trim().to_string();
172    }
173    if let Some(level) = request.min_level {
174        config.min_level = level;
175    }
176
177    // Generated on the first enable and stable thereafter. Document ids are namespaced by it, so a
178    // machine that has never opted in needs one before it can ship anything.
179    config.ensure_installation_id();
180
181    config.validate()?;
182    Ok(config)
183}
184
185/// Outcome of `enable` or `disable`.
186#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)]
187pub struct LogForwardResult {
188    /// Whether forwarding is enabled after the call.
189    pub enabled: bool,
190    /// True when the call found the setting already in the requested state.
191    pub already_in_state: bool,
192    pub endpoint: String,
193    pub min_level: LogLevel,
194    /// Nodes that will be tailed.
195    pub nodes_forwarding: Vec<ForwardingNode>,
196    /// Nodes that cannot be tailed, with reasons — most often because they have no log directory.
197    pub nodes_skipped: Vec<SkippedNode>,
198    /// Set when the config was persisted but no forwarder could be started because the daemon is
199    /// not running; forwarding begins when it next starts.
200    pub pending_daemon_start: bool,
201}
202
203#[cfg(test)]
204mod tests {
205    use super::*;
206
207    fn stored_with_token() -> LogForwardConfig {
208        LogForwardConfig {
209            enabled: false,
210            token: "stored-key".to_string(),
211            installation_id: "0123456789abcdef".to_string(),
212            ..LogForwardConfig::disabled()
213        }
214    }
215
216    #[test]
217    fn enabling_for_the_first_time_requires_a_token() {
218        let error = apply_enable(
219            &LogForwardConfig::disabled(),
220            &LogForwardEnableRequest::default(),
221        )
222        .unwrap_err()
223        .to_string();
224        assert!(error.contains("token"), "{error}");
225    }
226
227    /// Re-enabling after `disable` must not make the user find their key again.
228    #[test]
229    fn re_enabling_reuses_the_stored_token_and_settings() {
230        let stored = LogForwardConfig {
231            endpoint: "http://127.0.0.1:9999".to_string(),
232            min_level: LogLevel::Warn,
233            ..stored_with_token()
234        };
235
236        let config = apply_enable(&stored, &LogForwardEnableRequest::default()).unwrap();
237
238        assert!(config.enabled);
239        assert_eq!(config.token, "stored-key");
240        assert_eq!(config.endpoint, "http://127.0.0.1:9999");
241        assert_eq!(config.min_level, LogLevel::Warn);
242    }
243
244    /// The namespace is minted on the first enable, and never changes afterwards — regenerating it
245    /// would make a replayed batch look like new documents and duplicate them.
246    #[test]
247    fn enabling_mints_an_installation_id_and_later_enables_keep_it() {
248        let config = apply_enable(
249            &LogForwardConfig::disabled(),
250            &LogForwardEnableRequest {
251                token: Some("first-key".to_string()),
252                ..LogForwardEnableRequest::default()
253            },
254        )
255        .unwrap();
256        assert_eq!(config.installation_id.len(), 16);
257
258        let re_enabled = apply_enable(&config, &LogForwardEnableRequest::default()).unwrap();
259        assert_eq!(re_enabled.installation_id, config.installation_id);
260
261        // Even rotating the token must not change it.
262        let rotated = apply_enable(
263            &config,
264            &LogForwardEnableRequest {
265                token: Some("rotated-key".to_string()),
266                ..LogForwardEnableRequest::default()
267            },
268        )
269        .unwrap();
270        assert_eq!(rotated.installation_id, config.installation_id);
271    }
272
273    #[test]
274    fn a_supplied_token_endpoint_and_level_override_what_was_stored() {
275        let config = apply_enable(
276            &stored_with_token(),
277            &LogForwardEnableRequest {
278                token: Some("  rotated-key  ".to_string()),
279                endpoint: Some("http://localhost:8080".to_string()),
280                min_level: Some(LogLevel::Error),
281            },
282        )
283        .unwrap();
284
285        assert_eq!(config.token, "rotated-key", "surrounding space is trimmed");
286        assert_eq!(config.endpoint, "http://localhost:8080");
287        assert_eq!(config.min_level, LogLevel::Error);
288    }
289
290    #[test]
291    fn an_invalid_endpoint_is_rejected_before_anything_is_persisted() {
292        let error = apply_enable(
293            &stored_with_token(),
294            &LogForwardEnableRequest {
295                endpoint: Some("logs.autonomi.com".to_string()),
296                ..LogForwardEnableRequest::default()
297            },
298        )
299        .unwrap_err()
300        .to_string();
301        assert!(error.contains("http(s) URL"), "{error}");
302    }
303
304    #[test]
305    fn inactive_status_mirrors_the_config_without_exposing_the_token() {
306        let config = LogForwardConfig {
307            enabled: true,
308            token: "secret-api-key".to_string(),
309            ..LogForwardConfig::disabled()
310        };
311
312        let status = LogForwardStatus::inactive(&config);
313
314        assert!(status.enabled);
315        assert!(!status.active);
316        assert_eq!(status.endpoint, DEFAULT_ENDPOINT);
317        assert_eq!(status.min_level, LogLevel::Info);
318        assert_eq!(status.token_fingerprint, config.token_fingerprint());
319
320        let json = serde_json::to_string(&status).unwrap();
321        assert!(
322            !json.contains("secret-api-key"),
323            "status must never carry the token: {json}"
324        );
325    }
326
327    #[test]
328    fn inactive_status_of_a_disabled_config_has_no_fingerprint() {
329        let status = LogForwardStatus::inactive(&LogForwardConfig::disabled());
330        assert!(!status.enabled);
331        assert_eq!(status.token_fingerprint, None);
332    }
333
334    #[test]
335    fn skip_reason_points_at_the_flag_that_fixes_it() {
336        let skipped = SkippedNode::no_logging(3, "node3");
337        assert_eq!(skipped.node_id, 3);
338        assert_eq!(skipped.service, "node3");
339        assert!(skipped.reason.contains("--log-dir-path"));
340    }
341
342    #[test]
343    fn stats_start_at_zero() {
344        let stats = ForwardStats::default();
345        assert_eq!(stats.events_forwarded, 0);
346        assert_eq!(stats.batches_failed, 0);
347        assert_eq!(stats.last_success_unix, None);
348        assert_eq!(stats.last_error, None);
349    }
350}
ant_core/node/daemon/forward/mod.rs

ant_core/node/daemon/forward/
mod.rs
