feat(logging): enable file-backed request/response sources for enhanced API logging

- Introduced support for file-backed logging of API requests and responses to handle large payloads efficiently.
- Refactored `attachWebsocketLogSources` to `attachRequestLogSources` for broader request and response handling.
- Added new methods for appending request/response data to file-backed sources and updated existing logging workflows for compatibility.
- Improved cleanup and merge logic for file-backed sources during request processing.
- Updated tests to cover newly introduced file-backed logging functionality.
This commit is contained in:
Luis Pater
2026-06-05 01:48:05 +08:00
parent 387c783b32
commit 5753d1a089
9 changed files with 386 additions and 79 deletions
+1 -1
View File
@@ -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
+12 -3
View File
@@ -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)
}
@@ -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{
+99 -11
View File
@@ -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
+13 -2
View File
@@ -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)
+1 -1
View File
@@ -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.
+112 -15
View File
@@ -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
}
@@ -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("<empty>")); 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("<empty>")
}
builder.WriteString("\n\n")
requestText = builder.String()
}
} else {
builder.WriteString("<empty>")
if len(info.Body) > 0 {
builder.WriteString(string(info.Body))
} else {
builder.WriteString("<empty>")
}
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<missing>\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<missing>\n\n")); errWrite != nil {
log.WithError(errWrite).Warn("failed to append missing api request log part")
attempt.request = "=== API REQUEST 1 ===\n<missing>\n\n"
}
} else {
attempt.request = "=== API REQUEST 1 ===\n<missing>\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
}
+6
View File
@@ -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) {