Merge pull request #12832 from yifan-gu/rkt_patch_img

kubelet/rkt: update image related interfaces.

pkg/kubelet/rkt/image.go

@@ -1,51 +0,0 @@
-/*
-Copyright 2015 The Kubernetes Authors All rights reserved.
-
-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 rkt
-
-import (
-	"fmt"
-	"strings"
-)
-
-// image stores the appc image information.
-// TODO(yifan): Replace with schema.ImageManifest.
-type image struct {
-	// The hash of the image, it must be universal unique. (e.g. sha512-xxx)
-	id string
-	// The name of the image manifest.
-	name string
-	// The version of the image. (e.g. v2.0.8, latest)
-	version string
-}
-
-// parseString creates the image struct by parsing the string in the result of 'rkt images',
-// the input looks like:
-//
-// sha512-91e98d7f1679a097c878203c9659f2a26ae394656b3147963324c61fa3832f15	coreos.com/etcd:v2.0.9
-//
-func (im *image) parseString(input string) error {
-	idName := strings.Split(strings.TrimSpace(input), "\t")
-	if len(idName) != 2 {
-		return fmt.Errorf("invalid image information from 'rkt images': %q", input)
-	}
-	nameVersion := strings.Split(idName[1], ":")
-	if len(nameVersion) != 2 {
-		return fmt.Errorf("cannot parse the app name: %q", nameVersion)
-	}
-	im.id, im.name, im.version = idName[0], nameVersion[0], nameVersion[1]
-	return nil
-}

pkg/kubelet/rkt/rkt.go

@@ -73,6 +73,8 @@ const (
 	defaultGracePeriod = "1m"
 	// Duration to wait before expiring prepared pods.
 	defaultExpirePrepared = "1m"
+
+	defaultImageTag = "latest"
 )
 
 // runtime implements the Containerruntime for rkt. The implementation
@@ -92,6 +94,7 @@ type runtime struct {
 	prober              prober.Prober
 	readinessManager    *kubecontainer.ReadinessManager
 	volumeGetter        volumeGetter
+	imagePuller         kubecontainer.ImagePuller
 }
 
 var _ kubecontainer.Runtime = &runtime{}
@@ -147,6 +150,7 @@ func New(config *Config,
 		volumeGetter:        volumeGetter,
 	}
 	rkt.prober = prober.New(rkt, readinessManager, containerRefManager, recorder)
+	rkt.imagePuller = kubecontainer.NewImagePuller(recorder, rkt)
 
 	// Test the rkt version.
 	version, err := rkt.Version()
@@ -358,13 +362,28 @@ func setApp(app *appctypes.App, c *api.Container, opts *kubecontainer.RunContain
 	return setIsolators(app, c)
 }
 
+// parseImageName parses a docker image string into two parts: repo and tag.
+// If tag is empty, return the defaultImageTag.
+func parseImageName(image string) (string, string) {
+	repoToPull, tag := parsers.ParseRepositoryTag(image)
+	// If no tag was specified, use the default "latest".
+	if len(tag) == 0 {
+		tag = defaultImageTag
+	}
+	return repoToPull, tag
+}
+
 // getImageManifest invokes 'rkt image cat-manifest' to retrive the image manifest
 // for the image.
 func (r *runtime) getImageManifest(image string) (*appcschema.ImageManifest, error) {
 	var manifest appcschema.ImageManifest
 
-	// TODO(yifan): Assume docker images for now.
+	repoToPull, tag := parseImageName(image)
-	output, err := r.runCommand("image", "cat-manifest", "--quiet", dockerPrefix+image)
+	imgName, err := appctypes.SanitizeACIdentifier(repoToPull)
+	if err != nil {
+		return nil, err
+	}
+	output, err := r.runCommand("image", "cat-manifest", fmt.Sprintf("%s:%s", imgName, tag))
 	if err != nil {
 		return nil, err
 	}
@@ -375,11 +394,14 @@ func (r *runtime) getImageManifest(image string) (*appcschema.ImageManifest, err
 }
 
 // makePodManifest transforms a kubelet pod spec to the rkt pod manifest.
-func (r *runtime) makePodManifest(pod *api.Pod) (*appcschema.PodManifest, error) {
+func (r *runtime) makePodManifest(pod *api.Pod, pullSecrets []api.Secret) (*appcschema.PodManifest, error) {
 	var globalPortMappings []kubecontainer.PortMapping
 	manifest := appcschema.BlankPodManifest()
 
 	for _, c := range pod.Spec.Containers {
+		if err := r.imagePuller.PullImage(pod, &c, pullSecrets); err != nil {
+			return nil, err
+		}
 		imgManifest, err := r.getImageManifest(c.Image)
 		if err != nil {
 			return nil, err
@@ -393,7 +415,7 @@ func (r *runtime) makePodManifest(pod *api.Pod) (*appcschema.PodManifest, error)
 		if err != nil {
 			return nil, err
 		}
-		hash, err := appctypes.NewHash(img.id)
+		hash, err := appctypes.NewHash(img.ID)
 		if err != nil {
 			return nil, err
 		}
@@ -460,6 +482,9 @@ func newUnitOption(section, name, value string) *unit.UnitOption {
 	return &unit.UnitOption{Section: section, Name: name, Value: value}
 }
 
+// apiPodToruntimePod converts an api.Pod to kubelet/container.Pod.
+// we save the this for later reconstruction of the kubelet/container.Pod
+// such as in GetPods().
 func apiPodToruntimePod(uuid string, pod *api.Pod) *kubecontainer.Pod {
 	p := &kubecontainer.Pod{
 		ID:        pod.UID,
@@ -487,11 +512,11 @@ func apiPodToruntimePod(uuid string, pod *api.Pod) *kubecontainer.Pod {
 // On success, it will return a string that represents name of the unit file
 // and a boolean that indicates if the unit file needs to be reloaded (whether
 // the file is already existed).
-func (r *runtime) preparePod(pod *api.Pod) (string, bool, error) {
+func (r *runtime) preparePod(pod *api.Pod, pullSecrets []api.Secret) (string, bool, error) {
 	cmds := []string{"prepare", "--quiet", "--pod-manifest"}
 
 	// Generate the pod manifest from the pod spec.
-	manifest, err := r.makePodManifest(pod)
+	manifest, err := r.makePodManifest(pod, pullSecrets)
 	if err != nil {
 		return "", false, err
 	}
@@ -570,10 +595,10 @@ func (r *runtime) preparePod(pod *api.Pod) (string, bool, error) {
 
 // RunPod first creates the unit file for a pod, and then calls
 // StartUnit over d-bus.
-func (r *runtime) RunPod(pod *api.Pod) error {
+func (r *runtime) RunPod(pod *api.Pod, pullSecrets []api.Secret) error {
 	glog.V(4).Infof("Rkt starts to run pod: name %q.", pod.Name)
 
-	name, needReload, err := r.preparePod(pod)
+	name, needReload, err := r.preparePod(pod, pullSecrets)
 	if err != nil {
 		return err
 	}
@@ -727,9 +752,13 @@ func (r *runtime) Version() (kubecontainer.Version, error) {
 	return nil, fmt.Errorf("rkt: cannot determine the version")
 }
 
-// writeDockerAuthConfig writes the docker credentials to rkt auth config files.
+// TODO(yifan): This is very racy, unefficient, and unsafe, we need to provide
-// This enables rkt to pull docker images from docker registry with credentials.
+// different namespaces. See: https://github.com/coreos/rkt/issues/836.
 func (r *runtime) writeDockerAuthConfig(image string, credsSlice []docker.AuthConfiguration) error {
+	if len(credsSlice) == 0 {
+		return nil
+	}
+
 	creds := docker.AuthConfiguration{}
 	// TODO handle multiple creds
 	if len(credsSlice) >= 1 {
@@ -754,14 +783,9 @@ func (r *runtime) writeDockerAuthConfig(image string, credsSlice []docker.AuthCo
 			return err
 		}
 	}
-	f, err := os.Create(path.Join(localConfigDir, authDir, registry+".json"))
+
-	if err != nil {
-		glog.Errorf("rkt: Cannot create docker auth config file: %v", err)
-		return err
-	}
-	defer f.Close()
 	config := fmt.Sprintf(dockerAuthTemplate, registry, creds.Username, creds.Password)
-	if _, err := f.Write([]byte(config)); err != nil {
+	if err := ioutil.WriteFile(path.Join(authDir, registry+".json"), []byte(config), 0600); err != nil {
 		glog.Errorf("rkt: Cannot write docker auth config file: %v", err)
 		return err
 	}
@@ -778,12 +802,7 @@ func (r *runtime) PullImage(image kubecontainer.ImageSpec, pullSecrets []api.Sec
 	img := image.Image
 	// TODO(yifan): The credential operation is a copy from dockertools package,
 	// Need to resolve the code duplication.
-	repoToPull, tag := parsers.ParseRepositoryTag(img)
+	repoToPull, _ := parseImageName(img)
-	// If no tag was specified, use the default "latest".
-	if len(tag) == 0 {
-		tag = "latest"
-	}
-
 	keyring, err := credentialprovider.MakeDockerKeyring(pullSecrets, r.dockerKeyring)
 	if err != nil {
 		return err
@@ -800,39 +819,44 @@ func (r *runtime) PullImage(image kubecontainer.ImageSpec, pullSecrets []api.Sec
 		return err
 	}
 
-	output, err := r.runCommand("fetch", dockerPrefix+img)
+	if _, err := r.runCommand("fetch", dockerPrefix+img); err != nil {
-	if err != nil {
+		glog.Errorf("Failed to fetch: %v", err)
-		return fmt.Errorf("rkt: Failed to fetch image: %v:", output)
+		return err
 	}
 	return nil
 }
 
-// IsImagePresent returns true if the image is available on the machine.
+// TODO(yifan): Searching the image via 'rkt images' might not be the most efficient way.
-// TODO(yifan): 'rkt image' is now landed on master, use that once we bump up
-// the rkt version.
 func (r *runtime) IsImagePresent(image kubecontainer.ImageSpec) (bool, error) {
-	img := image.Image
+	repoToPull, tag := parseImageName(image.Image)
-	if _, err := r.runCommand("prepare", "--local=true", dockerPrefix+img); err != nil {
+	// TODO(yifan): Change appname to imagename. See https://github.com/coreos/rkt/issues/1295.
-		return false, nil
+	output, err := r.runCommand("image", "list", "--fields=appname", "--no-legend")
+	if err != nil {
+		return false, err
 	}
-	return true, nil
+	for _, line := range output {
-}
+		parts := strings.Split(strings.TrimSpace(line), ":")
 
-func (r *runtime) ListImages() ([]kubecontainer.Image, error) {
+		var imgName, imgTag string
-	return []kubecontainer.Image{}, fmt.Errorf("rkt: ListImages unimplemented")
+		switch len(parts) {
-}
+		case 1:
+			imgName, imgTag = parts[0], defaultImageTag
+		case 2:
+			imgName, imgTag = parts[0], parts[1]
+		default:
+			continue
+		}
 
-func (r *runtime) RemoveImage(image kubecontainer.ImageSpec) error {
+		if imgName == repoToPull && imgTag == tag {
-	return fmt.Errorf("rkt: RemoveImages unimplemented")
+			return true, nil
+		}
+	}
+	return false, nil
 }
 
 // SyncPod syncs the running pod to match the specified desired pod.
 func (r *runtime) SyncPod(pod *api.Pod, runningPod kubecontainer.Pod, podStatus api.PodStatus, pullSecrets []api.Secret) error {
 	podFullName := kubecontainer.GetPodFullName(pod)
-	if len(runningPod.Containers) == 0 {
-		glog.V(4).Infof("Pod %q is not running, will start it", podFullName)
-		return r.RunPod(pod)
-	}
 
 	// Add references to all containers.
 	unidentifiedContainers := make(map[types.UID]*kubecontainer.Container)
@@ -890,7 +914,7 @@ func (r *runtime) SyncPod(pod *api.Pod, runningPod kubecontainer.Pod, podStatus
 		if err := r.KillPod(runningPod); err != nil {
 			return err
 		}
-		if err := r.RunPod(pod); err != nil {
+		if err := r.RunPod(pod, pullSecrets); err != nil {
 			return err
 		}
 	}
@@ -1145,9 +1169,40 @@ func (r *runtime) getPodInfos() (map[string]*podInfo, error) {
 	return result, nil
 }
 
-// listImages lists all the available appc images on the machine by invoking 'rkt images'.
+// getImageByName tries to find the image info with the given image name.
-func (r *runtime) listImages() ([]image, error) {
+// TODO(yifan): Replace with 'rkt image cat-manifest'.
-	output, err := r.runCommand("images", "--no-legend=true", "--fields=key,appname")
+// imageName should be in the form of 'example.com/app:latest', which should matches
+// the result of 'rkt image list'. If the version is empty, then 'latest' is assumed.
+func (r *runtime) getImageByName(imageName string) (*kubecontainer.Image, error) {
+	// TODO(yifan): Print hash in 'rkt image cat-manifest'?
+	images, err := r.ListImages()
+	if err != nil {
+		return nil, err
+	}
+
+	nameVersion := strings.Split(imageName, ":")
+	switch len(nameVersion) {
+	case 1:
+		imageName += ":" + defaultImageTag
+	case 2:
+		break
+	default:
+		return nil, fmt.Errorf("invalid image name: %q, requires 'name[:version]'")
+	}
+
+	for _, img := range images {
+		for _, t := range img.Tags {
+			if t == imageName {
+				return &img, nil
+			}
+		}
+	}
+	return nil, fmt.Errorf("cannot find the image %q", imageName)
+}
+
+// ListImages lists all the available appc images on the machine by invoking 'rkt image list'.
+func (r *runtime) ListImages() ([]kubecontainer.Image, error) {
+	output, err := r.runCommand("image", "list", "--no-legend=true", "--fields=key,appname")
 	if err != nil {
 		return nil, err
 	}
@@ -1155,45 +1210,44 @@ func (r *runtime) listImages() ([]image, error) {
 		return nil, nil
 	}
 
-	var images []image
+	var images []kubecontainer.Image
 	for _, line := range output {
-		var img image
+		img, err := parseImageInfo(line)
-		if err := img.parseString(line); err != nil {
+		if err != nil {
 			glog.Warningf("rkt: Cannot parse image info from %q: %v", line, err)
 			continue
 		}
-		images = append(images, img)
+		images = append(images, *img)
 	}
 	return images, nil
 }
 
-// getImageByName tries to find the image info with the given image name.
+// parseImageInfo creates the kubecontainer.Image struct by parsing the string in the result of 'rkt image list',
-// TODO(yifan): Replace with 'rkt image cat-manifest'.
+// the input looks like:
-func (r *runtime) getImageByName(imageName string) (image, error) {
+//
-	// TODO(yifan): Print hash in 'rkt image cat-manifest'?
+// sha512-91e98d7f1679a097c878203c9659f2a26ae394656b3147963324c61fa3832f15	coreos.com/etcd:v2.0.9
-	images, err := r.listImages()
+//
-	if err != nil {
+func parseImageInfo(input string) (*kubecontainer.Image, error) {
-		return image{}, err
+	idName := strings.Split(strings.TrimSpace(input), "\t")
+	if len(idName) != 2 {
+		return nil, fmt.Errorf("invalid image information from 'rkt image list': %q", input)
 	}
-
+	return &kubecontainer.Image{
-	var name, version string
+		ID:   idName[0],
-	nameVersion := strings.Split(imageName, ":")
+		Tags: []string{idName[1]},
-
+	}, nil
-	name, err = appctypes.SanitizeACIdentifier(nameVersion[0])
+}
-	if err != nil {
+
-		return image{}, err
+// RemoveImage removes an on-disk image using 'rkt image rm'.
-	}
+// TODO(yifan): Use image ID to reference image.
-
+func (r *runtime) RemoveImage(image kubecontainer.ImageSpec) error {
-	if len(nameVersion) == 2 {
+	img, err := r.getImageByName(image.Image)
-		version = nameVersion[1]
+	if err != nil {
-	}
+		return err
-
+	}
-	for _, img := range images {
+
-		if img.name == name {
+	if _, err := r.runCommand("image", "rm", img.ID); err != nil {
-			if version == "" || img.version == version {
+		return err
-				return img, nil
+	}
-			}
+	return nil
-		}
-	}
-	return image{}, fmt.Errorf("cannot find the image %q", imageName)
 }