Repository: kubernetes/kubernetes (language: go)

Issue:
kube-apiserver logs watch requests before they end in 1.30
### What happened?


In 1.30, WATCH request is logged immediately when a watch is opened, with very low latency. There appears to be no trace of when the watch ends. This significantly impacts debugging capability for issues with establishing watches. 

Example for consecutive watches on pods opened by kube-scheduler - each appears to be lasting 8-9 minutes, everything works in the cluster, but the latency doesn't match watch length, the request is logged immediately, and with `resp=0`:

```
I0620 12:33:17.389016      11 httplog.go:132] "HTTP" verb="WATCH" URI="/api/v1/pods?allowWatchBookmarks=true&fieldSelector=status.phase%21%3DSucceeded%2Cstatus.phase%21%3DFailed&resourceVersion=97186&timeout=9m12s&timeoutSeconds=552&watch=true" latency="470.891µs" userAgent="kube-scheduler/v1.30.2 (linux/amd64) kubernetes/85fad5b/scheduler" audit-ID="2cf3b6d8-8c04-4065-8bb8-ccd209f24063" srcIP="[::1]:60566" apf_pl="workload-high" apf_fs="kube-scheduler" apf_iseats=1 apf_fseats=0 apf_additionalLatency="0s" apf_init_latency="252.569µs" apf_execution_time="253.277µs" resp=0
...
I0620 12:42:29.390797      11 get.go:261] "Starting watch" path="/api/v1/pods" resourceVersion="135790" labels="" fields="status.phase!=Failed,status.phase!=Succeeded" timeout="6m10s"
I0620 12:42:29.391006      11 httplog.go:132] "HTTP" verb="WATCH" URI="/api/v1/pods?allowWatchBookmarks=true&fieldSelector=status.phase%21%3DSucceeded%2Cstatus.phase%21%3DFailed&resourceVersion=135790&timeout=6m10s&timeoutSeconds=370&watch=true" latency="629.842µs" userAgent="kube-scheduler/v1.30.2 (linux/amd64) kubernetes/85fad5b/scheduler" audit-ID="619843d2-3f2a-4c2e-960a-17eecc71695b" srcIP="[::1]:60566" apf_pl="workload-high" apf_fs="kube-scheduler" apf_iseats=1 apf_fseats=0 apf_additionalLatency="0s" apf_init_latency="299.938µs" apf_execution_time="300.55µs" resp=0
```

### What did you expect to happen?

Watches are usually long (~minutes). In 1.29 and earlier log looks like this:

```
I0620 12:30:26.940451      11 get.go:260] "Starting watch" path="/api/v1/pods" resourceVersion="93431" labels="" fields="status.phase!=Failed,status.phase!=Succeeded" timeout="8m18s"
...
I0620 12:38:44.940817      11 httplog.go:132] "HTTP" verb="WATCH" URI="/api/v1/pods?allowWatchBookmarks=true&fieldSelector=status.phase%21%3DSucceeded%2Cstatus.phase%21%3DFailed&resourceVersion=93431&timeout=8m18s&timeoutSeconds=498&watch=true" latency="8m18.00060226s" userAgent="kube-scheduler/v1.29.6 (linux/amd64) kubernetes/43a444e/scheduler" audit-ID="c39f0303-10ad-4b51-baf4-87c592fcb85e" srcIP="[::1]:39438" apf_pl="workload-high" apf_fs="kube-scheduler" apf_iseats=1 apf_fseats=0 apf_additionalLatency="0s" apf_init_latency="280.462µs" apf_execution_time="281.612µs" resp=200
```

### How can we reproduce it (as minimally and precisely as possible)?

Compare kube-apiserver log in 1.29 and 1.30.

### Anything else we need to know?

Log line examples are taken from passing runs of gce-scalabiltiy-100 tests: 
- https://testgrid.k8s.io/sig-scalability-gce#gce-cos-1.29-scalability-100 : https://prow.k8s.io/view/gs/kubernetes-jenkins/logs/ci-kubernetes-e2e-gci-gce-scalability-1-29/1803759918289784832
- https://testgrid.k8s.io/sig-scalability-gce#gce-cos-1.30-scalability-100 : https://prow.k8s.io/view/gs/kubernetes-jenkins/logs/ci-kubernetes-e2e-gci-gce-scalability-1-30/1803759918403031040

### Kubernetes version

Observed in 1.30.2 and 1.30.1. To be verified if present in 1.30.0..

### Cloud provider

N/A


### OS version

N/A

### Install tools

N/A


### Container runtime (CRI) and version (if applicable)

N/A

### Related plugins (CNI, CSI, ...) and versions (if applicable)

N/A


Discussion / hints:
/cc @mborsz
/sig api-machinery
/cc

It looks like https://github.com/kubernetes/kubernetes/pull/120902 has moved watch to a defered goroutine. IIUC this can lead to effects like this. The problem is that it changes a behavior of httplog and removes us a way to track a length of watch calls which turned out to be useful in many debugging sessions in the past.

We either should fix httplog or provide some other log with equivalent information instead (i.e. when the watch ends).
fortunately we were cautious and this is behind a feature gate https://github.com/kubernetes/kubernetes/pull/120902#discussion_r1430722839

```
grep -r APIServingWithRoutine {pkg,staging}
pkg/features/kube_features.go:  genericfeatures.APIServingWithRoutine: {Default: true, PreRelease: featuregate.Beta},
staging/src/k8s.io/apiserver/pkg/server/config.go:      if utilfeature.DefaultFeatureGate.Enabled(genericfeatures.APIServingWithRoutine) {
staging/src/k8s.io/apiserver/pkg/features/kube_features.go:     APIServingWithRoutine featuregate.Feature = "APIServingWithRoutine"
staging/src/k8s.io/apiserver/pkg/features/kube_features.go:     APIServingWithRoutine: {Default: true, PreRelease: featuregate.Beta},
```

 /cc @linxiulei @wojtek-t 
/assign
> We either should fix httplog or provide some other log with equivalent information instead (i.e. when the watch ends).

@mborsz I tend to fix it for compatibility but also want to add an additional information so that we can see the latencies for both watch initialization (which allowed me to find https://github.com/kubernetes/kubernetes/pull/124692) and duration. On a different note, would adding a metric like `longrunning_duration` be better solution for your use cases?
>On a different note, would adding a metric like longrunning_duration be better solution for your use

Metric wouldn't work for the cases where this particular log was useful. (The cases I have in mind were mostly about figuring out _why_ watches opened by a particular component ended prematurely - metric would only confirm that they did, which we already knew from client side.)
> but also want to add an additional information so that we can see the latencies for both watch initialization (which allowed me to find https://github.com/kubernetes/kubernetes/pull/124692) and duration

Correction: actually there is `apf_execution_time` in the httplog, which is a proxy to initiation latency

Fix patch:
diff --git a/staging/src/k8s.io/apiserver/pkg/endpoints/handlers/get.go b/staging/src/k8s.io/apiserver/pkg/endpoints/handlers/get.go
index d928deed09c40..3cc4a728a57ba 100644
--- a/staging/src/k8s.io/apiserver/pkg/endpoints/handlers/get.go
+++ b/staging/src/k8s.io/apiserver/pkg/endpoints/handlers/get.go
@@ -41,7 +41,7 @@ import (
 	"k8s.io/apiserver/pkg/endpoints/request"
 	"k8s.io/apiserver/pkg/features"
 	"k8s.io/apiserver/pkg/registry/rest"
-	genericfilters "k8s.io/apiserver/pkg/server/filters"
+	"k8s.io/apiserver/pkg/server/routine"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/component-base/tracing"
 	"k8s.io/klog/v2"
@@ -285,7 +285,7 @@ func ListResource(r rest.Lister, rw rest.Watcher, scope *RequestScope, forceWatc
 			}
 
 			// Run watch serving in a separate goroutine to allow freeing current stack memory
-			t := genericfilters.TaskFrom(req.Context())
+			t := routine.TaskFrom(req.Context())
 			if t != nil {
 				t.Func = serve
 			} else {
diff --git a/staging/src/k8s.io/apiserver/pkg/server/config.go b/staging/src/k8s.io/apiserver/pkg/server/config.go
index 6f0ca1bcac4d8..62f077f24b3cb 100644
--- a/staging/src/k8s.io/apiserver/pkg/server/config.go
+++ b/staging/src/k8s.io/apiserver/pkg/server/config.go
@@ -64,6 +64,7 @@ import (
 	genericfilters "k8s.io/apiserver/pkg/server/filters"
 	"k8s.io/apiserver/pkg/server/healthz"
 	"k8s.io/apiserver/pkg/server/routes"
+	"k8s.io/apiserver/pkg/server/routine"
 	serverstore "k8s.io/apiserver/pkg/server/storage"
 	storagevalue "k8s.io/apiserver/pkg/storage/value"
 	"k8s.io/apiserver/pkg/storageversion"
@@ -1055,7 +1056,7 @@ func DefaultBuildHandlerChain(apiHandler http.Handler, c *Config) http.Handler {
 	// handler in current goroutine to minimize the stack memory usage. It must be
 	// after WithPanicRecover() to be protected from panics.
 	if c.FeatureGate.Enabled(genericfeatures.APIServingWithRoutine) {
-		handler = genericfilters.WithRoutine(handler, c.LongRunningFunc)
+		handler = routine.WithRoutine(handler, c.LongRunningFunc)
 	}
 	handler = genericapifilters.WithRequestInfo(handler, c.RequestInfoResolver)
 	handler = genericapifilters.WithRequestReceivedTimestamp(handler)
diff --git a/staging/src/k8s.io/apiserver/pkg/server/httplog/httplog.go b/staging/src/k8s.io/apiserver/pkg/server/httplog/httplog.go
index a4b4c5899fd1c..c64f5771d8ba6 100644
--- a/staging/src/k8s.io/apiserver/pkg/server/httplog/httplog.go
+++ b/staging/src/k8s.io/apiserver/pkg/server/httplog/httplog.go
@@ -31,6 +31,7 @@ import (
 	"k8s.io/apiserver/pkg/endpoints/metrics"
 	"k8s.io/apiserver/pkg/endpoints/request"
 	"k8s.io/apiserver/pkg/endpoints/responsewriter"
+	"k8s.io/apiserver/pkg/server/routine"
 	"k8s.io/klog/v2"
 )
 
@@ -125,10 +126,26 @@ func withLogging(handler http.Handler, stackTracePred StacktracePred, shouldLogR
 		rl := newLoggedWithStartTime(req, w, startTime)
 		rl.StacktraceWhen(stackTracePred)
 		req = req.WithContext(context.WithValue(ctx, respLoggerContextKey, rl))
-		defer rl.Log()
+
+		var logFunc func()
+		logFunc = rl.Log
+		defer func() {
+			if logFunc != nil {
+				logFunc()
+			}
+		}()
 
 		w = responsewriter.WrapForHTTP1Or2(rl)
 		handler.ServeHTTP(w, req)
+
+		// We need to ensure that the request is logged after it is processed.
+		// In case the request is executed in a separate goroutine created via
+		// WithRoutine handler in the handler chain (i.e. above handler.ServeHTTP()
+		// would return request is completely responsed), we want the logging to
+		// happen in that goroutine too, so we append it to the task.
+		if routine.AppendTask(ctx, &routine.Task{Func: rl.Log}) {
+			logFunc = nil
+		}
 	})
 }
 
diff --git a/staging/src/k8s.io/apiserver/pkg/server/filters/routine.go b/staging/src/k8s.io/apiserver/pkg/server/routine/routine.go
similarity index 83%
rename from staging/src/k8s.io/apiserver/pkg/server/filters/routine.go
rename to staging/src/k8s.io/apiserver/pkg/server/routine/routine.go
index 3f4dfa2bb228d..c4b23c5903431 100644
--- a/staging/src/k8s.io/apiserver/pkg/server/filters/routine.go
+++ b/staging/src/k8s.io/apiserver/pkg/server/routine/routine.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package filters
+package routine
 
 import (
 	"context"
@@ -35,6 +35,20 @@ func WithTask(parent context.Context, t *Task) context.Context {
 	return request.WithValue(parent, taskKey, t)
 }
 
+// AppendTask appends a task executed after completion of existing task.
+// It is a no-op if there is no existing task.
+func AppendTask(ctx context.Context, t *Task) bool {
+	if existTask := TaskFrom(ctx); existTask != nil && existTask.Func != nil {
+		existFunc := existTask.Func
+		existTask.Func = func() {
+			existFunc()
+			t.Func()
+		}
+		return true
+	}
+	return false
+}
+
 func TaskFrom(ctx context.Context) *Task {
 	t, _ := ctx.Value(taskKey).(*Task)
 	return t


Test patch:
diff --git a/staging/src/k8s.io/apiserver/pkg/server/filters/routine_test.go b/staging/src/k8s.io/apiserver/pkg/server/filters/wrap_test.go
similarity index 69%
rename from staging/src/k8s.io/apiserver/pkg/server/filters/routine_test.go
rename to staging/src/k8s.io/apiserver/pkg/server/filters/wrap_test.go
index fbfce6080699f..561a18a5581c6 100644
--- a/staging/src/k8s.io/apiserver/pkg/server/filters/routine_test.go
+++ b/staging/src/k8s.io/apiserver/pkg/server/filters/wrap_test.go
@@ -1,5 +1,5 @@
 /*
-Copyright 2023 The Kubernetes Authors.
+Copyright 2024 The Kubernetes Authors.
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
@@ -25,6 +25,7 @@ import (
 
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apiserver/pkg/endpoints/request"
+	"k8s.io/apiserver/pkg/server/routine"
 	"k8s.io/klog/v2"
 )
 
@@ -42,7 +43,7 @@ func TestPropogatingPanic(t *testing.T) {
 		APIPrefixes:          sets.NewString("api", "apis"),
 		GrouplessAPIPrefixes: sets.NewString("api"),
 	}
-	ts := httptest.NewServer(WithRoutine(WithPanicRecovery(handler, resolver), func(_ *http.Request, _ *request.RequestInfo) bool { return true }))
+	ts := httptest.NewServer(routine.WithRoutine(WithPanicRecovery(handler, resolver), func(_ *http.Request, _ *request.RequestInfo) bool { return true }))
 	defer ts.Close()
 	_, err := http.Get(ts.URL)
 	if err == nil {
@@ -57,23 +58,3 @@ func TestPropogatingPanic(t *testing.T) {
 		t.Errorf("unexpected out captured actual = %v", capturedOutput)
 	}
 }
-
-func TestExecutionWithRoutine(t *testing.T) {
-	var executed bool
-	handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		t := TaskFrom(r.Context())
-		t.Func = func() {
-			executed = true
-		}
-	})
-	ts := httptest.NewServer(WithRoutine(handler, func(_ *http.Request, _ *request.RequestInfo) bool { return true }))
-	defer ts.Close()
-
-	_, err := http.Get(ts.URL)
-	if err != nil {
-		t.Errorf("got unexpected error on request: %v", err)
-	}
-	if !executed {
-		t.Error("expected to execute")
-	}
-}
diff --git a/staging/src/k8s.io/apiserver/pkg/server/routine/routine_test.go b/staging/src/k8s.io/apiserver/pkg/server/routine/routine_test.go
new file mode 100644
index 0000000000000..1d4249b552084
--- /dev/null
+++ b/staging/src/k8s.io/apiserver/pkg/server/routine/routine_test.go
@@ -0,0 +1,99 @@
+/*
+Copyright 2023 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package routine
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"k8s.io/apiserver/pkg/endpoints/request"
+)
+
+func TestExecutionWithRoutine(t *testing.T) {
+	var executed bool
+	handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		t := TaskFrom(r.Context())
+		t.Func = func() {
+			executed = true
+		}
+	})
+	ts := httptest.NewServer(WithRoutine(handler, func(_ *http.Request, _ *request.RequestInfo) bool { return true }))
+	defer ts.Close()
+
+	_, err := http.Get(ts.URL)
+	if err != nil {
+		t.Errorf("got unexpected error on request: %v", err)
+	}
+	if !executed {
+		t.Error("expected to execute")
+	}
+}
+
+func TestAppendTask(t *testing.T) {
+	tests := []struct {
+		name          string
+		existingTask  bool
+		taskAppended  bool
+		shouldExecute bool
+	}{
+		{
+			name:          "append task when existing",
+			existingTask:  true,
+			taskAppended:  true,
+			shouldExecute: true,
+		},
+		{
+			name:          "not append task when no existing tasks",
+			existingTask:  false,
+			taskAppended:  false,
+			shouldExecute: false,
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			var executed, appended bool
+			taskToAppend := func() {
+				executed = true
+			}
+			handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				ctx := r.Context()
+				if test.existingTask {
+					t := TaskFrom(ctx)
+					t.Func = func() {}
+				}
+
+				appended = AppendTask(ctx, &Task{taskToAppend})
+			})
+			ts := httptest.NewServer(WithRoutine(handler, func(_ *http.Request, _ *request.RequestInfo) bool { return true }))
+			defer ts.Close()
+
+			_, err := http.Get(ts.URL)
+			if err != nil {
+				t.Errorf("got unexpected error on request: %v", err)
+			}
+
+			if test.taskAppended != appended {
+				t.Errorf("expected taskAppended: %t, got: %t", test.taskAppended, executed)
+			}
+
+			if test.shouldExecute != executed {
+				t.Errorf("expected shouldExecute: %t, got: %t", test.shouldExecute, executed)
+			}
+		})
+	}
+}