From e5e557e90257d5bc69f1fabf253f87491e0868b2 Mon Sep 17 00:00:00 2001 From: Jordan Liggitt Date: Tue, 14 Apr 2020 16:10:54 -0400 Subject: [PATCH] apiserver: add API server plumbing for adding warnings --- pkg/features/kube_features.go | 1 + pkg/kubeapiserver/server/insecure_handler.go | 1 + .../pkg/endpoints/filters/warning.go | 69 ++++++++++++ .../pkg/endpoints/filters/warning_test.go | 100 ++++++++++++++++++ .../apiserver/pkg/endpoints/installer.go | 35 ++++++ .../pkg/endpoints/warning/warning.go | 39 +++++++ .../apiserver/pkg/features/kube_features.go | 7 ++ .../src/k8s.io/apiserver/pkg/server/config.go | 1 + .../k8s.io/apiserver/pkg/warning/context.go | 59 +++++++++++ 9 files changed, 312 insertions(+) create mode 100644 staging/src/k8s.io/apiserver/pkg/endpoints/filters/warning.go create mode 100644 staging/src/k8s.io/apiserver/pkg/endpoints/filters/warning_test.go create mode 100644 staging/src/k8s.io/apiserver/pkg/endpoints/warning/warning.go create mode 100644 staging/src/k8s.io/apiserver/pkg/warning/context.go diff --git a/pkg/features/kube_features.go b/pkg/features/kube_features.go index a5e708268b8..645b4499d20 100644 --- a/pkg/features/kube_features.go +++ b/pkg/features/kube_features.go @@ -679,6 +679,7 @@ var defaultKubernetesFeatureGates = map[featuregate.Feature]featuregate.FeatureS genericfeatures.DryRun: {Default: true, PreRelease: featuregate.GA}, genericfeatures.ServerSideApply: {Default: true, PreRelease: featuregate.Beta}, genericfeatures.APIPriorityAndFairness: {Default: false, PreRelease: featuregate.Alpha}, + genericfeatures.WarningHeaders: {Default: true, PreRelease: featuregate.Beta}, // features that enable backwards compatibility but are scheduled to be removed // ... diff --git a/pkg/kubeapiserver/server/insecure_handler.go b/pkg/kubeapiserver/server/insecure_handler.go index c3b87bc9edb..ce8c387b847 100644 --- a/pkg/kubeapiserver/server/insecure_handler.go +++ b/pkg/kubeapiserver/server/insecure_handler.go @@ -41,6 +41,7 @@ func BuildInsecureHandlerChain(apiHandler http.Handler, c *server.Config) http.H handler = genericfilters.WithTimeoutForNonLongRunningRequests(handler, c.LongRunningFunc, c.RequestTimeout) handler = genericfilters.WithWaitGroup(handler, c.LongRunningFunc, c.HandlerChainWaitGroup) handler = genericapifilters.WithRequestInfo(handler, server.NewRequestInfoResolver(c)) + handler = genericapifilters.WithWarningRecorder(handler) handler = genericapifilters.WithCacheControl(handler) handler = genericfilters.WithPanicRecovery(handler) diff --git a/staging/src/k8s.io/apiserver/pkg/endpoints/filters/warning.go b/staging/src/k8s.io/apiserver/pkg/endpoints/filters/warning.go new file mode 100644 index 00000000000..323f634dcbf --- /dev/null +++ b/staging/src/k8s.io/apiserver/pkg/endpoints/filters/warning.go @@ -0,0 +1,69 @@ +/* +Copyright 2020 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 ( + "net/http" + "sync" + + "k8s.io/apimachinery/pkg/util/net" + "k8s.io/apiserver/pkg/warning" +) + +// WithWarningRecorder attaches a deduplicating k8s.io/apiserver/pkg/warning#WarningRecorder to the request context. +func WithWarningRecorder(handler http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + recorder := &recorder{writer: w} + req = req.WithContext(warning.WithWarningRecorder(req.Context(), recorder)) + handler.ServeHTTP(w, req) + }) +} + +type recorder struct { + lock sync.Mutex + recorded map[string]bool + writer http.ResponseWriter +} + +func (r *recorder) AddWarning(agent, text string) { + if len(text) == 0 { + return + } + + r.lock.Lock() + defer r.lock.Unlock() + + // init if needed + if r.recorded == nil { + r.recorded = map[string]bool{} + } + + // dedupe if already warned + if r.recorded[text] { + return + } + r.recorded[text] = true + + // TODO(liggitt): track total message characters written: + // * if this takes us over 4k truncate individual messages to 256 chars and regenerate headers + // * if we're already truncating truncate this message to 256 chars + // * if we're still over 4k omit this message + + if header, err := net.NewWarningHeader(299, agent, text); err == nil { + r.writer.Header().Add("Warning", header) + } +} diff --git a/staging/src/k8s.io/apiserver/pkg/endpoints/filters/warning_test.go b/staging/src/k8s.io/apiserver/pkg/endpoints/filters/warning_test.go new file mode 100644 index 00000000000..2198506a913 --- /dev/null +++ b/staging/src/k8s.io/apiserver/pkg/endpoints/filters/warning_test.go @@ -0,0 +1,100 @@ +/* +Copyright 2020 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 ( + "net/http/httptest" + "reflect" + "testing" +) + +func Test_recorder_AddWarning(t *testing.T) { + type args struct { + agent string + text string + } + tests := []struct { + name string + args []args + expect []string + }{ + { + name: "empty", + args: []args{}, + expect: nil, + }, + { + name: "empty values", + args: []args{{agent: "", text: ""}}, + expect: nil, + }, + { + name: "empty agent", + args: []args{{agent: "", text: "mytext"}}, + expect: []string{`299 - "mytext"`}, + }, + { + name: "simple", + args: []args{{agent: "myagent", text: "mytext"}}, + expect: []string{`299 myagent "mytext"`}, + }, + { + name: "duplicate text", + args: []args{ + {agent: "myagent", text: "mytext"}, + {agent: "myagent2", text: "mytext"}, + }, + expect: []string{`299 myagent "mytext"`}, + }, + { + name: "multiple", + args: []args{ + {agent: "", text: "mytext1"}, + {agent: "", text: "mytext2"}, + {agent: "", text: "mytext3"}, + }, + expect: []string{ + `299 - "mytext1"`, + `299 - "mytext2"`, + `299 - "mytext3"`, + }, + }, + + { + name: "invalid agent", + args: []args{{agent: "my agent", text: "mytext"}}, + expect: nil, + }, + { + name: "invalid text", + args: []args{{agent: "myagent", text: "my\ntext"}}, + expect: nil, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + responseRecorder := httptest.NewRecorder() + warningRecorder := &recorder{writer: responseRecorder} + for _, arg := range tt.args { + warningRecorder.AddWarning(arg.agent, arg.text) + } + if !reflect.DeepEqual(tt.expect, responseRecorder.Header()["Warning"]) { + t.Errorf("expected\n%#v\ngot\n%#v", tt.expect, responseRecorder.Header()["Warning"]) + } + }) + } +} diff --git a/staging/src/k8s.io/apiserver/pkg/endpoints/installer.go b/staging/src/k8s.io/apiserver/pkg/endpoints/installer.go index d743d44ba67..2ab3f01f964 100644 --- a/staging/src/k8s.io/apiserver/pkg/endpoints/installer.go +++ b/staging/src/k8s.io/apiserver/pkg/endpoints/installer.go @@ -39,6 +39,7 @@ import ( "k8s.io/apiserver/pkg/endpoints/handlers/fieldmanager" "k8s.io/apiserver/pkg/endpoints/handlers/negotiation" "k8s.io/apiserver/pkg/endpoints/metrics" + utilwarning "k8s.io/apiserver/pkg/endpoints/warning" "k8s.io/apiserver/pkg/features" "k8s.io/apiserver/pkg/registry/rest" utilfeature "k8s.io/apiserver/pkg/util/feature" @@ -622,6 +623,10 @@ func (a *APIInstaller) registerResourceHandlers(path string, storage rest.Storag verbOverrider, needOverride := storage.(StorageMetricsOverride) + // accumulate endpoint-level warnings + enableWarningHeaders := utilfeature.DefaultFeatureGate.Enabled(features.WarningHeaders) + warnings := []string{} + switch action.Verb { case "GET": // Get a resource. var handler restful.RouteFunction @@ -637,6 +642,9 @@ func (a *APIInstaller) registerResourceHandlers(path string, storage rest.Storag } else { handler = metrics.InstrumentRouteFunc(action.Verb, group, version, resource, subresource, requestScope, metrics.APIServerComponent, handler) } + if enableWarningHeaders { + handler = utilwarning.AddWarningsHandler(handler, warnings) + } doc := "read the specified " + kind if isSubresource { @@ -667,6 +675,9 @@ func (a *APIInstaller) registerResourceHandlers(path string, storage rest.Storag doc = "list " + subresource + " of objects of kind " + kind } handler := metrics.InstrumentRouteFunc(action.Verb, group, version, resource, subresource, requestScope, metrics.APIServerComponent, restfulListResource(lister, watcher, reqScope, false, a.minRequestTimeout)) + if enableWarningHeaders { + handler = utilwarning.AddWarningsHandler(handler, warnings) + } route := ws.GET(action.Path).To(handler). Doc(doc). Param(ws.QueryParameter("pretty", "If 'true', then the output is pretty printed.")). @@ -699,6 +710,9 @@ func (a *APIInstaller) registerResourceHandlers(path string, storage rest.Storag doc = "replace " + subresource + " of the specified " + kind } handler := metrics.InstrumentRouteFunc(action.Verb, group, version, resource, subresource, requestScope, metrics.APIServerComponent, restfulUpdateResource(updater, reqScope, admit)) + if enableWarningHeaders { + handler = utilwarning.AddWarningsHandler(handler, warnings) + } route := ws.PUT(action.Path).To(handler). Doc(doc). Param(ws.QueryParameter("pretty", "If 'true', then the output is pretty printed.")). @@ -729,6 +743,9 @@ func (a *APIInstaller) registerResourceHandlers(path string, storage rest.Storag supportedTypes = append(supportedTypes, string(types.ApplyPatchType)) } handler := metrics.InstrumentRouteFunc(action.Verb, group, version, resource, subresource, requestScope, metrics.APIServerComponent, restfulPatchResource(patcher, reqScope, admit, supportedTypes)) + if enableWarningHeaders { + handler = utilwarning.AddWarningsHandler(handler, warnings) + } route := ws.PATCH(action.Path).To(handler). Doc(doc). Param(ws.QueryParameter("pretty", "If 'true', then the output is pretty printed.")). @@ -751,6 +768,9 @@ func (a *APIInstaller) registerResourceHandlers(path string, storage rest.Storag handler = restfulCreateResource(creater, reqScope, admit) } handler = metrics.InstrumentRouteFunc(action.Verb, group, version, resource, subresource, requestScope, metrics.APIServerComponent, handler) + if enableWarningHeaders { + handler = utilwarning.AddWarningsHandler(handler, warnings) + } article := GetArticleForNoun(kind, " ") doc := "create" + article + kind if isSubresource { @@ -784,6 +804,9 @@ func (a *APIInstaller) registerResourceHandlers(path string, storage rest.Storag deleteReturnType = producedObject } handler := metrics.InstrumentRouteFunc(action.Verb, group, version, resource, subresource, requestScope, metrics.APIServerComponent, restfulDeleteResource(gracefulDeleter, isGracefulDeleter, reqScope, admit)) + if enableWarningHeaders { + handler = utilwarning.AddWarningsHandler(handler, warnings) + } route := ws.DELETE(action.Path).To(handler). Doc(doc). Param(ws.QueryParameter("pretty", "If 'true', then the output is pretty printed.")). @@ -807,6 +830,9 @@ func (a *APIInstaller) registerResourceHandlers(path string, storage rest.Storag doc = "delete collection of " + subresource + " of a " + kind } handler := metrics.InstrumentRouteFunc(action.Verb, group, version, resource, subresource, requestScope, metrics.APIServerComponent, restfulDeleteCollection(collectionDeleter, isCollectionDeleter, reqScope, admit)) + if enableWarningHeaders { + handler = utilwarning.AddWarningsHandler(handler, warnings) + } route := ws.DELETE(action.Path).To(handler). Doc(doc). Param(ws.QueryParameter("pretty", "If 'true', then the output is pretty printed.")). @@ -834,6 +860,9 @@ func (a *APIInstaller) registerResourceHandlers(path string, storage rest.Storag } doc += ". deprecated: use the 'watch' parameter with a list operation instead, filtered to a single item with the 'fieldSelector' parameter." handler := metrics.InstrumentRouteFunc(action.Verb, group, version, resource, subresource, requestScope, metrics.APIServerComponent, restfulListResource(lister, watcher, reqScope, true, a.minRequestTimeout)) + if enableWarningHeaders { + handler = utilwarning.AddWarningsHandler(handler, warnings) + } route := ws.GET(action.Path).To(handler). Doc(doc). Param(ws.QueryParameter("pretty", "If 'true', then the output is pretty printed.")). @@ -854,6 +883,9 @@ func (a *APIInstaller) registerResourceHandlers(path string, storage rest.Storag } doc += ". deprecated: use the 'watch' parameter with a list operation instead." handler := metrics.InstrumentRouteFunc(action.Verb, group, version, resource, subresource, requestScope, metrics.APIServerComponent, restfulListResource(lister, watcher, reqScope, true, a.minRequestTimeout)) + if enableWarningHeaders { + handler = utilwarning.AddWarningsHandler(handler, warnings) + } route := ws.GET(action.Path).To(handler). Doc(doc). Param(ws.QueryParameter("pretty", "If 'true', then the output is pretty printed.")). @@ -877,6 +909,9 @@ func (a *APIInstaller) registerResourceHandlers(path string, storage rest.Storag doc = "connect " + method + " requests to " + subresource + " of " + kind } handler := metrics.InstrumentRouteFunc(action.Verb, group, version, resource, subresource, requestScope, metrics.APIServerComponent, restfulConnectResource(connecter, reqScope, admit, path, isSubresource)) + if enableWarningHeaders { + handler = utilwarning.AddWarningsHandler(handler, warnings) + } route := ws.Method(method).Path(action.Path). To(handler). Doc(doc). diff --git a/staging/src/k8s.io/apiserver/pkg/endpoints/warning/warning.go b/staging/src/k8s.io/apiserver/pkg/endpoints/warning/warning.go new file mode 100644 index 00000000000..c0dde6e06dc --- /dev/null +++ b/staging/src/k8s.io/apiserver/pkg/endpoints/warning/warning.go @@ -0,0 +1,39 @@ +/* +Copyright 2020 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 warning + +import ( + restful "github.com/emicklei/go-restful" + + "k8s.io/apiserver/pkg/warning" +) + +// AddWarningsHandler returns a handler that adds the provided warnings to all requests, +// then delegates to the provided handler. +func AddWarningsHandler(handler restful.RouteFunction, warnings []string) restful.RouteFunction { + if len(warnings) == 0 { + return handler + } + + return func(req *restful.Request, res *restful.Response) { + ctx := req.Request.Context() + for _, msg := range warnings { + warning.AddWarning(ctx, "", msg) + } + handler(req, res) + } +} diff --git a/staging/src/k8s.io/apiserver/pkg/features/kube_features.go b/staging/src/k8s.io/apiserver/pkg/features/kube_features.go index 3fb19760b62..9869e60c265 100644 --- a/staging/src/k8s.io/apiserver/pkg/features/kube_features.go +++ b/staging/src/k8s.io/apiserver/pkg/features/kube_features.go @@ -143,6 +143,12 @@ const ( // // Allows label and field based indexes in apiserver watch cache to accelerate list operations. SelectorIndex featuregate.Feature = "SelectorIndex" + + // owner: @liggitt + // beta: v1.19 + // + // Allows sending warning headers in API responses. + WarningHeaders featuregate.Feature = "WarningHeaders" ) func init() { @@ -168,4 +174,5 @@ var defaultKubernetesFeatureGates = map[featuregate.Feature]featuregate.FeatureS APIPriorityAndFairness: {Default: false, PreRelease: featuregate.Alpha}, RemoveSelfLink: {Default: false, PreRelease: featuregate.Alpha}, SelectorIndex: {Default: false, PreRelease: featuregate.Alpha}, + WarningHeaders: {Default: true, PreRelease: featuregate.Beta}, } diff --git a/staging/src/k8s.io/apiserver/pkg/server/config.go b/staging/src/k8s.io/apiserver/pkg/server/config.go index 930ad28403d..ed680f8d5e5 100644 --- a/staging/src/k8s.io/apiserver/pkg/server/config.go +++ b/staging/src/k8s.io/apiserver/pkg/server/config.go @@ -678,6 +678,7 @@ func DefaultBuildHandlerChain(apiHandler http.Handler, c *Config) http.Handler { handler = genericfilters.WithProbabilisticGoaway(handler, c.GoawayChance) } handler = genericapifilters.WithAuditAnnotations(handler, c.AuditBackend, c.AuditPolicyChecker) + handler = genericapifilters.WithWarningRecorder(handler) handler = genericapifilters.WithCacheControl(handler) handler = genericfilters.WithPanicRecovery(handler) return handler diff --git a/staging/src/k8s.io/apiserver/pkg/warning/context.go b/staging/src/k8s.io/apiserver/pkg/warning/context.go new file mode 100644 index 00000000000..1b9dd54dfc0 --- /dev/null +++ b/staging/src/k8s.io/apiserver/pkg/warning/context.go @@ -0,0 +1,59 @@ +/* +Copyright 2020 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 warning + +import ( + "context" +) + +// The key type is unexported to prevent collisions +type key int + +const ( + // auditAnnotationsKey is the context key for the audit annotations. + warningRecorderKey key = iota +) + +// Recorder provides a method for recording warnings +type Recorder interface { + // AddWarning adds the specified warning to the response. + // agent must be valid UTF-8, and must not contain spaces, quotes, backslashes, or control characters. + // text must be valid UTF-8, and must not contain control characters. + AddWarning(agent, text string) +} + +// WithWarningRecorder returns a new context that wraps the provided context and contains the provided Recorder implementation. +// The returned context can be passed to AddWarning(). +func WithWarningRecorder(ctx context.Context, recorder Recorder) context.Context { + return context.WithValue(ctx, warningRecorderKey, recorder) +} +func warningRecorderFrom(ctx context.Context) (Recorder, bool) { + recorder, ok := ctx.Value(warningRecorderKey).(Recorder) + return recorder, ok +} + +// AddWarning records a warning for the specified agent and text to the Recorder added to the provided context using WithWarningRecorder(). +// If no Recorder exists in the provided context, this is a no-op. +// agent must be valid UTF-8, and must not contain spaces, quotes, backslashes, or control characters. +// text must be valid UTF-8, and must not contain control characters. +func AddWarning(ctx context.Context, agent string, text string) { + recorder, ok := warningRecorderFrom(ctx) + if !ok { + return + } + recorder.AddWarning(agent, text) +}