Merge pull request #3097 from brendandburns/img

pkg/kubelet/dockertools/docker.go

@@ -45,7 +45,9 @@ type DockerInterface interface {
 	StopContainer(id string, timeout uint) error
 	RemoveContainer(opts docker.RemoveContainerOptions) error
 	InspectImage(image string) (*docker.Image, error)
+	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)
 	CreateExec(docker.CreateExecOptions) (*docker.Exec, error)
@@ -620,3 +622,21 @@ func parseImageName(image string) (string, string) {
 type ContainerCommandRunner interface {
 	RunInContainer(containerID string, cmd []string) ([]byte, error)
 }
+
+func GetUnusedImages(client DockerInterface) ([]string, error) {
+	// IMPORTANT: this is _unsafe_ to do while there are active pulls
+	// See https://github.com/docker/docker/issues/8926 for details
+	images, err := client.ListImages(docker.ListImagesOptions{
+		Filters: map[string][]string{
+			"dangling": {"true"},
+		},
+	})
+	if err != nil {
+		return nil, err
+	}
+	result := make([]string, len(images))
+	for ix := range images {
+		result[ix] = images[ix].ID
+	}
+	return result, nil
+}

pkg/kubelet/dockertools/fake_docker_client.go

@@ -21,6 +21,7 @@ import (
 	"reflect"
 	"sync"
 
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
 	"github.com/fsouza/go-dockerclient"
 )
 
@@ -31,12 +32,14 @@ type FakeDockerClient struct {
 	Container     *docker.Container
 	ContainerMap  map[string]*docker.Container
 	Image         *docker.Image
+	Images        []docker.APIImages
 	Err           error
 	called        []string
 	Stopped       []string
 	pulled        []string
 	Created       []string
 	Removed       []string
+	RemovedImages util.StringSet
 	VersionInfo   docker.Env
 }
 
@@ -172,10 +175,20 @@ func (f *FakeDockerClient) Version() (*docker.Env, error) {
 func (f *FakeDockerClient) CreateExec(_ docker.CreateExecOptions) (*docker.Exec, error) {
 	return &docker.Exec{"12345678"}, nil
 }
+
 func (f *FakeDockerClient) StartExec(_ string, _ docker.StartExecOptions) error {
 	return nil
 }
 
+func (f *FakeDockerClient) ListImages(opts docker.ListImagesOptions) ([]docker.APIImages, error) {
+	return f.Images, f.Err
+}
+
+func (f *FakeDockerClient) RemoveImage(image string) error {
+	f.RemovedImages.Insert(image)
+	return f.Err
+}
+
 // FakeDockerPuller is a stub implementation of DockerPuller.
 type FakeDockerPuller struct {
 	sync.Mutex

pkg/kubelet/kubelet.go

@@ -109,6 +109,11 @@ type Kubelet struct {
 	dockerIDToRef map[dockertools.DockerID]*api.ObjectReference
 	refLock       sync.RWMutex
 
+	// Tracks active pulls.  Needed to protect image garbage collection
+	// See: https://github.com/docker/docker/issues/8926 for details
+	// TODO: Remove this when (if?) that issue is fixed.
+	pullLock sync.RWMutex
+
 	// Optional, no events will be sent without it
 	etcdClient tools.EtcdClient
 	// Optional, defaults to simple implementaiton
@@ -203,6 +208,36 @@ func (kl *Kubelet) purgeOldest(ids []string) error {
 	return nil
 }
 
+func (kl *Kubelet) GarbageCollectLoop() {
+	util.Forever(func() {
+		if err := kl.GarbageCollectContainers(); err != nil {
+			glog.Errorf("Garbage collect failed: %v", err)
+		}
+		if err := kl.GarbageCollectImages(); err != nil {
+			glog.Errorf("Garbage collect images failed: %v", err)
+		}
+	}, time.Minute*1)
+}
+
+func (kl *Kubelet) getUnusedImages() ([]string, error) {
+	kl.pullLock.Lock()
+	defer kl.pullLock.Unlock()
+	return dockertools.GetUnusedImages(kl.dockerClient)
+}
+
+func (kl *Kubelet) GarbageCollectImages() error {
+	images, err := kl.getUnusedImages()
+	if err != nil {
+		return err
+	}
+	for ix := range images {
+		if err := kl.dockerClient.RemoveImage(images[ix]); err != nil {
+			glog.Errorf("Failed to remove image: %s (%v)", images[ix], err)
+		}
+	}
+	return nil
+}
+
 // TODO: Also enforce a maximum total number of containers.
 func (kl *Kubelet) GarbageCollectContainers() error {
 	if kl.maxContainerCount == 0 {
@@ -607,10 +642,7 @@ func (kl *Kubelet) createNetworkContainer(pod *api.BoundPod) (dockertools.Docker
 		return "", err
 	}
 	if !ok {
-		if err := kl.dockerPuller.Pull(container.Image); err != nil {
-			if ref != nil {
-				record.Eventf(ref, "failed", "failed", "Failed to pull image %s", container.Image)
-			}
+		if err := kl.pullImage(container.Image, ref); err != nil {
 			return "", err
 		}
 	}
@@ -620,6 +652,18 @@ func (kl *Kubelet) createNetworkContainer(pod *api.BoundPod) (dockertools.Docker
 	return kl.runContainer(pod, container, nil, "")
 }
 
+func (kl *Kubelet) pullImage(img string, ref *api.ObjectReference) error {
+	kl.pullLock.RLock()
+	defer kl.pullLock.RUnlock()
+	if err := kl.dockerPuller.Pull(img); err != nil {
+		if ref != nil {
+			record.Eventf(ref, "failed", "failed", "Failed to pull image %s", img)
+		}
+		return err
+	}
+	return nil
+}
+
 // Kill all containers in a pod.  Returns the number of containers deleted and an error if one occurs.
 func (kl *Kubelet) killContainersInPod(pod *api.BoundPod, dockerContainers dockertools.DockerContainers) (int, error) {
 	podFullName := GetPodFullName(pod)

pkg/kubelet/kubelet_test.go

@@ -47,7 +47,9 @@ func init() {
 
 func newTestKubelet(t *testing.T) (*Kubelet, *tools.FakeEtcdClient, *dockertools.FakeDockerClient) {
 	fakeEtcdClient := tools.NewFakeEtcdClient(t)
-	fakeDocker := &dockertools.FakeDockerClient{}
+	fakeDocker := &dockertools.FakeDockerClient{
+		RemovedImages: util.StringSet{},
+	}
 
 	kubelet := &Kubelet{}
 	kubelet.dockerClient = fakeDocker
@@ -1698,3 +1700,26 @@ func TestSyncPodsWithPullPolicy(t *testing.T) {
 	}
 	fakeDocker.Unlock()
 }
+
+func TestGarbageCollectImages(t *testing.T) {
+	kubelet, _, fakeDocker := newTestKubelet(t)
+
+	fakeDocker.Images = []docker.APIImages{
+		{
+			ID: "foo",
+		},
+		{
+			ID: "bar",
+		},
+	}
+
+	if err := kubelet.GarbageCollectImages(); err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	if len(fakeDocker.RemovedImages) != 2 ||
+		!fakeDocker.RemovedImages.Has("foo") ||
+		!fakeDocker.RemovedImages.Has("bar") {
+		t.Errorf("unexpected images removed: %v", fakeDocker.RemovedImages)
+	}
+}

pkg/kubelet/util.go

@@ -24,7 +24,6 @@ import (
 	"path"
 	"strconv"
 	"strings"
-	"time"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/capabilities"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/client"
@@ -78,18 +77,6 @@ func ConnectToDockerOrDie(dockerEndpoint string) *docker.Client {
 }
 
 // TODO: move this into the kubelet itself
-func GarbageCollectLoop(k *Kubelet) {
-	func() {
-		util.Forever(func() {
-			err := k.GarbageCollectContainers()
-			if err != nil {
-				glog.Errorf("Garbage collect failed: %v", err)
-			}
-		}, time.Minute*1)
-	}()
-}
-
-// TODO: move this into the kubelet itself
 func MonitorCAdvisor(k *Kubelet, cp uint) {
 	defer util.HandleCrash()
 	// TODO: Monitor this connection, reconnect if needed?

pkg/standalone/standalone.go

@@ -269,7 +269,7 @@ func createAndInitKubelet(kc *KubeletConfig, pc *config.PodConfig) *kubelet.Kube
 
 	k.BirthCry()
 
-	go kubelet.GarbageCollectLoop(k)
+	go k.GarbageCollectLoop()
 	go kubelet.MonitorCAdvisor(k, kc.CAdvisorPort)
 	kubelet.InitHealthChecking(k)