update node-agent Heartbeat

Signed-off-by: gaoyuan <[email protected]>

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

@@ -1,7 +1,6 @@
 package app
 
 import (
-	"os"
 	"strings"
 
 	"github.com/spf13/cobra"
@@ -30,11 +29,12 @@ var RootCmd = &cobra.Command{
 
 func initConfig() {
 	// Tell Viper to automatically look for a .env file
-	viper.SetConfigFile("agent.env")
-	currentDir, _ := os.Getwd()
-	viper.AddConfigPath(currentDir)
-	viper.AddConfigPath("/srv/node-agent/agent.env")
-	viper.SetConfigType("toml")
+	//viper.SetConfigFile("agent.env")
+	viper.SetConfigFile("/srv/node-agent/agent.env")
+	//currentDir, _ := os.Getwd()
+	//viper.AddConfigPath(currentDir)
+	//viper.AddConfigPath("/srv/node-agent/agent.env")
+	//viper.SetConfigType("toml")
 	// If a agent.env file is found, read it in.
 	if err := viper.ReadInConfig(); err != nil {
 		log.Warnf("Load config file error, %s", err)

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

@@ -2,6 +2,7 @@ package serve
 
 import (
 	"bufio"
+	"context"
 	"crypto/sha256"
 	"crypto/tls"
 	"encoding/base64"
@@ -22,8 +23,12 @@ import (
 	"github.com/gorilla/websocket"
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/tools/clientcmd"
 
 	"github.com/kosmos.io/kosmos/cmd/kubenest/node-agent/app/logger"
+	"github.com/kosmos.io/kosmos/pkg/generated/clientset/versioned"
 )
 
 var (
@@ -36,6 +41,7 @@ var (
 	certFile string // SSL certificate file
 	keyFile  string // SSL key file
 	addr     string // server listen address
+	nodeName string // server nodename
 	log      = logger.GetLogger()
 )
 
@@ -52,9 +58,19 @@ func init() {
 	ServeCmd.PersistentFlags().StringVarP(&addr, "addr", "a", ":5678", "websocket service address")
 	ServeCmd.PersistentFlags().StringVarP(&certFile, "cert", "c", "cert.pem", "SSL certificate file")
 	ServeCmd.PersistentFlags().StringVarP(&keyFile, "key", "k", "key.pem", "SSL key file")
+	ServeCmd.PersistentFlags().StringVarP(&nodeName, "nodename", "n", "", "set nodename")
 }
 
 func serveCmdRun(_ *cobra.Command, _ []string) error {
+	//start heartbeatCheck Goroutine
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	if len(nodeName) == 0 {
+		nodeName = viper.GetString("NODE_NAME")
+	}
+	go heartbeatCheck(ctx, nodeName)
+
 	user := viper.GetString("WEB_USER")
 	password := viper.GetString("WEB_PASS")
 	port := viper.GetString("WEB_PORT")
@@ -65,9 +81,52 @@ func serveCmdRun(_ *cobra.Command, _ []string) error {
 	if port != "" {
 		addr = ":" + port
 	}
+
 	return Start(addr, certFile, keyFile, user, password)
 }
 
+func heartbeatCheck(ctx context.Context, nodeName string) {
+	kubeconfigPath := "/srv/node-agent/kubeconfigpath/kubeconfig"
+	config, err := clientcmd.BuildConfigFromFlags("", kubeconfigPath)
+	if err != nil {
+		log.Errorf("Failed to load kubeconfig from path %s:%v", kubeconfigPath, err)
+		return
+	}
+
+	kosmosClient, err := versioned.NewForConfig(config)
+	if err != nil {
+		log.Errorf("Failed to get config: %v", err)
+		return
+	}
+
+	ticker := time.NewTicker(10 * time.Second)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			log.Infof("Heartbeat for node %s stopped", nodeName)
+			return
+		case <-ticker.C:
+			node, err := kosmosClient.KosmosV1alpha1().GlobalNodes().Get(ctx, nodeName, metav1.GetOptions{})
+			if err != nil {
+				log.Errorf("Failed to get node: %v", err)
+			}
+			heartbeatTime := metav1.Now()
+			node.Status.Conditions = []corev1.NodeCondition{
+				{
+					LastHeartbeatTime: heartbeatTime,
+				},
+			}
+			if _, err := kosmosClient.KosmosV1alpha1().GlobalNodes().UpdateStatus(ctx, node, metav1.UpdateOptions{}); err != nil {
+				log.Errorf("update node %s status for globalnode failed, %v", node.Name, err)
+			} else {
+				log.Infof("GlobalnodeHeartbeat: successfully updated global node %s, Status.Conditions: %+v", node.Name, node.Status.Conditions)
+			}
+		}
+	}
+}
+
 // start server
 func Start(addr, certFile, keyFile, user, password string) error {
 	passwordHash := sha256.Sum256([]byte(password))
@@ -146,6 +205,7 @@ func Start(addr, certFile, keyFile, user, password string) error {
 		TLSConfig:         tlsConfig,
 		ReadHeaderTimeout: 10 * time.Second,
 	}
+
 	err := server.ListenAndServeTLS("", "")
 	if err != nil {
 		log.Errorf("failed to start server %v", err)

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

@@ -0,0 +1,44 @@
+package serve
+
+import (
+	"bytes"
+	"context"
+	"strings"
+	"testing"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"github.com/kosmos.io/kosmos/pkg/apis/kosmos/v1alpha1"
+)
+
+func TestHeartbeatCheck(t *testing.T) {
+	var logBuffer bytes.Buffer
+	log.SetOutput(&logBuffer)
+	defer log.SetOutput(nil)
+
+	globalNode := &v1alpha1.GlobalNode{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "test-node",
+		},
+		Status: v1alpha1.GlobalNodeStatus{
+			Conditions: []corev1.NodeCondition{},
+		},
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	heartbeatCheck(ctx, globalNode.Name)
+
+	logOutput := logBuffer.String()
+	if strings.Contains(logOutput, "Failed to load kubeconfig") {
+		t.Errorf("Heartbeat check failed: %s", logOutput)
+	}
+	if strings.Contains(logOutput, "Failed to get node") {
+		t.Errorf("Heartbeat check failed: %s", logOutput)
+	}
+	if strings.Contains(logOutput, "update node") && strings.Contains(logOutput, "failed") {
+		t.Errorf("Heartbeat check update failed: %s", logOutput)
+	}
+}

deploy/virtual-cluster-operator.yml

@@ -230,6 +230,11 @@ spec:
                 secretKeyRef:
                   name: node-agent-secret
                   key: password
+            - name: NODE_NAME
+              valueFrom:
+                fieldRef:
+                  apiVersion: v1
+                  fieldPath: spec.nodeName
           command: ["/bin/bash"]
           args:
             - "/app/init.sh"
@@ -240,6 +245,10 @@ spec:
             - mountPath: /host-systemd
               name: systemd-path
               readOnly: false
+            - mountPath: /app/kubeconfigpath
+              name: kubeconfig
+              subPath: kubeconfig  
+              readOnly: false
       containers:
         - name: install-agent
           image: cis-hub-huabei-3.cmecloud.cn/node-agent/node-agent:latest
@@ -282,6 +291,10 @@ spec:
           hostPath:
             path: /etc/systemd/system
             type: DirectoryOrCreate
+        - name: kubeconfig
+          secret:
+            secretName: virtual-cluster-operator
+
 ---
 apiVersion: v1
 kind: Secret

hack/node-agent/agent.env

@@ -1,3 +1,4 @@
 WEB_USER={{WEB_USER}}
 WEB_PASS={{WEB_PASS}}
 WEB_PORT={{WEB_PORT}}
+NODE_NAME={{NODE_NAME}}

hack/node-agent/init.sh

@@ -3,7 +3,9 @@
 WEB_USER="$WEB_USER" sed -i 's/^WEB_USER=.*/WEB_USER="'"$WEB_USER"'"/' /app/agent.env
 WEB_PASS="$WEB_PASS" sed -i 's/^WEB_PASS=.*/WEB_PASS="'"$WEB_PASS"'"/' /app/agent.env
 WEB_PORT="$WEB_PORT" sed -i 's/^WEB_PORT=.*/WEB_PORT="'"$WEB_PORT"'"/' /app/agent.env
+NODE_NAME="$NODE_NAME" sed -i 's/^NODE_NAME=.*/NODE_NAME="'"$NODE_NAME"'"/' /app/agent.env
+
 sha256sum /app/node-agent > /app/node-agent.sum
 sha256sum /host-path/node-agent >> /app/node-agent.sum
 rsync -avz /app/ /host-path/
-cp /app/node-agent.service /host-systemd/node-agent.service
+cp /app/node-agent.service /host-systemd/node-agent.service