Repository: traefik/traefik (language: go)

Issue:
DownstreamStatus is 0 in v3 when server-sent event response is aborted from client
### Welcome!

- [X] Yes, I've searched similar issues on [GitHub](https://github.com/traefik/traefik/issues) and didn't find any.
- [X] Yes, I've searched similar issues on the [Traefik community forum](https://community.traefik.io) and didn't find any.

### What did you do?

In Traefik v2.11, when client aborts server-sent events response stream, `DownstreamStatus` was `200` in the access log.
After upgrading to v3.0 the response itself is working as same as before, but in the access log `DownstreamStatus` is now logged as `0`.

[Reproducible docker compose setup](https://gist.github.com/dj-shin/5a4e69d8b6c09a303319d7d0082d98ec)

### What did you see instead?

v2.11 access log:
```
{
  "@timestamp": "2024-06-12T07:35:16.316813044Z",
  "ClientUsername": "-",
  "DownstreamContentSize": 4493,
  "DownstreamStatus": 200,
  "Duration": 694375616,
  "GzipRatio": 0,
  "OriginContentSize": 0,
  "OriginDuration": 0,
  "OriginStatus": 0,
  "Overhead": 694375616,
  "RequestContentSize": 438,
  "RequestCount": 543,
  "RequestMethod": "POST",
  "RequestPath": "/v1/chat/completions",
  "RequestPort": "-",
  "RequestProtocol": "HTTP/1.1",
  "RequestScheme": "http",
  "RetryAttempts": 0,
  "StartLocal": "2024-06-12T07:35:15.622296575Z",
  "StartUTC": "2024-06-12T07:35:15.622296575Z",
  "downstream_Access-Control-Allow-Credentials": "REDACTED",
  "downstream_Content-Type": "REDACTED",
  "downstream_Date": "REDACTED",
  "downstream_Server": "REDACTED",
  "downstream_Vary": "REDACTED",
  "entryPointName": "inference-server",
  "level": "info",
  "msg": "",
  "request_Accept": "REDACTED",
  "request_Authorization": "REDACTED",
  "request_Content-Length": "REDACTED",
  "request_Content-Type": "REDACTED",
  "request_User-Agent": "REDACTED",
  "request_X-Amzn-Trace-Id": "REDACTED",
  "request_X-Forwarded-Host": "REDACTED",
  "request_X-Forwarded-Port": "REDACTED",
  "request_X-Forwarded-Proto": "REDACTED",
  "request_X-Forwarded-Server": "REDACTED",
  "request_X-Real-Ip": "REDACTED",
  "request_X-Request-Id": "da53fa8d71fe4953928e12468f18ffd8",
  "time": "2024-06-12T07:35:16Z"
}
```

v3.0.2 access log:
```
{
  "@timestamp": "2024-06-12T07:48:16.096445637Z",
  "DownstreamContentSize": 0,
  "DownstreamStatus": 0,
  "Duration": 343245914,
  "OriginContentSize": 0,
  "OriginDuration": 0,
  "OriginStatus": 0,
  "Overhead": 343245914,
  "RequestContentSize": 0,
  "RequestCount": 2,
  "RequestMethod": "POST",
  "RequestPath": "/v1/chat/completions",
  "RequestPort": "-",
  "RequestProtocol": "HTTP/1.1",
  "RequestScheme": "http",
  "RetryAttempts": 0,
  "StartLocal": "2024-06-12T07:48:15.753032977Z",
  "StartUTC": "2024-06-12T07:48:15.753032977Z",
  "entryPointName": "inference-server",
  "level": "info",
  "msg": "",
  "request_Accept": "REDACTED",
  "request_Authorization": "REDACTED",
  "request_Content-Length": "REDACTED",
  "request_Content-Type": "REDACTED",
  "request_User-Agent": "REDACTED",
  "request_X-Amzn-Trace-Id": "REDACTED",
  "request_X-Forwarded-Host": "REDACTED",
  "request_X-Forwarded-Port": "REDACTED",
  "request_X-Forwarded-Proto": "REDACTED",
  "request_X-Forwarded-Server": "REDACTED",
  "request_X-Real-Ip": "REDACTED",
  "request_X-Request-Id": "e37e4460614b428cbf4c1d474f1bdc5c",
  "time": "2024-06-12T07:48:16Z"
}
```

### What version of Traefik are you using?

Tested in v3.0.1 & v3.0.2 and both behaviors where same.
v2.11 works fine.

### What is your environment & configuration?

```yaml
apiVersion: v1
data:
  traefik.yaml: |
    providers:
      file:
        directory: /etc/traefik/
        watch: true
    entryPoints:
      traefik:
        address: :8080
      inference-server:
        address: :8090
        transport:
          lifeCycle:
            graceTimeOut: 3600s
    api:
      insecure: true
      dashboard: true
      debug: true
    metrics:
      prometheus:
        entryPoint: traefik
    accessLog:
      filePath: "/logs/access.jsonl"
      format: json
      fields:
        defaultMode: keep
        headers:
          defaultMode: redact
          names:
            X-Request-Id: keep
    ping:
      entryPoint: "inference-server"
    log:
      level: DEBUG
kind: ConfigMap
```

Add more configuration information here.


### If applicable, please paste the log output in DEBUG level

_No response_


Discussion / hints:
Hello @dj-shin,

Thanks for reporting this!

Could you please also share a minimal dynamic configuration to reproduce the issue?
Hello @rtribotte, thank you for the fast check!

Here's our dynamic config:

```yaml
http:
  serversTransports:
    reuseConnection:
      maxIdleConnsPerHost: 5
      forwardingTimeouts:
        idleConnTimeout: 600s
  middlewares:
    CORS-Header:
      headers:
        accessControlAllowMethods:
          - GET
          - OPTIONS
          - PUT
          - POST
        accessControlAllowHeaders:
          - Authorization
          - Content-Type
        accessControlAllowOriginListRegex:
          - ...
        accessControlAllowCredentials: true
        accessControlMaxAge: 100
        addVaryHeader: true
  routers:
    router-test:
      entryPoints:
        - inference-server
      middlewares:
        - CORS-Header
        - auths
      service: test-cluster
      rule: PathPrefix(`/test`)
    router-inference:
      entryPoints:
      - inference-server
      middlewares:
      - CORS-Header
      - auths
      service: inference-cluster
      rule: PathPrefix(`/`)
  services:
    inference-cluster:
      loadBalancer:
        healthCheck:
          path: /ping
          interval: 15s
          timeout: 10s
        servers:
          - url: ...
        serversTransport: reuseConnection
    test-cluster:
      loadBalancer:
        servers:
        - url: ...
        serversTransport: reuseConnection
```
Hey @dj-shin ,

Thanks for the additional information.
If any community member can help us to move forward on it, we would love the help.
I’d love to help here, but I guess a minimal repro with a server/client setup similar to OP’s would go a long way. @dj-shin if possible, could you provide a minimal dockerised reproduction of your setup?
Thank you for your attention, @hood !
I've created a [docker compose setup](https://gist.github.com/dj-shin/5a4e69d8b6c09a303319d7d0082d98ec) to reproduce my issue. 

I've also tested in this setup too that v2.11 logs different output

Fix patch:
diff --git a/pkg/middlewares/accesslog/logger.go b/pkg/middlewares/accesslog/logger.go
index ed85a8888c..b5a533b8c2 100644
--- a/pkg/middlewares/accesslog/logger.go
+++ b/pkg/middlewares/accesslog/logger.go
@@ -196,16 +196,6 @@ func (h *Handler) ServeHTTP(rw http.ResponseWriter, req *http.Request, next http
 		},
 	}
 
-	defer func() {
-		if h.config.BufferingSize > 0 {
-			h.logHandlerChan <- handlerParams{
-				logDataTable: logDataTable,
-			}
-			return
-		}
-		h.logTheRoundTrip(logDataTable)
-	}()
-
 	reqWithDataTable := req.WithContext(context.WithValue(req.Context(), DataTableKey, logDataTable))
 
 	core[RequestCount] = nextRequestCount()
@@ -249,19 +239,30 @@ func (h *Handler) ServeHTTP(rw http.ResponseWriter, req *http.Request, next http
 		return
 	}
 
-	next.ServeHTTP(rw, reqWithDataTable)
+	defer func() {
+		logDataTable.DownstreamResponse = downstreamResponse{
+			headers: rw.Header().Clone(),
+		}
 
-	if _, ok := core[ClientUsername]; !ok {
-		core[ClientUsername] = usernameIfPresent(reqWithDataTable.URL)
-	}
+		logDataTable.DownstreamResponse.status = capt.StatusCode()
+		logDataTable.DownstreamResponse.size = capt.ResponseSize()
+		logDataTable.Request.size = capt.RequestSize()
 
-	logDataTable.DownstreamResponse = downstreamResponse{
-		headers: rw.Header().Clone(),
-	}
+		if _, ok := core[ClientUsername]; !ok {
+			core[ClientUsername] = usernameIfPresent(reqWithDataTable.URL)
+		}
 
-	logDataTable.DownstreamResponse.status = capt.StatusCode()
-	logDataTable.DownstreamResponse.size = capt.ResponseSize()
-	logDataTable.Request.size = capt.RequestSize()
+		if h.config.BufferingSize > 0 {
+			h.logHandlerChan <- handlerParams{
+				logDataTable: logDataTable,
+			}
+			return
+		}
+
+		h.logTheRoundTrip(logDataTable)
+	}()
+
+	next.ServeHTTP(rw, reqWithDataTable)
 }
 
 // Close closes the Logger (i.e. the file, drain logHandlerChan, etc).


Test patch:
diff --git a/pkg/middlewares/accesslog/logger_test.go b/pkg/middlewares/accesslog/logger_test.go
index d205bb9d79..01433c5f63 100644
--- a/pkg/middlewares/accesslog/logger_test.go
+++ b/pkg/middlewares/accesslog/logger_test.go
@@ -2,6 +2,7 @@ package accesslog
 
 import (
 	"bytes"
+	"context"
 	"crypto/tls"
 	"encoding/json"
 	"fmt"
@@ -22,6 +23,7 @@ import (
 	"github.com/stretchr/testify/require"
 	ptypes "github.com/traefik/paerser/types"
 	"github.com/traefik/traefik/v2/pkg/middlewares/capture"
+	"github.com/traefik/traefik/v2/pkg/middlewares/recovery"
 	"github.com/traefik/traefik/v2/pkg/types"
 )
 
@@ -519,6 +521,64 @@ func TestLoggerJSON(t *testing.T) {
 	}
 }
 
+func TestLogger_AbortedRequest(t *testing.T) {
+	expected := map[string]func(t *testing.T, value interface{}){
+		RequestContentSize:             assertFloat64(0),
+		RequestHost:                    assertString(testHostname),
+		RequestAddr:                    assertString(testHostname),
+		RequestMethod:                  assertString(testMethod),
+		RequestPath:                    assertString(""),
+		RequestProtocol:                assertString(testProto),
+		RequestScheme:                  assertString(testScheme),
+		RequestPort:                    assertString("-"),
+		DownstreamStatus:               assertFloat64(float64(200)),
+		DownstreamContentSize:          assertFloat64(float64(40)),
+		RequestRefererHeader:           assertString(testReferer),
+		RequestUserAgentHeader:         assertString(testUserAgent),
+		ServiceURL:                     assertString("http://stream"),
+		ServiceAddr:                    assertString("127.0.0.1"),
+		ServiceName:                    assertString("stream"),
+		ClientUsername:                 assertString(testUsername),
+		ClientHost:                     assertString(testHostname),
+		ClientPort:                     assertString(strconv.Itoa(testPort)),
+		ClientAddr:                     assertString(fmt.Sprintf("%s:%d", testHostname, testPort)),
+		"level":                        assertString("info"),
+		"msg":                          assertString(""),
+		RequestCount:                   assertFloat64NotZero(),
+		Duration:                       assertFloat64NotZero(),
+		Overhead:                       assertFloat64NotZero(),
+		RetryAttempts:                  assertFloat64(float64(0)),
+		"time":                         assertNotEmpty(),
+		StartLocal:                     assertNotEmpty(),
+		StartUTC:                       assertNotEmpty(),
+		"downstream_Content-Type":      assertString("text/plain"),
+		"downstream_Transfer-Encoding": assertString("chunked"),
+		"downstream_Cache-Control":     assertString("no-cache"),
+	}
+
+	config := &types.AccessLog{
+		FilePath: filepath.Join(t.TempDir(), logFileNameSuffix),
+		Format:   JSONFormat,
+	}
+	doLoggingWithAbortedStream(t, config)
+
+	logData, err := os.ReadFile(config.FilePath)
+	require.NoError(t, err)
+
+	jsonData := make(map[string]interface{})
+	err = json.Unmarshal(logData, &jsonData)
+	require.NoError(t, err)
+
+	assert.Equal(t, len(expected), len(jsonData))
+
+	for field, assertion := range expected {
+		assertion(t, jsonData[field])
+		if t.Failed() {
+			return
+		}
+	}
+}
+
 func TestNewLogHandlerOutputStdout(t *testing.T) {
 	testCases := []struct {
 		desc        string
@@ -852,3 +912,89 @@ func logWriterTestHandlerFunc(rw http.ResponseWriter, r *http.Request) {
 
 	rw.WriteHeader(testStatus)
 }
+
+func doLoggingWithAbortedStream(t *testing.T, config *types.AccessLog) {
+	t.Helper()
+
+	logger, err := NewHandler(config)
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		err := logger.Close()
+		require.NoError(t, err)
+	})
+
+	if config.FilePath != "" {
+		_, err = os.Stat(config.FilePath)
+		require.NoError(t, err, "logger should create "+config.FilePath)
+	}
+
+	reqContext, cancelRequest := context.WithCancel(context.Background())
+
+	req := &http.Request{
+		Header: map[string][]string{
+			"User-Agent": {testUserAgent},
+			"Referer":    {testReferer},
+		},
+		Proto:      testProto,
+		Host:       testHostname,
+		Method:     testMethod,
+		RemoteAddr: fmt.Sprintf("%s:%d", testHostname, testPort),
+		URL: &url.URL{
+			User: url.UserPassword(testUsername, ""),
+		},
+		Body: nil,
+	}
+
+	req = req.WithContext(reqContext)
+
+	chain := alice.New()
+	chain = chain.Append(func(next http.Handler) (http.Handler, error) {
+		return recovery.New(context.Background(), next)
+	})
+	chain = chain.Append(capture.Wrap)
+	chain = chain.Append(WrapHandler(logger))
+
+	service := NewFieldHandler(http.HandlerFunc(streamBackend), ServiceURL, "http://stream", nil)
+	service = NewFieldHandler(service, ServiceAddr, "127.0.0.1", nil)
+	service = NewFieldHandler(service, ServiceName, "stream", AddServiceFields)
+
+	handler, err := chain.Then(service)
+	require.NoError(t, err)
+
+	go func() {
+		time.Sleep(499 * time.Millisecond)
+		cancelRequest()
+	}()
+
+	handler.ServeHTTP(httptest.NewRecorder(), req)
+}
+
+func streamBackend(rw http.ResponseWriter, r *http.Request) {
+	// Get the Flusher to flush the response to the client
+	flusher, ok := rw.(http.Flusher)
+	if !ok {
+		http.Error(rw, "Streaming unsupported!", http.StatusInternalServerError)
+		return
+	}
+
+	// Set the headers for streaming
+	rw.Header().Set("Content-Type", "text/plain")
+	rw.Header().Set("Transfer-Encoding", "chunked")
+	rw.Header().Set("Cache-Control", "no-cache")
+
+	for {
+		time.Sleep(100 * time.Millisecond)
+
+		select {
+		case <-r.Context().Done():
+			panic(http.ErrAbortHandler)
+
+		default:
+			if _, err := fmt.Fprint(rw, "FOOBAR!!!!"); err != nil {
+				http.Error(rw, err.Error(), http.StatusInternalServerError)
+				return
+			}
+			flusher.Flush()
+		}
+	}
+}