Merge pull request #3144 from dperny/fix-volume-scheduling

Free unused volumes in more cases

manager/scheduler/scheduler.go

@@ -488,6 +488,18 @@ func (s *Scheduler) tick(ctx context.Context) {
 }
 
 func (s *Scheduler) applySchedulingDecisions(ctx context.Context, schedulingDecisions map[string]schedulingDecision) (successful, failed []schedulingDecision) {
+	// applySchedulingDecisions is the only place where we make store
+	// transactions in the scheduler. the scheduler is responsible for freeing
+	// volumes that are no longer in use. this means that volumes should be
+	// freed in this function. sometimes, there are no scheduling decisions to
+	// be made, so we return early in the if statement below.
+	//
+	// however, in all cases, any activity that results in a tick could result
+	// in needing volumes to be freed, even if nothing new is scheduled. this
+	// freeing of volumes should always happen *after* all of the scheduling
+	// decisions have been committed, hence the defer.
+	defer s.store.Batch(s.volumes.freeVolumes)
+
 	if len(schedulingDecisions) == 0 {
 		return
 	}
@@ -619,9 +631,7 @@ func (s *Scheduler) applySchedulingDecisions(ctx context.Context, schedulingDeci
 		}
 		// finally, every time we make new scheduling decisions, take the
 		// opportunity to release volumes.
-		return batch.Update(func(tx store.Tx) error {
-			return s.volumes.freeVolumes(tx)
-		})
+		return nil
 	})
 
 	if err != nil {

manager/scheduler/volumes.go

@@ -183,24 +183,33 @@ func (vs *volumeSet) releaseVolume(volumeID, taskID string) {
 //
 // TODO(dperny): this is messy and has a lot of overhead. it should be reworked
 // to something more streamlined.
-func (vs *volumeSet) freeVolumes(tx store.Tx) error {
+func (vs *volumeSet) freeVolumes(batch *store.Batch) error {
 	for volumeID, info := range vs.volumes {
-		v := store.GetVolume(tx, volumeID)
-		if v == nil {
-			continue
-		}
+		if err := batch.Update(func(tx store.Tx) error {
+			v := store.GetVolume(tx, volumeID)
+			if v == nil {
+				return nil
+			}
 
-		changed := false
-		for _, status := range v.PublishStatus {
-			if info.nodes[status.NodeID] == 0 && status.State == api.VolumePublishStatus_PUBLISHED {
-				status.State = api.VolumePublishStatus_PENDING_NODE_UNPUBLISH
-				changed = true
+			// when we are freeing a volume, we may update more than one of the
+			// volume's PublishStatuses. this means we can't simply put the
+			// Update call inside of the if statement; we need to know if we've
+			// changed anything once we've checked *all* of the statuses.
+			changed := false
+			for _, status := range v.PublishStatus {
+				if info.nodes[status.NodeID] == 0 && status.State == api.VolumePublishStatus_PUBLISHED {
+					status.State = api.VolumePublishStatus_PENDING_NODE_UNPUBLISH
+					changed = true
+				}
 			}
-		}
-		if changed {
-			if err := store.UpdateVolume(tx, v); err != nil {
-				return err
+			if changed {
+				if err := store.UpdateVolume(tx, v); err != nil {
+					return err
+				}
 			}
+			return nil
+		}); err != nil {
+			return err
 		}
 	}
 	return nil

manager/scheduler/volumes_test.go

@@ -652,9 +652,7 @@ var _ = Describe("volumeSet", func() {
 			vs.releaseVolume(volumes[0].ID, tasks[0].ID)
 			vs.releaseVolume(allVolume.ID, tasks[0].ID)
 
-			err := s.Update(func(tx store.Tx) error {
-				return vs.freeVolumes(tx)
-			})
+			err := s.Batch(vs.freeVolumes)
 			Expect(err).ToNot(HaveOccurred())
 
 			var freshVolumes []*api.Volume