github
/
k3s
mirror of https://github.com/k3s-io/k3s
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Remove APIResponseCompression

k3s-v1.14.6
Darren Shepherd 2018-10-05 11:48:16 -07:00 committed by Erik Wilson
parent ee7e4836df
commit a92b46730e
8 changed files with 5 additions and 327 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
pkg/features/kube_features.go
View File

@ -403,7 +403,6 @@ var defaultKubernetesFeatureGates = map[utilfeature.Feature]utilfeature.FeatureS
genericfeatures.ValidateProxyRedirects: {Default: true, PreRelease: utilfeature.Beta},
genericfeatures.AdvancedAuditing: {Default: true, PreRelease: utilfeature.GA},
genericfeatures.DynamicAuditing: {Default: false, PreRelease: utilfeature.Alpha},
genericfeatures.APIResponseCompression: {Default: false, PreRelease: utilfeature.Alpha},
genericfeatures.APIListChunking: {Default: true, PreRelease: utilfeature.Beta},
genericfeatures.DryRun: {Default: true, PreRelease: utilfeature.Beta},
genericfeatures.ServerSideApply: {Default: false, PreRelease: utilfeature.Alpha},

5
staging/src/k8s.io/apiserver/pkg/endpoints/groupversion.go
View File

@ -81,10 +81,6 @@ type APIGroupVersion struct {
MinRequestTimeout time.Duration
// EnableAPIResponseCompression indicates whether API Responses should support compression
// if the client requests it via Accept-Encoding
EnableAPIResponseCompression bool
// OpenAPIModels exposes the OpenAPI models to each individual handler.
OpenAPIModels openapiproto.Models
@ -102,7 +98,6 @@ func (g *APIGroupVersion) InstallREST(container *restful.Container) error {
group: g,
prefix: prefix,
minRequestTimeout: g.MinRequestTimeout,
enableAPIResponseCompression: g.EnableAPIResponseCompression,
}
apiResources, ws, registrationErrors := installer.Install()

8
staging/src/k8s.io/apiserver/pkg/endpoints/installer.go
View File

@ -40,7 +40,6 @@ import (
"k8s.io/apiserver/pkg/endpoints/metrics"
"k8s.io/apiserver/pkg/features"
"k8s.io/apiserver/pkg/registry/rest"
genericfilters "k8s.io/apiserver/pkg/server/filters"
utilfeature "k8s.io/apiserver/pkg/util/feature"
)
@ -53,7 +52,6 @@ type APIInstaller struct {
group *APIGroupVersion
prefix string // Path prefix where API resources are to be registered.
minRequestTimeout time.Duration
enableAPIResponseCompression bool
}
// Struct capturing information about an action ("GET", "POST", "WATCH", "PROXY", etc).
@ -623,9 +621,6 @@ func (a *APIInstaller) registerResourceHandlers(path string, storage rest.Storag
handler = metrics.InstrumentRouteFunc(action.Verb, group, version, resource, subresource, requestScope, metrics.APIServerComponent, handler)
}
if a.enableAPIResponseCompression {
handler = genericfilters.RestfulWithCompression(handler)
}
doc := "read the specified " + kind
if isSubresource {
doc = "read " + subresource + " of the specified " + kind
@ -655,9 +650,6 @@ 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 a.enableAPIResponseCompression {
handler = genericfilters.RestfulWithCompression(handler)
}
route := ws.GET(action.Path).To(handler).
Doc(doc).
Param(ws.QueryParameter("pretty", "If 'true', then the output is pretty printed.")).

7
staging/src/k8s.io/apiserver/pkg/features/kube_features.go
View File

@ -60,12 +60,6 @@ const (
// AuditSink API object.
DynamicAuditing utilfeature.Feature = "DynamicAuditing"
// owner: @ilackams
// alpha: v1.7
//
// Enables compression of REST responses (GET and LIST only)
APIResponseCompression utilfeature.Feature = "APIResponseCompression"
// owner: @smarterclayton
// alpha: v1.8
// beta: v1.9
@ -121,7 +115,6 @@ var defaultKubernetesFeatureGates = map[utilfeature.Feature]utilfeature.FeatureS
ValidateProxyRedirects: {Default: true, PreRelease: utilfeature.Beta},
AdvancedAuditing: {Default: true, PreRelease: utilfeature.GA},
DynamicAuditing: {Default: false, PreRelease: utilfeature.Alpha},
APIResponseCompression: {Default: false, PreRelease: utilfeature.Alpha},
APIListChunking: {Default: true, PreRelease: utilfeature.Beta},
DryRun: {Default: true, PreRelease: utilfeature.Beta},
ServerSideApply: {Default: false, PreRelease: utilfeature.Alpha},

9
staging/src/k8s.io/apiserver/pkg/server/config.go
View File

@ -53,13 +53,11 @@ import (
genericapifilters "k8s.io/apiserver/pkg/endpoints/filters"
apiopenapi "k8s.io/apiserver/pkg/endpoints/openapi"
apirequest "k8s.io/apiserver/pkg/endpoints/request"
"k8s.io/apiserver/pkg/features"
genericregistry "k8s.io/apiserver/pkg/registry/generic"
genericfilters "k8s.io/apiserver/pkg/server/filters"
"k8s.io/apiserver/pkg/server/healthz"
"k8s.io/apiserver/pkg/server/routes"
serverstore "k8s.io/apiserver/pkg/server/storage"
utilfeature "k8s.io/apiserver/pkg/util/feature"
"k8s.io/client-go/informers"
restclient "k8s.io/client-go/rest"
certutil "k8s.io/client-go/util/cert"
@ -171,10 +169,6 @@ type Config struct {
// Predicate which is true for paths of long-running http requests
LongRunningFunc apirequest.LongRunningRequestCheck
// EnableAPIResponseCompression indicates whether API Responses should support compression
// if the client requests it via Accept-Encoding
EnableAPIResponseCompression bool
// MergedResourceConfig indicates which groupVersion enabled and its resources enabled/disabled.
// This is composed of genericapiserver defaultAPIResourceConfig and those parsed from flags.
// If not specify any in flags, then genericapiserver will only enable defaultAPIResourceConfig.
@ -282,7 +276,6 @@ func NewConfig(codecs serializer.CodecFactory) *Config {
// the size ratio is 10:1, we set 100MB as the largest request
// body size to be accepted and decoded in a write request.
MaxRequestBodyBytes: int64(100 * 1024 * 1024),
EnableAPIResponseCompression: utilfeature.DefaultFeatureGate.Enabled(features.APIResponseCompression),
// Default to treating watch as a long-running operation
// Generic API servers have no inherent long-running subresources
@ -472,8 +465,6 @@ func (c completedConfig) New(name string, delegationTarget DelegationTarget) (*G
healthzChecks: c.HealthzChecks,
DiscoveryGroupManager: discovery.NewRootAPIsHandler(c.DiscoveryAddresses, c.Serializer),
enableAPIResponseCompression: c.EnableAPIResponseCompression,
maxRequestBodyBytes: c.MaxRequestBodyBytes,
}

181
staging/src/k8s.io/apiserver/pkg/server/filters/compression.go
View File

@ -1,181 +0,0 @@
/*
Copyright 2017 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 (
"compress/gzip"
"compress/zlib"
"errors"
"fmt"
"io"
"net/http"
"strings"
"github.com/emicklei/go-restful"
"k8s.io/apimachinery/pkg/util/runtime"
"k8s.io/apiserver/pkg/endpoints/request"
)
// Compressor is an interface to compression writers
type Compressor interface {
io.WriteCloser
Flush() error
}
const (
headerAcceptEncoding = "Accept-Encoding"
headerContentEncoding = "Content-Encoding"
encodingGzip = "gzip"
encodingDeflate = "deflate"
)
// WithCompression wraps an http.Handler with the Compression Handler
func WithCompression(handler http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
wantsCompression, encoding := wantsCompressedResponse(req)
w.Header().Set("Vary", "Accept-Encoding")
if wantsCompression {
compressionWriter, err := NewCompressionResponseWriter(w, encoding)
if err != nil {
handleError(w, req, err)
runtime.HandleError(fmt.Errorf("failed to compress HTTP response: %v", err))
return
}
compressionWriter.Header().Set("Content-Encoding", encoding)
handler.ServeHTTP(compressionWriter, req)
compressionWriter.(*compressionResponseWriter).Close()
} else {
handler.ServeHTTP(w, req)
}
})
}
// wantsCompressedResponse reads the Accept-Encoding header to see if and which encoding is requested.
func wantsCompressedResponse(req *http.Request) (bool, string) {
// don't compress watches
ctx := req.Context()
info, ok := request.RequestInfoFrom(ctx)
if !ok {
return false, ""
}
if !info.IsResourceRequest {
return false, ""
}
if info.Verb == "watch" {
return false, ""
}
header := req.Header.Get(headerAcceptEncoding)
gi := strings.Index(header, encodingGzip)
zi := strings.Index(header, encodingDeflate)
// use in order of appearance
switch {
case gi == -1:
return zi != -1, encodingDeflate
case zi == -1:
return gi != -1, encodingGzip
case gi < zi:
return true, encodingGzip
default:
return true, encodingDeflate
}
}
type compressionResponseWriter struct {
writer http.ResponseWriter
compressor Compressor
encoding string
}
// NewCompressionResponseWriter returns wraps w with a compression ResponseWriter, using the given encoding
func NewCompressionResponseWriter(w http.ResponseWriter, encoding string) (http.ResponseWriter, error) {
var compressor Compressor
switch encoding {
case encodingGzip:
compressor = gzip.NewWriter(w)
case encodingDeflate:
compressor = zlib.NewWriter(w)
default:
return nil, fmt.Errorf("%s is not a supported encoding type", encoding)
}
return &compressionResponseWriter{
writer: w,
compressor: compressor,
encoding: encoding,
}, nil
}
// compressionResponseWriter implements http.Responsewriter Interface
var _ http.ResponseWriter = &compressionResponseWriter{}
func (c *compressionResponseWriter) Header() http.Header {
return c.writer.Header()
}
// compress data according to compression method
func (c *compressionResponseWriter) Write(p []byte) (int, error) {
if c.compressorClosed() {
return -1, errors.New("compressing error: tried to write data using closed compressor")
}
c.Header().Set(headerContentEncoding, c.encoding)
defer c.compressor.Flush()
return c.compressor.Write(p)
}
func (c *compressionResponseWriter) WriteHeader(status int) {
c.writer.WriteHeader(status)
}
// CloseNotify is part of http.CloseNotifier interface
func (c *compressionResponseWriter) CloseNotify() <-chan bool {
return c.writer.(http.CloseNotifier).CloseNotify()
}
// Close the underlying compressor
func (c *compressionResponseWriter) Close() error {
if c.compressorClosed() {
return errors.New("Compressing error: tried to close already closed compressor")
}
c.compressor.Close()
c.compressor = nil
return nil
}
func (c *compressionResponseWriter) Flush() {
if c.compressorClosed() {
return
}
c.compressor.Flush()
}
func (c *compressionResponseWriter) compressorClosed() bool {
return nil == c.compressor
}
// RestfulWithCompression wraps WithCompression to be compatible with go-restful
func RestfulWithCompression(function restful.RouteFunction) restful.RouteFunction {
return restful.RouteFunction(func(request *restful.Request, response *restful.Response) {
handler := WithCompression(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
response.ResponseWriter = w
request.Request = req
function(request, response)
}))
handler.ServeHTTP(response.ResponseWriter, request.Request)
})
}

106
staging/src/k8s.io/apiserver/pkg/server/filters/compression_test.go
View File

@ -1,106 +0,0 @@
/*
Copyright 2017 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 (
"bytes"
"compress/gzip"
"io"
"io/ioutil"
"net/http"
"net/http/httptest"
"testing"
"k8s.io/apimachinery/pkg/util/sets"
"k8s.io/apiserver/pkg/endpoints/filters"
"k8s.io/apiserver/pkg/endpoints/request"
)
func TestCompression(t *testing.T) {
tests := []struct {
encoding string
watch bool
}{
{"", false},
{"gzip", true},
{"gzip", false},
}
responseData := []byte("1234")
for _, test := range tests {
handler := WithCompression(
http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
w.Write(responseData)
}),
)
handler = filters.WithRequestInfo(handler, newTestRequestInfoResolver())
server := httptest.NewServer(handler)
defer server.Close()
client := http.Client{
Transport: &http.Transport{
DisableCompression: true,
},
}
url := server.URL + "/api/v1/pods"
if test.watch {
url = url + "?watch=1"
}
request, err := http.NewRequest("GET", url, nil)
if err != nil {
t.Errorf("unexpected error: %v", err)
}
request.Header.Set("Accept-Encoding", test.encoding)
response, err := client.Do(request)
if err != nil {
t.Errorf("unexpected error: %v", err)
}
var reader io.Reader
if test.encoding == "gzip" && !test.watch {
if response.Header.Get("Content-Encoding") != "gzip" {
t.Fatal("expected response header Content-Encoding to be set to \"gzip\"")
}
if response.Header.Get("Vary") != "Accept-Encoding" {
t.Fatal("expected response header Vary to be set to \"Accept-Encoding\"")
}
reader, err = gzip.NewReader(response.Body)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
} else {
if response.Header.Get("Content-Encoding") == "gzip" {
t.Fatal("expected response header Content-Encoding not to be set")
}
reader = response.Body
}
body, err := ioutil.ReadAll(reader)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if !bytes.Equal(body, responseData) {
t.Fatalf("Expected response body %s to equal %s", body, responseData)
}
}
}
func newTestRequestInfoResolver() *request.RequestInfoFactory {
return &request.RequestInfoFactory{
APIPrefixes: sets.NewString("api", "apis"),
GrouplessAPIPrefixes: sets.NewString("api"),
}
}

5
staging/src/k8s.io/apiserver/pkg/server/genericapiserver.go
View File

@ -157,10 +157,6 @@ type GenericAPIServer struct {
// the create-on-update case
Authorizer authorizer.Authorizer
// enableAPIResponseCompression indicates whether API Responses should support compression
// if the client requests it via Accept-Encoding
enableAPIResponseCompression bool
// delegationTarget is the next delegate in the chain. This is never nil.
delegationTarget DelegationTarget
@ -466,7 +462,6 @@ func (s *GenericAPIServer) newAPIGroupVersion(apiGroupInfo *APIGroupInfo, groupV
Admit: s.admissionControl,
MinRequestTimeout: s.minRequestTimeout,
EnableAPIResponseCompression: s.enableAPIResponseCompression,
Authorizer: s.Authorizer,
}
}