代码拉取完成，页面将自动刷新
同步操作将从 src-openEuler/kubernetes 强制同步，此操作会覆盖自 Fork 仓库以来所做的任何修改，且无法恢复！！！
确定后同步将在后台操作，完成时将刷新页面，请耐心等待。
From 5ad72e2b135afb5fbe5112f901bdec79f8943611 Mon Sep 17 00:00:00 2001
From: liuxu <liuxu156@huawei.com>
Date: Fri, 26 Apr 2024 10:49:14 +0800
Subject: [PATCH] kubelet support exec and attach websocket protocol
Signed-off-by: liuxu <liuxu156@huawei.com>
---
pkg/kubelet/server/server.go | 43 +++-
.../pkg/cri/streaming/remotecommand/proxy.go | 212 ++++++++++++++++++
2 files changed, 247 insertions(+), 8 deletions(-)
create mode 100644 staging/src/k8s.io/kubelet/pkg/cri/streaming/remotecommand/proxy.go
diff --git a/pkg/kubelet/server/server.go b/pkg/kubelet/server/server.go
index 87a017f9..887e663a 100644
--- a/pkg/kubelet/server/server.go
+++ b/pkg/kubelet/server/server.go
@@ -834,51 +834,78 @@ func proxyStream(w http.ResponseWriter, r *http.Request, url *url.URL) {
// getAttach handles requests to attach to a container.
func (s *Server) getAttach(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
}
+
+ url, err := s.getAttachUrl(request, response, streamOpts)
+ if err != nil {
+ klog.Errorf("failed to get backend url %v", err)
+ return
+ }
+ if url.Scheme == "ws" || url.Scheme == "wss" {
+ remotecommandserver.ProxyToWebSocket(response.ResponseWriter, request.Request, url, streamOpts)
+ } else {
+ proxyStream(response.ResponseWriter, request.Request, url)
+ }
+}
+
+func (s *Server) getAttachUrl(request *restful.Request, response *restful.Response, streamOpts *remotecommandserver.Options) (*url.URL, error) {
+ params := getExecRequestParams(request)
pod, ok := s.host.GetPodByName(params.podNamespace, params.podName)
if !ok {
response.WriteError(http.StatusNotFound, fmt.Errorf("pod does not exist"))
- return
+ return nil, fmt.Errorf("pod not found")
}
podFullName := kubecontainer.GetPodFullName(pod)
url, err := s.host.GetAttach(request.Request.Context(), podFullName, params.podUID, params.containerName, *streamOpts)
if err != nil {
streaming.WriteError(err, response.ResponseWriter)
- return
+ return nil, err
}
- proxyStream(response.ResponseWriter, request.Request, url)
+ return url, nil
}
// 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
}
+ url, err := s.getExecUrl(request, response, streamOpts)
+ if err != nil {
+ klog.Errorf("failed to get backend url %v", err)
+ return
+ }
+ if url.Scheme == "ws" || url.Scheme == "wss" {
+ remotecommandserver.ProxyToWebSocket(response.ResponseWriter, request.Request, url, streamOpts)
+ } else {
+ proxyStream(response.ResponseWriter, request.Request, url)
+ }
+}
+
+func (s *Server) getExecUrl(request *restful.Request, response *restful.Response, streamOpts *remotecommandserver.Options) (*url.URL, error) {
+ params := getExecRequestParams(request)
pod, ok := s.host.GetPodByName(params.podNamespace, params.podName)
if !ok {
response.WriteError(http.StatusNotFound, fmt.Errorf("pod does not exist"))
- return
+ return nil, fmt.Errorf("pod not found")
}
podFullName := kubecontainer.GetPodFullName(pod)
url, err := s.host.GetExec(request.Request.Context(), podFullName, params.podUID, params.containerName, params.cmd, *streamOpts)
if err != nil {
streaming.WriteError(err, response.ResponseWriter)
- return
+ return nil, err
}
- proxyStream(response.ResponseWriter, request.Request, url)
+ return url, nil
}
// getRun handles requests to run a command inside a container.
diff --git a/staging/src/k8s.io/kubelet/pkg/cri/streaming/remotecommand/proxy.go b/staging/src/k8s.io/kubelet/pkg/cri/streaming/remotecommand/proxy.go
new file mode 100644
index 00000000..179d8183
--- /dev/null
+++ b/staging/src/k8s.io/kubelet/pkg/cri/streaming/remotecommand/proxy.go
@@ -0,0 +1,212 @@
+package remotecommand
+
+import (
+ "bytes"
+ "errors"
+ "fmt"
+ "io"
+ "net/http"
+ "net/url"
+ "strings"
+ "time"
+
+ "github.com/gorilla/websocket"
+ apierrors "k8s.io/apimachinery/pkg/api/errors"
+ "k8s.io/apimachinery/pkg/apis/meta/v1"
+ remotecommandconsts "k8s.io/apimachinery/pkg/util/remotecommand"
+ "k8s.io/apimachinery/pkg/util/runtime"
+ "k8s.io/client-go/util/exec"
+ "k8s.io/klog/v2"
+)
+
+var (
+ streamIdleTimeout = 4 * time.Hour
+ streamCreationTimeout = remotecommandconsts.DefaultStreamCreationTimeout
+)
+
+// proxyStreamToWebSocket proxies stream to url with websocket.
+func ProxyToWebSocket(w http.ResponseWriter, r *http.Request, url *url.URL, opts *Options) {
+ klog.V(8).Infof("start proxy request to websocket %+v", r)
+ ctx, ok := createStreams(
+ r,
+ w,
+ opts,
+ remotecommandconsts.SupportedStreamingProtocols,
+ streamIdleTimeout,
+ streamCreationTimeout)
+ if !ok {
+ msg := "failed to create stream to fontend"
+ klog.Error(msg)
+ http.Error(w, msg, http.StatusInternalServerError)
+ return
+ }
+ defer func() {
+ if err := ctx.conn.Close(); err != nil {
+ klog.Errorf("failed to close connection, %v", err)
+ }
+ }()
+
+ klog.V(8).Infof("start connecting to websocket %s", url.String())
+ backendConn, err := connectBackend(url.String(), "channel.k8s.io", r)
+ if err != nil {
+ msg := fmt.Sprintf("connectBackend failed: %v", err)
+ klog.Error(msg)
+ http.Error(w, msg, http.StatusInternalServerError)
+ return
+ }
+ defer backendConn.Close()
+
+ var errConnection error
+ frontendStdinToBackendComplete := make(chan struct{})
+ frontendResizeToBackendComplete := make(chan struct{})
+ backendToFrontendComplete := make(chan struct{})
+
+ go func() {
+ for {
+ _, msg, err := backendConn.ReadMessage()
+ if err != nil {
+ e, ok := err.(*websocket.CloseError)
+ if !ok || e.Code != websocket.CloseNormalClosure {
+ errConnection = err
+ }
+ break
+ }
+
+ if len(msg) < 1 {
+ errConnection = fmt.Errorf("received err msg from backEnd (the length less than 1), msg: %s", string(msg))
+ break
+ }
+
+ switch msg[0] {
+ case stdoutChannel:
+ _, err = ctx.stdoutStream.Write(msg[1:])
+ case stderrChannel:
+ _, err = ctx.stderrStream.Write(msg[1:])
+ case errorChannel:
+ err = ctx.writeStatus(apierrors.NewInternalError(errors.New(string(msg[1:]))))
+ default:
+ err = fmt.Errorf("received invalid msg from backEnd, msg: %s", string(msg))
+ }
+
+ if err != nil {
+ errConnection = err
+ break
+ }
+ }
+ close(backendToFrontendComplete)
+ }()
+
+ if opts.Stdin {
+ go func() {
+ r := &rwc{
+ c: backendConn,
+ index: stdinChannel,
+ }
+ _, err := io.Copy(r, ctx.stdinStream)
+ if err != nil && !strings.Contains(err.Error(), "use of closed network connection") {
+ errConnection = fmt.Errorf("copy data from frontend(stdinStream) to backend failed, err: %v", err)
+ }
+ close(frontendStdinToBackendComplete)
+ }()
+ }
+
+ if opts.TTY {
+ go func() {
+ r := &rwc{
+ c: backendConn,
+ index: resizeChannel,
+ }
+ _, err := io.Copy(r, ctx.resizeStream)
+ if err != nil && !strings.Contains(err.Error(), "use of closed network connection") {
+ errConnection = fmt.Errorf("copy data from frontend(resizeStream) to backend failed, err: %v", err)
+ }
+ close(frontendResizeToBackendComplete)
+ }()
+ }
+ select {
+ case <-backendToFrontendComplete:
+ case <-frontendStdinToBackendComplete:
+ case <-frontendResizeToBackendComplete:
+ }
+
+ select {
+ case <-backendToFrontendComplete:
+ case <-time.Tick(30 * time.Second):
+ klog.Errorf("Wait backend to frontend complete timeout")
+ }
+
+ if errConnection != nil {
+ klog.Errorf("SpdyProxy: the connection disconnected: %v", errConnection)
+ if exitErr, ok := errConnection.(exec.ExitError); ok && exitErr.Exited() {
+ rc := exitErr.ExitStatus()
+ ctx.writeStatus(&apierrors.StatusError{ErrStatus: v1.Status{
+ Status: v1.StatusFailure,
+ Reason: remotecommandconsts.NonZeroExitCodeReason,
+ Details: &v1.StatusDetails{
+ Causes: []v1.StatusCause{
+ {
+ Type: remotecommandconsts.ExitCodeCauseType,
+ Message: fmt.Sprintf("%d", rc),
+ },
+ },
+ },
+ Message: fmt.Sprintf("command terminated with non-zero exit code: %v", exitErr),
+ }})
+ } else if closeErr, ok := errConnection.(*websocket.CloseError); !ok || closeErr.Text != io.ErrUnexpectedEOF.Error() {
+ //ignore this ErrUnexpectedEOF because isulad always close the connection while reading a frame
+ err = fmt.Errorf("error executing command in container: %v", errConnection)
+ runtime.HandleError(err)
+ ctx.writeStatus(apierrors.NewInternalError(err))
+ }
+ } else {
+ ctx.writeStatus(&apierrors.StatusError{ErrStatus: v1.Status{
+ Status: v1.StatusSuccess,
+ }})
+ }
+}
+
+func connectBackend(addr, subprotocol string, r *http.Request) (*websocket.Conn, error) {
+ h := http.Header{}
+ originHeadValue := r.Header.Get("Origin")
+ if originHeadValue != "" {
+ h["Origin"] = []string{originHeadValue}
+ }
+ websocket.DefaultDialer.Subprotocols = []string{subprotocol}
+ websocket.DefaultDialer.ReadBufferSize = 128 * 1024
+ websocket.DefaultDialer.WriteBufferSize = 128 * 1024
+ ws, resp, err := websocket.DefaultDialer.Dial(addr, h)
+ if err == nil {
+ return ws, nil
+ }
+ msg := fmt.Errorf("dial failed: %v, response Body is nil", err)
+ if resp != nil && resp.Body != nil {
+ defer func() {
+ //websocket buffer size maybe not enough and cause panic
+ if e := recover(); e != nil {
+ msg = fmt.Errorf("dial failed: %v, response panic %v", err, e)
+ }
+ resp.Body.Close()
+ }()
+ var body bytes.Buffer
+ body.ReadFrom(resp.Body)
+ msg = fmt.Errorf("dial failed: %v, response is: %v", err, body.String())
+ }
+ return nil, msg
+}
+
+type rwc struct {
+ c *websocket.Conn
+ index byte
+}
+
+func (c *rwc) Write(p []byte) (int, error) {
+ frame := make([]byte, len(p)+1)
+ frame[0] = byte(c.index)
+ copy(frame[1:], p)
+
+ err := c.c.WriteMessage(websocket.BinaryMessage, frame)
+ if err != nil {
+ return 0, err
+ }
+ return len(p), nil
+}
--
2.34.1
此处可能存在不合适展示的内容，页面不予展示。您可通过相关编辑功能自查并修改。
如您确认内容无涉及 不当用语 / 纯广告导流 / 暴力 / 低俗色情 / 侵权 / 盗版 / 虚假 / 无价值内容或违法国家有关法律法规的内容，可点击提交进行申诉，我们将尽快为您处理。
马建仓 AI 助手