feat(collectors) basic implementation of boring collectors; sender via http(s) to hard coded api route

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)
+}

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)
+}

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++
+}

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
+}

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
+}

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))
+}

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))
+}

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: "<uptime_seconds> <idle_seconds>"
+	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))
+}

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
+}

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
+	}
+}

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"`
+}

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
+}