英伟达network-operator源码阅读

a-c-dream 发布于

GitHub - Mellanox/network-operator: Mellanox Network Operator

找不到包用这个

1
2
go clean --modcache
go mod download

结构

README.md：通常包含项目的概述、安装和使用说明。

docs/：可能包含详细的文档和设计说明。

pkg/：包含Go语言的源代码。

cmd/：包含项目的主程序入口。

deploy/：包含部署相关的YAML文件。

概念和一些信息

海光不需要加载

寒武纪使用sudo modprobe cambricon-peermem加载peermem

目标：通过修改英伟达network-operator实现自动加载peermem

OFED driver container

只会部署到支持Mellanox的node上

只要在主机上安装该镜像，容器将完成下面的功能

  • 重新加载 Mellanox OFED 提供的内核模块
  • 将容器的根目录挂载到 /run/mellanox/drivers/。如果该目录被映射到主机，该容器的内容就可以与主机或其他容器共享。它的一个用例是编译 Nvidia Peer Memory 客户端模块。

RDMA shared device plugin

支持 IB and RoCE HCA的plugin，作为daemonset运行

InfiniBand Kubernetes

InfiniBand Kubernetes 提供了一个守护进程 ib-kubernetes，它与 InfiniBand SR-IOV CNI 和 Multus CNI 一起工作，Pod 发生变化（创建/更新/删除）时，它会读取 Pod 的网络annotation并获取其对应的网络 CRD 并读取 PKey，对于带有注释 mellanox.infiniband.app 的 Pod，将新生成的 Guid 或 CRD cni-args 的 guid 字段中的预定义 Guid 添加到该 PKey。

secondaryNetwork

辅助网络，目前支持下面几种

Multus-CNI：支持 Kubernetes 中二级网络的委托 CNI 插件
CNI 插件： 目前仅支持 containernetworking-plugins
IP Over Infiniband（IPoIB）CNI 插件： 允许用户创建 IPoIB 子链接并将其移动到 pod。
IPAM CNI：IPAM CNI 和相关配置的去向

nicclusterpolicy_controller

NicClusterPolicyReconciler 结构体

NicClusterPolicyReconciler 是一个结构体，包含了用于管理 NicClusterPolicy 自定义资源状态的客户端、方案和各种提供者。

  • client.Client: 与 Kubernetes API 进行交互。
  • Scheme: 定义资源的运行时方案。
  • ClusterTypeProvider、StaticConfigProvider、DocaDriverImagesProvider: 提供必要的配置和驱动镜像。
  • MigrationCh: 用于处理迁移流程的通道。
  • stateManager: 管理自定义资源的状态同步。

Reconcile 方法

Reconcile方法是控制器的核心，负责：

  1. 处理迁移流程: 确保迁移通道被处理。
  2. 获取自定义资源: 根据请求检索 NicClusterPolicy 实例。
  3. 处理 NotFound 错误: 如果资源被删除，则进行清理。
  4. 检查资源名称: 确保只处理特定名称的资源。
  5. 创建状态服务目录: 初始化状态服务目录，并添加必要的信息。
  6. 节点信息提供者: 如果指定了 OFED 驱动程序，则收集节点信息并更新状态目录。
  7. 同步状态并更新状态: 同步状态并更新自定义资源状态。
  8. 处理 OFED 等待标签: 更新与 OFED 驱动程序就绪状态相关的节点标签。

辅助方法

  • requeue(): 触发带有延迟的重新排队。
  • handleMOFEDWaitLabels(): 更新节点标签，以指示 OFED 驱动程序是否应等待。
  • handleMOFEDWaitLabelsNoConfig(): 处理未配置 OFED 的场景，并相应地更新标签。
  • setOFEDWaitLabel(): 设置或删除节点上的 OFED 等待标签。
  • updateCrStatus(): 根据状态同步结果更新自定义资源的状态。
  • handleUnsupportedInstance(): 通过将其状态更新为忽略来处理不受支持的实例。

SetupWithManager 方法

SetupWithManager使用提供的管理器设置控制器。

  • 它创建一个状态管理器，并设置对 NicClusterPolicy 和相关资源的监视。
  • 它定义了事件的谓词和处理程序，以确保高效的协调。

此控制器确保 NicClusterPolicy 资源得到正确管理，同步 Kubernetes 集群中的期望状态与实际状态，并处理 OFED 驱动程序的节点特定配置。

peermem模块

Starting from v465 NVIDIA GPU driver includes a built-in nvidia_peermem module which is a replacement for nv_peer_mem module. NVIDIA GPU operator manages nvidia_peermem module loading.

由于现在peermem已经由NVIDIA GPU operator manages负责，所以用之前的代码。最低版本为0.1.0，最高版本为v23.5.0

还是从nicclusterpolicy_controller.go入手

nicclusterpolicy_controller.go

在Reconcile中，判断是否写了ofed，nvpeermem，rdma share等，如果写了将信息存入infoProvider，然后将infoProvider添加到sc中，最后调用r.stateManager.SyncState(instance, sc)进行同步。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
if instance.Spec.OFEDDriver != nil || instance.Spec.NVPeerDriver != nil ||
		instance.Spec.RdmaSharedDevicePlugin != nil || instance.Spec.SriovDevicePlugin != nil {
		// Create node infoProvider and add to the service catalog
		reqLogger.V(consts.LogLevelInfo).Info("Creating Node info provider")
		nodeList := &corev1.NodeList{}
		err = r.List(context.TODO(), nodeList, nodeinfo.MellanoxNICListOptions...)
		if err != nil {
			// Failed to get node list
			reqLogger.V(consts.LogLevelError).Info("Error occurred on LIST nodes request from API server.", "error:", err)
			return reconcile.Result{}, err
		}
		nodePtrList := make([]*corev1.Node, len(nodeList.Items))
		nodeNames := make([]*string, len(nodeList.Items))
		for i := range nodePtrList {
			nodePtrList[i] = &nodeList.Items[i]
			nodeNames[i] = &nodeList.Items[i].Name
		}
		reqLogger.V(consts.LogLevelDebug).Info("Node info provider with", "Nodes:", nodeNames)
		infoProvider := nodeinfo.NewProvider(nodePtrList)
		sc.Add(state.InfoTypeNodeInfo, infoProvider)
	}

接下来具体来看SyncState的实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
// SyncState attempts to reconcile the system by invoking Sync on each of the states
func (smgr *stateManager) SyncState(customResource interface{}, infoCatalog InfoCatalog) (Results, error) {
	// Sync groups of states, transition from one group to the other when a group finishes
	log.V(consts.LogLevelInfo).Info("Syncing system state")
	managerResult := Results{
		Status: SyncStateNotReady,
	}
	statesReady := true

	for i, stateGroup := range smgr.stateGroups {
		log.V(consts.LogLevelInfo).Info("Sync State group", "index", i)
		results := stateGroup.Sync(customResource, infoCatalog)
		managerResult.StatesStatus = append(managerResult.StatesStatus, results...)

		done, err := stateGroup.SyncDone()
		if err != nil {
			log.V(consts.LogLevelError).Info("Error while syncing states", "Error:", err)
			return managerResult, err
		}

		if !done {
			// We need to have at lease one not ready state to mark CR as notReady
			statesReady = false
			log.V(consts.LogLevelInfo).Info("State Group Not ready")
		} else {
			log.V(consts.LogLevelInfo).Info("Sync Completed successfully for State group", "index", i)
		}
	}
	if statesReady {
		// Done Syncing CR
		managerResult.Status = SyncStateReady
		log.V(consts.LogLevelInfo).Info("Sync Done for custom resource")
	}

	return managerResult, nil
}

这里又调用了下面的函数进行处理

1
2
3
stateGroup.Sync(customResource, infoCatalog)
sg.states[i].Sync(customResource, infoCatalog)
func (s *stateNVPeer) Sync(customResource interface{}, infoCatalog InfoCatalog) (SyncState, error)

最后发现关于peermem的代码在pkg/state/state_nv_peer.go

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
package state

import (
	"strconv"

	"github.com/pkg/errors"
	appsv1 "k8s.io/api/apps/v1"
	v1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/source"

	mellanoxv1alpha1 "github.com/Mellanox/network-operator/api/v1alpha1"
	"github.com/Mellanox/network-operator/pkg/config"
	"github.com/Mellanox/network-operator/pkg/consts"
	"github.com/Mellanox/network-operator/pkg/nodeinfo"
	"github.com/Mellanox/network-operator/pkg/render"
	"github.com/Mellanox/network-operator/pkg/utils"
)

const (
	stateNVPeerName        = "state-NV-Peer"
	stateNVPeerDescription = "Nvidia Peer Memory driver deployed in the cluster"

	// 从这个GPU驱动版本开始，nvPeerMem驱动程序不应部署
	maxCudaVersionMajor = 465
)

// NewStateNVPeer 创建一个新的NVPeer驱动状态
func NewStateNVPeer(k8sAPIClient client.Client, scheme *runtime.Scheme, manifestDir string) (State, error) {
	files, err := utils.GetFilesWithSuffix(manifestDir, render.ManifestFileSuffix...)
	if err != nil {
		return nil, errors.Wrap(err, "无法从清单目录获取文件")
	}

	renderer := render.NewRenderer(files)
	return &stateNVPeer{
		stateSkel: stateSkel{
			name:        stateNVPeerName,
			description: stateNVPeerDescription,
			client:      k8sAPIClient,
			scheme:      scheme,
			renderer:    renderer,
		}}, nil
}

// stateNVPeer 结构体
type stateNVPeer struct {
	stateSkel
}

// nvPeerRuntimeSpec 运行时配置
type nvPeerRuntimeSpec struct {
	runtimeSpec
	CPUArch        string
	OSName         string
	OSVer          string
	UseHostOFED    bool
	MaxCudaVersion int
}

// nvPeerManifestRenderData 清单渲染数据
type nvPeerManifestRenderData struct {
	CrSpec       *mellanoxv1alpha1.NVPeerDriverSpec
	NodeAffinity *v1.NodeAffinity
	RuntimeSpec  *nvPeerRuntimeSpec
}

// Sync 尝试让系统与State表示的所需状态匹配。同步操作必须相对较短且不能阻塞执行线程。
func (s *stateNVPeer) Sync(customResource interface{}, infoCatalog InfoCatalog) (SyncState, error) {
	cr := customResource.(*mellanoxv1alpha1.NicClusterPolicy)
	log.V(consts.LogLevelInfo).Info(
		"同步自定义资源", "状态:", s.name, "名称:", cr.Name, "命名空间:", cr.Namespace)

	if cr.Spec.NVPeerDriver == nil {
		// 如果不需要运行此状态或发生更新且需要删除已创建的资源。
		return s.handleStateObjectsDeletion()
	}

	// 填充清单渲染数据并渲染对象
	nodeInfo := infoCatalog.GetNodeInfoProvider()
	if nodeInfo == nil {
		return SyncStateError, errors.New("意外状态，目录不提供节点信息")
	}

	objs, err := s.getManifestObjects(cr, nodeInfo)
	if err != nil {
		return SyncStateNotReady, errors.Wrap(err, "无法从清单创建k8s对象")
	}
	if len(objs) == 0 {
		// getManifestObjects 返回没有对象，这意味着没有对象需要应用到集群中，
		// 因为（很可能）没有找到Mellanox/Nvidia硬件（没有找到Mellanox和Nvidia标签）。
		// 返回SyncStateNotReady以便我们重试同步。
		return SyncStateNotReady, nil
	}

	// 创建对象（如果不存在），更新对象（如果存在）
	err = s.createOrUpdateObjs(func(obj *unstructured.Unstructured) error {
		if err := controllerutil.SetControllerReference(cr, obj, s.scheme); err != nil {
			return errors.Wrap(err, "无法为对象设置控制器引用")
		}
		return nil
	}, objs)
	if err != nil {
		return SyncStateNotReady, errors.Wrap(err, "无法创建/更新对象")
	}

	// 检查是否应忽略nvPeerMem状态
	// 支持从nv_peer_mem模块升级到nvidia_peermem模块的解决方法
	if s.shouldIgnoreStatus(nodeInfo) {
		log.V(consts.LogLevelInfo).Info("检测到具有内置nvidia_peermem模块的GPU驱动程序版本。" +
			"忽略NV Peer驱动程序状态")
		return SyncStateIgnore, nil
	}

	// 检查对象状态
	syncState, err := s.getSyncState(objs)
	if err != nil {
		return SyncStateNotReady, errors.Wrap(err, "无法获取同步状态")
	}
	return syncState, nil
}

// GetWatchSources 获取应监视的源种类映射，按源种类名称键入
func (s *stateNVPeer) GetWatchSources() map[string]*source.Kind {
	wr := make(map[string]*source.Kind)
	wr["DaemonSet"] = &source.Kind{Type: &appsv1.DaemonSet{}}
	return wr
}

// getManifestObjects 获取清单对象
func (s *stateNVPeer) getManifestObjects(
	cr *mellanoxv1alpha1.NicClusterPolicy,
	nodeInfo nodeinfo.Provider) ([]*unstructured.Unstructured, error) {
	attrs := nodeInfo.GetNodesAttributes(
		nodeinfo.NewNodeLabelFilterBuilder().
			WithLabel(nodeinfo.NodeLabelMlnxNIC, "true").
			WithLabel(nodeinfo.NodeLabelNvGPU, "true").
			Build())
	if len(attrs) == 0 {
		log.V(consts.LogLevelInfo).Info("在集群中未找到具有Mellanox NIC和Nvidia GPU的节点。")
		return []*unstructured.Unstructured{}, nil
	}

	// TODO: 根据CPUXOS矩阵多次渲染daemonset（目前假设所有节点都是相同的）
	if err := s.checkAttributesExist(attrs[0],
		nodeinfo.AttrTypeCPUArch, nodeinfo.AttrTypeOSName, nodeinfo.AttrTypeOSVer); err != nil {
		return nil, err
	}

	renderData := &nvPeerManifestRenderData{
		CrSpec:       cr.Spec.NVPeerDriver,
		NodeAffinity: cr.Spec.NodeAffinity,
		RuntimeSpec: &nvPeerRuntimeSpec{
			runtimeSpec:    runtimeSpec{config.FromEnv().State.NetworkOperatorResourceNamespace},
			CPUArch:        attrs[0].Attributes[nodeinfo.AttrTypeCPUArch],
			OSName:         attrs[0].Attributes[nodeinfo.AttrTypeOSName],
			OSVer:          attrs[0].Attributes[nodeinfo.AttrTypeOSVer],
			UseHostOFED:    cr.Spec.OFEDDriver == nil,
			MaxCudaVersion: maxCudaVersionMajor,
		},
	}
	// 渲染对象
	log.V(consts.LogLevelDebug).Info("渲染对象", "数据:", renderData)
	objs, err := s.renderer.RenderObjects(&render.TemplatingData{Data: renderData})
	if err != nil {
		return nil, errors.Wrap(err, "渲染对象失败")
	}
	log.V(consts.LogLevelDebug).Info("已渲染", "对象:", objs)
	return objs, nil
}

// shouldIgnoreStatus 检查是否应忽略nvPeerMem DS状态
// 支持从nv_peer_mem模块切换到nvidia_peermem模块的解决方法，从v465版本开始，这个模块是GPU驱动程序的一部分。
// 如果集群中的所有节点GPU驱动程序版本大于465，则应忽略nvPeerMem状态。
// 如果某些节点仍使用GPU驱动程序版本小于465或没有节点具有GPU驱动程序，则应继续同步尝试。
// 我们不会删除DS，只是忽略其状态。这是为了在GPU驱动程序版本降级或在具有混合版本GPU驱动程序的环境中
// 节点在具有新版本的节点之后出现时能够恢复nv-peer-mem。
// NV PeerMem配置应从NicClusterPolicy中显式删除以删除DS。
func (s *stateNVPeer) shouldIgnoreStatus(nodeInfo nodeinfo.Provider) bool {
	attrs := nodeInfo.GetNodesAttributes(
		nodeinfo.NewNodeLabelNoValFilterBuilderr().
			WithLabel(nodeinfo.NodeLabelCudaVersionMajor).
			Build())
	if len(attrs) == 0 {
		// 没有驱动程序信息可用，应继续同步
		return false
	}
	for _, attr := range attrs {
		cudaVersion, err := strconv.Atoi(attr.Attributes[nodeinfo.AttrTypeCudaVersionMajor])
		if err != nil {
			// 未知的cuda版本，继续同步
			log.V(consts.LogLevelInfo).Info("检查GPU驱动程序版本失败")
			return false
		}
		if cudaVersion < maxCudaVersionMajor {
			// 找到旧的GPU驱动程序的节点，继续同步
			return false
		}
	}
	return true
}

pkg/state/state_skel.go

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
package state

import (
	"context"
	"encoding/json"
	"fmt"

	"github.com/pkg/errors"
	appsv1 "k8s.io/api/apps/v1"
	k8serrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/meta"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"

	"github.com/Mellanox/network-operator/pkg/consts"
	"github.com/Mellanox/network-operator/pkg/nodeinfo"
	"github.com/Mellanox/network-operator/pkg/render"
)

// 运行时配置
type runtimeSpec struct {
	Namespace string
}

// 状态骨架，嵌入实现State接口的结构体中，提供实现状态所需的常见结构和功能
type stateSkel struct {
	name        string
	description string

	client   client.Client
	scheme   *runtime.Scheme
	renderer render.Renderer
}

// Name 返回状态名称
func (s *stateSkel) Name() string {
	return s.name
}

// Description 返回状态描述
func (s *stateSkel) Description() string {
	return s.description
}

// 获取支持的GVK（Group Version Kind）
func getSupportedGVKs() []schema.GroupVersionKind {
	return []schema.GroupVersionKind{
		{Group: "", Kind: "ServiceAccount", Version: "v1"},
		{Group: "", Kind: "ConfigMap", Version: "v1"},
		{Group: "apps", Kind: "DaemonSet", Version: "v1"},
		{Group: "apps", Kind: "Deployment", Version: "v1"},
		{Group: "apiextensions.k8s.io", Kind: "CustomResourceDefinition", Version: "v1"},
		{Group: "rbac.authorization.k8s.io", Kind: "ClusterRole", Version: "v1"},
		{Group: "rbac.authorization.k8s.io", Kind: "ClusterRoleBinding", Version: "v1"},
		{Group: "rbac.authorization.k8s.io", Kind: "Role", Version: "v1"},
		{Group: "rbac.authorization.k8s.io", Kind: "RoleBinding", Version: "v1"},
		{Group: "k8s.cni.cncf.io", Kind: "NetworkAttachmentDefinition", Version: "v1"},
		{Group: "batch", Kind: "CronJob", Version: "v1"},
		{Group: "security.openshift.io", Kind: "SecurityContextConstraints", Version: "v1"},
	}
}

// 获取对象
func (s *stateSkel) getObj(obj *unstructured.Unstructured) error {
	log.V(consts.LogLevelInfo).Info("获取对象", "命名空间:", obj.GetNamespace(), "名称:", obj.GetName())
	err := s.client.Get(
		context.TODO(), types.NamespacedName{Name: obj.GetName(), Namespace: obj.GetNamespace()}, obj)
	if k8serrors.IsNotFound(err) {
		// 对象不存在
		log.V(consts.LogLevelInfo).Info("对象不存在")
	}
	return err
}

// 创建对象
func (s *stateSkel) createObj(obj *unstructured.Unstructured) error {
	s.checkDeleteSupported(obj)
	log.V(consts.LogLevelInfo).Info("创建对象", "命名空间:", obj.GetNamespace(), "名称:", obj.GetName())
	toCreate := obj.DeepCopy()
	if err := s.client.Create(context.TODO(), toCreate); err != nil {
		if k8serrors.IsAlreadyExists(err) {
			log.V(consts.LogLevelInfo).Info("对象已存在")
		}
		return err
	}
	log.V(consts.LogLevelInfo).Info("对象创建成功")
	return nil
}

// 检查是否支持删除对象
func (s *stateSkel) checkDeleteSupported(obj *unstructured.Unstructured) {
	for _, gvk := range getSupportedGVKs() {
		objGvk := obj.GroupVersionKind()
		if objGvk.Group == gvk.Group && objGvk.Version == gvk.Version && objGvk.Kind == gvk.Kind {
			return
		}
	}
	log.V(consts.LogLevelWarning).Info("对象不支持删除",
		"命名空间:", obj.GetNamespace(), "名称:", obj.GetName(), "GVK", obj.GroupVersionKind())
}

// 更新对象
func (s *stateSkel) updateObj(obj *unstructured.Unstructured) error {
	log.V(consts.LogLevelInfo).Info("更新对象", "命名空间:", obj.GetNamespace(), "名称:", obj.GetName())
	desired := obj.DeepCopy()
	if err := s.client.Update(context.TODO(), desired); err != nil {
		return errors.Wrap(err, "更新资源失败")
	}
	log.V(consts.LogLevelInfo).Info("对象更新成功")
	return nil
}

// 创建或更新对象
func (s *stateSkel) createOrUpdateObjs(
	setControllerReference func(obj *unstructured.Unstructured) error,
	objs []*unstructured.Unstructured) error {
	for _, desiredObj := range objs {
		log.V(consts.LogLevelInfo).Info("处理清单对象", "类型:", desiredObj.GetKind(), "名称", desiredObj.GetName())
		// 为对象设置控制器引用，以便在CR删除时进行清理
		if err := setControllerReference(desiredObj); err != nil {
			return errors.Wrap(err, "设置对象控制器引用失败")
		}

		s.addStateSpecificLabels(desiredObj)

		err := s.createObj(desiredObj)
		if err == nil {
			// 对象创建成功
			continue
		}
		if !k8serrors.IsAlreadyExists(err) {
			// 发生某些错误
			return err
		}

		currentObj := desiredObj.DeepCopy()
		if err := s.getObj(currentObj); err != nil {
			// 发生某些错误
			return err
		}

		if err := s.mergeObjects(desiredObj, currentObj); err != nil {
			return err
		}

		// 对象已存在，更新它
		if err := s.updateObj(desiredObj); err != nil {
			return err
		}
	}
	return nil
}

// 添加状态特定标签
func (s *stateSkel) addStateSpecificLabels(obj *unstructured.Unstructured) {
	labels := obj.GetLabels()
	if labels == nil {
		labels = make(map[string]string)
	}
	labels[consts.StateLabel] = s.name
	obj.SetLabels(labels)
}

// 处理状态对象的删除
func (s *stateSkel) handleStateObjectsDeletion() (SyncState, error) {
	log.V(consts.LogLevelInfo).Info(
		"CR中的状态规范为空，如果需要，删除现有对象", "状态:", s.name)
	found, err := s.deleteStateRelatedObjects()
	if err != nil {
		return SyncStateError, errors.Wrap(err, "删除k8s对象失败")
	}
	if found {
		log.V(consts.LogLevelInfo).Info("状态正在删除对象", "状态:", s.name)
		return SyncStateNotReady, nil
	}
	return SyncStateIgnore, nil
}

// 删除与状态相关的对象
func (s *stateSkel) deleteStateRelatedObjects() (bool, error) {
	stateLabel := map[string]string{
		consts.StateLabel: s.name,
	}
	found := false
	for _, gvk := range getSupportedGVKs() {
		l := &unstructured.UnstructuredList{}
		l.SetGroupVersionKind(gvk)
		err := s.client.List(context.Background(), l, client.MatchingLabels(stateLabel))
		if meta.IsNoMatchError(err) {
			continue
		}
		if err != nil {
			return false, err
		}
		if len(l.Items) > 0 {
			found = true
		}
		for _, obj := range l.Items {
			obj := obj
			if obj.GetDeletionTimestamp() == nil {
				err := s.client.Delete(context.TODO(), &obj)
				if err != nil {
					return true, err
				}
			}
		}
	}
	return found, nil
}

// 合并对象
func (s *stateSkel) mergeObjects(updated, current *unstructured.Unstructured) error {
	// 设置资源版本
	updated.SetResourceVersion(current.GetResourceVersion())

	gvk := updated.GroupVersionKind()
	if gvk.Group == "" && gvk.Kind == "ServiceAccount" {
		return s.mergeServiceAccount(updated, current)
	}
	return nil
}

// 对于ServiceAccount，保留现有的secrets
func (s *stateSkel) mergeServiceAccount(updated, current *unstructured.Unstructured) error {
	curSecrets, ok, err := unstructured.NestedSlice(current.Object, "secrets")
	if err != nil {
		return err
	}
	if ok {
		if err := unstructured.SetNestedField(updated.Object, curSecrets, "secrets"); err != nil {
			return err
		}
	}

	curImagePullSecrets, ok, err := unstructured.NestedSlice(current.Object, "imagePullSecrets")
	if err != nil {
		return err
	}
	if ok {
		if err := unstructured.SetNestedField(updated.Object, curImagePullSecrets, "imagePullSecrets"); err != nil {
			return err
		}
	}
	return nil
}

// 获取对象同步状态
func (s *stateSkel) getSyncState(objs []*unstructured.Unstructured) (SyncState, error) {
	for _, obj := range objs {
		if obj == nil {
			continue
		}
		gvk := obj.GroupVersionKind()
		if gvk.Group == "apps" && gvk.Kind == "DaemonSet" {
			ready, err := s.isDaemonSetReady(obj)
			if err != nil {
				return SyncStateError, errors.Wrapf(err, "检查DaemonSet就绪状态失败")
			}
			if !ready {
				return SyncStateNotReady, nil
			}
		}
	}
	return SyncStateReady, nil
}

// 检查DaemonSet是否就绪
func (s *stateSkel) isDaemonSetReady(uds *unstructured.Unstructured) (bool, error) {
	bs, err := uds.MarshalJSON()
	if err != nil {
		return false, err
	}

	ds := appsv1.DaemonSet{}
	err = json.Unmarshal(bs, &ds)
	if err != nil {
		return false, err
	}

	status := ds.Status
	log.V(consts.LogLevelDebug).Info(
		"检查DaemonSet就绪状态",
		"状态:", ds.Status,
		"期望副本数:", status.DesiredNumberScheduled,
		"当前副本数:", status.CurrentNumberScheduled,
		"可用副本数:", status.NumberAvailable,
		"已准备副本数:", status.NumberReady,
		"更新副本数:", status.UpdatedNumberScheduled)

	return status.DesiredNumberScheduled > 0 &&
		status.DesiredNumberScheduled == status.NumberReady &&
		status.UpdatedNumberScheduled == status.DesiredNumberScheduled &&
		status.NumberAvailable == status.DesiredNumberScheduled, nil
}

// 检查属性是否存在
func (s *stateSkel) checkAttributesExist(
	attrs nodeinfo.NodeAttributes, attrTypes ...nodeinfo.AttributeType) error {
	for _, attrType := range attrTypes {
		if _, ok := attrs[attrType]; !ok {
			return fmt.Errorf("缺少所需属性类型:%s", attrType)
		}
	}
	return nil
}

manifests/stage-nv-peer-mem-driver

  1. 这些文件定义了一个DaemonSet,用于在每个符合条件的节点上部署nv_peer_mem驱动。

  2. 主要的实现在0050_nv-peer-mem-driver-ds.yaml文件中:

    a) DaemonSet确保在每个满足nodeSelector条件的节点上运行一个Pod:

    1
    2
    3
    4
    nodeSelector:
      feature.node.kubernetes.io/pci-15b3.present: "true"
      nvidia.com/gpu.present: "true"
      network.nvidia.com/operator.mofed.wait: "false"

    b) 使用initContainer确保NVIDIA驱动已加载:

    1
    2
    3
    4
    initContainers:
    - name: gpu-driver-validation
      ...
      args: ["... until nvidia-smi; do echo waiting for nvidia drivers to be loaded; sleep 5; done"]

    c) 主容器负责加载nv_peer_mem模块:

    1
    2
    3
    containers:
      - image: {{ .CrSpec.Repository }}/{{ .CrSpec.Image }}-{{ .CrSpec.Version }}:{{ .RuntimeSpec.CPUArch }}-{{ .RuntimeSpec.OSName }}{{ .RuntimeSpec.OSVer }}
        name: nv-peer-mem-driver-container

    d) 使用livenessProbe和readinessProbe检查nv_peer_mem模块是否已加载:

    1
    2
    3
    4
    livenessProbe:
      exec:
        command:
          [sh, -c, 'lsmod | grep nv_peer_mem']

  3. 该DaemonSet会自动在符合条件的节点上部署和运行,从而实现nv_peer_mem的自动加载。

  4. 其他YAML文件(0010到0040)定义了必要的RBAC权限和安全上下文约束,确保Pod有足够的权限来加载内核模块。

  5. 注意到有一个版本检查,防止在CUDA驱动版本>=465的节点上调度此Pod,因为较新版本的CUDA驱动应该包含内置的nvidia_peermem模块:

    1
    2
    3
    4
    - key: nvidia.com/cuda.driver.major
      operator: Lt
      values:
        - "{{ .RuntimeSpec.MaxCudaVersion }}"

总之,这个实现通过Kubernetes的DaemonSet机制,在每个符合条件的节点上自动部署和运行一个负责加载nv_peer_mem模块的Pod,从而实现了nv_peer_mem的自动加载。

修改manifests/stage-nv-peer-mem-driver/0050_nv-peer-mem-driver-ds.yaml

通过阅读源码打算修改manifests/stage-nv-peer-mem-driver/0050_nv-peer-mem-driver-ds.yaml来实现寒武纪的peermem模块的加载。

在v23.5.0的代码基础上进行修改

非常抱歉我之前的误解。根据您的需求，我们需要执行 sudo modprobe cambricon-peermem 命令。让我为您重新说明需要修改的部分：

  1. 修改 initContainer 部分：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
spec:
  template:
    spec:
      initContainers:
      - name: load-cambricon-peermem
        image: {{ .CrSpec.Repository }}/{{ .CrSpec.Image }}-{{ .CrSpec.Version }}:{{ .RuntimeSpec.CPUArch }}-{{ .RuntimeSpec.OSName }}{{ .RuntimeSpec.OSVer }}
        command: ["/bin/sh", "-c"]
        args:
          - |
            set -e
            echo "Loading cambricon-peermem module..."
            modprobe cambricon-peermem || echo "Failed to load cambricon-peermem module"
            echo "cambricon-peermem module load attempt completed"
        securityContext:
          privileged: true
        volumeMounts:
          - name: host-modules
            mountPath: /lib/modules
            readOnly: true
  1. 在 volumes 部分添加：
1
2
3
4
5
volumes:
# ... (保持原有的卷配置)
- name: host-modules
  hostPath:
    path: /lib/modules
  1. 修改主容器的探针：

将原来检查 nv_peer_mem 的探针改为检查 cambricon-peermem：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
containers:
  - name: cambricon-peermem-driver-container
    # ... (其他配置保持不变)
    livenessProbe:
      exec:
        command:
          [sh, -c, 'lsmod | grep cambricon-peermem']
      periodSeconds: 30
      initialDelaySeconds: 30
      failureThreshold: 1
      successThreshold: 1
    readinessProbe:
      exec:
        command:
          [sh, -c, 'lsmod | grep cambricon-peermem']
      periodSeconds: 30
      initialDelaySeconds: 10
      failureThreshold: 1
  1. 更新 DaemonSet 的名称和标签（可选，但建议为了清晰）：
1
2
3
4
metadata:
  labels:
    app: cambricon-peermem-driver-{{ .RuntimeSpec.CPUArch }}-{{ .RuntimeSpec.OSName }}{{ .RuntimeSpec.OSVer }}
  name: cambricon-peermem-driver-{{ .RuntimeSpec.CPUArch }}-{{ .RuntimeSpec.OSName }}{{ .RuntimeSpec.OSVer }}-ds
  1. 更新节点选择器（如果需要）：

如果 cambricon-peermem 模块需要特定的硬件支持，您可能需要更新节点选择器。例如：

1
2
nodeSelector:
  cambricon.com/device: "true"

请注意，您可能需要根据实际情况调整节点选择器。

这些修改将使 DaemonSet 在每个符合条件的节点上执行 modprobe cambricon-peermem 命令，从而加载 cambricon-peermem 模块。同时，它还会检查模块是否成功加载。

记住，这种方法假设 cambricon-peermem 模块已经在主机上可用。如果模块不存在，modprobe 命令会失败。另外，使用 privileged: true 会给予容器很高的权限，这可能会带来安全风险。请确保在生产环境中谨慎使用，并相应地调整您的安全策略。

编译部署

要通过编译源码部署NVIDIA Network Operator，可以按照以下步骤进行：

  1. 克隆仓库：

    1
    2
    git clone https://github.com/Mellanox/network-operator.git
    cd network-operator

  2. 安装依赖：
    确保你的系统安装了Go编译器和Docker。

  3. 编译操作符：
    在项目根目录下运行：

    1
    make build

  4. 构建Docker镜像：

    1
    make docker-build

  5. 推送Docker镜像：
    将镜像推送到你的Docker仓库（例如Docker Hub）：

    1
    2
    docker tag network-operator:latest <your-repo>/network-operator:latest
    docker push <your-repo>/network-operator:latest

  6. 部署操作符：
    编辑部署文件，将镜像路径更新为你的仓库路径，然后应用这些文件：

    1
    kubectl apply -f deploy/

通过这些步骤，你可以从源码编译并部署NVIDIA Network Operator。如果需要详细信息，可以参考项目中的 README.md 文件或其他文档。

a-c-dream

a-c-dream

贵在坚持

109
13
0
CATEGORYS
TAGS