vendor: remove dep and use vndr
Signed-off-by: Antonio Murdaca <runcom@redhat.com>
This commit is contained in:
Antonio Murdaca
parent
16f44674a4
commit
148e72d81e
16131 changed files with 73815 additions and 4235138 deletions
65
vendor/k8s.io/kubernetes/pkg/kubelet/server/streaming/BUILD
generated
vendored
65
vendor/k8s.io/kubernetes/pkg/kubelet/server/streaming/BUILD
generated
vendored
|
|
@ -1,65 +0,0 @@
|
|||
package(default_visibility = ["//visibility:public"])
|
||||
|
||||
licenses(["notice"])
|
||||
|
||||
load(
|
||||
"@io_bazel_rules_go//go:def.bzl",
|
||||
"go_library",
|
||||
"go_test",
|
||||
)
|
||||
|
||||
go_library(
|
||||
name = "go_default_library",
|
||||
srcs = [
|
||||
"errors.go",
|
||||
"request_cache.go",
|
||||
"server.go",
|
||||
],
|
||||
tags = ["automanaged"],
|
||||
deps = [
|
||||
"//pkg/kubelet/api/v1alpha1/runtime:go_default_library",
|
||||
"//pkg/kubelet/server/portforward:go_default_library",
|
||||
"//pkg/kubelet/server/remotecommand:go_default_library",
|
||||
"//pkg/util/term:go_default_library",
|
||||
"//vendor:github.com/emicklei/go-restful",
|
||||
"//vendor:google.golang.org/grpc",
|
||||
"//vendor:google.golang.org/grpc/codes",
|
||||
"//vendor:k8s.io/apimachinery/pkg/types",
|
||||
"//vendor:k8s.io/client-go/util/clock",
|
||||
],
|
||||
)
|
||||
|
||||
go_test(
|
||||
name = "go_default_test",
|
||||
srcs = [
|
||||
"request_cache_test.go",
|
||||
"server_test.go",
|
||||
],
|
||||
library = ":go_default_library",
|
||||
tags = ["automanaged"],
|
||||
deps = [
|
||||
"//pkg/client/unversioned/remotecommand:go_default_library",
|
||||
"//pkg/kubelet/api/v1alpha1/runtime:go_default_library",
|
||||
"//pkg/kubelet/server/portforward:go_default_library",
|
||||
"//pkg/kubelet/server/remotecommand:go_default_library",
|
||||
"//pkg/util/term:go_default_library",
|
||||
"//vendor:github.com/stretchr/testify/assert",
|
||||
"//vendor:github.com/stretchr/testify/require",
|
||||
"//vendor:k8s.io/client-go/pkg/api",
|
||||
"//vendor:k8s.io/client-go/rest",
|
||||
"//vendor:k8s.io/client-go/util/clock",
|
||||
],
|
||||
)
|
||||
|
||||
filegroup(
|
||||
name = "package-srcs",
|
||||
srcs = glob(["**"]),
|
||||
tags = ["automanaged"],
|
||||
visibility = ["//visibility:private"],
|
||||
)
|
||||
|
||||
filegroup(
|
||||
name = "all-srcs",
|
||||
srcs = [":package-srcs"],
|
||||
tags = ["automanaged"],
|
||||
)
|
||||
221
vendor/k8s.io/kubernetes/pkg/kubelet/server/streaming/request_cache_test.go
generated
vendored
221
vendor/k8s.io/kubernetes/pkg/kubelet/server/streaming/request_cache_test.go
generated
vendored
|
|
@ -1,221 +0,0 @@
|
|||
/*
|
||||
Copyright 2016 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 streaming
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strconv"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"k8s.io/client-go/util/clock"
|
||||
)
|
||||
|
||||
func TestInsert(t *testing.T) {
|
||||
c, _ := newTestCache()
|
||||
|
||||
// Insert normal
|
||||
oldestTok, err := c.Insert(nextRequest())
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, oldestTok, TokenLen)
|
||||
assertCacheSize(t, c, 1)
|
||||
|
||||
// Insert until full
|
||||
for i := 0; i < MaxInFlight-2; i++ {
|
||||
tok, err := c.Insert(nextRequest())
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, tok, TokenLen)
|
||||
}
|
||||
assertCacheSize(t, c, MaxInFlight-1)
|
||||
|
||||
newestReq := nextRequest()
|
||||
newestTok, err := c.Insert(newestReq)
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, newestTok, TokenLen)
|
||||
assertCacheSize(t, c, MaxInFlight)
|
||||
require.Contains(t, c.tokens, oldestTok, "oldest request should still be cached")
|
||||
|
||||
// Consume newest token.
|
||||
req, ok := c.Consume(newestTok)
|
||||
assert.True(t, ok, "newest request should still be cached")
|
||||
assert.Equal(t, newestReq, req)
|
||||
require.Contains(t, c.tokens, oldestTok, "oldest request should still be cached")
|
||||
|
||||
// Insert again (still full)
|
||||
tok, err := c.Insert(nextRequest())
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, tok, TokenLen)
|
||||
assertCacheSize(t, c, MaxInFlight)
|
||||
|
||||
// Insert again (should evict)
|
||||
_, err = c.Insert(nextRequest())
|
||||
assert.Error(t, err, "should reject further requests")
|
||||
errResponse := httptest.NewRecorder()
|
||||
require.NoError(t, WriteError(err, errResponse))
|
||||
assert.Equal(t, errResponse.Code, http.StatusTooManyRequests)
|
||||
assert.Equal(t, strconv.Itoa(int(CacheTTL.Seconds())), errResponse.HeaderMap.Get("Retry-After"))
|
||||
|
||||
assertCacheSize(t, c, MaxInFlight)
|
||||
_, ok = c.Consume(oldestTok)
|
||||
assert.True(t, ok, "oldest request should be valid")
|
||||
}
|
||||
|
||||
func TestConsume(t *testing.T) {
|
||||
c, clock := newTestCache()
|
||||
|
||||
{ // Insert & consume.
|
||||
req := nextRequest()
|
||||
tok, err := c.Insert(req)
|
||||
require.NoError(t, err)
|
||||
assertCacheSize(t, c, 1)
|
||||
|
||||
cachedReq, ok := c.Consume(tok)
|
||||
assert.True(t, ok)
|
||||
assert.Equal(t, req, cachedReq)
|
||||
assertCacheSize(t, c, 0)
|
||||
}
|
||||
|
||||
{ // Insert & consume out of order
|
||||
req1 := nextRequest()
|
||||
tok1, err := c.Insert(req1)
|
||||
require.NoError(t, err)
|
||||
assertCacheSize(t, c, 1)
|
||||
|
||||
req2 := nextRequest()
|
||||
tok2, err := c.Insert(req2)
|
||||
require.NoError(t, err)
|
||||
assertCacheSize(t, c, 2)
|
||||
|
||||
cachedReq2, ok := c.Consume(tok2)
|
||||
assert.True(t, ok)
|
||||
assert.Equal(t, req2, cachedReq2)
|
||||
assertCacheSize(t, c, 1)
|
||||
|
||||
cachedReq1, ok := c.Consume(tok1)
|
||||
assert.True(t, ok)
|
||||
assert.Equal(t, req1, cachedReq1)
|
||||
assertCacheSize(t, c, 0)
|
||||
}
|
||||
|
||||
{ // Consume a second time
|
||||
req := nextRequest()
|
||||
tok, err := c.Insert(req)
|
||||
require.NoError(t, err)
|
||||
assertCacheSize(t, c, 1)
|
||||
|
||||
cachedReq, ok := c.Consume(tok)
|
||||
assert.True(t, ok)
|
||||
assert.Equal(t, req, cachedReq)
|
||||
assertCacheSize(t, c, 0)
|
||||
|
||||
_, ok = c.Consume(tok)
|
||||
assert.False(t, ok)
|
||||
assertCacheSize(t, c, 0)
|
||||
}
|
||||
|
||||
{ // Consume without insert
|
||||
_, ok := c.Consume("fooBAR")
|
||||
assert.False(t, ok)
|
||||
assertCacheSize(t, c, 0)
|
||||
}
|
||||
|
||||
{ // Consume expired
|
||||
tok, err := c.Insert(nextRequest())
|
||||
require.NoError(t, err)
|
||||
assertCacheSize(t, c, 1)
|
||||
|
||||
clock.Step(2 * CacheTTL)
|
||||
|
||||
_, ok := c.Consume(tok)
|
||||
assert.False(t, ok)
|
||||
assertCacheSize(t, c, 0)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGC(t *testing.T) {
|
||||
c, clock := newTestCache()
|
||||
|
||||
// When empty
|
||||
c.gc()
|
||||
assertCacheSize(t, c, 0)
|
||||
|
||||
tok1, err := c.Insert(nextRequest())
|
||||
require.NoError(t, err)
|
||||
assertCacheSize(t, c, 1)
|
||||
clock.Step(10 * time.Second)
|
||||
tok2, err := c.Insert(nextRequest())
|
||||
require.NoError(t, err)
|
||||
assertCacheSize(t, c, 2)
|
||||
|
||||
// expired: tok1, tok2
|
||||
// non-expired: tok3, tok4
|
||||
clock.Step(2 * CacheTTL)
|
||||
tok3, err := c.Insert(nextRequest())
|
||||
require.NoError(t, err)
|
||||
assertCacheSize(t, c, 1)
|
||||
clock.Step(10 * time.Second)
|
||||
tok4, err := c.Insert(nextRequest())
|
||||
require.NoError(t, err)
|
||||
assertCacheSize(t, c, 2)
|
||||
|
||||
_, ok := c.Consume(tok1)
|
||||
assert.False(t, ok)
|
||||
_, ok = c.Consume(tok2)
|
||||
assert.False(t, ok)
|
||||
_, ok = c.Consume(tok3)
|
||||
assert.True(t, ok)
|
||||
_, ok = c.Consume(tok4)
|
||||
assert.True(t, ok)
|
||||
|
||||
// When full, nothing is expired.
|
||||
for i := 0; i < MaxInFlight; i++ {
|
||||
_, err := c.Insert(nextRequest())
|
||||
require.NoError(t, err)
|
||||
}
|
||||
assertCacheSize(t, c, MaxInFlight)
|
||||
|
||||
// When everything is expired
|
||||
clock.Step(2 * CacheTTL)
|
||||
_, err = c.Insert(nextRequest())
|
||||
require.NoError(t, err)
|
||||
assertCacheSize(t, c, 1)
|
||||
}
|
||||
|
||||
func newTestCache() (*requestCache, *clock.FakeClock) {
|
||||
c := newRequestCache()
|
||||
fakeClock := clock.NewFakeClock(time.Now())
|
||||
c.clock = fakeClock
|
||||
return c, fakeClock
|
||||
}
|
||||
|
||||
func assertCacheSize(t *testing.T, cache *requestCache, expectedSize int) {
|
||||
tokenLen := len(cache.tokens)
|
||||
llLen := cache.ll.Len()
|
||||
assert.Equal(t, tokenLen, llLen, "inconsistent cache size! len(tokens)=%d; len(ll)=%d", tokenLen, llLen)
|
||||
assert.Equal(t, expectedSize, tokenLen, "unexpected cache size!")
|
||||
}
|
||||
|
||||
var requestUID = 0
|
||||
|
||||
func nextRequest() interface{} {
|
||||
requestUID++
|
||||
return requestUID
|
||||
}
|
||||
32
vendor/k8s.io/kubernetes/pkg/kubelet/server/streaming/server.go
generated
vendored
32
vendor/k8s.io/kubernetes/pkg/kubelet/server/streaming/server.go
generated
vendored
|
|
@ -80,7 +80,12 @@ type Config struct {
|
|||
// The streaming protocols the server supports (understands and permits). See
|
||||
// k8s.io/kubernetes/pkg/kubelet/server/remotecommand/constants.go for available protocols.
|
||||
// Only used for SPDY streaming.
|
||||
SupportedProtocols []string
|
||||
SupportedRemoteCommandProtocols []string
|
||||
|
||||
// The streaming protocols the server supports (understands and permits). See
|
||||
// k8s.io/kubernetes/pkg/kubelet/server/portforward/constants.go for available protocols.
|
||||
// Only used for SPDY streaming.
|
||||
SupportedPortForwardProtocols []string
|
||||
|
||||
// The config for serving over TLS. If nil, TLS will not be used.
|
||||
TLSConfig *tls.Config
|
||||
|
|
@ -89,9 +94,10 @@ type Config struct {
|
|||
// DefaultConfig provides default values for server Config. The DefaultConfig is partial, so
|
||||
// some fields like Addr must still be provided.
|
||||
var DefaultConfig = Config{
|
||||
StreamIdleTimeout: 4 * time.Hour,
|
||||
StreamCreationTimeout: remotecommand.DefaultStreamCreationTimeout,
|
||||
SupportedProtocols: remotecommand.SupportedStreamingProtocols,
|
||||
StreamIdleTimeout: 4 * time.Hour,
|
||||
StreamCreationTimeout: remotecommand.DefaultStreamCreationTimeout,
|
||||
SupportedRemoteCommandProtocols: remotecommand.SupportedStreamingProtocols,
|
||||
SupportedPortForwardProtocols: portforward.SupportedProtocols,
|
||||
}
|
||||
|
||||
// TODO(timstclair): Add auth(n/z) interface & handling.
|
||||
|
|
@ -248,7 +254,7 @@ func (s *server) serveExec(req *restful.Request, resp *restful.Response) {
|
|||
streamOpts,
|
||||
s.config.StreamIdleTimeout,
|
||||
s.config.StreamCreationTimeout,
|
||||
s.config.SupportedProtocols)
|
||||
s.config.SupportedRemoteCommandProtocols)
|
||||
}
|
||||
|
||||
func (s *server) serveAttach(req *restful.Request, resp *restful.Response) {
|
||||
|
|
@ -280,7 +286,7 @@ func (s *server) serveAttach(req *restful.Request, resp *restful.Response) {
|
|||
streamOpts,
|
||||
s.config.StreamIdleTimeout,
|
||||
s.config.StreamCreationTimeout,
|
||||
s.config.SupportedProtocols)
|
||||
s.config.SupportedRemoteCommandProtocols)
|
||||
}
|
||||
|
||||
func (s *server) servePortForward(req *restful.Request, resp *restful.Response) {
|
||||
|
|
@ -296,14 +302,22 @@ func (s *server) servePortForward(req *restful.Request, resp *restful.Response)
|
|||
return
|
||||
}
|
||||
|
||||
portForwardOptions, err := portforward.BuildV4Options(pf.Port)
|
||||
if err != nil {
|
||||
resp.WriteError(http.StatusBadRequest, err)
|
||||
return
|
||||
}
|
||||
|
||||
portforward.ServePortForward(
|
||||
resp.ResponseWriter,
|
||||
req.Request,
|
||||
s.runtime,
|
||||
pf.PodSandboxId,
|
||||
"", // unused: podUID
|
||||
portForwardOptions,
|
||||
s.config.StreamIdleTimeout,
|
||||
s.config.StreamCreationTimeout)
|
||||
s.config.StreamCreationTimeout,
|
||||
s.config.SupportedPortForwardProtocols)
|
||||
}
|
||||
|
||||
// criAdapter wraps the Runtime functions to conform to the remotecommand interfaces.
|
||||
|
|
@ -324,6 +338,6 @@ func (a *criAdapter) AttachContainer(podName string, podUID types.UID, container
|
|||
return a.Attach(container, in, out, err, tty, resize)
|
||||
}
|
||||
|
||||
func (a *criAdapter) PortForward(podName string, podUID types.UID, port uint16, stream io.ReadWriteCloser) error {
|
||||
return a.Runtime.PortForward(podName, int32(port), stream)
|
||||
func (a *criAdapter) PortForward(podName string, podUID types.UID, port int32, stream io.ReadWriteCloser) error {
|
||||
return a.Runtime.PortForward(podName, port, stream)
|
||||
}
|
||||
|
|
|
|||
422
vendor/k8s.io/kubernetes/pkg/kubelet/server/streaming/server_test.go
generated
vendored
422
vendor/k8s.io/kubernetes/pkg/kubelet/server/streaming/server_test.go
generated
vendored
|
|
@ -1,422 +0,0 @@
|
|||
/*
|
||||
Copyright 2016 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 streaming
|
||||
|
||||
import (
|
||||
"crypto/tls"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"net/url"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"k8s.io/client-go/pkg/api"
|
||||
restclient "k8s.io/client-go/rest"
|
||||
"k8s.io/kubernetes/pkg/client/unversioned/remotecommand"
|
||||
runtimeapi "k8s.io/kubernetes/pkg/kubelet/api/v1alpha1/runtime"
|
||||
kubeletportforward "k8s.io/kubernetes/pkg/kubelet/server/portforward"
|
||||
kubeletremotecommand "k8s.io/kubernetes/pkg/kubelet/server/remotecommand"
|
||||
"k8s.io/kubernetes/pkg/util/term"
|
||||
)
|
||||
|
||||
const (
|
||||
testAddr = "localhost:12345"
|
||||
testContainerID = "container789"
|
||||
testPodSandboxID = "pod0987"
|
||||
)
|
||||
|
||||
func TestGetExec(t *testing.T) {
|
||||
type testcase struct {
|
||||
cmd []string
|
||||
tty bool
|
||||
stdin bool
|
||||
}
|
||||
testcases := []testcase{
|
||||
{[]string{"echo", "foo"}, false, false},
|
||||
{[]string{"date"}, true, false},
|
||||
{[]string{"date"}, false, true},
|
||||
{[]string{"date"}, true, true},
|
||||
}
|
||||
serv, err := NewServer(Config{
|
||||
Addr: testAddr,
|
||||
}, nil)
|
||||
assert.NoError(t, err)
|
||||
|
||||
tlsServer, err := NewServer(Config{
|
||||
Addr: testAddr,
|
||||
TLSConfig: &tls.Config{},
|
||||
}, nil)
|
||||
assert.NoError(t, err)
|
||||
|
||||
const pathPrefix = "cri/shim"
|
||||
prefixServer, err := NewServer(Config{
|
||||
Addr: testAddr,
|
||||
BaseURL: &url.URL{
|
||||
Scheme: "http",
|
||||
Host: testAddr,
|
||||
Path: "/" + pathPrefix + "/",
|
||||
},
|
||||
}, nil)
|
||||
assert.NoError(t, err)
|
||||
|
||||
assertRequestToken := func(test testcase, cache *requestCache, token string) {
|
||||
req, ok := cache.Consume(token)
|
||||
require.True(t, ok, "token %s not found! testcase=%+v", token, test)
|
||||
assert.Equal(t, testContainerID, req.(*runtimeapi.ExecRequest).ContainerId, "testcase=%+v", test)
|
||||
assert.Equal(t, test.cmd, req.(*runtimeapi.ExecRequest).Cmd, "testcase=%+v", test)
|
||||
assert.Equal(t, test.tty, req.(*runtimeapi.ExecRequest).Tty, "testcase=%+v", test)
|
||||
assert.Equal(t, test.stdin, req.(*runtimeapi.ExecRequest).Stdin, "testcase=%+v", test)
|
||||
}
|
||||
containerID := testContainerID
|
||||
for _, test := range testcases {
|
||||
request := &runtimeapi.ExecRequest{
|
||||
ContainerId: containerID,
|
||||
Cmd: test.cmd,
|
||||
Tty: test.tty,
|
||||
Stdin: test.stdin,
|
||||
}
|
||||
{ // Non-TLS
|
||||
resp, err := serv.GetExec(request)
|
||||
assert.NoError(t, err, "testcase=%+v", test)
|
||||
expectedURL := "http://" + testAddr + "/exec/"
|
||||
assert.Contains(t, resp.Url, expectedURL, "testcase=%+v", test)
|
||||
token := strings.TrimPrefix(resp.Url, expectedURL)
|
||||
assertRequestToken(test, serv.(*server).cache, token)
|
||||
}
|
||||
|
||||
{ // TLS
|
||||
resp, err := tlsServer.GetExec(request)
|
||||
assert.NoError(t, err, "testcase=%+v", test)
|
||||
expectedURL := "https://" + testAddr + "/exec/"
|
||||
assert.Contains(t, resp.Url, expectedURL, "testcase=%+v", test)
|
||||
token := strings.TrimPrefix(resp.Url, expectedURL)
|
||||
assertRequestToken(test, tlsServer.(*server).cache, token)
|
||||
}
|
||||
|
||||
{ // Path prefix
|
||||
resp, err := prefixServer.GetExec(request)
|
||||
assert.NoError(t, err, "testcase=%+v", test)
|
||||
expectedURL := "http://" + testAddr + "/" + pathPrefix + "/exec/"
|
||||
assert.Contains(t, resp.Url, expectedURL, "testcase=%+v", test)
|
||||
token := strings.TrimPrefix(resp.Url, expectedURL)
|
||||
assertRequestToken(test, prefixServer.(*server).cache, token)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetAttach(t *testing.T) {
|
||||
type testcase struct {
|
||||
tty bool
|
||||
stdin bool
|
||||
}
|
||||
testcases := []testcase{
|
||||
{false, false},
|
||||
{true, false},
|
||||
{false, true},
|
||||
{true, true},
|
||||
}
|
||||
serv, err := NewServer(Config{
|
||||
Addr: testAddr,
|
||||
}, nil)
|
||||
assert.NoError(t, err)
|
||||
|
||||
tlsServer, err := NewServer(Config{
|
||||
Addr: testAddr,
|
||||
TLSConfig: &tls.Config{},
|
||||
}, nil)
|
||||
assert.NoError(t, err)
|
||||
|
||||
assertRequestToken := func(test testcase, cache *requestCache, token string) {
|
||||
req, ok := cache.Consume(token)
|
||||
require.True(t, ok, "token %s not found! testcase=%+v", token, test)
|
||||
assert.Equal(t, testContainerID, req.(*runtimeapi.AttachRequest).ContainerId, "testcase=%+v", test)
|
||||
assert.Equal(t, test.tty, req.(*runtimeapi.AttachRequest).Tty, "testcase=%+v", test)
|
||||
assert.Equal(t, test.stdin, req.(*runtimeapi.AttachRequest).Stdin, "testcase=%+v", test)
|
||||
}
|
||||
containerID := testContainerID
|
||||
for _, test := range testcases {
|
||||
request := &runtimeapi.AttachRequest{
|
||||
ContainerId: containerID,
|
||||
Stdin: test.stdin,
|
||||
Tty: test.tty,
|
||||
}
|
||||
{ // Non-TLS
|
||||
resp, err := serv.GetAttach(request)
|
||||
assert.NoError(t, err, "testcase=%+v", test)
|
||||
expectedURL := "http://" + testAddr + "/attach/"
|
||||
assert.Contains(t, resp.Url, expectedURL, "testcase=%+v", test)
|
||||
token := strings.TrimPrefix(resp.Url, expectedURL)
|
||||
assertRequestToken(test, serv.(*server).cache, token)
|
||||
}
|
||||
|
||||
{ // TLS
|
||||
resp, err := tlsServer.GetAttach(request)
|
||||
assert.NoError(t, err, "testcase=%+v", test)
|
||||
expectedURL := "https://" + testAddr + "/attach/"
|
||||
assert.Contains(t, resp.Url, expectedURL, "testcase=%+v", test)
|
||||
token := strings.TrimPrefix(resp.Url, expectedURL)
|
||||
assertRequestToken(test, tlsServer.(*server).cache, token)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetPortForward(t *testing.T) {
|
||||
podSandboxID := testPodSandboxID
|
||||
request := &runtimeapi.PortForwardRequest{
|
||||
PodSandboxId: podSandboxID,
|
||||
Port: []int32{1, 2, 3, 4},
|
||||
}
|
||||
|
||||
{ // Non-TLS
|
||||
serv, err := NewServer(Config{
|
||||
Addr: testAddr,
|
||||
}, nil)
|
||||
assert.NoError(t, err)
|
||||
resp, err := serv.GetPortForward(request)
|
||||
assert.NoError(t, err)
|
||||
expectedURL := "http://" + testAddr + "/portforward/"
|
||||
assert.True(t, strings.HasPrefix(resp.Url, expectedURL))
|
||||
token := strings.TrimPrefix(resp.Url, expectedURL)
|
||||
req, ok := serv.(*server).cache.Consume(token)
|
||||
require.True(t, ok, "token %s not found!", token)
|
||||
assert.Equal(t, testPodSandboxID, req.(*runtimeapi.PortForwardRequest).PodSandboxId)
|
||||
}
|
||||
|
||||
{ // TLS
|
||||
tlsServer, err := NewServer(Config{
|
||||
Addr: testAddr,
|
||||
TLSConfig: &tls.Config{},
|
||||
}, nil)
|
||||
assert.NoError(t, err)
|
||||
resp, err := tlsServer.GetPortForward(request)
|
||||
assert.NoError(t, err)
|
||||
expectedURL := "https://" + testAddr + "/portforward/"
|
||||
assert.True(t, strings.HasPrefix(resp.Url, expectedURL))
|
||||
token := strings.TrimPrefix(resp.Url, expectedURL)
|
||||
req, ok := tlsServer.(*server).cache.Consume(token)
|
||||
require.True(t, ok, "token %s not found!", token)
|
||||
assert.Equal(t, testPodSandboxID, req.(*runtimeapi.PortForwardRequest).PodSandboxId)
|
||||
}
|
||||
}
|
||||
|
||||
func TestServeExec(t *testing.T) {
|
||||
runRemoteCommandTest(t, "exec")
|
||||
}
|
||||
|
||||
func TestServeAttach(t *testing.T) {
|
||||
runRemoteCommandTest(t, "attach")
|
||||
}
|
||||
|
||||
func TestServePortForward(t *testing.T) {
|
||||
s, testServer := startTestServer(t)
|
||||
defer testServer.Close()
|
||||
|
||||
resp, err := s.GetPortForward(&runtimeapi.PortForwardRequest{
|
||||
PodSandboxId: testPodSandboxID,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
reqURL, err := url.Parse(resp.Url)
|
||||
require.NoError(t, err)
|
||||
|
||||
exec, err := remotecommand.NewExecutor(&restclient.Config{}, "POST", reqURL)
|
||||
require.NoError(t, err)
|
||||
streamConn, _, err := exec.Dial(kubeletportforward.PortForwardProtocolV1Name)
|
||||
require.NoError(t, err)
|
||||
defer streamConn.Close()
|
||||
|
||||
// Create the streams.
|
||||
headers := http.Header{}
|
||||
// Error stream is required, but unused in this test.
|
||||
headers.Set(api.StreamType, api.StreamTypeError)
|
||||
headers.Set(api.PortHeader, strconv.Itoa(testPort))
|
||||
_, err = streamConn.CreateStream(headers)
|
||||
require.NoError(t, err)
|
||||
// Setup the data stream.
|
||||
headers.Set(api.StreamType, api.StreamTypeData)
|
||||
headers.Set(api.PortHeader, strconv.Itoa(testPort))
|
||||
stream, err := streamConn.CreateStream(headers)
|
||||
require.NoError(t, err)
|
||||
|
||||
doClientStreams(t, "portforward", stream, stream, nil)
|
||||
}
|
||||
|
||||
// Run the remote command test.
|
||||
// commandType is either "exec" or "attach".
|
||||
func runRemoteCommandTest(t *testing.T, commandType string) {
|
||||
s, testServer := startTestServer(t)
|
||||
defer testServer.Close()
|
||||
|
||||
var reqURL *url.URL
|
||||
stdin := true
|
||||
containerID := testContainerID
|
||||
switch commandType {
|
||||
case "exec":
|
||||
resp, err := s.GetExec(&runtimeapi.ExecRequest{
|
||||
ContainerId: containerID,
|
||||
Cmd: []string{"echo"},
|
||||
Stdin: stdin,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
reqURL, err = url.Parse(resp.Url)
|
||||
require.NoError(t, err)
|
||||
case "attach":
|
||||
resp, err := s.GetAttach(&runtimeapi.AttachRequest{
|
||||
ContainerId: containerID,
|
||||
Stdin: stdin,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
reqURL, err = url.Parse(resp.Url)
|
||||
require.NoError(t, err)
|
||||
}
|
||||
|
||||
wg := sync.WaitGroup{}
|
||||
wg.Add(2)
|
||||
|
||||
stdinR, stdinW := io.Pipe()
|
||||
stdoutR, stdoutW := io.Pipe()
|
||||
stderrR, stderrW := io.Pipe()
|
||||
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
exec, err := remotecommand.NewExecutor(&restclient.Config{}, "POST", reqURL)
|
||||
require.NoError(t, err)
|
||||
|
||||
opts := remotecommand.StreamOptions{
|
||||
SupportedProtocols: kubeletremotecommand.SupportedStreamingProtocols,
|
||||
Stdin: stdinR,
|
||||
Stdout: stdoutW,
|
||||
Stderr: stderrW,
|
||||
Tty: false,
|
||||
TerminalSizeQueue: nil,
|
||||
}
|
||||
require.NoError(t, exec.Stream(opts))
|
||||
}()
|
||||
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
doClientStreams(t, commandType, stdinW, stdoutR, stderrR)
|
||||
}()
|
||||
|
||||
wg.Wait()
|
||||
|
||||
// Repeat request with the same URL should be a 404.
|
||||
resp, err := http.Get(reqURL.String())
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, http.StatusNotFound, resp.StatusCode)
|
||||
}
|
||||
|
||||
func startTestServer(t *testing.T) (Server, *httptest.Server) {
|
||||
var s Server
|
||||
testServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
s.ServeHTTP(w, r)
|
||||
}))
|
||||
cleanup := true
|
||||
defer func() {
|
||||
if cleanup {
|
||||
testServer.Close()
|
||||
}
|
||||
}()
|
||||
|
||||
testURL, err := url.Parse(testServer.URL)
|
||||
require.NoError(t, err)
|
||||
|
||||
rt := newFakeRuntime(t)
|
||||
config := DefaultConfig
|
||||
config.BaseURL = testURL
|
||||
s, err = NewServer(config, rt)
|
||||
require.NoError(t, err)
|
||||
|
||||
cleanup = false // Caller must close the test server.
|
||||
return s, testServer
|
||||
}
|
||||
|
||||
const (
|
||||
testInput = "abcdefg"
|
||||
testOutput = "fooBARbaz"
|
||||
testErr = "ERROR!!!"
|
||||
testPort = 12345
|
||||
)
|
||||
|
||||
func newFakeRuntime(t *testing.T) *fakeRuntime {
|
||||
return &fakeRuntime{
|
||||
t: t,
|
||||
}
|
||||
}
|
||||
|
||||
type fakeRuntime struct {
|
||||
t *testing.T
|
||||
}
|
||||
|
||||
func (f *fakeRuntime) Exec(containerID string, cmd []string, stdin io.Reader, stdout, stderr io.WriteCloser, tty bool, resize <-chan term.Size) error {
|
||||
assert.Equal(f.t, testContainerID, containerID)
|
||||
doServerStreams(f.t, "exec", stdin, stdout, stderr)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f *fakeRuntime) Attach(containerID string, stdin io.Reader, stdout, stderr io.WriteCloser, tty bool, resize <-chan term.Size) error {
|
||||
assert.Equal(f.t, testContainerID, containerID)
|
||||
doServerStreams(f.t, "attach", stdin, stdout, stderr)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f *fakeRuntime) PortForward(podSandboxID string, port int32, stream io.ReadWriteCloser) error {
|
||||
assert.Equal(f.t, testPodSandboxID, podSandboxID)
|
||||
assert.EqualValues(f.t, testPort, port)
|
||||
doServerStreams(f.t, "portforward", stream, stream, nil)
|
||||
return nil
|
||||
}
|
||||
|
||||
// Send & receive expected input/output. Must be the inverse of doClientStreams.
|
||||
// Function will block until the expected i/o is finished.
|
||||
func doServerStreams(t *testing.T, prefix string, stdin io.Reader, stdout, stderr io.Writer) {
|
||||
if stderr != nil {
|
||||
writeExpected(t, "server stderr", stderr, prefix+testErr)
|
||||
}
|
||||
readExpected(t, "server stdin", stdin, prefix+testInput)
|
||||
writeExpected(t, "server stdout", stdout, prefix+testOutput)
|
||||
}
|
||||
|
||||
// Send & receive expected input/output. Must be the inverse of doServerStreams.
|
||||
// Function will block until the expected i/o is finished.
|
||||
func doClientStreams(t *testing.T, prefix string, stdin io.Writer, stdout, stderr io.Reader) {
|
||||
if stderr != nil {
|
||||
readExpected(t, "client stderr", stderr, prefix+testErr)
|
||||
}
|
||||
writeExpected(t, "client stdin", stdin, prefix+testInput)
|
||||
readExpected(t, "client stdout", stdout, prefix+testOutput)
|
||||
}
|
||||
|
||||
// Read and verify the expected string from the stream.
|
||||
func readExpected(t *testing.T, streamName string, r io.Reader, expected string) {
|
||||
result := make([]byte, len(expected))
|
||||
_, err := io.ReadAtLeast(r, result, len(expected))
|
||||
assert.NoError(t, err, "stream %s", streamName)
|
||||
assert.Equal(t, expected, string(result), "stream %s", streamName)
|
||||
}
|
||||
|
||||
// Write and verify success of the data over the stream.
|
||||
func writeExpected(t *testing.T, streamName string, w io.Writer, data string) {
|
||||
n, err := io.WriteString(w, data)
|
||||
assert.NoError(t, err, "stream %s", streamName)
|
||||
assert.Equal(t, len(data), n, "stream %s", streamName)
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue