基于client-go实现pod 交互式terminal

基于client-go实现pod 交互式terminal

后端实现逻辑（golang）

package main

import (
	"errors"
	"fmt"
	"github.com/gin-gonic/gin"
	"github.com/gorilla/websocket"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/kubernetes/scheme"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd"
	"k8s.io/client-go/tools/remotecommand"
	"log"
	"net/http"
	"strconv"
	"strings"
	"unicode/utf8"
)

//初始化k8s客户端
func initialClientSet(path string) (*kubernetes.Clientset, *rest.Config, error) {
	config, err := clientcmd.BuildConfigFromFlags("", path)
	if err != nil {
		log.Fatal(err)
	}

	ClientSet, err := kubernetes.NewForConfig(config)
	if err != nil {
		log.Fatal(err)
		return nil, nil, err
	}
	return ClientSet, config, err
}

func initialWS(c *gin.Context) (*websocket.Conn, error) {
	var upgrader = websocket.Upgrader{
		ReadBufferSize:  1024,
		WriteBufferSize: 1024,
		CheckOrigin: func(r *http.Request) bool {
			return true
		},
	}
	//将http协议提升为ws
	ws, err := upgrader.Upgrade(c.Writer, c.Request, nil)
	if err != nil {
		fmt.Println(err)
		return nil, err
	}
	return ws, err
}

//为remotecommand.StreamOptions提供方法
type streamHandler struct {
	ws          *websocket.Conn                 //ws
	inputMsg    chan []byte                     //客户端输入数据
	resizeEvent chan remotecommand.TerminalSize //窗口调整事件
}

//获取调整窗口事件
func (handler *streamHandler) Next() *remotecommand.TerminalSize {
	resize := <-handler.resizeEvent
	return &resize
}

//从ws获取客户端输入的数据
func (handler *streamHandler) Read(p []byte) (size int, err error) {
	data, ok := <-handler.inputMsg
	if !ok {
		return 0, errors.New("I/O data reading failed")
	}
	copy(p, data)
	return len(data), nil
}

//将标准输出、错误写入ws（客户端）
func (handler *streamHandler) Write(p []byte) (int, error) {
	// 处理非utf8字符
	if !utf8.Valid(p) {
		bufStr := string(p)
		buf := make([]rune, 0, len(bufStr))
		for _, r := range bufStr {
			if r == utf8.RuneError {
				buf = append(buf, []rune("@")...)
			} else {
				buf = append(buf, r)
			}
		}
		p = []byte(string(buf))
	}
	err := handler.ws.WriteMessage(websocket.TextMessage, p)
	return len(p), err
}

//将字符串转换为int类型
func ToInt(str string) int {
	data, err := strconv.Atoi(str)
	if err != nil {
		fmt.Println(err)
	}
	return data
}

//处理ws输入数据
func executeTask(handler *streamHandler) {
	for {
		_, msg, err := handler.ws.ReadMessage()
		if err != nil {
			return
		}
		//心跳检测
		if string(msg) == "ping" {
			continue
		}
		//调整窗口宽高
		if strings.Contains(string(msg), "resize") {
			resizeSlice := strings.Split(string(msg), ":")
			rows, _ := strconv.Atoi(resizeSlice[1])
			cols, _ := strconv.Atoi(resizeSlice[2])
			handler.resizeEvent <- remotecommand.TerminalSize{
				Width:  uint16(cols),
				Height: uint16(rows),
			}
			continue
		}
		handler.inputMsg <- msg
	}
}

func podTerminal(c *gin.Context) {
	podName := c.Query("podName")
	namespace := c.Query("namespace")
	containerName := c.Query("containerName")
	cols := c.Query("cols")
	rows := c.Query("rows")

	ClientSet, config, err := initialClientSet("./kube/config")
	if err != nil {
		return
	}

	//初始化请求体
	req := ClientSet.CoreV1().RESTClient().Post().
		Resource("pods").
		Name(podName). //podName
		Namespace(namespace). //namespace
		SubResource("exec").
		VersionedParams(&corev1.PodExecOptions{
			Container: containerName, //containerName
			Command:   []string{"bash"},
			Stdin:     true,
			Stdout:    true,
			Stderr:    true,
			TTY:       true, // 启用终端
		}, scheme.ParameterCodec)

	// http转SPDY,添加X-Stream-Protocol-Version等相关header并发送请求
	exec, err := remotecommand.NewSPDYExecutor(config, "POST", req.URL())
	if err != nil {
		log.Println(err)
		return
	}

	ws, err := initialWS(c)
	defer func() {
		ws.Close()
		if err := recover(); err != nil {
			log.Println(err)
		}
	}()

	handler := &streamHandler{
		ws:          ws,
		inputMsg:    make(chan []byte, 1024),
		resizeEvent: make(chan remotecommand.TerminalSize, 1),
	}
	//将初次获取的窗口cols、rows写入channel
	handler.resizeEvent <- remotecommand.TerminalSize{Width: uint16(ToInt(cols)), Height: uint16(ToInt(rows))}

	//获取ws输入数据
	go executeTask(handler)

	if err := exec.Stream(remotecommand.StreamOptions{
		Stdin:             handler,
		Stdout:            handler,
		Stderr:            handler,
		Tty:               true,
		TerminalSizeQueue: handler,
	}); err != nil {
		ws.Close()
		return
	}
}

前端实现逻辑（vue3）

<template>
    <div id="terminal"></div>
</template>

<script>
  import { Terminal } from 'xterm'
  import { FitAddon } from 'xterm-addon-fit'
  import { AttachAddon } from 'xterm-addon-attach'
  import {onBeforeUnmount, onMounted } from "vue"
  import 'xterm/css/xterm.css'
   export default {
     name: 'web-terminal',
     setup(){
       //初始化ws连接
       let ws = new WebSocket("ws://localhost:9090?podName=xxxx?namespace=xxxx?containerName=xxx")
       ws.onopen = ()=>{
         console.log(Date(), 'onopen')
         heartCheck.start()
       }
       ws.onclose = ()=>{
         console.log(Date(), 'onclose')
         heartCheck.stop()
       }
       ws.onerror = ()=> {
         console.log(Date(), 'onerror')
       }

       //心跳检查
       const heartCheck = {
         timeout: 5000, // 5s发一次心跳
         //关闭心跳检查
         stop: function() {
           clearInterval(this.timer)
         },
         //开启心跳检查
         start: function() {
           this.timer = setInterval(function() {
             if (ws !== null && ws.readyState === 1) {
               ws.send('ping')
             }
           }, this.timeout)
         }
       }
       //页面挂载后初始化terminal功能
       onMounted(()=>{
         let webTerminal = document.getElementById('terminal')
         let terminal = new Terminal(
             {
               fontSize: 16
             }
         )
         let fitAddon = new FitAddon()
         terminal.loadAddon(fitAddon)
         terminal.open(webTerminal)
         try {
           fitAddon.fit()
         } catch (e) {
           console.error(e)
         }

         //加载attach插件，通过ws实现web终端与远程终端进行实时交互
         terminal.loadAddon(new AttachAddon(ws))

         //增加滚轮事件监听，用于调整web终端字体大小
         webTerminal.addEventListener("wheel", (e) => {
           if (e.ctrlKey) {
             e.preventDefault()
             if (e.deltaY < 0) {
               terminal.options.fontSize = ++self.fontSize
             } else {
               terminal.options.fontSize = --self.fontSize
             }
             try { fitAddon.fit() } catch (e) {/**/}
             if (ws !== null && ws.readyState === 1) {
               ws.send(`resize:${terminal.rows}:${terminal.cols}`)
             }
           }

           //为window添加窗口大小调整事件，用于实时调整终端窗口
           window.addEventListener('resize', () => {
             webTerminal.style.height = document.documentElement.clientHeight + 'px'
             try { fitAddon.fit() } catch (e) {/**/}
             if (ws !== null && ws.readyState === 1) {
               ws.send(`resize:${terminal.rows}:${terminal.cols}`)
             }
           })
         })
         
         onBeforeUnmount(()=>{
           if (ws !== null) {
             ws.close()
           }
           if (terminal !== null) {
             terminal.dispose()
           }
         })
       })

     }
   }
</script>

<style>
   #terminal{
     position: absolute;
     top: 0; right: 0; bottom: 0; left: 0;
   }
</style>