Only schedule executors on ready nodes (palantir#100)

* Only include ready nodes when scheduling executors

* Fix tests

* Fix version

* Add node ready handling on failover

Gopkg.toml

@@ -18,7 +18,7 @@ required = ["k8s.io/code-generator/cmd/client-gen"]
 
 [[constraint]]
   name = "github.com/palantir/k8s-spark-scheduler-lib"
-  version = "0.2.0"
+  version = "0.2.1"
 
 [[constraint]]
   version = "kubernetes-1.14.9"

internal/extender/extendertest/extender_test_utils.go

@@ -213,6 +213,12 @@ func NewNode(name string) v1.Node {
 				v1.ResourceCPU:    *resource.NewQuantity(8, resource.DecimalSI),
 				v1.ResourceMemory: *resource.NewQuantity(8*1024*1024*1024, resource.BinarySI),
 			},
+			Conditions: []v1.NodeCondition{
+				{
+					Type:   v1.NodeReady,
+					Status: v1.ConditionTrue,
+				},
+			},
 		},
 	}
 }

internal/extender/failover.go

@@ -48,7 +48,7 @@ func (s *SparkSchedulerExtender) syncResourceReservationsAndDemands(ctx context.
 	rrs := s.resourceReservations.List()
 	overhead := s.overheadComputer.GetOverhead(ctx, nodes)
 	softReservationOverhead := s.softReservationStore.UsedSoftReservationResources()
-	availableResources, orderedNodes := availableResourcesPerInstanceGroup(ctx, s.instanceGroupLabel, rrs, nodes, overhead, softReservationOverhead)
+	availableResources, orderedNodes := availableResourcesPerInstanceGroup(s.instanceGroupLabel, rrs, nodes, overhead, softReservationOverhead)
 	staleSparkPods := unreservedSparkPodsBySparkID(ctx, rrs, s.softReservationStore, pods)
 	svc1log.FromContext(ctx).Info("starting reconciliation", svc1log.SafeParam("appCount", len(staleSparkPods)))
 
@@ -278,7 +278,6 @@ func isNotScheduledSparkPod(pod *v1.Pod) bool {
 }
 
 func availableResourcesPerInstanceGroup(
-	ctx context.Context,
 	instanceGroupLabel string,
 	rrs []*v1beta1.ResourceReservation,
 	nodes []*v1.Node,
@@ -293,6 +292,17 @@ func availableResourcesPerInstanceGroup(
 		if n.Spec.Unschedulable {
 			continue
 		}
+
+		nodeReady := false
+		for _, n := range n.Status.Conditions {
+			if n.Type == v1.NodeReady && n.Status == v1.ConditionTrue {
+				nodeReady = true
+			}
+		}
+		if !nodeReady {
+			continue
+		}
+
 		instanceGroup := instanceGroup(n.Labels[instanceGroupLabel])
 		schedulableNodes[instanceGroup] = append(schedulableNodes[instanceGroup], n)
 	}

internal/extender/nodesorting_test.go

@@ -138,18 +138,21 @@ func TestAZAwareNodeSortingWorksIfZoneLabelIsMissing(t *testing.T) {
 			CPU:    two,
 			Memory: one,
 		},
+		Ready: true,
 	}
 	node2SchedulingMetadata := &resources.NodeSchedulingMetadata{
 		AvailableResources: &resources.Resources{
 			CPU:    two,
 			Memory: one,
 		},
+		Ready: true,
 	}
 	node3SchedulingMetadata := &resources.NodeSchedulingMetadata{
 		AvailableResources: &resources.Resources{
 			CPU:    one,
 			Memory: one,
 		},
+		Ready: true,
 	}
 
 	nodesSchedulingMetadata := resources.NodeGroupSchedulingMetadata{

internal/extender/resource.go

@@ -309,7 +309,7 @@ func (s *SparkSchedulerExtender) potentialNodes(availableNodesSchedulingMetadata
 		if _, ok := nodeNamesSet[nodeName]; ok {
 			driverNodeNames = append(driverNodeNames, nodeName)
 		}
-		if !availableNodesSchedulingMetadata[nodeName].Unschedulable {
+		if !availableNodesSchedulingMetadata[nodeName].Unschedulable && availableNodesSchedulingMetadata[nodeName].Ready {
 			executorNodeNames = append(executorNodeNames, nodeName)
 		}
 	}