diff --git a/cmd/epmon/main.go b/cmd/epmon/main.go index 0fac89c..ce8c2a7 100644 --- a/cmd/epmon/main.go +++ b/cmd/epmon/main.go @@ -22,10 +22,7 @@ import ( "github.com/epmon-dev/epmon/internal/metrics" "github.com/epmon-dev/epmon/internal/scheduler" "github.com/epmon-dev/epmon/internal/store" - // Side-effect import: registers the "sqlite" driver with the store - // registry. main never names the adapter type — cfg.Database.Driver - // picks it. Same pattern as database/sql drivers. - _ "github.com/epmon-dev/epmon/internal/store/sqlite" + "github.com/epmon-dev/epmon/internal/store/sqlite" ) const ( @@ -82,7 +79,7 @@ func runDefault(args []string, stdout, stderr io.Writer) int { return exitConfig } - st, err := store.Open(context.Background(), cfg.Database.Driver, cfg.Database.DSN) + st, err := store.Open(context.Background(), cfg.Database.Driver, sqliteDSN(cfg)) if err != nil { fmt.Fprintf(stderr, "epmon: %v\n", err) return exitStorage @@ -120,13 +117,7 @@ func runDefault(args []string, stdout, stderr io.Writer) int { root.Handle("/metrics", registry.Handler()) root.Handle("/", api.New(cfg, st, nil).Handler()) - srv := &http.Server{ - Addr: cfg.Server.Addr, - Handler: api.Log(root), - ReadTimeout: 10 * time.Second, - WriteTimeout: 15 * time.Second, - IdleTimeout: 60 * time.Second, - } + srv := newHTTPServer(cfg, api.Log(root)) // serveErr carries a bind/serve failure back to runDefault so boot can // report exitListen instead of hanging until the next signal. serveErr := make(chan error, 1) @@ -158,6 +149,29 @@ func runDefault(args []string, stdout, stderr io.Writer) int { return exitOK } +// sqliteDSN carries the configured SQLite busy budget inside the DSN via +// the adapter's documented pragma parameter. The store registry only +// passes an opaque connection string, so driver-specific tuning travels +// this way; other drivers receive the DSN untouched. +func sqliteDSN(cfg *config.Config) string { + if cfg.Database.Driver == "sqlite" { + return sqlite.WithBusyTimeout(cfg.Database.DSN, cfg.Storage.BusyTimeoutMs) + } + return cfg.Database.DSN +} + +// newHTTPServer builds the API server from the validated config timeouts. +// IdleTimeout has no knob and stays at the 60s default. +func newHTTPServer(cfg *config.Config, handler http.Handler) *http.Server { + return &http.Server{ + Addr: cfg.Server.Addr, + Handler: handler, + ReadTimeout: cfg.Server.ReadTimeout.Std(), + WriteTimeout: cfg.Server.WriteTimeout.Std(), + IdleTimeout: 60 * time.Second, + } +} + // getConfigArg returns the value after -config/--config in args, or the // default. Both dash forms (and --config=) are accepted: the README, // the usage strings and the Dockerfile all spell the single-dash form. diff --git a/cmd/epmon/server_test.go b/cmd/epmon/server_test.go new file mode 100644 index 0000000..e2d943d --- /dev/null +++ b/cmd/epmon/server_test.go @@ -0,0 +1,46 @@ +package main + +import ( + "net/http" + "strings" + "testing" + "time" + + "github.com/epmon-dev/epmon/internal/config" +) + +func TestNewHTTPServerUsesConfigTimeouts(t *testing.T) { + cfg := config.Default() + cfg.Server.Addr = "127.0.0.1:18080" + cfg.Server.ReadTimeout = config.Duration(7 * time.Second) + cfg.Server.WriteTimeout = config.Duration(11 * time.Second) + + srv := newHTTPServer(cfg, http.NewServeMux()) + if srv.Addr != "127.0.0.1:18080" { + t.Errorf("Addr = %q", srv.Addr) + } + if srv.ReadTimeout != 7*time.Second { + t.Errorf("ReadTimeout = %v, want 7s (must follow server.read_timeout)", srv.ReadTimeout) + } + if srv.WriteTimeout != 11*time.Second { + t.Errorf("WriteTimeout = %v, want 11s (must follow server.write_timeout)", srv.WriteTimeout) + } +} + +func TestSqliteDSNCarriesBusyTimeout(t *testing.T) { + cfg := config.Default() + cfg.Database.Driver = "sqlite" + cfg.Database.DSN = "epmon.db" + cfg.Storage.BusyTimeoutMs = 2500 + + got := sqliteDSN(cfg) + if !strings.Contains(got, "busy_timeout(2500)") { + t.Errorf("sqlite DSN = %q, want busy_timeout(2500) applied", got) + } + + cfg.Database.Driver = "postgres" + cfg.Database.DSN = "postgres://db/x" + if got := sqliteDSN(cfg); got != "postgres://db/x" { + t.Errorf("non-sqlite DSN = %q, want untouched", got) + } +} diff --git a/internal/config/config.go b/internal/config/config.go index 8a30bf4..3043325 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -275,6 +275,10 @@ type Service struct { FailureThreshold int `yaml:"failure_threshold" json:"failure_threshold"` Enabled *bool `yaml:"enabled" json:"enabled"` Aliases []string `yaml:"aliases" json:"aliases"` + // MaxBodyBytes caps body_contains inspection for this service. It is + // resolved from probes.max_body_bytes at load and hidden from file + // formats (yaml/json "-") so the fleet knob stays the only source. + MaxBodyBytes int64 `yaml:"-" json:"-"` } // TLSSkipVerify is the legacy alias for InsecureSkipVerify. @@ -338,18 +342,10 @@ type CORSConfig struct { AllowedOrigins []string `yaml:"allowed_origins" json:"allowed_origins"` } -// RetentionConfig holds day-window retention. -type RetentionConfig struct { - ChecksDays int `yaml:"checks_days" json:"checks_days"` - RollupsDays int `yaml:"rollups_days" json:"rollups_days"` -} - -// Storage selects the database file and write budgets. +// Storage tunes the bundled database adapter. database.dsn selects the +// file; only the adapter-relevant budgets live here. type Storage struct { - Path string `yaml:"path" json:"path"` - BusyTimeoutMs int `yaml:"busy_timeout_ms" json:"busy_timeout_ms"` - WriteTimeout Duration `yaml:"write_timeout" json:"write_timeout"` - Retention RetentionConfig `yaml:"retention" json:"retention"` + BusyTimeoutMs int `yaml:"busy_timeout_ms" json:"busy_timeout_ms"` } // History controls day-bucket timezone. @@ -357,16 +353,11 @@ type History struct { Timezone string `yaml:"timezone" json:"timezone"` } -// RateLimitConfig bounds write throughput. -type RateLimitConfig struct { - RequestsPerMinute int `yaml:"requests_per_minute" json:"requests_per_minute"` -} - -// API controls auth and pagination. +// API controls auth and pagination. Rate limiting lives under server +// (server.rate_limit_rpm/burst); there is no api.rate_limit key. type API struct { - AuthTokens []string `yaml:"auth_tokens" json:"auth_tokens"` - RateLimit RateLimitConfig `yaml:"rate_limit" json:"rate_limit"` - MaxPageSize int `yaml:"max_page_size" json:"max_page_size"` + AuthTokens []string `yaml:"auth_tokens" json:"auth_tokens"` + MaxPageSize int `yaml:"max_page_size" json:"max_page_size"` } // Probes holds fleet-wide probe defaults. @@ -468,13 +459,8 @@ func Default() *Config { cfg.Server.ReadTimeout = Duration(10 * time.Second) cfg.Server.WriteTimeout = Duration(10 * time.Second) cfg.Server.StatusPage.Enabled = &t - cfg.Storage.Path = "./epmon.db" cfg.Storage.BusyTimeoutMs = 5000 - cfg.Storage.WriteTimeout = Duration(5 * time.Second) - cfg.Storage.Retention.ChecksDays = 90 - cfg.Storage.Retention.RollupsDays = 730 cfg.History.Timezone = "UTC" - cfg.API.RateLimit.RequestsPerMinute = 120 cfg.API.MaxPageSize = 100 cfg.Probes.DefaultInterval = Duration(60 * time.Second) cfg.Probes.DefaultTimeout = Duration(10 * time.Second) @@ -550,27 +536,12 @@ func (c *Config) applyDefaults() error { if c.Server.StatusPage.Enabled == nil { c.Server.StatusPage.Enabled = boolPtr(true) } - if c.Storage.Path == "" { - c.Storage.Path = "./epmon.db" - } if c.Storage.BusyTimeoutMs == 0 { c.Storage.BusyTimeoutMs = 5000 } - if c.Storage.WriteTimeout.Std() == 0 { - c.Storage.WriteTimeout = Duration(5 * time.Second) - } - if c.Storage.Retention.ChecksDays == 0 { - c.Storage.Retention.ChecksDays = 90 - } - if c.Storage.Retention.RollupsDays == 0 { - c.Storage.Retention.RollupsDays = 730 - } if c.History.Timezone == "" { c.History.Timezone = "UTC" } - if c.API.RateLimit.RequestsPerMinute == 0 { - c.API.RateLimit.RequestsPerMinute = 120 - } if c.API.MaxPageSize == 0 { c.API.MaxPageSize = 100 } @@ -666,6 +637,9 @@ func (c *Config) applyDefaults() error { if s.FailureThreshold == 0 { s.FailureThreshold = c.Probes.FailureThreshold } + if s.MaxBodyBytes == 0 { + s.MaxBodyBytes = c.Probes.MaxBodyBytes + } if s.Name == "" { s.Name = s.ID } @@ -682,6 +656,25 @@ func (c *Config) applyDefaults() error { warnOnce("insecure-skip-verify:" + s.ID) } } + // Reserved knobs: validated but not yet honored by the runtime. Warn + // when explicitly set so operators don't silently tune dead settings. + // (applyDefaults already filled defaults above, so any non-default + // value here came from the file.) + if c.Logging.Level != "info" || c.Logging.Format != "json" { + warnOncef("unimplemented:logging", "logging.level/format have no effect yet (log output is fixed)") + } + if c.Probes.AutoIncidents != nil && !*c.Probes.AutoIncidents { + warnOncef("unimplemented:auto-incidents", "probes.auto_incidents=false has no effect yet (no automatic incidents exist)") + } + if c.Incidents.AutoResolve != nil && !*c.Incidents.AutoResolve { + warnOncef("unimplemented:auto-resolve", "incidents.auto_resolve=false has no effect yet") + } + if c.Server.StatusPage.Enabled != nil && !*c.Server.StatusPage.Enabled { + warnOncef("unimplemented:status-page", "server.status_page.enabled=false has no effect yet") + } + if c.Server.Metrics.RequireAuth { + warnOncef("unimplemented:metrics-auth", "server.metrics.require_auth=true has no effect yet (/metrics stays public)") + } return nil } @@ -728,24 +721,12 @@ func (c *Config) Validate() error { if c.Storage.BusyTimeoutMs < 100 || c.Storage.BusyTimeoutMs > 60000 { return fmt.Errorf("storage.busy_timeout_ms must be 100..60000") } - if c.Storage.WriteTimeout.Std() < time.Second || c.Storage.WriteTimeout.Std() > 30*time.Second { - return fmt.Errorf("storage.write_timeout must be 1s..30s") - } - if c.Storage.Retention.ChecksDays < 1 { - return fmt.Errorf("storage.retention.checks_days must be >= 1") - } - if c.Storage.Retention.RollupsDays < c.Storage.Retention.ChecksDays { - return fmt.Errorf("storage.retention.rollups_days must be >= checks_days") - } if _, err := time.LoadLocation(c.History.Timezone); err != nil { return fmt.Errorf("history.timezone: unknown IANA name %q", c.History.Timezone) } if c.API.MaxPageSize < 10 || c.API.MaxPageSize > 1000 { return fmt.Errorf("api.max_page_size must be 10..1000") } - if c.API.RateLimit.RequestsPerMinute < 1 { - return fmt.Errorf("api.rate_limit.requests_per_minute must be >= 1") - } if c.Database.RetentionDays < 1 { return fmt.Errorf("database.retention_days must be >= 1") } diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 97b0128..5890f1b 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -150,6 +150,47 @@ services: } } +func TestRemovedKnobsRejected(t *testing.T) { + // storage.path / storage.retention.* / api.rate_limit were validated + // but never read; they are gone, and the strict loader must say so + // loudly instead of accepting dead settings. + for name, body := range map[string]string{ + "storage.path": "storage: {path: other.db}\nservices:\n - {id: a, url: https://example.com}", + "storage.retention": "storage: {retention: {checks_days: 7}}\nservices:\n - {id: a, url: https://example.com}", + "api.rate_limit": "api: {rate_limit: {requests_per_minute: 10}}\nservices:\n - {id: a, url: https://example.com}", + } { + t.Run(name, func(t *testing.T) { + path := writeTemp(t, "config.yaml", body) + if _, err := Load(path); err == nil { + t.Errorf("expected unknown-field error for removed %s", name) + } + }) + } +} + +func TestResolvedAndReservedKnobs(t *testing.T) { + // probes.max_body_bytes resolves onto every service; reserved knobs + // (logging, auto_*, status_page, metrics auth) still load fine while + // warning on stderr instead of silently doing nothing. + path := writeTemp(t, "config.yaml", ` +probes: + max_body_bytes: 1024 +logging: + level: debug +server: + status_page: {enabled: false} +services: + - {id: a, url: https://example.com} +`) + cfg, err := Load(path) + if err != nil { + t.Fatalf("Load: %v", err) + } + if cfg.Services[0].MaxBodyBytes != 1024 { + t.Errorf("service MaxBodyBytes = %d, want 1024 from probes.max_body_bytes", cfg.Services[0].MaxBodyBytes) + } +} + func TestServerValidation(t *testing.T) { cases := map[string]string{ "tls half-set": "server: {tls_cert: /c.pem}\nservices:\n - {id: a, url: https://example.com}", diff --git a/internal/prober/prober.go b/internal/prober/prober.go index 5b1578d..a39d8f0 100644 --- a/internal/prober/prober.go +++ b/internal/prober/prober.go @@ -15,9 +15,19 @@ import ( "github.com/epmon-dev/epmon/internal/store" ) -// maxBody is the cap for body_contains inspection (4 MiB). +// maxBody is the backstop cap for body_contains inspection (4 MiB), +// used only when the service carries no resolved probes.max_body_bytes +// (e.g. programmatically built services that skipped config load). const maxBody = 4 << 20 +// bodyCap resolves the inspection cap for one service. +func bodyCap(svc config.Service) int64 { + if svc.MaxBodyBytes > 0 { + return svc.MaxBodyBytes + } + return maxBody +} + // clientKey distinguishes the transport configurations probes need. // Redirect policy joins the key so follow_redirects=false (#4) keeps its // own client instead of inheriting another service's policy. @@ -103,7 +113,7 @@ func Probe(ctx context.Context, svc config.Service) store.Check { return check } if svc.BodyContains != "" { - body, err := io.ReadAll(io.LimitReader(res.Body, maxBody)) + body, err := io.ReadAll(io.LimitReader(res.Body, bodyCap(svc))) if err != nil { check.Error = "body: " + shortErr(err) return check diff --git a/internal/prober/prober_test.go b/internal/prober/prober_test.go index 2d603f8..5c43c6a 100644 --- a/internal/prober/prober_test.go +++ b/internal/prober/prober_test.go @@ -5,6 +5,7 @@ import ( "net" "net/http" "net/http/httptest" + "strings" "sync/atomic" "testing" "time" @@ -139,6 +140,30 @@ func TestProbeReusesConnections(t *testing.T) { } } +// TestProbeBodyCap constructs a body whose needle sits past a 100-byte +// cap: the capped read misses it (down) while the 4MiB backstop would +// have matched (up), proving the service-resolved probes.max_body_bytes +// governs inspection. +func TestProbeBodyCap(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write(append([]byte(strings.Repeat("x", 200)), "needle"...)) + })) + defer srv.Close() + + s := svc(srv.URL) + s.BodyContains = "needle" + s.MaxBodyBytes = 100 + if c := Probe(t.Context(), s); c.Up { + t.Errorf("capped probe should miss the needle: %+v", c) + } + + uncapped := svc(srv.URL) + uncapped.BodyContains = "needle" + if c := Probe(t.Context(), uncapped); !c.Up { + t.Errorf("backstop probe should match the needle: %+v", c) + } +} + func TestProbeHeadersAndMethod(t *testing.T) { srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.Method != "HEAD" || r.Header.Get("X-Test") != "yes" { diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go index dd05cf5..f1a06ba 100644 --- a/internal/scheduler/scheduler.go +++ b/internal/scheduler/scheduler.go @@ -45,12 +45,19 @@ func randomPhase(interval time.Duration) time.Duration { // Run starts one loop per enabled service plus the daily purge. It returns immediately. func (s *Scheduler) Run(ctx context.Context) { + n := s.cfg.Probes.Concurrency + if n <= 0 { + n = 64 + } + // sem bounds concurrent probe executions across all loops so fleet + // size never sets the outbound concurrency by itself. + sem := make(chan struct{}, n) for _, svc := range s.cfg.Services { if !svc.EnabledOrDefault() { continue } s.wg.Add(1) - go s.loop(ctx, svc) + go s.loop(ctx, svc, sem) } s.wg.Add(1) go s.purgeLoop(ctx) @@ -59,7 +66,7 @@ func (s *Scheduler) Run(ctx context.Context) { // Stop waits for every loop to exit. func (s *Scheduler) Stop() { s.wg.Wait() } -func (s *Scheduler) loop(ctx context.Context, svc config.Service) { +func (s *Scheduler) loop(ctx context.Context, svc config.Service, sem chan struct{}) { defer s.wg.Done() logger := newDownLogger() // Consecutive-failure counting for failure_threshold: the raw probe is @@ -71,6 +78,12 @@ func (s *Scheduler) loop(ctx context.Context, svc config.Service) { } consecutive := 0 probe := func() { + select { + case sem <- struct{}{}: + defer func() { <-sem }() + case <-ctx.Done(): + return + } check := prober.Probe(ctx, svc) if err := s.checks.RecordCheck(ctx, check); err != nil { log.Printf("epmon: record %s: %v", svc.ID, err) diff --git a/internal/scheduler/scheduler_concurrency_test.go b/internal/scheduler/scheduler_concurrency_test.go new file mode 100644 index 0000000..58764ac --- /dev/null +++ b/internal/scheduler/scheduler_concurrency_test.go @@ -0,0 +1,98 @@ +package scheduler + +import ( + "context" + "net/http" + "net/http/httptest" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/epmon-dev/epmon/internal/config" + "github.com/epmon-dev/epmon/internal/store" +) + +type concurrencyRecorder struct { + mu sync.Mutex + n int +} + +func (r *concurrencyRecorder) RecordCheck(context.Context, store.Check) error { + r.mu.Lock() + defer r.mu.Unlock() + r.n++ + return nil +} + +func (r *concurrencyRecorder) Purge(context.Context, int, time.Time) (int64, error) { + return 0, nil +} + +type concurrencyObserver struct{} + +func (concurrencyObserver) ObserveCheck(string, bool, int64) {} +func (concurrencyObserver) ObserveProbe(string, bool, bool, time.Duration) {} +func (concurrencyObserver) ObserveSkipped(string) {} + +// TestSchedulerBoundsProbeConcurrency parks up to 5 service loops inside a +// blocking target and asserts no more than probes.concurrency=2 are ever +// inside a probe at once. The bound holds regardless of timing; the test +// only waits until contention is proven (active >= 3 attempted). +func TestSchedulerBoundsProbeConcurrency(t *testing.T) { + var active, maxSeen atomic.Int64 + release := make(chan struct{}) + target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + cur := active.Add(1) + for { + m := maxSeen.Load() + if cur <= m || maxSeen.CompareAndSwap(m, cur) { + break + } + } + select { + case <-release: + case <-time.After(10 * time.Second): + } + active.Add(-1) + })) + defer target.Close() + + spec := config.StatusSpec{} + if err := spec.UnmarshalJSON([]byte("[200]")); err != nil { + t.Fatal(err) + } + services := make([]config.Service, 0, 5) + for _, id := range []string{"a", "b", "c", "d", "e"} { + services = append(services, config.Service{ + ID: id, Name: id, URL: target.URL, Method: "GET", + Interval: config.Duration(10 * time.Millisecond), + Timeout: config.Duration(5 * time.Second), + ExpectStatus: spec, + }) + } + cfg := &config.Config{ + Services: services, + Probes: config.Probes{Concurrency: 2}, + } + s := New(cfg, &concurrencyRecorder{}, concurrencyObserver{}) + + ctx, cancel := context.WithCancel(context.Background()) + s.Run(ctx) + deadline := time.Now().Add(5 * time.Second) + for active.Load() < 2 && maxSeen.Load() < 2 && time.Now().Before(deadline) { + time.Sleep(5 * time.Millisecond) + } + // Give the remaining loops a chance to pile onto the semaphore. + time.Sleep(100 * time.Millisecond) + cancel() + close(release) + s.Stop() + + if got := maxSeen.Load(); got < 2 { + t.Fatalf("max concurrent probes = %d, contention never materialized", got) + } + if got := maxSeen.Load(); got > 2 { + t.Errorf("max concurrent probes = %d, want <= probes.concurrency=2", got) + } +} diff --git a/internal/store/sqlite/sqlite.go b/internal/store/sqlite/sqlite.go index f57d7e3..a23dfb8 100644 --- a/internal/store/sqlite/sqlite.go +++ b/internal/store/sqlite/sqlite.go @@ -10,6 +10,7 @@ import ( "context" "database/sql" "fmt" + "strconv" "strings" "time" "unicode/utf8" @@ -73,6 +74,21 @@ type Store struct { db *sql.DB } +// WithBusyTimeout returns dsn with a busy_timeout pragma applied, so +// concurrent writers wait up to ms milliseconds on locked pages instead +// of failing fast with "database is locked". An explicit busy_timeout +// already present in dsn is left untouched. +func WithBusyTimeout(dsn string, ms int) string { + if strings.Contains(dsn, "busy_timeout") { + return dsn + } + sep := "?" + if strings.Contains(dsn, "?") { + sep = "&" + } + return dsn + sep + "_pragma=busy_timeout(" + strconv.Itoa(ms) + ")" +} + // Open creates the file if needed and applies the schema. func Open(path string) (*Store, error) { db, err := sql.Open("sqlite", path) diff --git a/internal/store/sqlite/sqlite_test.go b/internal/store/sqlite/sqlite_test.go index d9b2f48..c26bc3a 100644 --- a/internal/store/sqlite/sqlite_test.go +++ b/internal/store/sqlite/sqlite_test.go @@ -433,6 +433,30 @@ func TestDailyHistoryTimezone(t *testing.T) { } } +func TestWithBusyTimeout(t *testing.T) { + for _, tc := range []struct { + dsn, want string + }{ + {"epmon.db", "epmon.db?_pragma=busy_timeout(5000)"}, + {"file:data.db?mode=ro", "file:data.db?mode=ro&_pragma=busy_timeout(5000)"}, + {"epmon.db?_pragma=busy_timeout(100)", "epmon.db?_pragma=busy_timeout(100)"}, + } { + if got := WithBusyTimeout(tc.dsn, 5000); got != tc.want { + t.Errorf("WithBusyTimeout(%q) = %q, want %q", tc.dsn, got, tc.want) + } + } + // The pragma DSN actually applies: open through it and read back. + st, err := Open(WithBusyTimeout(filepath.Join(t.TempDir(), "pragma.db"), 5000)) + if err != nil { + t.Fatal(err) + } + defer st.Close() + var v int + if err := st.db.QueryRow("PRAGMA busy_timeout").Scan(&v); err != nil || v != 5000 { + t.Errorf("busy_timeout = %d, %v; want 5000", v, err) + } +} + func TestSyncServices(t *testing.T) { st := openTest(t) ctx := t.Context()