creator/monlet
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add web auth, infinite-scroll, agent admission and review fixes across agent/server/web

Browse Source
This commit is contained in:
Stanislav Rossovskii
2026-05-28 23:45:12 +04:00
parent 37b1a1d6d6
commit 1026e9ebbe
75 changed files with 3021 additions and 916 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
agent/README.md
View File

@@ -26,6 +26,7 @@ See `config.example.toml`. Required fields: top-level `state_dir`, at least one
| `metrics.listen` | `127.0.0.1:9465` | low-cardinality only |
| `checks[].command` | required | Shell string executed as `/bin/sh -c`. Both forms accepted: `command = "check_disk --warn 80 --crit 90"` and triple-quoted `command = '''...'''`. Argv arrays are rejected. See [Command security contract](#command-security-contract). |
| `checks[].notifications_enabled` | `true` | Set `false` for checks that must not create server notifications. |
| `checks[].resource_limits` | `{}` | Optional per-check best-effort `ulimit` values: `cpu_time`, `memory`, `open_files`. |
Examples — one-line and multi-line forms:
@@ -44,6 +45,7 @@ set -eu
'''
interval = "60s"
timeout = "10s"
resource_limits = { cpu_time = "5s", memory = "256MiB", open_files = 128 }
```
## Command security contract
@@ -67,9 +69,25 @@ executes it as `/bin/sh -c <command>` on the host where the agent runs.
validation to keep secrets out of heartbeats.
- Each run uses `exec.CommandContext` with `Setpgid`. On timeout the whole
process group is killed and the check result becomes `critical`.
- Optional `resource_limits` are applied inside the child shell before the
configured command runs. `cpu_time` uses `ulimit -t` and maps SIGXCPU to
`critical`; `memory` uses virtual memory/address-space `ulimit -v`; `open_files`
uses `ulimit -n`. These limits are best-effort and are not a cgroup/RSS limit.
- Combined stdout+stderr is truncated by UTF-8 byte length to 8 KiB with marker
`...[truncated N bytes]` before being persisted, exposed, or notified.
## Config reload
Send `SIGHUP` or run `systemctl reload monlet-agent` to reload the local TOML file.
- Reload first validates the full new config. On error, the old config keeps running.
- Hot-reloadable: labels and checks, including `command`, `interval`, `timeout`,
`notifications_enabled`, `dedupe_key`, and `resource_limits`.
- Restart required: `agent_id`, `state_dir`, `server.*`, and `metrics.*`.
- A check already running during reload is allowed to finish under its old config.
Removed or changed checks do not start new old-config runs; changed checks start
with the new config after the in-flight run finishes.
## Behaviour
- Per-check scheduler with no-overlap: a tick is skipped if the previous run is still in flight (`monlet_agent_check_skipped_total`).
@@ -79,7 +97,9 @@ executes it as `/bin/sh -c <command>` on the host where the agent runs.
- `output` (stdout+stderr) is truncated by **UTF-8 byte length** to 8 KiB with marker `...[truncated N bytes]`.
- `event_id` is UUIDv7 generated on check completion and preserved across spool replay (ADR-0006 idempotency).
- Spool: per-event JSON files in `state_dir/spool/`, FIFO, limits 10 000 events / 50 MiB, drop oldest on overflow (ADR-0005). Drops are counted in `monlet_agent_events_dropped_total{reason}` with bounded reasons; one warning is logged per ~30s drop burst, not per event.
- Event flushes are split before POST so each JSON request body stays within the 1 MiB server limit.
- Retry backoff for heartbeat and events: 1s → 30s, exp ×2, jitter ±20%. 4xx are dropped (not retried) to avoid hot loop; dropped batches are counted as `reason=send_non_retryable` and logged once per batch.
- On shutdown the agent asks workers to stop cleanly; if they do not drain within 30s, the process exits non-zero so the service manager can restart or mark failure.
- Bearer auth header on both endpoints uses the local `state_dir/agent.key` value (ADR-0007).
- Heartbeats include configured labels plus reserved `monlet_agent_version=<binary version>`.
- Heartbeats include explicit feature flags: `push` from `server.enabled`, `metrics` from `metrics.enabled`.
@@ -97,6 +117,11 @@ Exposed on `metrics.listen` at `/metrics`. Labels are static (`check_id`, `statu
- `monlet_agent_check_last_success_timestamp_seconds{check_id}`
- `monlet_agent_check_interval_seconds{check_id}`
- `monlet_agent_checks_configured`
- `monlet_agent_config_load_success`
- `monlet_agent_config_reload_total{result}`
- `monlet_agent_config_last_reload_success_timestamp_seconds`
- `monlet_agent_resource_limit_apply_failures_total{check_id,resource}`
- `monlet_agent_events_channel_full_total`
- `monlet_agent_spool_events`, `monlet_agent_spool_bytes`
- `monlet_agent_send_attempts_total{kind}`, `monlet_agent_send_failures_total{kind}` (`kind` ∈ `heartbeat`, `events`)
- `monlet_agent_events_accepted_total`, `monlet_agent_events_deduplicated_total`

56
agent/cmd/monlet-agent/main.go
View File

@@ -8,6 +8,7 @@ import (
"os/signal"
"path/filepath"
"syscall"
"time"
"github.com/monlet/agent/internal/agentkey"
"github.com/monlet/agent/internal/app"
@@ -20,6 +21,13 @@ import (
var version = "0.1.0"
const shutdownTimeout = 30 * time.Second
type reloadTarget interface {
RecordReloadFailure()
Reload(*config.Config, string) error
}
func main() {
cfgPath := flag.String("config", "config.toml", "path to TOML config")
flag.Parse()
@@ -65,6 +73,12 @@ func main() {
ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
defer cancel()
startShutdownWatchdog(ctx, shutdownTimeout, log, os.Exit)
hup := make(chan os.Signal, 1)
signal.Notify(hup, syscall.SIGHUP)
defer signal.Stop(hup)
go runReloadLoop(ctx, *cfgPath, hup, a, log)
log.Info("monlet-agent starting", "agent_id", agentID, "version", version, "checks", len(cfg.Checks))
if err := a.Run(ctx); err != nil {
@@ -73,3 +87,45 @@ func main() {
}
log.Info("monlet-agent stopped")
}
func startShutdownWatchdog(ctx context.Context, timeout time.Duration, log *slog.Logger, exit func(int)) {
go func() {
<-ctx.Done()
t := time.NewTimer(timeout)
defer t.Stop()
<-t.C
log.Error("shutdown timeout exceeded", "timeout", timeout.String())
exit(1)
}()
}
func runReloadLoop(ctx context.Context, cfgPath string, hup <-chan os.Signal, target reloadTarget, log *slog.Logger) {
for {
select {
case <-ctx.Done():
return
case <-hup:
reloadOnce(cfgPath, target, log)
}
}
}
func reloadOnce(cfgPath string, target reloadTarget, log *slog.Logger) {
nextCfg, err := config.Load(cfgPath)
if err != nil {
target.RecordReloadFailure()
log.Warn("reload config failed", "err", err)
return
}
nextAgentID, err := ids.ResolveAgentID(nextCfg.AgentID, nextCfg.Hostname)
if err != nil {
target.RecordReloadFailure()
log.Warn("reload agent_id failed", "err", err)
return
}
if err := target.Reload(nextCfg, nextAgentID); err != nil {
log.Warn("reload rejected", "err", err)
return
}
log.Info("reload complete", "checks", len(nextCfg.Checks))
}

118
agent/cmd/monlet-agent/main_test.go Normal file
View File

@@ -0,0 +1,118 @@
package main
import (
"context"
"fmt"
"io"
"log/slog"
"os"
"path/filepath"
"syscall"
"testing"
"time"
"github.com/monlet/agent/internal/config"
)
type reloadCall struct {
cfg *config.Config
agentID string
}
type fakeReloadTarget struct {
reloads chan reloadCall
failures chan struct{}
}
func (f *fakeReloadTarget) RecordReloadFailure() {
f.failures <- struct{}{}
}
func (f *fakeReloadTarget) Reload(cfg *config.Config, agentID string) error {
f.reloads <- reloadCall{cfg: cfg, agentID: agentID}
return nil
}
func TestRunReloadLoopHandlesSIGHUP(t *testing.T) {
cfgPath := writeTestConfig(t)
target := &fakeReloadTarget{
reloads: make(chan reloadCall, 1),
failures: make(chan struct{}, 1),
}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
hup := make(chan os.Signal, 1)
go runReloadLoop(ctx, cfgPath, hup, target, discardLogger())
hup <- syscall.SIGHUP
select {
case call := <-target.reloads:
if call.agentID != "agent-test" || len(call.cfg.Checks) != 1 {
t.Fatalf("unexpected reload call: agent_id=%q cfg=%+v", call.agentID, call.cfg)
}
case <-time.After(time.Second):
t.Fatal("reload was not called")
}
}
func TestRunReloadLoopRecordsLoadFailure(t *testing.T) {
target := &fakeReloadTarget{
reloads: make(chan reloadCall, 1),
failures: make(chan struct{}, 1),
}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
hup := make(chan os.Signal, 1)
go runReloadLoop(ctx, filepath.Join(t.TempDir(), "missing.toml"), hup, target, discardLogger())
hup <- syscall.SIGHUP
select {
case <-target.failures:
case <-time.After(time.Second):
t.Fatal("reload failure was not recorded")
}
}
func TestShutdownWatchdogExitsAfterTimeout(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
exits := make(chan int, 1)
startShutdownWatchdog(ctx, time.Millisecond, discardLogger(), func(code int) {
exits <- code
})
cancel()
select {
case code := <-exits:
if code != 1 {
t.Fatalf("exit code: %d", code)
}
case <-time.After(time.Second):
t.Fatal("watchdog did not exit")
}
}
func writeTestConfig(t *testing.T) string {
t.Helper()
path := filepath.Join(t.TempDir(), "agent.toml")
body := fmt.Sprintf(`
agent_id = "agent-test"
state_dir = %q
[server]
enabled = true
url = "http://localhost"
[[checks]]
id = "c1"
command = "true"
interval = "10s"
timeout = "5s"
`, t.TempDir())
if err := os.WriteFile(path, []byte(body), 0o600); err != nil {
t.Fatal(err)
}
return path
}
func discardLogger() *slog.Logger {
return slog.New(slog.NewTextHandler(io.Discard, nil))
}

1
agent/config.example.toml
View File

@@ -24,6 +24,7 @@ command = '''
'''
interval = "60s"
timeout = "10s"
resource_limits = { cpu_time = "5s", memory = "256MiB", open_files = 128 }
[[checks]]
id = "uptime"

144
agent/internal/app/app.go
View File

@@ -3,6 +3,7 @@ package app
import (
"context"
"errors"
"fmt"
"log/slog"
"sync"
"time"
@@ -24,6 +25,9 @@ type App struct {
Log *slog.Logger
lastDropLog time.Time
cfgMu sync.RWMutex
schedulerMu sync.RWMutex
scheduler *scheduler.Manager
}
const (
@@ -36,47 +40,61 @@ const (
func (a *App) Run(ctx context.Context) error {
events := make(chan scheduler.Event, 256)
cfg := a.config()
// Account drops accumulated during spool Open (e.g. shrunk limits on restart).
a.recordSpoolDrops(a.Spool.Drops())
a.Metrics.ChecksConfigured.Set(float64(len(a.Cfg.Checks)))
a.Metrics.BuildInfo.WithLabelValues(a.Version, boolLabel(a.Cfg.PushesToServer()), boolLabel(a.Cfg.ExposesMetrics())).Set(1)
for _, c := range a.Cfg.Checks {
a.Metrics.ConfigLoadSuccess.Set(1)
a.Metrics.ChecksConfigured.Set(float64(len(cfg.Checks)))
a.Metrics.BuildInfo.WithLabelValues(a.Version, boolLabel(cfg.PushesToServer()), boolLabel(cfg.ExposesMetrics())).Set(1)
for _, c := range cfg.Checks {
a.Metrics.CheckInterval.WithLabelValues(c.ID).Set(c.Interval.Duration.Seconds())
}
var wg sync.WaitGroup
if a.Cfg.ExposesMetrics() {
if cfg.ExposesMetrics() {
wg.Add(1)
go func() {
defer wg.Done()
if err := a.Metrics.Serve(ctx, a.Cfg.Metrics.Listen); err != nil {
if err := a.Metrics.Serve(ctx, cfg.Metrics.Listen); err != nil {
a.Log.Error("metrics server", "err", err)
}
}()
}
mgr := scheduler.NewManager(ctx, a.AgentID, events, scheduler.Hooks{
OnSkipped: func(id string) { a.Metrics.CheckSkipped.WithLabelValues(id).Inc() },
OnResult: func(ev scheduler.Event) {
a.Metrics.CheckRuns.WithLabelValues(ev.CheckID, ev.Status).Inc()
a.Metrics.CheckDuration.WithLabelValues(ev.CheckID).Observe(float64(ev.DurationMs) / 1000.0)
a.Metrics.CheckStatus.WithLabelValues(ev.CheckID).Set(metrics.StatusCode(ev.Status))
a.Metrics.CheckExitCode.WithLabelValues(ev.CheckID).Set(float64(ev.ExitCode))
ts := float64(ev.ObservedAt.Unix())
a.Metrics.CheckLastRun.WithLabelValues(ev.CheckID).Set(ts)
if ev.Status == "ok" {
a.Metrics.CheckLastSuccess.WithLabelValues(ev.CheckID).Set(ts)
}
},
OnEventChannelFull: func() { a.Metrics.EventsChannelFull.Inc() },
OnResourceLimitFailure: func(id string, resource string) {
a.Metrics.ResourceLimitApplyFailures.WithLabelValues(id, resource).Inc()
},
OnStopped: func(id string) { a.Metrics.DeleteCheck(id) },
})
mgr.Update(cfg.Checks)
a.schedulerMu.Lock()
a.scheduler = mgr
a.schedulerMu.Unlock()
wg.Add(1)
go func() {
defer wg.Done()
scheduler.Run(ctx, a.AgentID, a.Cfg.Checks, events, scheduler.Hooks{
OnSkipped: func(id string) { a.Metrics.CheckSkipped.WithLabelValues(id).Inc() },
OnResult: func(ev scheduler.Event) {
a.Metrics.CheckRuns.WithLabelValues(ev.CheckID, ev.Status).Inc()
a.Metrics.CheckDuration.WithLabelValues(ev.CheckID).Observe(float64(ev.DurationMs) / 1000.0)
a.Metrics.CheckStatus.WithLabelValues(ev.CheckID).Set(metrics.StatusCode(ev.Status))
a.Metrics.CheckExitCode.WithLabelValues(ev.CheckID).Set(float64(ev.ExitCode))
ts := float64(ev.ObservedAt.Unix())
a.Metrics.CheckLastRun.WithLabelValues(ev.CheckID).Set(ts)
if ev.Status == "ok" {
a.Metrics.CheckLastSuccess.WithLabelValues(ev.CheckID).Set(ts)
}
},
})
<-ctx.Done()
mgr.Stop()
mgr.Wait()
}()
if a.Cfg.PushesToServer() {
if cfg.PushesToServer() {
wg.Add(1)
go func() {
defer wg.Done()
@@ -128,18 +146,19 @@ func (a *App) spoolWriter(ctx context.Context, events <-chan scheduler.Event) {
}
func (a *App) heartbeatLoop(ctx context.Context) {
interval := a.Cfg.Server.HeartbeatInterval.Duration
attempt := 0
for {
cfg := a.config()
interval := cfg.Server.HeartbeatInterval.Duration
hb := client.HeartbeatRequest{
AgentID: a.AgentID,
ObservedAt: time.Now().UTC(),
Hostname: a.Cfg.Hostname,
Hostname: cfg.Hostname,
Features: client.AgentFeatures{
Push: a.Cfg.PushesToServer(),
Metrics: a.Cfg.ExposesMetrics(),
Push: cfg.PushesToServer(),
Metrics: cfg.ExposesMetrics(),
},
Labels: a.Cfg.HeartbeatLabels(a.Version),
Labels: cfg.HeartbeatLabels(a.Version),
}
a.Metrics.SendAttempts.WithLabelValues("heartbeat").Inc()
err := a.Client.SendHeartbeat(ctx, hb)
@@ -167,9 +186,14 @@ func (a *App) heartbeatLoop(ctx context.Context) {
}
func (a *App) sendLoop(ctx context.Context) {
interval := a.Cfg.Server.BatchInterval.Duration
attempt := 0
for {
select {
case <-ctx.Done():
return
default:
}
interval := a.config().Server.BatchInterval.Duration
items, err := a.Spool.Peek(client.MaxBatchEvents)
if err != nil {
a.Log.Error("spool peek", "err", err)
@@ -217,6 +241,74 @@ func (a *App) sendLoop(ctx context.Context) {
}
}
// Reload applies a validated hot-reload config if immutable fields match.
func (a *App) Reload(newCfg *config.Config, newAgentID string) error {
oldCfg := a.config()
if err := validateReload(oldCfg, newCfg, a.AgentID, newAgentID); err != nil {
a.RecordReloadFailure()
return err
}
a.cfgMu.Lock()
a.Cfg = newCfg
a.cfgMu.Unlock()
a.Metrics.ConfigLoadSuccess.Set(1)
a.Metrics.ConfigReloadTotal.WithLabelValues("success").Inc()
a.Metrics.ConfigLastReloadSuccess.Set(float64(time.Now().Unix()))
a.Metrics.ChecksConfigured.Set(float64(len(newCfg.Checks)))
for _, c := range newCfg.Checks {
a.Metrics.CheckInterval.WithLabelValues(c.ID).Set(c.Interval.Duration.Seconds())
}
a.schedulerMu.RLock()
mgr := a.scheduler
a.schedulerMu.RUnlock()
if mgr != nil {
mgr.Update(newCfg.Checks)
}
return nil
}
// RecordReloadFailure updates config reload metrics after a failed reload attempt.
func (a *App) RecordReloadFailure() {
a.Metrics.ConfigLoadSuccess.Set(0)
a.Metrics.ConfigReloadTotal.WithLabelValues("failure").Inc()
}
func (a *App) config() *config.Config {
a.cfgMu.RLock()
defer a.cfgMu.RUnlock()
return a.Cfg
}
func validateReload(oldCfg, newCfg *config.Config, oldAgentID, newAgentID string) error {
if oldAgentID != newAgentID {
return fmt.Errorf("agent_id change requires restart")
}
if oldCfg.StateDir != newCfg.StateDir {
return fmt.Errorf("state_dir change requires restart")
}
if oldCfg.PushesToServer() != newCfg.PushesToServer() {
return fmt.Errorf("server.enabled change requires restart")
}
if oldCfg.Server.URL != newCfg.Server.URL {
return fmt.Errorf("server.url change requires restart")
}
if oldCfg.Server.HeartbeatInterval.Duration != newCfg.Server.HeartbeatInterval.Duration {
return fmt.Errorf("server.heartbeat_interval change requires restart")
}
if oldCfg.Server.BatchInterval.Duration != newCfg.Server.BatchInterval.Duration {
return fmt.Errorf("server.batch_interval change requires restart")
}
if oldCfg.ExposesMetrics() != newCfg.ExposesMetrics() {
return fmt.Errorf("metrics.enabled change requires restart")
}
if oldCfg.Metrics.Listen != newCfg.Metrics.Listen {
return fmt.Errorf("metrics.listen change requires restart")
}
return nil
}
// sleep returns true if context was cancelled while waiting.
func sleep(ctx context.Context, d time.Duration) bool {
t := time.NewTimer(d)

75
agent/internal/app/app_test.go
View File

@@ -182,6 +182,81 @@ func TestEndToEndPushAndReplay(t *testing.T) {
}
}
func TestReloadRejectsImmutableChangesAndKeepsOldConfig(t *testing.T) {
oldCfg := &config.Config{
AgentID: "a1",
Hostname: "h",
StateDir: "/tmp/old",
Server: config.ServerConfig{
Enabled: boolPtr(true),
URL: "http://old",
HeartbeatInterval: config.Duration{Duration: 10 * time.Second},
BatchInterval: config.Duration{Duration: 10 * time.Second},
},
Metrics: config.MetricsConfig{Enabled: false},
Checks: []config.CheckConfig{{
ID: "c1",
Command: "true",
Interval: config.Duration{Duration: time.Second},
Timeout: config.Duration{Duration: time.Second},
}},
}
newCfg := *oldCfg
newCfg.Server.URL = "http://new"
a := &App{
Cfg: oldCfg,
AgentID: "a1",
Metrics: metrics.New(),
}
if err := a.Reload(&newCfg, "a1"); err == nil {
t.Fatal("expected immutable reload rejection")
}
if a.config().Server.URL != "http://old" {
t.Fatalf("old config was replaced: %+v", a.config().Server)
}
}
func TestReloadAcceptsLabelsAndChecks(t *testing.T) {
oldCfg := &config.Config{
AgentID: "a1",
Hostname: "h",
StateDir: "/tmp/old",
Labels: map[string]string{"env": "old"},
Server: config.ServerConfig{
Enabled: boolPtr(true),
URL: "http://server",
HeartbeatInterval: config.Duration{Duration: 10 * time.Second},
BatchInterval: config.Duration{Duration: 10 * time.Second},
},
Metrics: config.MetricsConfig{Enabled: false},
Checks: []config.CheckConfig{{
ID: "c1",
Command: "true",
Interval: config.Duration{Duration: time.Second},
Timeout: config.Duration{Duration: time.Second},
}},
}
newCfg := *oldCfg
newCfg.Labels = map[string]string{"env": "new"}
newCfg.Checks = []config.CheckConfig{{
ID: "c2",
Command: "true",
Interval: config.Duration{Duration: 2 * time.Second},
Timeout: config.Duration{Duration: time.Second},
}}
a := &App{
Cfg: oldCfg,
AgentID: "a1",
Metrics: metrics.New(),
}
if err := a.Reload(&newCfg, "a1"); err != nil {
t.Fatal(err)
}
if a.config().Labels["env"] != "new" || len(a.config().Checks) != 1 || a.config().Checks[0].ID != "c2" {
t.Fatalf("reload did not apply: %+v", a.config())
}
}
func boolPtr(v bool) *bool {
return &v
}

103
agent/internal/client/client.go
View File

@@ -4,6 +4,7 @@ import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
@@ -15,6 +16,7 @@ import (
const (
MaxBatchEvents = 200
MaxBatchBytes = 1 << 20
)
type Client struct {
@@ -68,18 +70,81 @@ func (c *Client) SendEvents(ctx context.Context, events []scheduler.Event) (*Eve
return &EventBatchResponse{}, nil
}
if len(events) > MaxBatchEvents {
return nil, fmt.Errorf("batch exceeds %d events", MaxBatchEvents)
return nil, nonRetryablef("batch exceeds %d events", MaxBatchEvents)
}
wireEvents := make([]scheduler.Event, len(events))
for i, ev := range events {
wireEvents[i] = ev.WireEvent()
}
req := EventBatchRequest{AgentID: c.agentID, Events: wireEvents}
var resp EventBatchResponse
if err := c.post(ctx, "/api/v1/events", req, &resp); err != nil {
batches, err := c.splitEventBatches(req.Events)
if err != nil {
return nil, err
}
return &resp, nil
var total EventBatchResponse
for _, batch := range batches {
var resp EventBatchResponse
if err := c.post(ctx, "/api/v1/events", EventBatchRequest{AgentID: c.agentID, Events: batch}, &resp); err != nil {
return nil, err
}
total.Accepted += resp.Accepted
total.Deduplicated += resp.Deduplicated
}
return &total, nil
}
func (c *Client) splitEventBatches(events []scheduler.Event) ([][]scheduler.Event, error) {
overhead, err := c.eventBatchOverhead()
if err != nil {
return nil, err
}
batches := make([][]scheduler.Event, 0, 1)
current := make([]scheduler.Event, 0, len(events))
currentSize := overhead
for _, ev := range events {
eventSize, err := jsonSize(ev)
if err != nil {
return nil, err
}
nextSize := currentSize + eventSize
if len(current) > 0 {
nextSize++
}
if nextSize <= MaxBatchBytes {
current = append(current, ev)
currentSize = nextSize
continue
}
if len(current) == 0 {
return nil, nonRetryablef("single event batch exceeds %d bytes", MaxBatchBytes)
}
batches = append(batches, current)
current = []scheduler.Event{ev}
currentSize = overhead + eventSize
if currentSize > MaxBatchBytes {
return nil, nonRetryablef("single event batch exceeds %d bytes", MaxBatchBytes)
}
}
if len(current) > 0 {
batches = append(batches, current)
}
return batches, nil
}
func (c *Client) eventBatchOverhead() (int, error) {
size, err := jsonSize(EventBatchRequest{AgentID: c.agentID, Events: []scheduler.Event{}})
if err != nil {
return 0, err
}
return size, nil
}
func jsonSize(v any) (int, error) {
b, err := json.Marshal(v)
if err != nil {
return 0, err
}
return len(b), nil
}
func (c *Client) post(ctx context.Context, path string, body any, out any) error {
@@ -115,30 +180,28 @@ type HTTPError struct {
func (e *HTTPError) Error() string { return fmt.Sprintf("http %d: %s", e.Status, e.Body) }
type NonRetryableError struct{ Err error }
func (e *NonRetryableError) Error() string { return e.Err.Error() }
func (e *NonRetryableError) Unwrap() error { return e.Err }
func nonRetryablef(format string, args ...any) error {
return &NonRetryableError{Err: fmt.Errorf(format, args...)}
}
// Retryable reports whether the error should trigger a retry (network errors
// and 5xx are retryable; 4xx are dropped to avoid hot-loop on bad payload).
func Retryable(err error) bool {
if err == nil {
return false
}
var ne *NonRetryableError
if errors.As(err, &ne) {
return false
}
var he *HTTPError
if asErr(err, &he) {
if errors.As(err, &he) {
return he.Status >= 500
}
return true
}
func asErr(err error, target **HTTPError) bool {
for e := err; e != nil; {
if h, ok := e.(*HTTPError); ok {
*target = h
return true
}
u, ok := e.(interface{ Unwrap() error })
if !ok {
return false
}
e = u.Unwrap()
}
return false
}

72
agent/internal/client/client_test.go
View File

@@ -6,6 +6,7 @@ import (
"io"
"net/http"
"net/http/httptest"
"strings"
"testing"
"time"
@@ -57,6 +58,59 @@ func TestSendEventsBatch(t *testing.T) {
}
}
func TestSendEventsSplitsByJSONSize(t *testing.T) {
var (
requests int
total int
)
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
body, err := io.ReadAll(r.Body)
if err != nil {
t.Error(err)
http.Error(w, "read body", http.StatusInternalServerError)
return
}
if len(body) > MaxBatchBytes {
t.Errorf("body size %d exceeds %d", len(body), MaxBatchBytes)
http.Error(w, "too large", http.StatusRequestEntityTooLarge)
return
}
var got EventBatchRequest
if err := json.Unmarshal(body, &got); err != nil {
t.Error(err)
http.Error(w, "bad json", http.StatusBadRequest)
return
}
requests++
total += len(got.Events)
w.WriteHeader(http.StatusAccepted)
_ = json.NewEncoder(w).Encode(EventBatchResponse{Accepted: len(got.Events)})
}))
defer srv.Close()
events := make([]scheduler.Event, MaxBatchEvents)
for i := range events {
events[i] = scheduler.Event{
EventID: "event-" + time.Unix(int64(i), 0).UTC().Format("20060102150405"),
CheckID: "check",
ObservedAt: time.Unix(int64(i), 0).UTC(),
Status: "ok",
Output: strings.Repeat("x", 8192),
}
}
c := New(srv.URL, "tok", "a1")
resp, err := c.SendEvents(context.Background(), events)
if err != nil {
t.Fatal(err)
}
if requests < 2 {
t.Fatalf("expected split batch, got %d request", requests)
}
if total != MaxBatchEvents || resp.Accepted != MaxBatchEvents {
t.Fatalf("total=%d resp=%+v", total, resp)
}
}
func TestSendEventsTooBig(t *testing.T) {
c := New("http://x", "t", "a1")
big := make([]scheduler.Event, MaxBatchEvents+1)
@@ -64,6 +118,24 @@ func TestSendEventsTooBig(t *testing.T) {
if err == nil {
t.Fatal("expected error")
}
if Retryable(err) {
t.Fatal("oversized batch should not retry")
}
}
func TestSendEventsSingleEventTooLargeIsNotRetryable(t *testing.T) {
c := New("http://x", "t", "a1")
_, err := c.SendEvents(context.Background(), []scheduler.Event{{
EventID: "x",
CheckID: strings.Repeat("c", MaxBatchBytes),
Status: "ok",
}})
if err == nil {
t.Fatal("expected error")
}
if Retryable(err) {
t.Fatal("oversized single event should not retry")
}
}
func TestRetryableClassification(t *testing.T) {

85
agent/internal/config/config.go
View File

@@ -2,6 +2,7 @@ package config
import (
"fmt"
"math"
"os"
"regexp"
"strconv"
@@ -53,17 +54,28 @@ type MetricsConfig struct {
}
type CheckConfig struct {
ID string `toml:"id"`
Name string `toml:"name"`
Command string `toml:"command"`
Interval Duration `toml:"interval"`
Timeout Duration `toml:"timeout"`
NotificationsEnabled *bool `toml:"notifications_enabled"`
DedupeKey string `toml:"dedupe_key"`
ID string `toml:"id"`
Name string `toml:"name"`
Command string `toml:"command"`
Interval Duration `toml:"interval"`
Timeout Duration `toml:"timeout"`
NotificationsEnabled *bool `toml:"notifications_enabled"`
DedupeKey string `toml:"dedupe_key"`
ResourceLimits ResourceLimits `toml:"resource_limits"`
}
type Duration struct{ time.Duration }
// ByteSize stores a parsed TOML byte size.
type ByteSize struct{ Bytes int64 }
// ResourceLimits defines optional per-check OS resource limits.
type ResourceLimits struct {
CPUTime Duration `toml:"cpu_time"`
Memory ByteSize `toml:"memory"`
OpenFiles int `toml:"open_files"`
}
func (d *Duration) UnmarshalText(b []byte) error {
v, err := time.ParseDuration(string(b))
if err != nil {
@@ -73,6 +85,43 @@ func (d *Duration) UnmarshalText(b []byte) error {
return nil
}
func (s *ByteSize) UnmarshalText(b []byte) error {
raw := strings.TrimSpace(string(b))
if raw == "" {
return fmt.Errorf("empty byte size")
}
valueEnd := 0
for valueEnd < len(raw) && ((raw[valueEnd] >= '0' && raw[valueEnd] <= '9') || raw[valueEnd] == '.') {
valueEnd++
}
if valueEnd == 0 {
return fmt.Errorf("invalid byte size %q", raw)
}
value, err := strconv.ParseFloat(raw[:valueEnd], 64)
if err != nil || value <= 0 {
return fmt.Errorf("invalid byte size %q", raw)
}
unit := strings.ToLower(strings.TrimSpace(raw[valueEnd:]))
mul, ok := map[string]float64{
"": 1, "b": 1,
"k": 1000, "kb": 1000,
"m": 1000 * 1000, "mb": 1000 * 1000,
"g": 1000 * 1000 * 1000, "gb": 1000 * 1000 * 1000,
"kib": 1024,
"mib": 1024 * 1024,
"gib": 1024 * 1024 * 1024,
}[unit]
if !ok {
return fmt.Errorf("unknown byte size unit %q", unit)
}
bytes := value * mul
if bytes > math.MaxInt64 {
return fmt.Errorf("byte size %q is too large", raw)
}
s.Bytes = int64(bytes)
return nil
}
func Load(path string) (*Config, error) {
var c Config
md, err := toml.DecodeFile(path, &c)
@@ -174,6 +223,9 @@ func (c *Config) Validate() error {
if len(ch.DedupeKey) > 256 {
return fmt.Errorf("check %q: dedupe_key too long", ch.ID)
}
if err := validateResourceLimits(ch.ID, ch.ResourceLimits); err != nil {
return err
}
}
return nil
}
@@ -201,8 +253,23 @@ func (c CheckConfig) NotificationsOn() bool {
return c.NotificationsEnabled == nil || *c.NotificationsEnabled
}
func (c CheckConfig) Argv() []string {
return []string{"/bin/sh", "-c", c.Command}
func validateResourceLimits(checkID string, limits ResourceLimits) error {
if limits.CPUTime.Duration < 0 {
return fmt.Errorf("check %q: resource_limits.cpu_time must be > 0", checkID)
}
if limits.Memory.Bytes < 0 {
return fmt.Errorf("check %q: resource_limits.memory must be > 0", checkID)
}
if limits.Memory.Bytes > 0 && limits.Memory.Bytes < 1024*1024 {
return fmt.Errorf("check %q: resource_limits.memory must be >= 1MiB", checkID)
}
if limits.OpenFiles < 0 {
return fmt.Errorf("check %q: resource_limits.open_files must be > 0", checkID)
}
if limits.OpenFiles > 0 && limits.OpenFiles < 16 {
return fmt.Errorf("check %q: resource_limits.open_files must be >= 16", checkID)
}
return nil
}
func ValidateID(field, v string) error {

56
agent/internal/config/config_test.go
View File

@@ -48,6 +48,59 @@ func TestLoadMinimal(t *testing.T) {
}
}
func TestLoadResourceLimits(t *testing.T) {
c, err := Load(writeTmp(t, `
agent_id = "host-1"
state_dir = "/tmp/x"
[server]
enabled = true
url = "http://localhost"
[[checks]]
id = "c1"
command = "true"
interval = "10s"
timeout = "5s"
resource_limits = { cpu_time = "2s", memory = "256MiB", open_files = 128 }
`))
if err != nil {
t.Fatal(err)
}
limits := c.Checks[0].ResourceLimits
if limits.CPUTime.Duration != 2_000_000_000 || limits.Memory.Bytes != 256*1024*1024 || limits.OpenFiles != 128 {
t.Fatalf("unexpected limits: %+v", limits)
}
}
func TestValidateBadResourceLimits(t *testing.T) {
for name, line := range map[string]string{
"memory_too_small": `resource_limits = { memory = "512KiB" }`,
"open_files_too_small": `resource_limits = { open_files = 8 }`,
} {
t.Run(name, func(t *testing.T) {
_, err := Load(writeTmp(t, `
agent_id = "host-1"
state_dir = "/tmp/x"
[server]
enabled = true
url = "http://localhost"
[[checks]]
id = "c1"
command = "true"
interval = "10s"
timeout = "5s"
`+line+`
`))
if err == nil {
t.Fatal("expected resource limit error")
}
})
}
}
func TestLoadMultilineCommand(t *testing.T) {
c, err := Load(writeTmp(t, `
agent_id = "host-1"
@@ -72,9 +125,6 @@ timeout = "5s"
if !strings.Contains(c.Checks[0].Command, "echo ok") {
t.Fatalf("command not loaded: %q", c.Checks[0].Command)
}
if got := c.Checks[0].Argv(); len(got) != 3 || got[0] != "/bin/sh" || got[1] != "-c" {
t.Fatalf("argv: %#v", got)
}
}
func TestCommandArrayRejected(t *testing.T) {

77
agent/internal/metrics/metrics.go
View File

@@ -13,24 +13,29 @@ import (
type Metrics struct {
Registry *prometheus.Registry
CheckRuns *prometheus.CounterVec
CheckSkipped *prometheus.CounterVec
CheckDuration *prometheus.HistogramVec
CheckStatus *prometheus.GaugeVec
CheckExitCode *prometheus.GaugeVec
CheckLastRun *prometheus.GaugeVec
CheckLastSuccess *prometheus.GaugeVec
CheckInterval *prometheus.GaugeVec
ChecksConfigured prometheus.Gauge
SpoolDepth prometheus.Gauge
SpoolBytes prometheus.Gauge
SendAttempts *prometheus.CounterVec
SendFailures *prometheus.CounterVec
EventsAccepted prometheus.Counter
EventsDedup prometheus.Counter
EventsDropped *prometheus.CounterVec
LastHeartbeat prometheus.Gauge
BuildInfo *prometheus.GaugeVec
CheckRuns *prometheus.CounterVec
CheckSkipped *prometheus.CounterVec
CheckDuration *prometheus.HistogramVec
CheckStatus *prometheus.GaugeVec
CheckExitCode *prometheus.GaugeVec
CheckLastRun *prometheus.GaugeVec
CheckLastSuccess *prometheus.GaugeVec
CheckInterval *prometheus.GaugeVec
ChecksConfigured prometheus.Gauge
ConfigLoadSuccess prometheus.Gauge
ConfigReloadTotal *prometheus.CounterVec
ConfigLastReloadSuccess prometheus.Gauge
ResourceLimitApplyFailures *prometheus.CounterVec
EventsChannelFull prometheus.Counter
SpoolDepth prometheus.Gauge
SpoolBytes prometheus.Gauge
SendAttempts *prometheus.CounterVec
SendFailures *prometheus.CounterVec
EventsAccepted prometheus.Counter
EventsDedup prometheus.Counter
EventsDropped *prometheus.CounterVec
LastHeartbeat prometheus.Gauge
BuildInfo *prometheus.GaugeVec
}
// StatusCode maps a CheckResultEvent status string to a numeric gauge value.
@@ -115,6 +120,26 @@ func New() *Metrics {
Name: "monlet_agent_checks_configured",
Help: "Number of checks configured.",
})
m.ConfigLoadSuccess = prometheus.NewGauge(prometheus.GaugeOpts{
Name: "monlet_agent_config_load_success",
Help: "Whether the last agent config load or reload succeeded.",
})
m.ConfigReloadTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
Name: "monlet_agent_config_reload_total",
Help: "Number of config reload attempts by result.",
}, []string{"result"})
m.ConfigLastReloadSuccess = prometheus.NewGauge(prometheus.GaugeOpts{
Name: "monlet_agent_config_last_reload_success_timestamp_seconds",
Help: "Unix timestamp of the last successful config reload.",
})
m.ResourceLimitApplyFailures = prometheus.NewCounterVec(prometheus.CounterOpts{
Name: "monlet_agent_resource_limit_apply_failures_total",
Help: "Number of failed resource limit applications by check and resource.",
}, []string{"check_id", "resource"})
m.EventsChannelFull = prometheus.NewCounter(prometheus.CounterOpts{
Name: "monlet_agent_events_channel_full_total",
Help: "Number of times scheduler event delivery would block because the agent event channel was full.",
})
m.LastHeartbeat = prometheus.NewGauge(prometheus.GaugeOpts{
Name: "monlet_agent_last_heartbeat_timestamp_seconds",
Help: "Unix timestamp of the last successful heartbeat ack.",
@@ -126,6 +151,9 @@ func New() *Metrics {
reg.MustRegister(m.CheckRuns, m.CheckSkipped, m.CheckDuration,
m.CheckStatus, m.CheckExitCode, m.CheckLastRun, m.CheckLastSuccess,
m.CheckInterval, m.ChecksConfigured,
m.ConfigLoadSuccess, m.ConfigReloadTotal, m.ConfigLastReloadSuccess,
m.ResourceLimitApplyFailures,
m.EventsChannelFull,
m.SpoolDepth, m.SpoolBytes,
m.SendAttempts, m.SendFailures,
m.EventsAccepted, m.EventsDedup, m.EventsDropped,
@@ -133,6 +161,19 @@ func New() *Metrics {
return m
}
// DeleteCheck removes metric series for a check that no longer exists.
func (m *Metrics) DeleteCheck(checkID string) {
m.CheckRuns.DeletePartialMatch(prometheus.Labels{"check_id": checkID})
m.CheckSkipped.DeleteLabelValues(checkID)
m.CheckDuration.DeleteLabelValues(checkID)
m.CheckStatus.DeleteLabelValues(checkID)
m.CheckExitCode.DeleteLabelValues(checkID)
m.CheckLastRun.DeleteLabelValues(checkID)
m.CheckLastSuccess.DeleteLabelValues(checkID)
m.CheckInterval.DeleteLabelValues(checkID)
m.ResourceLimitApplyFailures.DeletePartialMatch(prometheus.Labels{"check_id": checkID})
}
// Serve blocks until ctx is done; the listener is closed on shutdown.
func (m *Metrics) Serve(ctx context.Context, addr string) error {
mux := http.NewServeMux()

2
agent/internal/metrics/metrics_test.go
View File

@@ -14,6 +14,7 @@ func TestMetricsExposeRegistered(t *testing.T) {
m := New()
m.CheckRuns.WithLabelValues("c1", "ok").Inc()
m.SpoolDepth.Set(7)
m.EventsChannelFull.Inc()
srv := httptest.NewServer(promhttp.HandlerFor(m.Registry, promhttp.HandlerOpts{}))
defer srv.Close()
resp, err := http.Get(srv.URL)
@@ -26,6 +27,7 @@ func TestMetricsExposeRegistered(t *testing.T) {
for _, want := range []string{
`monlet_agent_check_runs_total{check_id="c1",status="ok"} 1`,
`monlet_agent_spool_events 7`,
`monlet_agent_events_channel_full_total 1`,
} {
if !strings.Contains(s, want) {
t.Errorf("missing %q in output", want)

115
agent/internal/runner/runner.go
View File

@@ -6,6 +6,8 @@ import (
"errors"
"fmt"
"os/exec"
"strconv"
"strings"
"syscall"
"time"
"unicode/utf8"
@@ -14,8 +16,9 @@ import (
var replacementRune = []byte("<22>")
const (
MaxOutputBytes = 8192
truncMarkerFmt = "...[truncated %d bytes]"
MaxOutputBytes = 8192
truncMarkerFmt = "...[truncated %d bytes]"
limitFailPrefix = "__MONLET_RESOURCE_LIMIT_FAILED__:"
)
type Status string
@@ -28,16 +31,41 @@ const (
)
type Result struct {
Status Status
ExitCode int
DurationMs int64
Output string
OutputTruncated bool
Status Status
ExitCode int
DurationMs int64
Output string
OutputTruncated bool
ResourceLimitFailure string
}
// ResourceLimits contains best-effort shell ulimit values for a check command.
type ResourceLimits struct {
CPUTime time.Duration
MemoryBytes int64
OpenFiles int
}
func (l ResourceLimits) IsZero() bool {
return l.CPUTime == 0 && l.MemoryBytes == 0 && l.OpenFiles == 0
}
// Run executes argv with timeout. Stdout+stderr are captured into a single
// buffer; both are truncated to 8 KiB UTF-8 bytes if exceeded.
func Run(ctx context.Context, argv []string, timeout time.Duration) Result {
return runArgv(ctx, argv, timeout)
}
// RunCommand executes a shell command with optional resource limits.
func RunCommand(ctx context.Context, command string, timeout time.Duration, limits ResourceLimits) Result {
argv := []string{"/bin/sh", "-c", command}
if !limits.IsZero() {
argv = limitedShellArgv(command, limits)
}
return runArgv(ctx, argv, timeout)
}
func runArgv(ctx context.Context, argv []string, timeout time.Duration) Result {
start := time.Now()
cctx, cancel := context.WithTimeout(ctx, timeout)
defer cancel()
@@ -67,6 +95,23 @@ func Run(ctx context.Context, argv []string, timeout time.Duration) Result {
r.Output = fmt.Sprintf("critical: check timed out after %s", timeout)
return r
}
if isExitCode(err, 125) {
if resource, ok := detectLimitFailure(out); ok {
r.Status = StatusUnknown
r.ExitCode = -1
r.Output = "unknown: failed to apply resource limit: " + resource
r.OutputTruncated = false
r.ResourceLimitFailure = resource
return r
}
}
if isSignal(err, syscall.SIGXCPU) {
r.Status = StatusCritical
r.ExitCode = 2
r.Output = "critical: CPU time limit exceeded"
r.OutputTruncated = false
return r
}
if err != nil {
var ee *exec.ExitError
if errors.As(err, &ee) {
@@ -74,6 +119,8 @@ func Run(ctx context.Context, argv []string, timeout time.Duration) Result {
} else {
r.Status = StatusUnknown
r.ExitCode = -1
// Launch failure (e.g. shell missing); surface exec error, never check output.
r.Output, r.OutputTruncated = TruncateUTF8([]byte("unknown: failed to start check: "+err.Error()), MaxOutputBytes)
return r
}
}
@@ -81,6 +128,60 @@ func Run(ctx context.Context, argv []string, timeout time.Duration) Result {
return r
}
func isExitCode(err error, code int) bool {
var ee *exec.ExitError
return errors.As(err, &ee) && ee.ExitCode() == code
}
func limitedShellArgv(command string, limits ResourceLimits) []string {
cpu := ""
if limits.CPUTime > 0 {
cpu = strconv.FormatInt(int64((limits.CPUTime+time.Second-1)/time.Second), 10)
}
mem := ""
if limits.MemoryBytes > 0 {
mem = strconv.FormatInt((limits.MemoryBytes+1023)/1024, 10)
}
openFiles := ""
if limits.OpenFiles > 0 {
openFiles = strconv.Itoa(limits.OpenFiles)
}
script := `
cmd=$1
cpu=$2
mem=$3
open_files=$4
if [ -n "$cpu" ]; then ulimit -t "$cpu" 2>/dev/null || { printf '%s\n' '` + limitFailPrefix + `cpu_time'; exit 125; }; fi
if [ -n "$mem" ]; then ulimit -v "$mem" 2>/dev/null || { printf '%s\n' '` + limitFailPrefix + `memory'; exit 125; }; fi
if [ -n "$open_files" ]; then ulimit -n "$open_files" 2>/dev/null || { printf '%s\n' '` + limitFailPrefix + `open_files'; exit 125; }; fi
exec /bin/sh -c "$cmd"
`
return []string{"/bin/sh", "-c", script, "monlet-limit-wrapper", command, cpu, mem, openFiles}
}
func detectLimitFailure(output string) (string, bool) {
for _, line := range strings.Split(output, "\n") {
line = strings.TrimSpace(line)
if strings.HasPrefix(line, limitFailPrefix) {
resource := strings.TrimPrefix(line, limitFailPrefix)
if resource != "cpu_time" && resource != "memory" && resource != "open_files" {
resource = "unknown"
}
return resource, true
}
}
return "", false
}
func isSignal(err error, sig syscall.Signal) bool {
var ee *exec.ExitError
if !errors.As(err, &ee) {
return false
}
ws, ok := ee.Sys().(syscall.WaitStatus)
return ok && ws.Signaled() && ws.Signal() == sig
}
func MapExit(code int) Status {
switch code {
case 0:

39
agent/internal/runner/runner_test.go
View File

@@ -2,6 +2,7 @@ package runner
import (
"context"
"runtime"
"strings"
"testing"
"time"
@@ -114,3 +115,41 @@ func TestRunNoExec(t *testing.T) {
t.Fatalf("got %+v", r)
}
}
func TestRunCommandWithLimitsOK(t *testing.T) {
r := RunCommand(context.Background(), "echo ok", 2*time.Second, ResourceLimits{
CPUTime: time.Second,
OpenFiles: 64,
})
if r.Status != StatusOK || r.ExitCode != 0 {
t.Fatalf("got %+v", r)
}
if !strings.Contains(r.Output, "ok") {
t.Fatalf("output: %q", r.Output)
}
}
func TestRunCommandCPULimitExceeded(t *testing.T) {
if runtime.GOOS == "windows" {
t.Skip("ulimit is Unix-only")
}
r := RunCommand(context.Background(), "while :; do :; done", 5*time.Second, ResourceLimits{
CPUTime: time.Second,
})
if r.Status != StatusCritical || r.ExitCode != 2 {
t.Fatalf("got %+v", r)
}
if !strings.Contains(r.Output, "CPU time limit exceeded") {
t.Fatalf("output: %q", r.Output)
}
}
func TestDetectLimitFailureBoundsResourceLabel(t *testing.T) {
resource, ok := detectLimitFailure(limitFailPrefix + "dynamic-user-output")
if !ok {
t.Fatal("expected limit failure")
}
if resource != "unknown" {
t.Fatalf("resource label must be bounded, got %q", resource)
}
}

351
agent/internal/scheduler/scheduler.go
View File

@@ -2,6 +2,7 @@ package scheduler
import (
"context"
"reflect"
"sync"
"time"
@@ -33,82 +34,316 @@ func (e Event) WireEvent() Event {
}
type Hooks struct {
OnSkipped func(checkID string)
OnResult func(ev Event)
OnSkipped func(checkID string)
OnResult func(ev Event)
OnEventChannelFull func()
OnResourceLimitFailure func(checkID string, resource string)
OnStopped func(checkID string)
}
// Run starts one goroutine per check. Emits events to out. Blocks until ctx is done.
func Run(ctx context.Context, agentID string, checks []config.CheckConfig, out chan<- Event, h Hooks) {
var wg sync.WaitGroup
for i := range checks {
c := checks[i]
wg.Add(1)
go func() {
defer wg.Done()
runCheck(ctx, agentID, c, out, h)
}()
// Manager owns per-check workers and applies hot config updates.
type Manager struct {
ctx context.Context
agentID string
out chan<- Event
hooks Hooks
mu sync.Mutex
workers map[string]*checkWorker
draining map[string]struct{}
pending map[string]config.CheckConfig
all []*checkWorker
stopped bool
}
// NewManager creates a scheduler manager bound to the parent context.
func NewManager(ctx context.Context, agentID string, out chan<- Event, h Hooks) *Manager {
return &Manager{
ctx: ctx,
agentID: agentID,
out: out,
hooks: h,
workers: make(map[string]*checkWorker),
draining: make(map[string]struct{}),
pending: make(map[string]config.CheckConfig),
}
wg.Wait()
}
func runCheck(ctx context.Context, agentID string, c config.CheckConfig, out chan<- Event, h Hooks) {
t := time.NewTicker(c.Interval.Duration)
defer t.Stop()
var running bool
var mu sync.Mutex
// Update applies the current check set without killing in-flight runs.
func (m *Manager) Update(checks []config.CheckConfig) {
m.mu.Lock()
defer m.mu.Unlock()
if m.stopped {
return
}
next := make(map[string]config.CheckConfig, len(checks))
for _, c := range checks {
next[c.ID] = c
if w, ok := m.workers[c.ID]; ok {
if _, draining := m.draining[c.ID]; draining {
m.pending[c.ID] = c
} else {
w.Update(c)
}
continue
}
id := c.ID
w := newCheckWorker(m.ctx, m.agentID, c, m.out, m.hooks, func(stopped *checkWorker) {
m.workerStopped(id, stopped)
})
m.workers[c.ID] = w
m.all = append(m.all, w)
}
for id, w := range m.workers {
if _, ok := next[id]; ok {
continue
}
m.draining[id] = struct{}{}
delete(m.pending, id)
w.Stop()
}
}
exec := func() {
mu.Lock()
// Stop prevents new runs and lets in-flight checks finish or observe ctx cancel.
func (m *Manager) Stop() {
m.mu.Lock()
defer m.mu.Unlock()
if m.stopped {
return
}
m.stopped = true
for _, w := range m.workers {
w.Stop()
}
}
// Wait blocks until all workers have drained.
func (m *Manager) Wait() {
m.mu.Lock()
workers := append([]*checkWorker(nil), m.all...)
m.mu.Unlock()
for _, w := range workers {
w.Wait()
}
}
func (m *Manager) workerStopped(checkID string, w *checkWorker) {
m.mu.Lock()
defer m.mu.Unlock()
if current, ok := m.workers[checkID]; !ok || current != w {
return
}
delete(m.workers, checkID)
delete(m.draining, checkID)
if m.stopped {
delete(m.pending, checkID)
return
}
if cfg, ok := m.pending[checkID]; ok {
delete(m.pending, checkID)
next := newCheckWorker(m.ctx, m.agentID, cfg, m.out, m.hooks, func(stopped *checkWorker) {
m.workerStopped(checkID, stopped)
})
m.workers[checkID] = next
m.all = append(m.all, next)
}
}
type checkWorker struct {
ctx context.Context
agentID string
out chan<- Event
hooks Hooks
onDone func(*checkWorker)
updateCh chan config.CheckConfig
stopCh chan struct{}
doneCh chan struct{}
stopOnce sync.Once
wg sync.WaitGroup
}
func newCheckWorker(ctx context.Context, agentID string, c config.CheckConfig, out chan<- Event, h Hooks, onDone func(*checkWorker)) *checkWorker {
w := &checkWorker{
ctx: ctx,
agentID: agentID,
out: out,
hooks: h,
onDone: onDone,
updateCh: make(chan config.CheckConfig, 1),
stopCh: make(chan struct{}),
doneCh: make(chan struct{}, 1),
}
w.wg.Add(1)
go w.loop(c)
return w
}
func (w *checkWorker) Update(c config.CheckConfig) {
select {
case w.updateCh <- c:
default:
select {
case <-w.updateCh:
default:
}
w.updateCh <- c
}
}
func (w *checkWorker) Stop() {
w.stopOnce.Do(func() { close(w.stopCh) })
}
func (w *checkWorker) Wait() {
w.wg.Wait()
}
func (w *checkWorker) loop(initial config.CheckConfig) {
defer func() {
if w.onDone != nil {
w.onDone(w)
}
w.wg.Done()
}()
cfg := initial
t := time.NewTicker(cfg.Interval.Duration)
defer t.Stop()
running := false
stopping := false
runAfterCurrent := false
ctxDone := w.ctx.Done()
stopCh := w.stopCh
start := func(c config.CheckConfig) {
if running {
mu.Unlock()
if h.OnSkipped != nil {
h.OnSkipped(c.ID)
if w.hooks.OnSkipped != nil {
w.hooks.OnSkipped(c.ID)
}
return
}
running = true
mu.Unlock()
defer func() {
mu.Lock()
running = false
mu.Unlock()
go func() {
w.run(c)
w.doneCh <- struct{}{}
}()
res := runner.Run(ctx, c.Argv(), c.Timeout.Duration)
id, err := eventid.New()
if err != nil {
return
}
ev := Event{
EventID: id,
CheckID: c.ID,
ObservedAt: time.Now().UTC(),
Status: string(res.Status),
ExitCode: res.ExitCode,
DurationMs: res.DurationMs,
Output: res.Output,
OutputTruncated: res.OutputTruncated,
NotificationsEnabled: boolPtr(c.NotificationsOn()),
IncidentKey: incidentKey(agentID, c),
}
if h.OnResult != nil {
h.OnResult(ev)
}
select {
case out <- ev:
case <-ctx.Done():
}
}
// Fire first run immediately so smoke tests don't wait a full interval.
go exec()
start(cfg)
for {
select {
case <-ctx.Done():
return
case <-ctxDone:
stopping = true
ctxDone = nil
if !running {
if w.hooks.OnStopped != nil {
w.hooks.OnStopped(cfg.ID)
}
return
}
case <-stopCh:
stopping = true
stopCh = nil
if !running {
if w.hooks.OnStopped != nil {
w.hooks.OnStopped(cfg.ID)
}
return
}
case next := <-w.updateCh:
if reflect.DeepEqual(cfg, next) {
continue
}
cfg = next
t.Reset(cfg.Interval.Duration)
if running {
runAfterCurrent = true
} else {
start(cfg)
}
case <-w.doneCh:
running = false
if !stopping {
select {
case <-stopCh:
stopping = true
stopCh = nil
default:
}
}
if stopping {
if w.hooks.OnStopped != nil {
w.hooks.OnStopped(cfg.ID)
}
return
}
if runAfterCurrent {
runAfterCurrent = false
for {
select {
case next := <-w.updateCh:
if !reflect.DeepEqual(cfg, next) {
cfg = next
t.Reset(cfg.Interval.Duration)
}
default:
start(cfg)
goto nextLoop
}
}
nextLoop:
continue
}
case <-t.C:
go exec()
if stopping {
continue
}
start(cfg)
}
}
}
func (w *checkWorker) run(c config.CheckConfig) {
res := runner.RunCommand(w.ctx, c.Command, c.Timeout.Duration, runner.ResourceLimits{
CPUTime: c.ResourceLimits.CPUTime.Duration,
MemoryBytes: c.ResourceLimits.Memory.Bytes,
OpenFiles: c.ResourceLimits.OpenFiles,
})
if res.ResourceLimitFailure != "" && w.hooks.OnResourceLimitFailure != nil {
w.hooks.OnResourceLimitFailure(c.ID, res.ResourceLimitFailure)
}
id, err := eventid.New()
if err != nil {
return
}
ev := Event{
EventID: id,
CheckID: c.ID,
ObservedAt: time.Now().UTC(),
Status: string(res.Status),
ExitCode: res.ExitCode,
DurationMs: res.DurationMs,
Output: res.Output,
OutputTruncated: res.OutputTruncated,
NotificationsEnabled: boolPtr(c.NotificationsOn()),
IncidentKey: incidentKey(w.agentID, c),
}
if w.hooks.OnResult != nil {
w.hooks.OnResult(ev)
}
select {
case w.out <- ev:
case <-w.ctx.Done():
default:
if w.hooks.OnEventChannelFull != nil {
w.hooks.OnEventChannelFull()
}
select {
case w.out <- ev:
case <-w.ctx.Done():
case <-w.stopCh:
}
}
}

172
agent/internal/scheduler/scheduler_test.go
View File

@@ -2,6 +2,7 @@ package scheduler
import (
"context"
"strings"
"sync/atomic"
"testing"
"time"
@@ -9,6 +10,14 @@ import (
"github.com/monlet/agent/internal/config"
)
func runScheduler(ctx context.Context, agentID string, checks []config.CheckConfig, out chan<- Event, h Hooks) {
m := NewManager(ctx, agentID, out, h)
m.Update(checks)
<-ctx.Done()
m.Stop()
m.Wait()
}
func mkCheck(id, cmd string, interval, timeout time.Duration) config.CheckConfig {
return config.CheckConfig{
ID: id,
@@ -22,7 +31,7 @@ func TestEmitsEvents(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
defer cancel()
out := make(chan Event, 10)
go Run(ctx, "a1", []config.CheckConfig{mkCheck("c1", "echo hi", 200*time.Millisecond, 100*time.Millisecond)}, out, Hooks{})
go runScheduler(ctx, "a1", []config.CheckConfig{mkCheck("c1", "echo hi", 200*time.Millisecond, 100*time.Millisecond)}, out, Hooks{})
select {
case ev := <-out:
if ev.CheckID != "c1" || ev.Status != "ok" {
@@ -43,9 +52,168 @@ func TestNoOverlapSkips(t *testing.T) {
var skipped int32
hooks := Hooks{OnSkipped: func(string) { atomic.AddInt32(&skipped, 1) }}
c := mkCheck("slow", "sleep 1", 100*time.Millisecond, 900*time.Millisecond)
go Run(ctx, "a1", []config.CheckConfig{c}, out, hooks)
go runScheduler(ctx, "a1", []config.CheckConfig{c}, out, hooks)
<-ctx.Done()
if atomic.LoadInt32(&skipped) == 0 {
t.Fatal("expected at least one skipped tick")
}
}
func TestReportsFullEventChannel(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
out := make(chan Event, 1)
out <- Event{EventID: "occupied"}
var full int32
m := NewManager(ctx, "a1", out, Hooks{
OnEventChannelFull: func() { atomic.AddInt32(&full, 1) },
})
m.Update([]config.CheckConfig{mkCheck("c1", "echo hi", time.Hour, time.Second)})
deadline := time.Now().Add(time.Second)
for time.Now().Before(deadline) {
if atomic.LoadInt32(&full) > 0 {
break
}
time.Sleep(10 * time.Millisecond)
}
if atomic.LoadInt32(&full) == 0 {
t.Fatal("expected full channel hook")
}
<-out
select {
case ev := <-out:
if ev.CheckID != "c1" {
t.Fatalf("unexpected event: %+v", ev)
}
case <-ctx.Done():
t.Fatal("blocked event was not delivered")
}
m.Stop()
m.Wait()
}
func TestManagerReloadChangedCheckWaitsForRunning(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
out := make(chan Event, 10)
m := NewManager(ctx, "a1", out, Hooks{})
m.Update([]config.CheckConfig{mkCheck("c1", "sleep 0.3; echo old", time.Second, time.Second)})
time.Sleep(50 * time.Millisecond)
m.Update([]config.CheckConfig{mkCheck("c1", "echo new", time.Second, time.Second)})
var first, second Event
select {
case first = <-out:
case <-ctx.Done():
t.Fatal("no first event")
}
select {
case second = <-out:
case <-ctx.Done():
t.Fatal("no second event")
}
m.Stop()
m.Wait()
if !strings.Contains(first.Output, "old") {
t.Fatalf("first output: %q", first.Output)
}
if !strings.Contains(second.Output, "new") {
t.Fatalf("second output: %q", second.Output)
}
}
func TestManagerReloadRemovedCheckLetsRunningFinish(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
out := make(chan Event, 10)
stopped := make(chan string, 1)
m := NewManager(ctx, "a1", out, Hooks{OnStopped: func(id string) { stopped <- id }})
m.Update([]config.CheckConfig{mkCheck("c1", "sleep 0.2; echo old", time.Second, time.Second)})
time.Sleep(50 * time.Millisecond)
m.Update(nil)
select {
case ev := <-out:
if !strings.Contains(ev.Output, "old") {
t.Fatalf("output: %q", ev.Output)
}
case <-ctx.Done():
t.Fatal("removed running check did not finish")
}
select {
case id := <-stopped:
if id != "c1" {
t.Fatalf("stopped id: %q", id)
}
case <-ctx.Done():
t.Fatal("removed check did not stop")
}
m.Stop()
m.Wait()
}
func TestManagerRemoveThenReaddSameCheckDoesNotOverlap(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel()
out := make(chan Event, 10)
m := NewManager(ctx, "a1", out, Hooks{})
m.Update([]config.CheckConfig{mkCheck("c1", "sleep 0.25; echo old", time.Second, time.Second)})
time.Sleep(50 * time.Millisecond)
m.Update(nil)
m.Update([]config.CheckConfig{mkCheck("c1", "echo new", time.Second, time.Second)})
var first, second Event
select {
case first = <-out:
case <-ctx.Done():
t.Fatal("no first event")
}
select {
case second = <-out:
case <-ctx.Done():
t.Fatal("no second event")
}
m.Stop()
m.Wait()
if !strings.Contains(first.Output, "old") {
t.Fatalf("first output: %q", first.Output)
}
if !strings.Contains(second.Output, "new") {
t.Fatalf("second output: %q", second.Output)
}
}
func TestManagerUsesLatestPendingReloadAfterRunningFinishes(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel()
out := make(chan Event, 10)
m := NewManager(ctx, "a1", out, Hooks{})
m.Update([]config.CheckConfig{mkCheck("c1", "sleep 0.25; echo old", time.Second, time.Second)})
time.Sleep(50 * time.Millisecond)
m.Update([]config.CheckConfig{mkCheck("c1", "echo intermediate", time.Second, time.Second)})
m.Update([]config.CheckConfig{mkCheck("c1", "echo latest", time.Second, time.Second)})
var first, second Event
select {
case first = <-out:
case <-ctx.Done():
t.Fatal("no first event")
}
select {
case second = <-out:
case <-ctx.Done():
t.Fatal("no second event")
}
m.Stop()
m.Wait()
if !strings.Contains(first.Output, "old") {
t.Fatalf("first output: %q", first.Output)
}
if strings.Contains(second.Output, "intermediate") || !strings.Contains(second.Output, "latest") {
t.Fatalf("second output: %q", second.Output)
}
}

2
agent/systemd/monlet-agent.service.example
View File

@@ -6,8 +6,10 @@ Wants=network-online.target
[Service]
Type=simple
ExecStart=/usr/local/bin/monlet-agent -config /etc/monlet/agent.toml
ExecReload=/bin/kill -HUP $MAINPID
Restart=always
RestartSec=5s
TimeoutStopSec=35s
User=monlet
Group=monlet
StateDirectory=monlet-agent