diff --git a/cmd/agent/main.go b/cmd/agent/main.go new file mode 100644 index 0000000..de37e70 --- /dev/null +++ b/cmd/agent/main.go @@ -0,0 +1,25 @@ +package main + +import ( + "context" + "os" + "os/signal" + "syscall" + + "bugs.watch/agent/internal/agent" +) + +func main() { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + sigCh := make(chan os.Signal, 1) + signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM) + + go func() { + <-sigCh + cancel() + }() + + agent.Run(ctx) +} \ No newline at end of file diff --git a/internal/agent/agent.go b/internal/agent/agent.go new file mode 100644 index 0000000..761e49b --- /dev/null +++ b/internal/agent/agent.go @@ -0,0 +1,74 @@ +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/sender" +) + +const ( + sampleInterval = 5 * time.Second + windowDuration = 60 * time.Second +) + +func Run(ctx context.Context) { + log.Println("bugswatch agent starting") + + // Aggregation window + window := aggregate.NewWindow() + + // Register collectors + cols := []collectors.Collector{ + uptime.New(), + cpu.New(), + memory.New(), + disk.New(), + } + + // Sender (HTTP or logging) + snd := sender.NewHTTPSender("http://172.17.0.1:5173/api/health") + + ticker := time.NewTicker(sampleInterval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + log.Println("bugswatch agent shutting down") + + // Best-effort final flush with bounded timeout + shutdownCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + emitSnapshot(shutdownCtx, snd, window) + return + + case <-ticker.C: + for _, c := range cols { + c.Collect(window) + } + + if window.Ready(windowDuration) { + emitSnapshot(ctx, snd, window) + window = aggregate.NewWindow() + } + } + } +} + +func emitSnapshot( + ctx context.Context, + snd sender.Sender, + window *aggregate.Window, +) { + snap := window.Snapshot() + _ = snd.Send(ctx, snap) +} diff --git a/internal/aggregate/window.go b/internal/aggregate/window.go new file mode 100644 index 0000000..b111634 --- /dev/null +++ b/internal/aggregate/window.go @@ -0,0 +1,82 @@ +package aggregate + +import "time" + +type Window struct { + start time.Time + + cpuSum float64 + cpuCount uint64 + + memSum float64 + memCount uint64 + + diskSum float64 + diskCount uint64 + + uptime uint64 +} + +type Snapshot struct { + Timestamp time.Time + + CPUAvg float64 + MemAvg float64 + DiskAvg float64 + + Uptime uint64 +} + +func NewWindow() *Window { + return &Window{ + start: time.Now(), + } +} + +func (w *Window) AddCPU(value float64) { + w.cpuSum += value + w.cpuCount++ +} + +func (w *Window) AddMemory(value float64) { + w.memSum += value + w.memCount++ +} + +func (w *Window) SetUptime(value uint64) { + w.uptime = value +} + +func (w *Window) Ready(windowDuration time.Duration) bool { + return time.Since(w.start) >= windowDuration +} + +func (w *Window) Snapshot() Snapshot { + var cpuAvg float64 + if w.cpuCount > 0 { + cpuAvg = w.cpuSum / float64(w.cpuCount) + } + + var memAvg float64 + if w.memCount > 0 { + memAvg = w.memSum / float64(w.memCount) + } + + var diskAvg float64 + if w.diskCount > 0 { + diskAvg = w.diskSum / float64(w.diskCount) + } + + return Snapshot{ + Timestamp: time.Now(), + CPUAvg: cpuAvg, + MemAvg: memAvg, + DiskAvg: diskAvg, + Uptime: w.uptime, + } +} + +func (w *Window) AddDisk(value float64) { + w.diskSum += value + w.diskCount++ +} \ No newline at end of file diff --git a/internal/collectors/collector.go b/internal/collectors/collector.go new file mode 100644 index 0000000..5cfe89c --- /dev/null +++ b/internal/collectors/collector.go @@ -0,0 +1,9 @@ +package collectors + +import "bugs.watch/agent/internal/aggregate" + +// Collector collects one kind of system signal. +type Collector interface { + Collect(window *aggregate.Window) + Name() string +} diff --git a/internal/collectors/cpu/cpu.go b/internal/collectors/cpu/cpu.go new file mode 100644 index 0000000..5d27a5b --- /dev/null +++ b/internal/collectors/cpu/cpu.go @@ -0,0 +1,105 @@ +package cpu + +import ( + "bufio" + "os" + "strconv" + "strings" + + "bugs.watch/agent/internal/aggregate" +) + +type Sample struct { + Idle uint64 + Total uint64 +} + +// ReadSample reads CPU counters from /proc/stat. +// It returns cumulative counters since boot. +func ReadSample() (Sample, error) { + f, err := os.Open("/proc/stat") + if err != nil { + return Sample{}, err + } + defer f.Close() + + scanner := bufio.NewScanner(f) + if !scanner.Scan() { + if err := scanner.Err(); err != nil { + return Sample{}, err + } + return Sample{}, strconv.ErrSyntax + } + + // First line: + // cpu user nice system idle iowait irq softirq steal guest guest_nice + fields := strings.Fields(scanner.Text()) + if len(fields) < 5 || fields[0] != "cpu" { + return Sample{}, strconv.ErrSyntax + } + + var total uint64 + for i := 1; i < len(fields); i++ { + v, err := strconv.ParseUint(fields[i], 10, 64) + if err != nil { + return Sample{}, err + } + total += v + } + + idle, err := strconv.ParseUint(fields[4], 10, 64) + if err != nil { + return Sample{}, err + } + + return Sample{Idle: idle, Total: total}, nil +} + +// UsagePercent computes CPU usage percentage between two samples. +func UsagePercent(prev, curr Sample) float64 { + // Guard against counter resets or weirdness + if curr.Total <= prev.Total || curr.Idle < prev.Idle { + return 0 + } + + idleDelta := curr.Idle - prev.Idle + totalDelta := curr.Total - prev.Total + if totalDelta == 0 { + return 0 + } + + usedDelta := totalDelta - idleDelta + return (float64(usedDelta) / float64(totalDelta)) * 100.0 +} + +// Collector implements a stateful CPU usage collector. +// It stores the previous sample internally so the agent does not have to. +type Collector struct { + prev Sample + hasPrev bool +} + +func New() *Collector { + return &Collector{} +} + +func (c *Collector) Name() string { + return "cpu" +} + +// Collect reads current CPU counters and, if possible, emits a usage percent. +// On first run it will not emit (no previous sample yet). +func (c *Collector) Collect(window *aggregate.Window) { + curr, err := ReadSample() + if err != nil { + return + } + + if c.hasPrev { + usage := UsagePercent(c.prev, curr) + window.AddCPU(usage) + } + + c.prev = curr + c.hasPrev = true +} diff --git a/internal/collectors/disk/disk.go b/internal/collectors/disk/disk.go new file mode 100644 index 0000000..adfd838 --- /dev/null +++ b/internal/collectors/disk/disk.go @@ -0,0 +1,58 @@ +package disk + +import ( + "syscall" + + "bugs.watch/agent/internal/aggregate" +) + +type Stats struct { + TotalBytes uint64 + UsedBytes uint64 +} + +// Read returns disk usage statistics for the root filesystem. +func Read() (Stats, error) { + var fs syscall.Statfs_t + + if err := syscall.Statfs("/", &fs); err != nil { + return Stats{}, err + } + + total := fs.Blocks * uint64(fs.Bsize) + free := fs.Bavail * uint64(fs.Bsize) + used := total - free + + return Stats{ + TotalBytes: total, + UsedBytes: used, + }, nil +} + +// UsedPercent returns disk usage as a percentage. +func UsedPercent(stats Stats) float64 { + if stats.TotalBytes == 0 { + return 0 + } + return (float64(stats.UsedBytes) / float64(stats.TotalBytes)) * 100.0 +} + +// Collector implements the collectors.Collector interface for disk usage. +type Collector struct{} + +func New() *Collector { + return &Collector{} +} + +func (c *Collector) Name() string { + return "disk" +} + +func (c *Collector) Collect(window *aggregate.Window) { + stats, err := Read() + if err != nil { + return + } + + window.AddDisk(UsedPercent(stats)) +} diff --git a/internal/collectors/memory/memory.go b/internal/collectors/memory/memory.go new file mode 100644 index 0000000..0684e0d --- /dev/null +++ b/internal/collectors/memory/memory.go @@ -0,0 +1,92 @@ +package memory + +import ( + "bufio" + "os" + "strconv" + "strings" + + "bugs.watch/agent/internal/aggregate" +) + +type Stats struct { + TotalKB uint64 + AvailableKB uint64 +} + +// Read reads MemTotal and MemAvailable from /proc/meminfo. +// Values are in KB. +func Read() (Stats, error) { + f, err := os.Open("/proc/meminfo") + if err != nil { + return Stats{}, err + } + defer f.Close() + + var total uint64 + var available uint64 + + scanner := bufio.NewScanner(f) + for scanner.Scan() { + line := scanner.Text() + + if strings.HasPrefix(line, "MemTotal:") { + fields := strings.Fields(line) + if len(fields) >= 2 { + if v, err := strconv.ParseUint(fields[1], 10, 64); err == nil { + total = v + } + } + continue + } + + if strings.HasPrefix(line, "MemAvailable:") { + fields := strings.Fields(line) + if len(fields) >= 2 { + if v, err := strconv.ParseUint(fields[1], 10, 64); err == nil { + available = v + } + } + continue + } + } + + if err := scanner.Err(); err != nil { + return Stats{}, err + } + + return Stats{ + TotalKB: total, + AvailableKB: available, + }, nil +} + +// UsedPercent returns used memory percentage based on MemTotal and MemAvailable. +func UsedPercent(s Stats) float64 { + if s.TotalKB == 0 { + return 0 + } + + used := s.TotalKB - s.AvailableKB + return (float64(used) / float64(s.TotalKB)) * 100.0 +} + +// Collector implements a stateless memory usage collector. +type Collector struct{} + +func New() *Collector { + return &Collector{} +} + +func (c *Collector) Name() string { + return "memory" +} + +func (c *Collector) Collect(window *aggregate.Window) { + stats, err := Read() + if err != nil { + return + } + + window.AddMemory(UsedPercent(stats)) +} diff --git a/internal/collectors/uptime/uptime.go b/internal/collectors/uptime/uptime.go new file mode 100644 index 0000000..650d40f --- /dev/null +++ b/internal/collectors/uptime/uptime.go @@ -0,0 +1,41 @@ +package uptime + +import ( + "os" + "strconv" + "strings" + + "bugs.watch/agent/internal/aggregate" +) + +type Collector struct{} + +func New() *Collector { + return &Collector{} +} + +func (c *Collector) Name() string { + return "uptime" +} + +// Collect reads /proc/uptime and stores the latest uptime in seconds. +// Failure is silently ignored by design. +func (c *Collector) Collect(window *aggregate.Window) { + data, err := os.ReadFile("/proc/uptime") + if err != nil { + return + } + + // Format: " " + fields := strings.Fields(string(data)) + if len(fields) < 1 { + return + } + + seconds, err := strconv.ParseFloat(fields[0], 64) + if err != nil { + return + } + + window.SetUptime(uint64(seconds)) +} diff --git a/internal/sender/http.go b/internal/sender/http.go new file mode 100644 index 0000000..75a247f --- /dev/null +++ b/internal/sender/http.go @@ -0,0 +1,66 @@ +package sender + +import ( + "bytes" + "context" + "encoding/json" + "net/http" + "time" + + "bugs.watch/agent/internal/aggregate" +) + +type HTTPSender struct { + client *http.Client + endpoint string +} + +func NewHTTPSender(endpoint string) *HTTPSender { + return &HTTPSender{ + client: &http.Client{ + Timeout: 5 * time.Second, + }, + endpoint: endpoint, + } +} + +func (s *HTTPSender) Send(ctx context.Context, snap aggregate.Snapshot) error { + payload := SnapshotPayload{ + Timestamp: snap.Timestamp, + CPUAvg: snap.CPUAvg, + MemAvg: snap.MemAvg, + DiskAvg: snap.DiskAvg, + Uptime: snap.Uptime, + } + + body, err := json.Marshal(payload) + if err != nil { + return err + } + + req, err := http.NewRequestWithContext( + ctx, + http.MethodPost, + s.endpoint, + bytes.NewReader(body), + ) + if err != nil { + return err + } + + req.Header.Set("Content-Type", "application/json") + + resp, err := s.client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + + // We do not retry or branch yet. + // Any non-2xx is treated as a failure. + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return nil + } + + return nil +} diff --git a/internal/sender/logging.go b/internal/sender/logging.go new file mode 100644 index 0000000..6e19950 --- /dev/null +++ b/internal/sender/logging.go @@ -0,0 +1,33 @@ +package sender + +import ( + "context" + "log" + + "bugs.watch/agent/internal/aggregate" +) + +// LoggingSender sends snapshots to the local log. +// This is a stand-in for HTTP or disk-backed senders. +type LoggingSender struct{} + +func NewLoggingSender() *LoggingSender { + return &LoggingSender{} +} + +func (s *LoggingSender) Send(ctx context.Context, snap aggregate.Snapshot) error { + select { + case <-ctx.Done(): + // Context cancelled. Best effort: do not block shutdown. + return ctx.Err() + default: + log.Printf( + "SEND cpu=%.2f%% mem=%.2f%% disk=%.2f%% uptime=%ds", + snap.CPUAvg, + snap.MemAvg, + snap.DiskAvg, + snap.Uptime, + ) + return nil + } +} diff --git a/internal/sender/payload.go b/internal/sender/payload.go new file mode 100644 index 0000000..373bc29 --- /dev/null +++ b/internal/sender/payload.go @@ -0,0 +1,14 @@ +package sender + +import "time" + +// SnapshotPayload is the wire format sent to bugs.watch. +type SnapshotPayload struct { + Timestamp time.Time `json:"timestamp"` + + CPUAvg float64 `json:"cpu_avg"` + MemAvg float64 `json:"mem_avg"` + DiskAvg float64 `json:"disk_avg"` + + Uptime uint64 `json:"uptime"` +} diff --git a/internal/sender/sender.go b/internal/sender/sender.go new file mode 100644 index 0000000..c465374 --- /dev/null +++ b/internal/sender/sender.go @@ -0,0 +1,12 @@ +package sender + +import ( + "context" + + "bugs.watch/agent/internal/aggregate" +) + +// Sender sends snapshots somewhere. +type Sender interface { + Send(ctx context.Context, snap aggregate.Snapshot) error +}