77 lines
1.5 KiB
Go
77 lines
1.5 KiB
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)
|
|
}
|