Free unused volumes in more cases

Freeing volumes is only done on scheduling ticks, but not every
scheduling tick resulted in freeing volumes. This change makes it so
that volumes are freed every time a scheduling tick happens, regardless
of whether there were any scheduled task changes.

Signed-off-by: Drew Erny <derny@mirantis.com>

Author: dperny

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