make sleep interuptable

pkg/proc/basejob.go

@@ -61,6 +61,10 @@ func (job *baseJob) Signal(sig os.Signal) {
  )
 }
 
+func (job *baseJob) Reset() {
+ job.phase = JobPhase{}
+}
+
 func (job *baseJob) MarkForRestart() {
  job.restart = true
 }

pkg/proc/job_common.go

@@ -76,6 +76,11 @@ func (job *CommonJob) Run(ctx context.Context, _ chan<- error) error {
  }()
 
  for { // restart failed jobs as long mittnite is running
+ if job.stop {
+ return nil
+ }
+
+ job.ctx, job.interrupt = context.WithCancel(context.Background())
  err := job.startOnce(ctx, p)
  switch err {
  case nil:
@@ -98,7 +103,7 @@ func (job *CommonJob) Run(ctx context.Context, _ chan<- error) error {
  if maxAttempts == -1 || attempts < maxAttempts {
  currBackOff := backOff
  backOffRetries++
- if backOffRetries > maxBackOffRetries {
+ if backOffRetries >= maxBackOffRetries {
  backOff = calculateNextBackOff(currBackOff, maxBackOff)
  backOffRetries = 0
  }
@@ -110,7 +115,7 @@ func (job *CommonJob) Run(ctx context.Context, _ chan<- error) error {
  WithField("job.nextRestartIn", currBackOff.String()).
  Info("remaining attempts")
 
- time.Sleep(currBackOff)
+ job.crashLoopSleep(currBackOff)
  continue
  }
 
@@ -202,11 +207,13 @@ func (job *CommonJob) IsRunning() bool {
 func (job *CommonJob) Restart() {
  job.restart = true
  job.SignalAll(syscall.SIGTERM)
+ job.interrupt()
 }
 
 func (job *CommonJob) Stop() {
  job.stop = true
  job.SignalAll(syscall.SIGTERM)
+ job.interrupt()
 }
 
 func (job *CommonJob) Status() *CommonJobStatus {
@@ -240,3 +247,22 @@ func (job *CommonJob) executeWatchCommand(watchCmd *config.WatchCommand) error {
  Info("executing watch command")
  return cmd.Run()
 }
+
+func (job *CommonJob) crashLoopSleep(duration time.Duration) {
+ timeout := make(chan bool)
+
+ go func() {
+ defer close(timeout)
+ <-time.After(duration)
+ timeout <- true
+ }()
+
+ for {
+ select {
+ case <-timeout:
+ return
+ case <-job.ctx.Done():
+ return
+ }
+ }
+}

pkg/proc/runner.go

@@ -145,7 +145,7 @@ func (r *Runner) Init() error {
 
 func (r *Runner) exec() {
  for i := range r.jobs {
- r.startJob(r.jobs[i])
+ r.startJob(r.jobs[i], JobPhaseReasonUnknown)
  }
 }
 
@@ -155,7 +155,7 @@ func (r *Runner) jobExistsAndIsControllable(job *CommonJob) bool {
 
 func (r *Runner) addAndStartJob(job Job) {
  r.addJobIfNotExists(job)
- r.startJob(job)
+ r.startJob(job, job.GetPhase().Reason)
 }
 
 func (r *Runner) addJobIfNotExists(job Job) {
@@ -167,9 +167,10 @@ func (r *Runner) addJobIfNotExists(job Job) {
  r.jobs = append(r.jobs, job)
 }
 
-func (r *Runner) startJob(job Job) {
+func (r *Runner) startJob(job Job, initialPhase JobPhaseReason) {
+ job.GetPhase().Set(initialPhase)
  phase := job.GetPhase()
- if phase.Is(JobPhaseReasonStopped) || phase.Is(JobPhaseReasonFailed) {
+ if phase.Is(JobPhaseReasonStopped) || phase.Is(JobPhaseReasonFailed) || phase.Is(JobPhaseReasonCrashLooping) {
  return
  }
 

pkg/proc/runner_api_v1.go

@@ -63,15 +63,15 @@ func (r *Runner) apiV1JobMiddleware(next http.Handler) http.Handler {
 func (r *Runner) apiV1StartJob(writer http.ResponseWriter, req *http.Request) {
  job := req.Context().Value(contextKeyJob).(*CommonJob)
  if !job.IsRunning() {
- r.startJob(job)
+ r.startJob(job, JobPhaseReasonUnknown)
  }
  writer.WriteHeader(http.StatusOK)
 }
 
 func (r *Runner) apiV1RestartJob(writer http.ResponseWriter, req *http.Request) {
  job := req.Context().Value(contextKeyJob).(*CommonJob)
  if !job.IsRunning() {
- r.startJob(job)
+ r.startJob(job, JobPhaseReasonUnknown)
  } else {
  job.Restart()
  }

pkg/proc/types.go

@@ -50,6 +50,9 @@ func NewApi(listenAddress string) *Api {
 type baseJob struct {
  Config *config.BaseJobConfig
 
+ ctx context.Context
+ interrupt context.CancelFunc
+
  cmd *exec.Cmd
  restart bool
  stop bool
@@ -97,6 +100,7 @@ type Job interface {
  Init()
  Run(context.Context, chan<- error) error
  Watch()
+ Reset()
 
  GetPhase() *JobPhase
  GetName() string