From 83a253634d901a98ae619cb026240c5499be745e Mon Sep 17 00:00:00 2001 From: limengxuan <391013634@qq.com> Date: Tue, 14 Sep 2021 12:54:20 +0800 Subject: [PATCH 1/6] add mem and cores settings --- README.md | 0 .../templates/device-plugin/monitorrole.yaml | 2 +- .../device-plugin/monitorrolebinding.yaml | 3 +- .../templates/scheduler/configmap.yaml | 4 + .../templates/scheduler/deployment.yaml | 2 + deployments/4pd-vgpu/values.yaml | 2 + pkg/api/device_register.pb.go | 382 +++++++++++++-- pkg/api/device_register.proto | 11 +- pkg/device-plugin/nvidia.go | 248 +++++----- pkg/device-plugin/register.go | 200 ++++---- pkg/device-plugin/runtime.go | 127 ++--- pkg/k8sutil/pod.go | 78 ++- pkg/scheduler/nodes.go | 58 ++- pkg/scheduler/scheduler.go | 461 +++++++++--------- pkg/scheduler/score.go | 50 +- pkg/scheduler/webhook.go | 126 ++--- pkg/util/types.go | 40 +- pkg/util/util.go | 86 ++-- 18 files changed, 1182 insertions(+), 698 deletions(-) create mode 100644 README.md diff --git a/README.md b/README.md new file mode 100644 index 0000000000..e69de29bb2 diff --git a/deployments/4pd-vgpu/templates/device-plugin/monitorrole.yaml b/deployments/4pd-vgpu/templates/device-plugin/monitorrole.yaml index 70e2da8d38..1d6b8d78ce 100644 --- a/deployments/4pd-vgpu/templates/device-plugin/monitorrole.yaml +++ b/deployments/4pd-vgpu/templates/device-plugin/monitorrole.yaml @@ -1,7 +1,7 @@ apiVersion: rbac.authorization.k8s.io/v1 kind: ClusterRole metadata: - name: {{ include "4pd-vgpu.device-plugin" . }}-admission + name: {{ include "4pd-vgpu.device-plugin" . }}-monitor rules: - apiGroups: - "" diff --git a/deployments/4pd-vgpu/templates/device-plugin/monitorrolebinding.yaml b/deployments/4pd-vgpu/templates/device-plugin/monitorrolebinding.yaml index 53dc656a3b..46c0c0199c 100644 --- a/deployments/4pd-vgpu/templates/device-plugin/monitorrolebinding.yaml +++ b/deployments/4pd-vgpu/templates/device-plugin/monitorrolebinding.yaml @@ -8,7 +8,8 @@ metadata: roleRef: apiGroup: rbac.authorization.k8s.io kind: ClusterRole - name: cluster-admin + #name: cluster-admin + name: {{ include "4pd-vgpu.device-plugin" . }}-monitor subjects: - kind: ServiceAccount name: {{ include "4pd-vgpu.device-plugin" . }} diff --git a/deployments/4pd-vgpu/templates/scheduler/configmap.yaml b/deployments/4pd-vgpu/templates/scheduler/configmap.yaml index ad432f1535..4718fd3ad7 100644 --- a/deployments/4pd-vgpu/templates/scheduler/configmap.yaml +++ b/deployments/4pd-vgpu/templates/scheduler/configmap.yaml @@ -27,3 +27,7 @@ data: managedResources: - name: {{ .Values.resourceName }} ignoredByScheduler: true + - name: {{ .Values.resourceMem }} + ignoredByScheduler: true + - name: {{ .Values.resourcecores }} + ignoredByScheduler: true diff --git a/deployments/4pd-vgpu/templates/scheduler/deployment.yaml b/deployments/4pd-vgpu/templates/scheduler/deployment.yaml index 489c9b348c..35f0d8d115 100644 --- a/deployments/4pd-vgpu/templates/scheduler/deployment.yaml +++ b/deployments/4pd-vgpu/templates/scheduler/deployment.yaml @@ -49,6 +49,8 @@ spec: command: - scheduler - --resource-name={{ .Values.resourceName }} + - --resource-mem={{ .Values.resourceMem }} + - --resource-cores={{ .Values.resourceCores }} - --http_bind=0.0.0.0:443 - --grpc_bind=0.0.0.0:1080 - --cert_file=/tls/tls.crt diff --git a/deployments/4pd-vgpu/values.yaml b/deployments/4pd-vgpu/values.yaml index 8c22f120a4..b8583659b3 100644 --- a/deployments/4pd-vgpu/values.yaml +++ b/deployments/4pd-vgpu/values.yaml @@ -6,6 +6,8 @@ fullnameOverride: "" imagePullSecrets: [] # resourceName: "nvidia.com/gpu" +resourceMem: "nvidia.com/gpumem" +resourceCores: "nvidia.com/gpucores" schedulerName: "4pd-scheduler" podSecurityPolicy: diff --git a/pkg/api/device_register.pb.go b/pkg/api/device_register.pb.go index 2cd673df57..52e1d3ce07 100644 --- a/pkg/api/device_register.pb.go +++ b/pkg/api/device_register.pb.go @@ -29,7 +29,8 @@ const _ = proto.ProtoPackageIsVersion3 // please upgrade the proto package type DeviceInfo struct { Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"` Count int32 `protobuf:"varint,2,opt,name=count,proto3" json:"count,omitempty"` - Health bool `protobuf:"varint,3,opt,name=health,proto3" json:"health,omitempty"` + Devmem int32 `protobuf:"varint,3,opt,name=devmem,proto3" json:"devmem,omitempty"` + Health bool `protobuf:"varint,4,opt,name=health,proto3" json:"health,omitempty"` XXX_NoUnkeyedLiteral struct{} `json:"-"` XXX_unrecognized []byte `json:"-"` XXX_sizecache int32 `json:"-"` @@ -82,6 +83,13 @@ func (m *DeviceInfo) GetCount() int32 { return 0 } +func (m *DeviceInfo) GetDevmem() int32 { + if m != nil { + return m.Devmem + } + return 0 +} + func (m *DeviceInfo) GetHealth() bool { if m != nil { return m.Health @@ -230,22 +238,85 @@ func (m *GetContainerRequest) GetUuid() string { return "" } -type GetContainerReply struct { - DevList []string `protobuf:"bytes,1,rep,name=devList,proto3" json:"devList,omitempty"` - PodUID string `protobuf:"bytes,3,opt,name=podUID,proto3" json:"podUID,omitempty"` - CtrName string `protobuf:"bytes,4,opt,name=ctrName,proto3" json:"ctrName,omitempty"` - PodNamespace string `protobuf:"bytes,5,opt,name=podNamespace,proto3" json:"podNamespace,omitempty"` - PodName string `protobuf:"bytes,6,opt,name=podName,proto3" json:"podName,omitempty"` +type DeviceUsage struct { + Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"` + Devmem int32 `protobuf:"varint,2,opt,name=devmem,proto3" json:"devmem,omitempty"` + Cores int32 `protobuf:"varint,3,opt,name=cores,proto3" json:"cores,omitempty"` XXX_NoUnkeyedLiteral struct{} `json:"-"` XXX_unrecognized []byte `json:"-"` XXX_sizecache int32 `json:"-"` } +func (m *DeviceUsage) Reset() { *m = DeviceUsage{} } +func (m *DeviceUsage) String() string { return proto.CompactTextString(m) } +func (*DeviceUsage) ProtoMessage() {} +func (*DeviceUsage) Descriptor() ([]byte, []int) { + return fileDescriptor_f726eb77a5b37099, []int{4} +} +func (m *DeviceUsage) XXX_Unmarshal(b []byte) error { + return m.Unmarshal(b) +} +func (m *DeviceUsage) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + if deterministic { + return xxx_messageInfo_DeviceUsage.Marshal(b, m, deterministic) + } else { + b = b[:cap(b)] + n, err := m.MarshalToSizedBuffer(b) + if err != nil { + return nil, err + } + return b[:n], nil + } +} +func (m *DeviceUsage) XXX_Merge(src proto.Message) { + xxx_messageInfo_DeviceUsage.Merge(m, src) +} +func (m *DeviceUsage) XXX_Size() int { + return m.Size() +} +func (m *DeviceUsage) XXX_DiscardUnknown() { + xxx_messageInfo_DeviceUsage.DiscardUnknown(m) +} + +var xxx_messageInfo_DeviceUsage proto.InternalMessageInfo + +func (m *DeviceUsage) GetId() string { + if m != nil { + return m.Id + } + return "" +} + +func (m *DeviceUsage) GetDevmem() int32 { + if m != nil { + return m.Devmem + } + return 0 +} + +func (m *DeviceUsage) GetCores() int32 { + if m != nil { + return m.Cores + } + return 0 +} + +type GetContainerReply struct { + DevList []*DeviceUsage `protobuf:"bytes,1,rep,name=devList,proto3" json:"devList,omitempty"` + PodUID string `protobuf:"bytes,3,opt,name=podUID,proto3" json:"podUID,omitempty"` + CtrName string `protobuf:"bytes,4,opt,name=ctrName,proto3" json:"ctrName,omitempty"` + PodNamespace string `protobuf:"bytes,5,opt,name=podNamespace,proto3" json:"podNamespace,omitempty"` + PodName string `protobuf:"bytes,6,opt,name=podName,proto3" json:"podName,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + func (m *GetContainerReply) Reset() { *m = GetContainerReply{} } func (m *GetContainerReply) String() string { return proto.CompactTextString(m) } func (*GetContainerReply) ProtoMessage() {} func (*GetContainerReply) Descriptor() ([]byte, []int) { - return fileDescriptor_f726eb77a5b37099, []int{4} + return fileDescriptor_f726eb77a5b37099, []int{5} } func (m *GetContainerReply) XXX_Unmarshal(b []byte) error { return m.Unmarshal(b) @@ -274,7 +345,7 @@ func (m *GetContainerReply) XXX_DiscardUnknown() { var xxx_messageInfo_GetContainerReply proto.InternalMessageInfo -func (m *GetContainerReply) GetDevList() []string { +func (m *GetContainerReply) GetDevList() []*DeviceUsage { if m != nil { return m.DevList } @@ -314,36 +385,40 @@ func init() { proto.RegisterType((*RegisterRequest)(nil), "api.RegisterRequest") proto.RegisterType((*RegisterReply)(nil), "api.RegisterReply") proto.RegisterType((*GetContainerRequest)(nil), "api.GetContainerRequest") + proto.RegisterType((*DeviceUsage)(nil), "api.DeviceUsage") proto.RegisterType((*GetContainerReply)(nil), "api.GetContainerReply") } func init() { proto.RegisterFile("pkg/api/device_register.proto", fileDescriptor_f726eb77a5b37099) } var fileDescriptor_f726eb77a5b37099 = []byte{ - // 357 bytes of a gzipped FileDescriptorProto - 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x6c, 0x92, 0xcf, 0x4a, 0xeb, 0x40, - 0x14, 0xc6, 0xef, 0x24, 0x6d, 0x6f, 0x7b, 0x6c, 0xad, 0x8e, 0xa5, 0x0c, 0x05, 0x43, 0xc8, 0x2a, - 0xdd, 0xb4, 0x50, 0xc1, 0xb5, 0x68, 0x41, 0x2a, 0x22, 0x32, 0xe2, 0x5a, 0x62, 0x32, 0xb6, 0x83, - 0x31, 0x33, 0x26, 0x13, 0x21, 0x8f, 0xe0, 0x1b, 0xb8, 0xf1, 0x7d, 0x5c, 0xfa, 0x08, 0x52, 0x5f, - 0x44, 0x32, 0x49, 0xac, 0x2d, 0x5d, 0xe5, 0xfc, 0xe6, 0x3b, 0xff, 0xbe, 0x43, 0xe0, 0x50, 0x3e, - 0xce, 0xc7, 0x9e, 0xe4, 0xe3, 0x80, 0xbd, 0x70, 0x9f, 0xdd, 0xc5, 0x6c, 0xce, 0x13, 0xc5, 0xe2, - 0x91, 0x8c, 0x85, 0x12, 0xd8, 0xf4, 0x24, 0x77, 0x2e, 0x00, 0xa6, 0x5a, 0x9d, 0x45, 0x0f, 0x02, - 0xef, 0x82, 0xc1, 0x03, 0x82, 0x6c, 0xe4, 0xb6, 0xa8, 0xc1, 0x03, 0xdc, 0x83, 0xba, 0x2f, 0xd2, - 0x48, 0x11, 0xc3, 0x46, 0x6e, 0x9d, 0x16, 0x80, 0xfb, 0xd0, 0x58, 0x30, 0x2f, 0x54, 0x0b, 0x62, - 0xda, 0xc8, 0x6d, 0xd2, 0x92, 0x9c, 0x6b, 0xe8, 0xd2, 0x72, 0x04, 0x65, 0xcf, 0x29, 0x4b, 0x14, - 0xc6, 0x50, 0x8b, 0x44, 0xc0, 0xca, 0x96, 0x3a, 0xc6, 0x43, 0xf8, 0x5f, 0x2c, 0x94, 0x10, 0xc3, - 0x36, 0xdd, 0x9d, 0x49, 0x77, 0xe4, 0x49, 0x3e, 0x5a, 0xad, 0x41, 0x2b, 0xdd, 0xe9, 0x42, 0x67, - 0xd5, 0x51, 0x86, 0x99, 0x33, 0x84, 0x83, 0x73, 0xa6, 0xce, 0x44, 0xa4, 0x3c, 0x1e, 0xad, 0x8d, - 0x49, 0xd3, 0xdf, 0xcd, 0x75, 0xec, 0xbc, 0x23, 0xd8, 0x5f, 0xcf, 0x95, 0x61, 0x86, 0x89, 0x1e, - 0x7e, 0xc9, 0x13, 0x45, 0x90, 0x6d, 0xba, 0x2d, 0x5a, 0x61, 0xee, 0x4a, 0x8a, 0xe0, 0x76, 0x36, - 0xd5, 0xae, 0x5a, 0xb4, 0xa4, 0xbc, 0xc2, 0x57, 0xf1, 0x95, 0xf7, 0xc4, 0x48, 0x4d, 0x0b, 0x15, - 0x62, 0x07, 0xda, 0x52, 0x04, 0x79, 0x98, 0x48, 0xcf, 0x67, 0xa4, 0xae, 0xe5, 0xb5, 0xb7, 0xbc, - 0xba, 0x64, 0xd2, 0x28, 0xaa, 0x4b, 0x9c, 0xbc, 0x22, 0xe8, 0x14, 0x9e, 0x6f, 0x58, 0x9c, 0x7f, - 0xf0, 0x31, 0x34, 0x2b, 0xb7, 0xb8, 0xa7, 0x6f, 0xb2, 0x71, 0xce, 0x01, 0xde, 0x78, 0x95, 0x61, - 0xe6, 0x22, 0x7c, 0x02, 0xed, 0xbf, 0x46, 0x31, 0xd1, 0x59, 0x5b, 0xee, 0x34, 0xe8, 0x6f, 0x51, - 0x64, 0x98, 0x9d, 0xee, 0x7d, 0x2c, 0x2d, 0xf4, 0xb9, 0xb4, 0xd0, 0xd7, 0xd2, 0x42, 0x6f, 0xdf, - 0xd6, 0xbf, 0xfb, 0x86, 0xfe, 0x47, 0x8e, 0x7e, 0x02, 0x00, 0x00, 0xff, 0xff, 0x5d, 0xc5, 0x41, - 0x30, 0x44, 0x02, 0x00, 0x00, + // 403 bytes of a gzipped FileDescriptorProto + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x6c, 0x92, 0x51, 0x8e, 0xd3, 0x30, + 0x10, 0x86, 0x71, 0xba, 0xed, 0x6e, 0x67, 0x77, 0xe9, 0x62, 0x56, 0x2b, 0x6b, 0x25, 0xa2, 0x28, + 0x4f, 0x59, 0x1e, 0xba, 0x52, 0x91, 0x78, 0x46, 0x50, 0x09, 0x55, 0x20, 0x84, 0x8c, 0xfa, 0x8c, + 0xdc, 0x64, 0x68, 0x2d, 0xd2, 0xd8, 0x24, 0x4e, 0xa5, 0x1c, 0x81, 0x1b, 0x70, 0x0b, 0xae, 0xc1, + 0x23, 0x47, 0x40, 0xe5, 0x22, 0x28, 0x8e, 0x43, 0x9b, 0x6e, 0x9f, 0x32, 0xff, 0x8c, 0x3d, 0xff, + 0x7c, 0xe3, 0xc0, 0x33, 0xfd, 0x75, 0x79, 0x2f, 0xb4, 0xbc, 0x4f, 0x70, 0x23, 0x63, 0xfc, 0x9c, + 0xe3, 0x52, 0x16, 0x06, 0xf3, 0xb1, 0xce, 0x95, 0x51, 0xb4, 0x27, 0xb4, 0x0c, 0x17, 0x00, 0x53, + 0x5b, 0x9d, 0x65, 0x5f, 0x14, 0x7d, 0x0c, 0x9e, 0x4c, 0x18, 0x09, 0x48, 0x34, 0xe4, 0x9e, 0x4c, + 0xe8, 0x35, 0xf4, 0x63, 0x55, 0x66, 0x86, 0x79, 0x01, 0x89, 0xfa, 0xbc, 0x11, 0xf4, 0x06, 0x06, + 0x09, 0x6e, 0xd6, 0xb8, 0x66, 0x3d, 0x9b, 0x76, 0xaa, 0xce, 0xaf, 0x50, 0xa4, 0x66, 0xc5, 0x4e, + 0x02, 0x12, 0x9d, 0x71, 0xa7, 0xc2, 0x8f, 0x30, 0xe2, 0xce, 0x9a, 0xe3, 0xb7, 0x12, 0x0b, 0x43, + 0x29, 0x9c, 0x64, 0x2a, 0x41, 0x67, 0x65, 0x63, 0x7a, 0x07, 0xa7, 0xcd, 0xa0, 0x05, 0xf3, 0x82, + 0x5e, 0x74, 0x3e, 0x19, 0x8d, 0x85, 0x96, 0xe3, 0xdd, 0x78, 0xbc, 0xad, 0x87, 0x23, 0xb8, 0xdc, + 0x75, 0xd4, 0x69, 0x15, 0xde, 0xc1, 0xd3, 0xb7, 0x68, 0xde, 0xa8, 0xcc, 0x08, 0x99, 0x75, 0x6c, + 0xca, 0xf2, 0x3f, 0x91, 0x8d, 0xc3, 0x77, 0x70, 0xde, 0xb4, 0x9c, 0x17, 0x62, 0x89, 0x0f, 0x90, + 0x77, 0x70, 0x5e, 0x07, 0xce, 0xae, 0x22, 0xc7, 0xc2, 0x31, 0x37, 0x22, 0xfc, 0x49, 0xe0, 0x49, + 0xd7, 0x58, 0xa7, 0x15, 0x7d, 0x6e, 0x49, 0xde, 0xcb, 0xc2, 0x30, 0x62, 0x49, 0xae, 0xf6, 0x48, + 0xac, 0x2d, 0x6f, 0x0f, 0xd4, 0x7e, 0x5a, 0x25, 0xf3, 0xd9, 0xd4, 0x36, 0x1e, 0x72, 0xa7, 0x28, + 0x83, 0xd3, 0xd8, 0xe4, 0x1f, 0xc4, 0x1a, 0xed, 0x36, 0x87, 0xbc, 0x95, 0x34, 0x84, 0x0b, 0xad, + 0x92, 0x3a, 0x2c, 0xb4, 0x88, 0x91, 0xf5, 0x6d, 0xb9, 0x93, 0xab, 0x6f, 0x3b, 0xcd, 0x06, 0xcd, + 0x6d, 0x27, 0x27, 0xdf, 0x09, 0x5c, 0x36, 0x83, 0x7c, 0xc2, 0xbc, 0xfe, 0xd0, 0x97, 0x70, 0xd6, + 0x2e, 0x93, 0x5e, 0xdb, 0x41, 0x0f, 0x5e, 0xeb, 0x96, 0x1e, 0x64, 0x75, 0x5a, 0x45, 0x84, 0xbe, + 0x82, 0x8b, 0x7d, 0x74, 0xca, 0xec, 0xa9, 0x23, 0xcf, 0x70, 0x7b, 0x73, 0xa4, 0xa2, 0xd3, 0xea, + 0xf5, 0xd5, 0xaf, 0xad, 0x4f, 0x7e, 0x6f, 0x7d, 0xf2, 0x67, 0xeb, 0x93, 0x1f, 0x7f, 0xfd, 0x47, + 0x8b, 0x81, 0xfd, 0x35, 0x5f, 0xfc, 0x0b, 0x00, 0x00, 0xff, 0xff, 0x17, 0x1e, 0xdb, 0x7a, 0xbb, + 0x02, 0x00, 0x00, } // Reference imports to suppress errors if they are not otherwise used. @@ -529,6 +604,11 @@ func (m *DeviceInfo) MarshalToSizedBuffer(dAtA []byte) (int, error) { dAtA[i] = 0 } i-- + dAtA[i] = 0x20 + } + if m.Devmem != 0 { + i = encodeVarintDeviceRegister(dAtA, i, uint64(m.Devmem)) + i-- dAtA[i] = 0x18 } if m.Count != 0 { @@ -655,6 +735,50 @@ func (m *GetContainerRequest) MarshalToSizedBuffer(dAtA []byte) (int, error) { return len(dAtA) - i, nil } +func (m *DeviceUsage) Marshal() (dAtA []byte, err error) { + size := m.Size() + dAtA = make([]byte, size) + n, err := m.MarshalToSizedBuffer(dAtA[:size]) + if err != nil { + return nil, err + } + return dAtA[:n], nil +} + +func (m *DeviceUsage) MarshalTo(dAtA []byte) (int, error) { + size := m.Size() + return m.MarshalToSizedBuffer(dAtA[:size]) +} + +func (m *DeviceUsage) MarshalToSizedBuffer(dAtA []byte) (int, error) { + i := len(dAtA) + _ = i + var l int + _ = l + if m.XXX_unrecognized != nil { + i -= len(m.XXX_unrecognized) + copy(dAtA[i:], m.XXX_unrecognized) + } + if m.Cores != 0 { + i = encodeVarintDeviceRegister(dAtA, i, uint64(m.Cores)) + i-- + dAtA[i] = 0x18 + } + if m.Devmem != 0 { + i = encodeVarintDeviceRegister(dAtA, i, uint64(m.Devmem)) + i-- + dAtA[i] = 0x10 + } + if len(m.Id) > 0 { + i -= len(m.Id) + copy(dAtA[i:], m.Id) + i = encodeVarintDeviceRegister(dAtA, i, uint64(len(m.Id))) + i-- + dAtA[i] = 0xa + } + return len(dAtA) - i, nil +} + func (m *GetContainerReply) Marshal() (dAtA []byte, err error) { size := m.Size() dAtA = make([]byte, size) @@ -709,9 +833,14 @@ func (m *GetContainerReply) MarshalToSizedBuffer(dAtA []byte) (int, error) { } if len(m.DevList) > 0 { for iNdEx := len(m.DevList) - 1; iNdEx >= 0; iNdEx-- { - i -= len(m.DevList[iNdEx]) - copy(dAtA[i:], m.DevList[iNdEx]) - i = encodeVarintDeviceRegister(dAtA, i, uint64(len(m.DevList[iNdEx]))) + { + size, err := m.DevList[iNdEx].MarshalToSizedBuffer(dAtA[:i]) + if err != nil { + return 0, err + } + i -= size + i = encodeVarintDeviceRegister(dAtA, i, uint64(size)) + } i-- dAtA[i] = 0xa } @@ -743,6 +872,9 @@ func (m *DeviceInfo) Size() (n int) { if m.Count != 0 { n += 1 + sovDeviceRegister(uint64(m.Count)) } + if m.Devmem != 0 { + n += 1 + sovDeviceRegister(uint64(m.Devmem)) + } if m.Health { n += 2 } @@ -802,6 +934,28 @@ func (m *GetContainerRequest) Size() (n int) { return n } +func (m *DeviceUsage) Size() (n int) { + if m == nil { + return 0 + } + var l int + _ = l + l = len(m.Id) + if l > 0 { + n += 1 + l + sovDeviceRegister(uint64(l)) + } + if m.Devmem != 0 { + n += 1 + sovDeviceRegister(uint64(m.Devmem)) + } + if m.Cores != 0 { + n += 1 + sovDeviceRegister(uint64(m.Cores)) + } + if m.XXX_unrecognized != nil { + n += len(m.XXX_unrecognized) + } + return n +} + func (m *GetContainerReply) Size() (n int) { if m == nil { return 0 @@ -809,8 +963,8 @@ func (m *GetContainerReply) Size() (n int) { var l int _ = l if len(m.DevList) > 0 { - for _, s := range m.DevList { - l = len(s) + for _, e := range m.DevList { + l = e.Size() n += 1 + l + sovDeviceRegister(uint64(l)) } } @@ -923,6 +1077,25 @@ func (m *DeviceInfo) Unmarshal(dAtA []byte) error { } } case 3: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field Devmem", wireType) + } + m.Devmem = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowDeviceRegister + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + m.Devmem |= int32(b&0x7F) << shift + if b < 0x80 { + break + } + } + case 4: if wireType != 0 { return fmt.Errorf("proto: wrong wireType = %d for field Health", wireType) } @@ -1215,7 +1388,7 @@ func (m *GetContainerRequest) Unmarshal(dAtA []byte) error { } return nil } -func (m *GetContainerReply) Unmarshal(dAtA []byte) error { +func (m *DeviceUsage) Unmarshal(dAtA []byte) error { l := len(dAtA) iNdEx := 0 for iNdEx < l { @@ -1238,15 +1411,15 @@ func (m *GetContainerReply) Unmarshal(dAtA []byte) error { fieldNum := int32(wire >> 3) wireType := int(wire & 0x7) if wireType == 4 { - return fmt.Errorf("proto: GetContainerReply: wiretype end group for non-group") + return fmt.Errorf("proto: DeviceUsage: wiretype end group for non-group") } if fieldNum <= 0 { - return fmt.Errorf("proto: GetContainerReply: illegal tag %d (wire type %d)", fieldNum, wire) + return fmt.Errorf("proto: DeviceUsage: illegal tag %d (wire type %d)", fieldNum, wire) } switch fieldNum { case 1: if wireType != 2 { - return fmt.Errorf("proto: wrong wireType = %d for field DevList", wireType) + return fmt.Errorf("proto: wrong wireType = %d for field Id", wireType) } var stringLen uint64 for shift := uint(0); ; shift += 7 { @@ -1274,7 +1447,130 @@ func (m *GetContainerReply) Unmarshal(dAtA []byte) error { if postIndex > l { return io.ErrUnexpectedEOF } - m.DevList = append(m.DevList, string(dAtA[iNdEx:postIndex])) + m.Id = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex + case 2: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field Devmem", wireType) + } + m.Devmem = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowDeviceRegister + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + m.Devmem |= int32(b&0x7F) << shift + if b < 0x80 { + break + } + } + case 3: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field Cores", wireType) + } + m.Cores = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowDeviceRegister + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + m.Cores |= int32(b&0x7F) << shift + if b < 0x80 { + break + } + } + default: + iNdEx = preIndex + skippy, err := skipDeviceRegister(dAtA[iNdEx:]) + if err != nil { + return err + } + if (skippy < 0) || (iNdEx+skippy) < 0 { + return ErrInvalidLengthDeviceRegister + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...) + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} +func (m *GetContainerReply) Unmarshal(dAtA []byte) error { + l := len(dAtA) + iNdEx := 0 + for iNdEx < l { + preIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowDeviceRegister + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + wireType := int(wire & 0x7) + if wireType == 4 { + return fmt.Errorf("proto: GetContainerReply: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: GetContainerReply: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field DevList", wireType) + } + var msglen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowDeviceRegister + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + msglen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if msglen < 0 { + return ErrInvalidLengthDeviceRegister + } + postIndex := iNdEx + msglen + if postIndex < 0 { + return ErrInvalidLengthDeviceRegister + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.DevList = append(m.DevList, &DeviceUsage{}) + if err := m.DevList[len(m.DevList)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil { + return err + } iNdEx = postIndex case 3: if wireType != 2 { diff --git a/pkg/api/device_register.proto b/pkg/api/device_register.proto index 5846183a4e..9094511d5b 100644 --- a/pkg/api/device_register.proto +++ b/pkg/api/device_register.proto @@ -10,7 +10,8 @@ service DeviceService { message DeviceInfo { string id = 1; int32 count = 2; - bool health = 3; + int32 devmem = 3; + bool health = 4; } message RegisterRequest { @@ -25,8 +26,14 @@ message GetContainerRequest { string uuid = 1; } +message DeviceUsage { + string id = 1; + int32 devmem = 2; + int32 cores = 3; +} + message GetContainerReply { - repeated string devList = 1; + repeated DeviceUsage devList = 1; string podUID = 3; string ctrName = 4; string podNamespace = 5; diff --git a/pkg/device-plugin/nvidia.go b/pkg/device-plugin/nvidia.go index 3f609ae592..d89a354155 100644 --- a/pkg/device-plugin/nvidia.go +++ b/pkg/device-plugin/nvidia.go @@ -17,169 +17,173 @@ package device_plugin import ( - "fmt" - "log" - "os" - "strings" + "fmt" + "log" + "os" + "strings" - "github.com/NVIDIA/gpu-monitoring-tools/bindings/go/nvml" + "4pd.io/k8s-vgpu/pkg/util" + "github.com/NVIDIA/gpu-monitoring-tools/bindings/go/nvml" - pluginapi "k8s.io/kubelet/pkg/apis/deviceplugin/v1beta1" + pluginapi "k8s.io/kubelet/pkg/apis/deviceplugin/v1beta1" ) const ( - envDisableHealthChecks = "DP_DISABLE_HEALTHCHECKS" - allHealthChecks = "xids" + envDisableHealthChecks = "DP_DISABLE_HEALTHCHECKS" + allHealthChecks = "xids" ) // Device couples an underlying pluginapi.Device type with its device node paths type Device struct { - pluginapi.Device - Paths []string - Index string - Memory uint64 + pluginapi.Device + Paths []string + Index string + Memory uint64 } // ResourceManager provides an interface for listing a set of Devices and checking health on them type ResourceManager interface { - Devices() []*Device - CheckHealth(stop <-chan interface{}, devices []*Device, unhealthy chan<- *Device) + Devices() []*Device + CheckHealth(stop <-chan interface{}, devices []*Device, unhealthy chan<- *Device) } // GpuDeviceManager implements the ResourceManager interface for full GPU devices type GpuDeviceManager struct { - skipMigEnabledGPUs bool + skipMigEnabledGPUs bool } func check(err error) { - if err != nil { - log.Panicln("Fatal:", err) - } + if err != nil { + log.Panicln("Fatal:", err) + } } // NewGpuDeviceManager returns a reference to a new GpuDeviceManager func NewGpuDeviceManager(skipMigEnabledGPUs bool) *GpuDeviceManager { - return &GpuDeviceManager{ - skipMigEnabledGPUs: skipMigEnabledGPUs, - } + return &GpuDeviceManager{ + skipMigEnabledGPUs: skipMigEnabledGPUs, + } } // Devices returns a list of devices from the GpuDeviceManager func (g *GpuDeviceManager) Devices() []*Device { - n, err := nvml.GetDeviceCount() - check(err) + n, err := nvml.GetDeviceCount() + check(err) + if n > util.DeviceLimit { + n = util.DeviceLimit + } - var devs []*Device - for i := uint(0); i < n; i++ { - d, err := nvml.NewDevice(i) - check(err) + var devs []*Device + for i := uint(0); i < n; i++ { + d, err := nvml.NewDevice(i) + check(err) - migEnabled, err := d.IsMigEnabled() - check(err) + migEnabled, err := d.IsMigEnabled() + check(err) - if migEnabled && g.skipMigEnabledGPUs { - continue - } + if migEnabled && g.skipMigEnabledGPUs { + continue + } - devs = append(devs, buildDevice(d, []string{d.Path}, fmt.Sprintf("%v", i))) - } + devs = append(devs, buildDevice(d, []string{d.Path}, fmt.Sprintf("%v", i))) + } - return devs + return devs } // CheckHealth performs health checks on a set of devices, writing to the 'unhealthy' channel with any unhealthy devices func (g *GpuDeviceManager) CheckHealth(stop <-chan interface{}, devices []*Device, unhealthy chan<- *Device) { - checkHealth(stop, devices, unhealthy) + checkHealth(stop, devices, unhealthy) } func buildDevice(d *nvml.Device, paths []string, index string) *Device { - dev := Device{} - dev.ID = d.UUID - dev.Health = pluginapi.Healthy - dev.Paths = paths - dev.Index = index - dev.Memory = *d.Memory - if d.CPUAffinity != nil { - dev.Topology = &pluginapi.TopologyInfo{ - Nodes: []*pluginapi.NUMANode{ - { - ID: int64(*(d.CPUAffinity)), - }, - }, - } - } - return &dev + dev := Device{} + dev.ID = d.UUID + dev.Health = pluginapi.Healthy + dev.Paths = paths + dev.Index = index + dev.Memory = *d.Memory + if d.CPUAffinity != nil { + dev.Topology = &pluginapi.TopologyInfo{ + Nodes: []*pluginapi.NUMANode{ + { + ID: int64(*(d.CPUAffinity)), + }, + }, + } + } + return &dev } func checkHealth(stop <-chan interface{}, devices []*Device, unhealthy chan<- *Device) { - disableHealthChecks := strings.ToLower(os.Getenv(envDisableHealthChecks)) - if disableHealthChecks == "all" { - disableHealthChecks = allHealthChecks - } - if strings.Contains(disableHealthChecks, "xids") { - return - } - - eventSet := nvml.NewEventSet() - defer nvml.DeleteEventSet(eventSet) - - for _, d := range devices { - gpu, _, _, err := nvml.ParseMigDeviceUUID(d.ID) - if err != nil { - gpu = d.ID - } - - err = nvml.RegisterEventForDevice(eventSet, nvml.XidCriticalError, gpu) - if err != nil && strings.HasSuffix(err.Error(), "Not Supported") { - log.Printf("Warning: %s is too old to support healthchecking: %s. Marking it unhealthy.", d.ID, err) - unhealthy <- d - continue - } - check(err) - } - - for { - select { - case <-stop: - return - default: - } - - e, err := nvml.WaitForEvent(eventSet, 5000) - if err != nil && e.Etype != nvml.XidCriticalError { - continue - } - - // FIXME: formalize the full list and document it. - // http://docs.nvidia.com/deploy/xid-errors/index.html#topic_4 - // Application errors: the GPU should still be healthy - if e.Edata == 31 || e.Edata == 43 || e.Edata == 45 { - continue - } - - if e.UUID == nil || len(*e.UUID) == 0 { - // All devices are unhealthy - log.Printf("XidCriticalError: Xid=%d, All devices will go unhealthy.", e.Edata) - for _, d := range devices { - unhealthy <- d - } - continue - } - - for _, d := range devices { - // Please see https://github.com/NVIDIA/gpu-monitoring-tools/blob/148415f505c96052cb3b7fdf443b34ac853139ec/bindings/go/nvml/nvml.h#L1424 - // for the rationale why gi and ci can be set as such when the UUID is a full GPU UUID and not a MIG device UUID. - gpu, gi, ci, err := nvml.ParseMigDeviceUUID(d.ID) - if err != nil { - gpu = d.ID - gi = 0xFFFFFFFF - ci = 0xFFFFFFFF - } - - if gpu == *e.UUID && gi == *e.GpuInstanceId && ci == *e.ComputeInstanceId { - log.Printf("XidCriticalError: Xid=%d on Device=%s, the device will go unhealthy.", e.Edata, d.ID) - unhealthy <- d - } - } - } + disableHealthChecks := strings.ToLower(os.Getenv(envDisableHealthChecks)) + if disableHealthChecks == "all" { + disableHealthChecks = allHealthChecks + } + if strings.Contains(disableHealthChecks, "xids") { + return + } + + eventSet := nvml.NewEventSet() + defer nvml.DeleteEventSet(eventSet) + + for _, d := range devices { + gpu, _, _, err := nvml.ParseMigDeviceUUID(d.ID) + if err != nil { + gpu = d.ID + } + + err = nvml.RegisterEventForDevice(eventSet, nvml.XidCriticalError, gpu) + if err != nil && strings.HasSuffix(err.Error(), "Not Supported") { + log.Printf("Warning: %s is too old to support healthchecking: %s. Marking it unhealthy.", d.ID, err) + unhealthy <- d + continue + } + check(err) + } + + for { + select { + case <-stop: + return + default: + } + + e, err := nvml.WaitForEvent(eventSet, 5000) + if err != nil && e.Etype != nvml.XidCriticalError { + continue + } + + // FIXME: formalize the full list and document it. + // http://docs.nvidia.com/deploy/xid-errors/index.html#topic_4 + // Application errors: the GPU should still be healthy + if e.Edata == 31 || e.Edata == 43 || e.Edata == 45 { + continue + } + + if e.UUID == nil || len(*e.UUID) == 0 { + // All devices are unhealthy + log.Printf("XidCriticalError: Xid=%d, All devices will go unhealthy.", e.Edata) + for _, d := range devices { + unhealthy <- d + } + continue + } + + for _, d := range devices { + // Please see https://github.com/NVIDIA/gpu-monitoring-tools/blob/148415f505c96052cb3b7fdf443b34ac853139ec/bindings/go/nvml/nvml.h#L1424 + // for the rationale why gi and ci can be set as such when the UUID is a full GPU UUID and not a MIG device UUID. + gpu, gi, ci, err := nvml.ParseMigDeviceUUID(d.ID) + if err != nil { + gpu = d.ID + gi = 0xFFFFFFFF + ci = 0xFFFFFFFF + } + + if gpu == *e.UUID && gi == *e.GpuInstanceId && ci == *e.ComputeInstanceId { + log.Printf("XidCriticalError: Xid=%d on Device=%s, the device will go unhealthy.", e.Edata, d.ID) + unhealthy <- d + } + } + } } diff --git a/pkg/device-plugin/register.go b/pkg/device-plugin/register.go index 15202b5a8b..81b736208b 100644 --- a/pkg/device-plugin/register.go +++ b/pkg/device-plugin/register.go @@ -17,123 +17,133 @@ package device_plugin import ( - "context" - "fmt" - "k8s.io/klog/v2" - "time" + "context" + "fmt" + "time" - "4pd.io/k8s-vgpu/pkg/api" - "4pd.io/k8s-vgpu/pkg/device-plugin/config" - "google.golang.org/grpc" + "github.com/NVIDIA/gpu-monitoring-tools/bindings/go/nvml" + "k8s.io/klog/v2" + + "4pd.io/k8s-vgpu/pkg/api" + "4pd.io/k8s-vgpu/pkg/device-plugin/config" + "google.golang.org/grpc" ) type DevListFunc func() []*Device type DeviceRegister struct { - deviceCache *DeviceCache - unhealthy chan *Device - stopCh chan struct{} + deviceCache *DeviceCache + unhealthy chan *Device + stopCh chan struct{} } func NewDeviceRegister(deviceCache *DeviceCache) *DeviceRegister { - return &DeviceRegister{ - deviceCache: deviceCache, - unhealthy: make(chan *Device), - stopCh: make(chan struct{}), - } + return &DeviceRegister{ + deviceCache: deviceCache, + unhealthy: make(chan *Device), + stopCh: make(chan struct{}), + } } func (r *DeviceRegister) Start() { - r.deviceCache.AddNotifyChannel("register", r.unhealthy) - go r.WatchAndRegister() + r.deviceCache.AddNotifyChannel("register", r.unhealthy) + go r.WatchAndRegister() } func (r *DeviceRegister) Stop() { - close(r.stopCh) + close(r.stopCh) } func (r *DeviceRegister) apiDevices() *[]*api.DeviceInfo { - devs := r.deviceCache.GetCache() - res := make([]*api.DeviceInfo, 0, len(devs)) - for _, dev := range devs { - res = append(res, &api.DeviceInfo{ - Id: dev.ID, - Count: int32(config.DeviceSplitCount), - Health: dev.Health == "healthy", - }) - } - return &res + devs := r.deviceCache.GetCache() + res := make([]*api.DeviceInfo, 0, len(devs)) + for _, dev := range devs { + ndev, err := nvml.NewDeviceByUUID(dev.ID) + if err != nil { + fmt.Println("nvml new device by uuid error id=", dev.ID) + panic(0) + } else { + fmt.Println("nvml registered device id=", dev.ID, "memory=", *ndev.Memory) + } + res = append(res, &api.DeviceInfo{ + Id: dev.ID, + Count: int32(config.DeviceSplitCount), + Devmem: int32(*ndev.Memory), + Health: dev.Health == "healthy", + }) + } + return &res } func (r *DeviceRegister) Register(ctx context.Context) error { - conn, err := grpc.DialContext( - ctx, - config.SchedulerEndpoint, - grpc.WithInsecure(), - grpc.WithBlock(), - //grpc.WithConnectParams(grpc.ConnectParams{MinConnectTimeout: 3}), - ) - if err != nil { - return fmt.Errorf("connect scheduler error, %v", err) - } - client := api.NewDeviceServiceClient(conn) - register, err := client.Register(ctx) - if err != nil { - klog.Errorf("register error %v", err) - err = fmt.Errorf("client register error, %v", err) - return err - } - req := api.RegisterRequest{Node: config.NodeName, Devices: *r.apiDevices()} - err = register.Send(&req) - if err != nil { - klog.Errorf("register send error, %v", err) - return err - } - klog.V(3).Infof("register info %v", req.String()) - closeCh := make(chan struct{}) - go func() { - reply := api.RegisterReply{} - err := register.RecvMsg(reply) - if err != nil { - klog.Errorf("register recv error, %v", err) - } else { - klog.Errorf("register recv closed") - } - closeCh <- struct{}{} - }() - for { - select { - case <-r.unhealthy: - err = register.Send(&api.RegisterRequest{ - Node: config.NodeName, - Devices: *r.apiDevices(), - }) - if err != nil { - klog.Errorf("register send error, %v", err) - return err - } - klog.V(3).Infof("register info %v", req.String()) - case <-closeCh: - klog.Infof("register server closed") - return fmt.Errorf("register server closed") - case <-r.stopCh: - return nil - } - } + conn, err := grpc.DialContext( + ctx, + config.SchedulerEndpoint, + grpc.WithInsecure(), + grpc.WithBlock(), + //grpc.WithConnectParams(grpc.ConnectParams{MinConnectTimeout: 3}), + ) + if err != nil { + return fmt.Errorf("connect scheduler error, %v", err) + } + client := api.NewDeviceServiceClient(conn) + register, err := client.Register(ctx) + if err != nil { + klog.Errorf("register error %v", err) + err = fmt.Errorf("client register error, %v", err) + return err + } + req := api.RegisterRequest{Node: config.NodeName, Devices: *r.apiDevices()} + err = register.Send(&req) + if err != nil { + klog.Errorf("register send error, %v", err) + return err + } + klog.V(3).Infof("register info %v", req.String()) + closeCh := make(chan struct{}) + go func() { + reply := api.RegisterReply{} + err := register.RecvMsg(reply) + if err != nil { + klog.Errorf("register recv error, %v", err) + } else { + klog.Errorf("register recv closed") + } + closeCh <- struct{}{} + }() + for { + select { + case <-r.unhealthy: + err = register.Send(&api.RegisterRequest{ + Node: config.NodeName, + Devices: *r.apiDevices(), + }) + if err != nil { + klog.Errorf("register send error, %v", err) + return err + } + klog.V(3).Infof("register info %v", req.String()) + case <-closeCh: + klog.Infof("register server closed") + return fmt.Errorf("register server closed") + case <-r.stopCh: + return nil + } + } } func (r *DeviceRegister) WatchAndRegister() { - //ctx, cancel := context.WithTimeout(context.Background(), time.Second*30) - //defer cancel() - ctx := context.Background() - for { - err := r.Register(ctx) - if err != nil { - klog.Errorf("register error, %v", err) - time.Sleep(time.Second * 5) - } else { - klog.Infof("register stopped") - break - } - } + //ctx, cancel := context.WithTimeout(context.Background(), time.Second*30) + //defer cancel() + ctx := context.Background() + for { + err := r.Register(ctx) + if err != nil { + klog.Errorf("register error, %v", err) + time.Sleep(time.Second * 5) + } else { + klog.Infof("register stopped") + break + } + } } diff --git a/pkg/device-plugin/runtime.go b/pkg/device-plugin/runtime.go index df979f3ceb..8e97be24ac 100644 --- a/pkg/device-plugin/runtime.go +++ b/pkg/device-plugin/runtime.go @@ -17,75 +17,86 @@ package device_plugin import ( - "context" - "fmt" - "strings" + "context" + "fmt" + "strconv" - "4pd.io/k8s-vgpu/pkg/api" - "4pd.io/k8s-vgpu/pkg/device-plugin/config" - "google.golang.org/grpc" + "4pd.io/k8s-vgpu/pkg/api" + "4pd.io/k8s-vgpu/pkg/device-plugin/config" + "google.golang.org/grpc" ) type VGPURuntimeService struct { - deviceCache *DeviceCache + deviceCache *DeviceCache } func NewVGPURuntimeService(deviceCache *DeviceCache) *VGPURuntimeService { - return &VGPURuntimeService{deviceCache: deviceCache} + return &VGPURuntimeService{deviceCache: deviceCache} } func (s *VGPURuntimeService) GetDevice(ctx context.Context, req *api.GetDeviceRequest) (*api.GetDeviceReply, error) { - conn, err := grpc.DialContext( - ctx, - config.SchedulerEndpoint, - grpc.WithInsecure(), - grpc.WithBlock(), - //grpc.WithConnectParams(grpc.ConnectParams{MinConnectTimeout: 3}), - ) - if err != nil { - return nil, fmt.Errorf("connect scheduler error, %v", err) - } - client := api.NewDeviceServiceClient(conn) - sReq := api.GetContainerRequest{Uuid: req.CtrUUID} - sResp, err := client.GetContainer(ctx, &sReq) - if err != nil { - return nil, err - } - envs, err := s.containerEnvs(sResp.DevList) - if err != nil { - return nil, err - } - resp := api.GetDeviceReply{ - Envs: envs, - PodUID: sResp.PodUID, - CtrName: sResp.CtrName, - PodNamespace: sResp.PodNamespace, - PodName: sResp.PodName, - } - return &resp, nil + conn, err := grpc.DialContext( + ctx, + config.SchedulerEndpoint, + grpc.WithInsecure(), + grpc.WithBlock(), + //grpc.WithConnectParams(grpc.ConnectParams{MinConnectTimeout: 3}), + ) + if err != nil { + return nil, fmt.Errorf("connect scheduler error, %v", err) + } + client := api.NewDeviceServiceClient(conn) + sReq := api.GetContainerRequest{Uuid: req.CtrUUID} + sResp, err := client.GetContainer(ctx, &sReq) + if err != nil { + return nil, err + } + envs, err := s.containerEnvs(sResp.DevList) + if err != nil { + return nil, err + } + resp := api.GetDeviceReply{ + Envs: envs, + PodUID: sResp.PodUID, + CtrName: sResp.CtrName, + PodNamespace: sResp.PodNamespace, + PodName: sResp.PodName, + } + return &resp, nil } -func (s *VGPURuntimeService) containerEnvs(devIDs []string) (map[string]string, error) { - envs := make(map[string]string) - var devs []*Device - for _, id := range devIDs { - found := false - for _, d := range s.deviceCache.GetCache() { - if id == d.ID { - found = true - devs = append(devs, d) - break - } - } - if !found { - return nil, fmt.Errorf("device %v not found", id) - } - } +func (s *VGPURuntimeService) containerEnvs(devIDs []*api.DeviceUsage) (map[string]string, error) { + envs := make(map[string]string) + var devs []*Device + for _, id := range devIDs { + found := false + for _, d := range s.deviceCache.GetCache() { + if id.GetId() == d.ID { + found = true + devs = append(devs, d) + break + } + } + if !found { + return nil, fmt.Errorf("device %v not found", id) + } + } - envs["NVIDIA_VISIBLE_DEVICES"] = strings.Join(devIDs, ",") - for i, d := range devs { - limitKey := fmt.Sprintf("CUDA_DEVICE_MEMORY_LIMIT_%v", i) - envs[limitKey] = fmt.Sprintf("%vm", config.DeviceMemoryScaling*float64(d.Memory)/float64(config.DeviceSplitCount)) - } - return envs, nil + devenv := "" + for idx, val := range devIDs { + if idx == 0 { + devenv = "" + val.GetId() + } else { + devenv = devenv + "," + val.GetId() + } + } + //envs["NVIDIA_VISIBLE_DEVICES"] = strings.Join(devIDs, ",") + //fmt.Println("Assigneing NVIDIA_VISIBLE_DEVICES:", devenv) + envs["NVIDIA_VISIBLE_DEVICES"] = devenv + for i, d := range devIDs { + limitKey := fmt.Sprintf("CUDA_DEVICE_MEMORY_LIMIT_%v", i) + envs[limitKey] = strconv.Itoa(int(d.GetDevmem())) + "m" + //envs[limitKey] = fmt.Sprintf("%vm", config.DeviceMemoryScaling*float64(d.Memory)/float64(config.DeviceSplitCount)) + } + return envs, nil } diff --git a/pkg/k8sutil/pod.go b/pkg/k8sutil/pod.go index 90bee783a8..4678ef86ad 100644 --- a/pkg/k8sutil/pod.go +++ b/pkg/k8sutil/pod.go @@ -17,29 +17,75 @@ package k8sutil import ( - corev1 "k8s.io/api/core/v1" + "4pd.io/k8s-vgpu/pkg/util" + corev1 "k8s.io/api/core/v1" ) +func Resourcereqs(pod *corev1.Pod) (counts []util.ContainerDeviceRequest) { + resourceName := corev1.ResourceName(util.ResourceName) + resourceMem := corev1.ResourceName(util.ResourceMem) + resourceCores := corev1.ResourceName(util.ResourceCores) + counts = make([]util.ContainerDeviceRequest, len(pod.Spec.Containers)) + for i := 0; i < len(pod.Spec.Containers); i++ { + v, ok := pod.Spec.Containers[i].Resources.Limits[resourceName] + if !ok { + v, ok = pod.Spec.Containers[i].Resources.Requests[resourceName] + } + if ok { + if n, ok := v.AsInt64(); ok { + memnum := util.DefaultMem + mem, ok := pod.Spec.Containers[i].Resources.Limits[resourceMem] + if !ok { + mem, ok = pod.Spec.Containers[i].Resources.Requests[resourceMem] + } + if ok { + memnums, ok := mem.AsInt64() + if ok { + memnum = int(memnums) + } + } + corenum := util.DefaultCores + core, ok := pod.Spec.Containers[i].Resources.Limits[resourceCores] + if !ok { + core, ok = pod.Spec.Containers[i].Resources.Requests[resourceCores] + } + if ok { + corenums, ok := core.AsInt64() + if ok { + corenum = int(corenums) + } + } + counts[i] = util.ContainerDeviceRequest{ + Nums: int32(n), + Memreq: int32(memnum), + Coresreq: int32(corenum), + } + } + } + } + return counts +} + func ResourceNums(pod *corev1.Pod, resourceName corev1.ResourceName) (counts []int) { - counts = make([]int, len(pod.Spec.Containers)) - for i := 0; i < len(pod.Spec.Containers); i++ { - v, ok := pod.Spec.Containers[i].Resources.Limits[resourceName] - if !ok { - v, ok = pod.Spec.Containers[i].Resources.Requests[resourceName] - } - if ok { - if n, ok := v.AsInt64(); ok { - counts[i] = int(n) - } - } - } - return counts + counts = make([]int, len(pod.Spec.Containers)) + for i := 0; i < len(pod.Spec.Containers); i++ { + v, ok := pod.Spec.Containers[i].Resources.Limits[resourceName] + if !ok { + v, ok = pod.Spec.Containers[i].Resources.Requests[resourceName] + } + if ok { + if n, ok := v.AsInt64(); ok { + counts[i] = int(n) + } + } + } + return counts } func IsPodInTerminatedState(pod *corev1.Pod) bool { - return pod.Status.Phase == corev1.PodFailed || pod.Status.Phase == corev1.PodSucceeded + return pod.Status.Phase == corev1.PodFailed || pod.Status.Phase == corev1.PodSucceeded } func AllContainersCreated(pod *corev1.Pod) bool { - return len(pod.Status.ContainerStatuses) >= len(pod.Spec.Containers) + return len(pod.Status.ContainerStatuses) >= len(pod.Spec.Containers) } diff --git a/pkg/scheduler/nodes.go b/pkg/scheduler/nodes.go index 8b1241888f..aa7b6f56df 100644 --- a/pkg/scheduler/nodes.go +++ b/pkg/scheduler/nodes.go @@ -17,60 +17,64 @@ package scheduler import ( - "fmt" - "sync" + "fmt" + "sync" ) type DeviceInfo struct { - ID string - Count int32 - Health bool + ID string + Count int32 + Devmem int32 + Health bool } type NodeInfo struct { - ID string - Devices []DeviceInfo + ID string + Devices []DeviceInfo } type DeviceUsage struct { - id string - used int32 - count int32 - health bool + id string + used int32 + count int32 + usedmem int32 + totalmem int32 + usedcores int32 + health bool } type DeviceUsageList []*DeviceUsage type NodeUsage struct { - devices DeviceUsageList + devices DeviceUsageList } type nodeManager struct { - nodes map[string]NodeInfo - mutex sync.Mutex + nodes map[string]NodeInfo + mutex sync.Mutex } func (m *nodeManager) init() { - m.nodes = make(map[string]NodeInfo) + m.nodes = make(map[string]NodeInfo) } func (m *nodeManager) addNode(nodeID string, nodeInfo NodeInfo) { - m.mutex.Lock() - defer m.mutex.Unlock() - m.nodes[nodeID] = nodeInfo + m.mutex.Lock() + defer m.mutex.Unlock() + m.nodes[nodeID] = nodeInfo } func (m *nodeManager) delNode(nodeID string) { - m.mutex.Lock() - defer m.mutex.Unlock() - delete(m.nodes, nodeID) + m.mutex.Lock() + defer m.mutex.Unlock() + delete(m.nodes, nodeID) } func (m *nodeManager) GetNode(nodeID string) (NodeInfo, error) { - m.mutex.Lock() - defer m.mutex.Unlock() - if n, ok := m.nodes[nodeID]; ok { - return n, nil - } - return NodeInfo{}, fmt.Errorf("node %v not found", nodeID) + m.mutex.Lock() + defer m.mutex.Unlock() + if n, ok := m.nodes[nodeID]; ok { + return n, nil + } + return NodeInfo{}, fmt.Errorf("node %v not found", nodeID) } diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go index 455a3dd0eb..7a7467f360 100644 --- a/pkg/scheduler/scheduler.go +++ b/pkg/scheduler/scheduler.go @@ -17,112 +17,112 @@ package scheduler import ( - "context" - "encoding/json" - "fmt" - "sort" - "strconv" - "time" + "context" + "encoding/json" + "fmt" + "sort" + "strconv" + "time" - "4pd.io/k8s-vgpu/pkg/api" - "4pd.io/k8s-vgpu/pkg/k8sutil" - "4pd.io/k8s-vgpu/pkg/util" - corev1 "k8s.io/api/core/v1" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - k8stypes "k8s.io/apimachinery/pkg/types" - "k8s.io/client-go/informers" - "k8s.io/client-go/kubernetes" - listerscorev1 "k8s.io/client-go/listers/core/v1" - "k8s.io/client-go/tools/cache" - "k8s.io/klog/v2" - extenderv1 "k8s.io/kube-scheduler/extender/v1" + "4pd.io/k8s-vgpu/pkg/api" + "4pd.io/k8s-vgpu/pkg/k8sutil" + "4pd.io/k8s-vgpu/pkg/util" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + k8stypes "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/informers" + "k8s.io/client-go/kubernetes" + listerscorev1 "k8s.io/client-go/listers/core/v1" + "k8s.io/client-go/tools/cache" + "k8s.io/klog/v2" + extenderv1 "k8s.io/kube-scheduler/extender/v1" ) type Scheduler struct { - nodeManager - podManager + nodeManager + podManager - stopCh chan struct{} - kubeClient kubernetes.Interface - podLister listerscorev1.PodLister - nodeLister listerscorev1.NodeLister + stopCh chan struct{} + kubeClient kubernetes.Interface + podLister listerscorev1.PodLister + nodeLister listerscorev1.NodeLister } func NewScheduler() *Scheduler { - s := &Scheduler{ - stopCh: make(chan struct{}), - } - s.nodeManager.init() - s.podManager.init() - return s + s := &Scheduler{ + stopCh: make(chan struct{}), + } + s.nodeManager.init() + s.podManager.init() + return s } func check(err error) { - if err != nil { - klog.Fatal(err) - } + if err != nil { + klog.Fatal(err) + } } func (s *Scheduler) onAddPod(obj interface{}) { - pod, ok := obj.(*corev1.Pod) - if !ok { - klog.Errorf("unknown add object type") - return - } - nodeID, ok := pod.Annotations[util.AssignedNodeAnnotations] - if !ok { - return - } - ids, ok := pod.Annotations[util.AssignedIDsAnnotations] - if !ok { - return - } - if k8sutil.IsPodInTerminatedState(pod) { - s.delPod(pod) - return - } - podDev := util.DecodePodDevices(ids) - s.addPod(pod, nodeID, podDev) + pod, ok := obj.(*corev1.Pod) + if !ok { + klog.Errorf("unknown add object type") + return + } + nodeID, ok := pod.Annotations[util.AssignedNodeAnnotations] + if !ok { + return + } + ids, ok := pod.Annotations[util.AssignedIDsAnnotations] + if !ok { + return + } + if k8sutil.IsPodInTerminatedState(pod) { + s.delPod(pod) + return + } + podDev := util.DecodePodDevices(ids) + s.addPod(pod, nodeID, podDev) } func (s *Scheduler) onUpdatePod(_, newObj interface{}) { - s.onAddPod(newObj) + s.onAddPod(newObj) } func (s *Scheduler) onDelPod(obj interface{}) { - pod, ok := obj.(*corev1.Pod) - if !ok { - klog.Errorf("unknown add object type") - return - } - _, ok = pod.Annotations[util.AssignedNodeAnnotations] - if !ok { - return - } - s.delPod(pod) + pod, ok := obj.(*corev1.Pod) + if !ok { + klog.Errorf("unknown add object type") + return + } + _, ok = pod.Annotations[util.AssignedNodeAnnotations] + if !ok { + return + } + s.delPod(pod) } func (s *Scheduler) Start() { - kubeClient, err := k8sutil.NewClient() - check(err) - s.kubeClient = kubeClient - informerFactory := informers.NewSharedInformerFactoryWithOptions(s.kubeClient, time.Hour*1) - s.podLister = informerFactory.Core().V1().Pods().Lister() - s.nodeLister = informerFactory.Core().V1().Nodes().Lister() + kubeClient, err := k8sutil.NewClient() + check(err) + s.kubeClient = kubeClient + informerFactory := informers.NewSharedInformerFactoryWithOptions(s.kubeClient, time.Hour*1) + s.podLister = informerFactory.Core().V1().Pods().Lister() + s.nodeLister = informerFactory.Core().V1().Nodes().Lister() - informer := informerFactory.Core().V1().Pods().Informer() - informer.AddEventHandler(cache.ResourceEventHandlerFuncs{ - AddFunc: s.onAddPod, - UpdateFunc: s.onUpdatePod, - DeleteFunc: s.onDelPod, - }) + informer := informerFactory.Core().V1().Pods().Informer() + informer.AddEventHandler(cache.ResourceEventHandlerFuncs{ + AddFunc: s.onAddPod, + UpdateFunc: s.onUpdatePod, + DeleteFunc: s.onDelPod, + }) - informerFactory.Start(s.stopCh) - informerFactory.WaitForCacheSync(s.stopCh) + informerFactory.Start(s.stopCh) + informerFactory.WaitForCacheSync(s.stopCh) } func (s *Scheduler) Stop() { - close(s.stopCh) + close(s.stopCh) } //func (s *Scheduler) assignedNode(pod *corev1.Pod) string { @@ -132,164 +132,179 @@ func (s *Scheduler) Stop() { // return "" //} func (s *Scheduler) Register(stream api.DeviceService_RegisterServer) error { - var nodeID string - for { - req, err := stream.Recv() - if err != nil { - s.delNode(nodeID) - klog.Infof("node %v leave, %v", nodeID, err) - _ = stream.SendAndClose(&api.RegisterReply{}) - return err - } - klog.V(3).Infof("device register %v", req.String()) - nodeID = req.GetNode() - nodeInfo := NodeInfo{} - nodeInfo.ID = nodeID - nodeInfo.Devices = make([]DeviceInfo, len(req.Devices)) - for i := 0; i < len(req.Devices); i++ { - nodeInfo.Devices[i] = DeviceInfo{ - ID: req.Devices[i].GetId(), - Count: req.Devices[i].GetCount(), - Health: req.Devices[i].GetHealth(), - } - } - s.addNode(nodeID, nodeInfo) - klog.Infof("node %v come", nodeID) - } + var nodeID string + for { + req, err := stream.Recv() + if err != nil { + s.delNode(nodeID) + klog.Infof("node %v leave, %v", nodeID, err) + _ = stream.SendAndClose(&api.RegisterReply{}) + return err + } + klog.V(3).Infof("device register %v", req.String()) + nodeID = req.GetNode() + nodeInfo := NodeInfo{} + nodeInfo.ID = nodeID + nodeInfo.Devices = make([]DeviceInfo, len(req.Devices)) + for i := 0; i < len(req.Devices); i++ { + nodeInfo.Devices[i] = DeviceInfo{ + ID: req.Devices[i].GetId(), + Count: req.Devices[i].GetCount(), + Devmem: req.Devices[i].GetDevmem(), + Health: req.Devices[i].GetHealth(), + } + } + s.addNode(nodeID, nodeInfo) + klog.Infof("node %v come node info=", nodeID, nodeInfo) + } } func (s *Scheduler) GetContainer(_ context.Context, req *api.GetContainerRequest) (*api.GetContainerReply, error) { - pi, ctrIdx, err := s.getContainerByUUID(req.Uuid) - if err != nil { - return nil, err - } - if ctrIdx >= len(pi.devices) { - return nil, fmt.Errorf("container index error") - } - pod, err := s.podLister.Pods(pi.namespace).Get(pi.name) - if err != nil { - return nil, err - } - if pod == nil || ctrIdx >= len(pi.devices) { - return nil, fmt.Errorf("container not found") - } - rep := api.GetContainerReply{ - DevList: pi.devices[ctrIdx], - PodUID: string(pod.UID), - CtrName: pod.Spec.Containers[ctrIdx].Name, - PodNamespace: pod.Namespace, - PodName: pod.Name, - } - return &rep, nil + pi, ctrIdx, err := s.getContainerByUUID(req.Uuid) + if err != nil { + return nil, err + } + if ctrIdx >= len(pi.devices) { + return nil, fmt.Errorf("container index error") + } + pod, err := s.podLister.Pods(pi.namespace).Get(pi.name) + if err != nil { + return nil, err + } + if pod == nil || ctrIdx >= len(pi.devices) { + return nil, fmt.Errorf("container not found") + } + var devarray []*api.DeviceUsage + for _, val := range pi.devices[ctrIdx] { + devusage := api.DeviceUsage{} + devusage.Id = val.UUID + devusage.Devmem = val.Usedmem + devusage.Cores = val.Usedcores + devarray = append(devarray, &devusage) + } + rep := api.GetContainerReply{ + //DevList: pi.devices[ctrIdx], + DevList: devarray, + PodUID: string(pod.UID), + CtrName: pod.Spec.Containers[ctrIdx].Name, + PodNamespace: pod.Namespace, + PodName: pod.Name, + } + return &rep, nil } func (s *Scheduler) getNodesUsage(nodes *[]string) (*map[string]*NodeUsage, map[string]string, error) { - nodeMap := make(map[string]*NodeUsage) - failedNodes := make(map[string]string) - for _, nodeID := range *nodes { - node, err := s.GetNode(nodeID) - if err != nil { - klog.Errorf("get node %v device error, %v", nodeID, err) - failedNodes[nodeID] = fmt.Sprintf("node unregisterd") - continue - } + nodeMap := make(map[string]*NodeUsage) + failedNodes := make(map[string]string) + for _, nodeID := range *nodes { + node, err := s.GetNode(nodeID) + if err != nil { + klog.Errorf("get node %v device error, %v", nodeID, err) + failedNodes[nodeID] = fmt.Sprintf("node unregisterd") + continue + } - nodeInfo := &NodeUsage{} - for _, d := range node.Devices { - nodeInfo.devices = append(nodeInfo.devices, &DeviceUsage{ - id: d.ID, - used: 0, - count: d.Count, - health: d.Health, - }) - } - nodeMap[nodeID] = nodeInfo - } - for _, p := range s.pods { - node, ok := nodeMap[p.nodeID] - if !ok { - continue - } - for _, ds := range p.devices { - for _, deviceID := range ds { - for _, d := range node.devices { - if d.id == deviceID { - d.used++ - } - } - } - } - klog.V(5).Infof("usage: pod %v assigned %v %v", p.name, p.nodeID, p.devices) - } - return &nodeMap, failedNodes, nil + nodeInfo := &NodeUsage{} + for _, d := range node.Devices { + nodeInfo.devices = append(nodeInfo.devices, &DeviceUsage{ + id: d.ID, + used: 0, + count: d.Count, + usedmem: 0, + totalmem: d.Devmem, + usedcores: 0, + health: d.Health, + }) + } + nodeMap[nodeID] = nodeInfo + } + for _, p := range s.pods { + node, ok := nodeMap[p.nodeID] + if !ok { + continue + } + for _, ds := range p.devices { + for _, udevice := range ds { + for _, d := range node.devices { + if d.id == udevice.UUID { + d.used++ + d.usedmem += udevice.Usedmem + d.usedcores += udevice.Usedcores + } + } + } + } + klog.V(5).Infof("usage: pod %v assigned %v %v", p.name, p.nodeID, p.devices) + } + return &nodeMap, failedNodes, nil } func (s *Scheduler) Filter(args extenderv1.ExtenderArgs) (*extenderv1.ExtenderFilterResult, error) { - klog.Infof("schedule pod %v/%v[%v]", args.Pod.Namespace, args.Pod.Name, args.Pod.UID) - nums := k8sutil.ResourceNums(args.Pod, corev1.ResourceName(util.ResourceName)) - total := 0 - for _, n := range nums { - total += n - } - if total == 0 { - klog.V(1).Infof("pod %v not find resource %v", args.Pod.Name, util.ResourceName) - return &extenderv1.ExtenderFilterResult{ - NodeNames: args.NodeNames, - FailedNodes: nil, - Error: "", - }, nil - } - s.delPod(args.Pod) - nodeUsage, failedNodes, err := s.getNodesUsage(args.NodeNames) - if err != nil { - return nil, err - } - nodeScores, err := calcScore(nodeUsage, &failedNodes, nums) - if err != nil { - return nil, err - } - if len(*nodeScores) == 0 { - return &extenderv1.ExtenderFilterResult{ - FailedNodes: failedNodes, - }, nil - } - sort.Sort(nodeScores) - m := (*nodeScores)[len(*nodeScores)-1] - klog.Infof("schedule %v/%v to %v %v", args.Pod.Namespace, args.Pod.Name, m.nodeID, m.devices) - annotations := make(map[string]string) - annotations[util.AssignedNodeAnnotations] = m.nodeID - annotations[util.AssignedTimeAnnotations] = strconv.FormatInt(time.Now().Unix(), 10) - annotations[util.AssignedIDsAnnotations] = util.EncodePodDevices(m.devices) - s.addPod(args.Pod, m.nodeID, m.devices) - err = s.patchPodAnnotations(args.Pod, annotations) - if err != nil { - s.delPod(args.Pod) - return nil, err - } - res := extenderv1.ExtenderFilterResult{NodeNames: &[]string{m.nodeID}} - return &res, nil + klog.Infof("schedule pod %v/%v[%v]", args.Pod.Namespace, args.Pod.Name, args.Pod.UID) + nums := k8sutil.Resourcereqs(args.Pod) + total := 0 + for _, n := range nums { + total += int(n.Nums) + } + if total == 0 { + klog.V(1).Infof("pod %v not find resource %v", args.Pod.Name, util.ResourceName) + return &extenderv1.ExtenderFilterResult{ + NodeNames: args.NodeNames, + FailedNodes: nil, + Error: "", + }, nil + } + s.delPod(args.Pod) + nodeUsage, failedNodes, err := s.getNodesUsage(args.NodeNames) + if err != nil { + return nil, err + } + nodeScores, err := calcScore(nodeUsage, &failedNodes, nums) + if err != nil { + return nil, err + } + if len(*nodeScores) == 0 { + return &extenderv1.ExtenderFilterResult{ + FailedNodes: failedNodes, + }, nil + } + sort.Sort(nodeScores) + m := (*nodeScores)[len(*nodeScores)-1] + klog.Infof("schedule %v/%v to %v %v", args.Pod.Namespace, args.Pod.Name, m.nodeID, m.devices) + annotations := make(map[string]string) + annotations[util.AssignedNodeAnnotations] = m.nodeID + annotations[util.AssignedTimeAnnotations] = strconv.FormatInt(time.Now().Unix(), 10) + annotations[util.AssignedIDsAnnotations] = util.EncodePodDevices(m.devices) + s.addPod(args.Pod, m.nodeID, m.devices) + err = s.patchPodAnnotations(args.Pod, annotations) + if err != nil { + s.delPod(args.Pod) + return nil, err + } + res := extenderv1.ExtenderFilterResult{NodeNames: &[]string{m.nodeID}} + return &res, nil } func (s *Scheduler) patchPodAnnotations(pod *corev1.Pod, annotations map[string]string) error { - type patchMetadata struct { - Annotations map[string]string `json:"annotations,omitempty"` - } - type patchPod struct { - Metadata patchMetadata `json:"metadata"` - //Spec patchSpec `json:"spec,omitempty"` - } + type patchMetadata struct { + Annotations map[string]string `json:"annotations,omitempty"` + } + type patchPod struct { + Metadata patchMetadata `json:"metadata"` + //Spec patchSpec `json:"spec,omitempty"` + } - p := patchPod{} - p.Metadata.Annotations = annotations + p := patchPod{} + p.Metadata.Annotations = annotations - bytes, err := json.Marshal(p) - if err != nil { - return err - } - _, err = s.kubeClient.CoreV1().Pods(pod.Namespace). - Patch(context.Background(), pod.Name, k8stypes.StrategicMergePatchType, bytes, metav1.PatchOptions{}) - if err != nil { - klog.Infof("patch pod %v failed, %v", pod.Name, err) - } - return err + bytes, err := json.Marshal(p) + if err != nil { + return err + } + _, err = s.kubeClient.CoreV1().Pods(pod.Namespace). + Patch(context.Background(), pod.Name, k8stypes.StrategicMergePatchType, bytes, metav1.PatchOptions{}) + if err != nil { + klog.Infof("patch pod %v failed, %v", pod.Name, err) + } + return err } diff --git a/pkg/scheduler/score.go b/pkg/scheduler/score.go index 972e578b66..b05469c85e 100644 --- a/pkg/scheduler/score.go +++ b/pkg/scheduler/score.go @@ -17,6 +17,7 @@ package scheduler import ( + "fmt" "sort" "4pd.io/k8s-vgpu/pkg/util" @@ -54,38 +55,67 @@ func (l NodeScoreList) Less(i, j int) bool { return l[i].score < l[j].score } -func calcScore(nodes *map[string]*NodeUsage, errMap *map[string]string, nums []int) (*NodeScoreList, error) { +func viewStatus(usage NodeUsage) { + fmt.Println("viewing status") + for _, val := range usage.devices { + fmt.Println(val) + } +} + +func calcScore(nodes *map[string]*NodeUsage, errMap *map[string]string, nums []util.ContainerDeviceRequest) (*NodeScoreList, error) { res := make(NodeScoreList, 0, len(*nodes)) for nodeID, node := range *nodes { + viewStatus(*node) dn := len(node.devices) score := NodeScore{nodeID: nodeID, score: 0} for _, n := range nums { - if n == 0 { - score.devices = append(score.devices, []string{}) + if n.Nums == 0 { + score.devices = append(score.devices, util.ContainerDevices{}) continue } - if n > dn { + if int(n.Nums) > dn { break } sort.Sort(node.devices) - if node.devices[dn-n].count <= node.devices[dn-n].used { + if node.devices[dn-int(n.Nums)].count <= node.devices[dn-int(n.Nums)].used { break } total := int32(0) free := int32(0) - devs := make([]string, 0, n) + //devs := make([]string, 0, n) + devs := make([]util.ContainerDevice, 0, n.Nums) + countremains := 1 for i := len(node.devices) - 1; i >= 0; i-- { + if node.devices[i].count <= node.devices[i].used { + countremains = 0 + break + } + if node.devices[i].totalmem-node.devices[i].usedmem < n.Memreq { + continue + } + if 100-node.devices[i].usedcores < n.Coresreq { + continue + } total += node.devices[i].count free += node.devices[i].count - node.devices[i].used - if n > 0 { - n-- + if n.Nums > 0 { + n.Nums-- node.devices[i].used++ - devs = append(devs, node.devices[i].id) + node.devices[i].usedmem += n.Memreq + node.devices[i].usedcores += n.Coresreq + devs = append(devs, util.ContainerDevice{ + UUID: node.devices[i].id, + Usedmem: n.Memreq, + Usedcores: n.Coresreq, + }) } } + if countremains == 0 || n.Nums > 0 { + break + } score.devices = append(score.devices, devs) score.score += float32(free) / float32(total) - score.score += float32(dn - n) + score.score += float32(dn - int(n.Nums)) } if len(score.devices) == len(nums) { res = append(res, &score) diff --git a/pkg/scheduler/webhook.go b/pkg/scheduler/webhook.go index e32d7fcca4..5b151b35b4 100644 --- a/pkg/scheduler/webhook.go +++ b/pkg/scheduler/webhook.go @@ -17,75 +17,79 @@ package scheduler import ( - "context" - "encoding/json" - "fmt" - "net/http" + "context" + "encoding/json" + "fmt" + "net/http" - "4pd.io/k8s-vgpu/pkg/api" - "4pd.io/k8s-vgpu/pkg/k8sutil" - "4pd.io/k8s-vgpu/pkg/scheduler/config" - "4pd.io/k8s-vgpu/pkg/util" - corev1 "k8s.io/api/core/v1" - "k8s.io/apimachinery/pkg/runtime" - clientgoscheme "k8s.io/client-go/kubernetes/scheme" - "k8s.io/klog/v2" - "k8s.io/klog/v2/klogr" - "sigs.k8s.io/controller-runtime/pkg/webhook/admission" + "4pd.io/k8s-vgpu/pkg/api" + "4pd.io/k8s-vgpu/pkg/k8sutil" + "4pd.io/k8s-vgpu/pkg/scheduler/config" + "4pd.io/k8s-vgpu/pkg/util" + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/runtime" + clientgoscheme "k8s.io/client-go/kubernetes/scheme" + "k8s.io/klog/v2/klogr" + "sigs.k8s.io/controller-runtime/pkg/webhook/admission" ) type webhook struct { - decoder *admission.Decoder + decoder *admission.Decoder } func NewWebHook() (*admission.Webhook, error) { - schema := runtime.NewScheme() - if err := clientgoscheme.AddToScheme(schema); err != nil { - return nil, err - } - decoder, err := admission.NewDecoder(schema) - if err != nil { - return nil, err - } - wh := &admission.Webhook{Handler: &webhook{decoder: decoder}} - _ = wh.InjectLogger(klogr.New()) - return wh, nil + schema := runtime.NewScheme() + if err := clientgoscheme.AddToScheme(schema); err != nil { + return nil, err + } + decoder, err := admission.NewDecoder(schema) + if err != nil { + return nil, err + } + wh := &admission.Webhook{Handler: &webhook{decoder: decoder}} + _ = wh.InjectLogger(klogr.New()) + return wh, nil } func (h *webhook) Handle(_ context.Context, req admission.Request) admission.Response { - pod := &corev1.Pod{} - err := h.decoder.Decode(req, pod) - if err != nil { - return admission.Errored(http.StatusBadRequest, err) - } - if len(pod.Spec.Containers) == 0 { - return admission.Denied("pod has no containers") - } - klog.V(1).Infof("hook %v pod %v/%v", req.UID, req.Namespace, req.Name) - nums := k8sutil.ResourceNums(pod, corev1.ResourceName(util.ResourceName)) - total := 0 - // use request uid - uid := req.UID - for i := 0; i < len(nums); i++ { - if nums[i] == 0 { - continue - } - total += nums[i] - c := &pod.Spec.Containers[i] - c.Env = append(c.Env, corev1.EnvVar{ - Name: api.ContainerUID, - Value: fmt.Sprintf("%v/%v", uid, c.Name), - }) - } - if total == 0 { - return admission.Allowed(fmt.Sprintf("no resource %v", util.ResourceName)) - } - if len(config.SchedulerName) > 0 { - pod.Spec.SchedulerName = config.SchedulerName - } - marshaledPod, err := json.Marshal(pod) - if err != nil { - return admission.Errored(http.StatusInternalServerError, err) - } - return admission.PatchResponseFromRaw(req.Object.Raw, marshaledPod) + pod := &corev1.Pod{} + err := h.decoder.Decode(req, pod) + if err != nil { + return admission.Errored(http.StatusBadRequest, err) + } + if len(pod.Spec.Containers) == 0 { + return admission.Denied("pod has no containers") + } + //klog.V(1).Infof("hook %v pod %v/%v", req.UID, req.Namespace, req.Name) + fmt.Printf("hook %v pod %v/%v", req.UID, req.Namespace, req.Name) + nums := k8sutil.ResourceNums(pod, corev1.ResourceName(util.ResourceName)) + //gpu_mems := k8sutil.ResourceNums(pod, corev1.ResourceName("nvidia.com/gpu_device_memory")) + total := 0 + // use request uid + uid := req.UID + for i := 0; i < len(nums); i++ { + if nums[i] == 0 { + continue + } + //if gpu_mems[i] != 0 { + // fmt.Println("gpu_mem limit is", gpu_mems[i]) + //} + total += nums[i] + c := &pod.Spec.Containers[i] + c.Env = append(c.Env, corev1.EnvVar{ + Name: api.ContainerUID, + Value: fmt.Sprintf("%v/%v", uid, c.Name), + }) + } + if total == 0 { + return admission.Allowed(fmt.Sprintf("no resource %v", util.ResourceName)) + } + if len(config.SchedulerName) > 0 { + pod.Spec.SchedulerName = config.SchedulerName + } + marshaledPod, err := json.Marshal(pod) + if err != nil { + return admission.Errored(http.StatusInternalServerError, err) + } + return admission.PatchResponseFromRaw(req.Object.Raw, marshaledPod) } diff --git a/pkg/util/types.go b/pkg/util/types.go index 941206f57a..79f75af7b9 100644 --- a/pkg/util/types.go +++ b/pkg/util/types.go @@ -17,19 +17,26 @@ package util const ( - //ResourceName = "nvidia.com/gpu" - //ResourceName = "4pd.io/vgpu" - AssignedTimeAnnotations = "4pd.io/vgpu-time" - AssignedIDsAnnotations = "4pd.io/vgpu-ids" - AssignedNodeAnnotations = "4pd.io/vgpu-node" - - //TimeLayout = "ANSIC" - //DefaultTimeout = time.Second * 60 + //ResourceName = "nvidia.com/gpu" + //ResourceName = "4pd.io/vgpu" + AssignedTimeAnnotations = "4pd.io/vgpu-time" + AssignedIDsAnnotations = "4pd.io/vgpu-ids-new" + AssignedNodeAnnotations = "4pd.io/vgpu-node" + + //Set default mem to 5000m + DefaultMem = 5000 + DefaultCores = 0 + + DeviceLimit = 2 + //TimeLayout = "ANSIC" + //DefaultTimeout = time.Second * 60 ) var ( - ResourceName string - DebugMode bool + ResourceName string + ResourceMem string + ResourceCores string + DebugMode bool ) //type ContainerDevices struct { @@ -39,7 +46,18 @@ var ( //type PodDevices struct { // Containers []ContainerDevices `json:"containers,omitempty"` //} +type ContainerDevice struct { + UUID string + Usedmem int32 + Usedcores int32 +} + +type ContainerDeviceRequest struct { + Nums int32 + Memreq int32 + Coresreq int32 +} -type ContainerDevices []string +type ContainerDevices []ContainerDevice type PodDevices []ContainerDevices diff --git a/pkg/util/util.go b/pkg/util/util.go index 85d4833801..e7a554b969 100644 --- a/pkg/util/util.go +++ b/pkg/util/util.go @@ -17,48 +17,78 @@ package util import ( - "flag" - "os" - "strings" + "flag" + "fmt" + "os" + "strconv" + "strings" - "k8s.io/klog/v2" + "k8s.io/klog/v2" ) func GlobalFlagSet() *flag.FlagSet { - fs := flag.NewFlagSet(os.Args[0], flag.ExitOnError) - fs.StringVar(&ResourceName, "resource-name", "nvidia.com/gpu", "resource name") - fs.BoolVar(&DebugMode, "debug", false, "debug mode") - klog.InitFlags(fs) - return fs + fs := flag.NewFlagSet(os.Args[0], flag.ExitOnError) + fs.StringVar(&ResourceName, "resource-name", "nvidia.com/gpu", "resource name") + fs.StringVar(&ResourceMem, "resource-mem", "nvidia.com/gpumem", "gpu memory to allocate") + fs.StringVar(&ResourceCores, "resource-cores", "nvidia.com/gpucores", "cores percentage to use") + fs.BoolVar(&DebugMode, "debug", false, "debug mode") + klog.InitFlags(fs) + return fs } func EncodeContainerDevices(cd ContainerDevices) string { - return strings.Join(cd, ",") + tmp := "" + for _, val := range cd { + tmp += val.UUID + "," + strconv.Itoa(int(val.Usedmem)) + "," + strconv.Itoa(int(val.Usedcores)) + ":" + } + fmt.Println("Encoded container Devices=", tmp) + return tmp + //return strings.Join(cd, ",") } func EncodePodDevices(pd PodDevices) string { - var ss []string - for _, cd := range pd { - ss = append(ss, EncodeContainerDevices(cd)) - } - return strings.Join(ss, ";") + var ss []string + for _, cd := range pd { + ss = append(ss, EncodeContainerDevices(cd)) + } + return strings.Join(ss, ";") } func DecodeContainerDevices(str string) ContainerDevices { - if len(str) == 0 { - return ContainerDevices{} - } - return strings.Split(str, ",") + if len(str) == 0 { + return ContainerDevices{} + } + cd := strings.Split(str, ":") + contdev := ContainerDevices{} + tmpdev := ContainerDevice{} + //fmt.Println("before container device", str) + if len(str) == 0 { + return contdev + } + for _, val := range cd { + if strings.Contains(val, ",") { + //fmt.Println("cd is ", val) + tmpstr := strings.Split(val, ",") + tmpdev.UUID = tmpstr[0] + devmem, _ := strconv.ParseInt(tmpstr[1], 10, 32) + tmpdev.Usedmem = int32(devmem) + devcores, _ := strconv.ParseInt(tmpstr[2], 10, 32) + tmpdev.Usedcores = int32(devcores) + contdev = append(contdev, tmpdev) + } + } + //fmt.Println("Decoded container device", contdev) + return contdev } func DecodePodDevices(str string) PodDevices { - if len(str) == 0 { - return PodDevices{} - } - var pd PodDevices - for _, s := range strings.Split(str, ";") { - cd := DecodeContainerDevices(s) - pd = append(pd, cd) - } - return pd + if len(str) == 0 { + return PodDevices{} + } + var pd PodDevices + for _, s := range strings.Split(str, ";") { + cd := DecodeContainerDevices(s) + pd = append(pd, cd) + } + return pd } From a72cddc572360daaa1663ed33a8920be52e0fe31 Mon Sep 17 00:00:00 2001 From: limengxuan <391013634@qq.com> Date: Tue, 14 Sep 2021 16:29:06 +0800 Subject: [PATCH 2/6] add readme.md --- README.md | 172 +++++++++++++++++++++++++++++++ deployments/4pd-vgpu/values.yaml | 2 +- 2 files changed, 173 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index e69de29bb2..8a0fbf1370 100644 --- a/README.md +++ b/README.md @@ -0,0 +1,172 @@ +# vGPU device plugin for Kubernetes + + +## 目录 + +- [关于](#关于) +- [使用场景](#使用场景) +- [调度策略](#调度策略) +- [性能测试](#性能测试) +- [功能](#功能) +- [实验性功能](#实验性功能) +- [已知问题](#已知问题) +- [开发计划](#开发计划) +- [安装要求](#安装要求) +- [快速入门](#快速入门) + - [GPU节点准备](#GPU节点准备) + - [Kubernetes开启vGPU支持](#Kubernetes开启vGPU支持) + - [运行GPU任务](#运行GPU任务) +- [测试](#测试) +- [问题反馈及代码贡献](#问题反馈及代码贡献) + +## 关于 + +**k8s vGPU scheduler** 基于NVIDIA官方插件([NVIDIA/k8s-device-plugin](https://github.com/NVIDIA/k8s-device-plugin))，在保留官方功能的基础上，实现了对物理GPU进行切分，并对显存和计算单元进行限制，从而模拟出多张小的vGPU卡。k8s vGPU scheduler在原有显存分配方式的基础上，可以通过设置显存和算力更准确的分配到任务所需要的vGPU卡。在k8s集群中，基于这些切分后的vGPU进行调度，使不同的容器可以安全的共享同一张物理GPU，提高GPU的利用率。此外，插件还可以对显存做虚拟化处理（使用到的显存可以超过物理上的显存），运行一些超大显存需求的任务，或提高共享的任务数，可参考[性能测试报告](#性能测试)。 + +## 使用场景 + +1. 需要定制GPU申请的场合，如申请特定大小的vGPU，每个vGPU使用特定比例的算力。 +2. 在多个GPU节点组成的集群中，任务需要根据自身的显卡需求分配到合适的节点执行。 +3. 显存、计算单元利用率低的情况，如在一张GPU卡上运行10个tf-serving。 +4. 需要大量小显卡的情况，如教学场景把一张GPU提供给多个学生使用、云平台提供小GPU实例。 +5. 物理显存不足的情况，可以开启虚拟显存，如大batch、大模型的训练。 + +## 调度策略 + +调度策略为，在保证显存和算力满足需求的GPU中，优先选择任务数最少的GPU执行任务，这样做可以使任务均匀分配到所有的GPU中 + +## 性能测试 + +见[k8s-device-plugin]的性能测试部分 + +## 功能 + +- 指定每张物理GPU切分的最大vGPU的数量 +- 限制vGPU的显存 +- 限制vGPU的计算单元 +- 对已有程序零改动 + +## 实验性功能 + +- 虚拟显存 + + vGPU的显存总和可以超过GPU实际的显存，这时候超过的部分会放到内存里，对性能有一定的影响。 + +## 已知问题 + +- 目前仅支持计算任务，不支持视频编解码处理。 +- 暂时不支持MIG + +## 开发计划 + +- 支持视频编解码处理 +- 支持Multi-Instance GPUs (MIG) + +## 安装要求 + +* NVIDIA drivers >= 384.81 +* nvidia-docker version > 2.0 +* docker已配置nvidia作为默认runtime +* Kubernetes version >= 1.10 + +## 快速入门 + +### GPU节点准备 + +以下步骤要在所有GPU节点执行。这份README文档假定GPU节点已经安装NVIDIA驱动和`nvidia-docker`套件。 + +注意你需要安装的是`nvidia-docker2`而非`nvidia-container-toolkit`。因为新的`--gpus`选项kubernetes尚不支持。安装步骤举例： + +``` +# 加入套件仓库 +$ distribution=$(. /etc/os-release;echo $ID$VERSION_ID) +$ curl -s -L https://nvidia.github.io/nvidia-docker/gpgkey | sudo apt-key add - +$ curl -s -L https://nvidia.github.io/nvidia-docker/$distribution/nvidia-docker.list | sudo tee /etc/apt/sources.list.d/nvidia-docker.list + +$ sudo apt-get update && sudo apt-get install -y nvidia-docker2 +$ sudo systemctl restart docker +``` + +你需要在节点上将nvidia runtime做为你的docker runtime预设值。我们将编辑docker daemon的配置文件，此文件通常在`/etc/docker/daemon.json`路径： + +``` +{ + "default-runtime": "nvidia", + "runtimes": { + "nvidia": { + "path": "/usr/bin/nvidia-container-runtime", + "runtimeArgs": [] + } + } +} +``` + +> *如果 `runtimes` 字段没有出现, 前往的安装页面执行安装操作 [nvidia-docker](https://github.com/NVIDIA/nvidia-docker)* + +### Kubernetes开启vGPU支持 + +当你在所有GPU节点完成前面提到的准备动作，如果Kubernetes有已经存在的NVIDIA装置插件，需要先将它移除。然后，你需要下载整个项目，并进入deployments文件夹 + +``` +$ git clone https://gitlab.4pd.io/vgpu/k8s-vgpu.git +cd k8s-vgpu/deployments +``` + +在这个deployments文件中, 你可以在 `values.yaml/devicePlugin/extraArgs` 中使用以下的客制化参数： + +* `device-split-count:` + 整数类型，预设值是10。GPU的分割数，每一张GPU都不能分配超过其配置数目的任务。若其配置为N的话，每个GPU上最多可以同时存在N个任务。 +* `device-memory-scaling:` + 浮点数类型，预设值是1。NVIDIA装置显存使用比例，可以大于1（启用虚拟显存，实验功能）。对于有*M​*显存大小的NVIDIA GPU，如果我们配置`device-memory-scaling`参数为*S*，在部署了我们装置插件的Kubenetes集群中，这张GPU分出的vGPU将总共包含 *S \* M*显存。每张vGPU的显存大小也受`device-split-count`参数影响。在先前的例子中，如果`device-split-count`参数配置为*K*，那每一张vGPU最后会取得 *S \* M / K* 大小的显存。 + +配置完成后，随后使用helm安装整个chart + +``` +helm install vgpu 4pd-vgpu +``` + +通过kubectl get pods指令看到vgpu-4pd-vgpu-device-plugin与vgpu-4pd-vgpu-scheduler两个pod即为安装成功 + +``` +kubectl get pods +``` + +### 运行GPU任务 + +NVIDIA vGPUs 现在能透过资源类型`nvidia.com/gpu`被容器请求： + +``` +apiVersion: v1 +kind: Pod +metadata: + name: gpu-pod +spec: + containers: + - name: ubuntu-container + image: ubuntu:18.04 + command: ["bash", "-c", "sleep 86400"] + resources: + limits: + nvidia.com/gpu: 2 # 请求2个vGPUs + nvidia.com/gpumem: 3000 # 每个vGPU申请3000m显存 （可选） + nvidia.com/gpucores: 30 # 每个vGPU的算力为30%实际显卡的算力 （可选） +``` + +现在你可以在容器执行`nvidia-smi`命令，然后比较vGPU和实际GPU显存大小的不同。 + +> **注意:** 如果你使用插件装置时，如果没有请求vGPU资源，那容器所在机器的所有vGPU都将暴露给容器。 + +## 测试 + +- TensorFlow 1.14.0/2.4.1 +- torch1.1.0 +- mxnet 1.4.0 +- mindspore 1.1.1 + +以上框架均通过测试。 + +## 反馈和参与 + +* bug、疑惑、修改欢迎提在 [Github Issues](https://github.com/4paradigm/k8s-device-plugin/issues/new) +* 想了解更多或者有想法可以参与到[Discussions](https://github.com/4paradigm/k8s-device-plugin/discussions)和[slack](https://join.slack.com/t/k8s-device-plugin/shared_invite/zt-oi9zkr5c-LsMzNmNs7UYg6usc0OiWKw)交流 + diff --git a/deployments/4pd-vgpu/values.yaml b/deployments/4pd-vgpu/values.yaml index b8583659b3..0dbbc591a2 100644 --- a/deployments/4pd-vgpu/values.yaml +++ b/deployments/4pd-vgpu/values.yaml @@ -54,7 +54,7 @@ devicePlugin: monitorctrPath: /tmp/vgpu/containers imagePullPolicy: Always extraArgs: - - --device-split-count=2 + - --device-split-count=10 - -v=4 pluginPath: /var/lib/kubelet/device-plugins From 451f6be1b0990030c68040051880f491a794e7fe Mon Sep 17 00:00:00 2001 From: limengxuan <391013634@qq.com> Date: Tue, 14 Sep 2021 16:29:58 +0800 Subject: [PATCH 3/6] add readme.md --- README.md | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 8a0fbf1370..cd2e2b5c92 100644 --- a/README.md +++ b/README.md @@ -109,7 +109,7 @@ $ sudo systemctl restart docker ``` $ git clone https://gitlab.4pd.io/vgpu/k8s-vgpu.git -cd k8s-vgpu/deployments +$ cd k8s-vgpu/deployments ``` 在这个deployments文件中, 你可以在 `values.yaml/devicePlugin/extraArgs` 中使用以下的客制化参数： @@ -122,13 +122,13 @@ cd k8s-vgpu/deployments 配置完成后，随后使用helm安装整个chart ``` -helm install vgpu 4pd-vgpu +$ helm install vgpu 4pd-vgpu ``` 通过kubectl get pods指令看到vgpu-4pd-vgpu-device-plugin与vgpu-4pd-vgpu-scheduler两个pod即为安装成功 ``` -kubectl get pods +$ kubectl get pods ``` ### 运行GPU任务 From 720d8f269cda80d70512fbe45563e778aae78fd1 Mon Sep 17 00:00:00 2001 From: limengxuan <391013634@qq.com> Date: Tue, 14 Sep 2021 16:42:12 +0800 Subject: [PATCH 4/6] fix resourcecores error --- README.md | 4 +--- deployments/4pd-vgpu/templates/scheduler/configmap.yaml | 2 +- 2 files changed, 2 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index cd2e2b5c92..0d938bff4b 100644 --- a/README.md +++ b/README.md @@ -37,7 +37,7 @@ ## 性能测试 -见[k8s-device-plugin]的性能测试部分 +见[k8s-device-plugin的性能测试部分](https://github.com/4paradigm/k8s-device-plugin/blob/master/README_cn.md#性能测试) ## 功能 @@ -154,8 +154,6 @@ spec: 现在你可以在容器执行`nvidia-smi`命令，然后比较vGPU和实际GPU显存大小的不同。 -> **注意:** 如果你使用插件装置时，如果没有请求vGPU资源，那容器所在机器的所有vGPU都将暴露给容器。 - ## 测试 - TensorFlow 1.14.0/2.4.1 diff --git a/deployments/4pd-vgpu/templates/scheduler/configmap.yaml b/deployments/4pd-vgpu/templates/scheduler/configmap.yaml index 4718fd3ad7..cf96ad2561 100644 --- a/deployments/4pd-vgpu/templates/scheduler/configmap.yaml +++ b/deployments/4pd-vgpu/templates/scheduler/configmap.yaml @@ -29,5 +29,5 @@ data: ignoredByScheduler: true - name: {{ .Values.resourceMem }} ignoredByScheduler: true - - name: {{ .Values.resourcecores }} + - name: {{ .Values.resourceCores }} ignoredByScheduler: true From d7057e63da7b5b16eb7d65b3cf9c0dfdaa6b5c35 Mon Sep 17 00:00:00 2001 From: limengxuan <391013634@qq.com> Date: Tue, 14 Sep 2021 19:21:16 +0800 Subject: [PATCH 5/6] add default gpumem and default gpucores --- README.md | 12 ++- cmd/scheduler/main.go | 126 ++++++++++++++++--------------- deployments/4pd-vgpu/values.yaml | 2 + pkg/k8sutil/pod.go | 9 ++- pkg/scheduler/config/config.go | 8 +- pkg/util/types.go | 6 +- 6 files changed, 89 insertions(+), 74 deletions(-) diff --git a/README.md b/README.md index 0d938bff4b..24196976bd 100644 --- a/README.md +++ b/README.md @@ -1,4 +1,4 @@ -# vGPU device plugin for Kubernetes +# vGPU scheduler for Kubernetes ## 目录 @@ -21,7 +21,7 @@ ## 关于 -**k8s vGPU scheduler** 基于NVIDIA官方插件([NVIDIA/k8s-device-plugin](https://github.com/NVIDIA/k8s-device-plugin))，在保留官方功能的基础上，实现了对物理GPU进行切分，并对显存和计算单元进行限制，从而模拟出多张小的vGPU卡。k8s vGPU scheduler在原有显存分配方式的基础上，可以通过设置显存和算力更准确的分配到任务所需要的vGPU卡。在k8s集群中，基于这些切分后的vGPU进行调度，使不同的容器可以安全的共享同一张物理GPU，提高GPU的利用率。此外，插件还可以对显存做虚拟化处理（使用到的显存可以超过物理上的显存），运行一些超大显存需求的任务，或提高共享的任务数，可参考[性能测试报告](#性能测试)。 +**k8s vGPU scheduler** 基于4pd-k8s-device-plugin插件([4paradigm/k8s-device-plugin](https://github.com/4paradigm/k8s-device-plugin))，在保留原功能的基础上，添加了调度模块，以实现多个GPU节点间的负载均衡。k8s vGPU scheduler在原有显卡分配方式的基础上，可以进一步根据显存和算力来切分显卡。在k8s集群中，基于这些切分后的vGPU进行调度，使不同的容器可以安全的共享同一张物理GPU，提高GPU的利用率。此外，插件还可以对显存做虚拟化处理（使用到的显存可以超过物理上的显存），运行一些超大显存需求的任务，或提高共享的任务数，可参考[性能测试报告](#性能测试)。 ## 使用场景 @@ -119,6 +119,14 @@ $ cd k8s-vgpu/deployments * `device-memory-scaling:` 浮点数类型，预设值是1。NVIDIA装置显存使用比例，可以大于1（启用虚拟显存，实验功能）。对于有*M​*显存大小的NVIDIA GPU，如果我们配置`device-memory-scaling`参数为*S*，在部署了我们装置插件的Kubenetes集群中，这张GPU分出的vGPU将总共包含 *S \* M*显存。每张vGPU的显存大小也受`device-split-count`参数影响。在先前的例子中，如果`device-split-count`参数配置为*K*，那每一张vGPU最后会取得 *S \* M / K* 大小的显存。 +除此之外，你可以在 `values.yaml/scheduler/extender/extraArgs` 中使用以下客制化参数: + +* `default-mem:` + 整数类型，预设值为5000，表示不配置显存时使用的默认显存大小，单位为MB + +* `default-cores:` + 整数类型(0-100)，默认为0，表示不配置显卡使用比例时默认的使用比例。若设置为0，则代表任务可能会被分配到任一满足显存需求的GPU中，若设置为100，代表该任务独享整张显卡 + 配置完成后，随后使用helm安装整个chart ``` diff --git a/cmd/scheduler/main.go b/cmd/scheduler/main.go index 01c5df8e4d..ecdb0b92ef 100644 --- a/cmd/scheduler/main.go +++ b/cmd/scheduler/main.go @@ -16,84 +16,86 @@ package main import ( - "4pd.io/k8s-vgpu/pkg/util" - "4pd.io/k8s-vgpu/pkg/version" - "net" - "net/http" + "net" + "net/http" - pb "4pd.io/k8s-vgpu/pkg/api" - "4pd.io/k8s-vgpu/pkg/scheduler" - "4pd.io/k8s-vgpu/pkg/scheduler/config" - "4pd.io/k8s-vgpu/pkg/scheduler/routes" - "github.com/julienschmidt/httprouter" - "github.com/spf13/cobra" - "google.golang.org/grpc" - "k8s.io/klog/v2" + "4pd.io/k8s-vgpu/pkg/util" + "4pd.io/k8s-vgpu/pkg/version" + + pb "4pd.io/k8s-vgpu/pkg/api" + "4pd.io/k8s-vgpu/pkg/scheduler" + "4pd.io/k8s-vgpu/pkg/scheduler/config" + "4pd.io/k8s-vgpu/pkg/scheduler/routes" + "github.com/julienschmidt/httprouter" + "github.com/spf13/cobra" + "google.golang.org/grpc" + "k8s.io/klog/v2" ) //var version string var ( - tlsKeyFile string - tlsCertFile string - rootCmd = &cobra.Command{ - Use: "scheduler", - Short: "kubernetes vgpu scheduler", - Run: func(cmd *cobra.Command, args []string) { - start() - }, - } + tlsKeyFile string + tlsCertFile string + rootCmd = &cobra.Command{ + Use: "scheduler", + Short: "kubernetes vgpu scheduler", + Run: func(cmd *cobra.Command, args []string) { + start() + }, + } ) func init() { - rootCmd.Flags().SortFlags = false - rootCmd.PersistentFlags().SortFlags = false - - rootCmd.Flags().StringVar(&config.GrpcBind, "grpc_bind", "127.0.0.1:9090", "grpc server bind address") - rootCmd.Flags().StringVar(&config.HttpBind, "http_bind", "127.0.0.1:8080", "http server bind address") - rootCmd.Flags().StringVar(&tlsCertFile, "cert_file", "", "tls cert file") - rootCmd.Flags().StringVar(&tlsKeyFile, "key_file", "", "tls key file") - rootCmd.Flags().StringVar(&config.SchedulerName, "scheduler-name", "", "the name to be added to pod.spec.schedulerName if not empty") + rootCmd.Flags().SortFlags = false + rootCmd.PersistentFlags().SortFlags = false - rootCmd.PersistentFlags().AddGoFlagSet(util.GlobalFlagSet()) - rootCmd.AddCommand(version.VersionCmd) + rootCmd.Flags().StringVar(&config.GrpcBind, "grpc_bind", "127.0.0.1:9090", "grpc server bind address") + rootCmd.Flags().StringVar(&config.HttpBind, "http_bind", "127.0.0.1:8080", "http server bind address") + rootCmd.Flags().StringVar(&tlsCertFile, "cert_file", "", "tls cert file") + rootCmd.Flags().StringVar(&tlsKeyFile, "key_file", "", "tls key file") + rootCmd.Flags().StringVar(&config.SchedulerName, "scheduler-name", "", "the name to be added to pod.spec.schedulerName if not empty") + rootCmd.Flags().Int32Var(&config.DefaultMem, "default-mem", 5000, "default gpu device memory to allocate") + rootCmd.Flags().Int32Var(&config.DefaultCores, "default-cores", 0, "default gpu core percentage to allocate") + rootCmd.PersistentFlags().AddGoFlagSet(util.GlobalFlagSet()) + rootCmd.AddCommand(version.VersionCmd) } func start() { - sher := scheduler.NewScheduler() - sher.Start() - defer sher.Stop() + sher := scheduler.NewScheduler() + sher.Start() + defer sher.Stop() - // start grpc server - lisGrpc, _ := net.Listen("tcp", config.GrpcBind) - defer lisGrpc.Close() - s := grpc.NewServer() - pb.RegisterDeviceServiceServer(s, sher) - go func() { - err := s.Serve(lisGrpc) - if err != nil { - klog.Fatal(err) - } - }() + // start grpc server + lisGrpc, _ := net.Listen("tcp", config.GrpcBind) + defer lisGrpc.Close() + s := grpc.NewServer() + pb.RegisterDeviceServiceServer(s, sher) + go func() { + err := s.Serve(lisGrpc) + if err != nil { + klog.Fatal(err) + } + }() - // start http server - router := httprouter.New() - router.POST("/filter", routes.PredicateRoute(sher)) - router.POST("/webhook", routes.WebHookRoute()) - klog.Info("listen on ", config.HttpBind) - if len(tlsCertFile) == 0 || len(tlsKeyFile) == 0 { - if err := http.ListenAndServe(config.HttpBind, router); err != nil { - klog.Fatal("Listen and Serve error, ", err) - } - } else { - if err := http.ListenAndServeTLS(config.HttpBind, tlsCertFile, tlsKeyFile, router); err != nil { - klog.Fatal("Listen and Serve error, ", err) - } - } + // start http server + router := httprouter.New() + router.POST("/filter", routes.PredicateRoute(sher)) + router.POST("/webhook", routes.WebHookRoute()) + klog.Info("listen on ", config.HttpBind) + if len(tlsCertFile) == 0 || len(tlsKeyFile) == 0 { + if err := http.ListenAndServe(config.HttpBind, router); err != nil { + klog.Fatal("Listen and Serve error, ", err) + } + } else { + if err := http.ListenAndServeTLS(config.HttpBind, tlsCertFile, tlsKeyFile, router); err != nil { + klog.Fatal("Listen and Serve error, ", err) + } + } } func main() { - if err := rootCmd.Execute(); err != nil { - klog.Fatal(err) - } + if err := rootCmd.Execute(); err != nil { + klog.Fatal(err) + } } diff --git a/deployments/4pd-vgpu/values.yaml b/deployments/4pd-vgpu/values.yaml index 0dbbc591a2..e684ad114b 100644 --- a/deployments/4pd-vgpu/values.yaml +++ b/deployments/4pd-vgpu/values.yaml @@ -28,6 +28,8 @@ scheduler: imagePullPolicy: Always extraArgs: - --debug + - -default-mem=5000 + - -default-cores=0 - -v=4 podAnnotations: {} nodeSelector: {} diff --git a/pkg/k8sutil/pod.go b/pkg/k8sutil/pod.go index 4678ef86ad..0e80328ee0 100644 --- a/pkg/k8sutil/pod.go +++ b/pkg/k8sutil/pod.go @@ -17,6 +17,7 @@ package k8sutil import ( + "4pd.io/k8s-vgpu/pkg/scheduler/config" "4pd.io/k8s-vgpu/pkg/util" corev1 "k8s.io/api/core/v1" ) @@ -33,7 +34,7 @@ func Resourcereqs(pod *corev1.Pod) (counts []util.ContainerDeviceRequest) { } if ok { if n, ok := v.AsInt64(); ok { - memnum := util.DefaultMem + memnum := config.DefaultMem mem, ok := pod.Spec.Containers[i].Resources.Limits[resourceMem] if !ok { mem, ok = pod.Spec.Containers[i].Resources.Requests[resourceMem] @@ -41,10 +42,10 @@ func Resourcereqs(pod *corev1.Pod) (counts []util.ContainerDeviceRequest) { if ok { memnums, ok := mem.AsInt64() if ok { - memnum = int(memnums) + memnum = int32(memnums) } } - corenum := util.DefaultCores + corenum := config.DefaultCores core, ok := pod.Spec.Containers[i].Resources.Limits[resourceCores] if !ok { core, ok = pod.Spec.Containers[i].Resources.Requests[resourceCores] @@ -52,7 +53,7 @@ func Resourcereqs(pod *corev1.Pod) (counts []util.ContainerDeviceRequest) { if ok { corenums, ok := core.AsInt64() if ok { - corenum = int(corenums) + corenum = int32(corenums) } } counts[i] = util.ContainerDeviceRequest{ diff --git a/pkg/scheduler/config/config.go b/pkg/scheduler/config/config.go index f706d14709..9b5722efdb 100644 --- a/pkg/scheduler/config/config.go +++ b/pkg/scheduler/config/config.go @@ -17,7 +17,9 @@ package config var ( - HttpBind string - GrpcBind string - SchedulerName string + HttpBind string + GrpcBind string + SchedulerName string + DefaultMem int32 + DefaultCores int32 ) diff --git a/pkg/util/types.go b/pkg/util/types.go index 79f75af7b9..7340d5060f 100644 --- a/pkg/util/types.go +++ b/pkg/util/types.go @@ -24,10 +24,10 @@ const ( AssignedNodeAnnotations = "4pd.io/vgpu-node" //Set default mem to 5000m - DefaultMem = 5000 - DefaultCores = 0 + //DefaultMem = 5000 + //DefaultCores = 0 - DeviceLimit = 2 + DeviceLimit = 100 //TimeLayout = "ANSIC" //DefaultTimeout = time.Second * 60 ) From e61d0c94e6a63b500499a42266b8f5b2064967b7 Mon Sep 17 00:00:00 2001 From: limengxuan <391013634@qq.com> Date: Tue, 14 Sep 2021 19:41:47 +0800 Subject: [PATCH 6/6] modify default cores and default mem --- deployments/4pd-vgpu/values.yaml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/deployments/4pd-vgpu/values.yaml b/deployments/4pd-vgpu/values.yaml index e684ad114b..acdc673bc7 100644 --- a/deployments/4pd-vgpu/values.yaml +++ b/deployments/4pd-vgpu/values.yaml @@ -28,8 +28,8 @@ scheduler: imagePullPolicy: Always extraArgs: - --debug - - -default-mem=5000 - - -default-cores=0 + - --default-mem=5000 + - --default-cores=0 - -v=4 podAnnotations: {} nodeSelector: {}