diff --git a/config.example.yaml b/config.example.yaml index bb9307cc..4b30dd88 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -49,7 +49,7 @@ pprof: enable: false addr: "127.0.0.1:8316" -# When true, disable high-overhead HTTP middleware features to reduce per-request memory usage under high concurrency. +# When true, disable high-overhead request logging and HTTP middleware features to reduce per-request memory usage under high concurrency. commercial-mode: false # When true, write application logs to rotating files instead of stdout diff --git a/internal/api/middleware/request_logging.go b/internal/api/middleware/request_logging.go index 561219c4..0ee849ae 100644 --- a/internal/api/middleware/request_logging.go +++ b/internal/api/middleware/request_logging.go @@ -58,7 +58,7 @@ func RequestLoggingMiddleware(logger logging.RequestLogger) gin.HandlerFunc { wrapper.logOnErrorOnly = true } c.Writer = wrapper - attachWebsocketLogSources(c, logger, loggerEnabled) + attachRequestLogSources(c, logger, loggerEnabled) // Process the request c.Next() @@ -75,14 +75,23 @@ type fileBodySourceFactory interface { NewFileBodySource(prefix string) (*logging.FileBodySource, error) } -func attachWebsocketLogSources(c *gin.Context, logger logging.RequestLogger, loggerEnabled bool) { - if c == nil || !loggerEnabled || !isResponsesWebsocketUpgrade(c.Request) { +func attachRequestLogSources(c *gin.Context, logger logging.RequestLogger, loggerEnabled bool) { + if c == nil || !loggerEnabled { return } factory, ok := logger.(fileBodySourceFactory) if !ok || factory == nil { return } + if source, errSource := factory.NewFileBodySource("api-request"); errSource == nil { + c.Set(logging.APIRequestSourceContextKey, source) + } + if source, errSource := factory.NewFileBodySource("api-response"); errSource == nil { + c.Set(logging.APIResponseSourceContextKey, source) + } + if !isResponsesWebsocketUpgrade(c.Request) { + return + } if source, errSource := factory.NewFileBodySource("websocket-timeline"); errSource == nil { c.Set(logging.WebsocketTimelineSourceContextKey, source) } diff --git a/internal/api/middleware/request_logging_test.go b/internal/api/middleware/request_logging_test.go index c64b844a..ed1be2e0 100644 --- a/internal/api/middleware/request_logging_test.go +++ b/internal/api/middleware/request_logging_test.go @@ -144,7 +144,7 @@ func TestShouldCaptureRequestBody(t *testing.T) { } } -func TestAttachWebsocketLogSourcesUsesLoggerLogsDir(t *testing.T) { +func TestAttachRequestLogSourcesUsesLoggerLogsDir(t *testing.T) { gin.SetMode(gin.TestMode) logsDir := t.TempDir() @@ -154,7 +154,7 @@ func TestAttachWebsocketLogSourcesUsesLoggerLogsDir(t *testing.T) { c.Request = httptest.NewRequest(http.MethodGet, "/v1/responses", nil) c.Request.Header.Set("Upgrade", "websocket") - attachWebsocketLogSources(c, logger, true) + attachRequestLogSources(c, logger, true) defer cleanupFileBodySourcesFromContext(c) for _, key := range []string{ diff --git a/internal/api/middleware/response_writer.go b/internal/api/middleware/response_writer.go index 5eabd08d..aedce47c 100644 --- a/internal/api/middleware/response_writer.go +++ b/internal/api/middleware/response_writer.go @@ -282,9 +282,11 @@ func (w *ResponseWriterWrapper) Finalize(c *gin.Context) error { hasAPIError := len(slicesAPIResponseError) > 0 || finalStatusCode >= http.StatusBadRequest forceLog := w.logOnErrorOnly && hasAPIError && !w.logger.IsEnabled() websocketTimelineSource := w.extractWebsocketTimelineSource(c) + apiRequestSource := w.extractAPIRequestSource(c) + apiResponseSource := w.extractAPIResponseSource(c) apiWebsocketTimelineSource := w.extractAPIWebsocketTimelineSource(c) if !w.logger.IsEnabled() && !forceLog { - cleanupFileBodySources(websocketTimelineSource, apiWebsocketTimelineSource) + cleanupFileBodySources(websocketTimelineSource, apiRequestSource, apiResponseSource, apiWebsocketTimelineSource) return nil } @@ -303,33 +305,63 @@ func (w *ResponseWriterWrapper) Finalize(c *gin.Context) error { // Write API Request and Response to the streaming log before closing apiRequest := w.extractAPIRequest(c) - if len(apiRequest) > 0 { - _ = w.streamWriter.WriteAPIRequest(apiRequest) - } apiResponse := w.extractAPIResponse(c) - if len(apiResponse) > 0 { - _ = w.streamWriter.WriteAPIResponse(apiResponse) + if sourceWriter, ok := w.streamWriter.(interface { + WriteAPIRequestSource(*logging.FileBodySource) error + WriteAPIResponseSource(*logging.FileBodySource) error + }); ok { + if len(apiRequest) > 0 { + _ = w.streamWriter.WriteAPIRequest(apiRequest) + } + if apiRequestSource != nil && apiRequestSource.HasPayload() { + _ = sourceWriter.WriteAPIRequestSource(apiRequestSource) + } + if len(apiResponse) > 0 { + _ = w.streamWriter.WriteAPIResponse(apiResponse) + } + if apiResponseSource != nil && apiResponseSource.HasPayload() { + _ = sourceWriter.WriteAPIResponseSource(apiResponseSource) + } + } else { + var errMerge error + apiRequest, errMerge = mergeFileBodySource(apiRequest, apiRequestSource) + if errMerge != nil { + cleanupFileBodySources(websocketTimelineSource, apiResponseSource, apiWebsocketTimelineSource) + return errMerge + } + apiResponse, errMerge = mergeFileBodySource(apiResponse, apiResponseSource) + if errMerge != nil { + cleanupFileBodySources(websocketTimelineSource, apiWebsocketTimelineSource) + return errMerge + } + if len(apiRequest) > 0 { + _ = w.streamWriter.WriteAPIRequest(apiRequest) + } + if len(apiResponse) > 0 { + _ = w.streamWriter.WriteAPIResponse(apiResponse) + } } apiWebsocketTimeline := w.extractAPIWebsocketTimeline(c) var errMerge error apiWebsocketTimeline, errMerge = mergeFileBodySource(apiWebsocketTimeline, apiWebsocketTimelineSource) if errMerge != nil { - cleanupFileBodySources(websocketTimelineSource) + cleanupFileBodySources(websocketTimelineSource, apiRequestSource, apiResponseSource) return errMerge } - cleanupFileBodySources(websocketTimelineSource) if len(apiWebsocketTimeline) > 0 { _ = w.streamWriter.WriteAPIWebsocketTimeline(apiWebsocketTimeline) } if err := w.streamWriter.Close(); err != nil { w.streamWriter = nil + cleanupFileBodySources(websocketTimelineSource, apiRequestSource, apiResponseSource) return err } w.streamWriter = nil + cleanupFileBodySources(websocketTimelineSource, apiRequestSource, apiResponseSource) return nil } - return w.logRequest(w.extractRequestBody(c), finalStatusCode, w.cloneHeaders(), w.extractResponseBody(c), w.extractWebsocketTimeline(c), websocketTimelineSource, w.extractAPIRequest(c), w.extractAPIResponse(c), w.extractAPIWebsocketTimeline(c), apiWebsocketTimelineSource, w.extractAPIResponseTimestamp(c), slicesAPIResponseError, forceLog) + return w.logRequest(w.extractRequestBody(c), finalStatusCode, w.cloneHeaders(), w.extractResponseBody(c), w.extractWebsocketTimeline(c), websocketTimelineSource, w.extractAPIRequest(c), apiRequestSource, w.extractAPIResponse(c), apiResponseSource, w.extractAPIWebsocketTimeline(c), apiWebsocketTimelineSource, w.extractAPIResponseTimestamp(c), slicesAPIResponseError, forceLog) } func (w *ResponseWriterWrapper) cloneHeaders() map[string][]string { @@ -369,6 +401,14 @@ func (w *ResponseWriterWrapper) extractAPIResponse(c *gin.Context) []byte { return data } +func (w *ResponseWriterWrapper) extractAPIRequestSource(c *gin.Context) *logging.FileBodySource { + return extractFileBodySource(c, logging.APIRequestSourceContextKey) +} + +func (w *ResponseWriterWrapper) extractAPIResponseSource(c *gin.Context) *logging.FileBodySource { + return extractFileBodySource(c, logging.APIResponseSourceContextKey) +} + func (w *ResponseWriterWrapper) extractAPIWebsocketTimeline(c *gin.Context) []byte { apiTimeline, isExist := c.Get("API_WEBSOCKET_TIMELINE") if !isExist { @@ -460,15 +500,53 @@ func extractBodyOverride(c *gin.Context, key string) []byte { return nil } -func (w *ResponseWriterWrapper) logRequest(requestBody []byte, statusCode int, headers map[string][]string, body, websocketTimeline []byte, websocketTimelineSource *logging.FileBodySource, apiRequestBody, apiResponseBody, apiWebsocketTimeline []byte, apiWebsocketTimelineSource *logging.FileBodySource, apiResponseTimestamp time.Time, apiResponseErrors []*interfaces.ErrorMessage, forceLog bool) error { +func (w *ResponseWriterWrapper) logRequest(requestBody []byte, statusCode int, headers map[string][]string, body, websocketTimeline []byte, websocketTimelineSource *logging.FileBodySource, apiRequestBody []byte, apiRequestSource *logging.FileBodySource, apiResponseBody []byte, apiResponseSource *logging.FileBodySource, apiWebsocketTimeline []byte, apiWebsocketTimelineSource *logging.FileBodySource, apiResponseTimestamp time.Time, apiResponseErrors []*interfaces.ErrorMessage, forceLog bool) error { if w.requestInfo == nil { - cleanupFileBodySources(websocketTimelineSource, apiWebsocketTimelineSource) + cleanupFileBodySources(websocketTimelineSource, apiRequestSource, apiResponseSource, apiWebsocketTimelineSource) return nil } + if loggerWithAllSources, ok := w.logger.(interface { + LogRequestWithOptionsAndAllSources(string, string, map[string][]string, []byte, int, map[string][]string, []byte, []byte, *logging.FileBodySource, []byte, *logging.FileBodySource, []byte, *logging.FileBodySource, []byte, *logging.FileBodySource, []*interfaces.ErrorMessage, bool, string, time.Time, time.Time) error + }); ok { + return loggerWithAllSources.LogRequestWithOptionsAndAllSources( + w.requestInfo.URL, + w.requestInfo.Method, + w.requestInfo.Headers, + requestBody, + statusCode, + headers, + body, + websocketTimeline, + websocketTimelineSource, + apiRequestBody, + apiRequestSource, + apiResponseBody, + apiResponseSource, + apiWebsocketTimeline, + apiWebsocketTimelineSource, + apiResponseErrors, + forceLog, + w.requestInfo.RequestID, + w.requestInfo.Timestamp, + apiResponseTimestamp, + ) + } + if loggerWithSources, ok := w.logger.(interface { LogRequestWithOptionsAndSources(string, string, map[string][]string, []byte, int, map[string][]string, []byte, []byte, *logging.FileBodySource, []byte, []byte, []byte, *logging.FileBodySource, []*interfaces.ErrorMessage, bool, string, time.Time, time.Time) error }); ok { + var errMerge error + apiRequestBody, errMerge = mergeFileBodySource(apiRequestBody, apiRequestSource) + if errMerge != nil { + cleanupFileBodySources(websocketTimelineSource, apiResponseSource, apiWebsocketTimelineSource) + return errMerge + } + apiResponseBody, errMerge = mergeFileBodySource(apiResponseBody, apiResponseSource) + if errMerge != nil { + cleanupFileBodySources(websocketTimelineSource, apiWebsocketTimelineSource) + return errMerge + } return loggerWithSources.LogRequestWithOptionsAndSources( w.requestInfo.URL, w.requestInfo.Method, @@ -493,6 +571,16 @@ func (w *ResponseWriterWrapper) logRequest(requestBody []byte, statusCode int, h var errMerge error websocketTimeline, errMerge = mergeFileBodySource(websocketTimeline, websocketTimelineSource) + if errMerge != nil { + cleanupFileBodySources(apiRequestSource, apiResponseSource, apiWebsocketTimelineSource) + return errMerge + } + apiRequestBody, errMerge = mergeFileBodySource(apiRequestBody, apiRequestSource) + if errMerge != nil { + cleanupFileBodySources(apiResponseSource, apiWebsocketTimelineSource) + return errMerge + } + apiResponseBody, errMerge = mergeFileBodySource(apiResponseBody, apiResponseSource) if errMerge != nil { cleanupFileBodySources(apiWebsocketTimelineSource) return errMerge diff --git a/internal/api/server.go b/internal/api/server.go index 05bcd1cf..e81ca670 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -72,6 +72,17 @@ func defaultRequestLoggerFactory(cfg *config.Config, configPath string) logging. return logger } +func effectiveSDKConfig(cfg *config.Config) *config.SDKConfig { + if cfg == nil { + return nil + } + sdkCfg := cfg.SDKConfig + if cfg.CommercialMode { + sdkCfg.RequestLog = false + } + return &sdkCfg +} + // WithMiddleware appends additional Gin middleware during server construction. func WithMiddleware(mw ...gin.HandlerFunc) ServerOption { return func(cfg *serverOptionConfig) { @@ -257,7 +268,7 @@ func NewServer(cfg *config.Config, authManager *auth.Manager, accessManager *sdk // Create server instance s := &Server{ engine: engine, - handlers: handlers.NewBaseAPIHandlers(&cfg.SDKConfig, authManager), + handlers: handlers.NewBaseAPIHandlers(effectiveSDKConfig(cfg), authManager), cfg: cfg, accessManager: accessManager, requestLogger: requestLogger, @@ -1453,7 +1464,7 @@ func (s *Server) UpdateClients(cfg *config.Config) { // Save YAML snapshot for next comparison s.oldConfigYaml, _ = yaml.Marshal(cfg) - s.handlers.UpdateClients(&cfg.SDKConfig) + s.handlers.UpdateClients(effectiveSDKConfig(cfg)) if s.mgmt != nil { s.mgmt.SetConfig(cfg) diff --git a/internal/config/config.go b/internal/config/config.go index 0e193938..d0a59973 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -52,7 +52,7 @@ type Config struct { // Pprof config controls the optional pprof HTTP debug server. Pprof PprofConfig `yaml:"pprof" json:"pprof"` - // CommercialMode disables high-overhead HTTP middleware features to minimize per-request memory usage. + // CommercialMode disables high-overhead request logging and HTTP middleware features to minimize per-request memory usage. CommercialMode bool `yaml:"commercial-mode" json:"commercial-mode"` // LoggingToFile controls whether application logs are written to rotating files or stdout. diff --git a/internal/logging/request_logger.go b/internal/logging/request_logger.go index e1c7a9cc..9a21e7e0 100644 --- a/internal/logging/request_logger.go +++ b/internal/logging/request_logger.go @@ -35,6 +35,9 @@ var requestLogID atomic.Uint64 const ( WebsocketTimelineSourceContextKey = "WEBSOCKET_TIMELINE_SOURCE" + APIRequestSourceContextKey = "API_REQUEST_SOURCE" + APIResponseSourceContextKey = "API_RESPONSE_SOURCE" + APIResponseCapturedContextKey = "API_RESPONSE_CAPTURED" APIWebsocketTimelineSourceContextKey = "API_WEBSOCKET_TIMELINE_SOURCE" ) @@ -140,6 +143,46 @@ func (s *FileBodySource) AppendPart(data []byte) error { return writeErr } +// AppendBytes appends raw bytes to a single ordered part. +func (s *FileBodySource) AppendBytes(data []byte) error { + if s == nil { + return fmt.Errorf("file body source is nil") + } + if len(data) == 0 { + return nil + } + s.mu.Lock() + defer s.mu.Unlock() + if s.cleaned { + return fmt.Errorf("file body source has been cleaned") + } + if errMkdir := os.MkdirAll(s.dir, 0755); errMkdir != nil { + return errMkdir + } + + var file *os.File + var errOpen error + if len(s.paths) == 0 { + file, errOpen = os.CreateTemp(s.dir, "part-*.tmp") + if errOpen == nil { + s.paths = append(s.paths, file.Name()) + } + } else { + file, errOpen = os.OpenFile(s.paths[len(s.paths)-1], os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644) + } + if errOpen != nil { + return errOpen + } + + _, writeErr := file.Write(data) + if errClose := file.Close(); errClose != nil { + if writeErr == nil { + writeErr = errClose + } + } + return writeErr +} + // HasPayload reports whether any detail parts were recorded. func (s *FileBodySource) HasPayload() bool { if s == nil { @@ -520,20 +563,25 @@ func (l *FileRequestLogger) LogRequest(url, method string, requestHeaders map[st // LogRequestWithOptions logs a request with optional forced logging behavior. // The force flag allows writing error logs even when regular request logging is disabled. func (l *FileRequestLogger) LogRequestWithOptions(url, method string, requestHeaders map[string][]string, body []byte, statusCode int, responseHeaders map[string][]string, response, websocketTimeline, apiRequest, apiResponse, apiWebsocketTimeline []byte, apiResponseErrors []*interfaces.ErrorMessage, force bool, requestID string, requestTimestamp, apiResponseTimestamp time.Time) error { - return l.logRequestWithSources(url, method, requestHeaders, body, statusCode, responseHeaders, response, websocketTimeline, nil, apiRequest, apiResponse, apiWebsocketTimeline, nil, apiResponseErrors, force, requestID, requestTimestamp, apiResponseTimestamp) + return l.logRequestWithSources(url, method, requestHeaders, body, statusCode, responseHeaders, response, websocketTimeline, nil, apiRequest, nil, apiResponse, nil, apiWebsocketTimeline, nil, apiResponseErrors, force, requestID, requestTimestamp, apiResponseTimestamp) } func (l *FileRequestLogger) logRequest(url, method string, requestHeaders map[string][]string, body []byte, statusCode int, responseHeaders map[string][]string, response, websocketTimeline, apiRequest, apiResponse, apiWebsocketTimeline []byte, apiResponseErrors []*interfaces.ErrorMessage, force bool, requestID string, requestTimestamp, apiResponseTimestamp time.Time) error { - return l.logRequestWithSources(url, method, requestHeaders, body, statusCode, responseHeaders, response, websocketTimeline, nil, apiRequest, apiResponse, apiWebsocketTimeline, nil, apiResponseErrors, force, requestID, requestTimestamp, apiResponseTimestamp) + return l.logRequestWithSources(url, method, requestHeaders, body, statusCode, responseHeaders, response, websocketTimeline, nil, apiRequest, nil, apiResponse, nil, apiWebsocketTimeline, nil, apiResponseErrors, force, requestID, requestTimestamp, apiResponseTimestamp) } // LogRequestWithOptionsAndSources logs a request with optional file-backed large sections. func (l *FileRequestLogger) LogRequestWithOptionsAndSources(url, method string, requestHeaders map[string][]string, body []byte, statusCode int, responseHeaders map[string][]string, response, websocketTimeline []byte, websocketTimelineSource *FileBodySource, apiRequest, apiResponse, apiWebsocketTimeline []byte, apiWebsocketTimelineSource *FileBodySource, apiResponseErrors []*interfaces.ErrorMessage, force bool, requestID string, requestTimestamp, apiResponseTimestamp time.Time) error { - return l.logRequestWithSources(url, method, requestHeaders, body, statusCode, responseHeaders, response, websocketTimeline, websocketTimelineSource, apiRequest, apiResponse, apiWebsocketTimeline, apiWebsocketTimelineSource, apiResponseErrors, force, requestID, requestTimestamp, apiResponseTimestamp) + return l.logRequestWithSources(url, method, requestHeaders, body, statusCode, responseHeaders, response, websocketTimeline, websocketTimelineSource, apiRequest, nil, apiResponse, nil, apiWebsocketTimeline, apiWebsocketTimelineSource, apiResponseErrors, force, requestID, requestTimestamp, apiResponseTimestamp) } -func (l *FileRequestLogger) logRequestWithSources(url, method string, requestHeaders map[string][]string, body []byte, statusCode int, responseHeaders map[string][]string, response, websocketTimeline []byte, websocketTimelineSource *FileBodySource, apiRequest, apiResponse, apiWebsocketTimeline []byte, apiWebsocketTimelineSource *FileBodySource, apiResponseErrors []*interfaces.ErrorMessage, force bool, requestID string, requestTimestamp, apiResponseTimestamp time.Time) error { - defer cleanupFileBodySources(websocketTimelineSource, apiWebsocketTimelineSource) +// LogRequestWithOptionsAndAllSources logs a request with optional file-backed request and response sections. +func (l *FileRequestLogger) LogRequestWithOptionsAndAllSources(url, method string, requestHeaders map[string][]string, body []byte, statusCode int, responseHeaders map[string][]string, response, websocketTimeline []byte, websocketTimelineSource *FileBodySource, apiRequest []byte, apiRequestSource *FileBodySource, apiResponse []byte, apiResponseSource *FileBodySource, apiWebsocketTimeline []byte, apiWebsocketTimelineSource *FileBodySource, apiResponseErrors []*interfaces.ErrorMessage, force bool, requestID string, requestTimestamp, apiResponseTimestamp time.Time) error { + return l.logRequestWithSources(url, method, requestHeaders, body, statusCode, responseHeaders, response, websocketTimeline, websocketTimelineSource, apiRequest, apiRequestSource, apiResponse, apiResponseSource, apiWebsocketTimeline, apiWebsocketTimelineSource, apiResponseErrors, force, requestID, requestTimestamp, apiResponseTimestamp) +} + +func (l *FileRequestLogger) logRequestWithSources(url, method string, requestHeaders map[string][]string, body []byte, statusCode int, responseHeaders map[string][]string, response, websocketTimeline []byte, websocketTimelineSource *FileBodySource, apiRequest []byte, apiRequestSource *FileBodySource, apiResponse []byte, apiResponseSource *FileBodySource, apiWebsocketTimeline []byte, apiWebsocketTimelineSource *FileBodySource, apiResponseErrors []*interfaces.ErrorMessage, force bool, requestID string, requestTimestamp, apiResponseTimestamp time.Time) error { + defer cleanupFileBodySources(websocketTimelineSource, apiRequestSource, apiResponseSource, apiWebsocketTimelineSource) if !l.enabled && !force { return nil @@ -556,7 +604,9 @@ func (l *FileRequestLogger) logRequestWithSources(url, method string, requestHea websocketTimeline, websocketTimelineSource, apiRequest, + apiRequestSource, apiResponse, + apiResponseSource, apiWebsocketTimeline, apiWebsocketTimelineSource, apiResponseErrors, @@ -618,7 +668,9 @@ func (l *FileRequestLogger) logRequestWithSources(url, method string, requestHea websocketTimeline, websocketTimelineSource, apiRequest, + apiRequestSource, apiResponse, + apiResponseSource, apiWebsocketTimeline, apiWebsocketTimelineSource, apiResponseErrors, @@ -888,7 +940,9 @@ func (l *FileRequestLogger) writeNonStreamingLog( websocketTimeline []byte, websocketTimelineSource *FileBodySource, apiRequest []byte, + apiRequestSource *FileBodySource, apiResponse []byte, + apiResponseSource *FileBodySource, apiWebsocketTimeline []byte, apiWebsocketTimelineSource *FileBodySource, apiResponseErrors []*interfaces.ErrorMessage, @@ -904,7 +958,7 @@ func (l *FileRequestLogger) writeNonStreamingLog( } isWebsocketTranscript := hasSectionPayload(websocketTimeline) || hasFileBodySourcePayload(websocketTimelineSource) downstreamTransport := inferDownstreamTransport(requestHeaders, websocketTimeline, websocketTimelineSource) - upstreamTransport := inferUpstreamTransport(apiRequest, apiResponse, apiWebsocketTimeline, apiWebsocketTimelineSource, apiResponseErrors) + upstreamTransport := inferUpstreamTransport(apiRequest, apiRequestSource, apiResponse, apiResponseSource, apiWebsocketTimeline, apiWebsocketTimelineSource, apiResponseErrors) if errWrite := writeRequestInfoWithBody(w, url, method, requestHeaders, requestBody, requestBodyPath, requestTimestamp, downstreamTransport, upstreamTransport, !isWebsocketTranscript); errWrite != nil { return errWrite } @@ -914,13 +968,13 @@ func (l *FileRequestLogger) writeNonStreamingLog( if errWrite := writeAPISectionWithSource(w, "=== API WEBSOCKET TIMELINE ===\n", "=== API WEBSOCKET TIMELINE", apiWebsocketTimeline, apiWebsocketTimelineSource, time.Time{}); errWrite != nil { return errWrite } - if errWrite := writeAPISection(w, "=== API REQUEST ===\n", "=== API REQUEST", apiRequest, time.Time{}); errWrite != nil { + if errWrite := writePreformattedAPISectionWithSource(w, "=== API REQUEST ===\n", "=== API REQUEST", apiRequest, apiRequestSource, time.Time{}); errWrite != nil { return errWrite } if errWrite := writeAPIErrorResponses(w, apiResponseErrors); errWrite != nil { return errWrite } - if errWrite := writeAPISection(w, "=== API RESPONSE ===\n", "=== API RESPONSE", apiResponse, apiResponseTimestamp); errWrite != nil { + if errWrite := writePreformattedAPISectionWithSource(w, "=== API RESPONSE ===\n", "=== API RESPONSE", apiResponse, apiResponseSource, apiResponseTimestamp); errWrite != nil { return errWrite } if isWebsocketTranscript { @@ -1087,8 +1141,8 @@ func inferDownstreamTransport(headers map[string][]string, websocketTimeline []b return "http" } -func inferUpstreamTransport(apiRequest, apiResponse, apiWebsocketTimeline []byte, apiWebsocketTimelineSource *FileBodySource, _ []*interfaces.ErrorMessage) string { - hasHTTP := hasSectionPayload(apiRequest) || hasSectionPayload(apiResponse) +func inferUpstreamTransport(apiRequest []byte, apiRequestSource *FileBodySource, apiResponse []byte, apiResponseSource *FileBodySource, apiWebsocketTimeline []byte, apiWebsocketTimelineSource *FileBodySource, _ []*interfaces.ErrorMessage) string { + hasHTTP := hasSectionPayload(apiRequest) || hasFileBodySourcePayload(apiRequestSource) || hasSectionPayload(apiResponse) || hasFileBodySourcePayload(apiResponseSource) hasWS := hasSectionPayload(apiWebsocketTimeline) || hasFileBodySourcePayload(apiWebsocketTimelineSource) switch { case hasHTTP && hasWS: @@ -1178,6 +1232,25 @@ func writeAPISectionWithSource(w io.Writer, sectionHeader string, sectionPrefix return nil } +func writePreformattedAPISectionWithSource(w io.Writer, sectionHeader string, sectionPrefix string, payload []byte, source *FileBodySource, timestamp time.Time) error { + if !hasFileBodySourcePayload(source) { + return writeAPISection(w, sectionHeader, sectionPrefix, payload, timestamp) + } + if len(payload) > 0 { + if errWrite := writeAPISection(w, sectionHeader, sectionPrefix, payload, timestamp); errWrite != nil { + return errWrite + } + } + tracker := &trailingNewlineTrackingWriter{writer: w} + if errWrite := source.WriteTo(tracker); errWrite != nil { + return errWrite + } + if errWrite := writeSectionSpacing(w, tracker.trailingNewlines); errWrite != nil { + return errWrite + } + return nil +} + func writeAPIErrorResponses(w io.Writer, apiResponseErrors []*interfaces.ErrorMessage) error { for i := 0; i < len(apiResponseErrors); i++ { if apiResponseErrors[i] == nil { @@ -1288,7 +1361,7 @@ func (l *FileRequestLogger) formatLogContent(url, method string, headers map[str var content strings.Builder isWebsocketTranscript := hasSectionPayload(websocketTimeline) downstreamTransport := inferDownstreamTransport(headers, websocketTimeline, nil) - upstreamTransport := inferUpstreamTransport(apiRequest, apiResponse, apiWebsocketTimeline, nil, apiResponseErrors) + upstreamTransport := inferUpstreamTransport(apiRequest, nil, apiResponse, nil, apiWebsocketTimeline, nil, apiResponseErrors) // Request info content.WriteString(l.formatRequestInfo(url, method, headers, body, downstreamTransport, upstreamTransport, !isWebsocketTranscript)) @@ -1607,9 +1680,15 @@ type FileStreamingLogWriter struct { // apiRequest stores the upstream API request data. apiRequest []byte + // apiRequestSource stores file-backed upstream API request data. + apiRequestSource *FileBodySource + // apiResponse stores the upstream API response data. apiResponse []byte + // apiResponseSource stores file-backed upstream API response data. + apiResponseSource *FileBodySource + // apiWebsocketTimeline stores the upstream websocket event timeline. apiWebsocketTimeline []byte @@ -1679,6 +1758,15 @@ func (w *FileStreamingLogWriter) WriteAPIRequest(apiRequest []byte) error { return nil } +// WriteAPIRequestSource buffers a file-backed upstream API request for final writing. +func (w *FileStreamingLogWriter) WriteAPIRequestSource(apiRequestSource *FileBodySource) error { + if apiRequestSource == nil || !apiRequestSource.HasPayload() { + return nil + } + w.apiRequestSource = apiRequestSource + return nil +} + // WriteAPIResponse buffers the upstream API response details for later writing. // // Parameters: @@ -1694,6 +1782,15 @@ func (w *FileStreamingLogWriter) WriteAPIResponse(apiResponse []byte) error { return nil } +// WriteAPIResponseSource buffers a file-backed upstream API response for final writing. +func (w *FileStreamingLogWriter) WriteAPIResponseSource(apiResponseSource *FileBodySource) error { + if apiResponseSource == nil || !apiResponseSource.HasPayload() { + return nil + } + w.apiResponseSource = apiResponseSource + return nil +} + // WriteAPIWebsocketTimeline buffers the upstream websocket timeline for later writing. // // Parameters: @@ -1799,16 +1896,16 @@ func (w *FileStreamingLogWriter) asyncWriter() { } func (w *FileStreamingLogWriter) writeFinalLog(logFile *os.File) error { - if errWrite := writeRequestInfoWithBody(logFile, w.url, w.method, w.requestHeaders, nil, w.requestBodyPath, w.timestamp, "http", inferUpstreamTransport(w.apiRequest, w.apiResponse, w.apiWebsocketTimeline, nil, nil), true); errWrite != nil { + if errWrite := writeRequestInfoWithBody(logFile, w.url, w.method, w.requestHeaders, nil, w.requestBodyPath, w.timestamp, "http", inferUpstreamTransport(w.apiRequest, w.apiRequestSource, w.apiResponse, w.apiResponseSource, w.apiWebsocketTimeline, nil, nil), true); errWrite != nil { return errWrite } if errWrite := writeAPISection(logFile, "=== API WEBSOCKET TIMELINE ===\n", "=== API WEBSOCKET TIMELINE", w.apiWebsocketTimeline, time.Time{}); errWrite != nil { return errWrite } - if errWrite := writeAPISection(logFile, "=== API REQUEST ===\n", "=== API REQUEST", w.apiRequest, time.Time{}); errWrite != nil { + if errWrite := writePreformattedAPISectionWithSource(logFile, "=== API REQUEST ===\n", "=== API REQUEST", w.apiRequest, w.apiRequestSource, time.Time{}); errWrite != nil { return errWrite } - if errWrite := writeAPISection(logFile, "=== API RESPONSE ===\n", "=== API RESPONSE", w.apiResponse, w.apiResponseTimestamp); errWrite != nil { + if errWrite := writePreformattedAPISectionWithSource(logFile, "=== API RESPONSE ===\n", "=== API RESPONSE", w.apiResponse, w.apiResponseSource, w.apiResponseTimestamp); errWrite != nil { return errWrite } @@ -2040,7 +2137,7 @@ func (w *homeStreamingLogWriter) Close() error { responsePayload := w.responseBody.Bytes() var buf bytes.Buffer - upstreamTransport := inferUpstreamTransport(w.apiRequest, w.apiResponse, w.apiWebsocketTime, nil, nil) + upstreamTransport := inferUpstreamTransport(w.apiRequest, nil, w.apiResponse, nil, w.apiWebsocketTime, nil, nil) if errWrite := writeRequestInfoWithBody(&buf, w.url, w.method, w.requestHeaders, w.requestBody, "", w.timestamp, "http", upstreamTransport, true); errWrite != nil { return errWrite } diff --git a/internal/runtime/executor/helps/logging_helpers.go b/internal/runtime/executor/helps/logging_helpers.go index c3223058..94837d2c 100644 --- a/internal/runtime/executor/helps/logging_helpers.go +++ b/internal/runtime/executor/helps/logging_helpers.go @@ -44,6 +44,7 @@ type upstreamAttempt struct { index int request string response *strings.Builder + responseSource *logging.FileBodySource responseIntroWritten bool statusWritten bool headersWritten bool @@ -53,9 +54,13 @@ type upstreamAttempt struct { errorWritten bool } +func requestLogCaptureEnabled(cfg *config.Config) bool { + return cfg != nil && cfg.RequestLog && !cfg.CommercialMode +} + // RecordAPIRequest stores the upstream request metadata in Gin context for request logging. func RecordAPIRequest(ctx context.Context, cfg *config.Config, info UpstreamRequestLog) { - if cfg == nil || !cfg.RequestLog { + if !requestLogCaptureEnabled(cfg) { return } ginCtx := ginContextFrom(ctx) @@ -83,27 +88,57 @@ func RecordAPIRequest(ctx context.Context, cfg *config.Config, info UpstreamRequ builder.WriteString("\nHeaders:\n") writeHeaders(builder, info.Headers) builder.WriteString("\nBody:\n") - if len(info.Body) > 0 { - builder.WriteString(string(info.Body)) + + requestText := "" + if source, ok := apiRequestSource(ginCtx); ok { + if errWrite := source.AppendBytes([]byte(builder.String())); errWrite == nil { + if len(info.Body) > 0 { + if errBody := source.AppendBytes(info.Body); errBody != nil { + log.WithError(errBody).Warn("failed to append api request body log part") + } + } else if errEmpty := source.AppendBytes([]byte("")); errEmpty != nil { + log.WithError(errEmpty).Warn("failed to append empty api request log part") + } + if errEnd := source.AppendBytes([]byte("\n\n")); errEnd != nil { + log.WithError(errEnd).Warn("failed to append api request log terminator") + } + } else { + log.WithError(errWrite).Warn("failed to append api request log part") + if len(info.Body) > 0 { + builder.WriteString(string(info.Body)) + } else { + builder.WriteString("") + } + builder.WriteString("\n\n") + requestText = builder.String() + } } else { - builder.WriteString("") + if len(info.Body) > 0 { + builder.WriteString(string(info.Body)) + } else { + builder.WriteString("") + } + builder.WriteString("\n\n") + requestText = builder.String() } - builder.WriteString("\n\n") attempt := &upstreamAttempt{ - index: index, - request: builder.String(), - response: &strings.Builder{}, + index: index, + request: requestText, + response: &strings.Builder{}, + responseSource: apiResponseSourceOrNil(ginCtx), } attempts = append(attempts, attempt) ginCtx.Set(apiAttemptsKey, attempts) - updateAggregatedRequest(ginCtx, attempts) + if requestText != "" { + updateAggregatedRequest(ginCtx, attempts) + } } // RecordAPIResponseMetadata captures upstream response status/header information for the latest attempt. func RecordAPIResponseMetadata(ctx context.Context, cfg *config.Config, status int, headers http.Header) { logging.SetResponseHeaders(ctx, headers) - if cfg == nil || !cfg.RequestLog { + if !requestLogCaptureEnabled(cfg) { return } ginCtx := ginContextFrom(ctx) @@ -111,25 +146,27 @@ func RecordAPIResponseMetadata(ctx context.Context, cfg *config.Config, status i return } attempts, attempt := ensureAttempt(ginCtx) - ensureResponseIntro(attempt) + ensureResponseIntro(ginCtx, attempt) if status > 0 && !attempt.statusWritten { - attempt.response.WriteString(fmt.Sprintf("Status: %d\n", status)) + writeAttemptResponse(ginCtx, attempt, []byte(fmt.Sprintf("Status: %d\n", status))) attempt.statusWritten = true } if !attempt.headersWritten { - attempt.response.WriteString("Headers:\n") - writeHeaders(attempt.response, headers) + builder := &strings.Builder{} + builder.WriteString("Headers:\n") + writeHeaders(builder, headers) + writeAttemptResponse(ginCtx, attempt, []byte(builder.String())) attempt.headersWritten = true - attempt.response.WriteString("\n") + writeAttemptResponse(ginCtx, attempt, []byte("\n")) } - updateAggregatedResponse(ginCtx, attempts) + updateAggregatedResponseIfMemoryBacked(ginCtx, attempts) } // RecordAPIResponseError adds an error entry for the latest attempt when no HTTP response is available. func RecordAPIResponseError(ctx context.Context, cfg *config.Config, err error) { - if cfg == nil || !cfg.RequestLog || err == nil { + if !requestLogCaptureEnabled(cfg) || err == nil { return } ginCtx := ginContextFrom(ctx) @@ -137,24 +174,24 @@ func RecordAPIResponseError(ctx context.Context, cfg *config.Config, err error) return } attempts, attempt := ensureAttempt(ginCtx) - ensureResponseIntro(attempt) + ensureResponseIntro(ginCtx, attempt) if attempt.bodyStarted && !attempt.bodyHasContent { // Ensure body does not stay empty marker if error arrives first. attempt.bodyStarted = false } if attempt.errorWritten { - attempt.response.WriteString("\n") + writeAttemptResponse(ginCtx, attempt, []byte("\n")) } - attempt.response.WriteString(fmt.Sprintf("Error: %s\n", err.Error())) + writeAttemptResponse(ginCtx, attempt, []byte(fmt.Sprintf("Error: %s\n", err.Error()))) attempt.errorWritten = true - updateAggregatedResponse(ginCtx, attempts) + updateAggregatedResponseIfMemoryBacked(ginCtx, attempts) } // AppendAPIResponseChunk appends an upstream response chunk to Gin context for request logging. func AppendAPIResponseChunk(ctx context.Context, cfg *config.Config, chunk []byte) { - if cfg == nil || !cfg.RequestLog { + if !requestLogCaptureEnabled(cfg) { return } data := bytes.TrimSpace(chunk) @@ -166,16 +203,18 @@ func AppendAPIResponseChunk(ctx context.Context, cfg *config.Config, chunk []byt return } attempts, attempt := ensureAttempt(ginCtx) - ensureResponseIntro(attempt) + ensureResponseIntro(ginCtx, attempt) if !attempt.headersWritten { - attempt.response.WriteString("Headers:\n") - writeHeaders(attempt.response, nil) + builder := &strings.Builder{} + builder.WriteString("Headers:\n") + writeHeaders(builder, nil) + writeAttemptResponse(ginCtx, attempt, []byte(builder.String())) attempt.headersWritten = true - attempt.response.WriteString("\n") + writeAttemptResponse(ginCtx, attempt, []byte("\n")) } if !attempt.bodyStarted { - attempt.response.WriteString("Body:\n") + writeAttemptResponse(ginCtx, attempt, []byte("Body:\n")) attempt.bodyStarted = true } currentChunkIsSSEEvent := bytes.HasPrefix(data, []byte("event:")) @@ -185,18 +224,18 @@ func AppendAPIResponseChunk(ctx context.Context, cfg *config.Config, chunk []byt if attempt.prevWasSSEEvent && currentChunkIsSSEData { separator = "\n" } - attempt.response.WriteString(separator) + writeAttemptResponse(ginCtx, attempt, []byte(separator)) } - attempt.response.WriteString(string(data)) + writeAttemptResponse(ginCtx, attempt, data) attempt.bodyHasContent = true attempt.prevWasSSEEvent = currentChunkIsSSEEvent - updateAggregatedResponse(ginCtx, attempts) + updateAggregatedResponseIfMemoryBacked(ginCtx, attempts) } // RecordAPIWebsocketRequest stores an upstream websocket request event in Gin context. func RecordAPIWebsocketRequest(ctx context.Context, cfg *config.Config, info UpstreamRequestLog) { - if cfg == nil || !cfg.RequestLog { + if !requestLogCaptureEnabled(cfg) { return } ginCtx := ginContextFrom(ctx) @@ -229,7 +268,7 @@ func RecordAPIWebsocketRequest(ctx context.Context, cfg *config.Config, info Ups // RecordAPIWebsocketHandshake stores the upstream websocket handshake response metadata. func RecordAPIWebsocketHandshake(ctx context.Context, cfg *config.Config, status int, headers http.Header) { logging.SetResponseHeaders(ctx, headers) - if cfg == nil || !cfg.RequestLog { + if !requestLogCaptureEnabled(cfg) { return } ginCtx := ginContextFrom(ctx) @@ -253,7 +292,7 @@ func RecordAPIWebsocketHandshake(ctx context.Context, cfg *config.Config, status // RecordAPIWebsocketUpgradeRejection stores a rejected websocket upgrade as an HTTP attempt. func RecordAPIWebsocketUpgradeRejection(ctx context.Context, cfg *config.Config, info UpstreamRequestLog, status int, headers http.Header, body []byte) { logging.SetResponseHeaders(ctx, headers) - if cfg == nil || !cfg.RequestLog { + if !requestLogCaptureEnabled(cfg) { return } ginCtx := ginContextFrom(ctx) @@ -287,7 +326,7 @@ func WebsocketUpgradeRequestURL(rawURL string) string { // AppendAPIWebsocketResponse stores an upstream websocket response frame in Gin context. func AppendAPIWebsocketResponse(ctx context.Context, cfg *config.Config, payload []byte) { - if cfg == nil || !cfg.RequestLog { + if !requestLogCaptureEnabled(cfg) { return } data := bytes.TrimSpace(payload) @@ -311,7 +350,7 @@ func AppendAPIWebsocketResponse(ctx context.Context, cfg *config.Config, payload // RecordAPIWebsocketError stores an upstream websocket error event in Gin context. func RecordAPIWebsocketError(ctx context.Context, cfg *config.Config, stage string, err error) { - if cfg == nil || !cfg.RequestLog || err == nil { + if !requestLogCaptureEnabled(cfg) || err == nil { return } ginCtx := ginContextFrom(ctx) @@ -352,27 +391,61 @@ func ensureAttempt(ginCtx *gin.Context) ([]*upstreamAttempt, *upstreamAttempt) { attempts := getAttempts(ginCtx) if len(attempts) == 0 { attempt := &upstreamAttempt{ - index: 1, - request: "=== API REQUEST 1 ===\n\n\n", - response: &strings.Builder{}, + index: 1, + response: &strings.Builder{}, + responseSource: apiResponseSourceOrNil(ginCtx), + } + if source, ok := apiRequestSource(ginCtx); ok { + if errWrite := source.AppendBytes([]byte("=== API REQUEST 1 ===\n\n\n")); errWrite != nil { + log.WithError(errWrite).Warn("failed to append missing api request log part") + attempt.request = "=== API REQUEST 1 ===\n\n\n" + } + } else { + attempt.request = "=== API REQUEST 1 ===\n\n\n" } attempts = []*upstreamAttempt{attempt} ginCtx.Set(apiAttemptsKey, attempts) - updateAggregatedRequest(ginCtx, attempts) + if attempt.request != "" { + updateAggregatedRequest(ginCtx, attempts) + } } return attempts, attempts[len(attempts)-1] } -func ensureResponseIntro(attempt *upstreamAttempt) { +func ensureResponseIntro(ginCtx *gin.Context, attempt *upstreamAttempt) { if attempt == nil || attempt.response == nil || attempt.responseIntroWritten { return } - attempt.response.WriteString(fmt.Sprintf("=== API RESPONSE %d ===\n", attempt.index)) - attempt.response.WriteString(fmt.Sprintf("Timestamp: %s\n", time.Now().Format(time.RFC3339Nano))) - attempt.response.WriteString("\n") + writeAttemptResponse(ginCtx, attempt, []byte(fmt.Sprintf("=== API RESPONSE %d ===\n", attempt.index))) + writeAttemptResponse(ginCtx, attempt, []byte(fmt.Sprintf("Timestamp: %s\n", time.Now().Format(time.RFC3339Nano)))) + writeAttemptResponse(ginCtx, attempt, []byte("\n")) attempt.responseIntroWritten = true } +func writeAttemptResponse(ginCtx *gin.Context, attempt *upstreamAttempt, payload []byte) { + if attempt == nil || len(payload) == 0 { + return + } + if attempt.responseSource == nil { + attempt.responseSource = apiResponseSourceOrNil(ginCtx) + } + if attempt.responseSource != nil { + if errWrite := attempt.responseSource.AppendBytes(payload); errWrite == nil { + if ginCtx != nil { + ginCtx.Set(logging.APIResponseCapturedContextKey, true) + } + return + } else { + log.WithError(errWrite).Warn("failed to append api response log part") + attempt.responseSource = nil + } + } + if attempt.response == nil { + attempt.response = &strings.Builder{} + } + attempt.response.Write(payload) +} + func updateAggregatedRequest(ginCtx *gin.Context, attempts []*upstreamAttempt) { if ginCtx == nil { return @@ -384,6 +457,13 @@ func updateAggregatedRequest(ginCtx *gin.Context, attempts []*upstreamAttempt) { ginCtx.Set(apiRequestKey, []byte(builder.String())) } +func updateAggregatedResponseIfMemoryBacked(ginCtx *gin.Context, attempts []*upstreamAttempt) { + if apiResponseSourceOrNil(ginCtx) != nil { + return + } + updateAggregatedResponse(ginCtx, attempts) +} + func updateAggregatedResponse(ginCtx *gin.Context, attempts []*upstreamAttempt) { if ginCtx == nil { return @@ -408,6 +488,18 @@ func updateAggregatedResponse(ginCtx *gin.Context, attempts []*upstreamAttempt) ginCtx.Set(apiResponseKey, []byte(builder.String())) } +func apiRequestSource(ginCtx *gin.Context) (*logging.FileBodySource, bool) { + return fileBodySourceFromGin(ginCtx, logging.APIRequestSourceContextKey) +} + +func apiResponseSourceOrNil(ginCtx *gin.Context) *logging.FileBodySource { + source, ok := fileBodySourceFromGin(ginCtx, logging.APIResponseSourceContextKey) + if !ok { + return nil + } + return source +} + func appendAPIWebsocketTimeline(ginCtx *gin.Context, chunk []byte) { if ginCtx == nil { return @@ -440,10 +532,14 @@ func appendAPIWebsocketTimeline(ginCtx *gin.Context, chunk []byte) { } func apiWebsocketTimelineSource(ginCtx *gin.Context) (*logging.FileBodySource, bool) { + return fileBodySourceFromGin(ginCtx, logging.APIWebsocketTimelineSourceContextKey) +} + +func fileBodySourceFromGin(ginCtx *gin.Context, key string) (*logging.FileBodySource, bool) { if ginCtx == nil { return nil, false } - value, exists := ginCtx.Get(logging.APIWebsocketTimelineSourceContextKey) + value, exists := ginCtx.Get(key) if !exists { return nil, false } diff --git a/sdk/api/handlers/handlers.go b/sdk/api/handlers/handlers.go index 55b4d6ab..8b51d9ee 100644 --- a/sdk/api/handlers/handlers.go +++ b/sdk/api/handlers/handlers.go @@ -447,6 +447,12 @@ func (h *BaseAPIHandler) GetContextWithCancel(handler interfaces.APIHandler, c * logging.SetResponseStatus(cancelCtx, c.Writer.Status()) } if h.Cfg.RequestLog && len(params) == 1 { + if captured, exists := c.Get(logging.APIResponseCapturedContextKey); exists { + if capturedBool, ok := captured.(bool); ok && capturedBool { + cancel() + return + } + } if existing, exists := c.Get("API_RESPONSE"); exists { if existingBytes, ok := existing.([]byte); ok && len(bytes.TrimSpace(existingBytes)) > 0 { switch params[0].(type) {