github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-07-30 23:15:14 +00:00
Code Issues Projects Releases Wiki Activity

Merge pull request #23907 from Random-Liu/all-but-image-related-functions

Browse Source 
Automatic merge from submit-queue

Kubelet: Refactor all but image related functions in DockerInterface

For #23563.
Based on #23699 and #23844.

Only last 3 commits are new. This PR refactored all functions except image related functions, including:
* CreateExec
* StartExec
* InspectExec
* AttachToContainer
* Logs
* Info
* Version

@kubernetes/sig-node
This commit is contained in:
k8s-merge-robot 2016-04-21 20:57:38 -07:00
commit 06c2db4fe2
8 changed files with 179 additions and 286 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
pkg/kubelet/dockertools/docker.go
View File

@ -68,13 +68,13 @@ type DockerInterface interface {
ListImages(opts docker.ListImagesOptions) ([]docker.APIImages, error)
PullImage(opts docker.PullImageOptions, auth docker.AuthConfiguration) error
RemoveImage(image string) error
Logs(opts docker.LogsOptions) error
Version() (*docker.Env, error)
Info() (*docker.Env, error)
CreateExec(docker.CreateExecOptions) (*docker.Exec, error)
StartExec(string, docker.StartExecOptions) error
InspectExec(id string) (*docker.ExecInspect, error)
AttachToContainer(opts docker.AttachToContainerOptions) error
Logs(string, dockertypes.ContainerLogsOptions, StreamOptions) error
Version() (*dockertypes.Version, error)
Info() (*dockertypes.Info, error)
CreateExec(string, dockertypes.ExecConfig) (*dockertypes.ContainerExecCreateResponse, error)
StartExec(string, dockertypes.ExecStartCheck, StreamOptions) error
InspectExec(id string) (*dockertypes.ContainerExecInspect, error)
AttachToContainer(string, dockertypes.ContainerAttachOptions, StreamOptions) error
}
// KubeletContainerName encapsulates a pod name and a Kubernetes container name.

22
pkg/kubelet/dockertools/docker_test.go
View File

@ -157,28 +157,6 @@ func TestContainerNaming(t *testing.T) {
}
}
func TestVersion(t *testing.T) {
fakeDocker := NewFakeDockerClientWithVersion("1.1.3", "1.15")
manager := &DockerManager{client: fakeDocker}
version, err := manager.Version()
if err != nil {
t.Errorf("got error while getting docker server version - %s", err)
}
expectedVersion, _ := docker.NewAPIVersion("1.1.3")
if e, a := expectedVersion.String(), version.String(); e != a {
t.Errorf("invalid docker server version. expected: %v, got: %v", e, a)
}
version, err = manager.APIVersion()
if err != nil {
t.Errorf("got error while getting docker server version - %s", err)
}
expectedVersion, _ = docker.NewAPIVersion("1.15")
if e, a := expectedVersion.String(), version.String(); e != a {
t.Errorf("invalid docker server version. expected: %v, got: %v", e, a)
}
}
func TestParseImageName(t *testing.T) {
tests := []struct {
imageName string

13
pkg/kubelet/dockertools/exec.go
View File

@ -24,7 +24,6 @@ import (
"time"
dockertypes "github.com/docker/engine-api/types"
docker "github.com/fsouza/go-dockerclient"
"github.com/golang/glog"
kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
)
@ -100,27 +99,25 @@ func (*NsenterExecHandler) ExecInContainer(client DockerInterface, container *do
type NativeExecHandler struct{}
func (*NativeExecHandler) ExecInContainer(client DockerInterface, container *dockertypes.ContainerJSON, cmd []string, stdin io.Reader, stdout, stderr io.WriteCloser, tty bool) error {
createOpts := docker.CreateExecOptions{
Container: container.ID,
createOpts := dockertypes.ExecConfig{
Cmd: cmd,
AttachStdin: stdin != nil,
AttachStdout: stdout != nil,
AttachStderr: stderr != nil,
Tty: tty,
}
execObj, err := client.CreateExec(createOpts)
execObj, err := client.CreateExec(container.ID, createOpts)
if err != nil {
return fmt.Errorf("failed to exec in container - Exec setup failed - %v", err)
}
startOpts := docker.StartExecOptions{
Detach: false,
startOpts := dockertypes.ExecStartCheck{Detach: false, Tty: tty}
streamOpts := StreamOptions{
InputStream: stdin,
OutputStream: stdout,
ErrorStream: stderr,
Tty: tty,
RawTerminal: tty,
}
err = client.StartExec(execObj.ID, startOpts)
err = client.StartExec(execObj.ID, startOpts, streamOpts)
if err != nil {
return err
}

24
pkg/kubelet/dockertools/fake_docker_client.go
View File

@ -50,9 +50,9 @@ type FakeDockerClient struct {
Stopped []string
Removed []string
RemovedImages sets.String
VersionInfo docker.Env
Information docker.Env
ExecInspect *docker.ExecInspect
VersionInfo dockertypes.Version
Information dockertypes.Info
ExecInspect *dockertypes.ContainerExecInspect
execCmd []string
EnableSleep bool
}
@ -67,7 +67,7 @@ func NewFakeDockerClient() *FakeDockerClient {
func NewFakeDockerClientWithVersion(version, apiVersion string) *FakeDockerClient {
return &FakeDockerClient{
VersionInfo: docker.Env{fmt.Sprintf("Version=%s", version), fmt.Sprintf("ApiVersion=%s", apiVersion)},
VersionInfo: dockertypes.Version{Version: version, APIVersion: apiVersion},
Errors: make(map[string]error),
RemovedImages: sets.String{},
ContainerMap: make(map[string]*dockertypes.ContainerJSON),
@ -412,7 +412,7 @@ func (f *FakeDockerClient) RemoveContainer(id string, opts dockertypes.Container
// Logs is a test-spy implementation of DockerInterface.Logs.
// It adds an entry "logs" to the internal method call record.
func (f *FakeDockerClient) Logs(opts docker.LogsOptions) error {
func (f *FakeDockerClient) Logs(id string, opts dockertypes.ContainerLogsOptions, sopts StreamOptions) error {
f.Lock()
defer f.Unlock()
f.called = append(f.called, "logs")
@ -437,39 +437,39 @@ func (f *FakeDockerClient) PullImage(opts docker.PullImageOptions, auth docker.A
return err
}
func (f *FakeDockerClient) Version() (*docker.Env, error) {
func (f *FakeDockerClient) Version() (*dockertypes.Version, error) {
f.Lock()
defer f.Unlock()
return &f.VersionInfo, f.popError("version")
}
func (f *FakeDockerClient) Info() (*docker.Env, error) {
func (f *FakeDockerClient) Info() (*dockertypes.Info, error) {
return &f.Information, nil
}
func (f *FakeDockerClient) CreateExec(opts docker.CreateExecOptions) (*docker.Exec, error) {
func (f *FakeDockerClient) CreateExec(id string, opts dockertypes.ExecConfig) (*dockertypes.ContainerExecCreateResponse, error) {
f.Lock()
defer f.Unlock()
f.execCmd = opts.Cmd
f.called = append(f.called, "create_exec")
return &docker.Exec{ID: "12345678"}, nil
return &dockertypes.ContainerExecCreateResponse{ID: "12345678"}, nil
}
func (f *FakeDockerClient) StartExec(_ string, _ docker.StartExecOptions) error {
func (f *FakeDockerClient) StartExec(startExec string, opts dockertypes.ExecStartCheck, sopts StreamOptions) error {
f.Lock()
defer f.Unlock()
f.called = append(f.called, "start_exec")
return nil
}
func (f *FakeDockerClient) AttachToContainer(opts docker.AttachToContainerOptions) error {
func (f *FakeDockerClient) AttachToContainer(id string, opts dockertypes.ContainerAttachOptions, sopts StreamOptions) error {
f.Lock()
defer f.Unlock()
f.called = append(f.called, "attach")
return nil
}
func (f *FakeDockerClient) InspectExec(id string) (*docker.ExecInspect, error) {
func (f *FakeDockerClient) InspectExec(id string) (*dockertypes.ContainerExecInspect, error) {
return f.ExecInspect, f.popError("inspect_exec")
}

22
pkg/kubelet/dockertools/instrumented_docker.go
View File

@ -139,16 +139,16 @@ func (in instrumentedDockerInterface) RemoveImage(image string) error {
return err
}
func (in instrumentedDockerInterface) Logs(opts docker.LogsOptions) error {
func (in instrumentedDockerInterface) Logs(id string, opts dockertypes.ContainerLogsOptions, sopts StreamOptions) error {
const operation = "logs"
defer recordOperation(operation, time.Now())
err := in.client.Logs(opts)
err := in.client.Logs(id, opts, sopts)
recordError(operation, err)
return err
}
func (in instrumentedDockerInterface) Version() (*docker.Env, error) {
func (in instrumentedDockerInterface) Version() (*dockertypes.Version, error) {
const operation = "version"
defer recordOperation(operation, time.Now())
@ -157,7 +157,7 @@ func (in instrumentedDockerInterface) Version() (*docker.Env, error) {
return out, err
}
func (in instrumentedDockerInterface) Info() (*docker.Env, error) {
func (in instrumentedDockerInterface) Info() (*dockertypes.Info, error) {
const operation = "info"
defer recordOperation(operation, time.Now())
@ -166,25 +166,25 @@ func (in instrumentedDockerInterface) Info() (*docker.Env, error) {
return out, err
}
func (in instrumentedDockerInterface) CreateExec(opts docker.CreateExecOptions) (*docker.Exec, error) {
func (in instrumentedDockerInterface) CreateExec(id string, opts dockertypes.ExecConfig) (*dockertypes.ContainerExecCreateResponse, error) {
const operation = "create_exec"
defer recordOperation(operation, time.Now())
out, err := in.client.CreateExec(opts)
out, err := in.client.CreateExec(id, opts)
recordError(operation, err)
return out, err
}
func (in instrumentedDockerInterface) StartExec(startExec string, opts docker.StartExecOptions) error {
func (in instrumentedDockerInterface) StartExec(startExec string, opts dockertypes.ExecStartCheck, sopts StreamOptions) error {
const operation = "start_exec"
defer recordOperation(operation, time.Now())
err := in.client.StartExec(startExec, opts)
err := in.client.StartExec(startExec, opts, sopts)
recordError(operation, err)
return err
}
func (in instrumentedDockerInterface) InspectExec(id string) (*docker.ExecInspect, error) {
func (in instrumentedDockerInterface) InspectExec(id string) (*dockertypes.ContainerExecInspect, error) {
const operation = "inspect_exec"
defer recordOperation(operation, time.Now())
@ -193,11 +193,11 @@ func (in instrumentedDockerInterface) InspectExec(id string) (*docker.ExecInspec
return out, err
}
func (in instrumentedDockerInterface) AttachToContainer(opts docker.AttachToContainerOptions) error {
func (in instrumentedDockerInterface) AttachToContainer(id string, opts dockertypes.ContainerAttachOptions, sopts StreamOptions) error {
const operation = "attach"
defer recordOperation(operation, time.Now())
err := in.client.AttachToContainer(opts)
err := in.client.AttachToContainer(id, opts, sopts)
recordError(operation, err)
return err
}

100
pkg/kubelet/dockertools/kube_docker_client.go
View File

@ -23,7 +23,6 @@ import (
"fmt"
"io"
"io/ioutil"
"strconv"
"time"
dockermessage "github.com/docker/docker/pkg/jsonmessage"
@ -92,19 +91,6 @@ func convertFilters(filters map[string][]string) dockerfilters.Args {
return args
}
// convertEnv converts data to a go-dockerclient Env
func convertEnv(src interface{}) (*docker.Env, error) {
m := make(map[string]interface{})
if err := convertType(&src, &m); err != nil {
return nil, err
}
env := &docker.Env{}
for k, v := range m {
env.SetAuto(k, v)
}
return env, nil
}
func (k *kubeDockerClient) ListContainers(options dockertypes.ContainerListOptions) ([]dockertypes.Container, error) {
containers, err := k.client.ContainerList(getDefaultContext(), options)
if err != nil {
@ -232,61 +218,45 @@ func (d *kubeDockerClient) RemoveImage(image string) error {
return err
}
func (d *kubeDockerClient) Logs(opts docker.LogsOptions) error {
resp, err := d.client.ContainerLogs(getDefaultContext(), dockertypes.ContainerLogsOptions{
ContainerID: opts.Container,
ShowStdout: opts.Stdout,
ShowStderr: opts.Stderr,
Since: strconv.FormatInt(opts.Since, 10),
Timestamps: opts.Timestamps,
Follow: opts.Follow,
Tail: opts.Tail,
})
func (d *kubeDockerClient) Logs(id string, opts dockertypes.ContainerLogsOptions, sopts StreamOptions) error {
opts.ContainerID = id
resp, err := d.client.ContainerLogs(getDefaultContext(), opts)
if err != nil {
return err
}
defer resp.Close()
return d.redirectResponseToOutputStream(opts.RawTerminal, opts.OutputStream, opts.ErrorStream, resp)
return d.redirectResponseToOutputStream(sopts.RawTerminal, sopts.OutputStream, sopts.ErrorStream, resp)
}
func (d *kubeDockerClient) Version() (*docker.Env, error) {
func (d *kubeDockerClient) Version() (*dockertypes.Version, error) {
resp, err := d.client.ServerVersion(getDefaultContext())
if err != nil {
return nil, err
}
return convertEnv(resp)
return &resp, nil
}
func (d *kubeDockerClient) Info() (*docker.Env, error) {
func (d *kubeDockerClient) Info() (*dockertypes.Info, error) {
resp, err := d.client.Info(getDefaultContext())
if err != nil {
return nil, err
}
return convertEnv(resp)
return &resp, nil
}
func (d *kubeDockerClient) CreateExec(opts docker.CreateExecOptions) (*docker.Exec, error) {
cfg := dockertypes.ExecConfig{}
if err := convertType(&opts, &cfg); err != nil {
return nil, err
}
resp, err := d.client.ContainerExecCreate(getDefaultContext(), cfg)
// TODO(random-liu): Add unit test for exec and attach functions, just like what go-dockerclient did.
func (d *kubeDockerClient) CreateExec(id string, opts dockertypes.ExecConfig) (*dockertypes.ContainerExecCreateResponse, error) {
opts.Container = id
resp, err := d.client.ContainerExecCreate(getDefaultContext(), opts)
if err != nil {
return nil, err
}
exec := &docker.Exec{}
if err := convertType(&resp, exec); err != nil {
return nil, err
}
return exec, nil
return &resp, nil
}
func (d *kubeDockerClient) StartExec(startExec string, opts docker.StartExecOptions) error {
func (d *kubeDockerClient) StartExec(startExec string, opts dockertypes.ExecStartCheck, sopts StreamOptions) error {
if opts.Detach {
return d.client.ContainerExecStart(getDefaultContext(), startExec, dockertypes.ExecStartCheck{
Detach: opts.Detach,
Tty: opts.Tty,
})
return d.client.ContainerExecStart(getDefaultContext(), startExec, opts)
}
resp, err := d.client.ContainerExecAttach(getDefaultContext(), startExec, dockertypes.ExecConfig{
Detach: opts.Detach,
@ -296,43 +266,25 @@ func (d *kubeDockerClient) StartExec(startExec string, opts docker.StartExecOpti
return err
}
defer resp.Close()
if opts.Success != nil {
opts.Success <- struct{}{}
<-opts.Success
}
return d.holdHijackedConnection(opts.RawTerminal || opts.Tty, opts.InputStream, opts.OutputStream, opts.ErrorStream, resp)
return d.holdHijackedConnection(sopts.RawTerminal || opts.Tty, sopts.InputStream, sopts.OutputStream, sopts.ErrorStream, resp)
}
func (d *kubeDockerClient) InspectExec(id string) (*docker.ExecInspect, error) {
func (d *kubeDockerClient) InspectExec(id string) (*dockertypes.ContainerExecInspect, error) {
resp, err := d.client.ContainerExecInspect(getDefaultContext(), id)
if err != nil {
return nil, err
}
exec := &docker.ExecInspect{}
if err := convertType(&resp, exec); err != nil {
return nil, err
}
return exec, nil
return &resp, nil
}
func (d *kubeDockerClient) AttachToContainer(opts docker.AttachToContainerOptions) error {
resp, err := d.client.ContainerAttach(getDefaultContext(), dockertypes.ContainerAttachOptions{
ContainerID: opts.Container,
Stream: opts.Stream,
Stdin: opts.Stdin,
Stdout: opts.Stdout,
Stderr: opts.Stderr,
// TODO: How to deal with the *Logs* here? There is no *Logs* field in the engine-api.
})
func (d *kubeDockerClient) AttachToContainer(id string, opts dockertypes.ContainerAttachOptions, sopts StreamOptions) error {
opts.ContainerID = id
resp, err := d.client.ContainerAttach(getDefaultContext(), opts)
if err != nil {
return err
}
defer resp.Close()
if opts.Success != nil {
opts.Success <- struct{}{}
<-opts.Success
}
return d.holdHijackedConnection(opts.RawTerminal, opts.InputStream, opts.OutputStream, opts.ErrorStream, resp)
return d.holdHijackedConnection(sopts.RawTerminal, sopts.InputStream, sopts.OutputStream, sopts.ErrorStream, resp)
}
// redirectResponseToOutputStream redirect the response stream to stdout and stderr. When tty is true, all stream will
@ -389,6 +341,14 @@ func parseDockerTimestamp(s string) (time.Time, error) {
return time.Parse(time.RFC3339Nano, s)
}
// StreamOptions are the options used to configure the stream redirection
type StreamOptions struct {
RawTerminal bool
InputStream io.Reader
OutputStream io.Writer
ErrorStream io.Writer
}
// containerNotFoundError is the error returned by InspectContainer when container not found. We
// add this error type for testability. We don't use the original error returned by engine-api
// because dockertypes.containerNotFoundError is private, we can't create and inject it in our test.

173
pkg/kubelet/dockertools/manager.go
View File

@ -25,13 +25,11 @@ import (
"os"
"os/exec"
"path"
"regexp"
"strconv"
"strings"
"sync"
"time"
"github.com/coreos/go-semver/semver"
dockertypes "github.com/docker/engine-api/types"
dockercontainer "github.com/docker/engine-api/types/container"
dockerstrslice "github.com/docker/engine-api/types/strslice"
@ -220,7 +218,7 @@ func NewDockerManager(
glog.Errorf("Failed to execute Info() call to the Docker client: %v", err)
glog.Warningf("Using fallback default of /var/lib/docker for location of Docker runtime")
} else {
dockerRoot = dockerInfo.Get("DockerRootDir")
dockerRoot = dockerInfo.DockerRootDir
glog.Infof("Setting dockerRoot to %s", dockerRoot)
}
@ -274,23 +272,22 @@ func (dm *DockerManager) GetContainerLogs(pod *api.Pod, containerID kubecontaine
if logOptions.SinceTime != nil {
since = logOptions.SinceTime.Unix()
}
opts := docker.LogsOptions{
Container: containerID.ID,
Stdout: true,
Stderr: true,
OutputStream: stdout,
ErrorStream: stderr,
Timestamps: logOptions.Timestamps,
Since: since,
Follow: logOptions.Follow,
RawTerminal: false,
opts := dockertypes.ContainerLogsOptions{
ShowStdout: true,
ShowStderr: true,
Since: strconv.FormatInt(since, 10),
Timestamps: logOptions.Timestamps,
Follow: logOptions.Follow,
}
if logOptions.TailLines != nil {
opts.Tail = strconv.FormatInt(*logOptions.TailLines, 10)
}
err = dm.client.Logs(opts)
sopts := StreamOptions{
OutputStream: stdout,
ErrorStream: stderr,
RawTerminal: false,
}
err = dm.client.Logs(containerID.ID, opts, sopts)
return
}
@ -874,67 +871,48 @@ func getDockerNetworkMode(container *dockertypes.ContainerJSON) string {
}
// dockerVersion implementes kubecontainer.Version interface by implementing
// Compare() and String() (which is implemented by the underlying semver.Version)
// TODO: this code is the same as rktVersion and may make sense to be moved to
// somewhere shared.
type dockerVersion struct {
*semver.Version
// Compare() and String(). It could contain either server version or api version.
type dockerVersion string
func (v dockerVersion) String() string {
return string(v)
}
// Older versions of Docker could return non-semantically versioned values (distros like Fedora
// included partial values such as 1.8.1.fc21 which is not semver). Force those values to be semver.
var almostSemverRegexp = regexp.MustCompile(`^(\d+\.\d+\.\d+)\.(.*)$`)
func (v dockerVersion) Compare(other string) (int, error) {
return compare(string(v), other), nil
}
// newDockerVersion returns a semantically versioned docker version value
func newDockerVersion(version string) (dockerVersion, error) {
sem, err := semver.NewVersion(version)
if err != nil {
matches := almostSemverRegexp.FindStringSubmatch(version)
if matches == nil {
return dockerVersion{}, err
// compare is copied from engine-api, it compares two version strings, returns -1 if
// v1 < v2, 1 if v1 > v2, 0 otherwise.
// TODO(random-liu): Leveraging the version comparison in engine-api after bumping up
// the engine-api version. See #24076
func compare(v1, v2 string) int {
var (
currTab = strings.Split(v1, ".")
otherTab = strings.Split(v2, ".")
)
max := len(currTab)
if len(otherTab) > max {
max = len(otherTab)
}
for i := 0; i < max; i++ {
var currInt, otherInt int
if len(currTab) > i {
currInt, _ = strconv.Atoi(currTab[i])
}
if len(otherTab) > i {
otherInt, _ = strconv.Atoi(otherTab[i])
}
if currInt > otherInt {
return 1
}
if otherInt > currInt {
return -1
}
sem, err = semver.NewVersion(strings.Join(matches[1:], "-"))
}
return dockerVersion{sem}, err
}
func (r dockerVersion) Compare(other string) (int, error) {
v, err := newDockerVersion(other)
if err != nil {
return -1, err
}
if r.LessThan(*v.Version) {
return -1, nil
}
if v.Version.LessThan(*r.Version) {
return 1, nil
}
return 0, nil
}
// dockerVersion implementes kubecontainer.Version interface by implementing
// Compare() and String() on top og go-dockerclient's APIVersion. This version
// string doesn't conform to semantic versioning, as it is only "x.y"
type dockerAPIVersion docker.APIVersion
func (dv dockerAPIVersion) String() string {
return docker.APIVersion(dv).String()
}
func (dv dockerAPIVersion) Compare(other string) (int, error) {
a := docker.APIVersion(dv)
b, err := docker.NewAPIVersion(other)
if err != nil {
return 0, err
}
if a.LessThan(b) {
return -1, nil
}
if a.GreaterThan(b) {
return 1, nil
}
return 0, nil
return 0
}
func (dm *DockerManager) Type() string {
@ -942,33 +920,21 @@ func (dm *DockerManager) Type() string {
}
func (dm *DockerManager) Version() (kubecontainer.Version, error) {
env, err := dm.client.Version()
v, err := dm.client.Version()
if err != nil {
return nil, fmt.Errorf("docker: failed to get docker version: %v", err)
}
engineVersion := env.Get("Version")
version, err := newDockerVersion(engineVersion)
if err != nil {
glog.Errorf("docker: failed to parse docker server version %q: %v", engineVersion, err)
return nil, fmt.Errorf("docker: failed to parse docker server version %q: %v", engineVersion, err)
}
return version, nil
return dockerVersion(v.Version), nil
}
func (dm *DockerManager) APIVersion() (kubecontainer.Version, error) {
env, err := dm.client.Version()
v, err := dm.client.Version()
if err != nil {
return nil, fmt.Errorf("docker: failed to get docker version: %v", err)
}
apiVersion := env.Get("ApiVersion")
version, err := docker.NewAPIVersion(apiVersion)
if err != nil {
glog.Errorf("docker: failed to parse docker api version %q: %v", apiVersion, err)
return nil, fmt.Errorf("docker: failed to parse docker api version %q: %v", apiVersion, err)
}
return dockerAPIVersion(version), nil
return dockerVersion(v.APIVersion), nil
}
// Status returns error if docker daemon is unhealthy, nil otherwise.
@ -1014,27 +980,25 @@ func (dm *DockerManager) defaultSecurityOpt() ([]string, error) {
// RunInContainer run the command inside the container identified by containerID
func (dm *DockerManager) RunInContainer(containerID kubecontainer.ContainerID, cmd []string) ([]byte, error) {
glog.V(2).Infof("Using docker native exec to run cmd %+v inside container %s", cmd, containerID)
createOpts := docker.CreateExecOptions{
Container: containerID.ID,
createOpts := dockertypes.ExecConfig{
Cmd: cmd,
AttachStdin: false,
AttachStdout: true,
AttachStderr: true,
Tty: false,
}
execObj, err := dm.client.CreateExec(createOpts)
execObj, err := dm.client.CreateExec(containerID.ID, createOpts)
if err != nil {
return nil, fmt.Errorf("failed to run in container - Exec setup failed - %v", err)
}
var buf bytes.Buffer
startOpts := docker.StartExecOptions{
Detach: false,
Tty: false,
startOpts := dockertypes.ExecStartCheck{Detach: false, Tty: false}
streamOpts := StreamOptions{
OutputStream: &buf,
ErrorStream: &buf,
RawTerminal: false,
}
err = dm.client.StartExec(execObj.ID, startOpts)
err = dm.client.StartExec(execObj.ID, startOpts, streamOpts)
if err != nil {
glog.V(2).Infof("StartExec With error: %v", err)
return nil, err
@ -1061,7 +1025,7 @@ func (dm *DockerManager) RunInContainer(containerID kubecontainer.ContainerID, c
}
type dockerExitError struct {
Inspect *docker.ExecInspect
Inspect *dockertypes.ContainerExecInspect
}
func (d *dockerExitError) String() string {
@ -1098,19 +1062,20 @@ func (dm *DockerManager) ExecInContainer(containerID kubecontainer.ContainerID,
}
func (dm *DockerManager) AttachContainer(containerID kubecontainer.ContainerID, stdin io.Reader, stdout, stderr io.WriteCloser, tty bool) error {
opts := docker.AttachToContainerOptions{
Container: containerID.ID,
// TODO(random-liu): Do we really use the *Logs* field here?
opts := dockertypes.ContainerAttachOptions{
Stream: true,
Stdin: stdin != nil,
Stdout: stdout != nil,
Stderr: stderr != nil,
}
sopts := StreamOptions{
InputStream: stdin,
OutputStream: stdout,
ErrorStream: stderr,
Stream: true,
Logs: true,
Stdin: stdin != nil,
Stdout: stdout != nil,
Stderr: stderr != nil,
RawTerminal: tty,
}
return dm.client.AttachToContainer(opts)
return dm.client.AttachToContainer(containerID.ID, opts, sopts)
}
func noPodInfraContainerError(podName, podNamespace string) error {

97
pkg/kubelet/dockertools/manager_test.go
View File

@ -93,8 +93,13 @@ func (f *fakeRuntimeHelper) GeneratePodHostNameAndDomain(pod *api.Pod) (string,
return "", ""
}
func newTestDockerManagerWithHTTPClientWithVersion(fakeHTTPClient *fakeHTTP, version, apiVersion string) (*DockerManager, *FakeDockerClient) {
fakeDocker := NewFakeDockerClientWithVersion(version, apiVersion)
func createTestDockerManager(fakeHTTPClient *fakeHTTP, fakeDocker *FakeDockerClient) (*DockerManager, *FakeDockerClient) {
if fakeHTTPClient == nil {
fakeHTTPClient = &fakeHTTP{}
}
if fakeDocker == nil {
fakeDocker = NewFakeDockerClient()
}
fakeRecorder := &record.FakeRecorder{}
containerRefManager := kubecontainer.NewRefManager()
networkPlugin, _ := network.InitNetworkPlugin([]network.NetworkPlugin{}, "", nettest.NewFakeHost(nil))
@ -116,11 +121,16 @@ func newTestDockerManagerWithHTTPClientWithVersion(fakeHTTPClient *fakeHTTP, ver
}
func newTestDockerManagerWithHTTPClient(fakeHTTPClient *fakeHTTP) (*DockerManager, *FakeDockerClient) {
return newTestDockerManagerWithHTTPClientWithVersion(fakeHTTPClient, "1.8.1", "1.20")
return createTestDockerManager(fakeHTTPClient, nil)
}
func newTestDockerManagerWithVersion(version, apiVersion string) (*DockerManager, *FakeDockerClient) {
fakeDocker := NewFakeDockerClientWithVersion(version, apiVersion)
return createTestDockerManager(nil, fakeDocker)
}
func newTestDockerManager() (*DockerManager, *FakeDockerClient) {
return newTestDockerManagerWithHTTPClient(&fakeHTTP{})
return createTestDockerManager(nil, nil)
}
func matchString(t *testing.T, pattern, str string) bool {
@ -131,35 +141,6 @@ func matchString(t *testing.T, pattern, str string) bool {
return match
}
func TestNewDockerVersion(t *testing.T) {
cases := []struct {
value string
out string
err bool
}{
{value: "1", err: true},
{value: "1.8", err: true},
{value: "1.8.1", out: "1.8.1"},
{value: "1.8.1.fc21", out: "1.8.1-fc21"},
{value: "1.8.1.fc21.other", out: "1.8.1-fc21.other"},
{value: "1.8.1-fc21.other", out: "1.8.1-fc21.other"},
{value: "1.8.1-beta.12", out: "1.8.1-beta.12"},
}
for _, test := range cases {
v, err := newDockerVersion(test.value)
switch {
case err != nil && test.err:
continue
case (err != nil) != test.err:
t.Errorf("error for %q: expected %t, got %v", test.value, test.err, err)
continue
}
if v.String() != test.out {
t.Errorf("unexpected parsed version %q for %q", v, test.value)
}
}
}
func TestSetEntrypointAndCommand(t *testing.T) {
cases := []struct {
name string
@ -438,7 +419,7 @@ func TestKillContainerInPod(t *testing.T) {
func TestKillContainerInPodWithPreStop(t *testing.T) {
manager, fakeDocker := newTestDockerManager()
fakeDocker.ExecInspect = &docker.ExecInspect{
fakeDocker.ExecInspect = &dockertypes.ContainerExecInspect{
Running: false,
ExitCode: 0,
}
@ -1730,7 +1711,7 @@ func verifySyncResults(t *testing.T, expectedResults []*kubecontainer.SyncResult
}
func TestSeccompIsDisabledWithDockerV110(t *testing.T) {
dm, fakeDocker := newTestDockerManagerWithHTTPClientWithVersion(&fakeHTTP{}, "1.10.1", "1.22")
dm, fakeDocker := newTestDockerManagerWithVersion("1.10.1", "1.22")
pod := &api.Pod{
ObjectMeta: api.ObjectMeta{
UID: "12345678",
@ -1769,7 +1750,7 @@ func TestSeccompIsDisabledWithDockerV110(t *testing.T) {
}
func TestSecurityOptsAreNilWithDockerV19(t *testing.T) {
dm, fakeDocker := newTestDockerManagerWithHTTPClientWithVersion(&fakeHTTP{}, "1.9.1", "1.21")
dm, fakeDocker := newTestDockerManagerWithVersion("1.9.1", "1.21")
pod := &api.Pod{
ObjectMeta: api.ObjectMeta{
UID: "12345678",
@ -1808,10 +1789,6 @@ func TestSecurityOptsAreNilWithDockerV19(t *testing.T) {
}
func TestCheckVersionCompatibility(t *testing.T) {
apiVersion, err := docker.NewAPIVersion(minimumDockerAPIVersion)
if err != nil {
t.Fatalf("unexpected error %v", err)
}
type test struct {
version string
compatible bool
@ -1821,22 +1798,17 @@ func TestCheckVersionCompatibility(t *testing.T) {
{minimumDockerAPIVersion, true},
// Invalid apiversion
{"invalid_api_version", false},
}
for i := range apiVersion {
apiVersion[i]++
// Newer apiversion
tests = append(tests, test{apiVersion.String(), true})
apiVersion[i] -= 2
// Older apiversion
if apiVersion[i] >= 0 {
tests = append(tests, test{apiVersion.String(), false})
}
apiVersion[i]++
{"1.0.0", false},
// Newer apiversion
// NOTE(random-liu): We need to bump up the newer apiversion,
// if docker apiversion really reaches "9.9.9" someday. But I
// really doubt whether the test could live that long.
{"9.9.9", true},
}
for i, tt := range tests {
testCase := fmt.Sprintf("test case #%d test version %q", i, tt.version)
dm, fakeDocker := newTestDockerManagerWithHTTPClientWithVersion(&fakeHTTP{}, "", tt.version)
dm, fakeDocker := newTestDockerManagerWithVersion("", tt.version)
err := dm.checkVersionCompatibility()
assert.Equal(t, tt.compatible, err == nil, testCase)
if tt.compatible == true {
@ -1848,6 +1820,27 @@ func TestCheckVersionCompatibility(t *testing.T) {
}
}
func TestVersion(t *testing.T) {
expectedVersion := "1.8.1"
expectedAPIVersion := "1.20"
dm, _ := newTestDockerManagerWithVersion(expectedVersion, expectedAPIVersion)
version, err := dm.Version()
if err != nil {
t.Errorf("got error while getting docker server version - %v", err)
}
if e, a := expectedVersion, version.String(); e != a {
t.Errorf("expect docker server version %q, got %q", e, a)
}
apiVersion, err := dm.APIVersion()
if err != nil {
t.Errorf("got error while getting docker api version - %v", err)
}
if e, a := expectedAPIVersion, apiVersion.String(); e != a {
t.Errorf("expect docker api version %q, got %q", e, a)
}
}
func TestGetPodStatusNoSuchContainer(t *testing.T) {
const (
noSuchContainerID = "nosuchcontainer"