diff --git a/exec-agent/src/main.go b/exec-agent/src/main.go index 08f1c670ebf..144bdfb15e2 100644 --- a/exec-agent/src/main.go +++ b/exec-agent/src/main.go @@ -48,6 +48,9 @@ var ( authEnabled bool tokensExpirationTimeoutInMinutes uint + + processCleanupThresholdInMinutes int + processCleanupPeriodInMinutes int ) func init() { @@ -107,7 +110,7 @@ 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", @@ -115,16 +118,16 @@ func init() { "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 diff --git a/exec-agent/src/process/process.go b/exec-agent/src/process/process.go index 61aa887bd81..25075916bae 100644 --- a/exec-agent/src/process/process.go +++ b/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() -} diff --git a/exec-agent/src/process/process_cleaner.go b/exec-agent/src/process/process_cleaner.go new file mode 100644 index 00000000000..f21092bee71 --- /dev/null +++ b/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() +} diff --git a/exec-agent/src/process/process_cleaner_test.go b/exec-agent/src/process/process_cleaner_test.go new file mode 100644 index 00000000000..d3ee7e45f26 --- /dev/null +++ b/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") + } +} diff --git a/exec-agent/src/process/process_test.go b/exec-agent/src/process/process_test.go index b0f1964adeb..a535789922a 100644 --- a/exec-agent/src/process/process_test.go +++ b/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)