Kubectl exec 的作業原理解讀-有解無憂

對于經常和 Kubernetes 打交道的 YAML 工程師來說，最常用的命令就是 kubectl exec 了，通過它可以直接在容器內執行命令來除錯應用程式，如果你不滿足于只是用用而已，想了解 kubectl exec 的作業原理，那么本文值得你仔細讀一讀，本文將通過參考 kubectl、API Server、Kubelet 和容器運行時介面（CRI）Docker API 中的相關代碼來了解該命令是如何作業的，

kubectl exec 的作業原理用一張圖就可以表示：

kubectl exec

先來看一個例子：

?? → kubectl version --short 
Client Version: v1.15.0 
Server Version: v1.15.3

?? → kubectl run nginx --image=nginx --port=80 --generator=run-pod/v1
pod/nginx created

?? → kubectl get po     
NAME    READY   STATUS    RESTARTS   AGE 
nginx   1/1     Running   0          6s  

?? → kubectl exec nginx -- date
Sat Jan 25 18:47:52 UTC 2020

?? → kubectl exec -it nginx -- /bin/bash 
root@nginx:/#

第一個 kubectl exec 在容器內執行了 date 命令，第二個 kubectl exec 使用 -i 和 -t 引數進入了容器的互動式 shell，

重復第二個 kubectl exec 命令，列印更詳細的日志：

?? → kubectl -v=7 exec -it nginx -- /bin/bash                                                         
I0125 10:51:55.434043   28053 loader.go:359] Config loaded from file:  /home/isim/.kube/kind-config-linkerd
I0125 10:51:55.438595   28053 round_trippers.go:416] GET https://127.0.0.1:38545/api/v1/namespaces/default/pods/nginx
I0125 10:51:55.438607   28053 round_trippers.go:423] Request Headers:
I0125 10:51:55.438611   28053 round_trippers.go:426]     Accept: application/json, */*
I0125 10:51:55.438615   28053 round_trippers.go:426]     User-Agent: kubectl/v1.15.0 (linux/amd64) kubernetes/e8462b5
I0125 10:51:55.445942   28053 round_trippers.go:441] Response Status: 200 OK in 7 milliseconds
I0125 10:51:55.451050   28053 round_trippers.go:416] POST https://127.0.0.1:38545/api/v1/namespaces/default/pods/nginx/exec?command=%2Fbin%2Fbash&container=nginx&stdin=true&stdout=true&tty=true
I0125 10:51:55.451063   28053 round_trippers.go:423] Request Headers:
I0125 10:51:55.451067   28053 round_trippers.go:426]     X-Stream-Protocol-Version: v4.channel.k8s.io
I0125 10:51:55.451090   28053 round_trippers.go:426]     X-Stream-Protocol-Version: v3.channel.k8s.io
I0125 10:51:55.451096   28053 round_trippers.go:426]     X-Stream-Protocol-Version: v2.channel.k8s.io
I0125 10:51:55.451100   28053 round_trippers.go:426]     X-Stream-Protocol-Version: channel.k8s.ioI0125 10:51:55.451121   28053 round_trippers.go:426]     User-Agent: kubectl/v1.15.0 (linux/amd64) kubernetes/e8462b5
I0125 10:51:55.465690   28053 round_trippers.go:441] Response Status: 101 Switching Protocols in 14 milliseconds
root@nginx:/#

這里有兩個重要的 HTTP 請求：

GET 請求用來獲取 Pod 資訊，
POST 請求呼叫 Pod 的子資源 exec 在容器內執行命令，

子資源（subresource）隸屬于某個 K8S 資源，表示為父資源下方的子路徑，例如 /logs、/status、/scale、/exec 等，其中每個子資源支持的操作根據物件的不同而改變，

最后 API Server 回傳了 101 Ugrade 回應，向客戶端表示已切換到 SPDY 協議，

SPDY 允許在單個 TCP 連接上復用獨立的 stdin/stdout/stderr/spdy-error 流，

1. API Server 原始碼分析

請求首先會到底 API Server，先來看看 API Server 是如何注冊 rest.ExecRest 處理器來處理子資源請求 /exec 的，這個處理器用來確定 exec 要進入的節點，

API Server 啟動程序中做的第一件事就是指揮內嵌的 GenericAPIServer 加載早期的遺留 API（legacy API）：

if c.ExtraConfig.APIResourceConfigSource.VersionEnabled(apiv1.SchemeGroupVersion) {
	// ...
	if err := m.InstallLegacyAPI(&c, c.GenericConfig.RESTOptionsGetter, legacyRESTStorageProvider); err != nil {
		return nil, err
	}
}

在 API 加載程序中，會將型別 LegacyRESTStorage 實體化，創建一個 storage.PodStorage 實體：

podStorage, err := podstore.NewStorage(
	restOptionsGetter,
	nodeStorage.KubeletConnectionInfo,
	c.ProxyTransport,
	podDisruptionClient,
)
if err != nil {
	return LegacyRESTStorage{}, genericapiserver.APIGroupInfo{}, err
}

隨后 storeage.PodStorage 實體會被添加到 map restStorageMap 中，注意，該 map 將路徑 pods/exec 映射到了 podStorage 的 rest.ExecRest 處理器，

restStorageMap := map[string]rest.Storage{
	"pods":             podStorage.Pod,
	"pods/attach":      podStorage.Attach,
	"pods/status":      podStorage.Status,
	"pods/log":         podStorage.Log,
	"pods/exec":        podStorage.Exec,
	"pods/portforward": podStorage.PortForward,
	"pods/proxy":       podStorage.Proxy,
	"pods/binding":     podStorage.Binding,
	"bindings":         podStorage.LegacyBinding,

podstorage 為 pod 和子資源提供了 CURD 邏輯和策略的抽象，更多詳細資訊請查看內嵌的 genericregistry.Store

map restStorageMap 會成為實體 apiGroupInfo 的一部分，添加到 GenericAPIServer 中：

if err := s.installAPIResources(apiPrefix, apiGroupInfo, openAPIModels); err != nil {
	return err
}

// Install the version handler.
// Add a handler at /<apiPrefix> to enumerate the supported api versions.
s.Handler.GoRestfulContainer.Add(discovery.NewLegacyRootAPIHandler(s.discoveryAddresses, s.Serializer, apiPrefix).WebService())

其中 GoRestfulContainer.ServeMux 會將傳入的請求 URL 映射到不同的處理器，

接下來重點觀察處理器 therest.ExecRest 的作業原理，它的 Connect() 方法會呼叫函式 pod.ExecLocation() 來確定 pod 中容器的 exec 子資源的 URL：

// Connect returns a handler for the pod exec proxy
func (r *ExecREST) Connect(ctx context.Context, name string, opts runtime.Object, responder rest.Responder) (http.Handler, error) {
	execOpts, ok := opts.(*api.PodExecOptions)
	if !ok {
		return nil, fmt.Errorf("invalid options object: %#v", opts)
	}
	location, transport, err := pod.ExecLocation(r.Store, r.KubeletConn, ctx, name, execOpts)
	if err != nil {
		return nil, err
	}
	return newThrottledUpgradeAwareProxyHandler(location, transport, false, true, true, responder), nil
}

函式 pod.ExecLocation() 回傳的 URL 被 API Server 用來決定連接到哪個節點，

下面接著分析節點上的 Kubelet 原始碼，

2. Kubelet 原始碼分析

到了 Kubelet 這邊，我們需要關心兩點：

Kubelet 是如何注冊 exec 處理器的？
Kubelet 與 Docker API 如何互動？

Kubelet 的初始化程序非常復雜，主要涉及到兩個函式：

PreInitRuntimeService() : 使用 dockershim 包來初始化 CRI，
RunKubelet() : 注冊處理器，啟動 Kubelet 服務，

注冊處理器

當 Kubelet 啟動時，它的 RunKubelet() 函式會呼叫私有函式 startKubelet() 來啟動 kubelet.Kubelet 實體的 ListenAndServe() 方法，然后該方法會呼叫函式 ListenAndServeKubeletServer() ，使用建構式 NewServer() 來安裝『debugging』處理器：

// NewServer initializes and configures a kubelet.Server object to handle HTTP requests.
func NewServer(
	// ...
	criHandler http.Handler) Server {
	// ...
	if enableDebuggingHandlers {
		server.InstallDebuggingHandlers(criHandler)
		if enableContentionProfiling {
			goruntime.SetBlockProfileRate(1)
		}
	} else {
		server.InstallDebuggingDisabledHandlers()
	}
	return server
}

InstallDebuggingHandlers() 函式使用 getExec() 處理器來注冊 HTTP 請求模式：

// InstallDebuggingHandlers registers the HTTP request patterns that serve logs or run commands/containers
func (s *Server) InstallDebuggingHandlers(criHandler http.Handler) {
  // ...
  ws = new(restful.WebService)
	ws.
		Path("/exec")
	ws.Route(ws.GET("/{podNamespace}/{podID}/{containerName}").
		To(s.getExec).
		Operation("getExec"))
	ws.Route(ws.POST("/{podNamespace}/{podID}/{containerName}").
		To(s.getExec).
		Operation("getExec"))
	ws.Route(ws.GET("/{podNamespace}/{podID}/{uid}/{containerName}").
		To(s.getExec).
		Operation("getExec"))
	ws.Route(ws.POST("/{podNamespace}/{podID}/{uid}/{containerName}").
		To(s.getExec).
		Operation("getExec"))
	s.restfulCont.Add(ws)

其中 getExec() 處理器又會呼叫 s.host 實體中的 GetExec() 方法：

// getExec handles requests to run a command inside a container.
func (s *Server) getExec(request *restful.Request, response *restful.Response) {
  	// ...
	podFullName := kubecontainer.GetPodFullName(pod)
	url, err := s.host.GetExec(podFullName, params.podUID, params.containerName, params.cmd, *streamOpts)
	if err != nil {
		streaming.WriteError(err, response.ResponseWriter)
		return
	}
	// ...
}

s.host 被實體化為 kubelet.Kubelet 型別的一個實體，它嵌套參考了 StreamingRuntime 介面，該介面又被實體化為 kubeGenericRuntimeManager 的實體，即運行時管理器，該運行時管理器是 Kubelet 與 Docker API 互動的關鍵組件，GetExec() 方法就是由它實作的：

// GetExec gets the endpoint the runtime will serve the exec request from.
func (m *kubeGenericRuntimeManager) GetExec(id kubecontainer.ContainerID, cmd []string, stdin, stdout, stderr, tty bool) (*url.URL, error) {
	// ...
	resp, err := m.runtimeService.Exec(req)
	if err != nil {
		return nil, err
	}

	return url.Parse(resp.Url)
}

GetExec() 又會呼叫 runtimeService.Exec() 方法，進一步挖掘你會發現 runtimeService 是 CRI 包中定義的介面，kuberuntime.kubeGenericRuntimeManager 的 runtimeService 被實體化為 kuberuntime.instrumentedRuntimeService 型別，由它來實作 runtimeService.Exec() 方法：

func (in instrumentedRuntimeService) Exec(req *runtimeapi.ExecRequest) (*runtimeapi.ExecResponse, error) {
	const operation = "exec"
	defer recordOperation(operation, time.Now())

	resp, err := in.service.Exec(req)
	recordError(operation, err)
	return resp, err
}

instrumentedRuntimeService 實體的嵌套服務物件被實體化為 theremote.RemoteRuntimeService 型別的實體，該型別實作了 Exec() 方法：

// Exec prepares a streaming endpoint to execute a command in the container, and returns the address.
func (r *RemoteRuntimeService) Exec(req *runtimeapi.ExecRequest) (*runtimeapi.ExecResponse, error) {
	ctx, cancel := getContextWithTimeout(r.timeout)
	defer cancel()

	resp, err := r.runtimeClient.Exec(ctx, req)
	if err != nil {
		klog.Errorf("Exec %s '%s' from runtime service failed: %v", req.ContainerId, strings.Join(req.Cmd, " "), err)
		return nil, err
	}

	if resp.Url == "" {
		errorMessage := "URL is not set"
		klog.Errorf("Exec failed: %s", errorMessage)
		return nil, errors.New(errorMessage)
	}

	return resp, nil
}

Exec() 方法會向 /runtime.v1alpha2.RuntimeService/Exec 發起一個 gRPC 呼叫來讓運行時端準備一個流式通信的端點，該端點用于在容器中執行命令（關于如何將 Docker shim 設定為 gRPC 服務端的更多資訊請參考下一小節），

gRPC 服務端通過呼叫 RuntimeServiceServer.Exec() 方法來處理請求，該方法由 dockershim.dockerService 結構體實作：

// Exec prepares a streaming endpoint to execute a command in the container, and returns the address.
func (ds *dockerService) Exec(_ context.Context, req *runtimeapi.ExecRequest) (*runtimeapi.ExecResponse, error) {
	if ds.streamingServer == nil {
		return nil, streaming.NewErrorStreamingDisabled("exec")
	}
	_, err := checkContainerStatus(ds.client, req.ContainerId)
	if err != nil {
		return nil, err
	}
	return ds.streamingServer.GetExec(req)
}

第 10 行的 ThestreamingServer 是一個 streaming.Server 介面，它在建構式 dockershim.NewDockerService() 中被實體化：

// create streaming server if configured.
if streamingConfig != nil {
	var err error
	ds.streamingServer, err = streaming.NewServer(*streamingConfig, ds.streamingRuntime)
	if err != nil {
		return nil, err
	}
}

來看一下 GetExec() 方法的實作方式：

func (s *server) GetExec(req *runtimeapi.ExecRequest) (*runtimeapi.ExecResponse, error) {
	if err := validateExecRequest(req); err != nil {
		return nil, err
	}
	token, err := s.cache.Insert(req)
	if err != nil {
		return nil, err
	}
	return &runtimeapi.ExecResponse{
		Url: s.buildURL("exec", token),
	}, nil
}

可以看到這里只是向客戶端回傳一個簡單的 token 組合成的 URL，之所以生成一個 token 是因為用戶的命令中可能包含各種各樣的字符，各種長度的字符，需要格式化為一個簡單的 token，該 token 會快取在本地，后面真正的 exec 請求會攜帶這個 token，通過該 token 找到之前的具體請求，其中 restful.WebService 實體會將 pod exec 請求路由到這個端點：

// InstallDebuggingHandlers registers the HTTP request patterns that serve logs or run commands/containers
func (s *Server) InstallDebuggingHandlers(criHandler http.Handler) {
  // ...
  ws = new(restful.WebService)
	ws.
		Path("/exec")
	ws.Route(ws.GET("/{podNamespace}/{podID}/{containerName}").
		To(s.getExec).
		Operation("getExec"))
	ws.Route(ws.POST("/{podNamespace}/{podID}/{containerName}").
		To(s.getExec).
		Operation("getExec"))
	ws.Route(ws.GET("/{podNamespace}/{podID}/{uid}/{containerName}").
		To(s.getExec).
		Operation("getExec"))
	ws.Route(ws.POST("/{podNamespace}/{podID}/{uid}/{containerName}").
		To(s.getExec).
		Operation("getExec"))
	s.restfulCont.Add(ws)

創建 Docker shim

PreInitRuntimeService() 函式作為 gRPC 服務端，負責創建并啟動 Docker shim，在將dockershim.dockerService 型別實體化時，讓其嵌套的 streamingRuntime 實體參考 dockershim.NativeExecHandler 的實體（該實體實作了 dockershim.ExecHandler 介面），

ds := &dockerService{
	// ...
	streamingRuntime: &streamingRuntime{
		client:      client,
		execHandler: &NativeExecHandler{},
	},
	// ...
}

使用 Docker 的 exec API 在容器中執行命令的核心實作就是 NativeExecHandler.ExecInContainer() 方法：

func (*NativeExecHandler) ExecInContainer(client libdocker.Interface, container *dockertypes.ContainerJSON, cmd []string, stdin io.Reader, stdout, stderr io.WriteCloser, tty bool, resize <-chan remotecommand.TerminalSize, timeout time.Duration) error {
	// ...
	startOpts := dockertypes.ExecStartCheck{Detach: false, Tty: tty}
	streamOpts := libdocker.StreamOptions{
		InputStream:  stdin,
		OutputStream: stdout,
		ErrorStream:  stderr,
		RawTerminal:  tty,
		ExecStarted:  execStarted,
	}
	err = client.StartExec(execObj.ID, startOpts, streamOpts)
	if err != nil {
		return err
	}
	// ...

這里就是最終 Kubelet 呼叫 Docker exec API 的地方，

最后需要搞清楚的是 streamingServer 處理器如何處理 exec 請求，首先需要找到它的 exec 處理器，我們直接從建構式 streaming.NewServer() 開始往下找，因為這是將 /exec/{token} 路徑系結到 serveExec 處理器的地方：

ws := &restful.WebService{}
endpoints := []struct {
	path    string
	handler restful.RouteFunction
}{
	{"/exec/{token}", s.serveExec},
	{"/attach/{token}", s.serveAttach},
	{"/portforward/{token}", s.servePortForward},
}

所有發送到 dockershim.dockerService 實體的請求最終都會在 streamingServer 處理器上完成，因為 dockerService.ServeHTTP() 方法會呼叫 streamingServer 實體的 ServeHTTP() 方法，

serveExec 處理器會呼叫 remoteCommand.ServeExec() 函式，這個函式又是干嘛的呢？它會呼叫前面提到的 Executor.ExecInContainer() 方法，而 ExecInContainer() 方法是知道如何與 Docker exec API 通信的：

// ServeExec handles requests to execute a command in a container. After
// creating/receiving the required streams, it delegates the actual execution
// to the executor.
func ServeExec(w http.ResponseWriter, req *http.Request, executor Executor, podName string, uid types.UID, container string, cmd []string, streamOpts *Options, idleTimeout, streamCreationTimeout time.Duration, supportedProtocols []string) {
	// ...
	err := executor.ExecInContainer(podName, uid, container, cmd, ctx.stdinStream, ctx.stdoutStream, ctx.stderrStream, ctx.tty, ctx.resizeChan, 0)
	if err != nil {
	// ...
	} else {
	// ...	
	}
}

3. 總結

本文通過解讀 kubectl、API Server 和 CRI 的原始碼，幫助大家理解 kubectl exec 命令的作業原理，當然，這里并沒有涉及到 Docker exec API 的細節，也沒有涉及到 docker exec 的作業原理，

首先，kubectl 向 API Server 發出了 GET 和 POST 請求，API Server 回傳了 101 Ugrade 回應，向客戶端表示已切換到 SPDY 協議，

隨后 API Server 使用 storage.PodStorage 和 rest.ExecRest 來提供處理器的映射和執行邏輯，其中 rest.ExecRest 處理器決定 exec 要進入的節點，

最后 Kubelet 向 Docker shim 請求一個流式端點 URL，并將 exec 請求轉發到 Docker exec API，kubelet 再將這個 URL 以 Redirect 的方式回傳給 API Server，請求就會重定向到到對應 Streaming Server 上發起的 exec 請求，并維護長鏈，

雖然本文只關注了 kubectl exec 命令，但其他的子命令（例如 attach、port-forward、log 等等）也遵循了類似的實作模式：

kubectl

Kubernetes 1.18.2 1.17.5 1.16.9 1.15.12離線安裝包發布地址http://store.lameleg.com ，歡迎體驗，使用了最新的sealos v3.3.6版本，作了主機名決議配置優化，lvscare 掛載/lib/module解決開機啟動ipvs加載問題，修復lvscare社區netlink與3.10內核不兼容問題,sealos生成百年證書等特性，更多特性 https://github.com/fanux/sealos ，歡迎掃描下方的二維碼加入釘釘群，釘釘群已經集成sealos的機器人實時可以看到sealos的動態，

轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/10732.html

標籤：其他

上一篇：Kubernetes Ingress簡單入門

下一篇：分布式資料庫PolonDB 云端發力未來資料處理需求