Skip to content

K
k3s
Commit f5d9c430 authored by Kubernetes Submit Queue's avatar Kubernetes Submit Queue Committed by GitHub
Browse files
Options

Merge pull request #39298 from feiskyer/images

Automatic merge from submit-queue Kubelet: add image ref to ImageService interfaces This PR adds image ref (digest or ID, depending on runtime) to PullImage result, and pass image ref in CreateContainer instead of image name. It also * Adds image ref to CRI's PullImageResponse * Updates related image puller * Updates related testing utilities ~~One remaining issue is: it breaks some e2e tests because they [checks image repoTags](https://github.com/kubernetes/kubernetes/blob/master/test/e2e/framework/util.go#L1941) while docker always returns digest in this PR. Should we update e2e test or continue to return repoTags in `containerStatuses.image`?~~ Fixes #38833.
parents c5e04596 67a5bf84
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 223 additions and 138 deletions
pkg/kubelet/api/services.go
View file @ f5d9c430
......@@ -94,7 +94,7 @@ type ImageManagerService interface {
// ImageStatus returns the status of the image.
ImageStatus(image *runtimeapi.ImageSpec) (*runtimeapi.Image, error)
// PullImage pulls an image with the authentication config.
PullImage(image *runtimeapi.ImageSpec, auth *runtimeapi.AuthConfig) error
PullImage(image *runtimeapi.ImageSpec, auth *runtimeapi.AuthConfig) (string, error)
// RemoveImage removes the image.
RemoveImage(image *runtimeapi.ImageSpec) error
}
pkg/kubelet/api/testing/fake_image_service.go
View file @ f5d9c430
......@@ -91,7 +91,7 @@ func (r *FakeImageService) ImageStatus(image *runtimeapi.ImageSpec) (*runtimeapi
return r.Images[image.GetImage()], nil
}
func (r *FakeImageService) PullImage(image *runtimeapi.ImageSpec, auth *runtimeapi.AuthConfig) error {
func (r *FakeImageService) PullImage(image *runtimeapi.ImageSpec, auth *runtimeapi.AuthConfig) (string, error) {
r.Lock()
defer r.Unlock()
......@@ -104,7 +104,7 @@ func (r *FakeImageService) PullImage(image *runtimeapi.ImageSpec, auth *runtimea
r.Images[imageID] = r.makeFakeImage(image.GetImage())
}
return nil
return imageID, nil
}
func (r *FakeImageService) RemoveImage(image *runtimeapi.ImageSpec) error {
......
pkg/kubelet/api/v1alpha1/runtime/api.proto
View file @ f5d9c430
......@@ -421,9 +421,8 @@ message ListPodSandboxResponse {
}
// ImageSpec is an internal representation of an image. Currently, it wraps the
// value of a Container's Image field (e.g. imageName, imageName:tag, or
// imageName:digest), but in the future it will include more detailed
// information about the different image types.
// value of a Container's Image field (e.g. imageID or imageDigest), but in the
// future it will include more detailed information about the different image types.
message ImageSpec {
optional string image = 1;
}
......@@ -878,7 +877,11 @@ message PullImageRequest {
optional PodSandboxConfig sandbox_config = 3;
}
message PullImageResponse {}
message PullImageResponse {
// Reference to the image in use. For most runtimes, this should be an
// image ID or digest.
optional string image_ref = 1;
}
message RemoveImageRequest {
// Spec of the image to remove.
......
pkg/kubelet/container/runtime.go
View file @ f5d9c430
......@@ -146,10 +146,11 @@ type IndirectStreamingRuntime interface {
type ImageService interface {
// PullImage pulls an image from the network to local storage using the supplied
// secrets if necessary.
PullImage(image ImageSpec, pullSecrets []v1.Secret) error
// IsImagePresent checks whether the container image is already in the local storage.
IsImagePresent(image ImageSpec) (bool, error)
// secrets if necessary. It returns a reference (digest or ID) to the pulled image.
PullImage(image ImageSpec, pullSecrets []v1.Secret) (string, error)
// GetImageRef gets the reference (digest or ID) of the image which has already been in
// the local storage. It returns ("", nil) if the image isn't in the local storage.
GetImageRef(image ImageSpec) (string, error)
// Gets all images currently on the machine.
ListImages() ([]Image, error)
// Removes the specified image.
......
pkg/kubelet/container/testing/fake_runtime.go
View file @ f5d9c430
......@@ -348,25 +348,25 @@ func (f *FakeRuntime) GetContainerLogs(pod *v1.Pod, containerID ContainerID, log
return f.Err
}
func (f *FakeRuntime) PullImage(image ImageSpec, pullSecrets []v1.Secret) error {
func (f *FakeRuntime) PullImage(image ImageSpec, pullSecrets []v1.Secret) (string, error) {
f.Lock()
defer f.Unlock()
f.CalledFunctions = append(f.CalledFunctions, "PullImage")
return f.Err
return image.Image, f.Err
}
func (f *FakeRuntime) IsImagePresent(image ImageSpec) (bool, error) {
func (f *FakeRuntime) GetImageRef(image ImageSpec) (string, error) {
f.Lock()
defer f.Unlock()
f.CalledFunctions = append(f.CalledFunctions, "IsImagePresent")
f.CalledFunctions = append(f.CalledFunctions, "GetImageRef")
for _, i := range f.ImageList {
if i.ID == image.Image {
return true, nil
return i.ID, nil
}
}
return false, f.InspectErr
return "", f.InspectErr
}
func (f *FakeRuntime) ListImages() ([]Image, error) {
......
pkg/kubelet/container/testing/runtime_mock.go
View file @ f5d9c430
......@@ -105,14 +105,14 @@ func (r *Mock) GetContainerLogs(pod *v1.Pod, containerID ContainerID, logOptions
return args.Error(0)
}
func (r *Mock) PullImage(image ImageSpec, pullSecrets []v1.Secret) error {
func (r *Mock) PullImage(image ImageSpec, pullSecrets []v1.Secret) (string, error) {
args := r.Called(image, pullSecrets)
return args.Error(0)
return image.Image, args.Error(0)
}
func (r *Mock) IsImagePresent(image ImageSpec) (bool, error) {
func (r *Mock) GetImageRef(image ImageSpec) (string, error) {
args := r.Called(image)
return args.Get(0).(bool), args.Error(1)
return args.Get(0).(string), args.Error(1)
}
func (r *Mock) ListImages() ([]Image, error) {
......
pkg/kubelet/dockershim/docker_container.go
View file @ f5d9c430
......@@ -364,10 +364,14 @@ func (ds *dockerService) ContainerStatus(containerID string) (*runtimeapi.Contai
}
labels, annotations := extractLabels(r.Config.Labels)
imageName := r.Config.Image
if len(ir.RepoTags) > 0 {
imageName = ir.RepoTags[0]
}
return &runtimeapi.ContainerStatus{
Id: &r.ID,
Metadata: metadata,
Image: &runtimeapi.ImageSpec{Image: &r.Config.Image},
Image: &runtimeapi.ImageSpec{Image: &imageName},
ImageRef: &imageID,
Mounts: mounts,
ExitCode: &exitCode,
......
pkg/kubelet/dockershim/docker_image.go
View file @ f5d9c430
......@@ -63,8 +63,8 @@ func (ds *dockerService) ImageStatus(image *runtimeapi.ImageSpec) (*runtimeapi.I
}
// PullImage pulls an image with authentication config.
func (ds *dockerService) PullImage(image *runtimeapi.ImageSpec, auth *runtimeapi.AuthConfig) error {
return ds.client.PullImage(image.GetImage(),
func (ds *dockerService) PullImage(image *runtimeapi.ImageSpec, auth *runtimeapi.AuthConfig) (string, error) {
err := ds.client.PullImage(image.GetImage(),
dockertypes.AuthConfig{
Username: auth.GetUsername(),
Password: auth.GetPassword(),
......@@ -74,6 +74,11 @@ func (ds *dockerService) PullImage(image *runtimeapi.ImageSpec, auth *runtimeapi
},
dockertypes.ImagePullOptions{},
)
if err != nil {
return "", err
}
return dockertools.GetImageRef(ds.client, image.GetImage())
}
// RemoveImage removes the image.
......
pkg/kubelet/dockershim/remote/docker_service.go
View file @ f5d9c430
......@@ -200,11 +200,11 @@ func (d *dockerService) ImageStatus(ctx context.Context, r *runtimeapi.ImageStat
}
func (d *dockerService) PullImage(ctx context.Context, r *runtimeapi.PullImageRequest) (*runtimeapi.PullImageResponse, error) {
err := d.imageService.PullImage(r.GetImage(), r.GetAuth())
image, err := d.imageService.PullImage(r.GetImage(), r.GetAuth())
if err != nil {
return nil, err
}
return &runtimeapi.PullImageResponse{}, nil
return &runtimeapi.PullImageResponse{ImageRef: &image}, nil
}
func (d *dockerService) RemoveImage(ctx context.Context, r *runtimeapi.RemoveImageRequest) (*runtimeapi.RemoveImageResponse, error) {
......
pkg/kubelet/dockertools/docker.go
View file @ f5d9c430
......@@ -94,7 +94,7 @@ func SetContainerNamePrefix(prefix string) {
// DockerPuller is an abstract interface for testability. It abstracts image pull operations.
type DockerPuller interface {
Pull(image string, secrets []v1.Secret) error
IsImagePresent(image string) (bool, error)
GetImageRef(image string) (string, error)
}
// dockerPuller is the default implementation of DockerPuller.
......@@ -241,11 +241,11 @@ func (p dockerPuller) Pull(image string, secrets []v1.Secret) error {
err := p.client.PullImage(image, dockertypes.AuthConfig{}, opts)
if err == nil {
// Sometimes PullImage failed with no error returned.
exist, ierr := p.IsImagePresent(image)
imageRef, ierr := p.GetImageRef(image)
if ierr != nil {
glog.Warningf("Failed to inspect image %s: %v", image, ierr)
}
if !exist {
if imageRef == "" {
return fmt.Errorf("image pull failed for unknown error")
}
return nil
......@@ -277,15 +277,23 @@ func (p dockerPuller) Pull(image string, secrets []v1.Secret) error {
return utilerrors.NewAggregate(pullErrs)
}
func (p dockerPuller) IsImagePresent(image string) (bool, error) {
_, err := p.client.InspectImageByRef(image)
func (p dockerPuller) GetImageRef(image string) (string, error) {
resp, err := p.client.InspectImageByRef(image)
if err == nil {
return true, nil
if resp == nil {
return "", nil
}
imageRef := resp.ID
if len(resp.RepoDigests) > 0 {
imageRef = resp.RepoDigests[0]
}
return imageRef, nil
}
if _, ok := err.(imageNotFoundError); ok {
return false, nil
return "", nil
}
return false, err
return "", err
}
// Creates a name which can be reversed to identify both full pod name and container name.
......
pkg/kubelet/dockertools/docker_manager.go
View file @ f5d9c430
......@@ -430,10 +430,14 @@ func (dm *DockerManager) inspectContainer(id string, podName, podNamespace strin
}
}
imageName := iResult.Config.Image
if len(imgInspectResult.RepoTags) > 0 {
imageName = imgInspectResult.RepoTags[0]
}
status := kubecontainer.ContainerStatus{
Name: containerName,
RestartCount: containerInfo.RestartCount,
Image: iResult.Config.Image,
Image: imageName,
ImageID: imageID,
ID: kubecontainer.DockerID(id).ContainerID(),
ExitCode: iResult.State.ExitCode,
......@@ -590,6 +594,7 @@ func (dm *DockerManager) runContainer(
container *v1.Container,
opts *kubecontainer.RunContainerOptions,
ref *v1.ObjectReference,
imageRef string,
netMode string,
ipcMode string,
utsMode string,
......@@ -765,7 +770,7 @@ func (dm *DockerManager) runContainer(
Name: containerName,
Config: &dockercontainer.Config{
Env: makeEnvList(opts.Envs),
Image: container.Image,
Image: imageRef,
WorkingDir: container.WorkingDir,
Labels: labels,
// Interactive containers:
......@@ -958,14 +963,39 @@ func (dm *DockerManager) ListImages() ([]kubecontainer.Image, error) {
return images, nil
}
// GetImageRef returns the image digest if exists, or else returns the image ID.
// It is exported for reusing in dockershim.
func GetImageRef(client DockerInterface, image string) (string, error) {
img, err := client.InspectImageByRef(image)
if err != nil {
return "", err
}
if img == nil {
return "", fmt.Errorf("unable to inspect image %s", image)
}
// Returns the digest if it exist.
if len(img.RepoDigests) > 0 {
return img.RepoDigests[0], nil
}
return img.ID, nil
}
// PullImage pulls an image from network to local storage.
func (dm *DockerManager) PullImage(image kubecontainer.ImageSpec, secrets []v1.Secret) error {
return dm.dockerPuller.Pull(image.Image, secrets)
func (dm *DockerManager) PullImage(image kubecontainer.ImageSpec, secrets []v1.Secret) (string, error) {
err := dm.dockerPuller.Pull(image.Image, secrets)
if err != nil {
return "", err
}
return GetImageRef(dm.client, image.Image)
}
// IsImagePresent checks whether the container image is already in the local storage.
func (dm *DockerManager) IsImagePresent(image kubecontainer.ImageSpec) (bool, error) {
return dm.dockerPuller.IsImagePresent(image.Image)
// GetImageRef gets the reference (digest or ID) of the image which has already been in
// the local storage. It returns ("", nil) if the image isn't in the local storage.
func (dm *DockerManager) GetImageRef(image kubecontainer.ImageSpec) (string, error) {
return dm.dockerPuller.GetImageRef(image.Image)
}
// Removes the specified image.
......@@ -1683,7 +1713,7 @@ func (dm *DockerManager) applyOOMScoreAdj(pod *v1.Pod, container *v1.Container,
// Run a single container from a pod. Returns the docker container ID
// If do not need to pass labels, just pass nil.
func (dm *DockerManager) runContainerInPod(pod *v1.Pod, container *v1.Container, netMode, ipcMode, pidMode, podIP string, restartCount int) (kubecontainer.ContainerID, error) {
func (dm *DockerManager) runContainerInPod(pod *v1.Pod, container *v1.Container, netMode, ipcMode, pidMode, podIP, imageRef string, restartCount int) (kubecontainer.ContainerID, error) {
start := time.Now()
defer func() {
metrics.ContainerManagerLatency.WithLabelValues("runContainerInPod").Observe(metrics.SinceInMicroseconds(start))
......@@ -1708,7 +1738,7 @@ func (dm *DockerManager) runContainerInPod(pod *v1.Pod, container *v1.Container,
oomScoreAdj := dm.calculateOomScoreAdj(pod, container)
id, err := dm.runContainer(pod, container, opts, ref, netMode, ipcMode, utsMode, pidMode, restartCount, oomScoreAdj)
id, err := dm.runContainer(pod, container, opts, ref, imageRef, netMode, ipcMode, utsMode, pidMode, restartCount, oomScoreAdj)
if err != nil {
return kubecontainer.ContainerID{}, fmt.Errorf("runContainer: %v", err)
}
......@@ -1888,12 +1918,13 @@ func (dm *DockerManager) createPodInfraContainer(pod *v1.Pod) (kubecontainer.Doc
// No pod secrets for the infra container.
// The message isn't needed for the Infra container
if err, msg := dm.imagePuller.EnsureImageExists(pod, container, nil); err != nil {
imageRef, msg, err := dm.imagePuller.EnsureImageExists(pod, container, nil)
if err != nil {
return "", err, msg
}
// Currently we don't care about restart count of infra container, just set it to 0.
id, err := dm.runContainerInPod(pod, container, netNamespace, getIPCMode(pod), getPidMode(pod), "", 0)
id, err := dm.runContainerInPod(pod, container, netNamespace, getIPCMode(pod), getPidMode(pod), "", imageRef, 0)
if err != nil {
return "", kubecontainer.ErrRunContainer, err.Error()
}
......@@ -2305,7 +2336,7 @@ func (dm *DockerManager) SyncPod(pod *v1.Pod, _ v1.PodStatus, podStatus *kubecon
// tryContainerStart attempts to pull and start the container, returning an error and a reason string if the start
// was not successful.
func (dm *DockerManager) tryContainerStart(container *v1.Container, pod *v1.Pod, podStatus *kubecontainer.PodStatus, pullSecrets []v1.Secret, namespaceMode, pidMode, podIP string) (err error, reason string) {
err, msg := dm.imagePuller.EnsureImageExists(pod, container, pullSecrets)
imageRef, msg, err := dm.imagePuller.EnsureImageExists(pod, container, pullSecrets)
if err != nil {
return err, msg
}
......@@ -2331,7 +2362,7 @@ func (dm *DockerManager) tryContainerStart(container *v1.Container, pod *v1.Pod,
netMode = namespaceMode
}
_, err = dm.runContainerInPod(pod, container, netMode, namespaceMode, pidMode, podIP, restartCount)
_, err = dm.runContainerInPod(pod, container, netMode, namespaceMode, pidMode, podIP, imageRef, restartCount)
if err != nil {
// TODO(bburns) : Perhaps blacklist a container after N failures?
return kubecontainer.ErrRunContainer, err.Error()
......
pkg/kubelet/dockertools/docker_manager_test.go
View file @ f5d9c430
......@@ -112,8 +112,8 @@ func newFakeImageManager() images.ImageManager {
return &fakeImageManager{}
}
func (m *fakeImageManager) EnsureImageExists(pod *v1.Pod, container *v1.Container, pullSecrets []v1.Secret) (error, string) {
return nil, ""
func (m *fakeImageManager) EnsureImageExists(pod *v1.Pod, container *v1.Container, pullSecrets []v1.Secret) (string, string, error) {
return container.Image, "", nil
}
func createTestDockerManager(fakeHTTPClient *fakeHTTP, fakeDocker *FakeDockerClient) (*DockerManager, *FakeDockerClient) {
......@@ -647,9 +647,9 @@ func TestSyncPodCreatesNetAndContainerPullsImage(t *testing.T) {
verifyCalls(t, fakeDocker, []string{
// Create pod infra container.
"create", "start", "inspect_container", "inspect_container",
"inspect_image", "create", "start", "inspect_container", "inspect_container",
// Create container.
"create", "start", "inspect_container",
"inspect_image", "create", "start", "inspect_container",
})
fakeDocker.Lock()
......
pkg/kubelet/dockertools/docker_test.go
View file @ f5d9c430
......@@ -705,12 +705,12 @@ func (f *imageTrackingDockerClient) InspectImageByRef(name string) (image *docke
return
}
func TestIsImagePresent(t *testing.T) {
func TestGetImageRef(t *testing.T) {
cl := &imageTrackingDockerClient{NewFakeDockerClient(), ""}
puller := &dockerPuller{
client: cl,
}
_, _ = puller.IsImagePresent("abc:123")
_, _ = puller.GetImageRef("abc:123")
if cl.imageName != "abc:123" {
t.Errorf("expected inspection of image abc:123, instead inspected image %v", cl.imageName)
}
......
pkg/kubelet/dockertools/fake_docker_client.go
View file @ f5d9c430
......@@ -478,6 +478,10 @@ func (f *FakeDockerClient) PullImage(image string, auth dockertypes.AuthConfig,
err := f.popError("pull")
if err == nil {
authJson, _ := json.Marshal(auth)
f.Image = &dockertypes.ImageInspect{
ID: image,
RepoTags: []string{image},
}
f.pulled = append(f.pulled, fmt.Sprintf("%s using %s", image, string(authJson)))
}
return err
......@@ -592,18 +596,18 @@ func (f *FakeDockerPuller) Pull(image string, secrets []v1.Secret) (err error) {
return err
}
func (f *FakeDockerPuller) IsImagePresent(name string) (bool, error) {
func (f *FakeDockerPuller) GetImageRef(name string) (string, error) {
f.Lock()
defer f.Unlock()
if f.HasImages == nil {
return true, nil
return name, nil
}
for _, s := range f.HasImages {
if s == name {
return true, nil
return s, nil
}
}
return false, nil
return "", nil
}
func (f *FakeDockerClient) ImageHistory(id string) ([]dockertypes.ImageHistory, error) {
f.Lock()
......
pkg/kubelet/images/helpers.go
View file @ f5d9c430
......@@ -42,9 +42,9 @@ type throttledImageService struct {
limiter flowcontrol.RateLimiter
}
func (ts throttledImageService) PullImage(image kubecontainer.ImageSpec, secrets []v1.Secret) error {
func (ts throttledImageService) PullImage(image kubecontainer.ImageSpec, secrets []v1.Secret) (string, error) {
if ts.limiter.TryAccept() {
return ts.ImageService.PullImage(image, secrets)
}
return fmt.Errorf("pull QPS exceeded.")
return "", fmt.Errorf("pull QPS exceeded.")
}
pkg/kubelet/images/image_manager.go
View file @ f5d9c430
......@@ -81,8 +81,9 @@ func (m *imageManager) logIt(ref *v1.ObjectReference, eventtype, event, prefix,
}
}
// EnsureImageExists pulls the image for the specified pod and container.
func (m *imageManager) EnsureImageExists(pod *v1.Pod, container *v1.Container, pullSecrets []v1.Secret) (error, string) {
// EnsureImageExists pulls the image for the specified pod and container, and returns
// (imageRef, error message, error).
func (m *imageManager) EnsureImageExists(pod *v1.Pod, container *v1.Container, pullSecrets []v1.Secret) (string, string, error) {
logPrefix := fmt.Sprintf("%s/%s", pod.Name, container.Image)
ref, err := kubecontainer.GenerateContainerRef(pod, container)
if err != nil {
......@@ -94,26 +95,27 @@ func (m *imageManager) EnsureImageExists(pod *v1.Pod, container *v1.Container, p
if err != nil {
msg := fmt.Sprintf("Failed to apply default image tag %q: %v", container.Image, err)
m.logIt(ref, v1.EventTypeWarning, events.FailedToInspectImage, logPrefix, msg, glog.Warning)
return ErrInvalidImageName, msg
return "", msg, ErrInvalidImageName
}
spec := kubecontainer.ImageSpec{Image: image}
present, err := m.imageService.IsImagePresent(spec)
imageRef, err := m.imageService.GetImageRef(spec)
if err != nil {
msg := fmt.Sprintf("Failed to inspect image %q: %v", container.Image, err)
m.logIt(ref, v1.EventTypeWarning, events.FailedToInspectImage, logPrefix, msg, glog.Warning)
return ErrImageInspect, msg
return "", msg, ErrImageInspect
}
present := imageRef != ""
if !shouldPullImage(container, present) {
if present {
msg := fmt.Sprintf("Container image %q already present on machine", container.Image)
m.logIt(ref, v1.EventTypeNormal, events.PulledImage, logPrefix, msg, glog.Info)
return nil, ""
return imageRef, "", nil
} else {
msg := fmt.Sprintf("Container image %q is not present with pull policy of Never", container.Image)
m.logIt(ref, v1.EventTypeWarning, events.ErrImageNeverPullPolicy, logPrefix, msg, glog.Warning)
return ErrImageNeverPull, msg
return "", msg, ErrImageNeverPull
}
}
......@@ -121,24 +123,25 @@ func (m *imageManager) EnsureImageExists(pod *v1.Pod, container *v1.Container, p
if m.backOff.IsInBackOffSinceUpdate(backOffKey, m.backOff.Clock.Now()) {
msg := fmt.Sprintf("Back-off pulling image %q", container.Image)
m.logIt(ref, v1.EventTypeNormal, events.BackOffPullImage, logPrefix, msg, glog.Info)
return ErrImagePullBackOff, msg
return "", msg, ErrImagePullBackOff
}
m.logIt(ref, v1.EventTypeNormal, events.PullingImage, logPrefix, fmt.Sprintf("pulling image %q", container.Image), glog.Info)
errChan := make(chan error)
m.puller.pullImage(spec, pullSecrets, errChan)
if err := <-errChan; err != nil {
m.logIt(ref, v1.EventTypeWarning, events.FailedToPullImage, logPrefix, fmt.Sprintf("Failed to pull image %q: %v", container.Image, err), glog.Warning)
pullChan := make(chan pullResult)
m.puller.pullImage(spec, pullSecrets, pullChan)
imagePullResult := <-pullChan
if imagePullResult.err != nil {
m.logIt(ref, v1.EventTypeWarning, events.FailedToPullImage, logPrefix, fmt.Sprintf("Failed to pull image %q: %v", container.Image, imagePullResult.err), glog.Warning)
m.backOff.Next(backOffKey, m.backOff.Clock.Now())
if err == RegistryUnavailable {
if imagePullResult.err == RegistryUnavailable {
msg := fmt.Sprintf("image pull failed for %s because the registry is unavailable.", container.Image)
return err, msg
} else {
return ErrImagePull, err.Error()
return "", msg, imagePullResult.err
}
return "", imagePullResult.err.Error(), ErrImagePull
}
m.logIt(ref, v1.EventTypeNormal, events.PulledImage, logPrefix, fmt.Sprintf("Successfully pulled image %q", container.Image), glog.Info)
m.backOff.GC()
return nil, ""
return imagePullResult.imageRef, "", nil
}
// applyDefaultImageTag parses a docker image string, if it doesn't contain any tag or digest,
......
pkg/kubelet/images/image_manager_test.go
View file @ f5d9c430
......@@ -44,7 +44,7 @@ func pullerTestCases() []pullerTestCase {
{ // pull missing image
containerImage: "missing_image",
policy: v1.PullIfNotPresent,
calledFunctions: []string{"IsImagePresent", "PullImage"},
calledFunctions: []string{"GetImageRef", "PullImage"},
inspectErr: nil,
pullerErr: nil,
expectedErr: []error{nil}},
......@@ -52,35 +52,35 @@ func pullerTestCases() []pullerTestCase {
{ // image present, don't pull
containerImage: "present_image",
policy: v1.PullIfNotPresent,
calledFunctions: []string{"IsImagePresent"},
calledFunctions: []string{"GetImageRef"},
inspectErr: nil,
pullerErr: nil,
expectedErr: []error{nil, nil, nil}},
// image present, pull it
{containerImage: "present_image",
policy: v1.PullAlways,
calledFunctions: []string{"IsImagePresent", "PullImage"},
calledFunctions: []string{"GetImageRef", "PullImage"},
inspectErr: nil,
pullerErr: nil,
expectedErr: []error{nil, nil, nil}},
// missing image, error PullNever
{containerImage: "missing_image",
policy: v1.PullNever,
calledFunctions: []string{"IsImagePresent"},
calledFunctions: []string{"GetImageRef"},
inspectErr: nil,
pullerErr: nil,
expectedErr: []error{ErrImageNeverPull, ErrImageNeverPull, ErrImageNeverPull}},
// missing image, unable to inspect
{containerImage: "missing_image",
policy: v1.PullIfNotPresent,
calledFunctions: []string{"IsImagePresent"},
calledFunctions: []string{"GetImageRef"},
inspectErr: errors.New("unknown inspectError"),
pullerErr: nil,
expectedErr: []error{ErrImageInspect, ErrImageInspect, ErrImageInspect}},
// missing image, unable to fetch
{containerImage: "typo_image",
policy: v1.PullIfNotPresent,
calledFunctions: []string{"IsImagePresent", "PullImage"},
calledFunctions: []string{"GetImageRef", "PullImage"},
inspectErr: nil,
pullerErr: errors.New("404"),
expectedErr: []error{ErrImagePull, ErrImagePull, ErrImagePullBackOff, ErrImagePull, ErrImagePullBackOff, ErrImagePullBackOff}},
......@@ -126,7 +126,7 @@ func TestParallelPuller(t *testing.T) {
for tick, expected := range c.expectedErr {
fakeClock.Step(time.Second)
err, _ := puller.EnsureImageExists(pod, container, nil)
_, _, err := puller.EnsureImageExists(pod, container, nil)
fakeRuntime.AssertCalls(c.calledFunctions)
assert.Equal(t, expected, err, "in test %d tick=%d", i, tick)
}
......@@ -150,7 +150,7 @@ func TestSerializedPuller(t *testing.T) {
for tick, expected := range c.expectedErr {
fakeClock.Step(time.Second)
err, _ := puller.EnsureImageExists(pod, container, nil)
_, _, err := puller.EnsureImageExists(pod, container, nil)
fakeRuntime.AssertCalls(c.calledFunctions)
assert.Equal(t, expected, err, "in test %d tick=%d", i, tick)
}
......
pkg/kubelet/images/puller.go
View file @ f5d9c430
......@@ -24,8 +24,13 @@ import (
"k8s.io/kubernetes/pkg/util/wait"
)
type pullResult struct {
imageRef string
err error
}
type imagePuller interface {
pullImage(kubecontainer.ImageSpec, []v1.Secret, chan<- error)
pullImage(kubecontainer.ImageSpec, []v1.Secret, chan<- pullResult)
}
var _, _ imagePuller = &parallelImagePuller{}, &serialImagePuller{}
......@@ -38,9 +43,13 @@ func newParallelImagePuller(imageService kubecontainer.ImageService) imagePuller
return &parallelImagePuller{imageService}
}
func (pip *parallelImagePuller) pullImage(spec kubecontainer.ImageSpec, pullSecrets []v1.Secret, errChan chan<- error) {
func (pip *parallelImagePuller) pullImage(spec kubecontainer.ImageSpec, pullSecrets []v1.Secret, pullChan chan<- pullResult) {
go func() {
errChan <- pip.imageService.PullImage(spec, pullSecrets)
imageRef, err := pip.imageService.PullImage(spec, pullSecrets)
pullChan <- pullResult{
imageRef: imageRef,
err: err,
}
}()
}
......@@ -61,19 +70,23 @@ func newSerialImagePuller(imageService kubecontainer.ImageService) imagePuller {
type imagePullRequest struct {
spec kubecontainer.ImageSpec
pullSecrets []v1.Secret
errChan chan<- error
pullChan chan<- pullResult
}
func (sip *serialImagePuller) pullImage(spec kubecontainer.ImageSpec, pullSecrets []v1.Secret, errChan chan<- error) {
func (sip *serialImagePuller) pullImage(spec kubecontainer.ImageSpec, pullSecrets []v1.Secret, pullChan chan<- pullResult) {
sip.pullRequests <- &imagePullRequest{
spec: spec,
pullSecrets: pullSecrets,
errChan: errChan,
pullChan: pullChan,
}
}
func (sip *serialImagePuller) processImagePullRequests() {
for pullRequest := range sip.pullRequests {
pullRequest.errChan <- sip.imageService.PullImage(pullRequest.spec, pullRequest.pullSecrets)
imageRef, err := sip.imageService.PullImage(pullRequest.spec, pullRequest.pullSecrets)
pullRequest.pullChan <- pullResult{
imageRef: imageRef,
err: err,
}
}
}
pkg/kubelet/images/types.go
View file @ f5d9c430
......@@ -49,7 +49,7 @@ var (
// Implementations are expected to be thread safe.
type ImageManager interface {
// EnsureImageExists ensures that image specified in `container` exists.
EnsureImageExists(pod *v1.Pod, container *v1.Container, pullSecrets []v1.Secret) (error, string)
EnsureImageExists(pod *v1.Pod, container *v1.Container, pullSecrets []v1.Secret) (string, string, error)
// TODO(ronl): consolidating image managing and deleting operation in this interface
}
pkg/kubelet/kuberuntime/instrumented_services.go
View file @ f5d9c430
......@@ -239,13 +239,13 @@ func (in instrumentedImageManagerService) ImageStatus(image *runtimeapi.ImageSpe
return out, err
}
func (in instrumentedImageManagerService) PullImage(image *runtimeapi.ImageSpec, auth *runtimeapi.AuthConfig) error {
func (in instrumentedImageManagerService) PullImage(image *runtimeapi.ImageSpec, auth *runtimeapi.AuthConfig) (string, error) {
const operation = "pull_image"
defer recordOperation(operation, time.Now())
err := in.service.PullImage(image, auth)
imageRef, err := in.service.PullImage(image, auth)
recordError(operation, err)
return err
return imageRef, err
}
func (in instrumentedImageManagerService) RemoveImage(image *runtimeapi.ImageSpec) error {
......
pkg/kubelet/kuberuntime/kuberuntime_container.go
View file @ f5d9c430
......@@ -51,7 +51,7 @@ import (
// * run the post start lifecycle hooks (if applicable)
func (m *kubeGenericRuntimeManager) startContainer(podSandboxID string, podSandboxConfig *runtimeapi.PodSandboxConfig, container *v1.Container, pod *v1.Pod, podStatus *kubecontainer.PodStatus, pullSecrets []v1.Secret, podIP string) (string, error) {
// Step 1: pull the image.
err, msg := m.imagePuller.EnsureImageExists(pod, container, pullSecrets)
imageRef, msg, err := m.imagePuller.EnsureImageExists(pod, container, pullSecrets)
if err != nil {
return msg, err
}
......@@ -70,7 +70,7 @@ func (m *kubeGenericRuntimeManager) startContainer(podSandboxID string, podSandb
restartCount = containerStatus.RestartCount + 1
}
containerConfig, err := m.generateContainerConfig(container, pod, restartCount, podIP)
containerConfig, err := m.generateContainerConfig(container, pod, restartCount, podIP, imageRef)
if err != nil {
m.recorder.Eventf(ref, v1.EventTypeWarning, events.FailedToCreateContainer, "Failed to create container with error: %v", err)
return "Generate Container Config Failed", err
......@@ -129,7 +129,7 @@ func (m *kubeGenericRuntimeManager) startContainer(podSandboxID string, podSandb
}
// generateContainerConfig generates container config for kubelet runtime v1.
func (m *kubeGenericRuntimeManager) generateContainerConfig(container *v1.Container, pod *v1.Pod, restartCount int, podIP string) (*runtimeapi.ContainerConfig, error) {
func (m *kubeGenericRuntimeManager) generateContainerConfig(container *v1.Container, pod *v1.Pod, restartCount int, podIP, imageRef string) (*runtimeapi.ContainerConfig, error) {
opts, err := m.runtimeHelper.GenerateRunContainerOptions(pod, container, podIP)
if err != nil {
return nil, err
......@@ -156,7 +156,7 @@ func (m *kubeGenericRuntimeManager) generateContainerConfig(container *v1.Contai
Name: &container.Name,
Attempt: &restartCountUint32,
},
Image: &runtimeapi.ImageSpec{Image: &container.Image},
Image: &runtimeapi.ImageSpec{Image: &imageRef},
Command: command,
Args: args,
WorkingDir: &container.WorkingDir,
......
pkg/kubelet/kuberuntime/kuberuntime_image.go
View file @ f5d9c430
......@@ -28,16 +28,16 @@ import (
// PullImage pulls an image from the network to local storage using the supplied
// secrets if necessary.
func (m *kubeGenericRuntimeManager) PullImage(image kubecontainer.ImageSpec, pullSecrets []v1.Secret) error {
func (m *kubeGenericRuntimeManager) PullImage(image kubecontainer.ImageSpec, pullSecrets []v1.Secret) (string, error) {
img := image.Image
repoToPull, _, _, err := parsers.ParseImageName(img)
if err != nil {
return err
return "", err
}
keyring, err := credentialprovider.MakeDockerKeyring(pullSecrets, m.keyring)
if err != nil {
return err
return "", err
}
imgSpec := &runtimeapi.ImageSpec{Image: &img}
......@@ -45,13 +45,13 @@ func (m *kubeGenericRuntimeManager) PullImage(image kubecontainer.ImageSpec, pul
if !withCredentials {
glog.V(3).Infof("Pulling image %q without credentials", img)
err = m.imageService.PullImage(imgSpec, nil)
imageRef, err := m.imageService.PullImage(imgSpec, nil)
if err != nil {
glog.Errorf("Pull image %q failed: %v", img, err)
return err
return "", err
}
return nil
return imageRef, nil
}
var pullErrs []error
......@@ -66,26 +66,35 @@ func (m *kubeGenericRuntimeManager) PullImage(image kubecontainer.ImageSpec, pul
RegistryToken: &authConfig.RegistryToken,
}
err = m.imageService.PullImage(imgSpec, auth)
imageRef, err := m.imageService.PullImage(imgSpec, auth)
// If there was no error, return success
if err == nil {
return nil
return imageRef, nil
}
pullErrs = append(pullErrs, err)
}
return utilerrors.NewAggregate(pullErrs)
return "", utilerrors.NewAggregate(pullErrs)
}
// IsImagePresent checks whether the container image is already in the local storage.
func (m *kubeGenericRuntimeManager) IsImagePresent(image kubecontainer.ImageSpec) (bool, error) {
// GetImageRef gets the reference (digest or ID) of the image which has already been in
// the local storage. It returns ("", nil) if the image isn't in the local storage.
func (m *kubeGenericRuntimeManager) GetImageRef(image kubecontainer.ImageSpec) (string, error) {
status, err := m.imageService.ImageStatus(&runtimeapi.ImageSpec{Image: &image.Image})
if err != nil {
glog.Errorf("ImageStatus for image %q failed: %v", image, err)
return false, err
return "", err
}
if status == nil {
return "", nil
}
imageRef := status.GetId()
if len(status.RepoDigests) > 0 {
imageRef = status.RepoDigests[0]
}
return status != nil, nil
return imageRef, nil
}
// ListImages gets all images currently on the machine.
......
pkg/kubelet/kuberuntime/kuberuntime_image_test.go
View file @ f5d9c430
......@@ -28,8 +28,9 @@ func TestPullImage(t *testing.T) {
_, _, fakeManager, err := createTestRuntimeManager()
assert.NoError(t, err)
err = fakeManager.PullImage(kubecontainer.ImageSpec{Image: "busybox"}, nil)
imageRef, err := fakeManager.PullImage(kubecontainer.ImageSpec{Image: "busybox"}, nil)
assert.NoError(t, err)
assert.Equal(t, "busybox", imageRef)
images, err := fakeManager.ListImages()
assert.NoError(t, err)
......@@ -55,22 +56,22 @@ func TestListImages(t *testing.T) {
assert.Equal(t, expected.List(), actual.List())
}
func TestIsImagePresent(t *testing.T) {
func TestGetImageRef(t *testing.T) {
_, fakeImageService, fakeManager, err := createTestRuntimeManager()
assert.NoError(t, err)
image := "busybox"
fakeImageService.SetFakeImages([]string{image})
present, err := fakeManager.IsImagePresent(kubecontainer.ImageSpec{Image: image})
imageRef, err := fakeManager.GetImageRef(kubecontainer.ImageSpec{Image: image})
assert.NoError(t, err)
assert.Equal(t, true, present)
assert.Equal(t, image, imageRef)
}
func TestRemoveImage(t *testing.T) {
_, fakeImageService, fakeManager, err := createTestRuntimeManager()
assert.NoError(t, err)
err = fakeManager.PullImage(kubecontainer.ImageSpec{Image: "busybox"}, nil)
_, err = fakeManager.PullImage(kubecontainer.ImageSpec{Image: "busybox"}, nil)
assert.NoError(t, err)
assert.Equal(t, 1, len(fakeImageService.Images))
......
pkg/kubelet/kuberuntime/kuberuntime_manager_test.go
View file @ f5d9c430
......@@ -145,7 +145,7 @@ func makeFakeContainer(t *testing.T, m *kubeGenericRuntimeManager, template cont
sandboxConfig, err := m.generatePodSandboxConfig(template.pod, template.sandboxAttempt)
assert.NoError(t, err, "generatePodSandboxConfig for container template %+v", template)
containerConfig, err := m.generateContainerConfig(template.container, template.pod, template.attempt, "")
containerConfig, err := m.generateContainerConfig(template.container, template.pod, template.attempt, "", template.container.Image)
assert.NoError(t, err, "generateContainerConfig for container template %+v", template)
podSandboxID := apitest.BuildSandboxName(sandboxConfig.Metadata)
......
pkg/kubelet/remote/remote_image.go
View file @ f5d9c430
......@@ -79,20 +79,20 @@ func (r *RemoteImageService) ImageStatus(image *runtimeapi.ImageSpec) (*runtimea
}
// PullImage pulls an image with authentication config.
func (r *RemoteImageService) PullImage(image *runtimeapi.ImageSpec, auth *runtimeapi.AuthConfig) error {
func (r *RemoteImageService) PullImage(image *runtimeapi.ImageSpec, auth *runtimeapi.AuthConfig) (string, error) {
ctx, cancel := getContextWithTimeout(r.timeout)
defer cancel()
_, err := r.imageClient.PullImage(ctx, &runtimeapi.PullImageRequest{
resp, err := r.imageClient.PullImage(ctx, &runtimeapi.PullImageRequest{
Image: image,
Auth: auth,
})
if err != nil {
glog.Errorf("PullImage %q from image service failed: %v", image.GetImage(), err)
return err
return "", err
}
return nil
return resp.GetImageRef(), nil
}
// RemoveImage removes the image.
......
pkg/kubelet/rkt/image.go
View file @ f5d9c430
......@@ -45,18 +45,18 @@ import (
//
// http://issue.k8s.io/7203
//
func (r *Runtime) PullImage(image kubecontainer.ImageSpec, pullSecrets []v1.Secret) error {
func (r *Runtime) PullImage(image kubecontainer.ImageSpec, pullSecrets []v1.Secret) (string, error) {
img := image.Image
// TODO(yifan): The credential operation is a copy from dockertools package,
// Need to resolve the code duplication.
repoToPull, _, _, err := parsers.ParseImageName(img)
if err != nil {
return err
return "", err
}
keyring, err := credentialprovider.MakeDockerKeyring(pullSecrets, r.dockerKeyring)
if err != nil {
return err
return "", err
}
creds, ok := keyring.Lookup(repoToPull)
......@@ -66,7 +66,7 @@ func (r *Runtime) PullImage(image kubecontainer.ImageSpec, pullSecrets []v1.Secr
userConfigDir, err := ioutil.TempDir("", "rktnetes-user-config-dir-")
if err != nil {
return fmt.Errorf("rkt: Cannot create a temporary user-config directory: %v", err)
return "", fmt.Errorf("rkt: Cannot create a temporary user-config directory: %v", err)
}
defer os.RemoveAll(userConfigDir)
......@@ -74,7 +74,7 @@ func (r *Runtime) PullImage(image kubecontainer.ImageSpec, pullSecrets []v1.Secr
config.UserConfigDir = userConfigDir
if err := r.writeDockerAuthConfig(img, creds, userConfigDir); err != nil {
return err
return "", err
}
// Today, `--no-store` will fetch the remote image regardless of whether the content of the image
......@@ -82,14 +82,20 @@ func (r *Runtime) PullImage(image kubecontainer.ImageSpec, pullSecrets []v1.Secr
// the image pull policy is 'always'. The issue is tracked in https://github.com/coreos/rkt/issues/2937.
if _, err := r.cli.RunCommand(&config, "fetch", "--no-store", dockerPrefix+img); err != nil {
glog.Errorf("Failed to fetch: %v", err)
return err
return "", err
}
return nil
return r.getImageID(img)
}
func (r *Runtime) IsImagePresent(image kubecontainer.ImageSpec) (bool, error) {
func (r *Runtime) GetImageRef(image kubecontainer.ImageSpec) (string, error) {
images, err := r.listImages(image.Image, false)
return len(images) > 0, err
if err != nil {
return "", err
}
if len(images) == 0 {
return "", nil
}
return images[0].Id, nil
}
// ListImages lists all the available appc images on the machine by invoking 'rkt image list'.
......
pkg/kubelet/rkt/rkt.go
View file @ f5d9c430
......@@ -782,8 +782,9 @@ func (r *Runtime) newAppcRuntimeApp(pod *v1.Pod, podIP string, c v1.Container, r
if requiresPrivileged && !securitycontext.HasPrivilegedRequest(&c) {
return fmt.Errorf("cannot make %q: running a custom stage1 requires a privileged security context", format.Pod(pod))
}
if err, _ := r.imagePuller.EnsureImageExists(pod, &c, pullSecrets); err != nil {
return nil
imageRef, _, err := r.imagePuller.EnsureImageExists(pod, &c, pullSecrets)
if err != nil {
return err
}
imgManifest, err := r.getImageManifest(c.Image)
if err != nil {
......@@ -794,11 +795,7 @@ func (r *Runtime) newAppcRuntimeApp(pod *v1.Pod, podIP string, c v1.Container, r
imgManifest.App = new(appctypes.App)
}
imageID, err := r.getImageID(c.Image)
if err != nil {
return err
}
hash, err := appctypes.NewHash(imageID)
hash, err := appctypes.NewHash(imageRef)
if err != nil {
return err
}
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment