Merge pull request #1129 from Ace-Tang/gc_exec

feature: gc unused exec processes

Author: allencloud

daemon/mgr/container.go

@@ -150,6 +150,8 @@ func NewContainerManager(ctx context.Context, store *meta.Store, cli ctrd.APICli
 	mgr.Client.SetExitHooks(mgr.exitedAndRelease)
 	mgr.Client.SetExecExitHooks(mgr.execExitedAndRelease)
 
+	go mgr.execProcessGC()
+
 	return mgr, mgr.Restore(ctx)
 }
 
@@ -1742,3 +1744,32 @@ func (mgr *ContainerManager) setBaseFS(ctx context.Context, meta *ContainerMeta,
 	// io.containerd.runtime.v1.linux as a const used by runc
 	meta.BaseFS = filepath.Join(mgr.Config.HomeDir, "containerd/state", "io.containerd.runtime.v1.linux", namespaces.Default, info.Name, "rootfs")
 }
+
+// execProcessGC cleans unused exec processes config every 5 minutes.
+func (mgr *ContainerManager) execProcessGC() {
+	for range time.Tick(time.Duration(GCExecProcessTick) * time.Minute) {
+		execProcesses := mgr.ExecProcesses.Values()
+		cleaned := 0
+
+		for id, v := range execProcesses {
+			execConfig, ok := v.(*ContainerExecConfig)
+			if !ok {
+				logrus.Warnf("get incorrect exec config: %v", v)
+				continue
+			}
+			// if unused exec processes are found, we will tag them, and clean
+			// them in next loop, so that we can ensure exec process can get
+			// correct exit code.
+			if execConfig.WaitForClean {
+				cleaned++
+				mgr.ExecProcesses.Remove(id)
+			} else if !execConfig.Running {
+				execConfig.WaitForClean = true
+			}
+		}
+
+		if cleaned > 0 {
+			logrus.Debugf("clean %d unused exec process", cleaned)
+		}
+	}
+}

daemon/mgr/container_types.go

@@ -15,6 +15,12 @@ import (
 	"github.com/opencontainers/image-spec/specs-go/v1"
 )
 
+var (
+	// GCExecProcessTick is the time interval to trigger gc unused exec config,
+	// time unit is minute.
+	GCExecProcessTick = 5
+)
+
 const (
 	// DefaultStopTimeout is the timeout (in seconds) for the syscall signal used to stop a container.
 	DefaultStopTimeout = 10
@@ -43,6 +49,9 @@ type ContainerExecConfig struct {
 
 	// Error represents the exec process response error.
 	Error error
+
+	// WaitForClean means exec process can be removed.
+	WaitForClean bool
 }
 
 // AttachConfig wraps some infos of attaching.

pkg/collect/safe_map.go

@@ -25,10 +25,28 @@ func (m *SafeMap) Get(k string) *Value {
 	return &Value{v, ok}
 }
 
+// Values returns all key-values stored in map
+func (m *SafeMap) Values() map[string]interface{} {
+	m.RLock()
+	defer m.RUnlock()
+
+	nmap := make(map[string]interface{})
+	for k, v := range m.inner {
+		nmap[k] = v
+	}
+
+	return nmap
+}
+
 // Put stores a key-value pair into inner map safely.
 func (m *SafeMap) Put(k string, v interface{}) {
 	m.Lock()
 	defer m.Unlock()
+
+	if m.inner == nil {
+		return
+	}
+
 	m.inner[k] = v
 }
 

pkg/collect/safe_map_test.go

@@ -50,6 +50,27 @@ func TestSafeMapPutAndGet(t *testing.T) {
 	assert.Equal(t, value.data, []string{"asdfgh", "123344"})
 }
 
+// TestSafeMapDirectNew test functions should not panic.
+func TestSafeMapDirectNew(t *testing.T) {
+	assert := assert.New(t)
+	defer func() {
+		if err := recover(); err != nil {
+			t.Fatal(err)
+		}
+	}()
+
+	sm := &SafeMap{}
+	// test Put not panic
+	sm.Put("k", "v")
+
+	// test Remove not panic
+	sm.Remove("k")
+
+	// test Values not panic
+	values := sm.Values()
+	assert.Equal(values, map[string]interface{}{})
+}
+
 func TestSafeMapRemove(t *testing.T) {
 	safeMap := NewSafeMap()
 	assert.Equal(t, len(safeMap.inner), 0)