node-driver-registry | 南风未起｜云原生/Go技术分享博客

# 1.简介

# 1.1.作用

node-driver-registrar也属于sidecar容器，用于将csi plugin注册到kubelet volumeMgr，以实现volume的mount/unmount。

--- 注册流程
	1.registrar基于unix.socket连接csi plugin,调用GetPluginInfo获取driver名称
	2.registrar向kubelet-registration-path目录创建registry socket,暴露GetInfo和NotifyRegistrationStatus接口
	3.kubelet watch到/var/lib/kubelet/plugins_registry目录的socket,调用GetInfo获取csi plugin地址及driver名称
	4.kubelet基于csi plugin地址调用NodeGetInfo,获取nodeID等信息
	5.kubelet基于csi plugin信息更新node对象的annotation、label及status.allocatable,创建csiNode对象
	6.kubelet调用NotifyRegistrationStatus接口通知registrar完成csi plugin注册

1
2
3
4
5
6
7

注意

node-driver-registrar与csi plugin nodeserver容器一起部署，必须将socket目录挂载至容器及开放CRUD权限

# 1.2.入口

main函数负责校验启动参数，连接csi plugin的socket地址，调用GetPluginInfo获取driver名称，调用nodeRegister触发注册。

func main() {
	...

    // 未配置kubelet注册路径
	if *kubeletRegistrationPath == "" {
		os.Exit(1)
	}
	...

	// 设置地址
	if *healthzPort > 0 {
		addr = ":" + strconv.Itoa(*healthzPort)
	} else {
		addr = *httpEndpoint
	}

    // 建立socket连接
	csiConn, err := connection.ConnectWithoutMetrics(*csiAddress)
	...

	// 获取driver名称
	csiDriverName, err := csirpc.GetDriverName(ctx, csiConn)
	...
    
    // 关闭连接
	defer closeGrpcConnection(*csiAddress, csiConn)

	// 激活注册
	nodeRegister(csiDriverName, addr)
}

// This function is deprecated, prefer using Connect with `nil` as the metricsManager.
func ConnectWithoutMetrics(address string, options ...Option) (*grpc.ClientConn, error) {
	// Prepend default options
	options = append([]Option{WithTimeout(time.Second * 30)}, options...)
	return connect(address, options)
}

// connect is the internal implementation of Connect. It has more options to enable testing.
func connect(...) (*grpc.ClientConn, error) {
	...
    
	for _, option := range connectOptions {
		option(&o)
	}

	dialOptions := []grpc.DialOption{
		grpc.WithInsecure(),                    // 普通协议
		grpc.WithBackoffMaxDelay(time.Second),  // 间隔1s重试
		grpc.WithBlock(),                       // 阻塞至连接成功.
		grpc.WithIdleTimeout(time.Duration(0)), // 空闲不断开
	}

    // 30s超时
	if o.timeout > 0 {
		dialOptions = append(dialOptions, grpc.WithTimeout(o.timeout))
	}
	...
    
    // 设置log/metric/tracing拦截器
	dialOptions = append(dialOptions, grpc.WithChainUnaryInterceptor(interceptors...))

    // 设置unix socket地址
	unixPrefix := "unix://"
	if strings.HasPrefix(address, "/") {
		// It looks like filesystem path.
		address = unixPrefix + address
	}

    // unix socket分支
	if strings.HasPrefix(address, unixPrefix) {
		...

		dialOptions = append(dialOptions, grpc.WithDialer(func(string, time.Duration) (net.Conn, error) {
			// 连接中断
            if haveConnected && !lostConnection {
				// 执行重连
				if o.reconnect != nil {
					reconnect = o.reconnect()
				}
				lostConnection = true
			}
            
            // 禁止重连
			if !reconnect {
				return nil, errors.New("connection lost, reconnecting disabled")
			}
            
            // 尝试重新连接
			conn, err := net.DialTimeout("unix", address[len(unixPrefix):], timeout)
			if err == nil {
				// Connection reestablished.
				haveConnected = true
				lostConnection = false
			}
            
			return conn, err
		}))
    // 重连仅支持unix socket
	} else if o.reconnect != nil {
		return nil, errors.New("OnConnectionLoss callback only supported for unix:// addresses")
	}
	...
    
	go func() {
        // 尝试基于网络连接
		conn, err = grpc.Dial(address, dialOptions...)
		close(ready)
	}()
	...

	for {
		select {
        // 10s触发一次日志打印
		case <-ticker.C:
			klog.Warningf("Still connecting to %s", address)
		// 连接成功
		case <-ready:
			return conn, err
		}
	}
}

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

注意

node-driver-registrar会建立csi连接及获取pluginInfo，完成后向/var/lib/kubelet/plugins_registry注册socker

# 2.注册

# 2.1.register

nodeRegister()负责初始化registrationServer，向var/lib/kubelet/plugins_registry注册socket供kubelet调用。

func nodeRegister(csiDriverName, httpEndpoint string) {
	registrar := newRegistrationServer(csiDriverName, *kubeletRegistrationPath, supportedVersions)
    // var/lib/kubelet/plugins_registry/driver-reg.sock
	socketPath := buildSocketPath(csiDriverName)
	
    // 清理旧的socket
    util.CleanupSocketFile(socketPath)
    ...

	// 临时调整socket权限
	if runtime.GOOS == "linux" {
		// Default to only user accessible socket, caller can open up later if desired
		oldmask, _ = util.Umask(0077)
	}

    // 创建socket(0077)
	lis, err := net.Listen("unix", socketPath)
	...
    
    // 恢复权限
	if runtime.GOOS == "linux" {
		util.Umask(oldmask)
	}
	
    // 创建grpc server
	grpcServer := grpc.NewServer()

	// 注册api
	registerapi.RegisterRegistrationServer(grpcServer, registrar)

    // 注册health api,检测socket exist+conn测试
	go httpServer(socketPath, httpEndpoint, csiDriverName)
    
    // 退出监听
	go removeRegSocket(csiDriverName)
    
	// start service
	grpcServer.Serve(lis)
    ...

	// If gRPC server is gracefully shutdown, cleanup and exit
	os.Exit(0)
}

func removeRegSocket(csiDriverName string) {
	...
	signal.Notify(sigc, syscall.SIGTERM)
	<-sigc
    
    // 移除socket
	socketPath := buildSocketPath(csiDriverName)
	os.Remove(socketPath)
	...
}

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

注意

nodeRegister主要负责创建注册目录下的unix socket，驱动kubelet感知csi plugin及建立连接

# 2.2.restapi

node-driver-registrar暴露GetInfo和NotifyRegistrationStatus接口，前者用于获取csi plugin调用信息，后者用于通知注册结果。

// GetInfo is the RPC invoked by plugin watcher
func (e registrationServer) GetInfo(ctx Context, req *registerapi.InfoRequest) (*registerapi.PluginInfo, error){
	return &registerapi.PluginInfo{
		Type:              "CSIPlugin",
		Name:              e.driverName,
		Endpoint:          e.endpoint,		// csi plugin调用地址
		SupportedVersions: e.version,
	}, nil
}

func (e registrationServer) NotifyRegistrationStatus(ctx Context, status *registerapi.RegistrationStatus) (...){
	// 注册失败
    if !status.PluginRegistered {
		os.Exit(1)	// 退出重启会重建注册socket
	}

	return &registerapi.RegistrationStatusResponse{}, nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

注意

node-driver-registrar接口返回的是固定信息，注册失败会退出重启，重建注册socket，触发再次向kubelet注册