您好,登錄后才能下訂單哦!
k8s目前主要支持CPU和內(nèi)存兩種資源,為了支持用戶需要按需分配的其他硬件類型的資源的調(diào)度分配,k8s實現(xiàn)了設備插件框架(device plugin framework)來用于其他硬件類型的資源集成,比如現(xiàn)在機器學習要使用GPU等資源,今天來看下其內(nèi)部的關鍵實現(xiàn)
當我們要集成本地硬件的資源的時候,我們可以在當前節(jié)點上通過DaemonSet來運行一個GRPC服務,通過這個服務來進行本地硬件資源的上報與分配
當提供硬件服務需要與kubelet進行通信的時候,則首先需要進行注冊,注冊的方式,則是通過最原始的底層的socket文件,并且通過Linux文件系統(tǒng)的inotify機制,來實現(xiàn)服務的注冊
Watcher主要是負責感知當前節(jié)點上注冊的服務,當發(fā)現(xiàn)新的要注冊的插件服務,則會產(chǎn)生對應的事件,注冊到當前的kubelet中
這里的狀態(tài)主要是指的是否需要注冊,因為kubelet與對應的插件服務是通過網(wǎng)絡進行通信的,當網(wǎng)絡出現(xiàn)問題、或者對應的插件服務故障,則可能會導致服務注冊失敗,但此時對應的服務的socket還依舊存在,即對應的插件服務依舊存在
此時就會有兩種狀態(tài):期望狀態(tài)與實際狀態(tài), 因為socket存在所以服務的期望狀態(tài)其實是需要注冊這個插件服務,但是實際上因為某些原因,這個插件服務并沒有完成注冊,后續(xù)會不斷的通過期望狀態(tài),調(diào)整實際狀態(tài),從而達到一致
協(xié)調(diào)器則就是完成上述兩種狀態(tài)之間操作的核心,其通過調(diào)用對應插件的回調(diào)函數(shù),其實就是調(diào)用對應的grpc接口,來完成期望狀態(tài)與實際狀態(tài)的一致性
針對每種類型的插件,都會有對應的控制器,其實也就是實現(xiàn)對應設備注冊和反注冊并且完成底層資源的分配(Allocate)和收集(ListWatch)操作
type Watcher struct {
// 感知插件服務注冊的socket的路徑
path string
fs utilfs.Filesystem
// inotify監(jiān)測插件服務socket變化
fsWatcher *fsnotify.Watcher
stopped chan struct{}
// 存儲期望狀態(tài)
desiredStateOfWorld cache.DesiredStateOfWorld
}
初始化其實就是創(chuàng)建對應的目錄
func (w *Watcher) init() error {
klog.V(4).Infof("Ensuring Plugin directory at %s ", w.path)
if err := w.fs.MkdirAll(w.path, 0755); err != nil {
return fmt.Errorf("error (re-)creating root %s: %v", w.path, err)
}
return nil
}
go func(fsWatcher *fsnotify.Watcher) {
defer close(w.stopped)
for {
select {
case event := <-fsWatcher.Events:
//如果發(fā)現(xiàn)對應目錄的文件的變化,則會觸發(fā)對應的事件
if event.Op&fsnotify.Create == fsnotify.Create {
err := w.handleCreateEvent(event)
if err != nil {
klog.Errorf("error %v when handling create event: %s", err, event)
}
} else if event.Op&fsnotify.Remove == fsnotify.Remove {
w.handleDeleteEvent(event)
}
continue
case err := <-fsWatcher.Errors:
if err != nil {
klog.Errorf("fsWatcher received error: %v", err)
}
continue
case <-stopCh:
// In case of plugin watcher being stopped by plugin manager, stop
// probing the creation/deletion of plugin sockets.
// Also give all pending go routines a chance to complete
select {
case <-w.stopped:
case <-time.After(11 * time.Second):
klog.Errorf("timeout on stopping watcher")
}
w.fsWatcher.Close()
return
}
}
}(fsWatcher)
其實補償機制主要是在重新啟動kubelet的時候,需要將之前已經(jīng)存在的socket重新注冊到當前的kubelet中
func (w *Watcher) traversePluginDir(dir string) error {
return w.fs.Walk(dir, func(path string, info os.FileInfo, err error) error {
if err != nil {
if path == dir {
return fmt.Errorf("error accessing path: %s error: %v", path, err)
}
klog.Errorf("error accessing path: %s error: %v", path, err)
return nil
}
switch mode := info.Mode(); {
case mode.IsDir():
if err := w.fsWatcher.Add(path); err != nil {
return fmt.Errorf("failed to watch %s, err: %v", path, err)
}
case mode&os.ModeSocket != 0:
event := fsnotify.Event{
Name: path,
Op: fsnotify.Create,
}
//TODO: Handle errors by taking corrective measures
if err := w.handleCreateEvent(event); err != nil {
klog.Errorf("error %v when handling create event: %s", err, event)
}
default:
klog.V(5).Infof("Ignoring file %s with mode %v", path, mode)
}
return nil
})
}
注冊其實就只需要感知到的socket文件路徑傳遞給期望狀態(tài)進行管理
func (w *Watcher) handlePluginRegistration(socketPath string) error {
if runtime.GOOS == "windows" {
socketPath = util.NormalizePath(socketPath)
}
// 調(diào)用期望狀態(tài)進行更新
klog.V(2).Infof("Adding socket path or updating timestamp %s to desired state cache", socketPath)
err := w.desiredStateOfWorld.AddOrUpdatePlugin(socketPath)
if err != nil {
return fmt.Errorf("error adding socket path %s or updating timestamp to desired state cache: %v", socketPath, err)
}
return nil
}
注冊其實就只需要感知到的socket文件路徑傳遞給期望狀態(tài)進行管理
func (w *Watcher) handleDeleteEvent(event fsnotify.Event) {
klog.V(6).Infof("Handling delete event: %v", event)
socketPath := event.Name
klog.V(2).Infof("Removing socket path %s from desired state cache", socketPath)
w.desiredStateOfWorld.RemovePlugin(socketPath)
}
插件信息其實只是存儲了對應socket的路徑和最近更新的時間
type PluginInfo struct {
SocketPath string
Timestamp time.Time
}
期望狀態(tài)與實際狀態(tài)在數(shù)據(jù)結(jié)構(gòu)上都是一樣的,因為本質(zhì)上只是為了存儲插件的當前的狀態(tài)信息,即更新時間,這里不在贅述
type desiredStateOfWorld struct {
socketFileToInfo map[string]PluginInfo
sync.RWMutex
}
type actualStateOfWorld struct {
socketFileToInfo map[string]PluginInfo
sync.RWMutex
}
目前k8s中支持兩大類的插件的管理一類是DevicePlugin即我們本文說的這些都是這種概念,一類是CSIPlugin,其中針對每一類DRiver的處理其實內(nèi)部都是不一樣的,那其實在操作之前就要先感知到當前的Driver是那種類型的
OperationExecutor主要就是做這件事的,其根據(jù)不同的plugin類型,生成不同的要執(zhí)行的操作,即對應的Plugin類型獲取對應的handler,就生成了一個要執(zhí)行的操作
registerPluginFunc := func() error {
client, conn, err := dial(socketPath, dialTimeoutDuration)
if err != nil {
return fmt.Errorf("RegisterPlugin error -- dial failed at socket %s, err: %v", socketPath, err)
}
defer conn.Close()
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
infoResp, err := client.GetInfo(ctx, ®isterapi.InfoRequest{})
if err != nil {
return fmt.Errorf("RegisterPlugin error -- failed to get plugin info using RPC GetInfo at socket %s, err: %v", socketPath, err)
}
handler, ok := pluginHandlers[infoResp.Type]
if !ok {
if err := og.notifyPlugin(client, false, fmt.Sprintf("RegisterPlugin error -- no handler registered for plugin type: %s at socket %s", infoResp.Type, socketPath)); err != nil {
return fmt.Errorf("RegisterPlugin error -- failed to send error at socket %s, err: %v", socketPath, err)
}
return fmt.Errorf("RegisterPlugin error -- no handler registered for plugin type: %s at socket %s", infoResp.Type, socketPath)
}
if infoResp.Endpoint == "" {
infoResp.Endpoint = socketPath
}
if err := handler.ValidatePlugin(infoResp.Name, infoResp.Endpoint, infoResp.SupportedVersions); err != nil {
if err = og.notifyPlugin(client, false, fmt.Sprintf("RegisterPlugin error -- plugin validation failed with err: %v", err)); err != nil {
return fmt.Errorf("RegisterPlugin error -- failed to send error at socket %s, err: %v", socketPath, err)
}
return fmt.Errorf("RegisterPlugin error -- pluginHandler.ValidatePluginFunc failed")
}
err = actualStateOfWorldUpdater.AddPlugin(cache.PluginInfo{
SocketPath: socketPath,
Timestamp: timestamp,
})
if err != nil {
klog.Errorf("RegisterPlugin error -- failed to add plugin at socket %s, err: %v", socketPath, err)
}
// 調(diào)用插件的注冊回調(diào)函數(shù)
if err := handler.RegisterPlugin(infoResp.Name, infoResp.Endpoint, infoResp.SupportedVersions); err != nil {
return og.notifyPlugin(client, false, fmt.Sprintf("RegisterPlugin error -- plugin registration failed with err: %v", err))
}
if err := og.notifyPlugin(client, true, ""); err != nil {
return fmt.Errorf("RegisterPlugin error -- failed to send registration status at socket %s, err: %v", socketPath, err)
}
func dial(unixSocketPath string, timeout time.Duration) (registerapi.RegistrationClient, *grpc.ClientConn, error) {
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
c, err := grpc.DialContext(ctx, unixSocketPath, grpc.WithInsecure(), grpc.WithBlock(),
grpc.WithContextDialer(func(ctx context.Context, addr string) (net.Conn, error) {
return (&net.Dialer{}).DialContext(ctx, "unix", addr)
}),
)
if err != nil {
return nil, nil, fmt.Errorf("failed to dial socket %s, err: %v", unixSocketPath, err)
}
return registerapi.NewRegistrationClient(c), c, nil
}
今天就先到這里,下一章會繼續(xù)介紹如何組合上述組件以及默認的回調(diào)管理機制的實現(xiàn),進探究到這里謝謝大家,感謝分享點贊,反轉(zhuǎn)又不花錢
k8s源碼閱讀電子書地址: https://www.yuque.com/baxiaoshi/tyado3
免責聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點不代表本網(wǎng)站立場,如果涉及侵權(quán)請聯(lián)系站長郵箱:is@yisu.com進行舉報,并提供相關證據(jù),一經(jīng)查實,將立刻刪除涉嫌侵權(quán)內(nèi)容。