bugswatch-agent/internal/agent/agent.go

package agent

import (
	"context"
	"log"
	"time"

	"bugs.watch/agent/internal/aggregate"
	"bugs.watch/agent/internal/collectors"
	"bugs.watch/agent/internal/collectors/cpu"
	"bugs.watch/agent/internal/collectors/disk"
	"bugs.watch/agent/internal/collectors/memory"
	"bugs.watch/agent/internal/collectors/uptime"
	"bugs.watch/agent/internal/config"
	"bugs.watch/agent/internal/dbmanager"
	"bugs.watch/agent/internal/sender"
)

const (
	sampleInterval = 5 * time.Second
	windowDuration = 60 * time.Second
)

func Run(ctx context.Context, cfg config.Config) {
	log.Println("bugswatch agent starting")
	log.Printf("db config: %s", cfg.DBConfigPath)

	window := aggregate.NewWindow()

	cols := []collectors.Collector{
		uptime.New(),
		cpu.New(),
		memory.New(),
		disk.New(),
	}

	mgr := dbmanager.New(cfg.DBConfigPath)
	go mgr.Watch(ctx)

	snd := sender.NewHTTPSender(cfg)

	ticker := time.NewTicker(sampleInterval)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			log.Println("bugswatch agent shutting down")
			shutdownCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
			defer cancel()
			emitSnapshot(shutdownCtx, snd, window, mgr)
			return

		case <-ticker.C:
			for _, c := range cols {
				c.Collect(window)
			}
			mgr.Collect(window)

			if window.Ready(windowDuration) {
				emitSnapshot(ctx, snd, window, mgr)
				window = aggregate.NewWindow()
			}
		}
	}
}

func emitSnapshot(
	ctx context.Context,
	snd sender.Sender,
	window *aggregate.Window,
	mgr *dbmanager.Manager,
) {
	snap := window.Snapshot()
	_ = snd.Send(ctx, snap)
}