Merge pull request #5 from smartnews/fix-extra-node

chore: ignore handled pending pods

Author: Luke-Smartnews

pkg/controllers/provisioning/provisioner.go

@@ -310,20 +310,27 @@ func (p *Provisioner) Schedule(ctx context.Context) (scheduler.Results, error) {
 		return scheduler.Results{}, err
 	}
 	pods := append(pendingPods, deletingNodePods...)
+	// filter pods which are alredy handled in last 3 minute
+	targetPods := lo.FilterMap(pods, func(pod *v1.Pod, _ int) (*v1.Pod, bool) {
+		if p.isPodHandled(ctx, pod) {
+			return nil, false
+		}
+		return pod, true
+	})
 	// nothing to schedule, so just return success
-	if len(pods) == 0 {
+	if len(targetPods) == 0 {
 		return scheduler.Results{}, nil
 	}
-	s, err := p.NewScheduler(ctx, pods, nodes.Active())
+	s, err := p.NewScheduler(ctx, targetPods, nodes.Active())
 	if err != nil {
 		if errors.Is(err, ErrNodePoolsNotFound) {
 			logging.FromContext(ctx).Info(ErrNodePoolsNotFound)
 			return scheduler.Results{}, nil
 		}
 		return scheduler.Results{}, fmt.Errorf("creating scheduler, %w", err)
 	}
-	results := s.Solve(ctx, pods).TruncateInstanceTypes(scheduler.MaxInstanceTypes)
-	logging.FromContext(ctx).With("pods", pretty.Slice(lo.Map(pods, func(p *v1.Pod, _ int) string { return client.ObjectKeyFromObject(p).String() }), 5)).
+	results := s.Solve(ctx, targetPods).TruncateInstanceTypes(scheduler.MaxInstanceTypes)
+	logging.FromContext(ctx).With("pods", pretty.Slice(lo.Map(targetPods, func(p *v1.Pod, _ int) string { return client.ObjectKeyFromObject(p).String() }), 5)).
 		With("duration", time.Since(start)).
 		Infof("found provisionable pod(s)")
 	results.Record(ctx, p.recorder, p.cluster)
@@ -419,6 +426,31 @@ func (p *Provisioner) Validate(ctx context.Context, pod *v1.Pod) error {
 	)
 }
 
+func (p *Provisioner) isPodHandled(ctx context.Context, pod *v1.Pod) bool {
+	var events v1.EventList
+	filter := client.MatchingFields{
+		"namespace":           pod.Namespace,
+		"involvedObject.kind": "Pod",
+		"involvedObject.name": pod.Name,
+		"reason":              "HandledByKarpenter",
+	}
+	logging.FromContext(ctx).Debugf("get event for %s/%s", pod.Namespace, pod.Name)
+	if err := p.kubeClient.List(ctx, &events, filter); err == nil {
+		for _, event := range events.Items {
+			logging.FromContext(ctx).Debugf("found event %s/%s", pod.Namespace, event.Name)
+			// ignore the pod if it's already handled in 3 minute
+			if time.Now().Before(event.LastTimestamp.Time.Add(3 * time.Minute)) {
+				logging.FromContext(ctx).Infof("pod %s/%s is handled", pod.Namespace, pod.Name)
+				return true
+			}
+		}
+	} else {
+		logging.FromContext(ctx).Errorf("failed to get event for %s/%s: %w", pod.Namespace, pod.Name, err)
+	}
+	p.recorder.Publish(scheduler.PodHandledEvent(pod))
+	return false
+}
+
 // validateKarpenterManagedLabelCanExist provides a more clear error message in the event of scheduling a pod that specifically doesn't
 // want to run on a Karpenter node (e.g. a Karpenter controller replica).
 func validateKarpenterManagedLabelCanExist(p *v1.Pod) error {

pkg/controllers/provisioning/scheduling/events.go

@@ -59,3 +59,14 @@ func PodFailedToScheduleEvent(pod *v1.Pod, err error) events.Event {
 		DedupeTimeout:  5 * time.Minute,
 	}
 }
+
+func PodHandledEvent(pod *v1.Pod) events.Event {
+	return events.Event{
+		InvolvedObject: pod,
+		Type:           v1.EventTypeNormal,
+		Reason:         "HandledByKarpenter",
+		Message:        "Pod is handled by karpenter",
+		DedupeValues:   []string{string(pod.UID)},
+		DedupeTimeout:  5 * time.Minute,
+	}
+}

pkg/operator/operator.go

@@ -187,6 +187,18 @@ func NewOperator() (context.Context, *Operator) {
 	lo.Must0(mgr.GetFieldIndexer().IndexField(ctx, &v1.Node{}, "spec.providerID", func(o client.Object) []string {
 		return []string{o.(*v1.Node).Spec.ProviderID}
 	}), "failed to setup node provider id indexer")
+	lo.Must0(mgr.GetFieldIndexer().IndexField(ctx, &v1.Event{}, "involvedObject.kind", func(o client.Object) []string {
+		return []string{o.(*v1.Event).InvolvedObject.Kind}
+	}), "failed to setup event kind indexer")
+	lo.Must0(mgr.GetFieldIndexer().IndexField(ctx, &v1.Event{}, "involvedObject.name", func(o client.Object) []string {
+		return []string{o.(*v1.Event).InvolvedObject.Name}
+	}), "failed to setup event name indexer")
+	lo.Must0(mgr.GetFieldIndexer().IndexField(ctx, &v1.Event{}, "namespace", func(o client.Object) []string {
+		return []string{o.(*v1.Event).Namespace}
+	}), "failed to setup event namespace indexer")
+	lo.Must0(mgr.GetFieldIndexer().IndexField(ctx, &v1.Event{}, "reason", func(o client.Object) []string {
+		return []string{o.(*v1.Event).Reason}
+	}), "failed to setup event reason indexer")
 	lo.Must0(mgr.GetFieldIndexer().IndexField(ctx, &v1beta1.NodeClaim{}, "status.providerID", func(o client.Object) []string {
 		return []string{o.(*v1beta1.NodeClaim).Status.ProviderID}
 	}), "failed to setup nodeclaim provider id indexer")