feat: Add Node File Exchange (#8407)

Author: zhengkunwang223

agent/app/service/file.go

@@ -513,7 +513,7 @@ func (f *FileService) GetPathByType(pathType string) string {
 	if pathType == "websiteDir" {
 		value, _ := settingRepo.GetValueByKey("WEBSITE_DIR")
 		if value == "" {
-			return path.Join(global.Dir.BaseDir, "www")
+			return path.Join(global.Dir.BaseDir, "1panel", "www")
 		}
 		return value
 	}

agent/init/dir/dir.go

@@ -15,7 +15,7 @@ func Init() {
 	global.Dir.BaseDir, _ = fileOp.CreateDirWithPath(true, baseDir)
 	global.Dir.DataDir, _ = fileOp.CreateDirWithPath(true, path.Join(baseDir, "1panel"))
 	global.Dir.DbDir, _ = fileOp.CreateDirWithPath(true, path.Join(baseDir, "1panel/db"))
-	global.Dir.LogDir, _ = fileOp.CreateDirWithPath(true, path.Join(baseDir, "1panel/log"))
+	global.Dir.LogDir, _ = fileOp.CreateDirWithPath(true, path.Join(baseDir, "1panel/log/task"))
 	global.Dir.TmpDir, _ = fileOp.CreateDirWithPath(true, path.Join(baseDir, "1panel/tmp"))
 
 	global.Dir.AppDir, _ = fileOp.CreateDirWithPath(true, path.Join(baseDir, "1panel/apps"))

agent/utils/files/fileinfo.go

@@ -217,7 +217,6 @@ func sortFileList(list []FileSearchInfo, sortBy, sortOrder string) {
 }
 
 func (f *FileInfo) listChildren(option FileOption) error {
-	afs := &afero.Afero{Fs: f.Fs}
 	var (
 		files []FileSearchInfo
 		err   error
@@ -230,7 +229,7 @@ func (f *FileInfo) listChildren(option FileOption) error {
 			return err
 		}
 	} else {
-		files, err = f.getFiles(afs, option)
+		files, err = f.getFiles(option)
 		if err != nil {
 			return err
 		}
@@ -261,21 +260,24 @@ func (f *FileInfo) listChildren(option FileOption) error {
 	return nil
 }
 
-func (f *FileInfo) getFiles(afs *afero.Afero, option FileOption) ([]FileSearchInfo, error) {
-	dirFiles, err := afs.ReadDir(f.Path)
+func (f *FileInfo) getFiles(option FileOption) ([]FileSearchInfo, error) {
+	infos, err := os.ReadDir(f.Path)
 	if err != nil {
 		return nil, err
 	}
-
 	var (
 		dirs     []FileSearchInfo
 		fileList []FileSearchInfo
 	)
 
-	for _, file := range dirFiles {
+	for _, file := range infos {
+		fileInfo, err := file.Info()
+		if err != nil {
+			continue
+		}
 		info := FileSearchInfo{
 			Path:     f.Path,
-			FileInfo: file,
+			FileInfo: fileInfo,
 		}
 		if file.IsDir() {
 			dirs = append(dirs, info)

core/app/task/task.go

@@ -48,16 +48,13 @@ const (
 	TaskUpgrade = "TaskUpgrade"
 	TaskAddNode = "TaskAddNode"
 	TaskSync    = "TaskSync"
+	TaskRsync   = "TaskRsync"
 )
 
 const (
 	TaskScopeSystem = "System"
 	TaskScopeScript = "Script"
-)
-
-const (
-	TaskSuccess = "Success"
-	TaskFailed  = "Failed"
+	TaskScopeNodeFile = "NodeFile"
 )
 
 func GetTaskName(resourceName, operate, scope string) string {
@@ -72,7 +69,7 @@ func NewTask(name, operate, taskScope, taskID string, resourceID uint) (*Task, e
 	if taskID == "" {
 		taskID = uuid.New().String()
 	}
-	logItem := path.Join(global.CONF.Base.InstallDir, "1panel/log")
+	logItem := path.Join(global.CONF.Base.InstallDir, "1panel/log/task")
 	logDir := path.Join(logItem, taskScope)
 	if _, err := os.Stat(logDir); os.IsNotExist(err) {
 		if err = os.MkdirAll(logDir, constant.DirPerm); err != nil {

core/i18n/lang/zh.yaml

@@ -172,4 +172,12 @@ UserInfoAddr: "面板地址："
 UserInfoPassHelp: "提示：修改密码可执行命令："
 DBConnErr: "错误：初始化数据库连接失败，{{ .err }}"
 SystemVersion: "版本："
-SystemMode: "模式："
+SystemMode: "模式："
+
+#exchange
+LocalNodeIpFailed: "无法获取主节点 IP ，请编辑主节点增加 IP 地址和 SSH 认证信息"
+HandlePrivateKey: "处理节点私钥"
+HandlePublicKey: "处理节点公钥"
+ExchangeFile: "开始从 {{ .source }} 节点同步 {{ .sourcePath }} 到 {{ .dest  }} 节点 {{ .destPath }}"
+TaskRsync: "同步"
+NodeFile: "节点文件"

core/init/migration/helper/menu.go

@@ -40,6 +40,7 @@ func LoadMenus() string {
 				{ID: "111", Disabled: false, Title: "xpack.node.nodeManagement", IsShow: true, Label: "Node", Path: "/xpack/node"},
 				{ID: "113", Disabled: false, Title: "xpack.monitor.name", IsShow: true, Label: "MonitorDashboard", Path: "/xpack/monitor/dashboard"},
 				{ID: "114", Disabled: false, Title: "xpack.tamper.tamper", IsShow: true, Label: "Tamper", Path: "/xpack/tamper"},
+				{ID: "117", Disabled: false, Title: "xpack.exchange.exchange", IsShow: true, Label: "FileExange", Path: "/xpack/exchange/file"},
 				{ID: "116", Disabled: false, Title: "xpack.alert.alert", IsShow: true, Label: "XAlertDashboard", Path: "/xpack/alert/dashboard"},
 				{ID: "115", Disabled: false, Title: "xpack.setting.setting", IsShow: true, Label: "XSetting", Path: "/xpack/setting"},
 			}},

core/init/migration/migrations/init.go

@@ -95,7 +95,7 @@ var InitSetting = &gormigrate.Migration{
 			return err
 		}
 		val := `{"id":"1","label":"/xpack","isCheck":true,"title":"xpack.menu","children":[{"id":"2","label":"Dashboard","isCheck":true,"title":"xpack.waf.name","path":"/xpack/waf/dashboard"},{"id":"3","label":"Tamper","isCheck":true,"title":"xpack.tamper.tamper","path":"/xpack/tamper"},{"id":"4","label":"GPU","isCheck":true,"title":"xpack.gpu.gpu","path":"/xpack/gpu"},{"id":"5","label":"XSetting","isCheck":true,"title":"xpack.setting.setting","path":"/xpack/setting"},{"id":"6","label":"MonitorDashboard","isCheck":true,"title":"xpack.monitor.name","path":"/xpack/monitor/dashboard"},{"id":"7","label":"XAlertDashboard","isCheck":true,"title":"xpack.alert.alert","path":"/xpack/alert/dashboard"},{"id":"8","label":"Node","isCheck":true,"title":"xpack.node.nodeManagement","path":"/xpack/node"}]}`
-		if err := tx.Create(&model.Setting{Key: "XpackHideMenu", Value: val}).Error; err != nil {
+		if err := tx.Create(&model.Setting{Key: "HideMenu", Value: val}).Error; err != nil {
 			return err
 		}
 
@@ -303,9 +303,9 @@ var AddMFAInterval = &gormigrate.Migration{
 }
 
 var UpdateXpackHideMemu = &gormigrate.Migration{
-	ID: "20250227-update-xpack-hide-menu",
+	ID: "20250414-update-xpack-hide-menu",
 	Migrate: func(tx *gorm.DB) error {
-		if err := tx.Model(&model.Setting{}).Where("key = ?", "XpackHideMenu").Updates(map[string]interface{}{"key": "HideMenu", "value": helper.LoadMenus()}).Error; err != nil {
+		if err := tx.Model(&model.Setting{}).Where("key = ?", "HideMenu").Updates(map[string]interface{}{"key": "HideMenu", "value": helper.LoadMenus()}).Error; err != nil {
 			return err
 		}
 		return nil

core/init/router/proxy.go

@@ -19,13 +19,6 @@ import (
 	"github.com/gin-gonic/gin"
 )
 
-var wsUrl = map[string]struct{}{
-	"/api/v2/process/ws":         {},
-	"/api/v2/files/wget/process": {},
-
-	"/api/v2/containers/search/log": {},
-}
-
 func Proxy() gin.HandlerFunc {
 	return func(c *gin.Context) {
 		reqPath := c.Request.URL.Path
@@ -38,8 +31,8 @@ func Proxy() gin.HandlerFunc {
 			return
 		}
 		var currentNode string
-		if _, ok := wsUrl[reqPath]; ok {
-			currentNode = c.Query("currentNode")
+		if c.Query("operateNode") != "" {
+			currentNode = c.Query("operateNode")
 		} else {
 			currentNode = c.Request.Header.Get("CurrentNode")
 		}

core/utils/cmd/cmd.go

@@ -1,6 +1,9 @@
 package cmd
 
 import (
+	"bufio"
+	"fmt"
+	"io"
 	"os/exec"
 	"strings"
 )
@@ -20,3 +23,36 @@ func Which(name string) bool {
 	}
 	return true
 }
+
+func ExecWithStreamOutput(command string, outputCallback func(string)) error {
+	cmd := exec.Command("bash", "-c", command)
+
+	stdout, err := cmd.StdoutPipe()
+	if err != nil {
+		return fmt.Errorf("failed to get stdout: %w", err)
+	}
+
+	stderr, err := cmd.StderrPipe()
+	if err != nil {
+		return fmt.Errorf("failed to get stderr: %w", err)
+	}
+
+	if err := cmd.Start(); err != nil {
+		return fmt.Errorf("failed to start command: %w", err)
+	}
+
+	go streamReader(stdout, outputCallback)
+	go streamReader(stderr, outputCallback)
+
+	if err := cmd.Wait(); err != nil {
+		return fmt.Errorf("command finished with error: %w", err)
+	}
+	return nil
+}
+
+func streamReader(reader io.ReadCloser, callback func(string)) {
+	scanner := bufio.NewScanner(reader)
+	for scanner.Scan() {
+		callback(scanner.Text())
+	}
+}

core/utils/ssh/ssh.go

@@ -96,3 +96,94 @@ func makePrivateKeySigner(privateKey []byte, passPhrase []byte) (gossh.Signer, e
 	}
 	return gossh.ParsePrivateKey(privateKey)
 }
+
+func (c *SSHClient) RunWithStreamOutput(command string, outputCallback func(string)) error {
+	session, err := c.Client.NewSession()
+	if err != nil {
+		return fmt.Errorf("failed to create SSH session: %w", err)
+	}
+	defer session.Close()
+
+	stdout, err := session.StdoutPipe()
+	if err != nil {
+		return fmt.Errorf("failed to set up stdout pipe: %w", err)
+	}
+
+	stderr, err := session.StderrPipe()
+	if err != nil {
+		return fmt.Errorf("failed to set up stderr pipe: %w", err)
+	}
+
+	if err := session.Start(command); err != nil {
+		return fmt.Errorf("failed to start command: %w", err)
+	}
+
+	stdoutCh := make(chan string, 100)
+	stderrCh := make(chan string, 100)
+	doneCh := make(chan struct{})
+
+	go func() {
+		buffer := make([]byte, 1024)
+		for {
+			n, err := stdout.Read(buffer)
+			if err != nil {
+				close(stdoutCh)
+				return
+			}
+			if n > 0 {
+				stdoutCh <- string(buffer[:n])
+			}
+		}
+	}()
+
+	go func() {
+		buffer := make([]byte, 1024)
+		for {
+			n, err := stderr.Read(buffer)
+			if err != nil {
+				close(stderrCh)
+				return
+			}
+			if n > 0 {
+				stderrCh <- string(buffer[:n])
+			}
+		}
+	}()
+
+	go func() {
+		for {
+			select {
+			case stdoutOutput, ok := <-stdoutCh:
+				if !ok {
+					stdoutCh = nil
+					if stderrCh == nil {
+						close(doneCh)
+						return
+					}
+					continue
+				}
+				if outputCallback != nil {
+					outputCallback(stdoutOutput)
+				}
+
+			case stderrOutput, ok := <-stderrCh:
+				if !ok {
+					stderrCh = nil
+					if stdoutCh == nil {
+						close(doneCh)
+						return
+					}
+					continue
+				}
+				if outputCallback != nil {
+					outputCallback(stderrOutput)
+				}
+			}
+		}
+	}()
+
+	err = session.Wait()
+	<-doneCh
+
+	return err
+}