package monitor import ( "context" "crypto/tls" "fmt" "net/http" "regexp" "strings" "sync" "time" "gitea.lerkolabs.com/lerkolabs/uptop/internal/models" "gitea.lerkolabs.com/lerkolabs/uptop/internal/store" ) const ( maxLogEntries = 100 pollInterval = 5 * time.Second minCheckInterval = 5 minPushGrace = 60 * time.Second maintPruneInterval = 15 * time.Minute defaultMaintRetention = 7 * 24 * time.Hour dbWriteBuffer = 4096 dbPruneInterval = 10 * time.Minute ) type Engine struct { mu sync.RWMutex liveState map[int]models.Site logMu sync.RWMutex logStore []models.LogEntry activeMu sync.RWMutex isActive bool histMu sync.RWMutex histories map[int]*SiteHistory tokenIndex map[string]int // protected by mu probeResultsMu sync.RWMutex probeResults map[int]map[string]NodeResult aggStrategy AggregationStrategy alertHealthMu sync.RWMutex alertHealth map[int]AlertHealth recheckMu sync.RWMutex recheck map[int]chan struct{} maintCacheMu sync.RWMutex maintCache map[int]bool db store.Store insecureSkipVerify bool allowPrivateTargets bool maintRetention time.Duration strictClient *http.Client insecureClient *http.Client dbWrites chan dbWrite writerWG sync.WaitGroup checkerWG sync.WaitGroup cancel context.CancelFunc stopOnce sync.Once } func NewEngine(s store.Store) *Engine { return newEngine(s, false) } func NewEngineWithOpts(s store.Store, allowPrivateTargets bool) *Engine { return newEngine(s, allowPrivateTargets) } func newEngine(s store.Store, allowPrivateTargets bool) *Engine { dial := SafeDialContext(allowPrivateTargets) return &Engine{ liveState: make(map[int]models.Site), histories: make(map[int]*SiteHistory), tokenIndex: make(map[string]int), recheck: make(map[int]chan struct{}), probeResults: make(map[int]map[string]NodeResult), alertHealth: make(map[int]AlertHealth), aggStrategy: AggAnyDown, isActive: true, allowPrivateTargets: allowPrivateTargets, maintRetention: defaultMaintRetention, dbWrites: make(chan dbWrite, dbWriteBuffer), db: s, strictClient: &http.Client{ Transport: &http.Transport{ TLSClientConfig: &tls.Config{InsecureSkipVerify: false}, DialContext: dial, }, }, insecureClient: &http.Client{ Transport: &http.Transport{ TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, //nolint:gosec // intentional for IgnoreTLS sites DialContext: dial, }, }, } } // SetInsecureSkipVerify must be called before Start: the field is read by // checker goroutines without synchronization. func (e *Engine) SetInsecureSkipVerify(skip bool) { e.insecureSkipVerify = skip } // SetMaintRetention must be called before Start: the field is read by the // maintenance prune goroutine without synchronization. func (e *Engine) SetMaintRetention(d time.Duration) { e.maintRetention = d } var ansiRe = regexp.MustCompile(`\x1b\[[0-9;]*[a-zA-Z]`) func sanitizeLog(s string) string { s = ansiRe.ReplaceAllString(s, "") s = strings.ReplaceAll(s, "\n", "\\n") s = strings.ReplaceAll(s, "\r", "") return s } func fmtDurationShort(d time.Duration) string { if d < time.Minute { return fmt.Sprintf("%ds", int(d.Seconds())) } if d < time.Hour { return fmt.Sprintf("%dm", int(d.Minutes())) } if d < 24*time.Hour { return fmt.Sprintf("%dh %dm", int(d.Hours()), int(d.Minutes())%60) } return fmt.Sprintf("%dd %dh", int(d.Hours())/24, int(d.Hours())%24) } // appendLog adds a timestamped entry to the in-memory ring buffer and returns // it. It never touches the database, so it is safe to call from the db-write // drop/error path without recursing back through the write queue. func (e *Engine) appendLog(msg string) models.LogEntry { entry := models.LogEntry{ Message: sanitizeLog(msg), CreatedAt: time.Now(), } e.logMu.Lock() e.logStore = append([]models.LogEntry{entry}, e.logStore...) if len(e.logStore) > maxLogEntries { e.logStore = e.logStore[:maxLogEntries] } e.logMu.Unlock() return entry } func (e *Engine) AddLog(msg string) { entry := e.appendLog(msg) e.enqueueWrite(writeLog{message: entry.Message}) } // enqueueWrite hands a persistence task to the writer goroutine without // blocking the caller. If the queue is saturated the write is dropped and noted // in the in-memory log only (never re-enqueued, to avoid recursion via AddLog). func (e *Engine) enqueueWrite(w dbWrite) { select { case e.dbWrites <- w: default: e.appendLog(fmt.Sprintf("db write queue full, dropped %s", w.desc())) } } // dbWriter is the single goroutine that owns all writes. Serializing writes // through one path removes the fire-and-forget goroutine pile-up, surfaces // errors, and lets retention run on a timer instead of per-insert. It drains // any buffered writes on shutdown before returning. func (e *Engine) dbWriter(ctx context.Context) { defer e.writerWG.Done() pruneTicker := time.NewTicker(dbPruneInterval) defer pruneTicker.Stop() e.prune(ctx) for { select { case w := <-e.dbWrites: if err := w.exec(ctx, e.db); err != nil { e.appendLog(fmt.Sprintf("db %s write failed: %v", w.desc(), err)) } case <-pruneTicker.C: e.prune(ctx) case <-ctx.Done(): e.drainWrites() return } } } // drainWrites flushes everything still buffered, best-effort, at shutdown. func (e *Engine) drainWrites() { for { select { case w := <-e.dbWrites: if err := w.exec(context.Background(), e.db); err != nil { e.appendLog(fmt.Sprintf("db %s write failed (drain): %v", w.desc(), err)) } default: return } } } func (e *Engine) prune(ctx context.Context) { if err := e.db.PruneLogs(ctx); err != nil { e.appendLog(fmt.Sprintf("log prune failed: %v", err)) } if err := e.db.PruneCheckHistory(ctx); err != nil { e.appendLog(fmt.Sprintf("check-history prune failed: %v", err)) } if err := e.db.PruneStateChanges(ctx); err != nil { e.appendLog(fmt.Sprintf("state-change prune failed: %v", err)) } } // Stop signals the writer goroutine to drain and exit, then blocks until it // has. Call it before closing the store so no write races a closed DB. func (e *Engine) Stop() { e.stopOnce.Do(func() { if e.cancel != nil { e.cancel() } e.checkerWG.Wait() e.writerWG.Wait() e.drainWrites() }) } func (e *Engine) InitLogs() { entries, err := e.db.LoadLogs(context.Background(), maxLogEntries) if err != nil { return } if len(entries) == 0 { return } e.logMu.Lock() defer e.logMu.Unlock() e.logStore = entries } func (e *Engine) GetLogs() []models.LogEntry { e.logMu.RLock() defer e.logMu.RUnlock() logs := make([]models.LogEntry, len(e.logStore)) copy(logs, e.logStore) return logs } func (e *Engine) Start(ctx context.Context) { // e.cancel is invoked by Stop() to drain and halt the writer; gosec can't // trace the cross-method call, and cancelling the parent reaps this child // regardless, so the leak it warns about can't occur. ctx, e.cancel = context.WithCancel(ctx) //nolint:gosec // cancel is called in Stop() e.writerWG.Add(1) go e.dbWriter(ctx) e.checkerWG.Add(1) go func() { defer e.checkerWG.Done() for { select { case <-ctx.Done(): return default: } e.refreshMaintenanceCache(ctx) configs, err := e.db.GetSites(ctx) if err != nil { e.AddLog(fmt.Sprintf("Failed to load sites: %v", err)) select { case <-time.After(pollInterval): case <-ctx.Done(): return } continue } dbIDs := make(map[int]bool, len(configs)) for _, cfg := range configs { dbIDs[cfg.ID] = true e.mu.RLock() existing, exists := e.liveState[cfg.ID] e.mu.RUnlock() if !exists { e.mu.Lock() site := models.Site{SiteConfig: cfg, SiteState: models.SiteState{Status: models.StatusPending}} if h, ok := e.GetHistory(cfg.ID); ok && len(h.Statuses) > 0 { if h.Statuses[len(h.Statuses)-1] { site.Status = models.StatusUp } else { site.Status = models.StatusDown } if len(h.Latencies) > 0 { site.Latency = h.Latencies[len(h.Latencies)-1] } } e.liveState[cfg.ID] = site e.addToTokenIndex(site) e.mu.Unlock() e.checkerWG.Add(1) go func(id int) { defer e.checkerWG.Done() e.monitorRoutine(ctx, id) }(cfg.ID) } else if existing.SiteConfig != cfg { e.UpdateSiteConfig(cfg) } } e.mu.RLock() var vanished []int for id := range e.liveState { if !dbIDs[id] { vanished = append(vanished, id) } } e.mu.RUnlock() for _, id := range vanished { e.RemoveSite(id) e.AddLog(fmt.Sprintf("Monitor removed (no longer in DB): ID %d", id)) } select { case <-time.After(pollInterval): case <-ctx.Done(): return } } }() e.checkerWG.Add(1) go func() { defer e.checkerWG.Done() e.maintenancePruner(ctx) }() }