From b2d434b4c27ab20f293fc997656d5c314decba4c Mon Sep 17 00:00:00 2001 From: "courtmanr@gmail.com" Date: Tue, 2 Dec 2025 22:12:47 +0000 Subject: [PATCH] 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 | 74 ++++++++++ internal/dockeragent/agent.go | 45 +++++- internal/hostagent/agent.go | 41 +++++- internal/hostagent/agent_buffering_test.go | 156 +++++++++++++++++++++ 4 files changed, 314 insertions(+), 2 deletions(-) create mode 100644 internal/buffer/buffer.go create mode 100644 internal/hostagent/agent_buffering_test.go diff --git a/internal/buffer/buffer.go b/internal/buffer/buffer.go new file mode 100644 index 0000000..d614f21 --- /dev/null +++ b/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 +} diff --git a/internal/dockeragent/agent.go b/internal/dockeragent/agent.go index 8cdf22e..e98c13b 100644 --- a/internal/dockeragent/agent.go +++ b/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) { diff --git a/internal/hostagent/agent.go b/internal/hostagent/agent.go index b67d413..4ef2f04 100644 --- a/internal/hostagent/agent.go +++ b/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() diff --git a/internal/hostagent/agent_buffering_test.go b/internal/hostagent/agent_buffering_test.go new file mode 100644 index 0000000..d314c77 --- /dev/null +++ b/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) +}