基于kuebrnetes v1.17
简单来说，一个完整的streaming请求如下：
kube-apiserver 向 Kubelet 发送流式请求 /exec/
Kubelet 通过 CRI 接口向 CRI Shim 请求 Exec 的 URL
CRI Shim 向 Kubelet 返回 Exec URL
Kubelet 向 kube-apiserver 返回重定向的响应
kube-apiserver 重定向流式请求到 Exec URL，接着就是 CRI Shim 内部的 Streaming Server 跟 kube-apiserver 进行数据交互，完成 Exec 的请求和响应
在 v1.10 及更早版本中，容器运行时必需返回一个 API Server 可直接访问的 URL（通常跟 Kubelet 使用相同的监听地址）；而从 v1.11 开始，Kubelet 新增了 --redirect-container-streaming（默认为 false），默认不再转发而是代理 Streaming 请求，这样运行时可以返回一个 localhost 的 URL。通过 Kubelet 代理的好处是由 Kubelet 处理与 API server 通信之间的请求认证。
可以看到kubelet在处理exec请求的时候分为两步， 1. 首先获取streaming URL，2. 随后根据该URL建立流式请求。
获取streaming URLkubelet在启动时会初始化一个serve，注册好对应的handler， exec的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来进行处理：
// getExec handles requests to run a command inside a container. func (s *Server) getExec(request *restful.Request, response *restful.Response) { params := getExecRequestParams(request) streamOpts, err := remotecommandserver.NewOptions(request.Request) if err != nil { utilruntime.HandleError(err) response.WriteError(http.StatusBadRequest, err) return } pod, ok := s.host.GetPodByName(params.podNamespace, params.podName) if !ok { response.WriteError(http.StatusNotFound, fmt.Errorf("pod does not exist")) return } 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 } if s.redirectContainerStreaming { http.Redirect(response.ResponseWriter, request.Request, url.String(), http.StatusFound) return } proxyStream(response.ResponseWriter, request.Request, url) }以上代码可以看出，首先调用host.GetExec获取URL，然后判断是否开启重定向，如果开启则进行重定向，否则直接代理请求到该streaming URL。这里host对象对应的实现其实就是kubelet， 我们看下GetExec的实现：
// GetExec gets the URL the exec will be served from, or nil if the Kubelet will serve it. func (kl *Kubelet) GetExec(podFullName string, podUID types.UID, containerName string, cmd []string, streamOpts remotecommandserver.Options) (*url.URL, error) { container, err := kl.findContainer(podFullName, podUID, containerName) if err != nil { return nil, err } if container == nil { return nil, fmt.Errorf("container not found (%q)", containerName) } return kl.streamingRuntime.GetExec(container.ID, cmd, streamOpts.Stdin, streamOpts.Stdout, streamOpts.Stderr, streamOpts.TTY) }这里只是继续调用streamingRuntime的GetExec方法，streamingRuntime是个interface，具体的实现是kubeGenericRuntimeManager
// 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) { req := &runtimeapi.ExecRequest{ ContainerId: id.ID, Cmd: cmd, Tty: tty, Stdin: stdin, Stdout: stdout, Stderr: stderr, } resp, err := m.runtimeService.Exec(req) if err != nil { return nil, err } return url.Parse(resp.Url) }继而调用了runtimeService.Exec方法， 此处runtimeService根据CRI创建的remoteRuntimeService，简单来说就是对应CRI server的client端
// 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 }