Garbage collecting images in the Kubelet.

Integrated the imageManager into the Kubelet and applies the garbage
collection policy every 5 minutes. The default policy allows up to 90%
disk usage, after which images are garbage collected to bring limit back
down to 80%.

Fixes #157.

cmd/kubelet/app/server.go

@@ -76,6 +76,8 @@ type KubeletServer struct {
 	ClusterDNS                     util.IP
 	ReallyCrashForTesting          bool
 	StreamingConnectionIdleTimeout time.Duration
+	ImageGCHighThresholdPercent    int
+	ImageGCLowThresholdPercent     int
 }
 
 // NewKubeletServer will create a new KubeletServer with default values.
@@ -98,6 +100,8 @@ func NewKubeletServer() *KubeletServer {
 		CadvisorPort:                4194,
 		OOMScoreAdj:                 -900,
 		MasterServiceNamespace:      api.NamespaceDefault,
+		ImageGCHighThresholdPercent: 90,
+		ImageGCLowThresholdPercent:  80,
 	}
 }
 
@@ -133,6 +137,8 @@ func (s *KubeletServer) AddFlags(fs *pflag.FlagSet) {
 	fs.Var(&s.ClusterDNS, "cluster_dns", "IP address for a cluster DNS server.  If set, kubelet will configure all containers to use this for DNS resolution in addition to the host's DNS servers")
 	fs.BoolVar(&s.ReallyCrashForTesting, "really_crash_for_testing", s.ReallyCrashForTesting, "If true, crash with panics more often.")
 	fs.DurationVar(&s.StreamingConnectionIdleTimeout, "streaming_connection_idle_timeout", 0, "Maximum time a streaming connection can be idle before the connection is automatically closed.  Example: '5m'")
+	fs.IntVar(&s.ImageGCHighThresholdPercent, "image_gc_high_threshold", s.ImageGCHighThresholdPercent, "The percent of disk usage after which image garbage collection is always run. Default: 90%%")
+	fs.IntVar(&s.ImageGCLowThresholdPercent, "image_gc_low_threshold", s.ImageGCLowThresholdPercent, "The percent of disk usage before which image garbage collection is never run. Lowest disk usage to garbage collect to. Default: 80%%")
 }
 
 // Run runs the specified KubeletServer.  This should never exit.
@@ -158,6 +164,10 @@ func (s *KubeletServer) Run(_ []string) error {
 		return err
 	}
 
+	imageGCPolicy := kubelet.ImageGCPolicy{
+		HighThresholdPercent: s.ImageGCHighThresholdPercent,
+		LowThresholdPercent:  s.ImageGCLowThresholdPercent,
+	}
 	kcfg := KubeletConfig{
 		Address:                        s.Address,
 		AllowPrivileged:                s.AllowPrivileged,
@@ -187,6 +197,7 @@ func (s *KubeletServer) Run(_ []string) error {
 		MasterServiceNamespace:         s.MasterServiceNamespace,
 		VolumePlugins:                  ProbeVolumePlugins(),
 		StreamingConnectionIdleTimeout: s.StreamingConnectionIdleTimeout,
+		ImageGCPolicy:                  imageGCPolicy,
 	}
 
 	RunKubelet(&kcfg)
@@ -250,6 +261,11 @@ func SimpleRunKubelet(client *client.Client,
 	tlsOptions *kubelet.TLSOptions,
 	cadvisorInterface cadvisor.Interface,
 	configFilePath string) {
+
+	imageGCPolicy := kubelet.ImageGCPolicy{
+		HighThresholdPercent: 90,
+		LowThresholdPercent:  80,
+	}
 	kcfg := KubeletConfig{
 		KubeClient:             client,
 		DockerClient:           dockerClient,
@@ -271,6 +287,7 @@ func SimpleRunKubelet(client *client.Client,
 		TLSOptions:              tlsOptions,
 		CadvisorInterface:       cadvisorInterface,
 		ConfigFile:              configFilePath,
+		ImageGCPolicy:           imageGCPolicy,
 	}
 	RunKubelet(&kcfg)
 }
@@ -377,6 +394,7 @@ type KubeletConfig struct {
 	StreamingConnectionIdleTimeout time.Duration
 	Recorder                       record.EventRecorder
 	TLSOptions                     *kubelet.TLSOptions
+	ImageGCPolicy                  kubelet.ImageGCPolicy
 }
 
 func createAndInitKubelet(kc *KubeletConfig, pc *config.PodConfig) (*kubelet.Kubelet, error) {
@@ -416,7 +434,8 @@ func createAndInitKubelet(kc *KubeletConfig, pc *config.PodConfig) (*kubelet.Kub
 		kc.StreamingConnectionIdleTimeout,
 		kc.Recorder,
 		kc.CadvisorInterface,
-		kc.StatusUpdateFrequency)
+		kc.StatusUpdateFrequency,
+		kc.ImageGCPolicy)
 
 	if err != nil {
 		return nil, err
@@ -424,7 +443,7 @@ func createAndInitKubelet(kc *KubeletConfig, pc *config.PodConfig) (*kubelet.Kub
 
 	k.BirthCry()
 
-	go k.GarbageCollectLoop()
+	k.StartGarbageCollection()
 
 	return k, nil
 }

pkg/kubelet/image_manager.go

@@ -179,11 +179,18 @@ func (self *realImageManager) GarbageCollect() error {
 	}
 	usage := int64(fsInfo.Usage)
 	capacity := int64(fsInfo.Capacity)
-	usagePercent := int(usage * 100 / capacity)
+
+	// Check valid capacity.
+	if capacity == 0 {
+		// TODO(vmarmol): Surface event.
+		return fmt.Errorf("invalid capacity %d on device %q at mount point %q", capacity, fsInfo.Device, fsInfo.Mountpoint)
+	}
 
 	// If over the max threshold, free enough to place us at the lower threshold.
+	usagePercent := int(usage * 100 / capacity)
 	if usagePercent >= self.policy.HighThresholdPercent {
 		amountToFree := usage - (int64(self.policy.LowThresholdPercent) * capacity / 100)
+		glog.Infof("[ImageManager]: Disk usage on %q (%s) is at %d%% which is over the high threshold (%d%%). Trying to free %d bytes", fsInfo.Device, fsInfo.Mountpoint, usagePercent, self.policy.HighThresholdPercent, amountToFree)
 		freed, err := self.freeSpace(amountToFree)
 		if err != nil {
 			return err
@@ -234,6 +241,7 @@ func (self *realImageManager) freeSpace(bytesToFree int64) (int64, error) {
 		}
 
 		// Remove image. Continue despite errors.
+		glog.Infof("[ImageManager]: Removing image %q to free %d bytes", image.id, image.size)
 		err := self.dockerClient.RemoveImage(image.id)
 		if err != nil {
 			lastErr = err

pkg/kubelet/kubelet.go

@@ -121,7 +121,8 @@ func NewMainKubelet(
 	streamingConnectionIdleTimeout time.Duration,
 	recorder record.EventRecorder,
 	cadvisorInterface cadvisor.Interface,
-	statusUpdateFrequency time.Duration) (*Kubelet, error) {
+	statusUpdateFrequency time.Duration,
+	imageGCPolicy ImageGCPolicy) (*Kubelet, error) {
 	if rootDirectory == "" {
 		return nil, fmt.Errorf("invalid root directory %q", rootDirectory)
 	}
@@ -166,6 +167,10 @@ func NewMainKubelet(
 	if err != nil {
 		return nil, err
 	}
+	imageManager, err := newImageManager(dockerClient, cadvisorInterface, imageGCPolicy)
+	if err != nil {
+		return nil, fmt.Errorf("failed to initialize image manager: %v", err)
+	}
 
 	klet := &Kubelet{
 		hostname:                       hostname,
@@ -191,6 +196,7 @@ func NewMainKubelet(
 		recorder:                       recorder,
 		cadvisor:                       cadvisorInterface,
 		containerGC:                    containerGC,
+		imageManager:                   imageManager,
 	}
 
 	dockerCache, err := dockertools.NewDockerCache(dockerClient)
@@ -302,6 +308,9 @@ type Kubelet struct {
 
 	// Policy for handling garbage collection of dead containers.
 	containerGC containerGC
+
+	// Manager for images.
+	imageManager imageManager
 }
 
 // getRootDir returns the full path to the directory under which kubelet can
@@ -443,12 +452,19 @@ func (kl *Kubelet) listPodsFromDisk() ([]types.UID, error) {
 	return pods, nil
 }
 
-func (kl *Kubelet) GarbageCollectLoop() {
+// Starts garbage collection theads.
-	util.Forever(func() {
+func (kl *Kubelet) StartGarbageCollection() {
+	go util.Forever(func() {
 		if err := kl.containerGC.GarbageCollect(); err != nil {
-			glog.Errorf("Container garbage collect failed: %v", err)
+			glog.Errorf("Container garbage collection failed: %v", err)
 		}
-	}, time.Minute*1)
+	}, time.Minute)
+
+	go util.Forever(func() {
+		if err := kl.imageManager.GarbageCollect(); err != nil {
+			glog.Errorf("Image garbage collection failed: %v", err)
+		}
+	}, 5*time.Minute)
 }
 
 func (kl *Kubelet) getPodStatusFromCache(podFullName string) (api.PodStatus, bool) {