Repository: kubernetes
Issue: Audit log events for watch requests are incorrect when APIServingWithRoutine is enabled
Labels: kind/bug,sig/scalability,sig/api-machinery,triage/accepted

### What happened?

We recently noticed a significant difference in the number of audit log events related to watch requests on 1.30 clusters. Investigation found that each watch request generated two events for the ResponseStarted stage. Also, the response statuses show the synthetic "connection closed early" text, even though from the client's perspective, the watch was served normally.

What seems to be happening is a variant of https://github.com/kubernetes/kubernetes/pull/125626.

For context, the audit filter defers a function responsible for logging the ResponseCompleted audit event, and also for logging a synthetic ResponseStarted event in cases where the ResponseWriter is not touched in the course of calling the nested handlers. The same deferred function is also responsible for recovering from panics, logging an audit event for the panic, then allowing the panic to continue to propagate up the call stack.

With APIServingWithRoutine enabled, the watch handler nested within the audit filter does not immediately serve the response, it only saves a closure to the request context to be called later. So the execution sequence is something like:

1. RequestReceived
2. Closure saved to request context, holding a reference to a ResponseWriter adapter created by the audit filter
3. (Deferred) Synthetic ResponseStarted
4. ResponseCompleted
5. Watch task executes and its usage of the wrapped ResponseWriter triggers a second ResponseStarted event

These are the events I see in the audit log from a single watch request with the feature enabled. Note there are two events at the ResponseStarted stage and also that the responseStatus fields contain the synthetic event text "Connection closed early":

```json
{"kind":"Event","apiVersion":"audit.k8s.io/v1","level":"Metadata","auditID":"2fbf4326-31d4-4b2f-b214-69aca0023108","stage":"RequestReceived","requestURI":"/api/v1/namespaces?timeout=3s\u0026timeoutSeconds=3\u0026watch=true","verb":"watch","user":{"username":"system:apiserver","uid":"16aecab5-afd7-4b65-a9de-b19b0cf300f1","groups":["system:masters"]},"sourceIPs":["127.0.0.1"],"userAgent":"TestAuditWatch","objectRef":{"resource":"namespaces","apiVersion":"v1"},"requestReceivedTimestamp":"2024-07-30T16:09:28.502635Z","stageTimestamp":"2024-07-30T16:09:28.502635Z"}
{"kind":"Event","apiVersion":"audit.k8s.io/v1","level":"Metadata","auditID":"2fbf4326-31d4-4b2f-b214-69aca0023108","stage":"ResponseStarted","requestURI":"/api/v1/namespaces?timeout=3s\u0026timeoutSeconds=3\u0026watch=true","verb":"watch","user":{"username":"system:apiserver","uid":"16aecab5-afd7-4b65-a9de-b19b0cf300f1","groups":["system:masters"]},"sourceIPs":["127.0.0.1"],"userAgent":"TestAuditWatch","objectRef":{"resource":"namespaces","apiVersion":"v1"},"responseStatus":{"metadata":{},"status":"Success","message":"Connection closed early","code":200},"requestReceivedTimestamp":"2024-07-30T16:09:28.502635Z","stageTimestamp":"2024-07-30T16:09:28.502999Z","annotations":{"authorization.k8s.io/decision":"allow","authorization.k8s.io/reason":""}}
{"kind":"Event","apiVersion":"audit.k8s.io/v1","level":"Metadata","auditID":"2fbf4326-31d4-4b2f-b214-69aca0023108","stage":"ResponseComplete","requestURI":"/api/v1/namespaces?timeout=3s\u0026timeoutSeconds=3\u0026watch=true","verb":"watch","user":{"username":"system:apiserver","uid":"16aecab5-afd7-4b65-a9de-b19b0cf300f1","groups":["system:masters"]},"sourceIPs":["127.0.0.1"],"userAgent":"TestAuditWatch","objectRef":{"resource":"namespaces","apiVersion":"v1"},"responseStatus":{"metadata":{},"status":"Success","message":"Connection closed early","code":200},"requestReceivedTimestamp":"2024-07-30T16:09:28.502635Z","stageTimestamp":"2024-07-30T16:09:28.503017Z","annotations":{"authorization.k8s.io/decision":"allow","authorization.k8s.io/reason":""}}
{"kind":"Event","apiVersion":"audit.k8s.io/v1","level":"Metadata","auditID":"2fbf4326-31d4-4b2f-b214-69aca0023108","stage":"ResponseStarted","requestURI":"/api/v1/namespaces?timeout=3s\u0026timeoutSeconds=3\u0026watch=true","verb":"watch","user":{"username":"system:apiserver","uid":"16aecab5-afd7-4b65-a9de-b19b0cf300f1","groups":["system:masters"]},"sourceIPs":["127.0.0.1"],"userAgent":"TestAuditWatch","objectRef":{"resource":"namespaces","apiVersion":"v1"},"responseStatus":{"metadata":{},"status":"Success","message":"Connection closed early","code":200},"requestReceivedTimestamp":"2024-07-30T16:09:28.502635Z","stageTimestamp":"2024-07-30T16:09:28.503044Z","annotations":{"authorization.k8s.io/decision":"allow","authorization.k8s.io/reason":""}}
```

### What did you expect to happen?

The generated audit log events should look the same as they do with the feature disabled:

```json
{"kind":"Event","apiVersion":"audit.k8s.io/v1","level":"Metadata","auditID":"a7ff49e7-bf12-4dcb-bc0c-09aec5c3ee44","stage":"RequestReceived","requestURI":"/api/v1/namespaces?timeout=3s\u0026timeoutSeconds=3\u0026watch=true","verb":"watch","user":{"username":"system:apiserver","uid":"3a730033-bfdc-488f-aec1-5087d05e9a10","groups":["system:masters"]},"sourceIPs":["127.0.0.1"],"userAgent":"TestAuditWatch","objectRef":{"resource":"namespaces","apiVersion":"v1"},"requestReceivedTimestamp":"2024-07-30T16:11:57.119863Z","stageTimestamp":"2024-07-30T16:11:57.119863Z"}
{"kind":"Event","apiVersion":"audit.k8s.io/v1","level":"Metadata","auditID":"a7ff49e7-bf12-4dcb-bc0c-09aec5c3ee44","stage":"ResponseStarted","requestURI":"/api/v1/namespaces?timeout=3s\u0026timeoutSeconds=3\u0026watch=true","verb":"watch","user":{"username":"system:apiserver","uid":"3a730033-bfdc-488f-aec1-5087d05e9a10","groups":["system:masters"]},"sourceIPs":["127.0.0.1"],"userAgent":"TestAuditWatch","objectRef":{"resource":"namespaces","apiVersion":"v1"},"responseStatus":{"metadata":{},"code":200},"requestReceivedTimestamp":"2024-07-30T16:11:57.119863Z","stageTimestamp":"2024-07-30T16:11:57.120172Z","annotations":{"authorization.k8s.io/decision":"allow","authorization.k8s.io/reason":""}}
{"kind":"Event","apiVersion":"audit.k8s.io/v1","level":"Metadata","auditID":"a7ff49e7-bf12-4dcb-bc0c-09aec5c3ee44","stage":"ResponseComplete","requestURI":"/api/v1/namespaces?timeout=3s\u0026timeoutSeconds=3\u0026watch=true","verb":"watch","user":{"username":"system:apiserver","uid":"3a730033-bfdc-488f-aec1-5087d05e9a10","groups":["system:masters"]},"sourceIPs":["127.0.0.1"],"userAgent":"TestAuditWatch","objectRef":{"resource":"namespaces","apiVersion":"v1"},"responseStatus":{"metadata":{},"code":200},"requestReceivedTimestamp":"2024-07-30T16:11:57.119863Z","stageTimestamp":"2024-07-30T16:12:00.120388Z","annotations":{"authorization.k8s.io/decision":"allow","authorization.k8s.io/reason":""}}
```

### How can we reproduce it (as minimally and precisely as possible)?

I wrote a brief integration "test" to exercise this:

```go
func TestAuditWatch(t *testing.T) {
	featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultMutableFeatureGate, features.APIServingWithRoutine, true)

	logFile, err := os.CreateTemp("", "audit.log")
	if err != nil {
		t.Fatalf("Failed to create audit log file: %v", err)
	}
	defer utiltesting.CloseAndRemove(t, logFile)

	policyFile, err := os.CreateTemp("", "audit-policy.yaml")
	if err != nil {
		t.Fatalf("Failed to create audit policy file: %v", err)
	}
	defer utiltesting.CloseAndRemove(t, policyFile)

	if _, err := io.Copy(policyFile, strings.NewReader(`
apiVersion: audit.k8s.io/v1
kind: Policy
rules:
- level: Metadata
`)); err != nil {
		t.Fatal(err)
	}

	result := kubeapiservertesting.StartTestServerOrDie(t, nil,
		[]string{
			"--audit-log-mode", "blocking",
			"--audit-log-path", logFile.Name(),
			"--audit-log-version", "audit.k8s.io/v1",
			"--audit-policy-file", policyFile.Name(),
		},
		framework.SharedEtcd())
	defer result.TearDownFn()

	cfg := rest.CopyConfig(result.ClientConfig)
	cfg.UserAgent = "TestAuditWatch"
	kubeclient, err := clientset.NewForConfig(cfg)
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}

	w, err := kubeclient.CoreV1().Namespaces().Watch(context.Background(), metav1.ListOptions{TimeoutSeconds: ptr.To(int64(3))})
	if err != nil {
		t.Fatal(err)
	}
	defer w.Stop()
	for range w.ResultChan() {
		t.Log("ate watch event")
		// eat
	}

	dec := audit.Codecs.UniversalDecoder(auditv1.SchemeGroupVersion)
	scanner := bufio.NewScanner(logFile)
	for scanner.Scan() {
		var event auditinternal.Event
		if err := runtime.DecodeInto(dec, scanner.Bytes(), &event); err != nil {
			t.Fatal(err)
		}
		if event.UserAgent != cfg.UserAgent {
			continue
		}
		t.Log(string(scanner.Bytes()))
	}
	if scanner.Err() != nil {
		t.Fatal(err)
	}
}
```

### Anything else we need to know?

This could be patched in a similar way to the httplog issue, but for full parity there would also need to be a mechanism to recover from panics that originate from the deferred invocation of the task.

I'm also concerned about the effect of this feature on the priority and fairness filter, which releases the "seats" held during watch initialization via a deferred function.

/sig api-machinery
/sig scalability

### Kubernetes version

<details>

```console
$ kubectl version
# paste output here
```

</details>


### Cloud provider

<details>

</details>


### OS version

<details>

```console
# On Linux:
$ cat /etc/os-release
# paste output here
$ uname -a
# paste output here

# On Windows:
C:\> wmic os get Caption, Version, BuildNumber, OSArchitecture
# paste output here
```

</details>


### Install tools

<details>

</details>


### Container runtime (CRI) and version (if applicable)

<details>

</details>


### Related plugins (CNI, CSI, ...) and versions (if applicable)

<details>

</details>