Merge pull request kosmos-io#659 from OrangeBao/main

feat: adapt to the new node agent

Author: duanmengkk

.gitignore

@@ -30,4 +30,9 @@ kube-config
 
 __debug_bin*
 
-ignore_dir
+ignore_dir
+
+cmd/kubenest/node-agent/app.log
+cmd/kubenest/node-agent/cert.pem
+cmd/kubenest/node-agent/key.pem
+cmd/kubenest/node-agent/agent.env

cmd/kubenest/node-agent/app/serve/serve.go

@@ -155,12 +155,13 @@ func handleCheck(conn *websocket.Conn, params url.Values) {
 	listener, err := net.Listen("tcp", address)
 	if err != nil {
 		log.Infof("port not avalible %s %v", address, err)
-		_ = conn.WriteMessage(websocket.BinaryMessage, []byte("1"))
+		_ = conn.WriteMessage(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, fmt.Sprintf("%d", 1)))
 		return
 	}
 	defer listener.Close()
 	log.Infof("port avalible %s", address)
-	_ = conn.WriteMessage(websocket.BinaryMessage, []byte("0"))
+	// _ = conn.WriteMessage(websocket.BinaryMessage, []byte("0"))
+	_ = conn.WriteMessage(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, fmt.Sprintf("%d", 0)))
 }
 
 func handleTty(conn *websocket.Conn, queryParams url.Values) {

hack/k8s-in-k8s/kubelet_node_helper.sh

@@ -8,7 +8,6 @@ LOG_NAME=${2:-kubelet}
 JOIN_HOST=$2
 JOIN_TOKEN=$3
 JOIN_CA_HASH=$4
-CHECK_PORT=$2
 
 function unjoin() {
     # before unjoin, you need delete node by kubectl
@@ -331,13 +330,6 @@ function version() {
     echo "$SCRIPT_VERSION"
 }
 
-function port() {
-    echo "check port(1/1): use netstat check port :$CHECK_PORT"
-    output=$(netstat -ant | awk '{print $4}' | grep ":$CHECK_PORT" | wc -l)
-    # Do not modify the fixed format
-    echo "port:$CHECK_PORT/$output"
-}
-
 # See how we were called.
 case "$1" in
     unjoin)
@@ -346,9 +338,6 @@ case "$1" in
     join)
     join
     ;;
-    port)
-    port
-    ;;
     health)
     health
     ;;
@@ -365,6 +354,6 @@ case "$1" in
     version
     ;;
     *)
-    echo $"usage: $0 unjoin|join|health|log|check|version|revert|port"
+    echo $"usage: $0 unjoin|join|health|log|check|version|revert"
     exit 1
 esac

pkg/kubenest/controller/virtualcluster.node.controller/exector/exector.go

@@ -30,10 +30,11 @@ type ExectorReturn struct {
 	Reason  string
 	LastLog string
 	Text    string
+	Code    int
 }
 
 func (r *ExectorReturn) String() string {
-	return fmt.Sprintf("%d, %s, %s", r.Status, r.Reason, r.LastLog)
+	return fmt.Sprintf("%d, %s, %s, %d", r.Status, r.Reason, r.LastLog, r.Code)
 }
 
 type Exector interface {
@@ -98,7 +99,7 @@ func (h *ExectorHelper) DoExector(stopCh <-chan struct{}, exector Exector) *Exec
 func (h *ExectorHelper) DoExectorReal(stopCh <-chan struct{}, exector Exector) *ExectorReturn {
 	// default is error
 	result := &ExectorReturn{
-		FAILED, "init exector return status", "", "",
+		FAILED, "init exector return status", "", "", 0,
 	}
 
 	// nolint
@@ -133,6 +134,7 @@ func (h *ExectorHelper) DoExectorReal(stopCh <-chan struct{}, exector Exector) *
 						result.Reason = "command not found"
 						result.Text = cerr.Text
 					}
+					result.Code = cerr.Code
 				} else {
 					result.Reason = err.Error()
 				}

pkg/kubenest/controller/virtualcluster.node.controller/exector/remote_check.go

@@ -0,0 +1,20 @@
+package exector
+
+import (
+	"github.com/gorilla/websocket"
+)
+
+type CheckExector struct {
+	Port string
+}
+
+func (e *CheckExector) GetWebSocketOption() WebSocketOption {
+	rawQuery := "port=" + e.Port
+	return WebSocketOption{
+		Path:     "check/",
+		RawQuery: rawQuery,
+	}
+}
+
+func (e *CheckExector) SendHandler(_ *websocket.Conn, _ <-chan struct{}, _ chan struct{}, _ *ExectorReturn) {
+}

pkg/kubenest/controller/virtualcluster_init_controller.go

@@ -5,7 +5,6 @@ import (
 	"encoding/base64"
 	"fmt"
 	"sort"
-	"strings"
 	"sync"
 	"time"
 
@@ -694,14 +693,14 @@ func findAddress(node corev1.Node) (string, error) {
 // Return false to indicate that the port is not occupied
 func CheckPortOnHost(addr string, port int32) (bool, error) {
 	hostExectorHelper := exector.NewExectorHelper(addr, "")
-	joinCmdStrCmd := &exector.CMDExector{
-		Cmd: fmt.Sprintf("bash %s port %d", env.GetExectorShellName(), port),
+	checkCmd := &exector.CheckExector{
+		Port: fmt.Sprintf("%d", port),
 	}
 
 	var ret *exector.ExectorReturn
 	err := apiclient.TryRunCommand(func() error {
-		ret = hostExectorHelper.DoExector(context.TODO().Done(), joinCmdStrCmd)
-		if ret.Status != exector.SUCCESS {
+		ret = hostExectorHelper.DoExector(context.TODO().Done(), checkCmd)
+		if ret.Code != 1000 {
 			return fmt.Errorf("chekc port failed, err: %s", ret.String())
 		}
 		return nil
@@ -711,9 +710,12 @@ func CheckPortOnHost(addr string, port int32) (bool, error) {
 		klog.Errorf("check port on host error! addr:%s, port %d, err: %s", addr, port, err.Error())
 		return true, err
 	}
-	klog.V(4).Infof("check port on host, addr: %s, port %d, result: %s", addr, port, ret.String())
 
-	return !strings.HasSuffix(ret.LastLog, fmt.Sprintf("port:%d/0", port)), nil
+	if ret.Status != exector.SUCCESS {
+		return true, fmt.Errorf("pod[%d] is occupied", port)
+	} else {
+		return false, nil
+	}
 }
 
 func (c *VirtualClusterInitController) findHostAddresses() ([]string, error) {