目前，要在 Kubernetes (k8s) 中调度 GPU Pod，需要实施多种扩展方案，包括 Device Plugin、Extended Resource、scheduler extender、scheduler framework 或开发新的调度器。这些方案通常依赖 nodeSelector 或 nodeAffinity 来选择提供特定 GPU 模型的节点。此外，为了自动发现每个节点上的 GPU 模型，必须部署 Node Feature Discovery (NFD) 或 GPU Feature Discovery (GFD)。
后来，为了标准化将第三方设备暴露给容器的方式，引入了新的运行时规范 CDI（Container Device Interface）。当然，CDI 仅提供了 CRI 侧的解决方案。
幸运的是，从 k8s v1.26 开始，支持动态资源分配（DRA），它在调度侧提供了一致的资源请求和分配解决方案。然而，最初的 DRA with Control Plane Controller 方法存在一些缺陷，在 k8s v1.30 中通过引入 DRA with Structured Parameters 得到了改进，这也是本文的重点。
k8s 1.31 在此基础上做了进一步更改，本文将基于 k8s 1.31 版本介绍 DRA with Structured Parameters。
（对于之前的扩展方案，本文不再赘述；如果时间允许，我会再写一篇文章来介绍它们。）
现在，让我们直接进入本文的主题：DRA with Structured Parameters。
DRA with Structured Parameters
DRA 主要用于 Pod 之间或单个 Pod 内多个容器之间的资源请求和共享，不仅限于 GPU 资源。虽然它比之前的 Device Plugin 方法复杂得多，但 ResourceClaimTemplate/ResourceClaim/DeviceClass 这组对象类似于 PVC/PV/SC，提供了灵活且全面的资源定义，能够官方支持各种异构资源。
当然，用户仍需开发自己的资源驱动程序来处理这些资源的准备和维护。
关键概念
ResourceClaim
ResourceClaim 用于资源访问请求。调度器根据 Pod 中配置的 ResourceClaim 选择可用的节点。
---
apiVersion: resource.k8s.io/v1alpha3
kind: ResourceClaim
metadata:
name: small-white-cat-claim
spec:
devices:
requests:
- name: req-0
deviceClassName: resource.example.com
selectors:
- cel:
expression: |-
device.attributes["resource-driver.example.com"].color == "white" &&
device.attributes["resource-driver.example.com"].size == "small"
ResourceClaimTemplate
ResourceClaimTemplate 用于创建 ResourceClaim。它由 kube-controller-manager 的 resourceclaim 控制器管理。以下是创建它的代码片段：
func (ec *Controller) handleClaim(ctx context.Context, pod *v1.Pod, podClaim v1.PodResourceClaim, newPodClaims *map[string]string) error {
claimName, mustCheckOwner, err := resourceclaim.Name(pod, &podClaim)
switch {
case errors.Is(err, resourceclaim.ErrClaimNotFound):
......
}
// Before we create a new ResourceClaim, check if there is an orphaned one.
// This covers the case that the controller has created it, but then fails
// before it can update the pod status.
claim, err := ec.findPodResourceClaim(pod, podClaim)
if err != nil {
return fmt.Errorf("finding ResourceClaim for claim %s in pod %s/%s failed: %v", podClaim.Name, pod.Namespace, pod.Name, err)
}
if claim == nil {
template, err := ec.templateLister.ResourceClaimTemplates(pod.Namespace).Get(*templateName)
if err != nil {
return fmt.Errorf("resource claim template %q: %v", *templateName, err)
}
// Create the ResourceClaim with pod as owner, with a generated name that uses
// <pod>-<claim name> as base.
isTrue := true
annotations := template.Spec.ObjectMeta.Annotations
if annotations == nil {
annotations = make(map[string]string)
}
annotations[podResourceClaimAnnotation] = podClaim.Name
generateName := pod.Name + "-" + podClaim.Name + "-"
maxBaseLen := 57 // Leave space for hyphen and 5 random characters in a name with 63 characters.
if len(generateName) > maxBaseLen {
// We could leave truncation to the apiserver, but as
// it removes at the end, we would loose everything
// from the pod claim name when the pod name is long.
// We can do better and truncate both strings,
// proportional to their length.
generateName = pod.Name[0:len(pod.Name)*maxBaseLen/len(generateName)] +
"-" +
podClaim.Name[0:len(podClaim.Name)*maxBaseLen/len(generateName)]
}
claim = &resourceapi.ResourceClaim{
ObjectMeta: metav1.ObjectMeta{
GenerateName: generateName,
OwnerReferences: []metav1.OwnerReference{
{
APIVersion: "v1",
Kind: "Pod",
Name: pod.Name,
UID: pod.UID,
Controller: &isTrue,
BlockOwnerDeletion: &isTrue,
},
},
Annotations: annotations,
Labels: template.Spec.ObjectMeta.Labels,
},
Spec: template.Spec.Spec,
}
metrics.ResourceClaimCreateAttempts.Inc()
claimName := claim.Name
claim, err = ec.kubeClient.ResourceV1alpha3().ResourceClaims(pod.Namespace).Create(ctx, claim, metav1.CreateOptions{})
if err != nil {
metrics.ResourceClaimCreateFailures.Inc()
return fmt.Errorf("create ResourceClaim %s: %v", claimName, err)
}
logger.V(4).Info("Created ResourceClaim", "claim", klog.KObj(claim), "pod", klog.KObj(pod))
ec.claimCache.Mutation(claim)
}
为什么引入 ResourceClaimTemplate？
考虑以下 Pod Spec 中的示例：resourceClaims 字段允许你引用 ResourceClaim 或 ResourceClaimTemplate，但不能同时引用两者——它们是互斥的。当你通过 resourceClaimName 引用 ResourceClaim 时，所有使用此 Pod Spec 的 Pod（例如 Deployment 或 StatefulSet 中的 Pod）共享同一个 ResourceClaim 实例。该实例对应的是 Pod 所在命名空间中名为 resourceClaimName 的 ResourceClaim。
相反，当你通过 resourceClaimTemplateName 引用 ResourceClaimTemplate 时，每个 Pod 都会获得自己独立的 ResourceClaim 实例。这是因为 kube-controller-manager 的 resourceclaim 控制器会为每个 Pod 自动生成一个独立的 ResourceClaim。
以下是一个简化的 Pod Spec 示例：
–--
apiVersion: v1
kind: Pod
metadata:
name: pod-with-cats
spec:
containers:
- name: container0
image: ubuntu:20.04
command: ["sleep", "9999"]
resources:
claims:
- name: cat-0
- name: container1
image: ubuntu:20.04
command: ["sleep", "9999"]
resources:
claims:
- name: cat-1
resourceClaims:
- name: cat-0
resourceClaimName: small-white-cat-claim
- name: cat-1
resourceClaimTemplateName: large-black-cat-claim-template
DeviceClass
DeviceClass 包含设备的配置和选择器，其定义如下。
// +genclient
// +genclient:nonNamespaced
// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
// +k8s:prerelease-lifecycle-gen:introduced=1.31
// DeviceClass is a vendor- or admin-provided resource that contains
// device configuration and selectors. It can be referenced in
// the device requests of a claim to apply these presets.
// Cluster scoped.
//
// This is an alpha type and requires enabling the DynamicResourceAllocation
// feature gate.
type DeviceClass struct {
metav1.TypeMeta `json:",inline"`
// Standard object metadata
// +optional
metav1.ObjectMeta `json:"metadata,omitempty" protobuf:"bytes,1,opt,name=metadata"`
// Spec defines what can be allocated and how to configure it.
//
// This is mutable. Consumers have to be prepared for classes changing
// at any time, either because they get updated or replaced. Claim
// allocations are done once based on whatever was set in classes at
// the time of allocation.
//
// Changing the spec automatically increments the metadata.generation number.
Spec DeviceClassSpec `json:"spec" protobuf:"bytes,2,name=spec"`
}
// DeviceClassSpec is used in a [DeviceClass] to define what can be allocated
// and how to configure it.
type DeviceClassSpec struct {
// Each selector must be satisfied by a device which is claimed via this class.
//
// +optional
// +listType=atomic
Selectors []DeviceSelector `json:"selectors,omitempty" protobuf:"bytes,1,opt,name=selectors"`
// Config defines configuration parameters that apply to each device that is claimed via this class.
// Some classses may potentially be satisfied by multiple drivers, so each instance of a vendor
// configuration applies to exactly one driver.
//
// They are passed to the driver, but are not considered while allocating the claim.
//
// +optional
// +listType=atomic
Config []DeviceClassConfiguration `json:"config,omitempty" protobuf:"bytes,2,opt,name=config"`
// Only nodes matching the selector will be considered by the scheduler
// when trying to find a Node that fits a Pod when that Pod uses
// a claim that has not been allocated yet *and* that claim
// gets allocated through a control plane controller. It is ignored
// when the claim does not use a control plane controller
// for allocation.
//
// Setting this field is optional. If unset, all Nodes are candidates.
//
// This is an alpha field and requires enabling the DRAControlPlaneController
// feature gate.
//
// +optional
// +featureGate=DRAControlPlaneController
SuitableNodes *v1.NodeSelector `json:"suitableNodes,omitempty" protobuf:"bytes,3,opt,name=suitableNodes"`
}
当用户创建 ResourceClaim 或 ResourceClaimTemplate 时，需要通过 deviceClassName 指定对应的 DeviceClass。调度器使用 DeviceClass 的选择器匹配设备，从而为 Pod 分配节点，最后将 DeviceClass 中的 Device 配置填充到 ResourceClaim 实例的 Status 中。
if requestData.class != nil {
match, err := alloc.selectorsMatch(r, device, deviceID, requestData.class, requestData.class.Spec.Selectors)
if err != nil {
return false, err
}
if !match {
alloc.deviceMatchesRequest[matchKey] = false
return false, nil
}
}
request := &alloc.claimsToAllocate[r.claimIndex].Spec.Devices.Requests[r.requestIndex]
match, err := alloc.selectorsMatch(r, device, deviceID, nil, request.Selectors)
for claimIndex, allocationResult := range alloc.result {
claim := alloc.claimsToAllocate[claimIndex]
// Populate configs.
for requestIndex := range claim.Spec.Devices.Requests {
class := alloc.requestData[requestIndices{claimIndex: claimIndex, requestIndex: requestIndex}].class
if class != nil {
for _, config := range class.Spec.Config {
allocationResult.Devices.Config = append(allocationResult.Devices.Config, resourceapi.DeviceAllocationConfiguration{
Source: resourceapi.AllocationConfigSourceClass,
Requests: nil, // All of them...
DeviceConfiguration: config.DeviceConfiguration,
})
}
}
}
代码路径：pkg\scheduler\framework\plugins\dynamicresources\dynamicresources.go
func (pl *dynamicResources) bindClaim(ctx context.Context, state *stateData, index int, pod *v1.Pod, nodeName string) (patchedClaim *resourceapi.ResourceClaim, finalErr error) {
if allocation != nil {
if claim.Status.Allocation != nil {
return fmt.Errorf("claim %s got allocated elsewhere in the meantime", klog.KObj(claim))
}
......
claim.Status.Allocation = allocation
}
claim.Status.ReservedFor = append(claim.Status.ReservedFor, resourceapi.ResourceClaimConsumerReference{Resource: "pods", Name: pod.Name, UID: pod.UID})
updatedClaim, err := pl.clientset.ResourceV1alpha3().ResourceClaims(claim.Namespace).UpdateStatus(ctx, claim, metav1.UpdateOptions{})
ResourceSlice
Kubernetes 1.30 引入了 ResourceSlice 作为 DRA with Structured Parameters 的重要改进功能。ResourceSlice 实例的创建和维护由 kubelet 和供应商提供的资源驱动程序决定。此外，调度器在 Pod 调度期间根据 ResourceClaim 和 ResourceSlice 确定调度节点，从而承担了资源分配的责任。
kubelet 主要负责在启动、重启或删除 kubelet 插件时自动清除节点上的 ResourceSlice。
// NewPluginHandler returns new registration handler.
//
// Must only be called once per process because it manages global state.
// If a kubeClient is provided, then it synchronizes ResourceSlices
// with the resource information provided by plugins.
func NewRegistrationHandler(kubeClient kubernetes.Interface, getNode func() (*v1.Node, error)) *RegistrationHandler {
handler := &RegistrationHandler{
// The context and thus logger should come from the caller.
backgroundCtx: klog.NewContext(context.TODO(), klog.LoggerWithName(klog.TODO(), "DRA registration handler")),
kubeClient: kubeClient,
getNode: getNode,
}
// When kubelet starts up, no DRA driver has registered yet. None of
// the drivers are usable until they come back, which might not happen
// at all. Therefore it is better to not advertise any local resources
// because pods could get stuck on the node waiting for the driver
// to start up.
//
// This has to run in the background.
go handler.wipeResourceSlices("")
return handler
}
// DeRegisterPlugin is called when a plugin has removed its socket,
// signaling it is no longer available.
func (h *RegistrationHandler) DeRegisterPlugin(pluginName string) {
if p := draPlugins.delete(pluginName); p != nil {
logger := klog.FromContext(p.backgroundCtx)
logger.V(3).Info("Deregister DRA plugin", "endpoint", p.endpoint)
// Clean up the ResourceSlices for the deleted Plugin since it
// may have died without doing so itself and might never come
// back.
go h.wipeResourceSlices(pluginName)
return
}
logger := klog.FromContext(h.backgroundCtx)
logger.V(3).Info("Deregister DRA plugin not necessary, was already removed")
}
资源驱动程序（DRA 驱动）
资源驱动程序，也称为 DRA 驱动，需要由供应商或用户自行开发。它主要负责发布 ResourceSlice 资源，并作为 kubelet 插件处理 kubelet 触发的资源准备请求。最终通过 CDI 使用资源。
以下是 kubelet 发起资源准备请求的相关代码。
// PrepareResources attempts to prepare all of the required resources
// for the input container, issue NodePrepareResources rpc requests
// for each new resource requirement, process their responses and update the cached
// containerResources on success.
func (m *ManagerImpl) PrepareResources(pod *v1.Pod) error {
batches := make(map[string][]*drapb.Claim)
resourceClaims := make(map[types.UID]*resourceapi.ResourceClaim)
for i := range pod.Spec.ResourceClaims {
podClaim := &pod.Spec.ResourceClaims[i]
klog.V(3).InfoS("Processing resource", "pod", klog.KObj(pod), "podClaim", podClaim.Name)
claimName, mustCheckOwner, err := resourceclaim.Name(pod, podClaim)
......
resourceClaim, err := m.kubeClient.ResourceV1alpha3().ResourceClaims(pod.Namespace).Get(
context.TODO(),
*claimName,
metav1.GetOptions{})
......
// Atomically perform some operations on the claimInfo cache.
err = m.cache.withLock(func() error {
claimInfo, exists := m.cache.get(resourceClaim.Name, resourceClaim.Namespace)
......
// Loop through all drivers and prepare for calling NodePrepareResources.
claim := &drapb.Claim{
Namespace: claimInfo.Namespace,
UID: string(claimInfo.ClaimUID),
Name: claimInfo.ClaimName,
}
for driverName := range claimInfo.DriverState {
batches[driverName] = append(batches[driverName], claim)
}
return nil
})
if err != nil {
return fmt.Errorf("locked cache operation: %w", err)
}
}
// Call NodePrepareResources for all claims in each batch.
// If there is any error, processing gets aborted.
// We could try to continue, but that would make the code more complex.
for driverName, claims := range batches {
// Call NodePrepareResources RPC for all resource handles.
client, err := dra.NewDRAPluginClient(driverName)
if err != nil {
return fmt.Errorf("failed to get gRPC client for driver %s: %w", driverName, err)
}
response, err := client.NodePrepareResources(context.Background(), &drapb.NodePrepareResourcesRequest{Claims: claims})
......
}
DRA with Control Plane Controller 方法的缺陷
DRA with Control Plane Controller 方法的缺点是：在 Pod 调度期间，调度器会创建一个 PodSchedulingContext，资源驱动程序分配资源。由于 Pod 调度是顺序进行的，这不可避免地会增加队列中等待调度的其他 Pod 的延迟。
为了优化这一点，引入了结构化参数，允许资源驱动程序主动报告 ResourceSlices，使调度器在 Pod 调度期间能够直接从 ResourceSlices 做出调度决策。
总结
目前，AI 训练和推理可以像普通 CPU 容器一样由 Kubernetes 调度，但这是通过现有的 Kubernetes 扩展机制实现的，缺乏统一的解决方案。而且，GPU 资源分配的方案越复杂，用户需要自行实现的功能也越复杂。Kubernetes DRA 提供了一个统一的解决方案，供应商或用户只需实现资源驱动程序并提供 DriverClass 实例配置，由于它是官方支持的，因此无需担心 GPU 调度开源方案停止维护的问题。
DRA 尚未成熟，仍在持续改进中。让我们共同关注其最新发展动态。
