Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 2 additions & 6 deletions api/middleware/log.go
Original file line number Diff line number Diff line change
Expand Up @@ -8,14 +8,10 @@ import (

"github.com/gin-gonic/gin"
"opencsg.com/csghub-server/api/httpbase"
commonutils "opencsg.com/csghub-server/common/utils/common"
"opencsg.com/csghub-server/common/utils/trace"
)

// Status 499 is a non-standard code introduced by nginx to indicate
// "Client Closed Request" — the client disconnected before the server
// finished processing.
const StatusClientClosedRequest = 499

func Log() gin.HandlerFunc {
return func(ctx *gin.Context) {
if ctx.Request.URL.Path == "/healthz" {
Expand Down Expand Up @@ -43,7 +39,7 @@ func Log() gin.HandlerFunc {
// If the client disconnected (timeout or explicit cancel),
// override the logged status to 499.
if ctx.Request.Context().Err() == context.Canceled {
status = StatusClientClosedRequest
status = commonutils.StatusClientClosedRequest
}

// Derive a non-canceled context for the log call.
Expand Down
5 changes: 3 additions & 2 deletions api/middleware/log_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ import (
"github.com/gin-gonic/gin"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
commonutils "opencsg.com/csghub-server/common/utils/common"
)

// captureLog is a helper that installs a slog JSON handler writing to a buffer
Expand Down Expand Up @@ -82,7 +83,7 @@ func TestLog_ClientCancelBeforeHandler(t *testing.T) {

logs := readLogs()
require.Len(t, logs, 1, "must emit a log even when client canceled")
assert.Equal(t, float64(StatusClientClosedRequest), logs[0]["status"], "status should be 499 for client cancel")
assert.Equal(t, float64(commonutils.StatusClientClosedRequest), logs[0]["status"], "status should be 499 for client cancel")
assert.Equal(t, "/test", logs[0]["url"])
}

Expand Down Expand Up @@ -116,7 +117,7 @@ func TestLog_ClientCancelDuringSlowHandler(t *testing.T) {

logs := readLogs()
require.Len(t, logs, 1, "must emit a log when client cancels during slow handler")
assert.Equal(t, float64(StatusClientClosedRequest), logs[0]["status"], "status should be 499")
assert.Equal(t, float64(commonutils.StatusClientClosedRequest), logs[0]["status"], "status should be 499")
}

func TestLog_HandlerPanicWithRecovery(t *testing.T) {
Expand Down
12 changes: 12 additions & 0 deletions builder/proxy/reverse_proxy.go
Original file line number Diff line number Diff line change
@@ -1,12 +1,15 @@
package proxy

import (
"context"
"errors"
"log/slog"
"net/http"
"net/http/httputil"
"net/url"

"github.com/openai/openai-go/v3"
commonutils "opencsg.com/csghub-server/common/utils/common"
"opencsg.com/csghub-server/common/utils/trace"
)

Expand Down Expand Up @@ -54,6 +57,15 @@ func (rp *reverseProxyImpl) ServeHTTP(w http.ResponseWriter, r *http.Request, ap
}
}()
proxy := httputil.NewSingleHostReverseProxy(rp.target)
proxy.ErrorHandler = func(rw http.ResponseWriter, req *http.Request, err error) {
if errors.Is(err, context.Canceled) {
slog.InfoContext(context.WithoutCancel(req.Context()), "http: proxy request canceled", slog.Any("error", err))
rw.WriteHeader(commonutils.StatusClientClosedRequest)
return
}
slog.ErrorContext(req.Context(), "http: proxy error", slog.Any("error", err))
rw.WriteHeader(http.StatusBadGateway)
}
proxy.Director = func(req *http.Request) {
if len(svcHost) > 0 {
slog.Info("update reverse proxy header host", slog.Any("svc-host", svcHost))
Expand Down
20 changes: 20 additions & 0 deletions builder/proxy/reverse_proxy_test.go
Original file line number Diff line number Diff line change
@@ -1,11 +1,13 @@
package proxy

import (
"context"
"net/http"
"net/http/httptest"
"testing"

"github.com/stretchr/testify/require"
commonutils "opencsg.com/csghub-server/common/utils/common"
)

func TestReverseProxy_AcceptEncodingDefaultGzip(t *testing.T) {
Expand Down Expand Up @@ -47,6 +49,24 @@ func TestReverseProxy_AcceptEncodingDisabled(t *testing.T) {
require.Equal(t, "identity", downstreamAcceptEncoding)
}

func TestReverseProxy_ContextCanceledWritesClientClosed(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusOK)
}))
defer server.Close()

rp, err := NewReverseProxy(server.URL)
require.NoError(t, err)

ctx, cancel := context.WithCancel(context.Background())
cancel()
req := httptest.NewRequest(http.MethodGet, "/", nil).WithContext(ctx)
resp := httptest.NewRecorder()
rp.ServeHTTP(resp, req, "", "")

require.Equal(t, commonutils.StatusClientClosedRequest, resp.Code)
}

// import (
// "bytes"
// "io"
Expand Down
5 changes: 5 additions & 0 deletions common/utils/common/http_status.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
package common

// StatusClientClosedRequest is the non-standard nginx status code for a client
// closing the request before the server finishes processing it.
const StatusClientClosedRequest = 499
Loading