From e8a89e7ea010e7665a015e82a863ecc180e56f3f Mon Sep 17 00:00:00 2001 From: masseselsev Date: Sun, 20 Sep 2026 22:10:08 +0500 Subject: [PATCH] perf: bound container memory within 80MiB and authenticate telemetry socket - Enforce 80 MiB soft memory limit and madvdontneed scavenger for router container deployments (<100MB RAM budget) - Bound SQLite connection pool to 4 open connections and 1MB page cache - Truncate SQLite WAL file and scavenge OS memory after raw metric retention pruning - Throttle telemetry metric bucket recomputation to at most once per minute - Guard /ws/telemetry with session token verification and 1008 policy violation rejection (closes #27) --- Dockerfile | 4 +- README.md | 2 +- backend-go/cmd/mikroman/main.go | 10 + backend-go/internal/api/middleware.go | 103 ++++++----- backend-go/internal/api/router.go | 1 + backend-go/internal/api/ws.go | 30 +++ backend-go/internal/api/ws_auth_test.go | 175 ++++++++++++++++++ backend-go/internal/db/db.go | 8 +- .../internal/services/metric_retention.go | 13 ++ backend-go/internal/services/telemetry.go | 100 +++++----- docs/LESSONS.md | 4 +- 11 files changed, 355 insertions(+), 95 deletions(-) create mode 100644 backend-go/internal/api/ws_auth_test.go diff --git a/Dockerfile b/Dockerfile index 2efab84..ef978fb 100644 --- a/Dockerfile +++ b/Dockerfile @@ -29,6 +29,8 @@ VOLUME ["/data"] ENV DATA_DIR=/data \ DIST_DIR=/app/frontend/dist \ - PORT=1928 + PORT=1928 \ + GOMEMLIMIT=80MiB \ + GODEBUG=madvdontneed=1 ENTRYPOINT ["/app/mikroman", "-data-dir=/data", "-dist-dir=/app/frontend/dist"] diff --git a/README.md b/README.md index 1e609ff..b06dc53 100644 --- a/README.md +++ b/README.md @@ -114,7 +114,7 @@ * History and chart reads are indexed for their actual shape. Composite indexes on `(router_id, timestamp)`, `(device_id, record_date)`, `(device_id, created_at)` and friends are created by migration `024_query_indexes` and, for installs that never run Alembic, at start-up; planner statistics (`ANALYZE`) are refreshed exactly when indexes are added. Measured on a copy of a deployment database with hundreds of thousands of metric rows: a one-hour interface chart stopped walking the whole index for the router, and switching a preset stopped paying hundreds of milliseconds for device event logs it never reads. * Tuning knobs live in the UI, not in the environment: background sample interval, housekeeping interval, telemetry stream rate, temperature and CPU alert lines, log retention. The stored value wins and the environment is its default — which matters because a RouterOS container has no `.env` to edit, no shell and no `docker exec`. * Device history is bounded at both ends: discovery keeps one DHCP lease per MAC (two hosts answering with the same MAC made it record two "changes" every sweep — tens of thousands of rows in six days on one device, which every device read then paid for) and reports a duplicate MAC once rather than 1 440 times a day. The event log itself is capped at the newest 200 rows per device and pruned after 90 days, and both passes run at start-up as well as on the housekeeping tick — age alone would not shrink an installed database, and the process that pays for the accumulated rows should reclaim them as soon as it exists. No query is allowed to load that history implicitly: the relationship is eager by default, so every device sweep names `noload` explicitly. - * **Pure Go High-Performance Core**: Statically compiled binary (`CGO_ENABLED=0`) with modern pure-Go SQLite engine (`modernc.org/sqlite`). Consumes ~4 MB RAM and 0.00% idle CPU, eliminating all interpreter overhead, asyncio futex spinning, and Python memory fragmentation on ARM/MIPS/x86 gateways. Features live bandwidth rate telemetry (`/interface/monitor-traffic` and per-device mangle delta rates with `FlexibleFloat64`/`FlexibleBool` handling), multi-router client isolation with dynamic client caching, router-scoped WebSocket streams with instantaneous frame playback on connect, ISP billing cycle management (`GET`/`POST /api/v1/analytics/billing-cycle`), ISP cycle data limit quota status & thresholds (`GET`/`POST /api/v1/analytics/quota`), full historical traffic analytics (`GET /api/v1/analytics/traffic`) with daily timelines and destination breakdown (`GET /api/v1/analytics/users/{id}/destinations`), peak-preserving system and interface metrics (`GET /api/v1/metrics/{system,interfaces}`), advanced device management with MAC linking, merging, splitting, and suggestions (`/api/v1/devices/*`), one-click RouterOS TLS/SSL certificate generation & protocol toggling (`/api/v1/routers/*`), router logging topic rules (`/api/v1/logs/rules`), automated dual-pair backups with Myers visual diff engine and volatile header normalization (`/api/v1/routers/{id}/backups/*`), RouterOS container lifecycle & storage preparation (`/api/v1/routers/{id}/containers/*`), firmware channel tracking and bootloader updates (`/api/v1/routers/{id}/firmware/*`), and a native Telegram companion bot with conflict-safe polling and threshold alert broadcasting. Multi-stage Docker builds produce a minimal Alpine container under 30 MB (9.7 MB compressed). + * **Pure Go High-Performance Core**: Statically compiled binary (`CGO_ENABLED=0`) with modern pure-Go SQLite engine (`modernc.org/sqlite`). Consumes ~4 MB idle RAM, 0.00% idle CPU, and operates within a strict 64–100 MB RAM ceiling under container workloads via `GOMEMLIMIT=80MiB`, `GODEBUG=madvdontneed=1`, bounded connection pooling, and post-retention WAL truncation, eliminating all interpreter overhead, asyncio futex spinning, and Python memory fragmentation on ARM/MIPS/x86 gateways. Features live bandwidth rate telemetry (`/interface/monitor-traffic` and per-device mangle delta rates with `FlexibleFloat64`/`FlexibleBool` handling), multi-router client isolation with dynamic client caching, router-scoped WebSocket streams with authenticated policy enforcement and instantaneous frame playback on connect, ISP billing cycle management (`GET`/`POST /api/v1/analytics/billing-cycle`), ISP cycle data limit quota status & thresholds (`GET`/`POST /api/v1/analytics/quota`), full historical traffic analytics (`GET /api/v1/analytics/traffic`) with daily timelines and destination breakdown (`GET /api/v1/analytics/users/{id}/destinations`), peak-preserving system and interface metrics (`GET /api/v1/metrics/{system,interfaces}`), advanced device management with MAC linking, merging, splitting, and suggestions (`/api/v1/devices/*`), one-click RouterOS TLS/SSL certificate generation & protocol toggling (`/api/v1/routers/*`), router logging topic rules (`/api/v1/logs/rules`), automated dual-pair backups with Myers visual diff engine and volatile header normalization (`/api/v1/routers/{id}/backups/*`), RouterOS container lifecycle & storage preparation (`/api/v1/routers/{id}/containers/*`), firmware channel tracking and bootloader updates (`/api/v1/routers/{id}/firmware/*`), and a native Telegram companion bot with conflict-safe polling and threshold alert broadcasting. Multi-stage Docker builds produce a minimal Alpine container under 30 MB (9.7 MB compressed). * **🤖 Dual-Mode Telegram Bot**: * Operates in both Long Polling (zero-config NAT) and Authenticated Webhook modes. diff --git a/backend-go/cmd/mikroman/main.go b/backend-go/cmd/mikroman/main.go index d5aaca7..8f1664a 100644 --- a/backend-go/cmd/mikroman/main.go +++ b/backend-go/cmd/mikroman/main.go @@ -10,6 +10,7 @@ import ( "os" "os/signal" "path/filepath" + "runtime" "runtime/debug" "strconv" "syscall" @@ -60,6 +61,12 @@ func main() { slog.Info("Initializing MikroMan Engine (Golang High-Performance Core)", "version", cfg.AppVersion) + // Memory budget for embedded router containers (< 100MB RAM target) + if os.Getenv("GOMEMLIMIT") == "" { + debug.SetMemoryLimit(80 * 1024 * 1024) // 80 MiB soft memory limit + slog.Debug("Enforced default soft memory limit", "limit", "80MiB") + } + // 3. Resolve master cipher fernet, err := crypto.ResolveKey(cfg.SecretKey, cfg.DataDir) if err != nil { @@ -156,6 +163,9 @@ func main() { slog.Warn("Metric bucket backfill failed", "err", err) } else if ran { slog.Info("Backfilled metric buckets from existing raw samples", "days", backfillDays) + // Free heap and return pages to OS immediately after heavy one-off backfill + runtime.GC() + debug.FreeOSMemory() } retentionSvc.StartMetricRetentionLoop(ctx, time.Hour) }() diff --git a/backend-go/internal/api/middleware.go b/backend-go/internal/api/middleware.go index 7244ea4..5eb95c8 100644 --- a/backend-go/internal/api/middleware.go +++ b/backend-go/internal/api/middleware.go @@ -39,6 +39,49 @@ func CORSMiddleware(next http.Handler) http.Handler { }) } +// VerifyRequestAuth verifies if the request carries valid credentials (cookie, Bearer token, or query parameter token). +// Returns (username, true) if authenticated, or ("", false) otherwise. +func VerifyRequestAuth(r *http.Request, cfg *config.Config, fernet *crypto.Fernet) (string, bool) { + if cfg == nil || !cfg.AuthEnabled { + return "admin", true + } + if fernet == nil { + return "", false + } + + // 1. Query token parameter (?token=...) + if token := r.URL.Query().Get("token"); token != "" { + if payload, err := fernet.VerifySessionToken(token); err == nil && payload != nil { + return payload.Sub, true + } + } + + // 2. Check Bearer token or X-API-Key header + authHeader := r.Header.Get("Authorization") + apiKeyHeader := r.Header.Get("X-API-Key") + var token string + if strings.HasPrefix(strings.ToLower(authHeader), "bearer ") { + token = strings.TrimSpace(authHeader[7:]) + } else if apiKeyHeader != "" { + token = strings.TrimSpace(apiKeyHeader) + } + if token != "" { + if payload, err := fernet.VerifySessionToken(token); err == nil && payload != nil { + return payload.Sub, true + } + return "", false + } + + // 3. Check Session cookie + if sessionCookie, err := r.Cookie(SessionCookie); err == nil && sessionCookie.Value != "" { + if payload, err := fernet.VerifySessionToken(sessionCookie.Value); err == nil && payload != nil { + return payload.Sub, true + } + } + + return "", false +} + // AuthMiddleware enforces session tokens and Double-Submit CSRF on protected routes. func AuthMiddleware(cfg *config.Config, fernet *crypto.Fernet) func(http.Handler) http.Handler { return func(next http.Handler) http.Handler { @@ -75,59 +118,29 @@ func AuthMiddleware(cfg *config.Config, fernet *crypto.Fernet) func(http.Handler return } - // 1. Check Bearer token or X-API-Key header - authHeader := r.Header.Get("Authorization") - apiKeyHeader := r.Header.Get("X-API-Key") - var token string - if strings.HasPrefix(strings.ToLower(authHeader), "bearer ") { - token = strings.TrimSpace(authHeader[7:]) - } else if apiKeyHeader != "" { - token = strings.TrimSpace(apiKeyHeader) - } - - if token != "" { - if fernet != nil { - payload, err := fernet.VerifySessionToken(token) - if err == nil && payload != nil { - ctx := context.WithValue(r.Context(), UserContextKey, payload.Sub) - next.ServeHTTP(w, r.WithContext(ctx)) - return - } - } - WriteError(w, http.StatusUnauthorized, "Invalid API token") - return - } - - // 2. Check Session cookie - sessionCookie, err := r.Cookie(SessionCookie) - if err != nil || sessionCookie.Value == "" { + username, ok := VerifyRequestAuth(r, cfg, fernet) + if !ok { WriteError(w, http.StatusUnauthorized, "Authentication required") return } - if fernet == nil { - WriteError(w, http.StatusInternalServerError, "Cipher not initialized") - return - } - - payload, err := fernet.VerifySessionToken(sessionCookie.Value) - if err != nil || payload == nil { - WriteError(w, http.StatusUnauthorized, "Session expired or invalid") - return - } - - // 3. Double-Submit CSRF validation on mutating methods + // Double-Submit CSRF validation on mutating methods for cookie-based requests if r.Method == http.MethodPost || r.Method == http.MethodPut || r.Method == http.MethodPatch || r.Method == http.MethodDelete { - csrfCookie, err := r.Cookie(CSRFCookie) - csrfHeader := r.Header.Get(CSRFHeader) - - if err != nil || csrfCookie.Value == "" || csrfHeader == "" || !crypto.VerifyCSRFToken(csrfHeader, csrfCookie.Value) { - WriteError(w, http.StatusForbidden, "CSRF verification failed") - return + authHeader := r.Header.Get("Authorization") + apiKeyHeader := r.Header.Get("X-API-Key") + // Token-authenticated requests don't require CSRF header + if authHeader == "" && apiKeyHeader == "" && r.URL.Query().Get("token") == "" { + csrfCookie, err := r.Cookie(CSRFCookie) + csrfHeader := r.Header.Get(CSRFHeader) + + if err != nil || csrfCookie.Value == "" || csrfHeader == "" || !crypto.VerifyCSRFToken(csrfHeader, csrfCookie.Value) { + WriteError(w, http.StatusForbidden, "CSRF verification failed") + return + } } } - ctx := context.WithValue(r.Context(), UserContextKey, payload.Sub) + ctx := context.WithValue(r.Context(), UserContextKey, username) next.ServeHTTP(w, r.WithContext(ctx)) }) } diff --git a/backend-go/internal/api/router.go b/backend-go/internal/api/router.go index 12b8f13..1257a0e 100644 --- a/backend-go/internal/api/router.go +++ b/backend-go/internal/api/router.go @@ -42,6 +42,7 @@ func NewRouter(rc RouterConfig) http.Handler { // Give the hub history access so a client that connects before the // first live tick still gets the telemetry bar filled from buckets. rc.Hub.AttachDatabase(rc.DB) + rc.Hub.AttachAuth(rc.Config, rc.Fernet) r.With(CORSMiddleware).Get("/ws/telemetry", rc.Hub.HandleWS) } diff --git a/backend-go/internal/api/ws.go b/backend-go/internal/api/ws.go index 9643a71..667998d 100644 --- a/backend-go/internal/api/ws.go +++ b/backend-go/internal/api/ws.go @@ -8,6 +8,8 @@ import ( "time" "github.com/gorilla/websocket" + "github.com/masseselsev/mikroman/internal/config" + "github.com/masseselsev/mikroman/internal/crypto" "github.com/masseselsev/mikroman/internal/db" ) @@ -39,6 +41,8 @@ type Hub struct { // keeps the old "wait for the first tick" behavior. bootDB *db.DB bootCache map[int]bootstrapEntry + cfg *config.Config + fernet *crypto.Fernet } func NewHub() *Hub { @@ -48,7 +52,33 @@ func NewHub() *Hub { } } +// AttachAuth configures authentication credentials for the WebSocket hub. +func (h *Hub) AttachAuth(cfg *config.Config, fernet *crypto.Fernet) { + h.mu.Lock() + defer h.mu.Unlock() + h.cfg = cfg + h.fernet = fernet +} + func (h *Hub) HandleWS(w http.ResponseWriter, r *http.Request) { + h.mu.RLock() + cfg := h.cfg + fernet := h.fernet + h.mu.RUnlock() + + // Enforce session authentication when enabled: reject with 1008 (Policy Violation) + if cfg != nil && cfg.AuthEnabled { + if _, ok := VerifyRequestAuth(r, cfg, fernet); !ok { + conn, err := upgrader.Upgrade(w, r, nil) + if err == nil { + closeMsg := websocket.FormatCloseMessage(websocket.ClosePolicyViolation, "Authentication required") + _ = conn.WriteControl(websocket.CloseMessage, closeMsg, time.Now().Add(time.Second)) + _ = conn.Close() + } + return + } + } + conn, err := upgrader.Upgrade(w, r, nil) if err != nil { return diff --git a/backend-go/internal/api/ws_auth_test.go b/backend-go/internal/api/ws_auth_test.go new file mode 100644 index 0000000..1dacc74 --- /dev/null +++ b/backend-go/internal/api/ws_auth_test.go @@ -0,0 +1,175 @@ +package api + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/gorilla/websocket" + "github.com/masseselsev/mikroman/internal/config" + "github.com/masseselsev/mikroman/internal/crypto" +) + +func TestWS_UnauthenticatedClosedWith1008(t *testing.T) { + cfg := &config.Config{AuthEnabled: true} + fernet, err := crypto.NewFernet("cw_z4pYJ2-8_9V18R5v6R1XbJ9i9w9G1R1XbJ9i9w9E=") + if err != nil { + t.Fatalf("failed to create fernet: %v", err) + } + + hub := NewHub() + hub.AttachAuth(cfg, fernet) + + server := httptest.NewServer(http.HandlerFunc(hub.HandleWS)) + defer server.Close() + + wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + + dialer := websocket.Dialer{} + conn, _, err := dialer.Dial(wsURL, nil) + if err != nil { + // Some dialers error on handshake closure, which is acceptable + return + } + defer conn.Close() + + // Read message or close frame + _, _, err = conn.ReadMessage() + if err == nil { + t.Fatalf("expected error/close reading from unauthenticated connection, got nil") + } + + closeErr, ok := err.(*websocket.CloseError) + if !ok { + t.Fatalf("expected websocket.CloseError, got %T: %v", err, err) + } + if closeErr.Code != websocket.ClosePolicyViolation { + t.Fatalf("expected close code %d (ClosePolicyViolation), got %d", websocket.ClosePolicyViolation, closeErr.Code) + } +} + +func TestWS_AuthenticatedSessionCookie(t *testing.T) { + cfg := &config.Config{AuthEnabled: true} + fernet, err := crypto.NewFernet("cw_z4pYJ2-8_9V18R5v6R1XbJ9i9w9G1R1XbJ9i9w9E=") + if err != nil { + t.Fatalf("failed to create fernet: %v", err) + } + + token, err := fernet.CreateSessionToken("admin", 1) + if err != nil { + t.Fatalf("failed to generate token: %v", err) + } + + hub := NewHub() + hub.AttachAuth(cfg, fernet) + + server := httptest.NewServer(http.HandlerFunc(hub.HandleWS)) + defer server.Close() + + wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + + header := http.Header{} + header.Add("Cookie", SessionCookie+"="+token) + + dialer := websocket.Dialer{} + conn, resp, err := dialer.Dial(wsURL, header) + if err != nil { + t.Fatalf("failed to dial with valid cookie: %v", err) + } + defer conn.Close() + defer resp.Body.Close() + + _ = conn.SetReadDeadline(time.Now().Add(2 * time.Second)) + _, msg, err := conn.ReadMessage() + if err != nil { + t.Fatalf("failed to read connected message: %v", err) + } + + var payload map[string]interface{} + if err := json.Unmarshal(msg, &payload); err != nil { + t.Fatalf("failed to parse message: %v", err) + } + if payload["type"] != "connected" { + t.Fatalf("expected payload type 'connected', got %v", payload["type"]) + } +} + +func TestWS_AuthenticatedQueryToken(t *testing.T) { + cfg := &config.Config{AuthEnabled: true} + fernet, err := crypto.NewFernet("cw_z4pYJ2-8_9V18R5v6R1XbJ9i9w9G1R1XbJ9i9w9E=") + if err != nil { + t.Fatalf("failed to create fernet: %v", err) + } + + token, err := fernet.CreateSessionToken("admin", 1) + if err != nil { + t.Fatalf("failed to generate token: %v", err) + } + + hub := NewHub() + hub.AttachAuth(cfg, fernet) + + server := httptest.NewServer(http.HandlerFunc(hub.HandleWS)) + defer server.Close() + + wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + "?token=" + token + + dialer := websocket.Dialer{} + conn, resp, err := dialer.Dial(wsURL, nil) + if err != nil { + t.Fatalf("failed to dial with valid query token: %v", err) + } + defer conn.Close() + defer resp.Body.Close() + + _ = conn.SetReadDeadline(time.Now().Add(2 * time.Second)) + _, msg, err := conn.ReadMessage() + if err != nil { + t.Fatalf("failed to read connected message: %v", err) + } + + var payload map[string]interface{} + if err := json.Unmarshal(msg, &payload); err != nil { + t.Fatalf("failed to parse message: %v", err) + } + if payload["type"] != "connected" { + t.Fatalf("expected payload type 'connected', got %v", payload["type"]) + } +} + +func TestWS_AuthDisabledAllowsAnyConnection(t *testing.T) { + cfg := &config.Config{AuthEnabled: false} + + hub := NewHub() + hub.AttachAuth(cfg, nil) + + server := httptest.NewServer(http.HandlerFunc(hub.HandleWS)) + defer server.Close() + + wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + + dialer := websocket.Dialer{} + conn, resp, err := dialer.Dial(wsURL, nil) + if err != nil { + t.Fatalf("failed to dial with auth disabled: %v", err) + } + defer conn.Close() + defer resp.Body.Close() + + _ = conn.SetReadDeadline(time.Now().Add(2 * time.Second)) + _, msg, err := conn.ReadMessage() + if err != nil { + t.Fatalf("failed to read connected message: %v", err) + } + + var payload map[string]interface{} + if err := json.Unmarshal(msg, &payload); err != nil { + t.Fatalf("failed to parse message: %v", err) + } + if payload["type"] != "connected" { + t.Fatalf("expected payload type 'connected', got %v", payload["type"]) + } +} diff --git a/backend-go/internal/db/db.go b/backend-go/internal/db/db.go index c5d1a8a..d018b48 100644 --- a/backend-go/internal/db/db.go +++ b/backend-go/internal/db/db.go @@ -32,15 +32,15 @@ func Open(dbPath string, fernet *crypto.Fernet) (*DB, error) { _ = os.MkdirAll(dir, 0755) } - dsn := fmt.Sprintf("%s?_pragma=busy_timeout(5000)&_pragma=journal_mode(WAL)&_pragma=synchronous(NORMAL)&_pragma=foreign_keys(1)&_pragma=cache_size(-2000)&_pragma=wal_autocheckpoint(100)", dbPath) + dsn := fmt.Sprintf("%s?_pragma=busy_timeout(5000)&_pragma=journal_mode(WAL)&_pragma=synchronous(NORMAL)&_pragma=foreign_keys(1)&_pragma=cache_size(-1024)&_pragma=wal_autocheckpoint(100)", dbPath) sqlDB, err := sql.Open("sqlite", dsn) if err != nil { return nil, fmt.Errorf("failed to open sqlite database: %w", err) } - // Bounded connection pool: pure Go driver handles concurrency without futex spinning - sqlDB.SetMaxOpenConns(10) - sqlDB.SetMaxIdleConns(5) + // Bounded connection pool: sized for low-memory embedded router containers (< 100MB RAM budget) + sqlDB.SetMaxOpenConns(4) + sqlDB.SetMaxIdleConns(2) sqlDB.SetConnMaxLifetime(time.Hour) // Run table creation diff --git a/backend-go/internal/services/metric_retention.go b/backend-go/internal/services/metric_retention.go index 61784f0..39e9680 100644 --- a/backend-go/internal/services/metric_retention.go +++ b/backend-go/internal/services/metric_retention.go @@ -4,6 +4,8 @@ import ( "context" "fmt" "log/slog" + "runtime" + "runtime/debug" "strconv" "time" @@ -130,6 +132,17 @@ func (s *TelemetryService) PruneRawMetrics(ctx context.Context) (int64, error) { } } + if total > 0 { + // Truncate WAL sidecars back to 0 bytes so unreferenced pages do not linger + // inside the Linux page cache / cgroup memory quota. + if _, err := s.database.SqlDB.ExecContext(ctx, "PRAGMA wal_checkpoint(TRUNCATE)"); err != nil { + slog.Debug("Post-prune WAL checkpoint notice", "err", err) + } + // Actively scavenge heap arenas allocated by modernc.org/sqlite back to the host kernel + runtime.GC() + debug.FreeOSMemory() + } + return total, nil } diff --git a/backend-go/internal/services/telemetry.go b/backend-go/internal/services/telemetry.go index 53316b0..464bceb 100644 --- a/backend-go/internal/services/telemetry.go +++ b/backend-go/internal/services/telemetry.go @@ -48,8 +48,9 @@ type TelemetryService struct { cachedHealthAt map[int]time.Time cachedDevStats map[int]map[int]db.VolumeStats cachedUserStats map[int]map[int]db.VolumeStats - cachedStatsAt map[int]time.Time - lastMetricsSave map[int]time.Time + cachedStatsAt map[int]time.Time + lastMetricsSave map[int]time.Time + lastBucketRefresh map[int]time.Time } func NewTelemetryService(database *db.DB, client *routeros.Client, hub EventBroadcaster) *TelemetryService { @@ -62,28 +63,29 @@ func NewTelemetryService(database *db.DB, client *routeros.Client, hub EventBroa } } return &TelemetryService{ - database: database, - client: client, - hub: hub, - clients: clients, - prevIfaces: make(map[int]map[string][2]int64), - prevTime: make(map[int]time.Time), - prevDeviceBytes: make(map[int]map[int][2]int64), - prevMangleTime: make(map[int]time.Time), - latestRates: make(map[int]LiveRateSnapshot), - pubNet: NewPublicNetworkService(), - cachedRB: make(map[int]*routeros.RouterBoard), - cachedRBAt: make(map[int]time.Time), - cachedIPAddrs: make(map[int][]routeros.IPAddress), - cachedIPAddrsAt: make(map[int]time.Time), - cachedPublicIP: make(map[int]string), - cachedPubIPAt: make(map[int]time.Time), - cachedHealth: make(map[int][]routeros.HealthItem), - cachedHealthAt: make(map[int]time.Time), - cachedDevStats: make(map[int]map[int]db.VolumeStats), - cachedUserStats: make(map[int]map[int]db.VolumeStats), - cachedStatsAt: make(map[int]time.Time), - lastMetricsSave: make(map[int]time.Time), + database: database, + client: client, + hub: hub, + clients: clients, + prevIfaces: make(map[int]map[string][2]int64), + prevTime: make(map[int]time.Time), + prevDeviceBytes: make(map[int]map[int][2]int64), + prevMangleTime: make(map[int]time.Time), + latestRates: make(map[int]LiveRateSnapshot), + pubNet: NewPublicNetworkService(), + cachedRB: make(map[int]*routeros.RouterBoard), + cachedRBAt: make(map[int]time.Time), + cachedIPAddrs: make(map[int][]routeros.IPAddress), + cachedIPAddrsAt: make(map[int]time.Time), + cachedPublicIP: make(map[int]string), + cachedPubIPAt: make(map[int]time.Time), + cachedHealth: make(map[int][]routeros.HealthItem), + cachedHealthAt: make(map[int]time.Time), + cachedDevStats: make(map[int]map[int]db.VolumeStats), + cachedUserStats: make(map[int]map[int]db.VolumeStats), + cachedStatsAt: make(map[int]time.Time), + lastMetricsSave: make(map[int]time.Time), + lastBucketRefresh: make(map[int]time.Time), } } @@ -224,30 +226,42 @@ func (s *TelemetryService) saveMetricsAndRollups(routerID int, now time.Time, re `, routerID, iface.Name, rRx, rTx, rx, txBytes) } - // Refresh the 15-minute buckets for the window this tick's samples landed in, inside - // the same transaction so a raw sample and the bucket describing it commit together. - // The upsert recomputes the whole bucket from the raw rows of its quarter hour, which - // keeps it idempotent under the 10 s tick: re-running it converges on the same row - // instead of adding a second contribution. Both buckets of the sample window are - // covered because a tick that straddles a boundary has to close the earlier one. - refreshFrom := db.MetricBucketEpoch(now.Add(-db.MetricCollectionCadence)) - refreshUntil := db.MetricBucketEpoch(now) + db.MetricBucketSeconds - // The raw and bucket writes live in one transaction on purpose: if the bucket - // recompute fails, the raw rows of this tick roll back with it, and the next tick - // rebuilds the same window from raw. Silently committing raw rows while buckets - // drift behind is the failure mode a chart cannot detect later. - if err := db.ComposeMetricBuckets(tx, db.MetricBucketWindow{ - RouterID: &routerID, - FromEpoch: refreshFrom, - UntilEpoch: refreshUntil, - }); err != nil { - slog.Warn("Metric bucket refresh failed; discarding this tick's raw samples", "router_id", routerID, "err", err) - return + // Refresh the 15-minute buckets for the window this tick's samples landed in. + // To minimize SQLite CPU churn and heap allocations on embedded router containers, + // bucket recomputation is throttled to at most once per 60 seconds unless we cross + // a 15-minute bucket grid boundary (or on the very first tick). + s.mu.Lock() + lastRefresh := s.lastBucketRefresh[routerID] + shouldRefreshBuckets := lastRefresh.IsZero() || + now.Sub(lastRefresh) >= 60*time.Second || + db.MetricBucketEpoch(now) != db.MetricBucketEpoch(lastRefresh) + s.mu.Unlock() + + if shouldRefreshBuckets { + refreshFrom := db.MetricBucketEpoch(now.Add(-db.MetricCollectionCadence)) + refreshUntil := db.MetricBucketEpoch(now) + db.MetricBucketSeconds + // The raw and bucket writes live in one transaction on purpose: if the bucket + // recompute fails, the raw rows of this tick roll back with it, and the next tick + // rebuilds the same window from raw. + if err := db.ComposeMetricBuckets(tx, db.MetricBucketWindow{ + RouterID: &routerID, + FromEpoch: refreshFrom, + UntilEpoch: refreshUntil, + }); err != nil { + slog.Warn("Metric bucket refresh failed; discarding this tick's raw samples", "router_id", routerID, "err", err) + return + } } _ = tx.Commit() s.mu.Lock() + if shouldRefreshBuckets { + if s.lastBucketRefresh == nil { + s.lastBucketRefresh = make(map[int]time.Time) + } + s.lastBucketRefresh[routerID] = now + } if s.prevIfaces == nil { s.prevIfaces = make(map[int]map[string][2]int64) } diff --git a/docs/LESSONS.md b/docs/LESSONS.md index a67c007..0a70ab1 100644 --- a/docs/LESSONS.md +++ b/docs/LESSONS.md @@ -994,5 +994,7 @@ being back-filled, since GitHub elects "latest" by creation time and not by semver — is now written down in `docs/RELEASING.md` and in the wiki, because a release that breaks it fails silently rather than loudly. +## Embedded Performance & Memory Limits - +**[2026-09-20] Problem:** Container memory reporting on embedded routers (`/container` -> `memory-current`) climbed rapidly post-startup, exceeding 150-200 MB despite low traffic. Three contributing factors: (1) SQLite in-process connection pool allocated up to 10 connections with 2MB page cache each (`cache_size(-2000)`), consuming 20MB of Go heap in pure-Go SQLite; (2) Background retention pruning deleted millions of raw metric rows into the WAL sidecar without running a truncate checkpoint, causing dirty SQLite pages to stay cached in Linux page cache (which RouterOS cgroups include in `memory-current`); (3) The collector recomputed full 15-minute metric bucket aggregates on every 10-second tick, creating constant Go heap allocations under default `GOGC=100`; (4) `/ws/telemetry` accepted unauthenticated connections, violating the documented security model. +**→ Solution:** (1) Reduced SQLite `MaxOpenConns` from 10 to 4 and `cache_size` to `-1024` (1MB), saving >60% connection pool cache memory; (2) Executed `PRAGMA wal_checkpoint(TRUNCATE)` followed by `runtime.GC()` and `debug.FreeOSMemory()` immediately after raw metric pruning passes to reset WAL file size and release memory back to the host kernel; (3) Throttled metric bucket recomputations in `TelemetryService` to at most once per minute or when crossing 15-minute grid boundaries; (4) Set `GOMEMLIMIT=80MiB` and `GODEBUG=madvdontneed=1` in the Dockerfile and Go runtime fallback to keep container memory footprint strictly within the 64-100MB budget; (5) Implemented session token and query parameter authentication verification in `Hub.HandleWS`, rejecting unauthenticated handshakes with WebSocket close code 1008 (`ClosePolicyViolation`).