// Package logging provides request logging functionality for the CLI Proxy API server. // This file contains metrics integration for the FileStreamingLogWriter. package logging import ( "os" "path/filepath" "sync/atomic" "time" "github.com/router-for-me/CLIProxyAPI/v6/internal/domain/ports" ) // MetricsEnabledFileStreamingLogWriter wraps FileStreamingLogWriter with metrics collection type MetricsEnabledFileStreamingLogWriter struct { *FileStreamingLogWriter metrics ports.MetricsService dropCount atomic.Uint64 startTime time.Time bytesWritten atomic.Int64 } // NewMetricsEnabledFileStreamingLogWriter creates a new metrics-enabled streaming log writer func NewMetricsEnabledFileStreamingLogWriter( base *FileStreamingLogWriter, metrics ports.MetricsService, ) *MetricsEnabledFileStreamingLogWriter { if metrics == nil { metrics = &ports.NoOpMetricsService{} } writer := &MetricsEnabledFileStreamingLogWriter{ FileStreamingLogWriter: base, metrics: metrics, startTime: time.Now(), } // Report initial queue depth if base != nil && base.chunkChan != nil { metrics.RecordLogQueueDepth(len(base.chunkChan)) } return writer } // WriteChunkAsync writes a response chunk asynchronously with metrics tracking func (w *MetricsEnabledFileStreamingLogWriter) WriteChunkAsync(chunk []byte) { if w.FileStreamingLogWriter == nil { return } // Track queue depth before write if w.chunkChan != nil { w.metrics.RecordLogQueueDepth(len(w.chunkChan)) } // Try to write to channel select { case w.chunkChan <- chunk: // Success - record bytes w.bytesWritten.Add(int64(len(chunk))) default: // Channel full - record drop w.dropCount.Add(1) w.metrics.RecordLogDropCount(1) } // Update queue depth after write if w.chunkChan != nil { w.metrics.RecordLogQueueDepth(len(w.chunkChan)) } } // WriteStatus writes the response status with metrics tracking func (w *MetricsEnabledFileStreamingLogWriter) WriteStatus(status int, headers map[string][]string) error { if w.FileStreamingLogWriter == nil { return nil } return w.FileStreamingLogWriter.WriteStatus(status, headers) } // WriteAPIRequest writes the API request with metrics tracking func (w *MetricsEnabledFileStreamingLogWriter) WriteAPIRequest(apiRequest []byte) error { if w.FileStreamingLogWriter == nil { return nil } if len(apiRequest) > 0 { w.bytesWritten.Add(int64(len(apiRequest))) } return w.FileStreamingLogWriter.WriteAPIRequest(apiRequest) } // WriteAPIResponse writes the API response with metrics tracking func (w *MetricsEnabledFileStreamingLogWriter) WriteAPIResponse(apiResponse []byte) error { if w.FileStreamingLogWriter == nil { return nil } if len(apiResponse) > 0 { w.bytesWritten.Add(int64(len(apiResponse))) } return w.FileStreamingLogWriter.WriteAPIResponse(apiResponse) } // Close finalizes the log and records final metrics func (w *MetricsEnabledFileStreamingLogWriter) Close() error { if w.FileStreamingLogWriter == nil { return nil } // Record processing latency duration := time.Since(w.startTime) w.metrics.RecordLogProcessingLatency(duration) // Record total bytes written w.metrics.RecordLogWrite(w.bytesWritten.Load()) // Record any drops drops := w.dropCount.Load() if drops > 0 { w.metrics.RecordLogDropCount(drops) } // Report final queue depth (should be 0 or close to it) if w.chunkChan != nil { w.metrics.RecordLogQueueDepth(len(w.chunkChan)) } return w.FileStreamingLogWriter.Close() } // GetDropCount returns the number of dropped chunks func (w *MetricsEnabledFileStreamingLogWriter) GetDropCount() uint64 { return w.dropCount.Load() } // GetBytesWritten returns the total bytes written func (w *MetricsEnabledFileStreamingLogWriter) GetBytesWritten() int64 { return w.bytesWritten.Load() } // GetProcessingDuration returns the processing duration func (w *MetricsEnabledFileStreamingLogWriter) GetProcessingDuration() time.Duration { return time.Since(w.startTime) } // FileRequestLoggerWithMetrics wraps FileRequestLogger with metrics support type FileRequestLoggerWithMetrics struct { *FileRequestLogger metrics ports.MetricsService } // NewFileRequestLoggerWithMetrics creates a new FileRequestLogger with metrics support func NewFileRequestLoggerWithMetrics( base *FileRequestLogger, metrics ports.MetricsService, ) *FileRequestLoggerWithMetrics { if metrics == nil { metrics = &ports.NoOpMetricsService{} } return &FileRequestLoggerWithMetrics{ FileRequestLogger: base, metrics: metrics, } } // LogStreamingRequest creates a metrics-enabled streaming log writer func (l *FileRequestLoggerWithMetrics) LogStreamingRequest( url, method string, headers map[string][]string, body []byte, requestID string, ) (StreamingLogWriter, error) { if !l.enabled { return &NoOpStreamingLogWriter{}, nil } // Create the base writer using the parent implementation baseWriter, err := l.createBaseStreamingWriter(url, method, headers, body, requestID) if err != nil { l.metrics.RecordLogError(err) return nil, err } // Wrap with metrics return NewMetricsEnabledFileStreamingLogWriter(baseWriter, l.metrics), nil } // createBaseStreamingWriter creates the underlying FileStreamingLogWriter func (l *FileRequestLoggerWithMetrics) createBaseStreamingWriter( url, method string, headers map[string][]string, body []byte, requestID string, ) (*FileStreamingLogWriter, error) { // This mirrors the logic in FileRequestLogger.LogStreamingRequest if err := l.ensureLogsDir(); err != nil { return nil, err } filename := l.generateFilename(url, requestID) filePath := filepath.Join(l.logsDir, filename) requestHeaders := make(map[string][]string, len(headers)) for key, values := range headers { headerValues := make([]string, len(values)) copy(headerValues, values) requestHeaders[key] = headerValues } requestBodyPath, errTemp := l.writeRequestBodyTempFile(body) if errTemp != nil { return nil, errTemp } responseBodyFile, errCreate := os.CreateTemp(l.logsDir, "response-body-*.tmp") if errCreate != nil { _ = os.Remove(requestBodyPath) return nil, errCreate } responseBodyPath := responseBodyFile.Name() writer := &FileStreamingLogWriter{ logFilePath: filePath, url: url, method: method, timestamp: time.Now(), requestHeaders: requestHeaders, requestBodyPath: requestBodyPath, responseBodyPath: responseBodyPath, responseBodyFile: responseBodyFile, chunkChan: make(chan []byte, 100), closeChan: make(chan struct{}), errorChan: make(chan error, 1), } go writer.asyncWriter() return writer, nil } // GetMetrics returns the current metrics func (l *FileRequestLoggerWithMetrics) GetMetrics() *ports.LogMetrics { if l.metrics != nil { return l.metrics.GetMetrics() } return &ports.LogMetrics{} } // MetricsCollectorAdapter provides a callback-based interface for log writers type MetricsCollectorAdapter struct { metrics ports.MetricsService } // NewMetricsCollectorAdapter creates a new adapter for the given metrics service func NewMetricsCollectorAdapter(metrics ports.MetricsService) *MetricsCollectorAdapter { if metrics == nil { metrics = &ports.NoOpMetricsService{} } return &MetricsCollectorAdapter{metrics: metrics} } // OnQueueDepthChanged records queue depth changes func (a *MetricsCollectorAdapter) OnQueueDepthChanged(depth int) { a.metrics.RecordLogQueueDepth(depth) } // OnProcessingCompleted records processing completion func (a *MetricsCollectorAdapter) OnProcessingCompleted(duration time.Duration, bytesProcessed int64, err error) { a.metrics.RecordLogProcessingLatency(duration) a.metrics.RecordLogWrite(bytesProcessed) if err != nil { a.metrics.RecordLogError(err) } } // OnItemsDropped records dropped items func (a *MetricsCollectorAdapter) OnItemsDropped(count uint64) { a.metrics.RecordLogDropCount(count) } // Ensure MetricsCollectorAdapter implements the interface var _ ports.MetricsCollector = (*MetricsCollectorAdapter)(nil)