Add tests for process cleaner

exec-agent/src/main.go

@@ -48,6 +48,9 @@ var (
 
 	authEnabled                      bool
 	tokensExpirationTimeoutInMinutes uint
+
+	processCleanupThresholdInMinutes int
+	processCleanupPeriodInMinutes    int
 )
 
 func init() {
@@ -107,24 +110,24 @@ func init() {
 		"how much time machine tokens stay in cache(if auth is enabled)",
 	)
 
-	// activity tracking
+	// terminal configuration
 	flag.BoolVar(
 		&term.ActivityTrackingEnabled,
 		"enable-activity-tracking",
 		false,
 		"whether workspace master will be notified about terminal activity",
 	)
 
-	// process configuration
+	// process executor configuration
 	flag.IntVar(
-		&process.CleanupPeriodInMinutes,
+		&processCleanupPeriodInMinutes,
 		"process-cleanup-period",
-		5,
+		-1,
 		"how often processs cleanup job will be executed(in minutes)",
 	)
-	flag.IntVar(&process.CleanupThresholdInMinutes,
+	flag.IntVar(&processCleanupThresholdInMinutes,
 		"process-cleanup-threshold",
-		60,
+		-1,
 		`how much time will dead and unused process stay(in minutes),
 	if -1 passed then processes won't be cleaned at all. Please note that the time
 	of real cleanup is between configured threshold and threshold + process-cleanup-period.`,
@@ -160,8 +163,10 @@ func main() {
 	}
 	fmt.Println("  Process executor")
 	fmt.Printf("    - Logs dir: %s\n", process.LogsDir)
-	fmt.Printf("    - Cleanup job period: %dm\n", process.CleanupPeriodInMinutes)
-	fmt.Printf("    - Not used & dead processes stay for: %dm\n", process.CleanupThresholdInMinutes)
+	if processCleanupPeriodInMinutes > 0 {
+		fmt.Printf("    - Cleanup job period: %dm\n", processCleanupPeriodInMinutes)
+		fmt.Printf("    - Not used & dead processes stay for: %dm\n", processCleanupThresholdInMinutes)
+	}
 	if authEnabled || term.ActivityTrackingEnabled {
 		fmt.Println("  Workspace master server")
 		fmt.Printf("    - API endpoint: %s\n", apiEndpoint)
@@ -170,11 +175,25 @@ func main() {
 
 	term.ApiEndpoint = apiEndpoint
 
-	// cleanup logs dir
+	// process configuration
 	if err := os.RemoveAll(process.LogsDir); err != nil {
 		log.Fatal(err)
 	}
 
+	if processCleanupPeriodInMinutes > 0 {
+		if processCleanupThresholdInMinutes < 0 {
+			log.Fatal("Expected process cleanup threshold to be non negative value")
+		}
+		cleaner := process.NewCleaner(processCleanupPeriodInMinutes, processCleanupThresholdInMinutes)
+		cleaner.CleanPeriodically()
+	}
+
+	// terminal configuration
+	if term.ActivityTrackingEnabled {
+		go term.Activity.StartTracking()
+	}
+
+	// register routes and http handlers
 	router := httprouter.New()
 	router.NotFound = http.FileServer(http.Dir(staticDir))
 
@@ -201,13 +220,6 @@ func main() {
 		}
 	}
 
-	if process.CleanupThresholdInMinutes > 0 {
-		go process.CleanPeriodically()
-	}
-	if term.ActivityTrackingEnabled {
-		go term.Activity.StartTracking()
-	}
-
 	var handler http.Handler = router
 
 	// required authentication for all the requests, if it is configured

exec-agent/src/process/process.go

@@ -14,14 +14,14 @@ package process
 import (
 	"errors"
 	"fmt"
-	"github.com/eclipse/che/exec-agent/rpc"
-	"log"
 	"os"
 	"os/exec"
 	"sync"
 	"sync/atomic"
 	"syscall"
 	"time"
+
+	"github.com/eclipse/che/exec-agent/rpc"
 )
 
 const (
@@ -37,12 +37,10 @@ const (
 )
 
 var (
-	prevPid                   uint64 = 0
-	processes                        = &processesMap{items: make(map[uint64]*MachineProcess)}
-	logsDist                         = NewLogsDistributor()
-	LogsDir                   string
-	CleanupPeriodInMinutes    int
-	CleanupThresholdInMinutes int
+	prevPid   uint64 = 0
+	processes        = &processesMap{items: make(map[uint64]*MachineProcess)}
+	logsDist         = NewLogsDistributor()
+	LogsDir   string
 )
 
 type Command struct {
@@ -483,29 +481,3 @@ func notAlive(pid uint64) *NotAliveError {
 		Pid:   pid,
 	}
 }
-
-func CleanPeriodically() {
-	ticker := time.NewTicker(time.Duration(CleanupPeriodInMinutes) * time.Minute)
-	defer ticker.Stop()
-	for range ticker.C {
-		CleanOnce(CleanupThresholdInMinutes)
-	}
-}
-
-func CleanOnce(thresholdMinutes int) {
-	deadPoint := time.Now().Add(-time.Duration(thresholdMinutes) * time.Minute)
-	processes.Lock()
-	for _, mp := range processes.items {
-		mp.lastUsedLock.RLock()
-		if !mp.Alive && mp.lastUsed.Before(deadPoint) {
-			delete(processes.items, mp.Pid)
-			if err := os.Remove(mp.logfileName); err != nil {
-				if !os.IsNotExist(err) {
-					log.Printf("Couldn't remove process logs file, '%s'", mp.logfileName)
-				}
-			}
-		}
-		mp.lastUsedLock.RUnlock()
-	}
-	processes.Unlock()
-}

exec-agent/src/process/process_cleaner.go

@@ -0,0 +1,45 @@
+package process
+
+import (
+	"log"
+	"os"
+	"time"
+)
+
+type Cleaner struct {
+	CleanupPeriod    time.Duration
+	CleanupThreshold time.Duration
+}
+
+func NewCleaner(period int, threshold int) *Cleaner {
+	return &Cleaner{
+		time.Duration(period) * time.Minute,
+		time.Duration(threshold) * time.Minute,
+	}
+}
+
+func (c *Cleaner) CleanPeriodically() {
+	ticker := time.NewTicker(c.CleanupPeriod)
+	defer ticker.Stop()
+	for range ticker.C {
+		c.CleanOnce()
+	}
+}
+
+func (pc *Cleaner) CleanOnce() {
+	deadPoint := time.Now().Add(-pc.CleanupThreshold)
+	processes.Lock()
+	for _, mp := range processes.items {
+		mp.lastUsedLock.RLock()
+		if !mp.Alive && mp.lastUsed.Before(deadPoint) {
+			delete(processes.items, mp.Pid)
+			if err := os.Remove(mp.logfileName); err != nil {
+				if !os.IsNotExist(err) {
+					log.Printf("Couldn't remove process logs file, '%s'", mp.logfileName)
+				}
+			}
+		}
+		mp.lastUsedLock.RUnlock()
+	}
+	processes.Unlock()
+}

exec-agent/src/process/process_cleaner_test.go

@@ -0,0 +1,49 @@
+package process_test
+
+import (
+	"testing"
+
+	"github.com/eclipse/che/exec-agent/process"
+	"time"
+)
+
+func TestCleanWithZeroThreshold(t *testing.T) {
+	p := startAndWaitTestProcess(testCmd, t)
+	defer cleanupLogsDir()
+
+	process.NewCleaner(0, 0).CleanOnce()
+
+	_, err := process.Get(p.Pid)
+	if err == nil {
+		t.Fatal("Must not exist")
+	}
+	if _, ok := err.(*process.NoProcessError); !ok {
+		t.Fatal(err)
+	}
+}
+
+func TestCleansOnlyUnusedProcesses(t *testing.T) {
+	p1 := startAndWaitTestProcess(testCmd, t)
+	p2 := startAndWaitTestProcess(testCmd, t)
+
+	time.Sleep(500 * time.Millisecond)
+
+	// use one of the processes, so it is used now
+	process.Get(p1.Pid)
+
+	// cleanup immediately
+	(&process.Cleaner{CleanupPeriod: 0, CleanupThreshold: 500 * time.Millisecond}).CleanOnce()
+
+	_, err1 := process.Get(p1.Pid)
+	_, err2 := process.Get(p2.Pid)
+
+	// process 1 must be cleaned
+	if err1 != nil {
+		t.Fatalf("Expected process 2 to exist, but got an error: %s", err1.Error())
+	}
+
+	// process 2 must exist
+	if _, ok := err2.(*process.NoProcessError); !ok {
+		t.Fatal("Expected process 2 to be cleaned")
+	}
+}

exec-agent/src/process/process_test.go

@@ -196,19 +196,6 @@ func TestReadProcessLogs(t *testing.T) {
 	}
 }
 
-func TestCleanOnce(t *testing.T) {
-	p := startAndWaitTestProcess(testCmd, t)
-	defer cleanupLogsDir()
-	process.CleanOnce(0)
-	_, err := process.Get(p.Pid)
-	if err == nil {
-		t.Fatal("Must not exist")
-	}
-	if _, ok := err.(*process.NoProcessError); !ok {
-		t.Fatal(err)
-	}
-}
-
 func startAndWaitTestProcess(cmd string, t *testing.T) process.MachineProcess {
 	process.LogsDir = TmpFile()
 	events := make(chan *rpc.Event)