feat(agent): agent-run monitor scheduler

2026-07-21 14:19:59 +01:00
parent c3613c6c0a
commit 8b56911db6
3 changed files with 391 additions and 0 deletions
@@ -0,0 +1,165 @@
+// Package monitors runs agent-side service checks. It polls the server for the
+// monitors assigned to this agent (SyncMonitors), runs each on its own interval
+// using the local checker package, and reports results back (ReportChecks).
+package monitors
+
+import (
+	"context"
+	"log"
+	"sync"
+	"time"
+
+	"github.com/mrhid6/vantage/agent/internal/checker"
+	"github.com/mrhid6/vantage/agent/internal/config"
+	grpcclient "github.com/mrhid6/vantage/agent/internal/grpc"
+	"github.com/mrhid6/vantage/agent/internal/grpc/pb"
+)
+
+// syncInterval controls how often the agent re-fetches its assigned monitors.
+const syncInterval = 30 * time.Second
+
+type runner struct {
+	intervalSec int
+	cancel      context.CancelFunc
+}
+
+// Run starts the agent monitor loop and blocks until ctx is cancelled.
+func Run(ctx context.Context, cfg *config.Config) {
+	active := map[string]*runner{}
+	var mu sync.Mutex
+
+	// results is a shared channel every check writes to; a single reporter
+	// goroutine batches and ships them so we make one ReportChecks call per tick.
+	results := make(chan pb.CheckResult, 64)
+	go reporter(ctx, cfg, results)
+
+	syncOnce := func() {
+		specs, err := fetchSpecs(cfg)
+		if err != nil {
+			log.Printf("monitors: sync: %v", err)
+			return
+		}
+		want := map[string]pb.MonitorSpec{}
+		for _, s := range specs {
+			want[s.MonitorId] = s
+		}
+
+		mu.Lock()
+		defer mu.Unlock()
+		for id, r := range active {
+			s, ok := want[id]
+			if !ok || s.IntervalSec != r.intervalSec {
+				r.cancel()
+				delete(active, id)
+			}
+		}
+		for id, s := range want {
+			if _, ok := active[id]; ok {
+				continue
+			}
+			rctx, cancel := context.WithCancel(ctx)
+			active[id] = &runner{intervalSec: s.IntervalSec, cancel: cancel}
+			go runSpec(rctx, s, results)
+		}
+	}
+
+	syncOnce()
+	t := time.NewTicker(syncInterval)
+	defer t.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-t.C:
+			syncOnce()
+		}
+	}
+}
+
+func fetchSpecs(cfg *config.Config) ([]pb.MonitorSpec, error) {
+	client, err := grpcclient.New(cfg.ServerURL, cfg.TLS)
+	if err != nil {
+		return nil, err
+	}
+	defer client.Close()
+	return client.SyncMonitors(cfg.ServerID, cfg.AgentToken)
+}
+
+func runSpec(ctx context.Context, s pb.MonitorSpec, out chan<- pb.CheckResult) {
+	interval := time.Duration(s.IntervalSec) * time.Second
+	if interval <= 0 {
+		interval = 60 * time.Second
+	}
+	spec := checker.Spec{
+		Type:           s.Type,
+		URL:            s.URL,
+		Host:           s.Host,
+		Port:           s.Port,
+		Method:         s.Method,
+		ExpectedStatus: s.ExpectedStatus,
+		Keyword:        s.Keyword,
+		TLSWarnDays:    s.TLSWarnDays,
+		TimeoutSec:     s.IntervalSec,
+	}
+
+	run := func() {
+		res := checker.Run(ctx, spec)
+		cr := pb.CheckResult{MonitorId: s.MonitorId, Up: res.Up, LatencyMs: res.LatencyMs, Message: res.Message}
+		if res.CertExpiry != nil {
+			cr.CertExpiryUnix = res.CertExpiry.Unix()
+		}
+		select {
+		case out <- cr:
+		case <-ctx.Done():
+		}
+	}
+
+	run()
+	t := time.NewTicker(interval)
+	defer t.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-t.C:
+			run()
+		}
+	}
+}
+
+// reporter batches results on a short interval and ships each batch in one call.
+func reporter(ctx context.Context, cfg *config.Config, in <-chan pb.CheckResult) {
+	t := time.NewTicker(5 * time.Second)
+	defer t.Stop()
+	var batch []pb.CheckResult
+	flush := func() {
+		if len(batch) == 0 {
+			return
+		}
+		client, err := grpcclient.New(cfg.ServerURL, cfg.TLS)
+		if err != nil {
+			log.Printf("monitors: report dial: %v", err)
+			batch = nil
+			return
+		}
+		if err := client.ReportChecks(cfg.ServerID, cfg.AgentToken, batch); err != nil {
+			log.Printf("monitors: report: %v", err)
+		}
+		client.Close()
+		batch = nil
+	}
+	for {
+		select {
+		case <-ctx.Done():
+			flush()
+			return
+		case r := <-in:
+			batch = append(batch, r)
+			if len(batch) >= 32 {
+				flush()
+			}
+		case <-t.C:
+			flush()
+		}
+	}
+}