diff --git a/agent/sentry-agent/build.rs b/agent/sentry-agent/build.rs index 06cc559..62caa1e 100644 --- a/agent/sentry-agent/build.rs +++ b/agent/sentry-agent/build.rs @@ -2,7 +2,10 @@ fn main() -> Result<(), Box> { tonic_build::configure() .build_server(false) .compile_protos( - &["../../proto/sentry/logs/v1/logs.proto"], + &[ + "../../proto/sentry/logs/v1/logs.proto", + "../../proto/sentry/agent/v1/agent_control.proto", + ], &["../../proto"], )?; Ok(()) diff --git a/agent/sentry-agent/src/batch.rs b/agent/sentry-agent/src/batch.rs index 075063c..4b1f7fa 100644 --- a/agent/sentry-agent/src/batch.rs +++ b/agent/sentry-agent/src/batch.rs @@ -45,6 +45,20 @@ impl Batcher { } } + /// Unconditionally drains whatever is buffered, ignoring both + /// `max_size` and `flush_interval` -- for shutdown and for a + /// config hot-reload replacing this `Batcher` outright (Phase: + /// agent management's remote config editing), neither of which + /// should silently drop records just because the timeout hadn't + /// elapsed yet. + pub fn flush_all(&mut self) -> Option> { + if self.buf.is_empty() { + None + } else { + Some(self.drain()) + } + } + fn drain(&mut self) -> Vec { self.last_flush = Instant::now(); std::mem::replace(&mut self.buf, Vec::with_capacity(self.max_size)) @@ -106,4 +120,19 @@ mod tests { b.push(rec("a")); assert!(b.poll_timeout().is_none()); } + + // Regression test: shutdown used to call poll_timeout(), which + // silently drops anything buffered before flush_interval elapses -- + // real data loss on a graceful shutdown that happened to land + // between flushes. flush_all() is what shutdown (and hot-reload) + // must use instead. + #[test] + fn flush_all_drains_regardless_of_timeout() { + let mut b = Batcher::new(10, Duration::from_secs(999)); + b.push(rec("a")); + assert!(b.poll_timeout().is_none(), "sanity: timeout hasn't elapsed"); + let batch = b.flush_all().expect("flush_all should drain unconditionally"); + assert_eq!(batch.len(), 1); + assert!(b.flush_all().is_none(), "buffer should be empty after draining"); + } } diff --git a/agent/sentry-agent/src/grpc.rs b/agent/sentry-agent/src/grpc.rs index d2195bb..94e480f 100644 --- a/agent/sentry-agent/src/grpc.rs +++ b/agent/sentry-agent/src/grpc.rs @@ -1,12 +1,18 @@ use crate::config::{IngestConfig, TlsConfig}; +use crate::pb::agent::v1::{agent_control_client::AgentControlClient, CheckInRequest, CheckInResponse}; use crate::pb::{log_ingest_client::LogIngestClient, LogRecord, PushBatchRequest}; use anyhow::{Context, Result}; use tonic::transport::{Certificate, Channel, ClientTlsConfig, Identity}; -/// Establishes an mTLS gRPC channel to the ingest service. Agents never -/// talk to Redpanda directly — this is the only network egress the agent -/// has, by design (see /docs/architecture.md). -pub async fn connect(ingest: &IngestConfig, tls: &TlsConfig) -> Result> { +/// Establishes the one mTLS gRPC channel an agent has to ingest — +/// agents never talk to Redpanda directly, and never accept an inbound +/// connection either (see /docs/architecture.md and +/// /docs/agent-management-design.md). Returns the bare `Channel` rather +/// than a client wrapper so callers can build both `LogIngestClient` +/// (data plane) and `AgentControlClient` (control plane) from the same +/// connection — `Channel` is a cheap-to-clone handle, not the socket +/// itself, so there's no cost to sharing it across two client stubs. +pub async fn connect(ingest: &IngestConfig, tls: &TlsConfig) -> Result { let ca = tokio::fs::read(&tls.ca_cert) .await .with_context(|| format!("reading CA cert at {}", tls.ca_cert.display()))?; @@ -21,15 +27,13 @@ pub async fn connect(ingest: &IngestConfig, tls: &TlsConfig) -> Result, + req: CheckInRequest, +) -> Result { + let resp = client.check_in(req).await.context("CheckIn RPC failed")?; + Ok(resp.into_inner()) +} diff --git a/agent/sentry-agent/src/main.rs b/agent/sentry-agent/src/main.rs index b9dd287..eecf41b 100644 --- a/agent/sentry-agent/src/main.rs +++ b/agent/sentry-agent/src/main.rs @@ -8,12 +8,19 @@ mod service; pub mod pb { tonic::include_proto!("sentry.logs.v1"); + + pub mod agent { + pub mod v1 { + tonic::include_proto!("sentry.agent.v1"); + } + } } use anyhow::{Context, Result}; use batch::Batcher; use clap::Parser; use config::Config; +use pb::agent::v1::{agent_control_client::AgentControlClient, CheckInRequest, DesiredOverride, ReportedConfig}; use pb::{log_ingest_client::LogIngestClient, LogRecord, Severity}; use std::path::PathBuf; use std::time::Duration; @@ -85,31 +92,92 @@ pub async fn run_agent(config_path: Option) -> Result<()> { let host = cfg.agent.host.clone().unwrap_or_else(default_hostname); let service = cfg.agent.service.clone(); - let (tx, mut rx) = mpsc::channel(1024); - let source_handle = tokio::spawn(spawn_source(cfg.source.clone(), tx)); + // source_cfg is mutable: a remote DesiredOverride's journald_unit + // field (see apply_override below) can change it at runtime, which + // means aborting and respawning the source task with the new + // filter -- source_handle/rx are mutable for the same reason. + let mut source_cfg = cfg.source.clone(); + let (mut source_handle, mut rx) = spawn_source_task(source_cfg.clone()); - let mut client = grpc::connect(&cfg.ingest, &cfg.tls) + let channel = grpc::connect(&cfg.ingest, &cfg.tls) .await .context("connecting to ingest service")?; + let mut client = LogIngestClient::new(channel.clone()); + let mut control_client = AgentControlClient::new(channel); tracing::info!(endpoint = %cfg.ingest.endpoint, "connected to ingest service"); - let flush_interval = Duration::from_millis(cfg.batch.flush_interval_ms); - let mut batcher = Batcher::new(cfg.batch.max_size, flush_interval); + // Effective runtime settings, seeded from local config -- every one + // of these is mutable because a remote DesiredOverride can change + // it (see apply_override). The local agent.toml is never rewritten; + // an override lives only in memory here and reverts to agent.toml's + // own values on restart, re-syncing on the next successful CheckIn + // (see /docs/agent-management-design.md's merge-semantics section). + let mut batch_max_size = cfg.batch.max_size; + let mut flush_interval = Duration::from_millis(cfg.batch.flush_interval_ms); + let mut heartbeat_enabled = cfg.heartbeat.enabled; + let mut heartbeat_interval = cfg.heartbeat.interval; + // Empty until the first override is ever applied -- echoed back on + // every CheckIn as-is so the server can tell "pending" (an edit + // exists this agent hasn't picked up) from "applied." + let mut applied_override_version = String::new(); + + let mut batcher = Batcher::new(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))); + // -- it always fires on heartbeat_interval regardless of the batch + // settings or whether any real log traffic is flowing. Also drives + // CheckIn (see the arm below) unconditionally -- CheckIn keeps + // running even when heartbeat_enabled is false, since that's an + // agent's only path to ever receive a remote override that + // re-enables it; only the heartbeat log record itself is gated on + // heartbeat_enabled. + let mut heartbeat_ticker = tokio::time::interval(heartbeat_interval.max(Duration::from_millis(50))); loop { tokio::select! { - _ = heartbeat_ticker.tick(), if cfg.heartbeat.enabled => { - send_heartbeat(&mut client, &host, &service).await; + _ = heartbeat_ticker.tick() => { + if heartbeat_enabled { + send_heartbeat(&mut client, &host, &service).await; + } + + let reported = ReportedConfig { + agent_version: env!("CARGO_PKG_VERSION").to_string(), + source_kind: source_kind_name(&source_cfg), + source_detail: source_detail_summary(&source_cfg), + batch_max_size: batch_max_size as u64, + batch_flush_interval_ms: flush_interval.as_millis() as u64, + heartbeat_enabled, + heartbeat_interval_ms: heartbeat_interval.as_millis() as u64, + }; + match grpc::check_in(&mut control_client, CheckInRequest { + host: host.clone(), + service: service.clone(), + current_config: Some(reported), + applied_override_version: applied_override_version.clone(), + }).await { + Ok(resp) => { + if let Some(ov) = resp.has_override.then_some(resp.r#override).flatten() { + if ov.version != applied_override_version { + apply_override( + &ov, + &mut batch_max_size, &mut flush_interval, + &mut heartbeat_enabled, &mut heartbeat_interval, + &mut batcher, &mut ticker, &mut heartbeat_ticker, + &mut source_cfg, &mut source_handle, &mut rx, + &mut client, + ).await; + applied_override_version = ov.version.clone(); + tracing::info!(version = %applied_override_version, "applied remote config override"); + } + } + } + // A failed check-in is not fatal -- same graceful- + // degradation posture as a failed heartbeat/batch + // flush: an agent management feature being + // unreachable must never stop log collection. + Err(e) => tracing::debug!(error = %e, "check-in failed"), + } } maybe_line = rx.recv() => { let Some(raw) = maybe_line else { @@ -148,13 +216,113 @@ pub async fn run_agent(config_path: Option) -> Result<()> { } } - if let Some(batch) = batcher.poll_timeout() { + // flush_all(), not poll_timeout(): shutdown must send whatever's + // buffered unconditionally -- poll_timeout() only drains once + // flush_interval has elapsed, so anything buffered more recently + // than that would otherwise be silently dropped on every graceful + // shutdown that happens to land between flushes. Same reasoning + // applies to a config hot-reload replacing this batcher outright + // (see apply_override). + if let Some(batch) = batcher.flush_all() { flush(&mut client, batch).await; } source_handle.abort(); Ok(()) } +/// Applies a newly-received DesiredOverride to the agent's in-memory +/// runtime state -- see run_agent's CheckIn arm, and +/// /docs/agent-management-design.md's merge-semantics section for why +/// this never touches the local agent.toml file. Every field is +/// independently optional (unset = keep the current value); batch/ +/// heartbeat settings always get their batcher/ticker rebuilt together +/// when *any* override arrives, for simplicity, rather than tracking +/// which specific field changed -- this only runs when a human edits an +/// agent's config from the web UI, not on a hot path, so the extra +/// timer/allocation churn doesn't matter. +#[allow(clippy::too_many_arguments)] +async fn apply_override( + ov: &DesiredOverride, + batch_max_size: &mut usize, + flush_interval: &mut Duration, + heartbeat_enabled: &mut bool, + heartbeat_interval: &mut Duration, + batcher: &mut Batcher, + ticker: &mut tokio::time::Interval, + heartbeat_ticker: &mut tokio::time::Interval, + source_cfg: &mut config::SourceConfig, + source_handle: &mut tokio::task::JoinHandle<()>, + rx: &mut mpsc::Receiver, + client: &mut LogIngestClient, +) { + if let Some(v) = ov.batch_max_size { + *batch_max_size = v as usize; + } + if let Some(v) = ov.batch_flush_interval_ms { + *flush_interval = Duration::from_millis(v); + } + // Flush whatever the old batcher was holding before replacing it -- + // a hot-reload must never silently drop buffered-but-not-yet-due + // records, same reasoning as shutdown's flush_all() above. + if let Some(old) = batcher.flush_all() { + flush(client, old).await; + } + *batcher = Batcher::new(*batch_max_size, *flush_interval); + *ticker = tokio::time::interval((*flush_interval).max(Duration::from_millis(50))); + + if let Some(v) = ov.heartbeat_enabled { + *heartbeat_enabled = v; + } + if let Some(v) = ov.heartbeat_interval_ms { + *heartbeat_interval = Duration::from_millis(v); + } + *heartbeat_ticker = tokio::time::interval((*heartbeat_interval).max(Duration::from_millis(50))); + + // Only meaningful (and only ever sent by the server) when this + // agent's local source is journald -- ignored otherwise, per + // agent_control.proto's DesiredOverride.journald_unit comment. + // Changing it means aborting and respawning the source task: unlike + // batch/heartbeat, there's no way to change what journald::run is + // tailing without restarting that task. + if let Some(unit) = &ov.journald_unit { + if let config::SourceConfig::Journald { unit: current_unit } = source_cfg { + let new_unit = if unit.is_empty() { None } else { Some(unit.clone()) }; + if *current_unit != new_unit { + *source_cfg = config::SourceConfig::Journald { unit: new_unit }; + source_handle.abort(); + let (new_handle, new_rx) = spawn_source_task(source_cfg.clone()); + *source_handle = new_handle; + *rx = new_rx; + tracing::info!(unit = ?unit, "applied remote journald unit override, restarted source"); + } + } + } +} + +fn spawn_source_task(source_cfg: config::SourceConfig) -> (tokio::task::JoinHandle<()>, mpsc::Receiver) { + let (tx, rx) = mpsc::channel(1024); + let handle = tokio::spawn(spawn_source(source_cfg, tx)); + (handle, rx) +} + +fn source_kind_name(cfg: &config::SourceConfig) -> String { + match cfg { + config::SourceConfig::Journald { .. } => "journald".to_string(), + config::SourceConfig::File { .. } => "file".to_string(), + config::SourceConfig::EventLog { .. } => "eventlog".to_string(), + config::SourceConfig::Etw { .. } => "etw".to_string(), + } +} + +fn source_detail_summary(cfg: &config::SourceConfig) -> String { + match cfg { + config::SourceConfig::Journald { unit } => unit.clone().unwrap_or_else(|| "(whole journal)".to_string()), + config::SourceConfig::File { path, .. } => path.display().to_string(), + config::SourceConfig::EventLog { channels } => channels.join(","), + config::SourceConfig::Etw { providers } => providers.join(","), + } +} + // `tx` genuinely goes unused in one rare-but-valid combination: Windows // features enabled while targeting a non-Windows platform (e.g. sanity- // checking the Windows source arms compile shape from Linux, which is diff --git a/api/agents/handler.go b/api/agents/handler.go new file mode 100644 index 0000000..b630515 --- /dev/null +++ b/api/agents/handler.go @@ -0,0 +1,175 @@ +package agents + +import ( + "context" + "encoding/json" + "errors" + "log/slog" + "net/http" + + "github.com/sentry/sentry/api/authz" +) + +// store is the narrow interface Handler depends on -- *Store (store.go) +// is the production implementation; tests use a fake, same pattern as +// dashboards.store/queryapi's SQLRunner. +type store interface { + List(ctx context.Context, tenantID string) ([]Agent, error) + Get(ctx context.Context, tenantID, host string) (*Agent, error) + SetOverride(ctx context.Context, tenantID, host string, override ConfigOverride, updatedBy string) (*Agent, error) + ClearOverride(ctx context.Context, tenantID, host string) error +} + +type Handler struct { + logger *slog.Logger + store store + authorizer authz.Authorizer +} + +func NewHandler(logger *slog.Logger, store store, authorizer authz.Authorizer) *Handler { + return &Handler{logger: logger, store: store, authorizer: authorizer} +} + +// RegisterRoutes: viewing inventory is RoleViewer (same bar as viewing +// a dashboard); editing an agent's remote config is RoleEditor -- an +// operational-tuning action, not an admin-only one, matching the RBAC +// matrix's treatment of alert rules/notification targets rather than +// user/role management. +func (h *Handler) RegisterRoutes(mux *http.ServeMux) { + mux.HandleFunc("GET /agents", authz.RequireRole(h.authorizer, authz.RoleViewer, h.handleList)) + mux.HandleFunc("GET /agents/{host}", authz.RequireRole(h.authorizer, authz.RoleViewer, h.handleGet)) + mux.HandleFunc("PUT /agents/{host}/config", authz.RequireRole(h.authorizer, authz.RoleEditor, h.handleSetConfig)) + mux.HandleFunc("DELETE /agents/{host}/config", authz.RequireRole(h.authorizer, authz.RoleEditor, h.handleClearConfig)) +} + +// tenantID mirrors dashboards.Handler.tenantID exactly -- resolved from +// the authenticated identity, never from a client-supplied field +// (there isn't one here to begin with; host alone identifies an agent +// within a tenant). +func (h *Handler) tenantID(r *http.Request) string { + if id, ok := authz.IdentityFromContext(r.Context()); ok && id.TenantID != "" { + return id.TenantID + } + return "default" +} + +func (h *Handler) updatedBy(r *http.Request) string { + if id, ok := authz.IdentityFromContext(r.Context()); ok { + return id.UserID + } + return "" +} + +func (h *Handler) handleList(w http.ResponseWriter, r *http.Request) { + list, err := h.store.List(r.Context(), h.tenantID(r)) + if err != nil { + h.logger.Error("listing agents", "error", err) + writeError(w, http.StatusInternalServerError, "listing agents failed") + return + } + writeJSON(w, http.StatusOK, list) +} + +func (h *Handler) handleGet(w http.ResponseWriter, r *http.Request) { + a, err := h.store.Get(r.Context(), h.tenantID(r), r.PathValue("host")) + if err != nil { + h.writeStoreErr(w, err, "getting agent") + return + } + writeJSON(w, http.StatusOK, a) +} + +// setConfigRequest is deliberately the same shape as ConfigOverride +// (Handler just decodes straight into it) -- every field optional, +// unset means "no override for this field." A caller changing just one +// field (e.g. only heartbeat_interval_ms) must still send the fields +// they want to KEEP as an override alongside it, since SetOverride +// replaces the whole stored override -- the web UI's edit form always +// reads the agent's current DesiredOverride first and PUTs back the +// full merged set, same pattern any other "edit form that PUTs a whole +// resource" in this codebase already uses (e.g. dashboards' PUT). +func (h *Handler) handleSetConfig(w http.ResponseWriter, r *http.Request) { + var override ConfigOverride + if !decodeJSON(w, r, &override) { + return + } + if err := validateOverride(override); err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + + a, err := h.store.SetOverride(r.Context(), h.tenantID(r), r.PathValue("host"), override, h.updatedBy(r)) + if err != nil { + h.writeStoreErr(w, err, "setting agent config") + return + } + writeJSON(w, http.StatusOK, a) +} + +func (h *Handler) handleClearConfig(w http.ResponseWriter, r *http.Request) { + if err := h.store.ClearOverride(r.Context(), h.tenantID(r), r.PathValue("host")); err != nil { + h.writeStoreErr(w, err, "clearing agent config") + return + } + w.WriteHeader(http.StatusNoContent) +} + +// validateOverride rejects the two footguns a naive remote-config-edit +// feature could otherwise ship: a batch/heartbeat interval of 0 would +// mean "flush constantly"/"heartbeat constantly," hammering ingest and +// the agent's own CPU for no operator-intended reason -- floors match +// this codebase's other real floors (alerting's own +// eval_interval_seconds >= 30, found live during the heartbeat feature +// this builds on). There is deliberately no validation here for +// tls/ingest fields, because ConfigOverride has no such fields at all +// -- ingest connection details are not a remotely-editable dimension of +// an agent's config, full stop (see /docs/agent-management-design.md's +// security boundary section). +func validateOverride(o ConfigOverride) error { + if o.BatchMaxSize != nil && *o.BatchMaxSize < 1 { + return errors.New("batch_max_size must be at least 1") + } + if o.BatchFlushIntervalMS != nil && *o.BatchFlushIntervalMS < 100 { + return errors.New("batch_flush_interval_ms must be at least 100") + } + if o.HeartbeatIntervalMS != nil && *o.HeartbeatIntervalMS < 5000 { + return errors.New("heartbeat_interval_ms must be at least 5000 (5s)") + } + return nil +} + +func (h *Handler) writeStoreErr(w http.ResponseWriter, err error, action string) { + if errors.Is(err, ErrNotFound) { + writeError(w, http.StatusNotFound, "agent not found") + return + } + h.logger.Error(action, "error", err) + writeError(w, http.StatusInternalServerError, action+" failed") +} + +const maxBodyBytes = 1 << 20 // 1 MiB, same cap as queryapi/dashboards + +func decodeJSON(w http.ResponseWriter, r *http.Request, v any) bool { + r.Body = http.MaxBytesReader(w, r.Body, maxBodyBytes) + if err := json.NewDecoder(r.Body).Decode(v); err != nil { + writeError(w, http.StatusBadRequest, "invalid JSON body: "+err.Error()) + return false + } + return true +} + +func writeJSON(w http.ResponseWriter, status int, v any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(v) +} + +type errorResponse struct { + Error string `json:"error"` +} + +func writeError(w http.ResponseWriter, status int, msg string) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(errorResponse{Error: msg}) +} diff --git a/api/agents/handler_test.go b/api/agents/handler_test.go new file mode 100644 index 0000000..44ca76d --- /dev/null +++ b/api/agents/handler_test.go @@ -0,0 +1,218 @@ +package agents + +import ( + "bytes" + "context" + "encoding/json" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "testing" + + "github.com/sentry/sentry/api/authz" +) + +func discardLogger() *slog.Logger { + return slog.New(slog.NewTextHandler(io.Discard, nil)) +} + +// fakeStore enforces tenant scoping the same way store.go's real +// pgx-backed Store does (WHERE tenant_id = ...) -- a lookup for the +// right host under the wrong tenant behaves exactly like a missing +// host, never a distinguishable "found but wrong tenant" error, so +// handler_test.go's tenant-scoping tests exercise real behavior. +type fakeStore struct { + agents map[string]*Agent // keyed by tenantID+"/"+host +} + +func newFakeStore() *fakeStore { + return &fakeStore{agents: map[string]*Agent{}} +} + +func (f *fakeStore) put(a Agent) { + f.agents[a.TenantID+"/"+a.Host] = &a +} + +func (f *fakeStore) List(_ context.Context, tenantID string) ([]Agent, error) { + var out []Agent + for _, a := range f.agents { + if a.TenantID == tenantID { + out = append(out, *a) + } + } + return out, nil +} + +func (f *fakeStore) Get(_ context.Context, tenantID, host string) (*Agent, error) { + a, ok := f.agents[tenantID+"/"+host] + if !ok { + return nil, ErrNotFound + } + cp := *a + return &cp, nil +} + +func (f *fakeStore) SetOverride(_ context.Context, tenantID, host string, override ConfigOverride, updatedBy string) (*Agent, error) { + a, ok := f.agents[tenantID+"/"+host] + if !ok { + return nil, ErrNotFound + } + a.DesiredOverride = &override + a.DesiredOverrideVersion = "v-test" + a.Pending = true + a.UpdatedBy = updatedBy + cp := *a + return &cp, nil +} + +func (f *fakeStore) ClearOverride(_ context.Context, tenantID, host string) error { + a, ok := f.agents[tenantID+"/"+host] + if !ok { + return ErrNotFound + } + a.DesiredOverride = nil + a.DesiredOverrideVersion = "" + a.Pending = false + a.UpdatedBy = "" + return nil +} + +func newTestHandler(s *fakeStore) *Handler { + return NewHandler(discardLogger(), s, nil) +} + +func doRequest(t *testing.T, h *Handler, method, path string, body any) *httptest.ResponseRecorder { + t.Helper() + var req *http.Request + if body != nil { + b, err := json.Marshal(body) + if err != nil { + t.Fatalf("marshaling request body: %v", err) + } + req = httptest.NewRequest(method, path, bytes.NewReader(b)) + } else { + req = httptest.NewRequest(method, path, nil) + } + rec := httptest.NewRecorder() + mux := http.NewServeMux() + h.RegisterRoutes(mux) + mux.ServeHTTP(rec, req) + return rec +} + +func TestHandleListScopesToTenant(t *testing.T) { + s := newFakeStore() + s.put(Agent{TenantID: "default", Host: "web-01"}) + s.put(Agent{TenantID: "acme", Host: "web-02"}) + h := newTestHandler(s) + + rec := doRequest(t, h, "GET", "/agents", nil) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200", rec.Code) + } + var got []Agent + if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatalf("decoding response: %v", err) + } + if len(got) != 1 || got[0].Host != "web-01" { + t.Fatalf("unexpected list: %+v", got) + } +} + +func TestHandleGetNotFound(t *testing.T) { + h := newTestHandler(newFakeStore()) + rec := doRequest(t, h, "GET", "/agents/nope", nil) + if rec.Code != http.StatusNotFound { + t.Fatalf("status = %d, want 404", rec.Code) + } +} + +func TestHandleGetCrossTenantIsNotFound(t *testing.T) { + s := newFakeStore() + s.put(Agent{TenantID: "acme", Host: "web-01"}) + h := newTestHandler(s) // default tenant (no authorizer/identity) + + rec := doRequest(t, h, "GET", "/agents/web-01", nil) + if rec.Code != http.StatusNotFound { + t.Fatalf("status = %d, want 404 (agent belongs to a different tenant)", rec.Code) + } +} + +func TestHandleSetConfigRoundTrips(t *testing.T) { + s := newFakeStore() + s.put(Agent{TenantID: "default", Host: "web-01"}) + h := newTestHandler(s) + + interval := int64(30000) + rec := doRequest(t, h, "PUT", "/agents/web-01/config", ConfigOverride{HeartbeatIntervalMS: &interval}) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200, body=%s", rec.Code, rec.Body.String()) + } + var got Agent + if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatalf("decoding response: %v", err) + } + if got.DesiredOverride == nil || got.DesiredOverride.HeartbeatIntervalMS == nil || *got.DesiredOverride.HeartbeatIntervalMS != 30000 { + t.Fatalf("unexpected override: %+v", got.DesiredOverride) + } + if !got.Pending { + t.Fatal("expected pending=true right after setting a new override") + } +} + +func TestHandleSetConfigRejectsTooSmallHeartbeatInterval(t *testing.T) { + s := newFakeStore() + s.put(Agent{TenantID: "default", Host: "web-01"}) + h := newTestHandler(s) + + tooSmall := int64(100) + rec := doRequest(t, h, "PUT", "/agents/web-01/config", ConfigOverride{HeartbeatIntervalMS: &tooSmall}) + if rec.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want 400", rec.Code) + } +} + +func TestHandleSetConfigUnknownHostIsNotFound(t *testing.T) { + h := newTestHandler(newFakeStore()) + interval := int64(30000) + rec := doRequest(t, h, "PUT", "/agents/nope/config", ConfigOverride{HeartbeatIntervalMS: &interval}) + if rec.Code != http.StatusNotFound { + t.Fatalf("status = %d, want 404", rec.Code) + } +} + +func TestHandleClearConfig(t *testing.T) { + s := newFakeStore() + s.put(Agent{TenantID: "default", Host: "web-01", Pending: true}) + h := newTestHandler(s) + + rec := doRequest(t, h, "DELETE", "/agents/web-01/config", nil) + if rec.Code != http.StatusNoContent { + t.Fatalf("status = %d, want 204", rec.Code) + } + if s.agents["default/web-01"].Pending { + t.Fatal("expected override to be cleared") + } +} + +func TestRequireEditorRoleForConfigWrites(t *testing.T) { + s := newFakeStore() + s.put(Agent{TenantID: "default", Host: "web-01"}) + authorizer := fakeAuthorizer{role: authz.RoleViewer} + h := NewHandler(discardLogger(), s, authorizer) + + interval := int64(30000) + rec := doRequest(t, h, "PUT", "/agents/web-01/config", ConfigOverride{HeartbeatIntervalMS: &interval}) + if rec.Code != http.StatusForbidden { + t.Fatalf("status = %d, want 403 (Viewer must not be able to edit agent config)", rec.Code) + } +} + +type fakeAuthorizer struct { + role authz.Role +} + +func (f fakeAuthorizer) Authorize(*http.Request) (authz.Identity, error) { + return authz.Identity{TenantID: "default", UserID: "u1", Role: f.role}, nil +} diff --git a/api/agents/store.go b/api/agents/store.go new file mode 100644 index 0000000..4e95403 --- /dev/null +++ b/api/agents/store.go @@ -0,0 +1,209 @@ +// Package agents is the web-facing half of agent inventory/remote +// config (see /docs/agent-management-design.md) -- reads/writes the +// same `agents` table ingest's internal/agentregistry writes on every +// CheckIn RPC, the same shared-schema-different-services shape +// alerting and api already use for dashboards/alert_rules. +package agents + +import ( + "context" + "encoding/json" + "errors" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +var ErrNotFound = errors.New("not found") + +// ConfigOverride is the remotely-editable subset of an agent's config -- +// a plain-Go mirror of ingest/internal/agentregistry's overrideFields +// and agent_control.proto's DesiredOverride. Deliberately duplicated +// rather than imported across the module boundary, same convention as +// every other cross-module shared shape in this codebase (see +// grpcserver.TenantIDHeaderKey, enterprise/internal/apiconfig.AIConfig). +// Keep the three in sync by hand. +type ConfigOverride struct { + BatchMaxSize *int64 `json:"batch_max_size,omitempty"` + BatchFlushIntervalMS *int64 `json:"batch_flush_interval_ms,omitempty"` + HeartbeatEnabled *bool `json:"heartbeat_enabled,omitempty"` + HeartbeatIntervalMS *int64 `json:"heartbeat_interval_ms,omitempty"` + JournaldUnit *string `json:"journald_unit,omitempty"` +} + +type Agent struct { + ID string `json:"id"` + TenantID string `json:"tenant_id"` + Host string `json:"host"` + Service string `json:"service"` + AgentVersion string `json:"agent_version"` + SourceKind string `json:"source_kind"` + SourceDetail string `json:"source_detail"` + BatchMaxSize int64 `json:"batch_max_size"` + BatchFlushIntervalMS int64 `json:"batch_flush_interval_ms"` + HeartbeatEnabled bool `json:"heartbeat_enabled"` + HeartbeatIntervalMS int64 `json:"heartbeat_interval_ms"` + FirstSeenAt time.Time `json:"first_seen_at"` + LastSeenAt time.Time `json:"last_seen_at"` + DesiredOverride *ConfigOverride `json:"desired_override,omitempty"` + DesiredOverrideVersion string `json:"desired_override_version,omitempty"` + AppliedOverrideVersion string `json:"applied_override_version"` + // Pending is computed, not stored: an override exists + // (DesiredOverrideVersion != "") that the agent hasn't reported + // applying yet (AppliedOverrideVersion doesn't match). This is what + // the web UI's "pending"/"applied" indicator (task selected: + // "+ Remote config editing") reads directly, rather than + // recomputing the same string comparison itself. + Pending bool `json:"pending"` + UpdatedBy string `json:"updated_by,omitempty"` +} + +type Store struct { + pool *pgxpool.Pool +} + +func NewStore(pool *pgxpool.Pool) *Store { + return &Store{pool: pool} +} + +const selectColumns = ` + id, tenant_id, host, service, + reported_agent_version, reported_source_kind, reported_source_detail, + reported_batch_max_size, reported_batch_flush_ms, + reported_heartbeat_on, reported_heartbeat_ms, + first_seen_at, last_seen_at, + desired_override, desired_override_version, applied_override_version, updated_by` + +func (s *Store) List(ctx context.Context, tenantID string) ([]Agent, error) { + rows, err := s.pool.Query(ctx, ` + SELECT `+selectColumns+` + FROM agents WHERE tenant_id = $1 ORDER BY host`, tenantID) + if err != nil { + return nil, err + } + defer rows.Close() + + var out []Agent + for rows.Next() { + a, err := scanAgent(rows) + if err != nil { + return nil, err + } + out = append(out, a) + } + return out, rows.Err() +} + +func (s *Store) Get(ctx context.Context, tenantID, host string) (*Agent, error) { + rows, err := s.pool.Query(ctx, ` + SELECT `+selectColumns+` + FROM agents WHERE tenant_id = $1 AND host = $2`, tenantID, host) + if err != nil { + return nil, err + } + defer rows.Close() + + if !rows.Next() { + if err := rows.Err(); err != nil { + return nil, err + } + return nil, ErrNotFound + } + a, err := scanAgent(rows) + if err != nil { + return nil, err + } + return &a, nil +} + +// SetOverride writes a new desired override for host, generating a +// fresh version stamp -- overwrites any previous override wholesale +// (this is "set the desired config," not "patch a few fields into +// whatever was there," so a caller building a partial edit must have +// already merged it against the current value, same as any other PUT +// endpoint in this codebase). Returns ErrNotFound if the agent has +// never checked in (nothing to target an override at yet -- an +// override for a host ingest has never seen would be silently +// unreachable). +func (s *Store) SetOverride(ctx context.Context, tenantID, host string, override ConfigOverride, updatedBy string) (*Agent, error) { + version := newVersion() + data, err := json.Marshal(override) + if err != nil { + return nil, err + } + tag, err := s.pool.Exec(ctx, ` + UPDATE agents SET desired_override = $1, desired_override_version = $2, updated_by = $3 + WHERE tenant_id = $4 AND host = $5`, + data, version, updatedBy, tenantID, host) + if err != nil { + return nil, err + } + if tag.RowsAffected() == 0 { + return nil, ErrNotFound + } + return s.Get(ctx, tenantID, host) +} + +// ClearOverride reverts an agent to running its local agent.toml +// untouched -- the next CheckIn gets has_override=false. +func (s *Store) ClearOverride(ctx context.Context, tenantID, host string) error { + tag, err := s.pool.Exec(ctx, ` + UPDATE agents SET desired_override = NULL, desired_override_version = NULL, updated_by = NULL + WHERE tenant_id = $1 AND host = $2`, tenantID, host) + if err != nil { + return err + } + if tag.RowsAffected() == 0 { + return ErrNotFound + } + return nil +} + +type rowScanner interface { + Scan(dest ...any) error +} + +func scanAgent(row rowScanner) (Agent, error) { + var a Agent + var desiredOverride []byte + var desiredVersion, updatedBy *string + if err := row.Scan( + &a.ID, &a.TenantID, &a.Host, &a.Service, + &a.AgentVersion, &a.SourceKind, &a.SourceDetail, + &a.BatchMaxSize, &a.BatchFlushIntervalMS, + &a.HeartbeatEnabled, &a.HeartbeatIntervalMS, + &a.FirstSeenAt, &a.LastSeenAt, + &desiredOverride, &desiredVersion, &a.AppliedOverrideVersion, &updatedBy, + ); err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return Agent{}, ErrNotFound + } + return Agent{}, err + } + if updatedBy != nil { + a.UpdatedBy = *updatedBy + } + if desiredVersion != nil { + a.DesiredOverrideVersion = *desiredVersion + a.Pending = *desiredVersion != a.AppliedOverrideVersion + if len(desiredOverride) > 0 { + var override ConfigOverride + if err := json.Unmarshal(desiredOverride, &override); err != nil { + return Agent{}, err + } + a.DesiredOverride = &override + } + } + return a, nil +} + +// newVersion is an opaque, monotonically-informative-enough stamp for +// DesiredOverride.version -- a timestamp, not a counter, since Store +// has no prior version to increment from without an extra read. Never +// interpreted as a real time value by the agent (see +// agent_control.proto's DesiredOverride.version comment) -- just needs +// to change on every edit. +func newVersion() string { + return time.Now().UTC().Format(time.RFC3339Nano) +} diff --git a/api/cmd/api/main.go b/api/cmd/api/main.go index 0bed930..222e1cb 100644 --- a/api/cmd/api/main.go +++ b/api/cmd/api/main.go @@ -19,6 +19,7 @@ import ( "github.com/ClickHouse/clickhouse-go/v2" "github.com/jackc/pgx/v5/pgxpool" + "github.com/sentry/sentry/api/agents" "github.com/sentry/sentry/api/ai/aiapi" "github.com/sentry/sentry/api/ai/grounding" "github.com/sentry/sentry/api/ai/provider/ollama" @@ -119,6 +120,13 @@ func main() { // canEditDashboard's nil-permissions fallback -- only the "granted" // half of the matrix's "(own/granted)" qualifier is unavailable here. dashboardsHandler := dashboards.NewHandler(logger, dashboards.NewStore(pgPool), authorizer, nil) + // Same pgPool as dashboards above -- agents reads/writes the table + // ingest's internal/agentregistry upserts on every CheckIn RPC (see + // /docs/agent-management-design.md). Nothing here requires + // AGENT_REGISTRY_POSTGRES_ADDR to be set on ingest; these routes work + // unconditionally, they'll just show an empty inventory if ingest + // hasn't been configured to record check-ins. + agentsHandler := agents.NewHandler(logger, agents.NewStore(pgPool), authorizer) // One shared mux, CORS applied once around the whole thing -- see // httpserver's doc comment for why this changed from each @@ -126,6 +134,7 @@ func main() { mux := http.NewServeMux() queryHandler.RegisterRoutes(mux) dashboardsHandler.RegisterRoutes(mux) + agentsHandler.RegisterRoutes(mux) // AI routes (Phase 7) are only registered at all when OLLAMA_BASE_URL // is set -- an unconfigured deployment gets a plain 404 on /ai/* diff --git a/docker-compose.yml b/docker-compose.yml index 741deed..91293ba 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -162,6 +162,8 @@ services: condition: service_completed_successfully clickhouse-migrate: condition: service_completed_successfully + metadata-migrate: + condition: service_completed_successfully ports: - "4317:4317" # gRPC, mTLS — this is what the host-run agent connects to environment: @@ -181,6 +183,17 @@ services: # via `enterprise-auth -create-ingest-credential-tenant=`) or # be refused outright -- not turned on here since nothing in this # compose file provisions one. + # + # AGENT_REGISTRY_POSTGRES_ADDR enables agent inventory/remote + # config (see /docs/agent-management-design.md) -- same "sentry" + # shared Postgres role api/dashboards already uses (agent + # inventory carries no tamper-evidence requirement, unlike + # audit_log's dedicated restricted role). Set here (unlike + # ENTERPRISE_AUTH_URL above) since this feature has no multi- + # tenancy prerequisite -- it works the same in single-tenant core. + AGENT_REGISTRY_POSTGRES_ADDR: "metadata-postgres:5432" + AGENT_REGISTRY_POSTGRES_USERNAME: "sentry" + AGENT_REGISTRY_POSTGRES_PASSWORD: "sentry-dev-only" volumes: - ./hack/dev-certs/out:/etc/sentry-ingest:ro diff --git a/docs/agent-management-design.md b/docs/agent-management-design.md new file mode 100644 index 0000000..2baaf26 --- /dev/null +++ b/docs/agent-management-design.md @@ -0,0 +1,182 @@ +# Agent inventory, management, and remote config + +Extends `/docs/agent-heartbeat-monitoring.md` (heartbeat + absence +alerting). This adds three things to the web UI: an inventory view of +every agent that's checked in, read-only visibility into each agent's +actual running config, and the ability to remotely edit a narrow, +deliberately-scoped subset of that config. + +## Scope decision (confirmed before building) + +"Agent management" could mean several different things with very +different risk profiles. Confirmed up front: this covers inventory, +config visibility, and remote config *editing* — explicitly **not** +remote lifecycle commands (restart/stop/uninstall). That's a real +command-and-control channel across every managed host and deserves its +own security design (signed commands, strict RBAC, full audit trail) +before it's built, not something to fold in as a side effect of a +config-editing feature. + +## Why this is still pull, not push, on the wire + +The agent's transport is unchanged and still push-only: it dials +*out* to `ingest` over mTLS; nothing in the platform ever reaches into +an agent. See `/docs/agent-heartbeat-monitoring.md`'s "Design: why this +is a heartbeat, not a true pull" section for the full reasoning (NAT/ +dynamic-IP tolerance, no inbound port on any remote host). Remote config +editing extends that same posture: an operator's edit doesn't get +pushed to the agent at the moment it's saved. It sits in Postgres until +the agent's own next scheduled check-in asks "what should I be +running" and picks it up. The web UI's "pending" vs. "applied" badge +exists specifically to make that asynchrony visible rather than +implying something more immediate. + +## Wire shape: `AgentControl.CheckIn` + +New proto (`proto/sentry/agent/v1/agent_control.proto`), a second gRPC +service on the exact same mTLS channel/listener `LogIngest.PushBatch` +already uses — not a second protocol or connection the agent has to +maintain. Called on the agent's own heartbeat ticker (see +`agent/sentry-agent/src/main.rs`'s `heartbeat_ticker` arm), +independently of whether the heartbeat log record itself is enabled — +CheckIn keeps running even with `heartbeat.enabled = false`, since +that's an agent's only path to ever receive a remote override that +re-enables it. + +- **Request**: host/service identity, a `ReportedConfig` snapshot of + what the agent is actually running (version, source kind/detail, + batch settings, heartbeat settings), and the version of the last + override this agent successfully applied (empty if never). +- **Response**: `has_override` plus, when true, a `DesiredOverride` — + every field optional (unset = "no change to this field, keep local + config"), each independently overridable. + +## What's remotely editable, and what deliberately isn't + +Editable: `batch.max_size`, `batch.flush_interval_ms`, +`heartbeat.enabled`, `heartbeat.interval`, and — only when the agent's +local source is journald — the unit filter. + +**Never editable, permanently**: TLS material and the ingest endpoint. +`ReportedConfig` doesn't even carry these fields, and +`DesiredOverride` has no fields for them at all — this isn't a +validation rule that could be relaxed later, it's a shape decision. +Two reasons, both serious enough that this needed to be a design +boundary rather than a judgment call per edit: + +1. **A bad edit could permanently strand an agent.** Point an agent's + `ingest.endpoint` at an address that doesn't exist, or corrupt its + TLS config, and it can never call `CheckIn` again to receive a + correction — the one channel capable of fixing the mistake would be + exactly what broke. Every other editable field is safe by + construction: even a bad heartbeat interval or an over-aggressive + batch size degrades the agent's behavior without ever cutting off + its ability to receive the next correction. +2. **A compromised web session/API credential must not be able to + redirect where an agent's logs go.** If `ingest.endpoint` were + editable, an attacker with write access to this feature could point + agents at an address they control and exfiltrate log data. Keeping + connection details local-file-only means this feature's blast + radius is "an agent's operational tuning gets messed with," never + "an agent's data goes somewhere else." + +`validateOverride` (`api/agents/handler.go`) additionally floors +`batch_max_size >= 1`, `batch_flush_interval_ms >= 100`, and +`heartbeat_interval_ms >= 5000` — the same kind of real, found-by- +building-it floor as alerting's `eval_interval_seconds >= 30`, here to +stop a fat-fingered edit from telling an agent to flush constantly or +heartbeat constantly. + +## Merge semantics: an override is a live layer, not a rewrite + +The agent's local `agent.toml` is never rewritten. A remote override +lives only in the running process's memory +(`agent/sentry-agent/src/main.rs`'s `apply_override`) and is re-applied +fresh on every check-in that returns one — a restarted agent boots from +`agent.toml` alone and re-syncs whatever override is still set on its +next successful check-in. This was a deliberate simplicity choice over +persisting the override to disk: it avoids needing filesystem write +access on every managed host (not guaranteed, e.g. read-only base +images) and avoids a whole "reconcile a locally-cached override against +a freshly-fetched one at startup" state machine. The cost is that an +agent's effective config isn't fully recoverable from `agent.toml` +alone while an override is active — acceptable, since the web UI's +`GET /agents/{host}` is the source of truth for "what is this agent +actually running" regardless. + +Applying an override that changes `batch_max_size`/ +`batch_flush_interval_ms` rebuilds the `Batcher`/flush ticker outright. +Whatever was buffered under the old settings is flushed first +(`Batcher::flush_all`, new) rather than dropped — building this exposed +a real, independent, pre-existing bug: agent shutdown was calling +`poll_timeout()`, which only drains when `flush_interval` has already +elapsed, meaning records buffered more recently than that were silently +lost on every graceful shutdown that happened to land between flushes. +Fixed alongside this feature (`flush_all()` now used at both shutdown +and hot-reload) since it's the exact same correctness property in both +places. + +Changing the journald unit filter is the one override that can't just +swap a struct field — there's no way to change what +`source::journald::run` is tailing without restarting that task. Applying +it aborts the current source task and respawns a fresh one with the new +filter, swapping the channel `main.rs` reads from. Source *kind* +(journald vs. file vs. eventlog vs. etw) is never remotely switchable — +only narrowing/widening the filter within whatever source the host is +already configured for. + +## Data model + +`metadata/migrations/0037_create_agents.sql`: one `agents` table, one +row per `(tenant_id, host)`, in the same `sentry_metadata` Postgres +dashboards/alert_rules already live in — not a new database, matching +this project's established "shared schema, different services own +different tables" shape. `tenant_id` defaults to `'default'` for +single-tenant deployments with no `TenantResolver` configured on +ingest, same as every other tenant-scoped table since Phase 3. + +Two services write to it, cleanly split by concern: + +- **`ingest/internal/agentregistry`** (new) upserts on every `CheckIn`: + `last_seen_at`, the `reported_*` columns, and + `applied_override_version` (echoed from the agent). Reads back + whatever `desired_override`/`desired_override_version` is currently + stored to answer the RPC. Gated on `AGENT_REGISTRY_POSTGRES_ADDR` + being set — nil/off by default, same "off unless configured" shape as + `TenantResolver`; a deployment that hasn't opted in still accepts + `CheckIn` calls (agents never see an error), it just doesn't record + anything or ever return an override. +- **`api/agents`** (new) is the web-facing read/write side: `GET + /agents`, `GET /agents/{host}`, `PUT /agents/{host}/config` (replaces + the whole stored override — the web UI's edit form always reads the + current override first and submits the complete merged set, same + "PUT replaces the resource" convention every other edit form in this + codebase already uses), `DELETE /agents/{host}/config` (clears it, + reverting the agent to its local `agent.toml`). + +`ConfigOverride`'s JSON shape is duplicated three times — Go structs in +`ingest/internal/agentregistry` and `api/agents`, a proto message for +the wire — deliberately, matching this codebase's established +convention for shapes shared across module boundaries (see +`grpcserver.TenantIDHeaderKey`, `enterprise/internal/apiconfig.AIConfig`) +rather than coupling independently deployable services' builds +together. Keep the three in sync by hand. + +## RBAC + +Viewing inventory is `RoleViewer` (same bar as viewing a dashboard); +editing an agent's remote config is `RoleEditor` — treated as an +operational-tuning action matching alert rules/notification targets, +not an admin-only capability like user/role management. + +## Verified live + +See the runbook entry (task follow-up) for the full walkthrough: a real +agent binary, its heartbeat/CheckIn cadence pointed at a live +`ingest` with `AGENT_REGISTRY_POSTGRES_ADDR` configured, confirming (a) +the agent appears in `GET /agents` after its first check-in, (b) an +edit made via `PUT /agents/{host}/config` shows `pending: true` +immediately and `pending: false` after the agent's next check-in, and +(c) the edited setting (heartbeat interval) visibly takes effect in the +agent's own behavior — confirmed by the change in cadence of new +heartbeat rows landing in ClickHouse. diff --git a/enterprise/cmd/enterprise-api/main.go b/enterprise/cmd/enterprise-api/main.go index 7bbc250..2a3173a 100644 --- a/enterprise/cmd/enterprise-api/main.go +++ b/enterprise/cmd/enterprise-api/main.go @@ -42,6 +42,7 @@ import ( chdriver "github.com/ClickHouse/clickhouse-go/v2" "github.com/jackc/pgx/v5/pgxpool" + "github.com/sentry/sentry/api/agents" "github.com/sentry/sentry/api/ai/aiapi" "github.com/sentry/sentry/api/ai/provider/ollama" "github.com/sentry/sentry/api/ai/router" @@ -164,10 +165,12 @@ func main() { queryHandler := queryapi.NewHandler(logger, registry, search, cfg.QueryTimeout, auditLogger, authorizer) dashboardsHandler := dashboards.NewHandler(logger, dashboards.NewStore(pgPool), authorizer, rbacstore.NewDashboardPermissions(rbac)) + agentsHandler := agents.NewHandler(logger, agents.NewStore(pgPool), authorizer) mux := http.NewServeMux() queryHandler.RegisterRoutes(mux) // also registers GET /healthz dashboardsHandler.RegisterRoutes(mux) + agentsHandler.RegisterRoutes(mux) // Same "off unless OLLAMA_BASE_URL is set" gate as api/cmd/api -- // see that file's doc comment. diff --git a/ingest/cmd/ingest/main.go b/ingest/cmd/ingest/main.go index 8c35db7..e6889c1 100644 --- a/ingest/cmd/ingest/main.go +++ b/ingest/cmd/ingest/main.go @@ -19,10 +19,12 @@ import ( "os/signal" "syscall" + "github.com/jackc/pgx/v5/pgxpool" "golang.org/x/sync/errgroup" "github.com/sentry/sentry/ingest/clickhousewriter" "github.com/sentry/sentry/ingest/consumer" + "github.com/sentry/sentry/ingest/internal/agentregistry" "github.com/sentry/sentry/ingest/internal/config" "github.com/sentry/sentry/ingest/internal/grpcserver" "github.com/sentry/sentry/ingest/internal/producer" @@ -87,7 +89,31 @@ func main() { } else { logger.Info("ENTERPRISE_AUTH_URL not set -- ingest records carry no tenant_id, single-tenant behavior") } - srv := grpcserver.New(logger, cfg.GRPC, cfg.TLS, p, resolver) + + // agents stays nil (CheckIn always reports "no override," nothing + // recorded) unless AGENT_REGISTRY_POSTGRES_ADDR is configured -- + // same "off unless configured" shape as resolver above. Uses its + // own pgxpool rather than sharing one across mode=server/consumer + // -- consumer's half of this binary has no Postgres dependency at + // all today and shouldn't gain one just because server's did. + var agents grpcserver.AgentRegistry + if cfg.AgentRegistry.Postgres.Addr != "" { + dsn := fmt.Sprintf("postgres://%s:%s@%s/%s", + cfg.AgentRegistry.Postgres.Username, cfg.AgentRegistry.Postgres.Password, + cfg.AgentRegistry.Postgres.Addr, cfg.AgentRegistry.Postgres.Database) + pool, err := pgxpool.New(ctx, dsn) + if err != nil { + logger.Error("opening agent registry postgres pool", "error", err) + os.Exit(1) + } + defer pool.Close() + agents = agentregistry.New(pool) + logger.Info("agent registry configured", "postgres_addr", cfg.AgentRegistry.Postgres.Addr) + } else { + logger.Info("AGENT_REGISTRY_POSTGRES_ADDR not set -- agent check-ins are accepted but not recorded, no remote config") + } + + srv := grpcserver.New(logger, cfg.GRPC, cfg.TLS, p, resolver, agents) g.Go(func() error { return srv.Run(ctx) }) } diff --git a/ingest/go.mod b/ingest/go.mod index f801337..db2f97a 100644 --- a/ingest/go.mod +++ b/ingest/go.mod @@ -20,6 +20,10 @@ require ( github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/go-faster/city v1.0.1 // indirect github.com/go-faster/errors v0.7.1 // indirect + github.com/jackc/pgpassfile v1.0.0 // indirect + github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect + github.com/jackc/pgx/v5 v5.10.0 // indirect + github.com/jackc/puddle/v2 v2.2.2 // indirect github.com/klauspost/compress v1.19.1 // indirect github.com/paulmach/orb v0.13.0 // indirect github.com/pierrec/lz4/v4 v4.1.27 // indirect diff --git a/ingest/go.sum b/ingest/go.sum index 3d7b4e9..f5a3b85 100644 --- a/ingest/go.sum +++ b/ingest/go.sum @@ -6,6 +6,7 @@ github.com/andybalholm/brotli v1.2.2 h1:HzTuoo2ErYQqf5qvcJInB8uvqSVxRttzkFexPWtn github.com/andybalholm/brotli v1.2.2/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY= github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/go-faster/city v1.0.1 h1:4WAxSZ3V2Ws4QRDrscLEDcibJY8uf41H6AhXDrNDcGw= @@ -22,12 +23,21 @@ github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= +github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= +github.com/jackc/pgx/v5 v5.10.0 h1:VhSvgU2jSli8o3AqIEOTJr7rZwAEUVo4E4XhR94Zfr0= +github.com/jackc/pgx/v5 v5.10.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= +github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= +github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= github.com/klauspost/compress v1.19.1 h1:VsB4HPswih7mmZ8WleSFQ75c/Ui1M4trX5oAsJnhSlk= github.com/klauspost/compress v1.19.1/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= github.com/paulmach/orb v0.13.0 h1:r7n7mQGGF+cj/CbcivEj9J3HGK+XR+yXnvzRdq9saIw= github.com/paulmach/orb v0.13.0/go.mod h1:6scRWINywA2Jf05dcjOfLfxrUIMECvTSG2MVbRLxu/k= github.com/pierrec/lz4/v4 v4.1.27 h1:+PhzhWDrjRj89TH2sw43nE3+4+W8lSxIuQadEHZyjUk= github.com/pierrec/lz4/v4 v4.1.27/go.mod h1:EoQMVJgeeEOMsCqCzqFm2O0cJvljX2nGZjcRIPL34O4= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U= github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/segmentio/asm v1.2.1 h1:DTNbBqs57ioxAD4PrArqftgypG4/qNpXoJx8TVXxPR0= @@ -36,6 +46,9 @@ github.com/segmentio/kafka-go v0.4.51 h1:JgDPPG75tC1rWIS2Me6MwcvXJ6f49UQ4HjAOef7 github.com/segmentio/kafka-go v0.4.51/go.mod h1:Y1gn60kzLEEaW28YshXyk2+VCUKbJ3Qr6DrnT3i4+9E= github.com/shopspring/decimal v1.4.0 h1:bxl37RwXBklmTi0C79JfXCEBD1cqqHt0bbgBAGFp81k= github.com/shopspring/decimal v1.4.0/go.mod h1:gawqmDU56v4yIKSwfBSFip1HdCCXN8/+DMd9qYNcwME= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= github.com/xdg-go/pbkdf2 v1.0.0 h1:Su7DPu48wXMwC3bs7MCNG+z4FhcyEuz5dlvchbq0B0c= @@ -74,5 +87,7 @@ google.golang.org/grpc v1.83.0 h1:JeNZEKJFbQxArAMl+hiytHauacDNqJUllNfmIMmpqnQ= google.golang.org/grpc v1.83.0/go.mod h1:kDyl6SKsiHKt0uylY5gtn5cEjkrIOhQOGDgIc4JGwzQ= google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc= google.golang.org/protobuf v1.36.12/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/ingest/internal/agentregistry/agentregistry.go b/ingest/internal/agentregistry/agentregistry.go new file mode 100644 index 0000000..c3b4724 --- /dev/null +++ b/ingest/internal/agentregistry/agentregistry.go @@ -0,0 +1,108 @@ +// Package agentregistry is the Postgres-backed implementation of +// grpcserver.AgentRegistry -- ingest's half of agent inventory/remote +// config (see /docs/agent-management-design.md). Writes into the same +// sentry_metadata Postgres api reads/writes from for the web UI's +// inventory and edit-config views (api/agents), the same shared-schema- +// different-services shape alerting and api already use for dashboards/ +// alert_rules. +package agentregistry + +import ( + "context" + "encoding/json" + "fmt" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/sentry/sentry/ingest/internal/grpcserver" +) + +// defaultTenantID is used when no TenantResolver is configured (empty +// tenantID from grpcserver) -- the same 'default' tenant row every +// single-tenant deployment's Postgres already has, seeded by +// metadata/migrations/0019_seed_default_tenant.sql. +const defaultTenantID = "default" + +// overrideFields is the JSON shape stored in agents.desired_override. +// Deliberately duplicated in api/agents rather than imported -- these +// are two different Go modules, and this codebase's established +// convention (see enterprise/internal/apiconfig.AIConfig, +// grpcserver.TenantIDHeaderKey) is to duplicate a small shared shape +// across a module boundary rather than couple two independently +// deployable services' builds together. Keep the two in sync by hand. +type overrideFields struct { + BatchMaxSize *uint64 `json:"batch_max_size,omitempty"` + BatchFlushIntervalMS *uint64 `json:"batch_flush_interval_ms,omitempty"` + HeartbeatEnabled *bool `json:"heartbeat_enabled,omitempty"` + HeartbeatIntervalMS *uint64 `json:"heartbeat_interval_ms,omitempty"` + JournaldUnit *string `json:"journald_unit,omitempty"` +} + +type Registry struct { + pool *pgxpool.Pool +} + +func New(pool *pgxpool.Pool) *Registry { + return &Registry{pool: pool} +} + +var _ grpcserver.AgentRegistry = (*Registry)(nil) + +func (r *Registry) CheckIn(ctx context.Context, tenantID string, info grpcserver.AgentCheckIn) (grpcserver.AgentOverride, error) { + if tenantID == "" { + tenantID = defaultTenantID + } + + var ( + desiredOverride []byte + desiredVersion *string + ) + err := r.pool.QueryRow(ctx, ` + INSERT INTO agents ( + id, tenant_id, host, service, + reported_agent_version, reported_source_kind, reported_source_detail, + reported_batch_max_size, reported_batch_flush_ms, + reported_heartbeat_on, reported_heartbeat_ms, + first_seen_at, last_seen_at, applied_override_version + ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11, now(), now(), $12) + ON CONFLICT (tenant_id, host) DO UPDATE SET + service = EXCLUDED.service, + reported_agent_version = EXCLUDED.reported_agent_version, + reported_source_kind = EXCLUDED.reported_source_kind, + reported_source_detail = EXCLUDED.reported_source_detail, + reported_batch_max_size = EXCLUDED.reported_batch_max_size, + reported_batch_flush_ms = EXCLUDED.reported_batch_flush_ms, + reported_heartbeat_on = EXCLUDED.reported_heartbeat_on, + reported_heartbeat_ms = EXCLUDED.reported_heartbeat_ms, + last_seen_at = now(), + applied_override_version = EXCLUDED.applied_override_version + RETURNING desired_override, desired_override_version`, + uuid.NewString(), tenantID, info.Host, info.Service, + info.AgentVersion, info.SourceKind, info.SourceDetail, + info.BatchMaxSize, info.BatchFlushIntervalMS, + info.HeartbeatEnabled, info.HeartbeatIntervalMS, + info.AppliedOverrideVersion, + ).Scan(&desiredOverride, &desiredVersion) + if err != nil { + return grpcserver.AgentOverride{}, fmt.Errorf("agentregistry: upserting check-in: %w", err) + } + + if desiredVersion == nil || len(desiredOverride) == 0 { + return grpcserver.AgentOverride{HasOverride: false}, nil + } + + var fields overrideFields + if err := json.Unmarshal(desiredOverride, &fields); err != nil { + return grpcserver.AgentOverride{}, fmt.Errorf("agentregistry: parsing stored override: %w", err) + } + return grpcserver.AgentOverride{ + HasOverride: true, + BatchMaxSize: fields.BatchMaxSize, + BatchFlushIntervalMS: fields.BatchFlushIntervalMS, + HeartbeatEnabled: fields.HeartbeatEnabled, + HeartbeatIntervalMS: fields.HeartbeatIntervalMS, + JournaldUnit: fields.JournaldUnit, + Version: *desiredVersion, + }, nil +} diff --git a/ingest/internal/config/config.go b/ingest/internal/config/config.go index 42e5326..db7209d 100644 --- a/ingest/internal/config/config.go +++ b/ingest/internal/config/config.go @@ -23,6 +23,26 @@ type Config struct { // as every other optional enterprise integration point in this // codebase (e.g. api's own ENTERPRISE_AUTH_URL). EnterpriseAuthURL string + // AgentRegistry enables agent inventory/remote config + // (internal/agentregistry, internal/grpcserver.AgentRegistry) when + // Postgres.Addr is set -- same "off unless configured" shape as + // EnterpriseAuthURL above. Writes into the same sentry_metadata + // database api/web already use, via the same shared "sentry" role + // every other non-audit table in this schema uses (unlike + // audit_log's dedicated restricted role -- agent inventory carries + // no tamper-evidence requirement). + AgentRegistry AgentRegistryConfig +} + +type AgentRegistryConfig struct { + Postgres PostgresConfig +} + +type PostgresConfig struct { + Addr string + Database string + Username string + Password string } type GRPCConfig struct { @@ -77,6 +97,14 @@ func Load() (Config, error) { Password: getenv("CLICKHOUSE_PASSWORD", ""), }, EnterpriseAuthURL: getenv("ENTERPRISE_AUTH_URL", ""), + AgentRegistry: AgentRegistryConfig{ + Postgres: PostgresConfig{ + Addr: getenv("AGENT_REGISTRY_POSTGRES_ADDR", ""), + Database: getenv("AGENT_REGISTRY_POSTGRES_DATABASE", "sentry_metadata"), + Username: getenv("AGENT_REGISTRY_POSTGRES_USERNAME", "sentry"), + Password: getenv("AGENT_REGISTRY_POSTGRES_PASSWORD", ""), + }, + }, } maxSize, err := strconv.Atoi(getenv("CONSUMER_BATCH_MAX_SIZE", "500")) diff --git a/ingest/internal/grpcserver/server.go b/ingest/internal/grpcserver/server.go index 7012515..a00af08 100644 --- a/ingest/internal/grpcserver/server.go +++ b/ingest/internal/grpcserver/server.go @@ -39,6 +39,7 @@ import ( "google.golang.org/protobuf/proto" "github.com/sentry/sentry/ingest/internal/config" + agentv1 "github.com/sentry/sentry/proto/sentry/agent/v1" logsv1 "github.com/sentry/sentry/proto/sentry/logs/v1" ) @@ -52,12 +53,14 @@ const TenantIDHeaderKey = "tenant_id" type Server struct { logsv1.UnimplementedLogIngestServer + agentv1.UnimplementedAgentControlServer logger *slog.Logger grpcCfg config.GRPCConfig tlsCfg config.TLSConfig producer batchProducer resolver TenantResolver + agents AgentRegistry } // batchProducer is the subset of *producer.Producer this package depends @@ -80,8 +83,52 @@ type TenantResolver interface { ResolveTenant(ctx context.Context, token string) (tenantID string, err error) } -func New(logger *slog.Logger, grpcCfg config.GRPCConfig, tlsCfg config.TLSConfig, p batchProducer, resolver TenantResolver) *Server { - return &Server{logger: logger, grpcCfg: grpcCfg, tlsCfg: tlsCfg, producer: p, resolver: resolver} +// AgentRegistry records an agent's CheckIn (for the web UI's inventory +// view) and returns any remote config override an operator has set for +// it. nil is a deliberate no-op, same "off unless configured" shape as +// TenantResolver: CheckIn always succeeds and reports "no override" -- +// a deployment that hasn't configured AGENT_REGISTRY_POSTGRES_ADDR +// simply doesn't get agent inventory/management, exactly like one +// without ENTERPRISE_AUTH_URL doesn't get tenant-tagged records. +type AgentRegistry interface { + CheckIn(ctx context.Context, tenantID string, info AgentCheckIn) (AgentOverride, error) +} + +// AgentCheckIn is what an agent reports about itself on each CheckIn -- +// a plain-Go mirror of agentv1.ReportedConfig plus the identity/ +// tenant fields, kept separate from the proto type so AgentRegistry +// implementations (ingest/internal/agentregistry) don't need to import +// this package's gRPC-facing types just to satisfy the interface. +type AgentCheckIn struct { + Host string + Service string + AgentVersion string + SourceKind string + SourceDetail string + BatchMaxSize uint64 + BatchFlushIntervalMS uint64 + HeartbeatEnabled bool + HeartbeatIntervalMS uint64 + AppliedOverrideVersion string +} + +// AgentOverride is the remotely-editable subset of an agent's config, as +// currently stored for it -- a plain-Go mirror of agentv1.DesiredOverride. +// Every pointer field is nil when that field has no override set. +// HasOverride false means no override has ever been set at all (Version +// is meaningless in that case). +type AgentOverride struct { + HasOverride bool + BatchMaxSize *uint64 + BatchFlushIntervalMS *uint64 + HeartbeatEnabled *bool + HeartbeatIntervalMS *uint64 + JournaldUnit *string + Version string +} + +func New(logger *slog.Logger, grpcCfg config.GRPCConfig, tlsCfg config.TLSConfig, p batchProducer, resolver TenantResolver, agents AgentRegistry) *Server { + return &Server{logger: logger, grpcCfg: grpcCfg, tlsCfg: tlsCfg, producer: p, resolver: resolver, agents: agents} } // Run blocks serving gRPC until ctx is canceled, then gracefully stops. @@ -98,6 +145,7 @@ func (s *Server) Run(ctx context.Context) error { grpcSrv := grpc.NewServer(grpc.Creds(credentials.NewTLS(tlsConf))) logsv1.RegisterLogIngestServer(grpcSrv, s) + agentv1.RegisterAgentControlServer(grpcSrv, s) s.logger.Info("gRPC server listening", "addr", s.grpcCfg.ListenAddr) @@ -118,26 +166,10 @@ func (s *Server) PushBatch(ctx context.Context, req *logsv1.PushBatchRequest) (* return &logsv1.PushBatchResponse{Accepted: 0}, nil } - // tenantID stays empty (no header attached below) unless a resolver - // is actually configured -- single-tenant deployments never present - // a bearer credential and never need to. Once a resolver IS - // configured, a missing/invalid credential fails the whole batch - // closed rather than falling back to "no tenant" -- exactly the - // same fail-closed shape enterprise/internal/chrunner.Registry.RunSQL - // uses on the read side, applied here at the point data enters the - // system. - var tenantID string - if s.resolver != nil { - token, ok := bearerTokenFromContext(ctx) - if !ok { - return nil, status.Error(codes.Unauthenticated, "missing bearer credential") - } - resolved, err := s.resolver.ResolveTenant(ctx, token) - if err != nil { - s.logger.Error("resolving ingest tenant", "batch_id", req.GetBatchId(), "error", err) - return nil, status.Error(codes.Unauthenticated, "invalid ingest credential") - } - tenantID = resolved + tenantID, err := s.resolveTenant(ctx) + if err != nil { + s.logger.Error("resolving ingest tenant", "batch_id", req.GetBatchId(), "error", err) + return nil, err } msgs := make([]kafka.Message, 0, len(req.GetRecords())) @@ -174,6 +206,83 @@ func (s *Server) PushBatch(ctx context.Context, req *logsv1.PushBatchRequest) (* return &logsv1.PushBatchResponse{Accepted: uint32(len(req.GetRecords()))}, nil } +// resolveTenant is PushBatch's and CheckIn's shared tenant-resolution +// step, extracted so CheckIn gets the identical fail-closed behavior +// without duplicating it: empty tenantID (no resolver configured, the +// single-tenant default) is not an error, but a configured resolver +// that gets no/an invalid credential is -- exactly the same posture +// enterprise/internal/chrunner.Registry.RunSQL uses on the read side, +// applied here at the point data (or a check-in) enters the system. +func (s *Server) resolveTenant(ctx context.Context) (string, error) { + if s.resolver == nil { + return "", nil + } + token, ok := bearerTokenFromContext(ctx) + if !ok { + return "", status.Error(codes.Unauthenticated, "missing bearer credential") + } + resolved, err := s.resolver.ResolveTenant(ctx, token) + if err != nil { + return "", status.Error(codes.Unauthenticated, "invalid ingest credential") + } + return resolved, nil +} + +// CheckIn is AgentControl's one RPC (see agent_control.proto) -- agent- +// initiated, on its own heartbeat ticker. A nil AgentRegistry (no +// AGENT_REGISTRY_POSTGRES_ADDR configured) makes this a pure no-op that +// always reports "no override," so agents calling in against a +// deployment that hasn't opted into this feature see no behavior +// change at all. +func (s *Server) CheckIn(ctx context.Context, req *agentv1.CheckInRequest) (*agentv1.CheckInResponse, error) { + if req.GetHost() == "" { + return nil, status.Error(codes.InvalidArgument, "host must not be empty") + } + + tenantID, err := s.resolveTenant(ctx) + if err != nil { + s.logger.Error("resolving ingest tenant for check-in", "host", req.GetHost(), "error", err) + return nil, err + } + + if s.agents == nil { + return &agentv1.CheckInResponse{HasOverride: false}, nil + } + + cfg := req.GetCurrentConfig() + override, err := s.agents.CheckIn(ctx, tenantID, AgentCheckIn{ + Host: req.GetHost(), + Service: req.GetService(), + AgentVersion: cfg.GetAgentVersion(), + SourceKind: cfg.GetSourceKind(), + SourceDetail: cfg.GetSourceDetail(), + BatchMaxSize: cfg.GetBatchMaxSize(), + BatchFlushIntervalMS: cfg.GetBatchFlushIntervalMs(), + HeartbeatEnabled: cfg.GetHeartbeatEnabled(), + HeartbeatIntervalMS: cfg.GetHeartbeatIntervalMs(), + AppliedOverrideVersion: req.GetAppliedOverrideVersion(), + }) + if err != nil { + s.logger.Error("recording agent check-in", "host", req.GetHost(), "error", err) + return nil, status.Errorf(codes.Internal, "recording check-in: %v", err) + } + + if !override.HasOverride { + return &agentv1.CheckInResponse{HasOverride: false}, nil + } + return &agentv1.CheckInResponse{ + HasOverride: true, + Override: &agentv1.DesiredOverride{ + BatchMaxSize: override.BatchMaxSize, + BatchFlushIntervalMs: override.BatchFlushIntervalMS, + HeartbeatEnabled: override.HeartbeatEnabled, + HeartbeatIntervalMs: override.HeartbeatIntervalMS, + JournaldUnit: override.JournaldUnit, + Version: override.Version, + }, + }, nil +} + // bearerTokenFromContext reads the same "authorization: Bearer " // gRPC metadata shape HTTP's Authorization header uses -- an agent sets // this once per PushBatch call (see the agent's grpc.rs), not per diff --git a/ingest/internal/grpcserver/server_test.go b/ingest/internal/grpcserver/server_test.go index fec1f4b..a9a04f8 100644 --- a/ingest/internal/grpcserver/server_test.go +++ b/ingest/internal/grpcserver/server_test.go @@ -15,6 +15,7 @@ import ( "google.golang.org/protobuf/proto" "github.com/sentry/sentry/ingest/internal/config" + agentv1 "github.com/sentry/sentry/proto/sentry/agent/v1" logsv1 "github.com/sentry/sentry/proto/sentry/logs/v1" ) @@ -50,12 +51,40 @@ func (f *fakeResolver) ResolveTenant(_ context.Context, token string) (string, e return tenantID, nil } +// fakeAgentRegistry is an in-memory stand-in for +// ingest/internal/agentregistry.Registry, keyed by "tenantID/host" so +// tests can assert cross-tenant isolation the same way the real +// UNIQUE (tenant_id, host) constraint provides it. +type fakeAgentRegistry struct { + mu sync.Mutex + checkIns []AgentCheckIn + overrides map[string]AgentOverride + err error +} + +func (f *fakeAgentRegistry) CheckIn(_ context.Context, tenantID string, info AgentCheckIn) (AgentOverride, error) { + f.mu.Lock() + defer f.mu.Unlock() + if f.err != nil { + return AgentOverride{}, f.err + } + f.checkIns = append(f.checkIns, info) + if f.overrides == nil { + return AgentOverride{HasOverride: false}, nil + } + return f.overrides[tenantID+"/"+info.Host], nil +} + func newTestServer(p batchProducer) *Server { - return New(slog.New(slog.NewTextHandler(io.Discard, nil)), config.GRPCConfig{}, config.TLSConfig{}, p, nil) + return New(slog.New(slog.NewTextHandler(io.Discard, nil)), config.GRPCConfig{}, config.TLSConfig{}, p, nil, nil) } func newTestServerWithResolver(p batchProducer, resolver TenantResolver) *Server { - return New(slog.New(slog.NewTextHandler(io.Discard, nil)), config.GRPCConfig{}, config.TLSConfig{}, p, resolver) + return New(slog.New(slog.NewTextHandler(io.Discard, nil)), config.GRPCConfig{}, config.TLSConfig{}, p, resolver, nil) +} + +func newTestServerWithAgents(agents AgentRegistry) *Server { + return New(slog.New(slog.NewTextHandler(io.Discard, nil)), config.GRPCConfig{}, config.TLSConfig{}, &fakeProducer{}, nil, agents) } // contextWithBearerToken builds an incoming gRPC context carrying an @@ -253,3 +282,95 @@ func TestPushBatchWithResolverRejectsInvalidToken(t *testing.T) { t.Fatal("a batch with an invalid token must never reach the producer once a resolver is configured") } } + +func TestCheckInNilRegistryIsANoOp(t *testing.T) { + s := newTestServer(&fakeProducer{}) + + resp, err := s.CheckIn(context.Background(), &agentv1.CheckInRequest{Host: "h1", CurrentConfig: &agentv1.ReportedConfig{}}) + if err != nil { + t.Fatalf("CheckIn() error = %v", err) + } + if resp.GetHasOverride() { + t.Fatal("expected has_override=false with no AgentRegistry configured") + } +} + +func TestCheckInRejectsEmptyHost(t *testing.T) { + s := newTestServer(&fakeProducer{}) + + _, err := s.CheckIn(context.Background(), &agentv1.CheckInRequest{CurrentConfig: &agentv1.ReportedConfig{}}) + if status.Code(err) != codes.InvalidArgument { + t.Fatalf("CheckIn() error = %v, want InvalidArgument", err) + } +} + +func TestCheckInRecordsReportedConfig(t *testing.T) { + reg := &fakeAgentRegistry{} + s := newTestServerWithAgents(reg) + + _, err := s.CheckIn(context.Background(), &agentv1.CheckInRequest{ + Host: "web-01", + Service: "web", + CurrentConfig: &agentv1.ReportedConfig{ + AgentVersion: "0.1.0", + SourceKind: "journald", + BatchMaxSize: 500, + BatchFlushIntervalMs: 2000, + HeartbeatEnabled: true, + HeartbeatIntervalMs: 60000, + }, + AppliedOverrideVersion: "v3", + }) + if err != nil { + t.Fatalf("CheckIn() error = %v", err) + } + + reg.mu.Lock() + defer reg.mu.Unlock() + if len(reg.checkIns) != 1 { + t.Fatalf("expected 1 recorded check-in, got %d", len(reg.checkIns)) + } + got := reg.checkIns[0] + if got.Host != "web-01" || got.AgentVersion != "0.1.0" || got.BatchMaxSize != 500 || got.AppliedOverrideVersion != "v3" { + t.Fatalf("unexpected recorded check-in: %+v", got) + } +} + +// TestCheckInReturnsOverrideWhenSet is the regression test for the +// actual point of this RPC: an operator-set override for this specific +// host comes back in the response, correctly shaped. +func TestCheckInReturnsOverrideWhenSet(t *testing.T) { + // Keyed by "" (empty tenantID) rather than "default" -- the + // empty-to-"default" substitution is agentregistry.Registry's own + // Postgres-specific behavior (matching the seeded default tenant + // row), not something grpcserver itself does; this fake exercises + // grpcserver.CheckIn in isolation, so it sees the tenantID exactly + // as resolveTenant produced it (empty, since no resolver is + // configured for this test). + interval := uint64(30000) + reg := &fakeAgentRegistry{overrides: map[string]AgentOverride{ + "/web-01": {HasOverride: true, HeartbeatIntervalMS: &interval, Version: "v2"}, + }} + s := newTestServerWithAgents(reg) + + resp, err := s.CheckIn(context.Background(), &agentv1.CheckInRequest{Host: "web-01", CurrentConfig: &agentv1.ReportedConfig{}}) + if err != nil { + t.Fatalf("CheckIn() error = %v", err) + } + if !resp.GetHasOverride() { + t.Fatal("expected has_override=true") + } + if resp.GetOverride().GetHeartbeatIntervalMs() != 30000 || resp.GetOverride().GetVersion() != "v2" { + t.Fatalf("unexpected override: %+v", resp.GetOverride()) + } +} + +func TestCheckInWithResolverRejectsMissingToken(t *testing.T) { + resolver := &fakeResolver{tenantByToken: map[string]string{"real-token": "acme"}} + s := New(slog.New(slog.NewTextHandler(io.Discard, nil)), config.GRPCConfig{}, config.TLSConfig{}, &fakeProducer{}, resolver, &fakeAgentRegistry{}) + + _, err := s.CheckIn(context.Background(), &agentv1.CheckInRequest{Host: "web-01", CurrentConfig: &agentv1.ReportedConfig{}}) + if status.Code(err) != codes.Unauthenticated { + t.Fatalf("CheckIn() error = %v, want Unauthenticated", err) + } +} diff --git a/metadata/migrations/0037_create_agents.sql b/metadata/migrations/0037_create_agents.sql new file mode 100644 index 0000000..ebffb6f --- /dev/null +++ b/metadata/migrations/0037_create_agents.sql @@ -0,0 +1,37 @@ +-- Agent inventory + remote config (see /docs/agent-management-design.md). +-- One row per (tenant, host), upserted by ingest on every CheckIn RPC. +-- tenant_id defaults to 'default' for single-tenant deployments with no +-- TenantResolver configured, same pattern dashboards/alert_rules already +-- established in Phase 3/4. +-- +-- desired_override/desired_override_version are written by api (the web +-- UI's edit form); reported_* and last_seen_at are written by ingest (an +-- agent's own CheckIn). applied_override_version is written by ingest +-- too, but its value comes from the agent itself (CheckInRequest. +-- applied_override_version) -- it's what lets the web UI tell "pending" +-- (desired_override_version != applied_override_version) apart from +-- "applied" without either service needing to poll the other. +CREATE TABLE IF NOT EXISTS agents +( + id UUID PRIMARY KEY, + tenant_id TEXT NOT NULL REFERENCES tenants(id), + host TEXT NOT NULL, + service TEXT NOT NULL DEFAULT '', + reported_agent_version TEXT NOT NULL DEFAULT '', + reported_source_kind TEXT NOT NULL DEFAULT '', + reported_source_detail TEXT NOT NULL DEFAULT '', + reported_batch_max_size BIGINT NOT NULL DEFAULT 0, + reported_batch_flush_ms BIGINT NOT NULL DEFAULT 0, + reported_heartbeat_on BOOLEAN NOT NULL DEFAULT true, + reported_heartbeat_ms BIGINT NOT NULL DEFAULT 0, + first_seen_at TIMESTAMPTZ NOT NULL DEFAULT now(), + last_seen_at TIMESTAMPTZ NOT NULL DEFAULT now(), + -- NULL desired_override_version means no override has ever been set + -- -- CheckInResponse.has_override is false and the agent runs its + -- local agent.toml untouched. + desired_override JSONB, + desired_override_version TEXT, + applied_override_version TEXT NOT NULL DEFAULT '', + updated_by TEXT, + UNIQUE (tenant_id, host) +) diff --git a/proto/README.md b/proto/README.md index 9b86de6..25cb094 100644 --- a/proto/README.md +++ b/proto/README.md @@ -4,8 +4,14 @@ Shared `.proto` contracts. Source of truth for the agent↔ingest gRPC service; each language generates its own bindings from these files rather than sharing generated code across languages. -- `sentry/logs/v1/logs.proto` — `LogIngest.PushBatch`, the only RPC an - agent ever calls. +- `sentry/logs/v1/logs.proto` — `LogIngest.PushBatch`, the data-plane RPC + an agent calls to ship log records. +- `sentry/agent/v1/agent_control.proto` — `AgentControl.CheckIn`, the + control-plane RPC an agent calls (on its own heartbeat ticker, same + push-not-pull posture) to report its config and fetch any remote + override -- see /docs/agent-management-design.md. Same mTLS + connection/listener as `LogIngest`, a second service on it rather than + a second protocol. ## Go bindings @@ -16,7 +22,7 @@ as its own module (`github.com/sentry/sentry/proto`) that `/ingest` and (`/agent`) instead generates its bindings at build time via `tonic-build` (see `agent/sentry-agent/build.rs`) — no checked-in Rust output. -To regenerate the Go bindings after changing `logs.proto`: +To regenerate the Go bindings after changing either `.proto` file: ```sh go install google.golang.org/protobuf/cmd/protoc-gen-go@latest @@ -25,6 +31,6 @@ go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@latest cd proto protoc --go_out=. --go_opt=paths=source_relative \ --go-grpc_out=. --go-grpc_opt=paths=source_relative \ - sentry/logs/v1/logs.proto + sentry/logs/v1/logs.proto sentry/agent/v1/agent_control.proto go build ./... ``` diff --git a/proto/sentry/agent/v1/agent_control.pb.go b/proto/sentry/agent/v1/agent_control.pb.go new file mode 100644 index 0000000..3d42e3f --- /dev/null +++ b/proto/sentry/agent/v1/agent_control.pb.go @@ -0,0 +1,440 @@ +// Code generated by protoc-gen-go. DO NOT EDIT. +// versions: +// protoc-gen-go v1.36.12 +// protoc v7.35.1 +// source: sentry/agent/v1/agent_control.proto + +package agentv1 + +import ( + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" + reflect "reflect" + sync "sync" + unsafe "unsafe" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +// ReportedConfig is what an agent tells the platform about itself -- +// read-only, for inventory/visibility. Deliberately excludes tls/ingest +// endpoint fields: those are never reported and never remotely +// overridable (see DesiredOverride's comment) -- reporting the ingest +// endpoint back to itself would be redundant (that's exactly the +// connection this request arrived over), and TLS material has no +// business leaving the host at all. +type ReportedConfig struct { + state protoimpl.MessageState `protogen:"open.v1"` + AgentVersion string `protobuf:"bytes,1,opt,name=agent_version,json=agentVersion,proto3" json:"agent_version,omitempty"` + SourceKind string `protobuf:"bytes,2,opt,name=source_kind,json=sourceKind,proto3" json:"source_kind,omitempty"` // "journald", "file", "eventlog", "etw" + SourceDetail string `protobuf:"bytes,3,opt,name=source_detail,json=sourceDetail,proto3" json:"source_detail,omitempty"` // human-readable summary: unit name, file path, or channel list + BatchMaxSize uint64 `protobuf:"varint,4,opt,name=batch_max_size,json=batchMaxSize,proto3" json:"batch_max_size,omitempty"` + BatchFlushIntervalMs uint64 `protobuf:"varint,5,opt,name=batch_flush_interval_ms,json=batchFlushIntervalMs,proto3" json:"batch_flush_interval_ms,omitempty"` + HeartbeatEnabled bool `protobuf:"varint,6,opt,name=heartbeat_enabled,json=heartbeatEnabled,proto3" json:"heartbeat_enabled,omitempty"` + HeartbeatIntervalMs uint64 `protobuf:"varint,7,opt,name=heartbeat_interval_ms,json=heartbeatIntervalMs,proto3" json:"heartbeat_interval_ms,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ReportedConfig) Reset() { + *x = ReportedConfig{} + mi := &file_sentry_agent_v1_agent_control_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ReportedConfig) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ReportedConfig) ProtoMessage() {} + +func (x *ReportedConfig) ProtoReflect() protoreflect.Message { + mi := &file_sentry_agent_v1_agent_control_proto_msgTypes[0] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ReportedConfig.ProtoReflect.Descriptor instead. +func (*ReportedConfig) Descriptor() ([]byte, []int) { + return file_sentry_agent_v1_agent_control_proto_rawDescGZIP(), []int{0} +} + +func (x *ReportedConfig) GetAgentVersion() string { + if x != nil { + return x.AgentVersion + } + return "" +} + +func (x *ReportedConfig) GetSourceKind() string { + if x != nil { + return x.SourceKind + } + return "" +} + +func (x *ReportedConfig) GetSourceDetail() string { + if x != nil { + return x.SourceDetail + } + return "" +} + +func (x *ReportedConfig) GetBatchMaxSize() uint64 { + if x != nil { + return x.BatchMaxSize + } + return 0 +} + +func (x *ReportedConfig) GetBatchFlushIntervalMs() uint64 { + if x != nil { + return x.BatchFlushIntervalMs + } + return 0 +} + +func (x *ReportedConfig) GetHeartbeatEnabled() bool { + if x != nil { + return x.HeartbeatEnabled + } + return false +} + +func (x *ReportedConfig) GetHeartbeatIntervalMs() uint64 { + if x != nil { + return x.HeartbeatIntervalMs + } + return 0 +} + +type CheckInRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + Host string `protobuf:"bytes,1,opt,name=host,proto3" json:"host,omitempty"` + Service string `protobuf:"bytes,2,opt,name=service,proto3" json:"service,omitempty"` + CurrentConfig *ReportedConfig `protobuf:"bytes,3,opt,name=current_config,json=currentConfig,proto3" json:"current_config,omitempty"` + // The DesiredOverride.version this agent last successfully applied, + // empty if it has never applied one. Lets the server distinguish + // "pending" (an edit exists the agent hasn't picked up yet) from + // "applied" for the web UI, without the agent needing to know + // anything about that distinction itself. + AppliedOverrideVersion string `protobuf:"bytes,4,opt,name=applied_override_version,json=appliedOverrideVersion,proto3" json:"applied_override_version,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *CheckInRequest) Reset() { + *x = CheckInRequest{} + mi := &file_sentry_agent_v1_agent_control_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *CheckInRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*CheckInRequest) ProtoMessage() {} + +func (x *CheckInRequest) ProtoReflect() protoreflect.Message { + mi := &file_sentry_agent_v1_agent_control_proto_msgTypes[1] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use CheckInRequest.ProtoReflect.Descriptor instead. +func (*CheckInRequest) Descriptor() ([]byte, []int) { + return file_sentry_agent_v1_agent_control_proto_rawDescGZIP(), []int{1} +} + +func (x *CheckInRequest) GetHost() string { + if x != nil { + return x.Host + } + return "" +} + +func (x *CheckInRequest) GetService() string { + if x != nil { + return x.Service + } + return "" +} + +func (x *CheckInRequest) GetCurrentConfig() *ReportedConfig { + if x != nil { + return x.CurrentConfig + } + return nil +} + +func (x *CheckInRequest) GetAppliedOverrideVersion() string { + if x != nil { + return x.AppliedOverrideVersion + } + return "" +} + +// DesiredOverride is the remotely-editable subset of an agent's config +// -- batch/heartbeat tuning, and, for journald sources, the unit +// filter. Every field is optional: unset means "no override for this +// field, keep whatever agent.toml says locally" -- a partial edit only +// touches the fields it sets. Never includes tls/ingest: those stay +// local-file-only, permanently, a deliberate security boundary (see +// /docs/agent-management-design.md) so a bad or malicious remote edit +// can never strand an agent or redirect where its logs go. +type DesiredOverride struct { + state protoimpl.MessageState `protogen:"open.v1"` + BatchMaxSize *uint64 `protobuf:"varint,1,opt,name=batch_max_size,json=batchMaxSize,proto3,oneof" json:"batch_max_size,omitempty"` + BatchFlushIntervalMs *uint64 `protobuf:"varint,2,opt,name=batch_flush_interval_ms,json=batchFlushIntervalMs,proto3,oneof" json:"batch_flush_interval_ms,omitempty"` + HeartbeatEnabled *bool `protobuf:"varint,3,opt,name=heartbeat_enabled,json=heartbeatEnabled,proto3,oneof" json:"heartbeat_enabled,omitempty"` + HeartbeatIntervalMs *uint64 `protobuf:"varint,4,opt,name=heartbeat_interval_ms,json=heartbeatIntervalMs,proto3,oneof" json:"heartbeat_interval_ms,omitempty"` + // Only meaningful when the agent's local source is journald; ignored + // otherwise. Empty string means "no unit filter" (tail the whole + // journal), same semantics as the local config's own unit field. + JournaldUnit *string `protobuf:"bytes,5,opt,name=journald_unit,json=journaldUnit,proto3,oneof" json:"journald_unit,omitempty"` + // Opaque version stamp the platform assigns on every edit. The + // agent's only obligation is to echo it back as + // CheckInRequest.applied_override_version once applied -- it never + // interprets the value itself. + Version string `protobuf:"bytes,6,opt,name=version,proto3" json:"version,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *DesiredOverride) Reset() { + *x = DesiredOverride{} + mi := &file_sentry_agent_v1_agent_control_proto_msgTypes[2] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *DesiredOverride) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*DesiredOverride) ProtoMessage() {} + +func (x *DesiredOverride) ProtoReflect() protoreflect.Message { + mi := &file_sentry_agent_v1_agent_control_proto_msgTypes[2] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use DesiredOverride.ProtoReflect.Descriptor instead. +func (*DesiredOverride) Descriptor() ([]byte, []int) { + return file_sentry_agent_v1_agent_control_proto_rawDescGZIP(), []int{2} +} + +func (x *DesiredOverride) GetBatchMaxSize() uint64 { + if x != nil && x.BatchMaxSize != nil { + return *x.BatchMaxSize + } + return 0 +} + +func (x *DesiredOverride) GetBatchFlushIntervalMs() uint64 { + if x != nil && x.BatchFlushIntervalMs != nil { + return *x.BatchFlushIntervalMs + } + return 0 +} + +func (x *DesiredOverride) GetHeartbeatEnabled() bool { + if x != nil && x.HeartbeatEnabled != nil { + return *x.HeartbeatEnabled + } + return false +} + +func (x *DesiredOverride) GetHeartbeatIntervalMs() uint64 { + if x != nil && x.HeartbeatIntervalMs != nil { + return *x.HeartbeatIntervalMs + } + return 0 +} + +func (x *DesiredOverride) GetJournaldUnit() string { + if x != nil && x.JournaldUnit != nil { + return *x.JournaldUnit + } + return "" +} + +func (x *DesiredOverride) GetVersion() string { + if x != nil { + return x.Version + } + return "" +} + +type CheckInResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + // False when no override has ever been set for this agent -- it + // should be running whatever agent.toml already has, untouched. + HasOverride bool `protobuf:"varint,1,opt,name=has_override,json=hasOverride,proto3" json:"has_override,omitempty"` + Override *DesiredOverride `protobuf:"bytes,2,opt,name=override,proto3" json:"override,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *CheckInResponse) Reset() { + *x = CheckInResponse{} + mi := &file_sentry_agent_v1_agent_control_proto_msgTypes[3] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *CheckInResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*CheckInResponse) ProtoMessage() {} + +func (x *CheckInResponse) ProtoReflect() protoreflect.Message { + mi := &file_sentry_agent_v1_agent_control_proto_msgTypes[3] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use CheckInResponse.ProtoReflect.Descriptor instead. +func (*CheckInResponse) Descriptor() ([]byte, []int) { + return file_sentry_agent_v1_agent_control_proto_rawDescGZIP(), []int{3} +} + +func (x *CheckInResponse) GetHasOverride() bool { + if x != nil { + return x.HasOverride + } + return false +} + +func (x *CheckInResponse) GetOverride() *DesiredOverride { + if x != nil { + return x.Override + } + return nil +} + +var File_sentry_agent_v1_agent_control_proto protoreflect.FileDescriptor + +const file_sentry_agent_v1_agent_control_proto_rawDesc = "" + + "\n" + + "#sentry/agent/v1/agent_control.proto\x12\x0fsentry.agent.v1\"\xb9\x02\n" + + "\x0eReportedConfig\x12#\n" + + "\ragent_version\x18\x01 \x01(\tR\fagentVersion\x12\x1f\n" + + "\vsource_kind\x18\x02 \x01(\tR\n" + + "sourceKind\x12#\n" + + "\rsource_detail\x18\x03 \x01(\tR\fsourceDetail\x12$\n" + + "\x0ebatch_max_size\x18\x04 \x01(\x04R\fbatchMaxSize\x125\n" + + "\x17batch_flush_interval_ms\x18\x05 \x01(\x04R\x14batchFlushIntervalMs\x12+\n" + + "\x11heartbeat_enabled\x18\x06 \x01(\bR\x10heartbeatEnabled\x122\n" + + "\x15heartbeat_interval_ms\x18\a \x01(\x04R\x13heartbeatIntervalMs\"\xc0\x01\n" + + "\x0eCheckInRequest\x12\x12\n" + + "\x04host\x18\x01 \x01(\tR\x04host\x12\x18\n" + + "\aservice\x18\x02 \x01(\tR\aservice\x12F\n" + + "\x0ecurrent_config\x18\x03 \x01(\v2\x1f.sentry.agent.v1.ReportedConfigR\rcurrentConfig\x128\n" + + "\x18applied_override_version\x18\x04 \x01(\tR\x16appliedOverrideVersion\"\x98\x03\n" + + "\x0fDesiredOverride\x12)\n" + + "\x0ebatch_max_size\x18\x01 \x01(\x04H\x00R\fbatchMaxSize\x88\x01\x01\x12:\n" + + "\x17batch_flush_interval_ms\x18\x02 \x01(\x04H\x01R\x14batchFlushIntervalMs\x88\x01\x01\x120\n" + + "\x11heartbeat_enabled\x18\x03 \x01(\bH\x02R\x10heartbeatEnabled\x88\x01\x01\x127\n" + + "\x15heartbeat_interval_ms\x18\x04 \x01(\x04H\x03R\x13heartbeatIntervalMs\x88\x01\x01\x12(\n" + + "\rjournald_unit\x18\x05 \x01(\tH\x04R\fjournaldUnit\x88\x01\x01\x12\x18\n" + + "\aversion\x18\x06 \x01(\tR\aversionB\x11\n" + + "\x0f_batch_max_sizeB\x1a\n" + + "\x18_batch_flush_interval_msB\x14\n" + + "\x12_heartbeat_enabledB\x18\n" + + "\x16_heartbeat_interval_msB\x10\n" + + "\x0e_journald_unit\"r\n" + + "\x0fCheckInResponse\x12!\n" + + "\fhas_override\x18\x01 \x01(\bR\vhasOverride\x12<\n" + + "\boverride\x18\x02 \x01(\v2 .sentry.agent.v1.DesiredOverrideR\boverride2\\\n" + + "\fAgentControl\x12L\n" + + "\aCheckIn\x12\x1f.sentry.agent.v1.CheckInRequest\x1a .sentry.agent.v1.CheckInResponseB8Z6github.com/sentry/sentry/proto/sentry/agent/v1;agentv1b\x06proto3" + +var ( + file_sentry_agent_v1_agent_control_proto_rawDescOnce sync.Once + file_sentry_agent_v1_agent_control_proto_rawDescData []byte +) + +func file_sentry_agent_v1_agent_control_proto_rawDescGZIP() []byte { + file_sentry_agent_v1_agent_control_proto_rawDescOnce.Do(func() { + file_sentry_agent_v1_agent_control_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_sentry_agent_v1_agent_control_proto_rawDesc), len(file_sentry_agent_v1_agent_control_proto_rawDesc))) + }) + return file_sentry_agent_v1_agent_control_proto_rawDescData +} + +var file_sentry_agent_v1_agent_control_proto_msgTypes = make([]protoimpl.MessageInfo, 4) +var file_sentry_agent_v1_agent_control_proto_goTypes = []any{ + (*ReportedConfig)(nil), // 0: sentry.agent.v1.ReportedConfig + (*CheckInRequest)(nil), // 1: sentry.agent.v1.CheckInRequest + (*DesiredOverride)(nil), // 2: sentry.agent.v1.DesiredOverride + (*CheckInResponse)(nil), // 3: sentry.agent.v1.CheckInResponse +} +var file_sentry_agent_v1_agent_control_proto_depIdxs = []int32{ + 0, // 0: sentry.agent.v1.CheckInRequest.current_config:type_name -> sentry.agent.v1.ReportedConfig + 2, // 1: sentry.agent.v1.CheckInResponse.override:type_name -> sentry.agent.v1.DesiredOverride + 1, // 2: sentry.agent.v1.AgentControl.CheckIn:input_type -> sentry.agent.v1.CheckInRequest + 3, // 3: sentry.agent.v1.AgentControl.CheckIn:output_type -> sentry.agent.v1.CheckInResponse + 3, // [3:4] is the sub-list for method output_type + 2, // [2:3] is the sub-list for method input_type + 2, // [2:2] is the sub-list for extension type_name + 2, // [2:2] is the sub-list for extension extendee + 0, // [0:2] is the sub-list for field type_name +} + +func init() { file_sentry_agent_v1_agent_control_proto_init() } +func file_sentry_agent_v1_agent_control_proto_init() { + if File_sentry_agent_v1_agent_control_proto != nil { + return + } + file_sentry_agent_v1_agent_control_proto_msgTypes[2].OneofWrappers = []any{} + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: unsafe.Slice(unsafe.StringData(file_sentry_agent_v1_agent_control_proto_rawDesc), len(file_sentry_agent_v1_agent_control_proto_rawDesc)), + NumEnums: 0, + NumMessages: 4, + NumExtensions: 0, + NumServices: 1, + }, + GoTypes: file_sentry_agent_v1_agent_control_proto_goTypes, + DependencyIndexes: file_sentry_agent_v1_agent_control_proto_depIdxs, + MessageInfos: file_sentry_agent_v1_agent_control_proto_msgTypes, + }.Build() + File_sentry_agent_v1_agent_control_proto = out.File + file_sentry_agent_v1_agent_control_proto_goTypes = nil + file_sentry_agent_v1_agent_control_proto_depIdxs = nil +} diff --git a/proto/sentry/agent/v1/agent_control.proto b/proto/sentry/agent/v1/agent_control.proto new file mode 100644 index 0000000..ed08dcd --- /dev/null +++ b/proto/sentry/agent/v1/agent_control.proto @@ -0,0 +1,79 @@ +syntax = "proto3"; + +package sentry.agent.v1; + +option go_package = "github.com/sentry/sentry/proto/sentry/agent/v1;agentv1"; + +// AgentControl is the control-plane counterpart to logs.v1.LogIngest's +// data-plane PushBatch -- the same mTLS channel/connection an agent +// already has open to ingest, a second gRPC service on the same +// listener rather than a second protocol or connection the agent would +// need to maintain (see /docs/agent-management-design.md). CheckIn is +// agent-initiated, called on the agent's own heartbeat ticker: there is +// still no path for the platform to reach into an agent uninvited. An +// agent asks "what should I be running" on its own schedule -- the same +// push-not-pull posture the heartbeat feature this builds on already +// established. +service AgentControl { + rpc CheckIn(CheckInRequest) returns (CheckInResponse); +} + +// ReportedConfig is what an agent tells the platform about itself -- +// read-only, for inventory/visibility. Deliberately excludes tls/ingest +// endpoint fields: those are never reported and never remotely +// overridable (see DesiredOverride's comment) -- reporting the ingest +// endpoint back to itself would be redundant (that's exactly the +// connection this request arrived over), and TLS material has no +// business leaving the host at all. +message ReportedConfig { + string agent_version = 1; + string source_kind = 2; // "journald", "file", "eventlog", "etw" + string source_detail = 3; // human-readable summary: unit name, file path, or channel list + uint64 batch_max_size = 4; + uint64 batch_flush_interval_ms = 5; + bool heartbeat_enabled = 6; + uint64 heartbeat_interval_ms = 7; +} + +message CheckInRequest { + string host = 1; + string service = 2; + ReportedConfig current_config = 3; + // The DesiredOverride.version this agent last successfully applied, + // empty if it has never applied one. Lets the server distinguish + // "pending" (an edit exists the agent hasn't picked up yet) from + // "applied" for the web UI, without the agent needing to know + // anything about that distinction itself. + string applied_override_version = 4; +} + +// DesiredOverride is the remotely-editable subset of an agent's config +// -- batch/heartbeat tuning, and, for journald sources, the unit +// filter. Every field is optional: unset means "no override for this +// field, keep whatever agent.toml says locally" -- a partial edit only +// touches the fields it sets. Never includes tls/ingest: those stay +// local-file-only, permanently, a deliberate security boundary (see +// /docs/agent-management-design.md) so a bad or malicious remote edit +// can never strand an agent or redirect where its logs go. +message DesiredOverride { + optional uint64 batch_max_size = 1; + optional uint64 batch_flush_interval_ms = 2; + optional bool heartbeat_enabled = 3; + optional uint64 heartbeat_interval_ms = 4; + // Only meaningful when the agent's local source is journald; ignored + // otherwise. Empty string means "no unit filter" (tail the whole + // journal), same semantics as the local config's own unit field. + optional string journald_unit = 5; + // Opaque version stamp the platform assigns on every edit. The + // agent's only obligation is to echo it back as + // CheckInRequest.applied_override_version once applied -- it never + // interprets the value itself. + string version = 6; +} + +message CheckInResponse { + // False when no override has ever been set for this agent -- it + // should be running whatever agent.toml already has, untouched. + bool has_override = 1; + DesiredOverride override = 2; +} diff --git a/proto/sentry/agent/v1/agent_control_grpc.pb.go b/proto/sentry/agent/v1/agent_control_grpc.pb.go new file mode 100644 index 0000000..aa58eb7 --- /dev/null +++ b/proto/sentry/agent/v1/agent_control_grpc.pb.go @@ -0,0 +1,143 @@ +// Code generated by protoc-gen-go-grpc. DO NOT EDIT. +// versions: +// - protoc-gen-go-grpc v1.6.2 +// - protoc v7.35.1 +// source: sentry/agent/v1/agent_control.proto + +package agentv1 + +import ( + context "context" + grpc "google.golang.org/grpc" + codes "google.golang.org/grpc/codes" + status "google.golang.org/grpc/status" +) + +// This is a compile-time assertion to ensure that this generated file +// is compatible with the grpc package it is being compiled against. +// Requires gRPC-Go v1.64.0 or later. +const _ = grpc.SupportPackageIsVersion9 + +const ( + AgentControl_CheckIn_FullMethodName = "/sentry.agent.v1.AgentControl/CheckIn" +) + +// AgentControlClient is the client API for AgentControl service. +// +// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream. +// +// AgentControl is the control-plane counterpart to logs.v1.LogIngest's +// data-plane PushBatch -- the same mTLS channel/connection an agent +// already has open to ingest, a second gRPC service on the same +// listener rather than a second protocol or connection the agent would +// need to maintain (see /docs/agent-management-design.md). CheckIn is +// agent-initiated, called on the agent's own heartbeat ticker: there is +// still no path for the platform to reach into an agent uninvited. An +// agent asks "what should I be running" on its own schedule -- the same +// push-not-pull posture the heartbeat feature this builds on already +// established. +type AgentControlClient interface { + CheckIn(ctx context.Context, in *CheckInRequest, opts ...grpc.CallOption) (*CheckInResponse, error) +} + +type agentControlClient struct { + cc grpc.ClientConnInterface +} + +func NewAgentControlClient(cc grpc.ClientConnInterface) AgentControlClient { + return &agentControlClient{cc} +} + +func (c *agentControlClient) CheckIn(ctx context.Context, in *CheckInRequest, opts ...grpc.CallOption) (*CheckInResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(CheckInResponse) + err := c.cc.Invoke(ctx, AgentControl_CheckIn_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + +// AgentControlServer is the server API for AgentControl service. +// All implementations must embed UnimplementedAgentControlServer +// for forward compatibility. +// +// AgentControl is the control-plane counterpart to logs.v1.LogIngest's +// data-plane PushBatch -- the same mTLS channel/connection an agent +// already has open to ingest, a second gRPC service on the same +// listener rather than a second protocol or connection the agent would +// need to maintain (see /docs/agent-management-design.md). CheckIn is +// agent-initiated, called on the agent's own heartbeat ticker: there is +// still no path for the platform to reach into an agent uninvited. An +// agent asks "what should I be running" on its own schedule -- the same +// push-not-pull posture the heartbeat feature this builds on already +// established. +type AgentControlServer interface { + CheckIn(context.Context, *CheckInRequest) (*CheckInResponse, error) + mustEmbedUnimplementedAgentControlServer() +} + +// UnimplementedAgentControlServer must be embedded to have +// forward compatible implementations. +// +// NOTE: this should be embedded by value instead of pointer to avoid a nil +// pointer dereference when methods are called. +type UnimplementedAgentControlServer struct{} + +func (UnimplementedAgentControlServer) CheckIn(context.Context, *CheckInRequest) (*CheckInResponse, error) { + return nil, status.Error(codes.Unimplemented, "method CheckIn not implemented") +} +func (UnimplementedAgentControlServer) mustEmbedUnimplementedAgentControlServer() {} +func (UnimplementedAgentControlServer) testEmbeddedByValue() {} + +// UnsafeAgentControlServer may be embedded to opt out of forward compatibility for this service. +// Use of this interface is not recommended, as added methods to AgentControlServer will +// result in compilation errors. +type UnsafeAgentControlServer interface { + mustEmbedUnimplementedAgentControlServer() +} + +func RegisterAgentControlServer(s grpc.ServiceRegistrar, srv AgentControlServer) { + // If the following call panics, it indicates UnimplementedAgentControlServer was + // embedded by pointer and is nil. This will cause panics if an + // unimplemented method is ever invoked, so we test this at initialization + // time to prevent it from happening at runtime later due to I/O. + if t, ok := srv.(interface{ testEmbeddedByValue() }); ok { + t.testEmbeddedByValue() + } + s.RegisterService(&AgentControl_ServiceDesc, srv) +} + +func _AgentControl_CheckIn_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(CheckInRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(AgentControlServer).CheckIn(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: AgentControl_CheckIn_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(AgentControlServer).CheckIn(ctx, req.(*CheckInRequest)) + } + return interceptor(ctx, in, info, handler) +} + +// AgentControl_ServiceDesc is the grpc.ServiceDesc for AgentControl service. +// It's only intended for direct use with grpc.RegisterService, +// and not to be introspected or modified (even as a copy) +var AgentControl_ServiceDesc = grpc.ServiceDesc{ + ServiceName: "sentry.agent.v1.AgentControl", + HandlerType: (*AgentControlServer)(nil), + Methods: []grpc.MethodDesc{ + { + MethodName: "CheckIn", + Handler: _AgentControl_CheckIn_Handler, + }, + }, + Streams: []grpc.StreamDesc{}, + Metadata: "sentry/agent/v1/agent_control.proto", +} diff --git a/web/src/lib/api.ts b/web/src/lib/api.ts index c1d1cc4..354fd44 100644 --- a/web/src/lib/api.ts +++ b/web/src/lib/api.ts @@ -473,3 +473,55 @@ export function createNotificationTarget(input: { return alertingRequest('/targets', { method: 'POST', body: JSON.stringify(input) }); } +// ---- Agent inventory + remote config ---- +// See /docs/agent-management-design.md. An agent only appears here +// after it's checked in at least once (GET /agents/{host} 404s until +// then) -- there's no "pre-register a host" step, inventory is purely +// observed from real check-ins. +export type ConfigOverride = { + batch_max_size?: number; + batch_flush_interval_ms?: number; + heartbeat_enabled?: boolean; + heartbeat_interval_ms?: number; + journald_unit?: string; +}; + +export type Agent = { + id: string; + tenant_id: string; + host: string; + service: string; + agent_version: string; + source_kind: string; + source_detail: string; + batch_max_size: number; + batch_flush_interval_ms: number; + heartbeat_enabled: boolean; + heartbeat_interval_ms: number; + first_seen_at: string; + last_seen_at: string; + desired_override?: ConfigOverride; + desired_override_version?: string; + applied_override_version: string; + pending: boolean; + updated_by?: string; +}; + +export function listAgents(): Promise { + return request('/agents').then((a) => a ?? []); +} + +export function getAgent(host: string): Promise { + return request(`/agents/${encodeURIComponent(host)}`); +} + +export function setAgentConfig(host: string, override: ConfigOverride): Promise { + return request(`/agents/${encodeURIComponent(host)}/config`, { + method: 'PUT', + body: JSON.stringify(override) + }); +} + +export function clearAgentConfig(host: string): Promise { + return request(`/agents/${encodeURIComponent(host)}/config`, { method: 'DELETE' }); +} diff --git a/web/src/lib/components/NavSidebar.svelte b/web/src/lib/components/NavSidebar.svelte index 693b88b..2dd0487 100644 --- a/web/src/lib/components/NavSidebar.svelte +++ b/web/src/lib/components/NavSidebar.svelte @@ -15,6 +15,7 @@ { href: '/dashboards', label: 'Dashboards', icon: '▤' }, { href: '/alerts', label: 'Alerts', icon: '▲' }, { href: '/data-sources', label: 'Data Sources', icon: '◈' }, + { href: '/agents', label: 'Agents', icon: '●' }, { href: '/settings', label: 'Settings', icon: '⚙' } ]; diff --git a/web/src/routes/agents/+page.svelte b/web/src/routes/agents/+page.svelte new file mode 100644 index 0000000..408380d --- /dev/null +++ b/web/src/routes/agents/+page.svelte @@ -0,0 +1,131 @@ + + +
+

Agents

+

+ Linux/Windows log collection agents that have checked in at least once. +

+ {#if error}

Error: {error}

{/if} + + {#if loading} +
+ {#each Array(3) as _, i (i)} + + {/each} +
+ {:else if agents.length === 0} + + {:else} + + + + + + + + + + + + + {#each agents as a (a.id)} + + + + + + + + + {/each} + +
HostServiceVersionLast seenStatusConfig
{a.host}{a.service}{a.agent_version || '—'}{relativeTime(a.last_seen_at)} + {#if isStale(a)} + stale + {:else} + healthy + {/if} + + {#if a.pending} + pending + {:else if a.desired_override} + applied + {/if} +
+ {/if} +
+ + diff --git a/web/src/routes/agents/+page.ts b/web/src/routes/agents/+page.ts new file mode 100644 index 0000000..30f9b52 --- /dev/null +++ b/web/src/routes/agents/+page.ts @@ -0,0 +1,3 @@ +// No route params, data comes from a client-side fetch -- same shape as +// dashboards/+page.ts. +export const prerender = true; diff --git a/web/src/routes/agents/[host]/+page.svelte b/web/src/routes/agents/[host]/+page.svelte new file mode 100644 index 0000000..7c34f39 --- /dev/null +++ b/web/src/routes/agents/[host]/+page.svelte @@ -0,0 +1,237 @@ + + +
+ ← Agents +

{host}

+ + {#if loading} + + {:else if error} +

Error: {error}

+ {:else if agent} +
+

Reported

+
+
Service
+
{agent.service}
+
Version
+
{agent.agent_version || '—'}
+
Source
+
{agent.source_kind}{agent.source_detail ? ` (${agent.source_detail})` : ''}
+
First seen
+
{relativeTime(agent.first_seen_at)}
+
Last seen
+
{relativeTime(agent.last_seen_at)}
+
+
+ +
+

+ Remote config + {#if agent.pending} + pending — agent hasn't applied this yet + {:else if agent.desired_override} + applied + {/if} +

+

+ Changes here don't touch the agent's local config file -- they're an override the agent fetches and applies + on its own schedule (its heartbeat interval), and revert automatically if the agent restarts before its next + check-in re-syncs them. Connection details (TLS, ingest endpoint) are never remotely editable. +

+ +
+ + +
+
+ + +
+
+ +
+
+ + +
+ {#if agent.source_kind === 'journald'} +
+ + +
+ {/if} + + {#if saveError}

Error: {saveError}

{/if} + +
+ + {#if agent.desired_override} + + {/if} +
+
+ {/if} +
+ + diff --git a/web/src/routes/agents/[host]/+page.ts b/web/src/routes/agents/[host]/+page.ts new file mode 100644 index 0000000..f9544f3 --- /dev/null +++ b/web/src/routes/agents/[host]/+page.ts @@ -0,0 +1,4 @@ +// The host param doesn't exist at build time -- same reasoning as +// dashboards/[id]/+page.ts. +export const prerender = false; +export const ssr = false;