feat: improve batch processing by grouping files by exporter and selecting oldest for efficient sending

Author: guiyumin

internal/report/sender.go

@@ -10,7 +10,6 @@ import (
 	"net/http"
 	"net/url"
 	"path/filepath"
-	"sort"
 	"strings"
 	"time"
 
@@ -149,24 +148,15 @@ func (s *Sender) drainLoop() {
 			continue
 		}
 
-		// Group files by time window (5s buckets) for efficient batching
-		timeWindows := s.groupFilesByTimeWindow(files, 5*time.Second)
+		// NEW APPROACH: Group files by exporter, pick oldest from each
+		// This ensures all exporters are represented in each batch
+		batch := s.selectOldestFromEachExporter(files, s.config.Buffer.BatchSize)
 
-		// Process first time window (oldest files first)
-		if len(timeWindows) > 0 {
-			firstWindow := timeWindows[0]
-
-			// Limit batch size to configured batch_size
-			batchSize := len(firstWindow)
-			if batchSize > s.config.Buffer.BatchSize {
-				batchSize = s.config.Buffer.BatchSize
-			}
-
-			batch := firstWindow[:batchSize]
+		if len(batch) > 0 {
 			if err := s.processBatch(batch); err != nil {
 				// Failed to send - keep files and retry after delay
 				logger.Debug("Failed to process batch, will retry",
-					logger.Int("batch_size", batchSize),
+					logger.Int("batch_size", len(batch)),
 					logger.Err(err))
 			}
 		}
@@ -314,71 +304,37 @@ func (s *Sender) processBatch(filePaths []string) error {
 	return nil
 }
 
-// groupFilesByTimeWindow groups files into time buckets (e.g., 5s windows)
-// This allows batching multiple exporters that scraped at similar times
-// Returns a list of time windows (oldest first), each containing file paths
-func (s *Sender) groupFilesByTimeWindow(filePaths []string, windowSize time.Duration) [][]string {
-	// Map: timestamp bucket -> file paths
-	windows := make(map[int64][]string)
+// selectOldestFromEachExporter picks the oldest file from each exporter directory
+// This ensures all exporters are represented in each batch, preventing one exporter
+// from blocking others if it has a backlog
+func (s *Sender) selectOldestFromEachExporter(filePaths []string, maxBatch int) []string {
+	// Group files by exporter (directory name)
+	byExporter := make(map[string][]string)
 
 	for _, filePath := range filePaths {
-		// Parse timestamp from filename: YYYYMMDD-HHMMSS-...
-		timestamp, err := parseTimestampFromFilename(filePath)
-		if err != nil {
-			logger.Warn("Failed to parse timestamp from filename, skipping",
-				logger.String("file", filePath),
-				logger.Err(err))
-			continue
-		}
+		// Extract exporter name from path: buffer/<exporter>/file.prom
+		dir := filepath.Dir(filePath)
+		exporterName := filepath.Base(dir)
 
-		// Bucket by time window (e.g., 5s buckets)
-		bucket := timestamp.Unix() / int64(windowSize.Seconds())
-		windows[bucket] = append(windows[bucket], filePath)
+		byExporter[exporterName] = append(byExporter[exporterName], filePath)
 	}
 
-	// Convert to sorted list of windows (oldest first)
-	buckets := make([]int64, 0, len(windows))
-	for bucket := range windows {
-		buckets = append(buckets, bucket)
+	// Pick oldest file from each exporter (files are already sorted chronologically)
+	batch := make([]string, 0, len(byExporter))
+	for _, files := range byExporter {
+		if len(files) > 0 {
+			batch = append(batch, files[0]) // First file is oldest
+		}
 	}
-	sort.Slice(buckets, func(i, j int) bool {
-		return buckets[i] < buckets[j]
-	})
 
-	result := make([][]string, 0, len(buckets))
-	for _, bucket := range buckets {
-		result = append(result, windows[bucket])
+	// Limit to maxBatch if needed
+	if len(batch) > maxBatch {
+		batch = batch[:maxBatch]
 	}
 
-	return result
+	return batch
 }
 
-// parseTimestampFromFilename extracts timestamp from buffer filename
-// Format: buffer/<exporter>/YYYYMMDD-HHMMSS-<server_id>.prom
-func parseTimestampFromFilename(filePath string) (time.Time, error) {
-	filename := filepath.Base(filePath)
-
-	// Remove .prom extension
-	if !strings.HasSuffix(filename, ".prom") {
-		return time.Time{}, fmt.Errorf("invalid file extension")
-	}
-
-	// Extract timestamp part (first two segments: YYYYMMDD-HHMMSS)
-	parts := strings.SplitN(strings.TrimSuffix(filename, ".prom"), "-", 3)
-	if len(parts) < 2 {
-		return time.Time{}, fmt.Errorf("invalid filename format")
-	}
-
-	timeStr := parts[0] + "-" + parts[1]
-
-	// Parse timestamp
-	timestamp, err := time.Parse("20060102-150405", timeStr)
-	if err != nil {
-		return time.Time{}, fmt.Errorf("failed to parse timestamp: %w", err)
-	}
-
-	return timestamp, nil
-}
 
 // randomDelay waits for a random duration between 0 and the configured interval
 // This distributes load across the interval window