Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #114925 from tkashem/watch-termination
apiserver: terminate watch with rate limiting during shutdown
- Loading branch information
Showing
10 changed files
with
606 additions
and
61 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
55 changes: 55 additions & 0 deletions
55
staging/src/k8s.io/apiserver/pkg/endpoints/request/server_shutdown_signal.go
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,55 @@ | ||
| /* | ||
| 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 request | ||
|
|
||
| import ( | ||
| "context" | ||
| ) | ||
|
|
||
| // The serverShutdownSignalKeyType type is unexported to prevent collisions | ||
| type serverShutdownSignalKeyType int | ||
|
|
||
| // serverShutdownSignalKey is the context key for storing the | ||
| // watch termination interface instance for a WATCH request. | ||
| const serverShutdownSignalKey serverShutdownSignalKeyType = iota | ||
|
|
||
| // ServerShutdownSignal is associated with the request context so | ||
| // the request handler logic has access to signals rlated to | ||
| // the server shutdown events | ||
| type ServerShutdownSignal interface { | ||
| // Signaled when the apiserver is not receiving any new request | ||
| ShuttingDown() <-chan struct{} | ||
| } | ||
|
|
||
| // ServerShutdownSignalFrom returns the ServerShutdownSignal instance | ||
| // associated with the request context. | ||
| // If there is no ServerShutdownSignal asscoaied with the context, | ||
| // nil is returned. | ||
| func ServerShutdownSignalFrom(ctx context.Context) ServerShutdownSignal { | ||
| ev, _ := ctx.Value(serverShutdownSignalKey).(ServerShutdownSignal) | ||
| return ev | ||
| } | ||
|
|
||
| // WithServerShutdownSignal returns a new context that stores | ||
| // the ServerShutdownSignal interface instance. | ||
| func WithServerShutdownSignal(parent context.Context, window ServerShutdownSignal) context.Context { | ||
| if ServerShutdownSignalFrom(parent) != nil { | ||
| return parent // Avoid double registering. | ||
| } | ||
|
|
||
| return context.WithValue(parent, serverShutdownSignalKey, window) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
62 changes: 62 additions & 0 deletions
62
staging/src/k8s.io/apiserver/pkg/server/filters/watch_termination.go
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,62 @@ | ||
| /* | ||
| 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 filters | ||
|
|
||
| import ( | ||
| "errors" | ||
| "net/http" | ||
|
|
||
| "k8s.io/apiserver/pkg/endpoints/handlers/responsewriters" | ||
| apirequest "k8s.io/apiserver/pkg/endpoints/request" | ||
| "k8s.io/klog/v2" | ||
| ) | ||
|
|
||
| func WithWatchTerminationDuringShutdown(handler http.Handler, termination apirequest.ServerShutdownSignal, wg RequestWaitGroup) http.Handler { | ||
| if termination == nil || wg == nil { | ||
| klog.Warningf("watch termination during shutdown not attached to the handler chain") | ||
| return handler | ||
| } | ||
| return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { | ||
| ctx := req.Context() | ||
| requestInfo, ok := apirequest.RequestInfoFrom(ctx) | ||
| if !ok { | ||
| // if this happens, the handler chain isn't setup correctly because there is no request info | ||
| responsewriters.InternalError(w, req, errors.New("no RequestInfo found in the context")) | ||
| return | ||
| } | ||
| if !watchVerbs.Has(requestInfo.Verb) { | ||
| handler.ServeHTTP(w, req) | ||
| return | ||
| } | ||
|
|
||
| if err := wg.Add(1); err != nil { | ||
| // When apiserver is shutting down, signal clients to retry | ||
| // There is a good chance the client hit a different server, so a tight retry is good for client responsiveness. | ||
| waitGroupWriteRetryAfterToResponse(w) | ||
| return | ||
| } | ||
|
|
||
| // attach ServerShutdownSignal to the watch request so that the | ||
| // watch handler loop can return as soon as the server signals | ||
| // that it is shutting down. | ||
| ctx = apirequest.WithServerShutdownSignal(req.Context(), termination) | ||
| req = req.WithContext(ctx) | ||
|
|
||
| defer wg.Done() | ||
| handler.ServeHTTP(w, req) | ||
| }) | ||
| } |
Oops, something went wrong.