Always add jobs to periodic tracker

nomad/fsm.go

@@ -208,15 +208,22 @@ func (n *nomadFSM) applyUpsertJob(buf []byte, index uint64) interface{} {
 		return err
 	}
 
-	// If it is periodic, insert it into the periodic runner and record the
-	// time it was inserted.
-	if req.Job.IsPeriodic() {
-		if err := n.periodicDispatcher.Add(req.Job); err != nil {
-			n.logger.Printf("[ERR] nomad.fsm: periodicDispatcher.Add failed: %v", err)
-			return err
-		}
+	// We always add the job to the periodic dispatcher because there is the
+	// possibility that the periodic spec was removed and then we should stop
+	// tracking it.
+	if err := n.periodicDispatcher.Add(req.Job); err != nil {
+		n.logger.Printf("[ERR] nomad.fsm: periodicDispatcher.Add failed: %v", err)
+		return err
+	}
 
-		// Record the insertion time as a launch.
+	// If it is periodic, record the time it was inserted. This is necessary for
+	// recovering during leader election. It is possible that from the time it
+	// is added to when it was suppose to launch, leader election occurs and the
+	// job was not launched. In this case, we use the insertion time to
+	// determine if a launch was missed.
+	if req.Job.IsPeriodic() {
+		// Record the insertion time as a launch. We overload the launch table
+		// such that the first entry is the insertion time.
 		launch := &structs.PeriodicLaunch{ID: req.Job.ID, Launch: time.Now()}
 		if err := n.state.UpsertPeriodicLaunch(index, launch); err != nil {
 			n.logger.Printf("[ERR] nomad.fsm: UpsertPeriodicLaunch failed: %v", err)

nomad/periodic_test.go

@@ -128,6 +128,31 @@ func TestPeriodicDispatch_Add_UpdateJob(t *testing.T) {
 	}
 }
 
+func TestPeriodicDispatch_Add_RemoveJob(t *testing.T) {
+	t.Parallel()
+	p, _ := testPeriodicDispatcher()
+	job := mock.PeriodicJob()
+	if err := p.Add(job); err != nil {
+		t.Fatalf("Add failed %v", err)
+	}
+
+	tracked := p.Tracked()
+	if len(tracked) != 1 {
+		t.Fatalf("Add didn't track the job: %v", tracked)
+	}
+
+	// Update the job to be non-periodic and add it again.
+	job.Periodic = nil
+	if err := p.Add(job); err != nil {
+		t.Fatalf("Add failed %v", err)
+	}
+
+	tracked = p.Tracked()
+	if len(tracked) != 0 {
+		t.Fatalf("Add didn't remove: %v", tracked)
+	}
+}
+
 func TestPeriodicDispatch_Add_TriggersUpdate(t *testing.T) {
 	t.Parallel()
 	p, m := testPeriodicDispatcher()