NVIDIA Device Plugin

a-c-dream 发布于

下面分析NVIDIA Device Plugin的源码

Overview

NVIDIA提供的这个Device Plugin当然是服务于GPU的，它可能是社区中最经典的Device Plugin了，所以下面的分析就以它为例。

它以DaemonSet的形式在K8s集群上运行：

  • 暴露各Nodes的GPUs数目；
  • 跟踪GPUs的健康情况；
  • 帮助运行启用了GPU的容器。

目录结构

1
2
3
4
5
6
7
8
9
api
cmd *
demo
deployments
docker
hack
internal
tests  测试
vendor  依赖的外部包
1
2
3
4
5
6
7
8
9
api：此目录通常包含与项目相关的HTTP API定义。它可能包含路由、端点、请求和响应结构等。
cmd：这是命令行工具或应用的代码所在位置。在Go中，main包通常位于cmd目录下。
demo：这是一个示例或演示目录，通常包含演示如何使用该项目的代码。
deployments：与部署相关的脚本、配置和工具可能位于此目录中。
docker：如果您使用Docker来运行或部署您的项目，那么此目录可能包含Dockerfile、docker-compose文件等。
hack：这是一个Hackathon目录，用于存放一些在Hackathon中使用的代码或文件。
internal：这是一个内部包目录，通常包含项目内部使用的私有包或模块。这些包不能从其他项目导入。
test：这是测试代码的目录。它可能包含单元测试、集成测试等。
vendor：在较早的Go版本中，依赖项被下载到vendor目录中。但从Go 1.11开始，Go Modules成为官方依赖管理工具，因此vendor目录的使用已经减少。

主要的代码在cmd文件夹下，下面来看cmd/nvidia-device-plugin中的代码

main.go

1
c.Flags = []cli.Flag{

在main函数中主要基于cli实现了命令行功能

定义了一些flag

名称 类型 作用
mig-strategy StringFlag 在支持 MIG 设备的 GPU 上公开 MIG 设备所需的策略：[none |single|mixed]
fail-on-init-error BoolFlag 如果在初始化过程中遇到错误，则插件失败，否则无限期阻止
nvidia-driver-root StringFlag NVIDIA 驱动程序安装的根路径（典型值为“/”或“/run/nvidia/driver”）
pass-device-specs BoolFlag 将 DeviceSpecs 列表传递给 Allocate（） 上的 kubelet
device-list-strategy StringSliceFlag 将设备列表传递到底层运行时所需的策略：[envvar |volume-mounts|cdi-annotations]
device-id-strategy StringFlag 将设备 ID 传递到底层运行时所需的策略：[uuid|index]
gds-enabled BoolFlag 确保容器以 NVIDIA_GDS=enabled 启动
mofed-enabled BoolFlag 确保容器以 NVIDIA_MOFED=enabled 启动
cdi-annotation-prefix StringFlag 用于 CDI 容器注释键的前缀
nvidia-ctk-path StringFlag 在生成的 CDI 规范中用于 nvidia-ctk 的路径
container-driver-root StringFlag NVIDIA 驱动程序根目录在容器中挂载的路径;用于生成 CDI 规范 
1
func start(c *cli.Context, flags []cli.Flag) error {

在这个函数中首先使用fsnotify来监控文件系统变化，监控的目录为/var/lib/kubelet/device-plugins/

调用的newFSWatcher在watchers.go中

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
func newFSWatcher(files ...string) (*fsnotify.Watcher, error) {
	watcher, err := fsnotify.NewWatcher()
	if err != nil {
		return nil, err
	}

	for _, f := range files {
		err = watcher.Add(f)
		if err != nil {
			watcher.Close()
			return nil, err
		}
	}

	return watcher, nil
}

这里是基于fsnotify封装的一个watcher。他可以监控files参数中的所有目录

通过sigs := newOSWatcher(syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)监听是否有syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT信号

newOSWatcher的定义也在watchers.go中

1
2
3
4
5
6
func newOSWatcher(sigs ...os.Signal) chan os.Signal {
	sigChan := make(chan os.Signal, 1)
	signal.Notify(sigChan, sigs...)

	return sigChan
}

具体实现流程如下

1.创建一个新的通道sigChan，它可以接收一个信号。
2.使用signal.Notify函数，将指定的信号（sigs）发送到sigChan通道。这意味着当操作系统接收到这些信号之一时，sigChan通道会接收到这个信号。
3.最后，函数返回sigChan通道，这样调用者就可以监听这些信号了。

restart：

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
// If we are restarting, stop plugins from previous run.
	if restarting {
		err := stopPlugins(plugins)
		if err != nil {
			return fmt.Errorf("error stopping plugins from previous run: %v", err)
		}
	}

	klog.Info("Starting Plugins.")
	plugins, restartPlugins, err := startPlugins(c, flags, restarting)
	if err != nil {
		return fmt.Errorf("error starting plugins: %v", err)
	}

	//如果启动失败，30s后再次启动
	if restartPlugins {
		klog.Infof("Failed to start one or more plugins. Retrying in 30s...")
		restartTimeout = time.After(30 * time.Second)
	}

	restarting = true

	// Start an infinite loop, waiting for several indicators to either log
	// some messages, trigger a restart of the plugins, or exit the program.
	//无限循环，监听文件系统和信号
	for {
		select {
		// If the restart timeout has expired, then restart the plugins
		case <-restartTimeout:
			goto restart

		// Detect a kubelet restart by watching for a newly created
		// 'pluginapi.KubeletSocket' file. When this occurs, restart this loop,
		// restarting all of the plugins in the process.
		// 监听/var/lib/kubelet/device-plugins/kubelet.sock，如果创建重启
		case event := <-watcher.Events:
			if event.Name == pluginapi.KubeletSocket && event.Op&fsnotify.Create == fsnotify.Create {
				klog.Infof("inotify: %s created, restarting.", pluginapi.KubeletSocket)
				goto restart
			}

		// Watch for any other fs errors and log them.
		case err := <-watcher.Errors:
			klog.Infof("inotify: %s", err)

		// Watch for any signals from the OS. On SIGHUP, restart this loop,
		// restarting all of the plugins in the process. On all other
		// signals, exit the loop and exit the program.
		case s := <-sigs:
			switch s {
			case syscall.SIGHUP:
				klog.Info("Received SIGHUP, restarting.")
				goto restart
			default:
				klog.Infof("Received signal \"%v\", shutting down.", s)
				goto exit
			}
		}
	}

exit:

exit比较简单，直接调用stopPlugins停止所有插件

startPlugins

下面来看startPlugins函数，函数原型为func startPlugins(c *cli.Context, flags []cli.Flag, restarting bool) ([]plugin.Interface, bool, error)

首先，加载配置文件。然后使用默认的资源更新配置信息。以JSON的形式输出配置信息。最后启动所有插件。

stopPlugins

stopPlugins的实现比较简单，遍历所有plugins，然后依次停止即可。

1
2
3
4
5
6
7
8
9
10
11
12
 这段代码是一个实现了NVIDIA设备插件的Go语言程序。它实现了一个用于在Kubernetes中管理NVIDIA GPU设备的插件，通过对设备的动态分配和管理，实现对GPU资源的有效利用。

具体实现了以下功能：
1. 加载配置文件：读取命令行参数或环境变量，生成有效的配置参数。
2. 监控文件系统变化：使用fsnotify包监控文件系统的变化，包括kubelet的socket文件和其他文件的创建、删除等操作。
3. 启动和停止插件：根据配置和设备列表启动NVIDIA设备插件，并在需要时重新启动插件。
4. 提供gRPC服务：针对每个插件，启动gRPC服务，用于与Kubelet通信，包括设备的注册和注销等操作。
5. 设备资源管理：根据配置信息，管理和调度可用的设备资源，确保资源的正确分配和利用。

除此之外，程序还包括了对配置文件的加载、校验和打印，以及对资源重命名功能的临时禁用等。

总体来说，这段代码实现了一个完整的NVIDIA设备插件程序，能够有效地管理和分配集群中的GPU设备资源。

参考链接：

https://blog.nowcoder.net/n/2be3d64786a54c40901ba885babff860

https://www.nowcoder.com/discuss/353147783991468032

a-c-dream

a-c-dream

贵在坚持

109
13
0
CATEGORYS
TAGS