mirror of https://github.com/k3s-io/k3s
Mike Danese
parent
c7414323d8
commit
7b4722964d
|
|
@ -72,7 +72,7 @@ spec:
|
|||
- name: GF_AUTH_ANONYMOUS_ORG_ROLE
|
||||
value: Admin
|
||||
- name: GF_SERVER_ROOT_URL
|
||||
value: /api/v1/proxy/namespaces/kube-system/services/monitoring-grafana/
|
||||
value: /api/v1/namespaces/kube-system/services/monitoring-grafana/proxy/
|
||||
ports:
|
||||
- name: ui
|
||||
containerPort: 3000
|
||||
|
|
|
|||
|
|
@ -30,7 +30,7 @@ spec:
|
|||
- name: ELASTICSEARCH_URL
|
||||
value: http://elasticsearch-logging:9200
|
||||
- name: SERVER_BASEPATH
|
||||
value: /api/v1/proxy/namespaces/kube-system/services/kibana-logging
|
||||
value: /api/v1/namespaces/kube-system/services/kibana-logging/proxy
|
||||
- name: XPACK_MONITORING_ENABLED
|
||||
value: "false"
|
||||
- name: XPACK_SECURITY_ENABLED
|
||||
|
|
|
|||
|
|
@ -72,10 +72,10 @@ fi
|
|||
|
||||
HEAPSTER_VERSION="v0.18.2"
|
||||
MASTER_PPROF_PATH=""
|
||||
HEAPSTER_PPROF_PATH="/api/v1/proxy/namespaces/kube-system/services/monitoring-heapster"
|
||||
HEAPSTER_PPROF_PATH="/api/v1/namespaces/kube-system/services/monitoring-heapster/proxy"
|
||||
KUBELET_PPROF_PATH_PREFIX="/api/v1/proxy/nodes"
|
||||
SCHEDULER_PPROF_PATH_PREFIX="/api/v1/proxy/namespaces/kube-system/pods/kube-scheduler"
|
||||
CONTROLLER_MANAGER_PPROF_PATH_PREFIX="/api/v1/proxy/namespaces/kube-system/pods/kube-controller-manager"
|
||||
SCHEDULER_PPROF_PATH_PREFIX="/api/v1/namespaces/kube-system/pods/kube-scheduler/proxy"
|
||||
CONTROLLER_MANAGER_PPROF_PATH_PREFIX="/api/v1/namespaces/kube-system/pods/kube-controller-manager/proxy"
|
||||
|
||||
eval set -- "${args}"
|
||||
|
||||
|
|
@ -292,7 +292,7 @@ for component in ${profile_components}; do
|
|||
|
||||
if [[ "${component}" == "kubelet" ]]; then
|
||||
for node in ${kubelet_addresses//[,;]/' '}; do
|
||||
grab_profiles_from_component "${requested_profiles}" "${mem_pprof_flags}" "${binary}" "${tunnel_port}" "${path}/${node}" "${output_dir}/${component}" "${timestamp}"
|
||||
grab_profiles_from_component "${requested_profiles}" "${mem_pprof_flags}" "${binary}" "${tunnel_port}" "${path}/${node}/proxy" "${output_dir}/${component}" "${timestamp}"
|
||||
done
|
||||
else
|
||||
grab_profiles_from_component "${requested_profiles}" "${mem_pprof_flags}" "${binary}" "${tunnel_port}" "${path}" "${output_dir}/${component}" "${timestamp}"
|
||||
|
|
|
|||
|
|
@ -12,7 +12,6 @@ go_test(
|
|||
"apiserver_test.go",
|
||||
"audit_test.go",
|
||||
"installer_test.go",
|
||||
"proxy_test.go",
|
||||
"watch_test.go",
|
||||
],
|
||||
embed = [":go_default_library"],
|
||||
|
|
@ -35,7 +34,6 @@ go_test(
|
|||
"//vendor/k8s.io/apimachinery/pkg/runtime/serializer/streaming:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/util/diff:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/util/net:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/util/wait:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/watch:go_default_library",
|
||||
|
|
|
|||
|
|
@ -594,26 +594,6 @@ func (storage *SimpleRESTStorage) Watcher() *watch.FakeWatcher {
|
|||
return storage.fakeWatch
|
||||
}
|
||||
|
||||
// Implement Redirector.
|
||||
var _ = rest.Redirector(&SimpleRESTStorage{})
|
||||
|
||||
// Implement Redirector.
|
||||
func (storage *SimpleRESTStorage) ResourceLocation(ctx request.Context, id string) (*url.URL, http.RoundTripper, error) {
|
||||
storage.checkContext(ctx)
|
||||
// validate that the namespace context on the request matches the expected input
|
||||
storage.requestedResourceNamespace = request.NamespaceValue(ctx)
|
||||
if storage.expectedResourceNamespace != storage.requestedResourceNamespace {
|
||||
return nil, nil, fmt.Errorf("Expected request namespace %s, but got namespace %s", storage.expectedResourceNamespace, storage.requestedResourceNamespace)
|
||||
}
|
||||
storage.requestedResourceLocationID = id
|
||||
if err := storage.errors["resourceLocation"]; err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
// Make a copy so the internal URL never gets mutated
|
||||
locationCopy := *storage.resourceLocation
|
||||
return &locationCopy, storage.resourceLocationTransport, nil
|
||||
}
|
||||
|
||||
// Implement Connecter
|
||||
type ConnecterRESTStorage struct {
|
||||
connectHandler http.Handler
|
||||
|
|
@ -1526,18 +1506,30 @@ func TestMetadata(t *testing.T) {
|
|||
matches := map[string]int{}
|
||||
for _, w := range ws {
|
||||
for _, r := range w.Routes() {
|
||||
t.Logf("%v %v %#v", r.Method, r.Path, r.Produces)
|
||||
s := strings.Join(r.Produces, ",")
|
||||
i := matches[s]
|
||||
matches[s] = i + 1
|
||||
}
|
||||
}
|
||||
|
||||
if matches["text/plain,application/json,application/yaml,application/vnd.kubernetes.protobuf"] == 0 ||
|
||||
matches["application/json,application/yaml,application/vnd.kubernetes.protobuf,application/json;stream=watch,application/vnd.kubernetes.protobuf;stream=watch"] == 0 ||
|
||||
matches["application/json,application/yaml,application/vnd.kubernetes.protobuf"] == 0 ||
|
||||
matches["*/*"] == 0 ||
|
||||
len(matches) != 5 {
|
||||
t.Errorf("unexpected mime types: %v", matches)
|
||||
cs := []func() bool{
|
||||
func() bool {
|
||||
return matches["text/plain,application/json,application/yaml,application/vnd.kubernetes.protobuf"] == 0
|
||||
},
|
||||
func() bool {
|
||||
return matches["application/json,application/yaml,application/vnd.kubernetes.protobuf,application/json;stream=watch,application/vnd.kubernetes.protobuf;stream=watch"] == 0
|
||||
},
|
||||
func() bool {
|
||||
return matches["application/json,application/yaml,application/vnd.kubernetes.protobuf"] == 0
|
||||
},
|
||||
func() bool {
|
||||
return len(matches) != 4
|
||||
},
|
||||
}
|
||||
for i, c := range cs {
|
||||
if c() {
|
||||
t.Errorf("[%d]unexpected mime types: %#v", i, matches)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -44,7 +44,6 @@ go_library(
|
|||
"get.go",
|
||||
"namer.go",
|
||||
"patch.go",
|
||||
"proxy.go",
|
||||
"response.go",
|
||||
"rest.go",
|
||||
"update.go",
|
||||
|
|
@ -65,11 +64,8 @@ go_library(
|
|||
"//vendor/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/runtime/serializer/streaming:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/util/httpstream:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/util/json:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/util/mergepatch:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/util/net:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/util/proxy:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/util/runtime:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library",
|
||||
"//vendor/k8s.io/apimachinery/pkg/util/strategicpatch:go_default_library",
|
||||
|
|
|
|||
|
|
@ -1,287 +0,0 @@
|
|||
/*
|
||||
Copyright 2014 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 handlers
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"math/rand"
|
||||
"net/http"
|
||||
"net/http/httputil"
|
||||
"net/url"
|
||||
"path"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
apierrors "k8s.io/apimachinery/pkg/api/errors"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/apimachinery/pkg/util/httpstream"
|
||||
"k8s.io/apimachinery/pkg/util/net"
|
||||
proxyutil "k8s.io/apimachinery/pkg/util/proxy"
|
||||
"k8s.io/apiserver/pkg/endpoints/handlers/responsewriters"
|
||||
"k8s.io/apiserver/pkg/endpoints/metrics"
|
||||
"k8s.io/apiserver/pkg/endpoints/request"
|
||||
"k8s.io/apiserver/pkg/registry/rest"
|
||||
"k8s.io/apiserver/pkg/server/httplog"
|
||||
|
||||
"github.com/golang/glog"
|
||||
)
|
||||
|
||||
// ProxyHandler provides a http.Handler which will proxy traffic to locations
|
||||
// specified by items implementing Redirector.
|
||||
type ProxyHandler struct {
|
||||
Prefix string
|
||||
Storage map[string]rest.Storage
|
||||
Serializer runtime.NegotiatedSerializer
|
||||
Mapper request.RequestContextMapper
|
||||
}
|
||||
|
||||
func (r *ProxyHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
|
||||
reqStart := time.Now()
|
||||
|
||||
var httpCode int
|
||||
var requestInfo *request.RequestInfo
|
||||
defer RecordMetrics(w, req, requestInfo, httpCode, reqStart)
|
||||
|
||||
ctx, ok := r.Mapper.Get(req)
|
||||
if !ok {
|
||||
responsewriters.InternalError(w, req, errors.New("Error getting request context"))
|
||||
httpCode = http.StatusInternalServerError
|
||||
return
|
||||
}
|
||||
|
||||
requestInfo, ok = request.RequestInfoFrom(ctx)
|
||||
if !ok {
|
||||
responsewriters.InternalError(w, req, errors.New("Error getting RequestInfo from context"))
|
||||
httpCode = http.StatusInternalServerError
|
||||
return
|
||||
}
|
||||
|
||||
metrics.RecordLongRunning(req, requestInfo, func() {
|
||||
httpCode = r.serveHTTP(w, req, ctx, requestInfo)
|
||||
})
|
||||
}
|
||||
|
||||
// serveHTTP performs proxy handling and returns the status code of the operation.
|
||||
func (r *ProxyHandler) serveHTTP(w http.ResponseWriter, req *http.Request, ctx request.Context, requestInfo *request.RequestInfo) int {
|
||||
proxyHandlerTraceID := rand.Int63()
|
||||
|
||||
if !requestInfo.IsResourceRequest {
|
||||
responsewriters.NotFound(w, req)
|
||||
return http.StatusNotFound
|
||||
}
|
||||
namespace, resource, parts := requestInfo.Namespace, requestInfo.Resource, requestInfo.Parts
|
||||
|
||||
ctx = request.WithNamespace(ctx, namespace)
|
||||
if len(parts) < 2 {
|
||||
responsewriters.NotFound(w, req)
|
||||
return http.StatusNotFound
|
||||
}
|
||||
id := parts[1]
|
||||
remainder := ""
|
||||
if len(parts) > 2 {
|
||||
proxyParts := parts[2:]
|
||||
remainder = strings.Join(proxyParts, "/")
|
||||
if strings.HasSuffix(req.URL.Path, "/") {
|
||||
// The original path had a trailing slash, which has been stripped
|
||||
// by KindAndNamespace(). We should add it back because some
|
||||
// servers (like etcd) require it.
|
||||
remainder = remainder + "/"
|
||||
}
|
||||
}
|
||||
storage, ok := r.Storage[resource]
|
||||
if !ok {
|
||||
httplog.LogOf(req, w).Addf("'%v' has no storage object", resource)
|
||||
responsewriters.NotFound(w, req)
|
||||
return http.StatusNotFound
|
||||
}
|
||||
|
||||
gv := schema.GroupVersion{Group: requestInfo.APIGroup, Version: requestInfo.APIVersion}
|
||||
|
||||
redirector, ok := storage.(rest.Redirector)
|
||||
if !ok {
|
||||
httplog.LogOf(req, w).Addf("'%v' is not a redirector", resource)
|
||||
return responsewriters.ErrorNegotiated(ctx, apierrors.NewMethodNotSupported(schema.GroupResource{Resource: resource}, "proxy"), r.Serializer, gv, w, req)
|
||||
}
|
||||
|
||||
location, roundTripper, err := redirector.ResourceLocation(ctx, id)
|
||||
if err != nil {
|
||||
httplog.LogOf(req, w).Addf("Error getting ResourceLocation: %v", err)
|
||||
return responsewriters.ErrorNegotiated(ctx, err, r.Serializer, gv, w, req)
|
||||
}
|
||||
if location == nil {
|
||||
httplog.LogOf(req, w).Addf("ResourceLocation for %v returned nil", id)
|
||||
responsewriters.NotFound(w, req)
|
||||
return http.StatusNotFound
|
||||
}
|
||||
|
||||
if roundTripper != nil {
|
||||
glog.V(5).Infof("[%x: %v] using transport %T...", proxyHandlerTraceID, req.URL, roundTripper)
|
||||
}
|
||||
|
||||
// Default to http
|
||||
if location.Scheme == "" {
|
||||
location.Scheme = "http"
|
||||
}
|
||||
// Add the subpath
|
||||
if len(remainder) > 0 {
|
||||
location.Path = singleJoiningSlash(location.Path, remainder)
|
||||
}
|
||||
// Start with anything returned from the storage, and add the original request's parameters
|
||||
values := location.Query()
|
||||
for k, vs := range req.URL.Query() {
|
||||
for _, v := range vs {
|
||||
values.Add(k, v)
|
||||
}
|
||||
}
|
||||
location.RawQuery = values.Encode()
|
||||
|
||||
// WithContext creates a shallow clone of the request with the new context.
|
||||
newReq := req.WithContext(context.Background())
|
||||
newReq.Header = net.CloneHeader(req.Header)
|
||||
newReq.URL = location
|
||||
|
||||
// TODO convert this entire proxy to an UpgradeAwareProxy similar to
|
||||
// https://github.com/openshift/origin/blob/master/pkg/util/httpproxy/upgradeawareproxy.go.
|
||||
// That proxy needs to be modified to support multiple backends, not just 1.
|
||||
if r.tryUpgrade(ctx, w, req, newReq, location, roundTripper, gv) {
|
||||
return http.StatusSwitchingProtocols
|
||||
}
|
||||
|
||||
// Redirect requests of the form "/{resource}/{name}" to "/{resource}/{name}/"
|
||||
// This is essentially a hack for http://issue.k8s.io/4958.
|
||||
// Note: Keep this code after tryUpgrade to not break that flow.
|
||||
if len(parts) == 2 && !strings.HasSuffix(req.URL.Path, "/") {
|
||||
var queryPart string
|
||||
if len(req.URL.RawQuery) > 0 {
|
||||
queryPart = "?" + req.URL.RawQuery
|
||||
}
|
||||
w.Header().Set("Location", req.URL.Path+"/"+queryPart)
|
||||
w.WriteHeader(http.StatusMovedPermanently)
|
||||
return http.StatusMovedPermanently
|
||||
}
|
||||
|
||||
start := time.Now()
|
||||
glog.V(4).Infof("[%x] Beginning proxy %s...", proxyHandlerTraceID, req.URL)
|
||||
defer func() {
|
||||
glog.V(4).Infof("[%x] Proxy %v finished %v.", proxyHandlerTraceID, req.URL, time.Now().Sub(start))
|
||||
}()
|
||||
|
||||
proxy := httputil.NewSingleHostReverseProxy(&url.URL{Scheme: location.Scheme, Host: location.Host})
|
||||
alreadyRewriting := false
|
||||
if roundTripper != nil {
|
||||
_, alreadyRewriting = roundTripper.(*proxyutil.Transport)
|
||||
glog.V(5).Infof("[%x] Not making a rewriting transport for proxy %s...", proxyHandlerTraceID, req.URL)
|
||||
}
|
||||
if !alreadyRewriting {
|
||||
glog.V(5).Infof("[%x] making a transport for proxy %s...", proxyHandlerTraceID, req.URL)
|
||||
prepend := path.Join(r.Prefix, resource, id)
|
||||
if len(namespace) > 0 {
|
||||
prepend = path.Join(r.Prefix, "namespaces", namespace, resource, id)
|
||||
}
|
||||
pTransport := &proxyutil.Transport{
|
||||
Scheme: req.URL.Scheme,
|
||||
Host: req.URL.Host,
|
||||
PathPrepend: prepend,
|
||||
RoundTripper: roundTripper,
|
||||
}
|
||||
roundTripper = pTransport
|
||||
}
|
||||
proxy.Transport = roundTripper
|
||||
proxy.FlushInterval = 200 * time.Millisecond
|
||||
proxy.ServeHTTP(w, newReq)
|
||||
return 0
|
||||
}
|
||||
|
||||
// tryUpgrade returns true if the request was handled.
|
||||
func (r *ProxyHandler) tryUpgrade(ctx request.Context, w http.ResponseWriter, req, newReq *http.Request, location *url.URL, transport http.RoundTripper, gv schema.GroupVersion) bool {
|
||||
if !httpstream.IsUpgradeRequest(req) {
|
||||
return false
|
||||
}
|
||||
// Only append X-Forwarded-For in the upgrade path, since httputil.NewSingleHostReverseProxy
|
||||
// handles this in the non-upgrade path.
|
||||
net.AppendForwardedForHeader(newReq)
|
||||
|
||||
backendConn, err := proxyutil.DialURL(location, transport)
|
||||
if err != nil {
|
||||
responsewriters.ErrorNegotiated(ctx, err, r.Serializer, gv, w, req)
|
||||
return true
|
||||
}
|
||||
defer backendConn.Close()
|
||||
|
||||
// TODO should we use _ (a bufio.ReadWriter) instead of requestHijackedConn
|
||||
// when copying between the client and the backend? Docker doesn't when they
|
||||
// hijack, just for reference...
|
||||
requestHijackedConn, _, err := w.(http.Hijacker).Hijack()
|
||||
if err != nil {
|
||||
responsewriters.ErrorNegotiated(ctx, err, r.Serializer, gv, w, req)
|
||||
return true
|
||||
}
|
||||
defer requestHijackedConn.Close()
|
||||
|
||||
if err = newReq.Write(backendConn); err != nil {
|
||||
responsewriters.ErrorNegotiated(ctx, err, r.Serializer, gv, w, req)
|
||||
return true
|
||||
}
|
||||
|
||||
done := make(chan struct{}, 2)
|
||||
|
||||
go func() {
|
||||
_, err := io.Copy(backendConn, requestHijackedConn)
|
||||
if err != nil && !strings.Contains(err.Error(), "use of closed network connection") {
|
||||
glog.Errorf("Error proxying data from client to backend: %v", err)
|
||||
}
|
||||
done <- struct{}{}
|
||||
}()
|
||||
|
||||
go func() {
|
||||
_, err := io.Copy(requestHijackedConn, backendConn)
|
||||
if err != nil && !strings.Contains(err.Error(), "use of closed network connection") {
|
||||
glog.Errorf("Error proxying data from backend to client: %v", err)
|
||||
}
|
||||
done <- struct{}{}
|
||||
}()
|
||||
|
||||
<-done
|
||||
return true
|
||||
}
|
||||
|
||||
// borrowed from net/http/httputil/reverseproxy.go
|
||||
func singleJoiningSlash(a, b string) string {
|
||||
aslash := strings.HasSuffix(a, "/")
|
||||
bslash := strings.HasPrefix(b, "/")
|
||||
switch {
|
||||
case aslash && bslash:
|
||||
return a + b[1:]
|
||||
case !aslash && !bslash:
|
||||
return a + "/" + b
|
||||
}
|
||||
return a + b
|
||||
}
|
||||
|
||||
func RecordMetrics(w http.ResponseWriter, req *http.Request, requestInfo *request.RequestInfo, httpCode int, reqStart time.Time) {
|
||||
responseLength := 0
|
||||
if rw, ok := w.(*metrics.ResponseWriterDelegator); ok {
|
||||
responseLength = rw.ContentLength()
|
||||
if httpCode == 0 {
|
||||
httpCode = rw.Status()
|
||||
}
|
||||
}
|
||||
metrics.Record(req, requestInfo, w.Header().Get("Content-Type"), httpCode, responseLength, time.Now().Sub(reqStart))
|
||||
}
|
||||
|
|
@ -96,13 +96,6 @@ func (a *APIInstaller) Install() ([]metav1.APIResource, *restful.WebService, []e
|
|||
var errors []error
|
||||
ws := a.newWebService()
|
||||
|
||||
proxyHandler := (&handlers.ProxyHandler{
|
||||
Prefix: a.prefix + "/proxy/",
|
||||
Storage: a.group.Storage,
|
||||
Serializer: a.group.Serializer,
|
||||
Mapper: a.group.Context,
|
||||
})
|
||||
|
||||
// Register the paths in a deterministic (sorted) order to get a deterministic swagger spec.
|
||||
paths := make([]string, len(a.group.Storage))
|
||||
var i int = 0
|
||||
|
|
@ -112,7 +105,7 @@ func (a *APIInstaller) Install() ([]metav1.APIResource, *restful.WebService, []e
|
|||
}
|
||||
sort.Strings(paths)
|
||||
for _, path := range paths {
|
||||
apiResource, err := a.registerResourceHandlers(path, a.group.Storage[path], ws, proxyHandler)
|
||||
apiResource, err := a.registerResourceHandlers(path, a.group.Storage[path], ws)
|
||||
if err != nil {
|
||||
errors = append(errors, fmt.Errorf("error in registering resource: %s, %v", path, err))
|
||||
}
|
||||
|
|
@ -193,7 +186,7 @@ func (a *APIInstaller) restMapping(resource string) (*meta.RESTMapping, error) {
|
|||
return a.group.Mapper.RESTMapping(fqKindToRegister.GroupKind(), fqKindToRegister.Version)
|
||||
}
|
||||
|
||||
func (a *APIInstaller) registerResourceHandlers(path string, storage rest.Storage, ws *restful.WebService, proxyHandler http.Handler) (*metav1.APIResource, error) {
|
||||
func (a *APIInstaller) registerResourceHandlers(path string, storage rest.Storage, ws *restful.WebService) (*metav1.APIResource, error) {
|
||||
admit := a.group.Admit
|
||||
context := a.group.Context
|
||||
if context == nil {
|
||||
|
|
@ -240,7 +233,6 @@ func (a *APIInstaller) registerResourceHandlers(path string, storage rest.Storag
|
|||
updater, isUpdater := storage.(rest.Updater)
|
||||
patcher, isPatcher := storage.(rest.Patcher)
|
||||
watcher, isWatcher := storage.(rest.Watcher)
|
||||
_, isRedirector := storage.(rest.Redirector)
|
||||
connecter, isConnecter := storage.(rest.Connecter)
|
||||
storageMeta, isMetadata := storage.(rest.StorageMetadata)
|
||||
if !isMetadata {
|
||||
|
|
@ -426,12 +418,6 @@ func (a *APIInstaller) registerResourceHandlers(path string, storage rest.Storag
|
|||
actions = appendIf(actions, action{"PATCH", itemPath, nameParams, namer, false}, isPatcher)
|
||||
actions = appendIf(actions, action{"DELETE", itemPath, nameParams, namer, false}, isDeleter)
|
||||
actions = appendIf(actions, action{"WATCH", "watch/" + itemPath, nameParams, namer, false}, isWatcher)
|
||||
// We add "proxy" subresource to remove the need for the generic top level prefix proxy.
|
||||
// The generic top level prefix proxy is deprecated in v1.2, and will be removed in 1.3, or 1.4 at the latest.
|
||||
// TODO: DEPRECATED in v1.2.
|
||||
actions = appendIf(actions, action{"PROXY", "proxy/" + itemPath + "/{path:*}", proxyParams, namer, false}, isRedirector)
|
||||
// TODO: DEPRECATED in v1.2.
|
||||
actions = appendIf(actions, action{"PROXY", "proxy/" + itemPath, nameParams, namer, false}, isRedirector)
|
||||
actions = appendIf(actions, action{"CONNECT", itemPath, nameParams, namer, false}, isConnecter)
|
||||
actions = appendIf(actions, action{"CONNECT", itemPath + "/{path:*}", proxyParams, namer, false}, isConnecter && connectSubpath)
|
||||
break
|
||||
|
|
@ -478,12 +464,6 @@ func (a *APIInstaller) registerResourceHandlers(path string, storage rest.Storag
|
|||
actions = appendIf(actions, action{"PATCH", itemPath, nameParams, namer, false}, isPatcher)
|
||||
actions = appendIf(actions, action{"DELETE", itemPath, nameParams, namer, false}, isDeleter)
|
||||
actions = appendIf(actions, action{"WATCH", "watch/" + itemPath, nameParams, namer, false}, isWatcher)
|
||||
// We add "proxy" subresource to remove the need for the generic top level prefix proxy.
|
||||
// The generic top level prefix proxy is deprecated in v1.2, and will be removed in 1.3, or 1.4 at the latest.
|
||||
// TODO: DEPRECATED in v1.2.
|
||||
actions = appendIf(actions, action{"PROXY", "proxy/" + itemPath + "/{path:*}", proxyParams, namer, false}, isRedirector)
|
||||
// TODO: DEPRECATED in v1.2.
|
||||
actions = appendIf(actions, action{"PROXY", "proxy/" + itemPath, nameParams, namer, false}, isRedirector)
|
||||
actions = appendIf(actions, action{"CONNECT", itemPath, nameParams, namer, false}, isConnecter)
|
||||
actions = appendIf(actions, action{"CONNECT", itemPath + "/{path:*}", proxyParams, namer, false}, isConnecter && connectSubpath)
|
||||
|
||||
|
|
@ -812,18 +792,6 @@ func (a *APIInstaller) registerResourceHandlers(path string, storage rest.Storag
|
|||
}
|
||||
addParams(route, action.Params)
|
||||
routes = append(routes, route)
|
||||
// We add "proxy" subresource to remove the need for the generic top level prefix proxy.
|
||||
// The generic top level prefix proxy is deprecated in v1.2, and will be removed in 1.3, or 1.4 at the latest.
|
||||
// TODO: DEPRECATED in v1.2.
|
||||
case "PROXY": // Proxy requests to a resource.
|
||||
// Accept all methods as per http://issue.k8s.io/3996
|
||||
routes = append(routes, buildProxyRoute(ws, "GET", a.prefix, action.Path, kind, resource, subresource, namespaced, requestScope, hasSubresource, action.Params, proxyHandler, operationSuffix))
|
||||
routes = append(routes, buildProxyRoute(ws, "PUT", a.prefix, action.Path, kind, resource, subresource, namespaced, requestScope, hasSubresource, action.Params, proxyHandler, operationSuffix))
|
||||
routes = append(routes, buildProxyRoute(ws, "POST", a.prefix, action.Path, kind, resource, subresource, namespaced, requestScope, hasSubresource, action.Params, proxyHandler, operationSuffix))
|
||||
routes = append(routes, buildProxyRoute(ws, "PATCH", a.prefix, action.Path, kind, resource, subresource, namespaced, requestScope, hasSubresource, action.Params, proxyHandler, operationSuffix))
|
||||
routes = append(routes, buildProxyRoute(ws, "DELETE", a.prefix, action.Path, kind, resource, subresource, namespaced, requestScope, hasSubresource, action.Params, proxyHandler, operationSuffix))
|
||||
routes = append(routes, buildProxyRoute(ws, "HEAD", a.prefix, action.Path, kind, resource, subresource, namespaced, requestScope, hasSubresource, action.Params, proxyHandler, operationSuffix))
|
||||
routes = append(routes, buildProxyRoute(ws, "OPTIONS", a.prefix, action.Path, kind, resource, subresource, namespaced, requestScope, hasSubresource, action.Params, proxyHandler, operationSuffix))
|
||||
case "CONNECT":
|
||||
for _, method := range connecter.ConnectMethods() {
|
||||
connectProducedObject := storageMeta.ProducesObject(method)
|
||||
|
|
@ -906,27 +874,6 @@ func routeFunction(handler http.Handler) restful.RouteFunction {
|
|||
}
|
||||
}
|
||||
|
||||
func buildProxyRoute(ws *restful.WebService,
|
||||
method, prefix, path, kind, resource, subresource, namespaced, requestScope string,
|
||||
hasSubresource bool,
|
||||
params []*restful.Parameter,
|
||||
proxyHandler http.Handler,
|
||||
operationSuffix string) *restful.RouteBuilder {
|
||||
doc := "proxy " + method + " requests to " + kind
|
||||
if hasSubresource {
|
||||
doc = "proxy " + method + " requests to " + subresource + " of " + kind
|
||||
}
|
||||
handler := metrics.InstrumentRouteFunc("PROXY", resource, subresource, requestScope, routeFunction(proxyHandler))
|
||||
proxyRoute := ws.Method(method).Path(path).To(handler).
|
||||
Doc(doc).
|
||||
Operation("proxy" + strings.Title(method) + namespaced + kind + strings.Title(subresource) + operationSuffix).
|
||||
Produces("*/*").
|
||||
Consumes("*/*").
|
||||
Writes("string")
|
||||
addParams(proxyRoute, params)
|
||||
return proxyRoute
|
||||
}
|
||||
|
||||
func addParams(route *restful.RouteBuilder, params []*restful.Parameter) {
|
||||
for _, param := range params {
|
||||
route.Param(param)
|
||||
|
|
|
|||
|
|
@ -1,570 +0,0 @@
|
|||
/*
|
||||
Copyright 2014 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 endpoints
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"compress/gzip"
|
||||
"crypto/tls"
|
||||
"crypto/x509"
|
||||
"fmt"
|
||||
"io"
|
||||
"io/ioutil"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"net/http/httputil"
|
||||
"net/url"
|
||||
"reflect"
|
||||
"strconv"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"golang.org/x/net/websocket"
|
||||
utilnet "k8s.io/apimachinery/pkg/util/net"
|
||||
"k8s.io/apiserver/pkg/registry/rest"
|
||||
)
|
||||
|
||||
func TestProxyRequestContentLengthAndTransferEncoding(t *testing.T) {
|
||||
chunk := func(data []byte) []byte {
|
||||
out := &bytes.Buffer{}
|
||||
chunker := httputil.NewChunkedWriter(out)
|
||||
for _, b := range data {
|
||||
if _, err := chunker.Write([]byte{b}); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
}
|
||||
chunker.Close()
|
||||
out.Write([]byte("\r\n"))
|
||||
return out.Bytes()
|
||||
}
|
||||
|
||||
zip := func(data []byte) []byte {
|
||||
out := &bytes.Buffer{}
|
||||
zipper := gzip.NewWriter(out)
|
||||
if _, err := zipper.Write(data); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
zipper.Close()
|
||||
return out.Bytes()
|
||||
}
|
||||
|
||||
sampleData := []byte("abcde")
|
||||
|
||||
table := map[string]struct {
|
||||
reqHeaders http.Header
|
||||
reqBody []byte
|
||||
|
||||
expectedHeaders http.Header
|
||||
expectedBody []byte
|
||||
}{
|
||||
"content-length": {
|
||||
reqHeaders: http.Header{
|
||||
"Content-Length": []string{"5"},
|
||||
},
|
||||
reqBody: sampleData,
|
||||
|
||||
expectedHeaders: http.Header{
|
||||
"Content-Length": []string{"5"},
|
||||
"Content-Encoding": nil, // none set
|
||||
"Transfer-Encoding": nil, // none set
|
||||
},
|
||||
expectedBody: sampleData,
|
||||
},
|
||||
|
||||
"content-length + identity transfer-encoding": {
|
||||
reqHeaders: http.Header{
|
||||
"Content-Length": []string{"5"},
|
||||
"Transfer-Encoding": []string{"identity"},
|
||||
},
|
||||
reqBody: sampleData,
|
||||
|
||||
expectedHeaders: http.Header{
|
||||
"Content-Length": []string{"5"},
|
||||
"Content-Encoding": nil, // none set
|
||||
"Transfer-Encoding": nil, // gets removed
|
||||
},
|
||||
expectedBody: sampleData,
|
||||
},
|
||||
|
||||
"content-length + gzip content-encoding": {
|
||||
reqHeaders: http.Header{
|
||||
"Content-Length": []string{strconv.Itoa(len(zip(sampleData)))},
|
||||
"Content-Encoding": []string{"gzip"},
|
||||
},
|
||||
reqBody: zip(sampleData),
|
||||
|
||||
expectedHeaders: http.Header{
|
||||
"Content-Length": []string{strconv.Itoa(len(zip(sampleData)))},
|
||||
"Content-Encoding": []string{"gzip"},
|
||||
"Transfer-Encoding": nil, // none set
|
||||
},
|
||||
expectedBody: zip(sampleData),
|
||||
},
|
||||
|
||||
"chunked transfer-encoding": {
|
||||
reqHeaders: http.Header{
|
||||
"Transfer-Encoding": []string{"chunked"},
|
||||
},
|
||||
reqBody: chunk(sampleData),
|
||||
|
||||
expectedHeaders: http.Header{
|
||||
"Content-Length": nil, // none set
|
||||
"Content-Encoding": nil, // none set
|
||||
"Transfer-Encoding": nil, // Transfer-Encoding gets removed
|
||||
},
|
||||
expectedBody: sampleData, // sample data is unchunked
|
||||
},
|
||||
|
||||
"chunked transfer-encoding + gzip content-encoding": {
|
||||
reqHeaders: http.Header{
|
||||
"Content-Encoding": []string{"gzip"},
|
||||
"Transfer-Encoding": []string{"chunked"},
|
||||
},
|
||||
reqBody: chunk(zip(sampleData)),
|
||||
|
||||
expectedHeaders: http.Header{
|
||||
"Content-Length": nil, // none set
|
||||
"Content-Encoding": []string{"gzip"},
|
||||
"Transfer-Encoding": nil, // gets removed
|
||||
},
|
||||
expectedBody: zip(sampleData), // sample data is unchunked, but content-encoding is preserved
|
||||
},
|
||||
|
||||
// "Transfer-Encoding: gzip" is not supported by go
|
||||
// See http/transfer.go#fixTransferEncoding (https://golang.org/src/net/http/transfer.go#L427)
|
||||
// Once it is supported, this test case should succeed
|
||||
//
|
||||
// "gzip+chunked transfer-encoding": {
|
||||
// reqHeaders: http.Header{
|
||||
// "Transfer-Encoding": []string{"chunked,gzip"},
|
||||
// },
|
||||
// reqBody: chunk(zip(sampleData)),
|
||||
//
|
||||
// expectedHeaders: http.Header{
|
||||
// "Content-Length": nil, // no content-length headers
|
||||
// "Transfer-Encoding": nil, // Transfer-Encoding gets removed
|
||||
// },
|
||||
// expectedBody: sampleData,
|
||||
// },
|
||||
}
|
||||
|
||||
successfulResponse := "backend passed tests"
|
||||
for k, item := range table {
|
||||
// Start the downstream server
|
||||
downstreamServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
|
||||
// Verify headers
|
||||
for header, v := range item.expectedHeaders {
|
||||
if !reflect.DeepEqual(v, req.Header[header]) {
|
||||
t.Errorf("%s: Expected headers for %s to be %v, got %v", k, header, v, req.Header[header])
|
||||
}
|
||||
}
|
||||
|
||||
// Read body
|
||||
body, err := ioutil.ReadAll(req.Body)
|
||||
if err != nil {
|
||||
t.Errorf("%s: unexpected error %v", k, err)
|
||||
}
|
||||
req.Body.Close()
|
||||
|
||||
// Verify length
|
||||
if req.ContentLength > 0 && req.ContentLength != int64(len(body)) {
|
||||
t.Errorf("%s: ContentLength was %d, len(data) was %d", k, req.ContentLength, len(body))
|
||||
}
|
||||
|
||||
// Verify content
|
||||
if !bytes.Equal(item.expectedBody, body) {
|
||||
t.Errorf("%s: Expected %q, got %q", k, string(item.expectedBody), string(body))
|
||||
}
|
||||
|
||||
// Write successful response
|
||||
w.Write([]byte(successfulResponse))
|
||||
}))
|
||||
defer downstreamServer.Close()
|
||||
|
||||
// Start the proxy server
|
||||
serverURL, _ := url.Parse(downstreamServer.URL)
|
||||
simpleStorage := &SimpleRESTStorage{
|
||||
errors: map[string]error{},
|
||||
resourceLocation: serverURL,
|
||||
expectedResourceNamespace: "default",
|
||||
}
|
||||
namespaceHandler := handleNamespaced(map[string]rest.Storage{"foo": simpleStorage})
|
||||
server := newTestServer(namespaceHandler)
|
||||
defer server.Close()
|
||||
|
||||
// Dial the proxy server
|
||||
conn, err := net.Dial(server.Listener.Addr().Network(), server.Listener.Addr().String())
|
||||
if err != nil {
|
||||
t.Errorf("%s: unexpected error %v", k, err)
|
||||
continue
|
||||
}
|
||||
defer conn.Close()
|
||||
|
||||
// Add standard http 1.1 headers
|
||||
if item.reqHeaders == nil {
|
||||
item.reqHeaders = http.Header{}
|
||||
}
|
||||
item.reqHeaders.Add("Connection", "close")
|
||||
item.reqHeaders.Add("Host", server.Listener.Addr().String())
|
||||
|
||||
// We directly write to the connection to bypass the Go library's manipulation of the Request.Header.
|
||||
// Write the request headers
|
||||
post := fmt.Sprintf("POST /%s/%s/%s/proxy/namespaces/default/foo/id/some/dir HTTP/1.1\r\n", prefix, newGroupVersion.Group, newGroupVersion.Version)
|
||||
if _, err := fmt.Fprint(conn, post); err != nil {
|
||||
t.Fatalf("%s: unexpected error %v", k, err)
|
||||
}
|
||||
for header, values := range item.reqHeaders {
|
||||
for _, value := range values {
|
||||
if _, err := fmt.Fprintf(conn, "%s: %s\r\n", header, value); err != nil {
|
||||
t.Fatalf("%s: unexpected error %v", k, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
// Header separator
|
||||
if _, err := fmt.Fprint(conn, "\r\n"); err != nil {
|
||||
t.Fatalf("%s: unexpected error %v", k, err)
|
||||
}
|
||||
// Body
|
||||
if _, err := conn.Write(item.reqBody); err != nil {
|
||||
t.Fatalf("%s: unexpected error %v", k, err)
|
||||
}
|
||||
|
||||
// Read response
|
||||
response, err := ioutil.ReadAll(conn)
|
||||
if err != nil {
|
||||
t.Errorf("%s: unexpected error %v", k, err)
|
||||
continue
|
||||
}
|
||||
if !strings.HasSuffix(string(response), successfulResponse) {
|
||||
t.Errorf("%s: Did not get successful response: %s", k, string(response))
|
||||
continue
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestProxy(t *testing.T) {
|
||||
table := []struct {
|
||||
method string
|
||||
path string
|
||||
reqBody string
|
||||
respBody string
|
||||
respContentType string
|
||||
reqNamespace string
|
||||
}{
|
||||
{"GET", "/some/dir", "", "answer", "text/css", "default"},
|
||||
{"GET", "/some/dir", "", "<html><head></head><body>answer</body></html>", "text/html", "default"},
|
||||
{"POST", "/some/other/dir", "question", "answer", "text/css", "default"},
|
||||
{"PUT", "/some/dir/id", "different question", "answer", "text/css", "default"},
|
||||
{"DELETE", "/some/dir/id", "", "ok", "text/css", "default"},
|
||||
{"GET", "/some/dir/id", "", "answer", "text/css", "other"},
|
||||
{"GET", "/trailing/slash/", "", "answer", "text/css", "default"},
|
||||
{"GET", "/", "", "answer", "text/css", "default"},
|
||||
}
|
||||
|
||||
for _, item := range table {
|
||||
downstreamServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
|
||||
gotBody, err := ioutil.ReadAll(req.Body)
|
||||
if err != nil {
|
||||
t.Errorf("%v - unexpected error %v", item.method, err)
|
||||
}
|
||||
if e, a := item.reqBody, string(gotBody); e != a {
|
||||
t.Errorf("%v - expected %v, got %v", item.method, e, a)
|
||||
}
|
||||
if e, a := item.path, req.URL.Path; e != a {
|
||||
t.Errorf("%v - expected %v, got %v", item.method, e, a)
|
||||
}
|
||||
w.Header().Set("Content-Type", item.respContentType)
|
||||
var out io.Writer = w
|
||||
if strings.Contains(req.Header.Get("Accept-Encoding"), "gzip") {
|
||||
// The proxier can ask for gzip'd data; we need to provide it with that
|
||||
// in order to test our processing of that data.
|
||||
w.Header().Set("Content-Encoding", "gzip")
|
||||
gzw := gzip.NewWriter(w)
|
||||
out = gzw
|
||||
defer gzw.Close()
|
||||
}
|
||||
fmt.Fprint(out, item.respBody)
|
||||
}))
|
||||
defer downstreamServer.Close()
|
||||
|
||||
serverURL, _ := url.Parse(downstreamServer.URL)
|
||||
simpleStorage := &SimpleRESTStorage{
|
||||
errors: map[string]error{},
|
||||
resourceLocation: serverURL,
|
||||
expectedResourceNamespace: item.reqNamespace,
|
||||
}
|
||||
|
||||
namespaceHandler := handleNamespaced(map[string]rest.Storage{"foo": simpleStorage})
|
||||
namespaceServer := newTestServer(namespaceHandler)
|
||||
defer namespaceServer.Close()
|
||||
|
||||
// test each supported URL pattern for finding the redirection resource in the proxy in a particular namespace
|
||||
serverPatterns := []struct {
|
||||
server *httptest.Server
|
||||
proxyTestPattern string
|
||||
}{
|
||||
{namespaceServer, "/" + prefix + "/" + newGroupVersion.Group + "/" + newGroupVersion.Version + "/proxy/namespaces/" + item.reqNamespace + "/foo/id" + item.path},
|
||||
}
|
||||
|
||||
for _, serverPattern := range serverPatterns {
|
||||
server := serverPattern.server
|
||||
proxyTestPattern := serverPattern.proxyTestPattern
|
||||
req, err := http.NewRequest(
|
||||
item.method,
|
||||
server.URL+proxyTestPattern,
|
||||
strings.NewReader(item.reqBody),
|
||||
)
|
||||
if err != nil {
|
||||
t.Errorf("%v - unexpected error %v", item.method, err)
|
||||
continue
|
||||
}
|
||||
resp, err := http.DefaultClient.Do(req)
|
||||
if err != nil {
|
||||
t.Errorf("%v - unexpected error %v", item.method, err)
|
||||
continue
|
||||
}
|
||||
gotResp, err := ioutil.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
t.Errorf("%v - unexpected error %v", item.method, err)
|
||||
}
|
||||
resp.Body.Close()
|
||||
if e, a := item.respBody, string(gotResp); e != a {
|
||||
t.Errorf("%v - expected %v, got %v. url: %#v", item.method, e, a, req.URL)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestProxyUpgrade(t *testing.T) {
|
||||
|
||||
localhostPool := x509.NewCertPool()
|
||||
if !localhostPool.AppendCertsFromPEM(localhostCert) {
|
||||
t.Errorf("error setting up localhostCert pool")
|
||||
}
|
||||
|
||||
testcases := map[string]struct {
|
||||
ServerFunc func(http.Handler) *httptest.Server
|
||||
ProxyTransport http.RoundTripper
|
||||
}{
|
||||
"http": {
|
||||
ServerFunc: httptest.NewServer,
|
||||
ProxyTransport: nil,
|
||||
},
|
||||
"https (invalid hostname + InsecureSkipVerify)": {
|
||||
ServerFunc: func(h http.Handler) *httptest.Server {
|
||||
cert, err := tls.X509KeyPair(exampleCert, exampleKey)
|
||||
if err != nil {
|
||||
t.Errorf("https (invalid hostname): proxy_test: %v", err)
|
||||
}
|
||||
ts := httptest.NewUnstartedServer(h)
|
||||
ts.TLS = &tls.Config{
|
||||
Certificates: []tls.Certificate{cert},
|
||||
}
|
||||
ts.StartTLS()
|
||||
return ts
|
||||
},
|
||||
ProxyTransport: utilnet.SetTransportDefaults(&http.Transport{TLSClientConfig: &tls.Config{InsecureSkipVerify: true}}),
|
||||
},
|
||||
"https (valid hostname + RootCAs)": {
|
||||
ServerFunc: func(h http.Handler) *httptest.Server {
|
||||
cert, err := tls.X509KeyPair(localhostCert, localhostKey)
|
||||
if err != nil {
|
||||
t.Errorf("https (valid hostname): proxy_test: %v", err)
|
||||
}
|
||||
ts := httptest.NewUnstartedServer(h)
|
||||
ts.TLS = &tls.Config{
|
||||
Certificates: []tls.Certificate{cert},
|
||||
}
|
||||
ts.StartTLS()
|
||||
return ts
|
||||
},
|
||||
ProxyTransport: utilnet.SetTransportDefaults(&http.Transport{TLSClientConfig: &tls.Config{RootCAs: localhostPool}}),
|
||||
},
|
||||
"https (valid hostname + RootCAs + custom dialer)": {
|
||||
ServerFunc: func(h http.Handler) *httptest.Server {
|
||||
cert, err := tls.X509KeyPair(localhostCert, localhostKey)
|
||||
if err != nil {
|
||||
t.Errorf("https (valid hostname): proxy_test: %v", err)
|
||||
}
|
||||
ts := httptest.NewUnstartedServer(h)
|
||||
ts.TLS = &tls.Config{
|
||||
Certificates: []tls.Certificate{cert},
|
||||
}
|
||||
ts.StartTLS()
|
||||
return ts
|
||||
},
|
||||
ProxyTransport: utilnet.SetTransportDefaults(&http.Transport{Dial: net.Dial, TLSClientConfig: &tls.Config{RootCAs: localhostPool}}),
|
||||
},
|
||||
}
|
||||
|
||||
for k, tc := range testcases {
|
||||
|
||||
backendServer := tc.ServerFunc(websocket.Handler(func(ws *websocket.Conn) {
|
||||
defer ws.Close()
|
||||
body := make([]byte, 5)
|
||||
ws.Read(body)
|
||||
ws.Write([]byte("hello " + string(body)))
|
||||
}))
|
||||
defer backendServer.Close()
|
||||
|
||||
serverURL, _ := url.Parse(backendServer.URL)
|
||||
simpleStorage := &SimpleRESTStorage{
|
||||
errors: map[string]error{},
|
||||
resourceLocation: serverURL,
|
||||
resourceLocationTransport: tc.ProxyTransport,
|
||||
expectedResourceNamespace: "myns",
|
||||
}
|
||||
|
||||
namespaceHandler := handleNamespaced(map[string]rest.Storage{"foo": simpleStorage})
|
||||
|
||||
server := newTestServer(namespaceHandler)
|
||||
defer server.Close()
|
||||
|
||||
ws, err := websocket.Dial("ws://"+server.Listener.Addr().String()+"/"+prefix+"/"+newGroupVersion.Group+"/"+newGroupVersion.Version+"/proxy/namespaces/myns/foo/123", "", "http://127.0.0.1/")
|
||||
if err != nil {
|
||||
t.Errorf("%s: websocket dial err: %s", k, err)
|
||||
continue
|
||||
}
|
||||
defer ws.Close()
|
||||
|
||||
if _, err := ws.Write([]byte("world")); err != nil {
|
||||
t.Errorf("%s: write err: %s", k, err)
|
||||
continue
|
||||
}
|
||||
|
||||
response := make([]byte, 20)
|
||||
n, err := ws.Read(response)
|
||||
if err != nil {
|
||||
t.Errorf("%s: read err: %s", k, err)
|
||||
continue
|
||||
}
|
||||
if e, a := "hello world", string(response[0:n]); e != a {
|
||||
t.Errorf("%s: expected '%#v', got '%#v'", k, e, a)
|
||||
continue
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestRedirectOnMissingTrailingSlash(t *testing.T) {
|
||||
table := []struct {
|
||||
// The requested path
|
||||
path string
|
||||
// The path requested on the proxy server.
|
||||
proxyServerPath string
|
||||
// query string
|
||||
query string
|
||||
}{
|
||||
{"/trailing/slash/", "/trailing/slash/", ""},
|
||||
{"/", "/", "test1=value1&test2=value2"},
|
||||
// "/" should be added at the end.
|
||||
{"", "/", "test1=value1&test2=value2"},
|
||||
// "/" should not be added at a non-root path.
|
||||
{"/some/path", "/some/path", ""},
|
||||
}
|
||||
|
||||
for _, item := range table {
|
||||
downstreamServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
|
||||
if req.URL.Path != item.proxyServerPath {
|
||||
t.Errorf("Unexpected request on path: %s, expected path: %s, item: %v", req.URL.Path, item.proxyServerPath, item)
|
||||
}
|
||||
if req.URL.RawQuery != item.query {
|
||||
t.Errorf("Unexpected query on url: %s, expected: %s", req.URL.RawQuery, item.query)
|
||||
}
|
||||
}))
|
||||
defer downstreamServer.Close()
|
||||
|
||||
serverURL, _ := url.Parse(downstreamServer.URL)
|
||||
simpleStorage := &SimpleRESTStorage{
|
||||
errors: map[string]error{},
|
||||
resourceLocation: serverURL,
|
||||
expectedResourceNamespace: "ns",
|
||||
}
|
||||
|
||||
handler := handleNamespaced(map[string]rest.Storage{"foo": simpleStorage})
|
||||
server := newTestServer(handler)
|
||||
defer server.Close()
|
||||
|
||||
proxyTestPattern := "/" + prefix + "/" + newGroupVersion.Group + "/" + newGroupVersion.Version + "/proxy/namespaces/ns/foo/id" + item.path
|
||||
req, err := http.NewRequest(
|
||||
"GET",
|
||||
server.URL+proxyTestPattern+"?"+item.query,
|
||||
strings.NewReader(""),
|
||||
)
|
||||
if err != nil {
|
||||
t.Errorf("unexpected error %v", err)
|
||||
continue
|
||||
}
|
||||
// Note: We are using a default client here, that follows redirects.
|
||||
resp, err := http.DefaultClient.Do(req)
|
||||
if err != nil {
|
||||
t.Errorf("unexpected error %v", err)
|
||||
continue
|
||||
}
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
t.Errorf("Unexpected errorCode: %v, expected: 200. Response: %v, item: %v", resp.StatusCode, resp, item)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// exampleCert was generated from crypto/tls/generate_cert.go with the following command:
|
||||
// go run generate_cert.go --rsa-bits 512 --host example.com --ca --start-date "Jan 1 00:00:00 1970" --duration=1000000h
|
||||
var exampleCert = []byte(`-----BEGIN CERTIFICATE-----
|
||||
MIIBdzCCASGgAwIBAgIRAOVTAdPnfbS5V85mfS90TfIwDQYJKoZIhvcNAQELBQAw
|
||||
EjEQMA4GA1UEChMHQWNtZSBDbzAgFw03MDAxMDEwMDAwMDBaGA8yMDg0MDEyOTE2
|
||||
MDAwMFowEjEQMA4GA1UEChMHQWNtZSBDbzBcMA0GCSqGSIb3DQEBAQUAA0sAMEgC
|
||||
QQCoVSqeu8TBvF+70T7Jm4340YQNhds6IxjRoifenYodAO1dnKGrcbF266DJGunh
|
||||
nIjQH7B12tduhl0fLK4Ezf7/AgMBAAGjUDBOMA4GA1UdDwEB/wQEAwICpDATBgNV
|
||||
HSUEDDAKBggrBgEFBQcDATAPBgNVHRMBAf8EBTADAQH/MBYGA1UdEQQPMA2CC2V4
|
||||
YW1wbGUuY29tMA0GCSqGSIb3DQEBCwUAA0EAk1kVa5uZ/AzwYDVcS9bpM/czwjjV
|
||||
xq3VeSCfmNa2uNjbFvodmCRwZOHUvipAMGCUCV6j5vMrJ8eMj8tCQ36W9A==
|
||||
-----END CERTIFICATE-----`)
|
||||
|
||||
var exampleKey = []byte(`-----BEGIN RSA PRIVATE KEY-----
|
||||
MIIBOgIBAAJBAKhVKp67xMG8X7vRPsmbjfjRhA2F2zojGNGiJ96dih0A7V2coatx
|
||||
sXbroMka6eGciNAfsHXa126GXR8srgTN/v8CAwEAAQJASdzdD7vKsUwMIejGCUb1
|
||||
fAnLTPfAY3lFCa+CmR89nE22dAoRDv+5RbnBsZ58BazPNJHrsVPRlfXB3OQmSQr0
|
||||
SQIhANoJhs+xOJE/i8nJv0uAbzKyiD1YkvRkta0GpUOULyAVAiEAxaQus3E/SuqD
|
||||
P7y5NeJnE7X6XkyC35zrsJRkz7orE8MCIHdDjsI8pjyNDeGqwUCDWE/a6DrmIDwe
|
||||
emHSqMN2YvChAiEAnxLCM9NWaenOsaIoP+J1rDuvw+4499nJKVqGuVrSCRkCIEqK
|
||||
4KSchPMc3x8M/uhw9oWTtKFmjA/PPh0FsWCdKrEy
|
||||
-----END RSA PRIVATE KEY-----`)
|
||||
|
||||
// localhostCert was generated from crypto/tls/generate_cert.go with the following command:
|
||||
// go run generate_cert.go --rsa-bits 512 --host 127.0.0.1,::1,example.com --ca --start-date "Jan 1 00:00:00 1970" --duration=1000000h
|
||||
var localhostCert = []byte(`-----BEGIN CERTIFICATE-----
|
||||
MIIBjzCCATmgAwIBAgIRAKpi2WmTcFrVjxrl5n5YDUEwDQYJKoZIhvcNAQELBQAw
|
||||
EjEQMA4GA1UEChMHQWNtZSBDbzAgFw03MDAxMDEwMDAwMDBaGA8yMDg0MDEyOTE2
|
||||
MDAwMFowEjEQMA4GA1UEChMHQWNtZSBDbzBcMA0GCSqGSIb3DQEBAQUAA0sAMEgC
|
||||
QQC9fEbRszP3t14Gr4oahV7zFObBI4TfA5i7YnlMXeLinb7MnvT4bkfOJzE6zktn
|
||||
59zP7UiHs3l4YOuqrjiwM413AgMBAAGjaDBmMA4GA1UdDwEB/wQEAwICpDATBgNV
|
||||
HSUEDDAKBggrBgEFBQcDATAPBgNVHRMBAf8EBTADAQH/MC4GA1UdEQQnMCWCC2V4
|
||||
YW1wbGUuY29thwR/AAABhxAAAAAAAAAAAAAAAAAAAAABMA0GCSqGSIb3DQEBCwUA
|
||||
A0EAUsVE6KMnza/ZbodLlyeMzdo7EM/5nb5ywyOxgIOCf0OOLHsPS9ueGLQX9HEG
|
||||
//yjTXuhNcUugExIjM/AIwAZPQ==
|
||||
-----END CERTIFICATE-----`)
|
||||
|
||||
// localhostKey is the private key for localhostCert.
|
||||
var localhostKey = []byte(`-----BEGIN RSA PRIVATE KEY-----
|
||||
MIIBOwIBAAJBAL18RtGzM/e3XgavihqFXvMU5sEjhN8DmLtieUxd4uKdvsye9Phu
|
||||
R84nMTrOS2fn3M/tSIezeXhg66quOLAzjXcCAwEAAQJBAKcRxH9wuglYLBdI/0OT
|
||||
BLzfWPZCEw1vZmMR2FF1Fm8nkNOVDPleeVGTWoOEcYYlQbpTmkGSxJ6ya+hqRi6x
|
||||
goECIQDx3+X49fwpL6B5qpJIJMyZBSCuMhH4B7JevhGGFENi3wIhAMiNJN5Q3UkL
|
||||
IuSvv03kaPR5XVQ99/UeEetUgGvBcABpAiBJSBzVITIVCGkGc7d+RCf49KTCIklv
|
||||
bGWObufAR8Ni4QIgWpILjW8dkGg8GOUZ0zaNA6Nvt6TIv2UWGJ4v5PoV98kCIQDx
|
||||
rIiZs5QbKdycsv9gQJzwQAogC8o04X3Zz3dsoX+h4A==
|
||||
-----END RSA PRIVATE KEY-----`)
|
||||
|
|
@ -131,8 +131,6 @@ test/e2e/kubectl/kubectl.go: "should support proxy with --port 0 "
|
|||
test/e2e/kubectl/kubectl.go: "should support --unix-socket=/path "
|
||||
test/e2e/network/dns.go: "should provide DNS for the cluster "
|
||||
test/e2e/network/dns.go: "should provide DNS for services "
|
||||
test/e2e/network/proxy.go: "should proxy logs on node with explicit kubelet port "
|
||||
test/e2e/network/proxy.go: "should proxy logs on node "
|
||||
test/e2e/network/proxy.go: "should proxy logs on node with explicit kubelet port using proxy subresource "
|
||||
test/e2e/network/proxy.go: "should proxy logs on node using proxy subresource "
|
||||
test/e2e/network/proxy.go: "should proxy through a service and a pod "
|
||||
|
|
|
|||
|
|
@ -68,7 +68,7 @@ func CheckCadvisorHealthOnAllNodes(c clientset.Interface, timeout time.Duration)
|
|||
for _, node := range nodeList.Items {
|
||||
// cadvisor is not accessible directly unless its port (4194 by default) is exposed.
|
||||
// Here, we access '/stats/' REST endpoint on the kubelet which polls cadvisor internally.
|
||||
statsResource := fmt.Sprintf("api/v1/proxy/nodes/%s/stats/", node.Name)
|
||||
statsResource := fmt.Sprintf("api/v1/nodes/%s/proxy/stats/", node.Name)
|
||||
By(fmt.Sprintf("Querying stats from node %s using url %s", node.Name, statsResource))
|
||||
_, err = c.CoreV1().RESTClient().Get().AbsPath(statsResource).Timeout(timeout).Do().Raw()
|
||||
if err != nil {
|
||||
|
|
|
|||
|
|
@ -61,22 +61,6 @@ var _ = SIGDescribe("Proxy", func() {
|
|||
f := framework.NewFramework("proxy", options, nil)
|
||||
prefix := "/api/" + version
|
||||
|
||||
// Port here has to be kept in sync with default kubelet port.
|
||||
/*
|
||||
Testname: proxy-prefix-node-logs-port
|
||||
Description: Ensure that proxy on node logs works with generic top
|
||||
level prefix proxy and explicit kubelet port.
|
||||
*/
|
||||
framework.ConformanceIt("should proxy logs on node with explicit kubelet port ", func() { nodeProxyTest(f, prefix+"/proxy/nodes/", ":10250/logs/") })
|
||||
|
||||
/*
|
||||
Testname: proxy-prefix-node-logs
|
||||
Description: Ensure that proxy on node logs works with generic top
|
||||
level prefix proxy.
|
||||
*/
|
||||
framework.ConformanceIt("should proxy logs on node ", func() { nodeProxyTest(f, prefix+"/proxy/nodes/", "/logs/") })
|
||||
It("should proxy to cadvisor", func() { nodeProxyTest(f, prefix+"/proxy/nodes/", ":4194/containers/") })
|
||||
|
||||
/*
|
||||
Testname: proxy-subresource-node-logs-port
|
||||
Description: Ensure that proxy on node logs works with node proxy
|
||||
|
|
@ -185,36 +169,15 @@ var _ = SIGDescribe("Proxy", func() {
|
|||
|
||||
// table constructors
|
||||
// Try proxying through the service and directly to through the pod.
|
||||
svcProxyURL := func(scheme, port string) string {
|
||||
return prefix + "/proxy/namespaces/" + f.Namespace.Name + "/services/" + net.JoinSchemeNamePort(scheme, service.Name, port)
|
||||
}
|
||||
subresourceServiceProxyURL := func(scheme, port string) string {
|
||||
return prefix + "/namespaces/" + f.Namespace.Name + "/services/" + net.JoinSchemeNamePort(scheme, service.Name, port) + "/proxy"
|
||||
}
|
||||
podProxyURL := func(scheme, port string) string {
|
||||
return prefix + "/proxy/namespaces/" + f.Namespace.Name + "/pods/" + net.JoinSchemeNamePort(scheme, pods[0].Name, port)
|
||||
}
|
||||
subresourcePodProxyURL := func(scheme, port string) string {
|
||||
return prefix + "/namespaces/" + f.Namespace.Name + "/pods/" + net.JoinSchemeNamePort(scheme, pods[0].Name, port) + "/proxy"
|
||||
}
|
||||
|
||||
// construct the table
|
||||
expectations := map[string]string{
|
||||
svcProxyURL("", "portname1") + "/": "foo",
|
||||
svcProxyURL("", "80") + "/": "foo",
|
||||
svcProxyURL("", "portname2") + "/": "bar",
|
||||
svcProxyURL("", "81") + "/": "bar",
|
||||
|
||||
svcProxyURL("http", "portname1") + "/": "foo",
|
||||
svcProxyURL("http", "80") + "/": "foo",
|
||||
svcProxyURL("http", "portname2") + "/": "bar",
|
||||
svcProxyURL("http", "81") + "/": "bar",
|
||||
|
||||
svcProxyURL("https", "tlsportname1") + "/": "tls baz",
|
||||
svcProxyURL("https", "443") + "/": "tls baz",
|
||||
svcProxyURL("https", "tlsportname2") + "/": "tls qux",
|
||||
svcProxyURL("https", "444") + "/": "tls qux",
|
||||
|
||||
subresourceServiceProxyURL("", "portname1") + "/": "foo",
|
||||
subresourceServiceProxyURL("http", "portname1") + "/": "foo",
|
||||
subresourceServiceProxyURL("", "portname2") + "/": "bar",
|
||||
|
|
@ -222,14 +185,6 @@ var _ = SIGDescribe("Proxy", func() {
|
|||
subresourceServiceProxyURL("https", "tlsportname1") + "/": "tls baz",
|
||||
subresourceServiceProxyURL("https", "tlsportname2") + "/": "tls qux",
|
||||
|
||||
podProxyURL("", "1080") + "/": `<a href="` + podProxyURL("", "1080") + `/rewriteme">test</a>`,
|
||||
podProxyURL("", "160") + "/": "foo",
|
||||
podProxyURL("", "162") + "/": "bar",
|
||||
|
||||
podProxyURL("http", "1080") + "/": `<a href="` + podProxyURL("http", "1080") + `/rewriteme">test</a>`,
|
||||
podProxyURL("http", "160") + "/": "foo",
|
||||
podProxyURL("http", "162") + "/": "bar",
|
||||
|
||||
subresourcePodProxyURL("", "") + "/": `<a href="` + subresourcePodProxyURL("", "") + `/rewriteme">test</a>`,
|
||||
subresourcePodProxyURL("", "1080") + "/": `<a href="` + subresourcePodProxyURL("", "1080") + `/rewriteme">test</a>`,
|
||||
subresourcePodProxyURL("http", "1080") + "/": `<a href="` + subresourcePodProxyURL("http", "1080") + `/rewriteme">test</a>`,
|
||||
|
|
|
|||
|
|
@ -232,7 +232,7 @@ func getKubeletConfigOkCondition(cs []apiv1.NodeCondition) *apiv1.NodeCondition
|
|||
|
||||
// Causes the test to fail, or returns a status 200 response from the /configz endpoint
|
||||
func pollConfigz(timeout time.Duration, pollInterval time.Duration) *http.Response {
|
||||
endpoint := fmt.Sprintf("http://127.0.0.1:8080/api/v1/proxy/nodes/%s/configz", framework.TestContext.NodeName)
|
||||
endpoint := fmt.Sprintf("http://127.0.0.1:8080/api/v1/nodes/%s/proxy/configz", framework.TestContext.NodeName)
|
||||
client := &http.Client{}
|
||||
req, err := http.NewRequest("GET", endpoint, nil)
|
||||
framework.ExpectNoError(err)
|
||||
|
|
|
|||
|
|
@ -100,6 +100,10 @@ func pathWithPrefix(prefix, resource, namespace, name string) string {
|
|||
return testapi.Default.ResourcePathWithPrefix(prefix, resource, namespace, name)
|
||||
}
|
||||
|
||||
func pathWithSubResource(resource, namespace, name, subresource string) string {
|
||||
return testapi.Default.SubResourcePath(resource, namespace, name, subresource)
|
||||
}
|
||||
|
||||
func timeoutPath(resource, namespace, name string) string {
|
||||
return addTimeoutFlag(testapi.Default.ResourcePath(resource, namespace, name))
|
||||
}
|
||||
|
|
@ -326,7 +330,7 @@ func getTestRequests(namespace string) []struct {
|
|||
// whenever a service is created, but this test does not run that controller)
|
||||
{"POST", timeoutPath("endpoints", namespace, ""), emptyEndpoints, integration.Code201},
|
||||
// Should return service unavailable when endpoint.subset is empty.
|
||||
{"GET", pathWithPrefix("proxy", "services", namespace, "a") + "/", "", integration.Code503},
|
||||
{"GET", pathWithSubResource("services", namespace, "a", "proxy") + "/", "", integration.Code503},
|
||||
{"PUT", timeoutPath("services", namespace, "a"), aService, integration.Code200},
|
||||
{"GET", path("services", namespace, "a"), "", integration.Code200},
|
||||
{"DELETE", timeoutPath("endpoints", namespace, "a"), "", integration.Code200},
|
||||
|
|
@ -379,7 +383,7 @@ func getTestRequests(namespace string) []struct {
|
|||
{"DELETE", timeoutPath("foo", namespace, ""), "", integration.Code404},
|
||||
|
||||
// Special verbs on nodes
|
||||
{"GET", pathWithPrefix("proxy", "nodes", namespace, "a"), "", integration.Code404},
|
||||
{"GET", pathWithSubResource("nodes", namespace, "a", "proxy"), "", integration.Code404},
|
||||
{"GET", pathWithPrefix("redirect", "nodes", namespace, "a"), "", integration.Code404},
|
||||
// TODO: test .../watch/..., which doesn't end before the test timeout.
|
||||
// TODO: figure out how to create a node so that it can successfully proxy/redirect.
|
||||
|
|
|
|||
Loading…
Reference in New Issue