manager: sync rdma resource to node

Signed-off-by: wangjianyu.wjy <[email protected]>
koordinator-sh · Dec 5, 2024 · 30a4b7a · 30a4b7a
1 parent b36d230
commit 30a4b7a
Show file tree

Hide file tree

Showing 7 changed files with 1,002 additions and 0 deletions.
diff --git a/pkg/slo-controller/noderesource/plugins/gpudeviceresource/plugin.go b/pkg/slo-controller/noderesource/plugins/gpudeviceresource/plugin.go
@@ -151,6 +151,17 @@ func (p *Plugin) Calculate(_ *configuration.ColocationStrategy, node *corev1.Nod
 		return p.resetGPUNodeResource()
 	}
 
+	existsGPU := false
+	for _, d := range device.Spec.Devices {
+		if d.Type == schedulingv1alpha1.GPU && d.Health {
+			existsGPU = true
+		}
+	}
+	if !existsGPU {
+		klog.V(5).InfoS("gpu not found in device, reset gpu resources on node", "node", node.Name)
+		return p.resetGPUNodeResource()
+	}
+
 	// TODO: calculate NUMA-level resources against NRT
 	return p.calculate(node, device)
 }

diff --git a/pkg/slo-controller/noderesource/plugins/gpudeviceresource/plugin_test.go b/pkg/slo-controller/noderesource/plugins/gpudeviceresource/plugin_test.go
@@ -602,6 +602,14 @@ func TestPluginCalculate(t *testing.T) {
 			},
 		},
 	}
+	deviceMissingGPU := &schedulingv1alpha1.Device{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: testNode.Name,
+		},
+		Spec: schedulingv1alpha1.DeviceSpec{
+			Devices: []schedulingv1alpha1.DeviceInfo{},
+		},
+	}
 	type fields struct {
 		client ctrlclient.Client
 	}
@@ -838,6 +846,43 @@ func TestPluginCalculate(t *testing.T) {
 			},
 			wantErr: false,
 		},
+		{
+			name: "calculate resetting device resources",
+			fields: fields{
+				client: fake.NewClientBuilder().WithScheme(testScheme).WithObjects(testNode, deviceMissingGPU).Build(),
+			},
+			args: args{
+				node: testNode,
+			},
+			want: []framework.ResourceItem{
+				{
+					Name:    extension.ResourceGPU,
+					Reset:   true,
+					Message: ResetResourcesMsg,
+				},
+				{
+					Name:    extension.ResourceGPUCore,
+					Reset:   true,
+					Message: ResetResourcesMsg,
+				},
+				{
+					Name:    extension.ResourceGPUMemory,
+					Reset:   true,
+					Message: ResetResourcesMsg,
+				},
+				{
+					Name:    extension.ResourceGPUMemoryRatio,
+					Reset:   true,
+					Message: ResetResourcesMsg,
+				},
+				{
+					Name:    extension.ResourceGPUShared,
+					Reset:   true,
+					Message: ResetResourcesMsg,
+				},
+			},
+			wantErr: false,
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {

diff --git a/pkg/slo-controller/noderesource/plugins/rdmadevicereource/device_event_handler.go b/pkg/slo-controller/noderesource/plugins/rdmadevicereource/device_event_handler.go
@@ -0,0 +1,71 @@
+/*
+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 rdmadeviceresource
+
+import (
+	"context"
+	"reflect"
+
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/util/workqueue"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+
+	schedulingv1alpha1 "github.com/koordinator-sh/koordinator/apis/scheduling/v1alpha1"
+)
+
+var _ handler.EventHandler = &DeviceHandler{}
+
+type DeviceHandler struct{}
+
+func (d *DeviceHandler) Create(ctx context.Context, e event.CreateEvent, q workqueue.RateLimitingInterface) {
+	device := e.Object.(*schedulingv1alpha1.Device)
+	q.Add(reconcile.Request{
+		NamespacedName: types.NamespacedName{
+			Name: device.Name,
+		},
+	})
+}
+
+func (d *DeviceHandler) Update(ctx context.Context, e event.UpdateEvent, q workqueue.RateLimitingInterface) {
+	newDevice := e.ObjectNew.(*schedulingv1alpha1.Device)
+	oldDevice := e.ObjectOld.(*schedulingv1alpha1.Device)
+	if reflect.DeepEqual(newDevice.Spec, oldDevice.Spec) {
+		return
+	}
+	q.Add(reconcile.Request{
+		NamespacedName: types.NamespacedName{
+			Name: newDevice.Name,
+		},
+	})
+}
+
+func (d *DeviceHandler) Delete(ctx context.Context, e event.DeleteEvent, q workqueue.RateLimitingInterface) {
+	device, ok := e.Object.(*schedulingv1alpha1.Device)
+	if !ok {
+		return
+	}
+	q.Add(reconcile.Request{
+		NamespacedName: types.NamespacedName{
+			Name: device.Name,
+		},
+	})
+}
+
+func (d *DeviceHandler) Generic(ctx context.Context, e event.GenericEvent, q workqueue.RateLimitingInterface) {
+}
diff --git a/pkg/slo-controller/noderesource/plugins/rdmadevicereource/device_event_handler_test.go b/pkg/slo-controller/noderesource/plugins/rdmadevicereource/device_event_handler_test.go
@@ -0,0 +1,166 @@
+/*
+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 rdmadeviceresource
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/util/workqueue"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+
+	schedulingv1alpha1 "github.com/koordinator-sh/koordinator/apis/scheduling/v1alpha1"
+)
+
+func Test_EnqueueRequestForNodeMetricMetric(t *testing.T) {
+	tests := []struct {
+		name      string
+		fn        func(handler handler.EventHandler, q workqueue.RateLimitingInterface)
+		hasEvent  bool
+		eventName string
+	}{
+		{
+			name: "create device event",
+			fn: func(handler handler.EventHandler, q workqueue.RateLimitingInterface) {
+				handler.Create(context.TODO(), event.CreateEvent{
+					Object: &schedulingv1alpha1.Device{
+						ObjectMeta: metav1.ObjectMeta{
+							Name: "node1",
+						},
+					},
+				}, q)
+			},
+			hasEvent:  true,
+			eventName: "node1",
+		},
+		{
+			name: "delete device event",
+			fn: func(handler handler.EventHandler, q workqueue.RateLimitingInterface) {
+				handler.Delete(context.TODO(), event.DeleteEvent{
+					Object: &schedulingv1alpha1.Device{
+						ObjectMeta: metav1.ObjectMeta{
+							Name: "node1",
+						},
+					},
+				}, q)
+			},
+			hasEvent:  true,
+			eventName: "node1",
+		},
+		{
+			name: "delete event not device",
+			fn: func(handler handler.EventHandler, q workqueue.RateLimitingInterface) {
+				handler.Delete(context.TODO(), event.DeleteEvent{
+					Object: &corev1.Node{
+						ObjectMeta: metav1.ObjectMeta{
+							Name: "node1",
+						},
+					},
+				}, q)
+			},
+			hasEvent: false,
+		},
+		{
+			name: "generic event ignore",
+			fn: func(handler handler.EventHandler, q workqueue.RateLimitingInterface) {
+				handler.Generic(context.TODO(), event.GenericEvent{}, q)
+			},
+			hasEvent: false,
+		},
+		{
+			name: "update device event",
+			fn: func(handler handler.EventHandler, q workqueue.RateLimitingInterface) {
+				handler.Update(context.TODO(), event.UpdateEvent{
+					ObjectOld: &schedulingv1alpha1.Device{
+						ObjectMeta: metav1.ObjectMeta{
+							Name:            "node1",
+							ResourceVersion: "100",
+						},
+						Spec: schedulingv1alpha1.DeviceSpec{
+							Devices: []schedulingv1alpha1.DeviceInfo{
+								{},
+							},
+						},
+					},
+					ObjectNew: &schedulingv1alpha1.Device{
+						ObjectMeta: metav1.ObjectMeta{
+							Name:            "node1",
+							ResourceVersion: "101",
+						},
+						Spec: schedulingv1alpha1.DeviceSpec{
+							Devices: []schedulingv1alpha1.DeviceInfo{
+								{},
+								{},
+							},
+						},
+					},
+				}, q)
+			},
+			hasEvent:  true,
+			eventName: "node1",
+		},
+		{
+			name: "update device event ignore",
+			fn: func(handler handler.EventHandler, q workqueue.RateLimitingInterface) {
+				handler.Update(context.TODO(), event.UpdateEvent{
+					ObjectOld: &schedulingv1alpha1.Device{
+						ObjectMeta: metav1.ObjectMeta{
+							Name:            "node1",
+							ResourceVersion: "100",
+						},
+						Spec: schedulingv1alpha1.DeviceSpec{
+							Devices: []schedulingv1alpha1.DeviceInfo{
+								{},
+							},
+						},
+					},
+					ObjectNew: &schedulingv1alpha1.Device{
+						ObjectMeta: metav1.ObjectMeta{
+							Name:            "node1",
+							ResourceVersion: "100",
+						},
+						Spec: schedulingv1alpha1.DeviceSpec{
+							Devices: []schedulingv1alpha1.DeviceInfo{
+								{},
+							},
+						},
+					},
+				}, q)
+			},
+			hasEvent: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			queue := workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter())
+			h := &DeviceHandler{}
+			tt.fn(h, queue)
+			assert.Equal(t, tt.hasEvent, queue.Len() > 0, "unexpected event")
+			if tt.hasEvent {
+				assert.True(t, queue.Len() >= 0, "expected event")
+				e, _ := queue.Get()
+				assert.Equal(t, tt.eventName, e.(reconcile.Request).Name)
+			}
+		})
+	}
+
+}