koord-scheduler: support dispatching scheduling error (#1324)

Signed-off-by: Joseph <joseph.t.lee@outlook.com>

cmd/koord-scheduler/app/server.go

@@ -399,7 +399,13 @@ func Setup(ctx context.Context, opts *options.Options, outOfTreeRegistryOptions
 	schedAdapter := frameworkExtenderFactory.Scheduler()
 
 	eventhandlers.AddScheduleEventHandler(sched, schedAdapter, frameworkExtenderFactory.KoordinatorSharedInformerFactory())
-	eventhandlers.AddReservationErrorHandler(sched, schedAdapter, frameworkExtenderFactory.KoordinatorClientSet(), frameworkExtenderFactory.KoordinatorSharedInformerFactory())
+	reservationErrorHandler := eventhandlers.MakeReservationErrorHandler(
+		sched,
+		schedAdapter,
+		frameworkExtenderFactory.KoordinatorClientSet(),
+		frameworkExtenderFactory.KoordinatorSharedInformerFactory(),
+	)
+	frameworkExtenderFactory.RegisterErrorHandler(reservationErrorHandler)
 
 	return &cc, sched, frameworkExtenderFactory, nil
 }

pkg/scheduler/frameworkext/errorhandler_dispatcher.go

@@ -0,0 +1,47 @@
+/*
+Copyright 2022 The Koordinator Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package frameworkext
+
+import "k8s.io/kubernetes/pkg/scheduler/framework"
+
+type ErrorHandler func(*framework.QueuedPodInfo, error) bool
+
+type errorHandlerDispatcher struct {
+	handlers       []ErrorHandler
+	defaultHandler func(*framework.QueuedPodInfo, error)
+}
+
+func newErrorHandlerDispatcher() *errorHandlerDispatcher {
+	return &errorHandlerDispatcher{}
+}
+
+func (d *errorHandlerDispatcher) setDefaultHandler(handler func(*framework.QueuedPodInfo, error)) {
+	d.defaultHandler = handler
+}
+
+func (d *errorHandlerDispatcher) RegisterErrorHandler(handler ErrorHandler) {
+	d.handlers = append(d.handlers, handler)
+}
+
+func (d *errorHandlerDispatcher) Error(podInfo *framework.QueuedPodInfo, err error) {
+	for _, handler := range d.handlers {
+		if handler(podInfo, err) {
+			return
+		}
+	}
+	d.defaultHandler(podInfo, err)
+}

pkg/scheduler/frameworkext/errorhandler_dispatcher_test.go

@@ -0,0 +1,90 @@
+/*
+Copyright 2022 The Koordinator Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package frameworkext
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/kubernetes/pkg/scheduler/framework"
+)
+
+func TestErrorHandlerDispatcher(t *testing.T) {
+	dispatcher := newErrorHandlerDispatcher()
+	enterDefaultHandler := false
+	dispatcher.setDefaultHandler(func(info *framework.QueuedPodInfo, err error) {
+		enterDefaultHandler = true
+	})
+
+	handler1Processed := false
+	dispatcher.RegisterErrorHandler(func(info *framework.QueuedPodInfo, err error) bool {
+		if info.Pod.Name == "handler1" {
+			handler1Processed = true
+			return true
+		}
+		return false
+	})
+
+	handler2Processed := false
+	dispatcher.RegisterErrorHandler(func(info *framework.QueuedPodInfo, err error) bool {
+		if info.Pod.Name == "handler2" {
+			handler2Processed = true
+			return true
+		}
+		return false
+	})
+	dispatcher.Error(&framework.QueuedPodInfo{
+		PodInfo: &framework.PodInfo{
+			Pod: &corev1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "none",
+				},
+			},
+		},
+	}, nil)
+	assert.True(t, enterDefaultHandler)
+	enterDefaultHandler = false
+
+	dispatcher.Error(&framework.QueuedPodInfo{
+		PodInfo: &framework.PodInfo{
+			Pod: &corev1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "handler2",
+				},
+			},
+		},
+	}, nil)
+	assert.False(t, handler1Processed)
+	assert.True(t, handler2Processed)
+	assert.False(t, enterDefaultHandler)
+	handler2Processed = false
+
+	dispatcher.Error(&framework.QueuedPodInfo{
+		PodInfo: &framework.PodInfo{
+			Pod: &corev1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "handler1",
+				},
+			},
+		},
+	}, nil)
+	assert.True(t, handler1Processed)
+	assert.False(t, handler2Processed)
+	assert.False(t, enterDefaultHandler)
+}

pkg/scheduler/frameworkext/eventhandlers/reservation_handler.go

@@ -42,41 +42,40 @@ import (
 // Register schedulingv1alpha1 scheme to report event
 var _ = schedulingv1alpha1.AddToScheme(scheme.Scheme)
 
-func AddReservationErrorHandler(
+func MakeReservationErrorHandler(
 	sched *scheduler.Scheduler,
 	schedAdapter frameworkext.Scheduler,
 	koordClientSet koordclientset.Interface,
 	koordSharedInformerFactory koordinatorinformers.SharedInformerFactory,
-) {
-	defaultErrorFn := sched.Error
+) frameworkext.ErrorHandler {
 	reservationLister := koordSharedInformerFactory.Scheduling().V1alpha1().Reservations().Lister()
 	reservationErrorFn := makeReservationErrorFunc(schedAdapter, reservationLister)
-	sched.Error = func(podInfo *framework.QueuedPodInfo, schedulingErr error) {
+	return func(podInfo *framework.QueuedPodInfo, schedulingErr error) bool {
 		pod := podInfo.Pod
 		// if the pod is not a reserve pod, use the default error handler
 		if !reservationutil.IsReservePod(pod) {
-			defaultErrorFn(podInfo, schedulingErr)
-			return
+			return false
 		}
 
 		reservationErrorFn(podInfo, schedulingErr)
 
 		rName := reservationutil.GetReservationNameFromReservePod(pod)
 		r, err := reservationLister.Get(rName)
 		if err != nil {
-			return
+			return true
 		}
 
 		fwk, ok := sched.Profiles[pod.Spec.SchedulerName]
 		if !ok {
 			klog.Errorf("profile not found for scheduler name %q", pod.Spec.SchedulerName)
-			return
+			return true
 		}
 
 		msg := truncateMessage(schedulingErr.Error())
 		fwk.EventRecorder().Eventf(r, nil, corev1.EventTypeWarning, "FailedScheduling", "Scheduling", msg)
 
 		updateReservationStatus(koordClientSet, reservationLister, rName, schedulingErr)
+		return true
 	}
 }
 

pkg/scheduler/frameworkext/eventhandlers/reservation_handler_test.go

@@ -98,7 +98,7 @@ func TestAddReservationErrorHandler(t *testing.T) {
 		koordClientSet := koordfake.NewSimpleClientset(testR)
 		koordSharedInformerFactory := koordinatorinformers.NewSharedInformerFactory(koordClientSet, 0)
 
-		AddReservationErrorHandler(sched, internalHandler, koordClientSet, koordSharedInformerFactory)
+		handler := MakeReservationErrorHandler(sched, internalHandler, koordClientSet, koordSharedInformerFactory)
 
 		koordSharedInformerFactory.Start(nil)
 		koordSharedInformerFactory.WaitForCacheSync(nil)
@@ -108,7 +108,7 @@ func TestAddReservationErrorHandler(t *testing.T) {
 		}
 
 		expectedErr := errors.New(strings.Repeat("test error", validation.NoteLengthLimit))
-		sched.Error(queuedPodInfo, expectedErr)
+		handler(queuedPodInfo, expectedErr)
 
 		r, err := koordClientSet.SchedulingV1alpha1().Reservations().Get(context.TODO(), testR.Name, metav1.GetOptions{})
 		assert.NoError(t, err)

pkg/scheduler/frameworkext/framework_extender.go

@@ -38,6 +38,7 @@ var _ FrameworkExtender = &frameworkExtenderImpl{}
 
 type frameworkExtenderImpl struct {
 	framework.Framework
+	*errorHandlerDispatcher
 
 	schedulerFn func() Scheduler
 
@@ -67,6 +68,7 @@ func NewFrameworkExtender(f *FrameworkExtenderFactory, fw framework.Framework) F
 
 	frameworkExtender := &frameworkExtenderImpl{
 		Framework:                        fw,
+		errorHandlerDispatcher:           f.errorHandlerDispatcher,
 		schedulerFn:                      schedulerFn,
 		snapshotGeneration:               snapshotGeneration,
 		koordinatorClientSet:             f.KoordinatorClientSet(),

pkg/scheduler/frameworkext/framework_extender_factory.go

@@ -75,6 +75,7 @@ type FrameworkExtenderFactory struct {
 	koordinatorSharedInformerFactory koordinatorinformers.SharedInformerFactory
 	profiles                         map[string]FrameworkExtender
 	scheduler                        Scheduler
+	*errorHandlerDispatcher
 }
 
 func NewFrameworkExtenderFactory(options ...Option) (*FrameworkExtenderFactory, error) {
@@ -94,6 +95,7 @@ func NewFrameworkExtenderFactory(options ...Option) (*FrameworkExtenderFactory,
 		koordinatorClientSet:             handleOptions.koordinatorClientSet,
 		koordinatorSharedInformerFactory: handleOptions.koordinatorSharedInformerFactory,
 		profiles:                         map[string]FrameworkExtender{},
+		errorHandlerDispatcher:           newErrorHandlerDispatcher(),
 	}, nil
 }
 
@@ -133,6 +135,8 @@ func (f *FrameworkExtenderFactory) InitScheduler(sched *scheduler.Scheduler) {
 	f.scheduler = &SchedulerAdapter{
 		Scheduler: sched,
 	}
+	f.errorHandlerDispatcher.setDefaultHandler(sched.Error)
+	sched.Error = f.errorHandlerDispatcher.Error
 }
 
 func (f *FrameworkExtenderFactory) Run() {

pkg/scheduler/frameworkext/interface.go

@@ -38,6 +38,7 @@ type ExtendedHandle interface {
 	Scheduler() Scheduler
 	KoordinatorClientSet() koordinatorclientset.Interface
 	KoordinatorSharedInformerFactory() koordinatorinformers.SharedInformerFactory
+	RegisterErrorHandler(handler ErrorHandler)
 }
 
 // FrameworkExtender extends the K8s Scheduling Framework interface to provide more extension methods to support Koordinator.