scheduler: Fix accounting when task ends up in multiple groups

Tasks are split between task groups based on common specs. This allows
nodes to only be ranked once per group, not once per task.

This logic doesn't work correctly because maps are marshalled in a
random order. Currently, the same task can end up in a multiple groups
(say, if it's updated multiple times, and the marshalling ends up being
different). Add a check to make sure we don't try to schedule the same
task twice within the same batch, which can cause resource accounting to
be wrong.

Note this doesn't fix the brokenness of task spec deduplication based on
marshalling the protobuf. This is a fix for the symptom that can be
backported, and I'm going to replace the marshalling stuff in a
different PR.

Signed-off-by: Aaron Lehmann <[email protected]>

manager/scheduler/scheduler.go

@@ -602,6 +602,12 @@ func (s *Scheduler) scheduleNTasksOnNodes(ctx context.Context, n int, taskGroup
 	nodeIter := 0
 	nodeCount := len(nodes)
 	for taskID, t := range taskGroup {
+		// Skip tasks which were already scheduled because they ended
+		// up in two groups at once.
+		if _, exists := schedulingDecisions[taskID]; exists {
+			continue
+		}
+
 		node := &nodes[nodeIter%nodeCount]
 
 		log.G(ctx).WithField("task.id", t.ID).Debugf("assigning to node %s", node.ID)