feat: implement Write-Ahead Log pattern for metrics buffering and enhance buffer handling

Author: guiyumin

cmd/setup.go

@@ -16,20 +16,19 @@ var (
 	quickMode bool
 
 	// Config flags for quick mode
-	flagEndpointURL      string
-	flagServerID         string
-	flagInterval         string
-	flagTimeout          string
-	flagBufferEnabled    bool
-	flagBufferDir        string
-	flagBufferRetention  int
-	flagLogLevel         string
-	flagLogOutput        string
-	flagLogFile          string
-	flagLogMaxSize       int
-	flagLogMaxBackups    int
-	flagLogMaxAge        int
-	flagLogCompress      bool
+	flagEndpointURL     string
+	flagServerID        string
+	flagInterval        string
+	flagTimeout         string
+	flagBufferDir       string
+	flagBufferRetention int
+	flagLogLevel        string
+	flagLogOutput       string
+	flagLogFile         string
+	flagLogMaxSize      int
+	flagLogMaxBackups   int
+	flagLogMaxAge       int
+	flagLogCompress     bool
 )
 
 // setupCmd represents the setup command
@@ -56,8 +55,7 @@ func init() {
 	setupCmd.Flags().StringVar(&flagServerID, "server-id", "", "Server ID (auto-generated UUID if not provided)")
 	setupCmd.Flags().StringVar(&flagInterval, "interval", "5s", "Metric collection interval (5s, 10s, 30s, or 1m)")
 
-	// Buffer configuration flags
-	setupCmd.Flags().BoolVar(&flagBufferEnabled, "buffer-enabled", true, "Enable offline buffering")
+	// Buffer configuration flags (buffer is always enabled)
 	setupCmd.Flags().StringVar(&flagBufferDir, "buffer-dir", "/var/lib/node-pulse/buffer", "Buffer directory path")
 	setupCmd.Flags().IntVar(&flagBufferRetention, "buffer-retention", 48, "Buffer retention in hours")
 
@@ -161,8 +159,7 @@ func runQuickMode(existing *installer.ExistingInstall) error {
 		ServerID: finalServerID,
 		Interval: flagInterval,
 
-		// Buffer options
-		BufferEnabled:        flagBufferEnabled,
+		// Buffer options (always enabled)
 		BufferPath:           flagBufferDir,
 		BufferRetentionHours: flagBufferRetention,
 
@@ -178,12 +175,12 @@ func runQuickMode(existing *installer.ExistingInstall) error {
 
 	fmt.Println()
 	fmt.Printf("Configuration summary:\n")
-	fmt.Printf("  Endpoint:       %s\n", opts.Endpoint)
-	fmt.Printf("  Server ID:      %s\n", opts.ServerID)
-	fmt.Printf("  Interval:       %s\n", opts.Interval)
-	fmt.Printf("  Timeout:        %s\n", opts.Timeout)
-	fmt.Printf("  Buffer enabled: %v\n", opts.BufferEnabled)
-	fmt.Printf("  Log level:      %s\n", opts.LogLevel)
+	fmt.Printf("  Endpoint:     %s\n", opts.Endpoint)
+	fmt.Printf("  Server ID:    %s\n", opts.ServerID)
+	fmt.Printf("  Interval:     %s\n", opts.Interval)
+	fmt.Printf("  Timeout:      %s\n", opts.Timeout)
+	fmt.Printf("  Buffer path:  %s\n", opts.BufferPath)
+	fmt.Printf("  Log level:    %s\n", opts.LogLevel)
 	fmt.Println()
 
 	// Perform installation

cmd/setup_tui.go

@@ -666,10 +666,7 @@ func (m setupTUIModel) viewReview() string {
 	summary.WriteString(labelStyle.Render("Server ID:") + " " + valueStyle.Render(m.config.ServerID) + "\n")
 	summary.WriteString(labelStyle.Render("Interval:") + " " + valueStyle.Render(m.config.Interval) + "\n")
 	summary.WriteString(labelStyle.Render("Timeout:") + " " + valueStyle.Render(m.config.Timeout) + "\n")
-	bufferStatus := "Disabled"
-	if m.config.BufferEnabled {
-		bufferStatus = fmt.Sprintf("Enabled (%dh retention)", m.config.BufferRetentionHours)
-	}
+	bufferStatus := fmt.Sprintf("%s (%dh retention)", m.config.BufferPath, m.config.BufferRetentionHours)
 	summary.WriteString(labelStyle.Render("Buffer:") + " " + valueStyle.Render(bufferStatus) + "\n")
 	summary.WriteString(labelStyle.Render("Logging:") + " " + valueStyle.Render(fmt.Sprintf("%s → %s", m.config.LogLevel, m.config.LogOutput)) + "\n")
 	summary.WriteString(labelStyle.Render("Config Path:") + " " + valueStyle.Render("/etc/node-pulse/nodepulse.yml"))
@@ -902,29 +899,16 @@ func (m setupTUIModel) handleEnter() (tea.Model, tea.Cmd) {
 		m.config.Timeout = timeout
 		m.err = nil
 
-		// Move to buffer screen
-		m.screen = ScreenBuffer
-		if m.config.BufferEnabled {
-			m.textInput.SetValue("yes")
-		} else {
-			m.textInput.SetValue("no")
-		}
-		m.textInput.Placeholder = "yes/no"
+		// Skip buffer screen (buffer is always enabled now) and go to logging
+		m.screen = ScreenLogging
+		m.textInput.SetValue(m.config.LogLevel)
+		m.textInput.Placeholder = "debug, info, warn, error"
 		m.textInput.Focus()
 		return m, textinput.Blink
 
 	case ScreenBuffer:
-		// Parse buffer settings
-		input := strings.ToLower(strings.TrimSpace(m.textInput.Value()))
-		if input != "yes" && input != "no" {
-			m.err = fmt.Errorf("enter 'yes' or 'no'")
-			return m, nil
-		}
-
-		m.config.BufferEnabled = (input == "yes")
-		m.err = nil
-
-		// Move to logging screen
+		// Buffer screen is deprecated (buffer is always enabled)
+		// Skip to logging screen
 		m.screen = ScreenLogging
 		m.textInput.SetValue(m.config.LogLevel)
 		m.textInput.Placeholder = "debug, info, warn, error"
@@ -1005,14 +989,10 @@ func (m setupTUIModel) handleBack() (tea.Model, tea.Cmd) {
 		return m, textinput.Blink
 
 	case ScreenLogging:
-		// Go back to buffer
-		m.screen = ScreenBuffer
-		m.textInput.Placeholder = "yes/no"
-		if m.config.BufferEnabled {
-			m.textInput.SetValue("yes")
-		} else {
-			m.textInput.SetValue("no")
-		}
+		// Go back to timeout (buffer screen is skipped)
+		m.screen = ScreenTimeout
+		m.textInput.Placeholder = "3s"
+		m.textInput.SetValue(m.config.Timeout)
 		m.textInput.Focus()
 		return m, textinput.Blink
 

cmd/start.go

@@ -88,6 +88,10 @@ func runAgent(cmd *cobra.Command, args []string) error {
 	}
 	defer sender.Close()
 
+	// Start background draining goroutine (WAL pattern)
+	// This continuously sends buffered reports with random jitter
+	sender.StartDraining()
+
 	// Setup signal handling for graceful shutdown
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -110,19 +114,20 @@ func runAgent(cmd *cobra.Command, args []string) error {
 		logger.Duration("interval", cfg.Agent.Interval),
 		logger.String("endpoint", cfg.Server.Endpoint))
 
-	// Collect and send immediately on start
+	// Collect and save to buffer immediately on start
 	if err := collectAndSend(sender, cfg.Agent.ServerID); err != nil {
-		logger.Error("Collection and send failed", logger.Err(err))
+		logger.Error("Collection failed", logger.Err(err))
 	}
 
-	// Then continue with ticker
+	// Then continue with ticker (collect and save to buffer at interval)
+	// Background goroutine will drain buffer with random jitter
 	for {
 		select {
 		case <-ctx.Done():
 			return nil
 		case <-ticker.C:
 			if err := collectAndSend(sender, cfg.Agent.ServerID); err != nil {
-				logger.Error("Collection and send failed", logger.Err(err))
+				logger.Error("Collection failed", logger.Err(err))
 			}
 		}
 	}
@@ -139,16 +144,16 @@ func collectAndSend(sender *report.Sender, serverID string) error {
 	stats := metrics.GetGlobalStats()
 	stats.RecordCollection(metricsReport)
 
-	// Send report
+	// Save to buffer (WAL pattern - actual sending happens in background with jitter)
 	if err := sender.Send(metricsReport); err != nil {
-		// Record failure
+		// Record failure (failed to save to buffer)
 		stats.RecordFailure()
-		return fmt.Errorf("failed to send report: %w", err)
+		return fmt.Errorf("failed to save to buffer: %w", err)
 	}
 
-	// Record success
+	// Record success (successfully buffered)
 	stats.RecordSuccess()
-	logger.Info("Report sent successfully")
+	logger.Info("Report collected and buffered")
 	return nil
 }
 

cmd/status.go

@@ -59,18 +59,14 @@ func runStatus(cmd *cobra.Command, args []string) error {
 	fmt.Printf("Agent:         %s\n", serviceStatus)
 	fmt.Println()
 
-	// Buffer Status
-	if cfg.Buffer.Enabled {
-		bufferCount, err := countBufferFiles(cfg.Buffer.Path)
-		if err != nil {
-			fmt.Printf("Buffer:        enabled (error checking: %v)\n", err)
-		} else if bufferCount > 0 {
-			fmt.Printf("Buffer:        %d report(s) pending in %s\n", bufferCount, cfg.Buffer.Path)
-		} else {
-			fmt.Printf("Buffer:        enabled, no pending reports\n")
-		}
+	// Buffer Status (always enabled in new architecture)
+	bufferCount, err := countBufferFiles(cfg.Buffer.Path)
+	if err != nil {
+		fmt.Printf("Buffer:        error checking: %v\n", err)
+	} else if bufferCount > 0 {
+		fmt.Printf("Buffer:        %d report(s) pending in %s\n", bufferCount, cfg.Buffer.Path)
 	} else {
-		fmt.Printf("Buffer:        disabled\n")
+		fmt.Printf("Buffer:        no pending reports\n")
 	}
 	fmt.Println()
 

internal/config/config.go

@@ -34,8 +34,8 @@ type AgentConfig struct {
 }
 
 // BufferConfig represents buffer settings
+// Note: Buffer is always enabled in the new architecture (write-ahead log pattern)
 type BufferConfig struct {
-	Enabled        bool   `mapstructure:"enabled"`
 	Path           string `mapstructure:"path"`
 	RetentionHours int    `mapstructure:"retention_hours"`
 }
@@ -50,7 +50,6 @@ var (
 			Interval: 5 * time.Second,
 		},
 		Buffer: BufferConfig{
-			Enabled:        true,
 			Path:           "/var/lib/node-pulse/buffer",
 			RetentionHours: 48,
 		},
@@ -121,7 +120,6 @@ func setDefaults(v *viper.Viper) {
 	v.SetDefault("server.endpoint", defaultConfig.Server.Endpoint)
 	v.SetDefault("server.timeout", defaultConfig.Server.Timeout)
 	v.SetDefault("agent.interval", defaultConfig.Agent.Interval)
-	v.SetDefault("buffer.enabled", defaultConfig.Buffer.Enabled)
 	v.SetDefault("buffer.path", defaultConfig.Buffer.Path)
 	v.SetDefault("buffer.retention_hours", defaultConfig.Buffer.RetentionHours)
 	v.SetDefault("logging.level", defaultConfig.Logging.Level)
@@ -175,13 +173,12 @@ func validate(cfg *Config) error {
 		return fmt.Errorf("agent.interval must be one of: 5s, 10s, 30s, 1m")
 	}
 
-	if cfg.Buffer.Enabled {
-		if cfg.Buffer.Path == "" {
-			return fmt.Errorf("buffer.path is required when buffer is enabled")
-		}
-		if cfg.Buffer.RetentionHours <= 0 {
-			return fmt.Errorf("buffer.retention_hours must be positive")
-		}
+	// Buffer is always enabled now
+	if cfg.Buffer.Path == "" {
+		return fmt.Errorf("buffer.path is required")
+	}
+	if cfg.Buffer.RetentionHours <= 0 {
+		return fmt.Errorf("buffer.retention_hours must be positive")
 	}
 
 	return nil
@@ -227,23 +224,13 @@ func isAlphanumeric(c rune) bool {
 
 // EnsureBufferDir creates the buffer directory if it doesn't exist
 func (c *Config) EnsureBufferDir() error {
-	if !c.Buffer.Enabled {
-		return nil
-	}
-
 	if err := os.MkdirAll(c.Buffer.Path, 0755); err != nil {
 		return fmt.Errorf("failed to create buffer directory: %w", err)
 	}
 
 	return nil
 }
 
-// GetBufferFilePath returns the path for the current hour's buffer file
-func (c *Config) GetBufferFilePath(t time.Time) string {
-	filename := fmt.Sprintf("%s.jsonl", t.Format("2006-01-02-15"))
-	return filepath.Join(c.Buffer.Path, filename)
-}
-
 // ConfigExists checks if a configuration file exists in any of the standard locations
 func ConfigExists(configPath string) bool {
 	// If explicit config path provided, check only that

internal/installer/installer.go

@@ -33,8 +33,7 @@ type ConfigOptions struct {
 	ServerID string
 	Interval string
 
-	// Buffer options
-	BufferEnabled       bool
+	// Buffer options (buffer is always enabled in new architecture)
 	BufferPath          string
 	BufferRetentionHours int
 
@@ -207,8 +206,7 @@ func DefaultConfigOptions() ConfigOptions {
 		// Agent defaults
 		Interval: "5s",
 
-		// Buffer defaults
-		BufferEnabled:       true,
+		// Buffer defaults (always enabled)
 		BufferPath:          DefaultBufferPath,
 		BufferRetentionHours: 48,
 
@@ -236,7 +234,6 @@ func WriteConfigFile(opts ConfigOptions) error {
 			"interval":  opts.Interval,
 		},
 		"buffer": map[string]interface{}{
-			"enabled":         opts.BufferEnabled,
 			"path":            opts.BufferPath,
 			"retention_hours": opts.BufferRetentionHours,
 		},