Backport of Multiple instances of a periodic job are run simultaneously, when prohibit_overlap is true into release/1.4.x

nomad/leader.go

@@ -768,7 +768,8 @@ func (s *Server) restorePeriodicDispatcher() error {
 			logger.Error("force run of periodic job failed", "job", job.NamespacedID(), "error", err)
 			return fmt.Errorf("force run of periodic job %q failed: %v", job.NamespacedID(), err)
 		}
-		logger.Debug("periodic job force runned during leadership establishment", "job", job.NamespacedID())
+
+		logger.Debug("periodic job force ran during leadership establishment", "job", job.NamespacedID())
 	}
 
 	return nil

nomad/leader_test.go

@@ -438,13 +438,34 @@ func TestLeader_PeriodicDispatcher_Restore_NoEvals(t *testing.T) {
 	}
 }
 
+type mockJobEvalDispatcher struct {
+	called, children bool
+	evalToReturn     *structs.Evaluation
+}
+
+func (mjed *mockJobEvalDispatcher) DispatchJob(job *structs.Job) (*structs.Evaluation, error) {
+	mjed.called = true
+	return mjed.evalToReturn, nil
+}
+
+func (mjed *mockJobEvalDispatcher) RunningChildren(job *structs.Job) (bool, error) {
+	return mjed.children, nil
+}
+
+func testPeriodicJobWithOverlapEnabled(times ...time.Time) *structs.Job {
+	job := testPeriodicJob(times...)
+	job.Periodic.ProhibitOverlap = true
+	return job
+}
+
 func TestLeader_PeriodicDispatcher_Restore_Evals(t *testing.T) {
 	ci.Parallel(t)
 
 	s1, cleanupS1 := TestServer(t, func(c *Config) {
 		c.NumSchedulers = 0
 	})
 	defer cleanupS1()
+
 	testutil.WaitForLeader(t, s1.RPC)
 
 	// Inject a periodic job that triggered once in the past, should trigger now
@@ -464,6 +485,158 @@ func TestLeader_PeriodicDispatcher_Restore_Evals(t *testing.T) {
 		t.Fatalf("err: %v", err)
 	}
 
+	// Create an eval for the past launch.
+	eval, _ := s1.periodicDispatcher.createEval(job, past)
+
+	md := &mockJobEvalDispatcher{
+		children:     false,
+		evalToReturn: eval,
+	}
+
+	s1.periodicDispatcher.dispatcher = md
+
+	// Flush the periodic dispatcher, ensuring that no evals will be created.
+	s1.periodicDispatcher.SetEnabled(false)
+
+	// Sleep till after the job should have been launched.
+	time.Sleep(3 * time.Second)
+
+	// Restore the periodic dispatcher.
+	s1.periodicDispatcher.SetEnabled(true)
+
+	s1.restorePeriodicDispatcher()
+
+	// Ensure the job is tracked.
+	tuple := structs.NamespacedID{
+		ID:        job.ID,
+		Namespace: job.Namespace,
+	}
+	if _, tracked := s1.periodicDispatcher.tracked[tuple]; !tracked {
+		t.Fatalf("periodic job not restored")
+	}
+
+	// Check that an eval was made.
+	ws := memdb.NewWatchSet()
+	last, err := s1.fsm.State().PeriodicLaunchByID(ws, job.Namespace, job.ID)
+	if err != nil || last == nil {
+		t.Fatalf("failed to get periodic launch time: %v", err)
+	}
+	if last.Launch == past {
+		t.Fatalf("restorePeriodicDispatcher did not force launch")
+	}
+
+	if md.called != true {
+		t.Fatalf("failed to force job evaluation")
+	}
+}
+
+func TestLeader_PeriodicDispatcher_No_Overlaps_No_Running_Job(t *testing.T) {
+	ci.Parallel(t)
+
+	s1, cleanupS1 := TestServer(t, func(c *Config) {
+		c.NumSchedulers = 0
+	})
+	defer cleanupS1()
+	testutil.WaitForLeader(t, s1.RPC)
+
+	// Inject a periodic job that triggered once in the past, should trigger now
+	// and once in the future.
+	now := time.Now()
+	past := now.Add(-1 * time.Second)
+	future := now.Add(10 * time.Second)
+
+	job := testPeriodicJobWithOverlapEnabled(past, now, future)
+	req := structs.JobRegisterRequest{
+		Job: job,
+		WriteRequest: structs.WriteRequest{
+			Namespace: job.Namespace,
+		},
+	}
+	_, _, err := s1.raftApply(structs.JobRegisterRequestType, req)
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+
+	// Create an eval for the past launch.
+	eval, _ := s1.periodicDispatcher.createEval(job, past)
+
+	md := &mockJobEvalDispatcher{
+		children:     false,
+		evalToReturn: eval,
+	}
+
+	s1.periodicDispatcher.dispatcher = md
+
+	// Flush the periodic dispatcher, ensuring that no evals will be created.
+	s1.periodicDispatcher.SetEnabled(false)
+
+	// Sleep till after the job should have been launched.
+	time.Sleep(3 * time.Second)
+
+	// Restore the periodic dispatcher.
+	s1.periodicDispatcher.SetEnabled(true)
+	s1.restorePeriodicDispatcher()
+
+	// Ensure the job is tracked.
+	tuple := structs.NamespacedID{
+		ID:        job.ID,
+		Namespace: job.Namespace,
+	}
+	if _, tracked := s1.periodicDispatcher.tracked[tuple]; !tracked {
+		t.Fatalf("periodic job not restored")
+	}
+
+	// Check that an eval was made.
+	ws := memdb.NewWatchSet()
+	last, err := s1.fsm.State().PeriodicLaunchByID(ws, job.Namespace, job.ID)
+	if err != nil || last == nil {
+		t.Fatalf("failed to get periodic launch time: %v", err)
+	}
+
+	if last.Launch == past {
+		t.Fatalf("restorePeriodicDispatcher did not force launch")
+	}
+
+	if md.called != true {
+		t.Fatalf("failed to force job evaluation")
+	}
+}
+
+func TestLeader_PeriodicDispatcher_No_Overlaps_Running_Job(t *testing.T) {
+	ci.Parallel(t)
+
+	s1, cleanupS1 := TestServer(t, func(c *Config) {
+		c.NumSchedulers = 0
+	})
+	defer cleanupS1()
+	testutil.WaitForLeader(t, s1.RPC)
+
+	md := &mockJobEvalDispatcher{
+		children:     true,
+		evalToReturn: nil,
+	}
+
+	s1.periodicDispatcher.dispatcher = md
+
+	// Inject a periodic job that triggered once in the past, should trigger now
+	// and once in the future.
+	now := time.Now()
+	past := now.Add(-1 * time.Second)
+	future := now.Add(10 * time.Second)
+	job := testPeriodicJob(past, now, future)
+
+	req := structs.JobRegisterRequest{
+		Job:            job,
+		PolicyOverride: true,
+		WriteRequest: structs.WriteRequest{
+			Namespace: job.Namespace,
+		},
+	}
+	_, _, err := s1.raftApply(structs.JobRegisterRequestType, req)
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+
 	// Create an eval for the past launch.
 	s1.periodicDispatcher.createEval(job, past)
 
@@ -492,9 +665,14 @@ func TestLeader_PeriodicDispatcher_Restore_Evals(t *testing.T) {
 	if err != nil || last == nil {
 		t.Fatalf("failed to get periodic launch time: %v", err)
 	}
+
 	if last.Launch == past {
 		t.Fatalf("restorePeriodicDispatcher did not force launch")
 	}
+
+	if md.called != false {
+		t.Fatalf("job evaluation forced when job is already running")
+	}
 }
 
 func TestLeader_PeriodicDispatch(t *testing.T) {

nomad/periodic.go

@@ -299,10 +299,10 @@ func (p *PeriodicDispatch) removeLocked(jobID structs.NamespacedID) error {
 // subsequent eval.
 func (p *PeriodicDispatch) ForceEval(namespace, jobID string) (*structs.Evaluation, error) {
 	p.l.Lock()
+	defer p.l.Unlock()
 
 	// Do nothing if not enabled
 	if !p.enabled {
-		p.l.Unlock()
 		return nil, fmt.Errorf("periodic dispatch disabled")
 	}
 
@@ -312,11 +312,9 @@ func (p *PeriodicDispatch) ForceEval(namespace, jobID string) (*structs.Evaluati
 	}
 	job, tracked := p.tracked[tuple]
 	if !tracked {
-		p.l.Unlock()
 		return nil, fmt.Errorf("can't force run non-tracked job %q (%s)", jobID, namespace)
 	}
 
-	p.l.Unlock()
 	return p.createEval(job, time.Now().In(job.Periodic.GetLocation()))
 }
 
@@ -356,6 +354,7 @@ func (p *PeriodicDispatch) run(ctx context.Context, updateCh <-chan struct{}) {
 // based on the passed launch time.
 func (p *PeriodicDispatch) dispatch(job *structs.Job, launchTime time.Time) {
 	p.l.Lock()
+	defer p.l.Unlock()
 
 	nextLaunch, err := job.Periodic.Next(launchTime)
 	if err != nil {
@@ -370,19 +369,16 @@ func (p *PeriodicDispatch) dispatch(job *structs.Job, launchTime time.Time) {
 		running, err := p.dispatcher.RunningChildren(job)
 		if err != nil {
 			p.logger.Error("failed to determine if periodic job has running children", "job", job.NamespacedID(), "error", err)
-			p.l.Unlock()
 			return
 		}
 
 		if running {
 			p.logger.Debug("skipping launch of periodic job because job prohibits overlap", "job", job.NamespacedID())
-			p.l.Unlock()
 			return
 		}
 	}
 
 	p.logger.Debug(" launching job", "job", job.NamespacedID(), "launch_time", launchTime)
-	p.l.Unlock()
 	p.createEval(job, launchTime)
 }