Merge pull request #353 from Meesho/async-data-logger-metrics

metric updates and GCS uploader changes

Author: neehar-mavuduru

asyncloguploader/config.go

@@ -21,8 +21,11 @@ type Config struct {
 	FlushTimeout  time.Duration // Wait for write completion before flush (default: 10ms)
 
 	// Upload configuration
-	UploadChannel   chan<- string    // Optional: channel for completed files
-	GCSUploadConfig *GCSUploadConfig // Optional: GCS upload configuration
+	GCSUploadConfig *GCSUploadConfig // Optional: GCS upload configuration (uploader scans log dir)
+
+	// MetricTags are application-provided tags propagated to all metric emissions
+	// (e.g., from metric.BuildTag(metric.NewTag("service", "x"), ...))
+	MetricTags []string
 }
 
 // GCSUploadConfig holds configuration for GCS uploader
@@ -34,7 +37,7 @@ type GCSUploadConfig struct {
 	MaxRetries          int           // Max retry attempts (default: 3)
 	RetryDelay          time.Duration // Delay between retries (default: 5s)
 	GRPCPoolSize        int           // gRPC connection pool size (default: 64)
-	ChannelBufferSize   int           // Upload channel buffer size (default: 100)
+	ScanInterval        time.Duration // How often to scan log dir for .log files (default: 10s)
 }
 
 // DefaultConfig returns a configuration with baseline defaults
@@ -47,7 +50,6 @@ func DefaultConfig(logPath string) Config {
 		PreallocateFileSize: 0, // Disabled by default
 		FlushInterval:       10 * time.Second,
 		FlushTimeout:        10 * time.Millisecond,
-		UploadChannel:       nil, // Optional
 		GCSUploadConfig:     nil, // Optional
 	}
 }
@@ -62,7 +64,7 @@ func DefaultGCSUploadConfig(bucket string) GCSUploadConfig {
 		MaxRetries:          3,
 		RetryDelay:          5 * time.Second,
 		GRPCPoolSize:        64,
-		ChannelBufferSize:   100,
+		ScanInterval:        10 * time.Second,
 	}
 }
 
@@ -130,8 +132,8 @@ func (g *GCSUploadConfig) Validate() error {
 		g.GRPCPoolSize = 64
 	}
 
-	if g.ChannelBufferSize <= 0 {
-		g.ChannelBufferSize = 100
+	if g.ScanInterval <= 0 {
+		g.ScanInterval = 10 * time.Second
 	}
 
 	return nil

asyncloguploader/file_rename.go

@@ -0,0 +1,28 @@
+package asyncloguploader
+
+import (
+	"os"
+	"strings"
+	"time"
+
+	"github.com/Meesho/go-core/metric"
+)
+
+const renameRetryCount = 3
+const renameRetryDelay = 10 * time.Millisecond
+
+// renameTmpToLog renames a .log.tmp file to .log with retries.
+// On failure after retries, emits MetricFileRenameFailed and returns nil (ignore per spec).
+func renameTmpToLog(tmpPath string, metricTags []string) {
+	logPath := strings.TrimSuffix(tmpPath, ".tmp")
+	if logPath == tmpPath {
+		return // not a .tmp file, nothing to do
+	}
+	for i := 0; i < renameRetryCount; i++ {
+		if err := os.Rename(tmpPath, logPath); err == nil {
+			return
+		}
+		time.Sleep(renameRetryDelay)
+	}
+	metric.Incr(MetricFileRenameFailed, getBaseTags(metricTags))
+}

asyncloguploader/file_writer_default.go

@@ -11,6 +11,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/Meesho/go-core/metric"
 	logger "github.com/rs/zerolog/log"
 )
 
@@ -39,21 +40,23 @@ type SizeFileWriter struct {
 	// Last write duration (for metrics tracking)
 	lastPwritevDuration atomic.Int64 // Nanoseconds
 
-	// Channel for completed files (for GCS upload)
-	completedFileChan chan<- string
+	// Metric tags for emissions
+	metricTags []string
 }
 
-// NewSizeFileWriter creates a new SizeFileWriter (non-Linux fallback)
-func NewSizeFileWriter(config Config, completedFileChan chan<- string) (*SizeFileWriter, error) {
+// NewSizeFileWriter creates a new SizeFileWriter (non-Linux fallback).
+// Files are written with .log.tmp suffix during active write; renamed to .log on rotation/close.
+func NewSizeFileWriter(config Config, metricTags []string) (*SizeFileWriter, error) {
 	// Extract base directory and filename
 	baseDir, baseFileName, err := extractBasePathSize(config.LogFilePath)
 	if err != nil {
 		return nil, fmt.Errorf("failed to extract base path: %w", err)
 	}
 
 	// Generate timestamped filename for initial file
+	// Use .log.tmp during active write; rename to .log on rotation/close
 	timestamp := time.Now().Format("2006-01-02_15-04-05")
-	initialPath := filepath.Join(baseDir, fmt.Sprintf("%s_%s.log", baseFileName, timestamp))
+	initialPath := filepath.Join(baseDir, fmt.Sprintf("%s_%s.log.tmp", baseFileName, timestamp))
 
 	// Open initial file (always starts at offset 0 for new files)
 	file, err := openDirectIOSize(initialPath, config.PreallocateFileSize)
@@ -69,7 +72,7 @@ func NewSizeFileWriter(config Config, completedFileChan chan<- string) (*SizeFil
 		baseDir:             baseDir,
 		baseFileName:        baseFileName,
 		preallocateFileSize: config.PreallocateFileSize,
-		completedFileChan:   completedFileChan,
+		metricTags:          getBaseTags(metricTags),
 	}
 
 	// New files always start at offset 0
@@ -93,7 +96,6 @@ func (fw *SizeFileWriter) WriteVectored(buffers [][]byte) (int, error) {
 	// Get current offset
 	offset := fw.fileOffset.Load()
 
-	// Write sequentially (non-Linux fallback)
 	writeStart := time.Now()
 	totalWritten := 0
 	for _, buf := range buffers {
@@ -103,14 +105,17 @@ func (fw *SizeFileWriter) WriteVectored(buffers [][]byte) (int, error) {
 		n, err := fw.file.WriteAt(buf, offset+int64(totalWritten))
 		if err != nil {
 			logger.Error().Err(err).Msgf("write failed (file=%s offset=%d)", fw.filePath, offset+int64(totalWritten))
-			fw.lastPwritevDuration.Store(time.Since(writeStart).Nanoseconds())
+			writeDuration := time.Since(writeStart)
+			fw.lastPwritevDuration.Store(writeDuration.Nanoseconds())
+			metric.Timing(MetricFileWriterWriteDuration, writeDuration, fw.metricTags)
 			return totalWritten, err
 		}
 		totalWritten += n
 	}
 	writeDuration := time.Since(writeStart)
 
 	fw.lastPwritevDuration.Store(writeDuration.Nanoseconds())
+	metric.Timing(MetricFileWriterWriteDuration, writeDuration, fw.metricTags)
 	fw.fileOffset.Add(int64(totalWritten))
 
 	return totalWritten, nil
@@ -129,7 +134,6 @@ func (fw *SizeFileWriter) Close() error {
 	// We need to complete the rotation: swap files, then close both
 	if fw.nextFile != nil && fw.file != nil {
 		// Complete the rotation by swapping files
-		// This will send the current file to upload channel
 		if err := fw.swapFiles(); err != nil && firstErr == nil {
 			firstErr = fmt.Errorf("failed to complete rotation during close: %w", err)
 		}
@@ -167,16 +171,9 @@ func (fw *SizeFileWriter) Close() error {
 			firstErr = err
 		}
 
-		// Send completed file to upload channel (non-blocking) if it has data
-		if hasData && fw.completedFileChan != nil {
-			select {
-			case fw.completedFileChan <- completedFilePath:
-				// Successfully sent to channel
-			default:
-				// Channel full - log warning but don't block close
-				logger.Warn().Msgf("upload channel full, skipping upload (file=%s)", completedFilePath)
-				fmt.Printf("[WARNING] Upload channel full, skipping upload for %s\n", completedFilePath)
-			}
+		// Rename .tmp to .log (retry 3x, emit metric on failure)
+		if hasData {
+			renameTmpToLog(completedFilePath, fw.metricTags)
 		}
 
 		fw.file = nil
@@ -234,7 +231,7 @@ func (fw *SizeFileWriter) rotateIfNeeded() error {
 // createNextFile creates a new file for rotation
 func (fw *SizeFileWriter) createNextFile() error {
 	timestamp := time.Now().Format("2006-01-02_15-04-05")
-	nextPath := filepath.Join(fw.baseDir, fmt.Sprintf("%s_%s.log", fw.baseFileName, timestamp))
+	nextPath := filepath.Join(fw.baseDir, fmt.Sprintf("%s_%s.log.tmp", fw.baseFileName, timestamp))
 
 	file, err := openDirectIOSize(nextPath, fw.preallocateFileSize)
 	if err != nil {
@@ -281,27 +278,21 @@ func (fw *SizeFileWriter) swapFiles() error {
 		return fmt.Errorf("failed to close current file: %w", err)
 	}
 
-	// Send completed file to upload channel (non-blocking)
-	if fw.completedFileChan != nil {
-		select {
-		case fw.completedFileChan <- completedFilePath:
-		default:
-			logger.Warn().Msgf("upload channel full, skipping upload (file=%s)", completedFilePath)
-			fmt.Printf("[WARNING] Upload channel full, skipping upload for %s\n", completedFilePath)
-		}
-	}
+	// Rename .tmp to .log (retry 3x, emit metric on failure)
+	renameTmpToLog(completedFilePath, fw.metricTags)
 
 	// Swap next file to current
 	fw.file = fw.nextFile
 	fw.fd = fw.nextFd
 	fw.filePath = fw.nextFilePath
 	fw.fileOffset.Store(0)
 
-	// Clear next file fields
 	fw.nextFile = nil
 	fw.nextFd = 0
 	fw.nextFilePath = ""
 
+	metric.Incr(MetricFileWriterRotationCount, fw.metricTags)
+
 	return nil
 }
 

asyncloguploader/file_writer_linux.go

@@ -11,6 +11,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/Meesho/go-core/metric"
 	logger "github.com/rs/zerolog/log"
 	"golang.org/x/sys/unix"
 )
@@ -46,22 +47,24 @@ type SizeFileWriter struct {
 	// Last Pwritev duration (for metrics tracking)
 	lastPwritevDuration atomic.Int64 // Nanoseconds
 
-	// Channel for completed files (for GCS upload)
-	completedFileChan chan<- string
+	// Metric tags for emissions (e.g., from Config.MetricTags + event_name)
+	metricTags []string
 }
 
-// NewSizeFileWriter creates a new SizeFileWriter with the given configuration
-// completedFileChan is optional - if provided, completed files will be sent to this channel for upload
-func NewSizeFileWriter(config Config, completedFileChan chan<- string) (*SizeFileWriter, error) {
+// NewSizeFileWriter creates a new SizeFileWriter with the given configuration.
+// Files are written with .log.tmp suffix during active write; renamed to .log on rotation/close.
+// metricTags are propagated to metric emissions (e.g., from MergeMetricTags(config.MetricTags, eventName))
+func NewSizeFileWriter(config Config, metricTags []string) (*SizeFileWriter, error) {
 	// Extract base directory and filename
 	baseDir, baseFileName, err := extractBasePathSize(config.LogFilePath)
 	if err != nil {
 		return nil, fmt.Errorf("failed to extract base path: %w", err)
 	}
 
 	// Generate timestamped filename for initial file (consistent naming)
+	// Use .log.tmp during active write; rename to .log on rotation/close
 	timestamp := time.Now().Format("2006-01-02_15-04-05")
-	initialPath := filepath.Join(baseDir, fmt.Sprintf("%s_%s.log", baseFileName, timestamp))
+	initialPath := filepath.Join(baseDir, fmt.Sprintf("%s_%s.log.tmp", baseFileName, timestamp))
 
 	// Open initial file with preallocation (always starts at offset 0 for new files)
 	file, err := openDirectIOSize(initialPath, config.PreallocateFileSize)
@@ -77,7 +80,7 @@ func NewSizeFileWriter(config Config, completedFileChan chan<- string) (*SizeFil
 		baseDir:             baseDir,
 		baseFileName:        baseFileName,
 		preallocateFileSize: config.PreallocateFileSize,
-		completedFileChan:   completedFileChan,
+		metricTags:          getBaseTags(metricTags),
 	}
 
 	// New files always start at offset 0
@@ -108,15 +111,14 @@ func (fw *SizeFileWriter) WriteVectored(buffers [][]byte) (int, error) {
 	n, err := writevAlignedWithOffset(fw.fd, buffers, offset)
 	pwritevDuration := time.Since(pwritevStart)
 
-	// Store write duration for metrics
 	fw.lastPwritevDuration.Store(pwritevDuration.Nanoseconds())
+	metric.Timing(MetricFileWriterWriteDuration, pwritevDuration, fw.metricTags)
 
 	if err != nil {
 		logger.Error().Err(err).Msgf("writev failed (file=%s offset=%d)", fw.filePath, offset)
 		return n, err
 	}
 
-	// Update offset atomically after successful write
 	fw.fileOffset.Add(int64(n))
 
 	return n, nil
@@ -135,7 +137,6 @@ func (fw *SizeFileWriter) Close() error {
 	// We need to complete the rotation: swap files, then close both
 	if fw.nextFile != nil && fw.file != nil {
 		// Complete the rotation by swapping files
-		// This will send the current file to upload channel
 		if err := fw.swapFiles(); err != nil && firstErr == nil {
 			firstErr = fmt.Errorf("failed to complete rotation during close: %w", err)
 		}
@@ -173,16 +174,9 @@ func (fw *SizeFileWriter) Close() error {
 			firstErr = err
 		}
 
-		// Send completed file to upload channel (non-blocking) if it has data
-		if hasData && fw.completedFileChan != nil {
-			select {
-			case fw.completedFileChan <- completedFilePath:
-				// Successfully sent to channel
-			default:
-				// Channel full - log warning but don't block close
-				logger.Warn().Msgf("upload channel full, skipping upload (file=%s)", completedFilePath)
-				fmt.Printf("[WARNING] Upload channel full, skipping upload for %s\n", completedFilePath)
-			}
+		// Rename .tmp to .log (retry 3x, emit metric on failure)
+		if hasData {
+			renameTmpToLog(completedFilePath, fw.metricTags)
 		}
 
 		fw.file = nil
@@ -249,9 +243,9 @@ func (fw *SizeFileWriter) rotateIfNeeded() error {
 
 // createNextFile creates a new file for rotation with preallocation
 func (fw *SizeFileWriter) createNextFile() error {
-	// Generate timestamped filename: {baseFileName}_{YYYY-MM-DD_HH-MM-SS}.log
+	// Generate timestamped filename: {baseFileName}_{YYYY-MM-DD_HH-MM-SS}.log.tmp
 	timestamp := time.Now().Format("2006-01-02_15-04-05")
-	nextPath := filepath.Join(fw.baseDir, fmt.Sprintf("%s_%s.log", fw.baseFileName, timestamp))
+	nextPath := filepath.Join(fw.baseDir, fmt.Sprintf("%s_%s.log.tmp", fw.baseFileName, timestamp))
 
 	// Try to open new file with preallocation
 	file, err := openDirectIOSize(nextPath, fw.preallocateFileSize)
@@ -302,37 +296,29 @@ func (fw *SizeFileWriter) swapFiles() error {
 		}
 	}
 
-	// Store current file path before closing (for upload)
+	// Store current file path before closing
 	completedFilePath := fw.filePath
 
 	// Close current file
 	if err := fw.file.Close(); err != nil {
 		return fmt.Errorf("failed to close current file: %w", err)
 	}
 
-	// Send completed file to upload channel (non-blocking)
-	if fw.completedFileChan != nil {
-		select {
-		case fw.completedFileChan <- completedFilePath:
-			// Successfully sent to channel
-		default:
-			// Channel full - log warning but don't block rotation
-			logger.Warn().Msgf("upload channel full, skipping upload (file=%s)", completedFilePath)
-			fmt.Printf("[WARNING] Upload channel full, skipping upload for %s\n", completedFilePath)
-		}
-	}
+	// Rename .tmp to .log (retry 3x, emit metric on failure)
+	renameTmpToLog(completedFilePath, fw.metricTags)
 
 	// Swap next file to current
 	fw.file = fw.nextFile
 	fw.fd = fw.nextFd
 	fw.filePath = fw.nextFilePath
 	fw.fileOffset.Store(0) // Reset offset for new file
 
-	// Clear next file fields
 	fw.nextFile = nil
 	fw.nextFd = 0
 	fw.nextFilePath = ""
 
+	metric.Incr(MetricFileWriterRotationCount, fw.metricTags)
+
 	return nil
 }