feat: Implement offline buffering for host and docker agents

- Add internal/buffer package with generic ring buffer
- Add buffering logic to host agent for failed reports
- Add buffering logic to docker agent for failed reports
- Add BufferCapacity configuration option
- Add integration tests for buffering logic

internal/buffer/buffer.go

@@ -0,0 +1,74 @@
+package buffer
+
+import (
+	"sync"
+)
+
+// Queue is a thread-safe ring buffer for storing failed reports.
+type Queue[T any] struct {
+	mu       sync.Mutex
+	data     []T
+	capacity int
+}
+
+// New creates a new Queue with the specified capacity.
+func New[T any](capacity int) *Queue[T] {
+	return &Queue[T]{
+		data:     make([]T, 0, capacity),
+		capacity: capacity,
+	}
+}
+
+// Push adds an item to the queue. If the queue is full, the oldest item is dropped.
+func (q *Queue[T]) Push(item T) {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
+	if len(q.data) >= q.capacity {
+		// Drop oldest (shift left)
+		q.data = q.data[1:]
+	}
+	q.data = append(q.data, item)
+}
+
+// Pop removes and returns the oldest item from the queue.
+// Returns zero value and false if empty.
+func (q *Queue[T]) Pop() (T, bool) {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
+	if len(q.data) == 0 {
+		var zero T
+		return zero, false
+	}
+
+	item := q.data[0]
+	q.data = q.data[1:]
+	return item, true
+}
+
+// Peek returns the oldest item without removing it.
+func (q *Queue[T]) Peek() (T, bool) {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
+	if len(q.data) == 0 {
+		var zero T
+		return zero, false
+	}
+	return q.data[0], true
+}
+
+// Len returns the current number of items.
+func (q *Queue[T]) Len() int {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	return len(q.data)
+}
+
+// IsEmpty returns true if the queue is empty.
+func (q *Queue[T]) IsEmpty() bool {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	return len(q.data) == 0
+}

internal/dockeragent/agent.go

@@ -27,6 +27,7 @@ import (
 	"github.com/docker/docker/api/types/filters"
 	systemtypes "github.com/docker/docker/api/types/system"
 	"github.com/docker/docker/client"
+	"github.com/rcourtman/pulse-go-rewrite/internal/buffer"
 	"github.com/rcourtman/pulse-go-rewrite/internal/hostmetrics"
 	"github.com/rcourtman/pulse-go-rewrite/internal/utils"
 	agentsdocker "github.com/rcourtman/pulse-go-rewrite/pkg/agents/docker"
@@ -61,6 +62,7 @@ type Config struct {
 	CollectDiskMetrics bool
 	LogLevel           zerolog.Level
 	Logger             *zerolog.Logger
+	BufferCapacity     int // Number of reports to buffer when offline (default: 60)
 }
 
 var allowedContainerStates = map[string]string{
@@ -103,6 +105,7 @@ type Agent struct {
 	hostID              string
 	prevContainerCPU    map[string]cpuSample
 	preCPUStatsFailures int
+	reportBuffer        *buffer.Queue[agentsdocker.Report]
 }
 
 // ErrStopRequested indicates the agent should terminate gracefully after acknowledging a stop command.
@@ -235,6 +238,11 @@ func New(cfg Config) (*Agent, error) {
 		agentVersion = Version
 	}
 
+	bufferCapacity := cfg.BufferCapacity
+	if bufferCapacity <= 0 {
+		bufferCapacity = 60
+	}
+
 	agent := &Agent{
 		cfg:              cfg,
 		docker:           dockerClient,
@@ -252,6 +260,7 @@ func New(cfg Config) (*Agent, error) {
 		allowedStates:    make(map[string]struct{}, len(stateFilters)),
 		stateFilters:     stateFilters,
 		prevContainerCPU: make(map[string]cpuSample),
+		reportBuffer:     buffer.New[agentsdocker.Report](bufferCapacity),
 	}
 
 	for _, state := range stateFilters {
@@ -583,7 +592,41 @@ func (a *Agent) collectOnce(ctx context.Context) error {
 		return err
 	}
 
-	return a.sendReport(ctx, report)
+	if err := a.sendReport(ctx, report); err != nil {
+		if errors.Is(err, ErrStopRequested) {
+			return nil
+		}
+		a.logger.Warn().Err(err).Msg("Failed to send docker report, buffering")
+		a.reportBuffer.Push(report)
+		return nil
+	}
+
+	a.flushBuffer(ctx)
+	return nil
+}
+
+func (a *Agent) flushBuffer(ctx context.Context) {
+	if a.reportBuffer.IsEmpty() {
+		return
+	}
+
+	a.logger.Info().Int("count", a.reportBuffer.Len()).Msg("Flushing buffered docker reports")
+
+	for !a.reportBuffer.IsEmpty() {
+		report, ok := a.reportBuffer.Peek()
+		if !ok {
+			break
+		}
+
+		if err := a.sendReport(ctx, report); err != nil {
+			if errors.Is(err, ErrStopRequested) {
+				return
+			}
+			a.logger.Warn().Err(err).Msg("Failed to flush buffered docker report, stopping flush")
+			return
+		}
+		a.reportBuffer.Pop()
+	}
 }
 
 func (a *Agent) buildReport(ctx context.Context) (agentsdocker.Report, error) {

internal/hostagent/agent.go

@@ -12,6 +12,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/rcourtman/pulse-go-rewrite/internal/buffer"
 	"github.com/rcourtman/pulse-go-rewrite/internal/hostmetrics"
 	"github.com/rcourtman/pulse-go-rewrite/internal/mdadm"
 	"github.com/rcourtman/pulse-go-rewrite/internal/sensors"
@@ -34,6 +35,7 @@ type Config struct {
 	RunOnce            bool
 	LogLevel           zerolog.Level
 	Logger             *zerolog.Logger
+	BufferCapacity     int // Number of reports to buffer when offline (default: 60)
 }
 
 // Agent is responsible for collecting host metrics and shipping them to Pulse.
@@ -55,6 +57,7 @@ type Agent struct {
 	agentVersion    string
 	interval        time.Duration
 	trimmedPulseURL string
+	reportBuffer    *buffer.Queue[agentshost.Report]
 }
 
 const defaultInterval = 30 * time.Second
@@ -164,6 +167,11 @@ func New(cfg Config) (*Agent, error) {
 		agentVersion = Version
 	}
 
+	bufferCapacity := cfg.BufferCapacity
+	if bufferCapacity <= 0 {
+		bufferCapacity = 60
+	}
+
 	return &Agent{
 		cfg:             cfg,
 		logger:          logger,
@@ -181,6 +189,7 @@ func New(cfg Config) (*Agent, error) {
 		agentVersion:    agentVersion,
 		interval:        cfg.Interval,
 		trimmedPulseURL: pulseURL,
+		reportBuffer:    buffer.New[agentshost.Report](bufferCapacity),
 	}, nil
 }
 
@@ -222,8 +231,14 @@ func (a *Agent) process(ctx context.Context) error {
 		return fmt.Errorf("build report: %w", err)
 	}
 	if err := a.sendReport(ctx, report); err != nil {
-		return fmt.Errorf("send report: %w", err)
+		a.logger.Warn().Err(err).Msg("Failed to send report, buffering")
+		a.reportBuffer.Push(report)
+		return nil
 	}
+
+	// If successful, try to flush buffer
+	a.flushBuffer(ctx)
+
 	a.logger.Debug().
 		Str("hostname", report.Host.Hostname).
 		Str("platform", report.Host.Platform).
@@ -231,6 +246,30 @@ func (a *Agent) process(ctx context.Context) error {
 	return nil
 }
 
+func (a *Agent) flushBuffer(ctx context.Context) {
+	if a.reportBuffer.IsEmpty() {
+		return
+	}
+
+	a.logger.Info().Int("count", a.reportBuffer.Len()).Msg("Flushing buffered reports")
+
+	for !a.reportBuffer.IsEmpty() {
+		// Peek first
+		report, ok := a.reportBuffer.Peek()
+		if !ok {
+			break
+		}
+
+		if err := a.sendReport(ctx, report); err != nil {
+			a.logger.Warn().Err(err).Msg("Failed to flush buffered report, stopping flush")
+			return
+		}
+
+		// Pop only on success
+		a.reportBuffer.Pop()
+	}
+}
+
 func (a *Agent) buildReport(ctx context.Context) (agentshost.Report, error) {
 	collectCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
 	defer cancel()

internal/hostagent/agent_buffering_test.go

@@ -0,0 +1,156 @@
+package hostagent
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/rcourtman/pulse-go-rewrite/pkg/agents/host"
+	"github.com/rs/zerolog"
+)
+
+type testWriter struct {
+	t *testing.T
+}
+
+func (w *testWriter) Write(p []byte) (n int, err error) {
+	w.t.Log(string(p))
+	return len(p), nil
+}
+
+func TestAgentBuffering(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test in short mode")
+	}
+
+	// 1. Setup Mock Server
+	var (
+		mu              sync.Mutex
+		receivedReports []host.Report
+		shouldFail      bool
+	)
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		mu.Lock()
+		defer mu.Unlock()
+
+		t.Logf("Server received request: %s %s", r.Method, r.URL.Path)
+
+		if shouldFail {
+			t.Log("Server simulating failure (500)")
+			w.WriteHeader(http.StatusInternalServerError)
+			return
+		}
+
+		if r.URL.Path != "/api/agents/host/report" {
+			t.Logf("Server 404 for path: %s", r.URL.Path)
+			w.WriteHeader(http.StatusNotFound)
+			return
+		}
+
+		var report host.Report
+		if err := json.NewDecoder(r.Body).Decode(&report); err != nil {
+			t.Logf("Server failed to decode body: %v", err)
+			w.WriteHeader(http.StatusBadRequest)
+			return
+		}
+
+		t.Logf("Server accepted report from %s", report.Host.Hostname)
+		receivedReports = append(receivedReports, report)
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer server.Close()
+
+	// 2. Configure Agent
+	// Use testWriter to capture logs in test output
+	logger := zerolog.New(zerolog.ConsoleWriter{Out: &testWriter{t}}).Level(zerolog.DebugLevel)
+
+	cfg := Config{
+		PulseURL:         server.URL,
+		APIToken:         "test-token",
+		Interval:         250 * time.Millisecond,
+		HostnameOverride: "test-host",
+		Logger:           &logger,
+	}
+
+	agent, err := New(cfg)
+	if err != nil {
+		t.Fatalf("Failed to create agent: %v", err)
+	}
+
+	// 3. Run Agent in background
+	ctx, cancel := context.WithCancel(context.Background())
+	var wg sync.WaitGroup
+	wg.Add(1)
+
+	go func() {
+		defer wg.Done()
+		t.Log("Starting agent...")
+		if err := agent.Run(ctx); err != nil && err != context.Canceled {
+			t.Errorf("Agent run failed: %v", err)
+		}
+		t.Log("Agent stopped")
+	}()
+
+	// 4. Wait for initial successful report
+	t.Log("Waiting for initial report...")
+	time.Sleep(2 * time.Second)
+
+	mu.Lock()
+	initialCount := len(receivedReports)
+	mu.Unlock()
+
+	t.Logf("Initial reports received: %d", initialCount)
+	if initialCount == 0 {
+		t.Fatal("Expected at least one initial report")
+	}
+
+	// 5. Simulate Outage
+	t.Log("Simulating outage...")
+	mu.Lock()
+	shouldFail = true
+	mu.Unlock()
+
+	// Wait for a few cycles (should buffer)
+	time.Sleep(3 * time.Second)
+
+	mu.Lock()
+	// Should not have received any new reports during outage
+	if len(receivedReports) > initialCount {
+		t.Errorf("Received reports during outage! Expected %d, got %d", initialCount, len(receivedReports))
+	}
+	mu.Unlock()
+
+	// 6. Recover
+	t.Log("Recovering server...")
+	mu.Lock()
+	shouldFail = false
+	mu.Unlock()
+
+	// Wait for flush (flush happens after next successful report)
+	time.Sleep(3 * time.Second)
+
+	// Stop agent
+	cancel()
+	wg.Wait()
+
+	mu.Lock()
+	finalCount := len(receivedReports)
+	mu.Unlock()
+
+	t.Logf("Final reports received: %d", finalCount)
+
+	// We expect: initial + (buffered during outage) + (new ones after recovery)
+	// If collection is slow (e.g. 1s), we might get 1-2 buffered.
+	// We just want to ensure we got MORE than just the initial ones.
+	// And ideally more than just initial + 1 (which would be just the next report).
+	if finalCount <= initialCount+1 {
+		t.Errorf("Buffered reports were not flushed? Initial: %d, Final: %d", initialCount, finalCount)
+	}
+
+	t.Logf("Test passed. Initial: %d, Final: %d", initialCount, finalCount)
+}