Add agent heartbeat monitoring and fix a query-language lexer bug
Agents now send an independent "still alive" record on a configurable schedule (seconds/minutes/hours, [heartbeat] in agent.toml), separate from real log traffic and tagged with a sentry.heartbeat attribute. No new wire protocol -- it's an ordinary record through the same PushBatch RPC/mTLS identity every log line already uses. Unavailability alerting reuses the existing absence-condition alert rule type unchanged; no new alerting code was needed. See /docs/agent-heartbeat-monitoring.md for the design and how to build the alert rule. While verifying the alert rule live, found that the query language's lexer never treated '-' as part of an identifier, so any unquoted hyphenated filter value -- including the reference doc's own canonical example, `host!=host-03` -- failed to parse at all. Fixed in api/internal/querylang/lexer/lexer.go with regression tests; a leading '-' still lexes as its own token so earliest=-1h/sort -count are unaffected.
This commit is contained in:
@@ -38,6 +38,19 @@ kind = "journald"
|
||||
max_size = 500
|
||||
flush_interval_ms = 2000
|
||||
|
||||
[heartbeat]
|
||||
# How often this agent proves it's still alive to the platform, sent as
|
||||
# its own record independent of whatever real log traffic is flowing --
|
||||
# pair with an "absence" alert rule on the sentry.heartbeat attribute to
|
||||
# get paged when a host goes quiet. Accepts a plain number + unit: s
|
||||
# (seconds), m (minutes), or h (hours) -- same vocabulary as
|
||||
# earliest=/latest= in the query language. See
|
||||
# /docs/agent-heartbeat-monitoring.md.
|
||||
enabled = true
|
||||
interval = "60s"
|
||||
# interval = "5m"
|
||||
# interval = "1h"
|
||||
|
||||
[ingest]
|
||||
endpoint = "https://ingest.internal:4317"
|
||||
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
use anyhow::{Context, Result};
|
||||
use serde::Deserialize;
|
||||
use serde::{Deserialize, Deserializer};
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::time::Duration;
|
||||
|
||||
#[cfg(not(windows))]
|
||||
const DEFAULT_CONFIG_PATH: &str = "/etc/sentry-agent/agent.toml";
|
||||
@@ -13,6 +14,7 @@ pub struct Config {
|
||||
pub agent: AgentConfig,
|
||||
pub source: SourceConfig,
|
||||
pub batch: BatchConfig,
|
||||
pub heartbeat: HeartbeatConfig,
|
||||
pub ingest: IngestConfig,
|
||||
pub tls: TlsConfig,
|
||||
}
|
||||
@@ -141,6 +143,100 @@ impl Default for BatchConfig {
|
||||
}
|
||||
}
|
||||
|
||||
/// Sent independently of `batch` -- a heartbeat is a punctual liveness
|
||||
/// signal, not log data, so it bypasses `Batcher` entirely (see
|
||||
/// main.rs's `send_heartbeat`) rather than waiting on `max_size`/
|
||||
/// `flush_interval_ms` like real records do. This is the operator-facing
|
||||
/// "polling resolution" knob: how often this agent proves it's still
|
||||
/// alive, which a `condition_type = "absence"` alert rule on the
|
||||
/// `sentry.heartbeat` attribute (see /docs/agent-heartbeat-monitoring.md)
|
||||
/// turns into "alert when this host goes quiet."
|
||||
#[derive(Debug, Clone, Deserialize)]
|
||||
#[serde(default)]
|
||||
pub struct HeartbeatConfig {
|
||||
pub enabled: bool,
|
||||
#[serde(deserialize_with = "deserialize_duration")]
|
||||
pub interval: Duration,
|
||||
}
|
||||
|
||||
impl Default for HeartbeatConfig {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
enabled: true,
|
||||
interval: Duration::from_secs(60),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Parses a human-friendly duration string with an explicit unit suffix
|
||||
/// -- "30s", "5m", "1h" -- deliberately the same s/m/h vocabulary
|
||||
/// `earliest=`/`latest=` use in the query language
|
||||
/// (/docs/query-language-reference.md), so the interval you set here and
|
||||
/// the window you write in the matching alert rule's query read the same
|
||||
/// way. Kept as a small hand-rolled parser rather than pulling in a
|
||||
/// duration-parsing crate for this one field -- this is the
|
||||
/// statically-linked edge agent every "no glibc runtime deps" constraint
|
||||
/// in CLAUDE.md is about keeping lean, and the grammar needed here is a
|
||||
/// handful of lines.
|
||||
fn deserialize_duration<'de, D>(deserializer: D) -> Result<Duration, D::Error>
|
||||
where
|
||||
D: Deserializer<'de>,
|
||||
{
|
||||
let s = String::deserialize(deserializer)?;
|
||||
parse_duration(&s).map_err(serde::de::Error::custom)
|
||||
}
|
||||
|
||||
fn parse_duration(s: &str) -> Result<Duration, String> {
|
||||
let s = s.trim();
|
||||
let (num, unit) = s.split_at(s.len().saturating_sub(1));
|
||||
let n: u64 = num
|
||||
.parse()
|
||||
.map_err(|_| format!("expected a duration like \"30s\", \"5m\", or \"1h\", got {s:?}"))?;
|
||||
match unit {
|
||||
"s" => Ok(Duration::from_secs(n)),
|
||||
"m" => Ok(Duration::from_secs(n * 60)),
|
||||
"h" => Ok(Duration::from_secs(n * 3600)),
|
||||
_ => Err(format!("expected a time unit of s, m, or h after {n}, got {s:?}")),
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod heartbeat_config_tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn parses_seconds_minutes_hours() {
|
||||
assert_eq!(parse_duration("30s").unwrap(), Duration::from_secs(30));
|
||||
assert_eq!(parse_duration("5m").unwrap(), Duration::from_secs(300));
|
||||
assert_eq!(parse_duration("2h").unwrap(), Duration::from_secs(7200));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_missing_or_unknown_unit() {
|
||||
assert!(parse_duration("30").is_err());
|
||||
assert!(parse_duration("30x").is_err());
|
||||
assert!(parse_duration("").is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn default_is_60_seconds_and_enabled() {
|
||||
let cfg = HeartbeatConfig::default();
|
||||
assert!(cfg.enabled);
|
||||
assert_eq!(cfg.interval, Duration::from_secs(60));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn toml_field_parses_via_deserialize() {
|
||||
#[derive(Deserialize)]
|
||||
struct Wrapper {
|
||||
#[serde(default)]
|
||||
heartbeat: HeartbeatConfig,
|
||||
}
|
||||
let w: Wrapper = toml::from_str("[heartbeat]\nenabled = true\ninterval = \"90s\"\n").unwrap();
|
||||
assert_eq!(w.heartbeat.interval, Duration::from_secs(90));
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Deserialize)]
|
||||
#[serde(default)]
|
||||
pub struct IngestConfig {
|
||||
|
||||
@@ -97,8 +97,20 @@ pub async fn run_agent(config_path: Option<PathBuf>) -> Result<()> {
|
||||
let mut batcher = Batcher::new(cfg.batch.max_size, flush_interval);
|
||||
let mut ticker = tokio::time::interval(flush_interval.max(Duration::from_millis(50)));
|
||||
|
||||
// Heartbeat's own ticker, independent of the batch flush ticker above
|
||||
// -- it always fires on cfg.heartbeat.interval regardless of
|
||||
// cfg.batch's settings or whether any real log traffic is flowing.
|
||||
// Built unconditionally even when disabled (tokio::time::interval
|
||||
// doesn't fail on construction); the `if cfg.heartbeat.enabled`
|
||||
// select! guard is what actually turns it off, so a disabled
|
||||
// heartbeat costs nothing beyond one idle timer.
|
||||
let mut heartbeat_ticker = tokio::time::interval(cfg.heartbeat.interval.max(Duration::from_millis(50)));
|
||||
|
||||
loop {
|
||||
tokio::select! {
|
||||
_ = heartbeat_ticker.tick(), if cfg.heartbeat.enabled => {
|
||||
send_heartbeat(&mut client, &host, &service).await;
|
||||
}
|
||||
maybe_line = rx.recv() => {
|
||||
let Some(raw) = maybe_line else {
|
||||
tracing::warn!("source exited, flushing remaining batch and shutting down");
|
||||
@@ -191,6 +203,40 @@ async fn spawn_source(source: config::SourceConfig, tx: source::LineSender) {
|
||||
}
|
||||
}
|
||||
|
||||
/// Sends a single synthetic record through the same `PushBatch` RPC and
|
||||
/// mTLS identity as real log data -- no new proto message, no new ingest
|
||||
/// code, no new ClickHouse schema. Bypasses `Batcher` (see the heartbeat
|
||||
/// ticker's own comment above): a heartbeat that got queued behind
|
||||
/// `batch.max_size` or `batch.flush_interval_ms` would defeat the point
|
||||
/// of a punctual "still alive" signal. Distinguished from a real log
|
||||
/// record purely by the `sentry.heartbeat` attribute -- `service` stays
|
||||
/// the agent's real configured service so it doesn't pollute
|
||||
/// service-based dashboards/faceting with a fake value. See
|
||||
/// /docs/agent-heartbeat-monitoring.md for how an absence alert rule
|
||||
/// turns a run of missed heartbeats into a notification.
|
||||
async fn send_heartbeat(client: &mut LogIngestClient<Channel>, host: &str, service: &str) {
|
||||
let record = LogRecord {
|
||||
timestamp_unix_nano: now_unix_nanos(),
|
||||
host: host.to_string(),
|
||||
service: service.to_string(),
|
||||
severity: Severity::Info as i32,
|
||||
message: "agent heartbeat".to_string(),
|
||||
attributes: std::collections::HashMap::from([("sentry.heartbeat".to_string(), "true".to_string())]),
|
||||
record_id: String::new(),
|
||||
};
|
||||
match grpc::send_batch(client, format!("heartbeat-{}", batch_id()), vec![record]).await {
|
||||
Ok(_) => tracing::debug!(host, "heartbeat sent"),
|
||||
Err(e) => tracing::warn!(error = %e, host, "heartbeat send failed"),
|
||||
}
|
||||
}
|
||||
|
||||
fn now_unix_nanos() -> i64 {
|
||||
std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.map(|d| d.as_nanos() as i64)
|
||||
.unwrap_or(0)
|
||||
}
|
||||
|
||||
async fn flush(client: &mut LogIngestClient<Channel>, batch: Vec<LogRecord>) {
|
||||
let n = batch.len();
|
||||
let batch_id = batch_id();
|
||||
|
||||
Reference in New Issue
Block a user