【NPU】Ascend Device Plugin — Part 2 gRPC 服务层 超深度源码分析之二
·
3.2 plugin.go — PluginServer gRPC 接口实现
3.2.1 ListAndWatch — 设备列表上报
逐行解析：
func (ps *PluginServer) ListAndWatch(empty *v1beta1.Empty,
stream v1beta1.DevicePlugin_ListAndWatchServer) error {
serverRestartCount := ps.restartTimes.Load() - 1
hwlog.RunLog.Infof("receive ListAndWatch from kubelet, deviceType=%s, server restartTimes=%d",
ps.deviceType, serverRestartCount)
// 标记 server 正在运行
ps.isRunning.Store(true)
// 首次立即上报设备信息
ps.reportDeviceInfo(stream)
for {
select {
case <-stream.Context().Done():
// gRPC 流被 kubelet 关闭
hwlog.RunLog.Warnf("grpc stream closed, deviceType=%s", ps.deviceType)
if ps.isRunning.Load() {
// server 应该继续运行，设置重启标志
ps.isRunning.Store(false)
ps.SetRestartFlag(true)
}
return nil
case <-ps.stop:
// 收到停止信号（来自 Stop() 调用）
ps.isRunning.Store(false)
return nil
case <-ps.reciChan:
// 收到设备变更通知，重新上报
ps.reportDeviceInfo(stream)
// 检查发送策略（连续失败时的恢复策略）
strategy := ps.strategyForSendStats()
ps.handleConsecutiveErrorStrategy(strategy, serverRestartCount)
if strategy != common.EmptyStrategy {
return nil // 需要重启或重新注册，退出 ListAndWatch
}
}
}
}
3.2.2 responseToKubelet — 构造设备列表响应
三种模式对比：
| 模式 | 条件 | 设备名来源 | 健康状态来源 |
|---|---|---|---|
| 动态切分 | PresetVDevice=false |
AICore 设备名 | getUnhealthyAICore() 计算 |
| Volcano | PresetVDevice=true && UseVolcanoType && !IsVirtualDev |
klt 设备名（通过 vol→klt 映射） | cachedDevices 健康状态 |
| 标准预置 | PresetVDevice=true（其他） |
cachedDevices 设备名 | cachedDevices 健康状态 |
3.2.3 Allocate — 设备分配
关键代码解析：
func (ps *PluginServer) Allocate(ctx context.Context, requests *v1beta1.AllocateRequest) (
*v1beta1.AllocateResponse, error) {
// 1. 校验请求合法性
if err := ps.checkAllocateRequest(requests); err != nil {
return nil, err
}
// 2. 获取所有 NPU 信息（用于 A5 LogicID 转换）
allNPUInfo, err := ps.manager.GetNPUs()
if err != nil {
return nil, err
}
resps := new(v1beta1.AllocateResponse)
for _, rqt := range requests.ContainerRequests {
// 3. 设备名转换（公共名→内部名）
allocateDevices := customname.ReplaceDeviceInnerName(ps.deviceType, rqt.DevicesIDs)
var npuInfoConfigDir string
usePodAnnotation := false
// 4. Volcano 调度场景：如果请求数 != 设备总数 或 动态切分模式
if (len(allocateDevices) != len(allNPUInfo.AllDevs) || !common.ParamOption.PresetVDevice) &&
common.ParamOption.UseVolcanoType {
usePodAnnotation = true
allocateDevices, npuInfoConfigDir, err = ps.useVolcano(rqt.DevicesIDs)
if err != nil {
return nil, err
}
}
// 5. 获取设备 ID 列表（物理 ID）
_, ascendVisibleDevices, err := common.GetDeviceListID(allocateDevices, ps.ascendRuntimeOptions)
if err != nil {
return nil, err
}
// 6. A5 场景：物理 ID → 逻辑 ID 转换（除非已通过 Pod 注解获取）
finalVisibleDevices := getFinalVisibleDevices(ascendVisibleDevices, allNPUInfo, usePodAnnotation)
// 7. 构造容器分配响应
resp := new(v1beta1.ContainerAllocateResponse)
ps.mountShareDeviceConfig(resp, finalVisibleDevices, npuInfoConfigDir) // 软共享配置
ps.setNPUDeviceMount(resp, finalVisibleDevices) // 设备挂载
ps.setHcclTopoFilePathEnv(resp, allNPUInfo) // HCCL 拓扑路径
ps.SetSlowNodeNoticeEnv(resp) // 慢节点通知
resps.ContainerResponses = append(resps.ContainerResponses, resp)
}
return resps, nil
}
3.2.4 doWithVolcanoSchedule — Volcano 调度设备分配
3.2.5 getUnhealthyAICore — 动态切分不健康 AICore 计算
func (ps *PluginServer) getUnhealthyAICore() sets.String {
// 1. 收集不健康的物理 ID 和所有 AICore 设备名
unhealthyPhyID := sets.Int{}
allAICore := make(sets.String, len(ps.cachedDevices))
for _, device := range ps.cachedDevices {
if device.Health == v1beta1.Unhealthy {
unhealthyPhyID.Insert(int(device.PhyID))
}
allAICore.Insert(device.DeviceName)
}
// 2. 获取实际被 Pod 使用的 AICore（从 Pod 资源查询）
realUsedAICore, err := ps.GetRealUsedAICore()
if err != nil {
return sets.String{}
}
// 3. 标记不健康芯片上已使用的 AICore 为不健康
unhealthyAICore := sets.String{}
usedAICore := sets.String{}
for k, r := range realUsedAICore {
phyID, _, err := common.GetDeviceID(r, "")
if err != nil {
continue
}
if unhealthyPhyID.Has(phyID) {
unhealthyAICore.Insert(k) // 不健康芯片上的 AICore 标记为不健康
}
usedAICore.Insert(k)
}
// 4. 计算还需要标记多少个空闲 AICore 为不健康
// 每个不健康芯片有 chipAICore 个 AICore，已标记一部分，剩余的从空闲 AICore 中选择
leftUnhealthyAICoreNum := unhealthyPhyID.Len()*int(ps.manager.GetChipAICore()) - unhealthyAICore.Len()
if leftUnhealthyAICoreNum < 0 {
return unhealthyAICore
}
// 5. 从空闲 AICore 中选取补充不健康数量
freeAICore := allAICore.Difference(usedAICore)
freeList := freeAICore.List()
for count := 0; count < leftUnhealthyAICoreNum; count++ {
unhealthyAICore.Insert(freeList[count])
}
return unhealthyAICore
}
3.2.6 软共享设备配置写入
// 配置文件内容格式（key=value）：
// PhysicalNPUId=0
// VirtualNPUId=1
// AICoreQuota=4
// HBMQuota=8
// ShmId=0
// SchedulingPolicy=1
3.2.7 设备挂载
func (ps *PluginServer) setNPUDeviceMount(resp *v1beta1.ContainerAllocateResponse,
ascendVisibleDevices []int) {
if !common.ParamOption.UseAscendDocker {
// 原始挂载模式：直接挂载 /dev/davinciN 设备文件
hwlog.RunLog.Info("device-plugin will use origin mount way")
mountDefaultDevice(resp, ps.defaultDevs) // 挂载默认设备（hi_ai_manager 等）
mountDevice(resp, ascendVisibleDevices, ps.ascendRuntimeOptions) // 挂载 NPU 设备
return
}
// ascend-docker 模式：通过环境变量传递设备信息，由 ascend-docker-runtime 挂载
common.SetAscendRuntimeEnv(ascendVisibleDevices, ps.ascendRuntimeOptions, resp)
hwlog.RunLog.Info("device-plugin will use ascend-docker to mount")
}
func mountDevice(resp *v1beta1.ContainerAllocateResponse, devices []int, ascendRuntimeOptions string) {
for _, deviceID := range devices {
// /dev/davinciN （物理设备）或 /dev/vdavinciN （虚拟设备）
containerPath, hostPath := getDevPath(fmt.Sprintf("%d", deviceID), ascendRuntimeOptions)
resp.Devices = append(resp.Devices, &v1beta1.DeviceSpec{
HostPath: hostPath,
ContainerPath: containerPath,
Permissions: "rw",
})
}
// 如果有 NPU 设备，额外挂载 uburma 和 ummu 设备
if len(devices) != 0 {
mountUBDevice(resp)
}
}
3.2.8 策略与错误恢复
func (ps *PluginServer) strategyForSendStats() string {
if ps.LastSendSuccess() {
return common.EmptyStrategy // 上次发送成功，无需特殊策略
}
if ps.deviceSyncStat.GetConsecutiveFailures() >= common.FailureCountThresholdForRestart {
// 连续失败超过重启阈值 → 重启整个 device plugin 进程
return common.ReStartDevicePluginStrategy
}
if ps.deviceSyncStat.GetConsecutiveFailures() >= common.FailureCountThresholdForReRegistry {
// 连续失败超过重注册阈值 → 重新注册到 kubelet
return common.ReRegistryStrategy
}
return common.EmptyStrategy
}
3.3 server.go — gRPC Server 启停与注册
3.3.1 Start — 启动流程
3.3.2 serve — gRPC Server 创建
func (ps *PluginServer) serve(socketWatcher *common.FileWatch) error {
// 1. 创建网络监听器（unix socket）
netListener, err := createNetListener(socketWatcher, ps.deviceType)
if err != nil {
return err
}
// 2. 配置 gRPC keepalive 参数
keepAlive := keepalive.ServerParameters{
Time: common.GrpcKeepAliveTime, // 空闲多久后发 ping
Timeout: common.GrpcKeepAliveTimeout, // ping 等待响应超时
}
// 3. 创建 gRPC Server，配置最大消息大小、并发流、keepalive
ps.grpcServer = grpc.NewServer(
grpc.MaxRecvMsgSize(common.MaxGRPCRecvMsgSize),
grpc.MaxConcurrentStreams(common.MaxGRPCConcurrentStreams),
grpc.KeepaliveParams(keepAlive),
)
// 4. 注册 DevicePluginServer
v1beta1.RegisterDevicePluginServer(ps.grpcServer, ps)
// 5. 异步启动 gRPC Server
go func() {
if err := ps.grpcServer.Serve(netListener); err != nil {
hwlog.RunLog.Errorf("GRPC server for '%s' crashed with error: %v", ps.deviceType, err)
}
}()
// 6. 等待 gRPC Server 就绪
for len(ps.grpcServer.GetServiceInfo()) <= 0 {
time.Sleep(time.Second)
}
hwlog.RunLog.Infof("device plugin (%s) start serving.", ps.deviceType)
return nil
}
3.3.3 createNetListener — Socket 创建
func createNetListener(socketWatcher *common.FileWatch, deviceType string) (net.Listener, error) {
// 1. 验证 device-plugin 目录路径和权限
realSocketPath, ok := common.VerifyPathAndPermission(v1beta1.DevicePluginPath, waitKubectlSockCreateTime)
if !ok {
return nil, fmt.Errorf("socket path verify failed")
}
// 2. 添加文件监听
if err := socketWatcher.WatchFile(realSocketPath); err != nil {
return nil, err
}
// 3. 构造 socket 文件路径：/var/lib/kubelet/device-plugin/{deviceType}.sock
pluginSocketPath := path.Join(realSocketPath, fmt.Sprintf("%s.sock", deviceType))
// 4. 如果 socket 文件已存在，先删除
if _, err := os.Stat(pluginSocketPath); err == nil {
if err = os.Remove(pluginSocketPath); err != nil {
return nil, err
}
}
// 5. 监听 unix socket
netListen, err := net.Listen("unix", pluginSocketPath)
if err != nil {
return nil, err
}
// 6. 设置 socket 文件权限和属主
if err = os.Chmod(pluginSocketPath, common.SocketChmod); err != nil {
return nil, err
}
if err = os.Lchown(pluginSocketPath, common.RootUID, common.RootGID); err != nil {
return nil, err
}
// 7. 使用限流器包装 listener（防止单 IP 连接过多）
return limiter.LimitListener(netListen, common.MaxConcurrentLimit,
common.MaxIPConnectionLimit, common.CacheSize)
}
3.3.4 register — kubelet 注册
func (ps *PluginServer) register() error {
// 1. 验证 kubelet socket 路径
realKubeletSockPath, ok := common.VerifyPathAndPermission(v1beta1.KubeletSocket, 0)
if !ok {
return fmt.Errorf("check kubelet socket file path failed")
}
// 2. 通过 gRPC 连接 kubelet
conn, err := grpc.Dial(realKubeletSockPath, grpc.WithInsecure(),
grpc.WithContextDialer(func(ctx context.Context, addr string) (net.Conn, error) {
if deadline, ok := ctx.Deadline(); ok {
return net.DialTimeout("unix", addr, time.Until(deadline))
}
return net.DialTimeout("unix", addr, 0)
}))
if err != nil {
return fmt.Errorf("connect to kubelet fail: %v", err)
}
defer conn.Close()
// 3. 构造注册请求
client := v1beta1.NewRegistrationClient(conn)
resourceName := customname.ReplaceDevicePublicType(ps.deviceType,
api.ResourceNamePrefix+ps.deviceType)
reqt := &v1beta1.RegisterRequest{
Version: v1beta1.Version, // Kubernetes Device Plugin API 版本
Endpoint: fmt.Sprintf("%s.sock", ps.deviceType), // socket 文件名
ResourceName: resourceName, // 资源名称（如 huawei.com/Ascend910A）
}
// 4. 发送注册请求
if _, err = client.Register(context.Background(), reqt); err != nil {
return fmt.Errorf("register to kubelet fail: %v", err)
}
return nil
}
3.4 types.go — 类型定义
// InterfaceServer 接口：定义了 PluginServer 的生命周期方法
type InterfaceServer interface {
Start(*common.FileWatch) error // 启动 gRPC Server 并注册到 kubelet
Stop() // 停止 gRPC Server
GetRestartFlag() bool // 获取重启标志
SetRestartFlag(bool) // 设置重启标志
LastSendSuccess() bool // 上次发送是否成功
}
// PluginServer 实现 Kubernetes Device Plugin v1beta1 接口
type PluginServer struct {
manager device.DevManager // 设备管理器（910/310/310P）
grpcServer *grpc.Server // gRPC Server 实例
isRunning *common.AtomicBool // ListAndWatch 是否在运行
cachedDevices []common.NpuDevice // 缓存的设备列表（用于 ListAndWatch）
deviceType string // 设备类型（如 Ascend910A、Ascend310P-2c-100）
ascendRuntimeOptions string // 运行时选项（空 或 "virtual"）
defaultDevs []string // 默认挂载设备（hi_ai_manager 等）
allocMapLock sync.RWMutex // klt2RealDevMap 读写锁
cachedLock sync.RWMutex // cachedDevices 读写锁
reciChan chan interface{} // 设备变更通知通道
stop chan interface{} // 停止通道
klt2RealDevMap map[string]string // kubelet 分配 → 实际设备映射
restart bool // 是否需要重启
deviceSyncStat *common.SendStats // 发送统计（成功/失败次数）
restartTimes atomic.Uint64 // 重启次数计数
podLock sync.Mutex // Pod 操作锁（Volcano 调度时使用）
}
// PodDevice Pod 设备信息
type PodDevice struct {
ResourceName string // 资源名称（如 huawei.com/Ascend910A）
DeviceIds []string // 设备 ID 列表
}
// PodResource 实现 kubelet Pod Resources 查询客户端
type PodResource struct {
conn *grpc.ClientConn // gRPC 连接
client v1alpha1.PodResourcesListerClient // Pod Resources 客户端
}
// stepTimeCM 慢节点步时间 ConfigMap 结构
type stepTimeCM struct {
Data stepTimeData `json:"data"`
}
type stepTimeData struct {
PerfDumpPath string `json:"PerfDumpPath"` // 性能 dump 路径
PerfDumpConfig string `json:"PerfDumpConfig"` // 性能 dump 配置
}
3.5 pod_resource.go — Pod 资源查询
3.5.1 GetPodResource — 获取 Pod 设备分配
3.5.2 IsPodMoveComplete — 检查 Pod 迁移完成
在热复位中的作用：
当设备不健康需要热复位时，必须先确认该设备上运行的 Pod 已完成迁移或销毁，否则不能执行复位。IsPodMoveComplete 通过查询 kubelet Pod Resources API，检查该设备是否仍被某个活跃 Pod 使用。
3.6 dpu.go — DPU 网卡健康监测
3.6.1 ListenDpu — DPU 状态周期查询
3.6.2 updateDpuHealthy — DPU 健康状态更新
func (hdm *HwDevManager) updateDpuHealthy(groupDevice map[string][]*common.NpuDevice) {
// 1. 构建 NPU→DPU 映射和 DPU→operstate 映射
uniqueMap := make(map[string]dpucontrol.BaseDpuInfo, api.NpuCountPerNode)
npuToDpusMap := make(map[string][]string, api.NpuCountPerNode)
// ... 填充映射 ...
// 2. 查询每个 DPU 的 operstate
dpuOperstateMap := make(map[string]string, api.NpuCountPerNode)
for _, dpu := range uniqueMap {
state, err := ethtool.GetInterfaceOperState(dpu.DeviceName)
if err != nil {
state = api.DpuStatusDown // 查询失败视为 down
}
dpuOperstateMap[dpu.DeviceName] = state
}
// 3. 为每个 NPU 计算 DPU 健康状态
for _, devices := range groupDevice {
for _, device := range devices {
device.DpuHealth = getDpuFaultInfoOfNpu(device.DeviceName, npuToDpusMap, dpuOperstateMap)
}
}
}
// getDpuFaultInfoOfNpu 判断单个 NPU 的 DPU 健康状态
// - PCIe 场景：一个 NPU 对应一个 DPU，DPU 故障 → NPU 不健康
// - UB 场景：一个 NPU 对应两个 DPU，至少一个 DPU 正常 → NPU 健康；全故障 → 不健康；一个故障 → 子健康
func getDpuFaultInfoOfNpu(npuName string, npuToDpuMap map[string][]string,
dpuOperstateMap map[string]string) string {
dpuFaultCount := 0
for npu, dpus := range npuToDpuMap {
if !isNpuMatched(npuName, npu) {
continue
}
for _, dpu := range dpus {
if dpuOperstateMap[dpu] != api.DpuStatusUp {
dpuFaultCount++
}
}
if dpuFaultCount == len(dpus) {
return v1beta1.Unhealthy // 全部 DPU 故障 → 不健康
} else if dpuFaultCount == common.OneDpuFault {
return api.DpuSubHealthy // 一个故障 → 子健康
}
return v1beta1.Healthy // 全部正常 → 健康
}
return v1beta1.Healthy
}
3.7 manager_v2.go — A5 扩展管理功能
3.7.1 getCardType — 板卡类型识别
func (hdm *HwDevManager) getCardType() (string, error) {
// 获取第一块设备的板卡信息
boardInfo, err := hdm.manager.GetDmgr().GetBoardInfo(hdm.allInfo.AllDevs[common.FirstDevice].LogicID)
if err != nil {
return "", err
}
// 仅对特定 BoardId 进行识别（A5300I 系列板卡）
if boardInfo.BoardId != npuCommon.A5300IBoardId &&
boardInfo.BoardId != npuCommon.A5300IBoardId2 &&
boardInfo.BoardId != npuCommon.A5300IBoardId3 {
return "", nil // 非目标板卡，返回空
}
// 通过主板 ID 区分 1P 卡和 4P 卡
mainBoardId := hdm.manager.GetDmgr().GetMainBoardId()
if mainBoardId == common.A5300IMainBoardId {
return common.A5300ICardName, nil // 1P 卡
}
if mainBoardId == common.A5300I4PMainBoardId {
return common.A54P300ICardName, nil // 4P 卡
}
return "", nil
}
3.7.2 getLevelList — Rank Table 拓扑生成
3.7.3 getROCEAddrList — RoCE 地址获取
func (hdm *HwDevManager) getROCEAddrList(dev *common.NpuDevice) []api.RankAddrItem {
// 通过 DPU 管理器获取 NPU 对应的 DPU IP 列表
dpuIPList, err := hdm.getNpuCorrespDpuInfo(dev)
if err != nil {
return []api.RankAddrItem{}
}
// 将每个 DPU IP 转换为 RankAddrItem
rankAddrList := make([]api.RankAddrItem, 0)
for _, ip := range dpuIPList {
rankAddrList = append(rankAddrList, api.RankAddrItem{
AddrType: "IPV4",
Addr: ip,
Ports: []string{},
PlaneId: api.DefaultRandAddrPlaneID,
})
}
return rankAddrList
}
3.8 plugin_v2.go — HCCL 拓扑环境变量
// setHcclTopoFilePathEnv 为容器设置 HCCL 拓扑文件路径环境变量
func (ps *PluginServer) setHcclTopoFilePathEnv(resp *v1beta1.ContainerAllocateResponse,
allNPUInfo common.NpuAllInfo) {
// 仅 A5 设备需要设置
if common.ParamOption.RealCardType != api.Ascend910A5 {
return
}
if len(allNPUInfo.AllDevs) == 0 {
return
}
// 根据板卡类型或 SuperPod 类型获取拓扑文件路径 key
productTypeKey := ps.getProductTypeKey(common.ParamOption.CardType, allNPUInfo)
if productTypeKey == -1 {
return
}
if resp.Envs == nil {
resp.Envs = make(map[string]string)
}
// 从映射表中查找拓扑文件路径
if path, exist := hcclTopoFilePathMap[productTypeKey]; exist {
resp.Envs[common.HcclTopoFilePathKey] = path
return
}
}
拓扑文件路径映射表：
var hcclTopoFilePathMap = map[int8]string{
common.ProductTypeServer: common.Server8PTopoPath, // 8P 服务器
common.ProductType1D: common.Pod1DTopoPath, // 1D Pod
common.ProductType2D: common.Pod2DTopoPath, // 2D Pod
common.ProductType16PServer: common.Server16PTopoPath, // 16P 服务器
common.ProductType32PServer: common.Server32PTopoPath, // 32P 服务器
common.ProductType1PCard: common.Card1PTopoPath, // 1P 卡
common.ProductType4PCard: common.Card4PTopoPath, // 4P 卡
}
3.9 npu_base_v2.go — NPU 基础拓扑信息
3.9.1 ProductBase — 产品场景判断
3.9.2 getRankLevelInfoKeyArr — 层级网络类型
func (n *NpuBase) getRankLevelInfoKeyArr() []string {
if n.productInfo == nil {
return []string{}
}
// 1. 优先从 cardType 映射表查找（标准卡）
if arr, ok := rankLevelInfoKeyArrMap[n.productInfo.cardType]; ok {
return arr
}
// 2. Pod 场景：4 层全部使用 UB/UBG/RoCE
if n.productInfo.isPodScene() {
return []string{
api.LevelInfoTypeUB, // Level 0: UB（Die-to-Die）
api.LevelInfoTypeUB, // Level 1: UB（Server内）
api.LevelInfoTypeUBG, // Level 2: UBG（Pod内跨服务器）
api.LevelInfoTypeRoCE, // Level 3: RoCE（跨 Pod）
}
}
// 3. 超级服务器场景
if n.productInfo.isSuperServer() {
return []string{
api.LevelInfoTypeUB, // Level 0: UB
api.LevelInfoTypeUB, // Level 1: UB
api.LevelInfoTypeUBoE, // Level 2: UBoE（UB over Ethernet）
api.LevelInfoTypeRoCE, // Level 3: RoCE
}
}
// 4. 普通服务器场景（无 Level 1）
if n.productInfo.isServer() {
return []string{
api.LevelInfoTypeUB, // Level 0: UB
api.LevelInfoTypeIgnore, // Level 1: 忽略
api.LevelInfoTypeUBoE, // Level 2: UBoE
api.LevelInfoTypeRoCE, // Level 3: RoCE
}
}
return []string{}
}
3.9.3 EID 解析与端口计算
3.9.4 getRandAddrByFuncEntityID — Rank 地址生成
func (n *NpuBase) getRandAddrByFuncEntityID(phyID int32, feID uint, netType string,
rankLevel int) []api.RankAddrItem {
// 1. 从缓存获取 URMA 设备信息
urmaDevInfoAll, exist := n.urmaDevInfoMap[phyID]
if !exist {
return nil
}
rankAddrList := make([]api.RankAddrItem, 0)
for _, devInfo := range urmaDevInfoAll {
// 2. 根据 feID 和 rankLevel 筛选 EID 列表
eidList := n.getEidListByFeIDAndRankLevel(feID, &devInfo, rankLevel)
for i := 0; i < len(eidList); i++ {
eid := eidList[i].Eid
eidStr := hex.EncodeToString(eid.Raw[:])
// 3. 获取 EID 对应的端口列表（含缓存）
portList, err := n.GetPortListByEid(phyID, eidStr, rankLevel)
if err != nil {
continue
}
// 4. 根据网络类型构造 RankAddrItem
item := n.createRankAddrItem(netType, eid, portList)
rankAddrList = append(rankAddrList, item)
}
}
return rankAddrList
}
3.9.5 createRankAddrItem — 地址项构造
func (n *NpuBase) createRankAddrItem(netType string, eid apiCommon.Eid,
ports []string) api.RankAddrItem {
planeId := api.DefaultRandAddrPlaneID
if netType == api.LevelInfoTypeUB {
// UB 类型：从 EID 最后一个字节计算 dieId 作为 planeId
val := int(eid.Raw[len(eid.Raw)-1])
dieId := 0
if val <= common.PhyLimit {
dieId = ((val - 1) / common.PhyPortNumPerDie) % common.DieNumPerDev
} else if common.LogicLowerLimit <= val && val <= common.LogicUpperLimit {
dieId = ((val - common.LogicLowerLimit) / common.LogicPortNumPerDie) % common.DieNumPerDev
}
planeId = strconv.Itoa(dieId)
}
if netType == api.LevelInfoTypeUBG || netType == api.LevelInfoTypeUB {
// UB/UBG：地址类型为 EID（十六进制字符串）
return api.RankAddrItem{
AddrType: addrTypeEID, // "EID"
Addr: hex.EncodeToString(eid.Raw[:]),
Ports: ports,
PlaneId: planeId,
}
}
if netType == api.LevelInfoTypeUBoE {
// UBoE：从 EID 后 4 字节提取 IPv4 地址
ipv4Bytes := eid.Raw[len(eid.Raw)-apiCommon.MaxUBCNAByteLen : len(eid.Raw)]
ipv4Str := fmt.Sprintf("%d.%d.%d.%d", ipv4Bytes[0], ipv4Bytes[1], ipv4Bytes[2], ipv4Bytes[3])
return api.RankAddrItem{
AddrType: addrTypeIPV4, // "IPV4"
Addr: ipv4Str,
Ports: ports,
PlaneId: planeId,
}
}
return api.RankAddrItem{}
}
3.9.6 getPortsList — 端口列表获取
3.10 types_v2.go — V2 类型定义
// Peer 拓扑对端信息
type Peer struct {
LocalId int `json:"local_id"` // 本地设备 ID
}
// TopoInfo 拓扑文件信息（从 JSON 文件解析）
type TopoInfo struct {
Version string `json:"version"` // 拓扑文件版本
HardwareType string `json:"hardware_type"` // 硬件类型
PeerCount int `json:"peer_count"` // 对端数量
PeerList []Peer `json:"peer_list"` // 对端列表
EdgeList []Edge `json:"edge_list"` // 边列表（连接信息）
}
// Edge 拓扑边信息（设备间连接）
type Edge struct {
NetLayer int `json:"net_layer"` // 网络层级（0-3）
LinkType string `json:"link_type"` // 链路类型（peer2net 等）
TopoType string `json:"topo_type"` // 拓扑类型
TopoInstanceId int `json:"topo_instance_id"` // 拓扑实例 ID
TopoAttr string `json:"topo_attr"` // 拓扑属性
LocalA int `json:"local_a"` // 本地设备 A
LocalAPorts []string `json:"local_a_ports"` // 本地设备 A 端口列表
LocalB int `json:"local_b"` // 本地设备 B
LocalBPorts []string `json:"local_b_ports"` // 本地设备 B 端口列表
Protocols []string `json:"protocols"` // 协议列表
Position string `json:"position"` // 位置信息
}
四、核心流程总览
4.1 设备插件完整生命周期
4.2 设备分配完整流程
4.3 故障处理完整流程
五、架构总结
5.1 设计模式
| 模式 | 应用 |
|---|---|
| 策略模式 | responseToKubelet 三种模式（动态切分/Volcano/标准） |
| 观察者模式 | PluginServer.Notify() → reciChan → ListAndWatch |
| 工厂模式 | NewHwDevManager 根据设备类型创建不同 manager |
| 状态机 | strategyForSendStats 发送状态机（OK→ReRegistry→RestartPlugin） |
| 代理模式 | PodResource 代理 kubelet Pod Resources API |
| 缓存模式 | NpuBase.eidPortMap EID→Port 缓存、ProductBase.topoInfo 拓扑文件缓存 |
5.2 并发控制
5.3 关键常量
| 常量 | 值 | 含义 |
|---|---|---|
memoryRadix |
1024 | 内存单位进制（字节→GB） |
nodeAnnotationUpdateInterval |
60 | 节点注解更新间隔（秒） |
serverIndexKey |
“serverIndex” | 节点注解 key |
serverTypeKey |
“serverType” | 节点注解 key |
cardTypeKey |
“cardType” | 节点注解 key |
listenDpuInterval |
5s | DPU 查询间隔 |
maxUpdateInterval |
6h | DPU ConfigMap 最大更新间隔 |
socketPath |
/var/lib/kubelet/pod-resources/kubelet.sock | Pod Resources socket |
callTimeout |
2s | Pod Resources API 超时 |
waitKubectlSockCreateTime |
5min | 等待 kubelet socket 创建超时 |
本文档严格基于源码分析，涵盖 10 个源文件共约 4750 行代码的逐行级深度解析。
鲲鹏昇腾开发者社区是面向全社会开放的“联接全球计算开发者，聚合华为+生态”的社区，内容涵盖鲲鹏、昇腾资源，帮助开发者快速获取所需的知识、经验、软件、工具、算力，支撑开发者易学、好用、成功，成为核心开发者。
更多推荐
9
0- 0
已为社区贡献18条内容
所有评论(0)