feat: 网元主机添加redis连接终端控制

src/framework/redis/conn.go

@@ -59,3 +59,21 @@ func (c *ConnRedis) Close() {
 		c.Client.Close()
 	}
 }
+
+// RunCMD 执行单次命令 "GET key"
+func (c *ConnRedis) RunCMD(cmd string) (any, error) {
+	if c.Client == nil {
+		return "", fmt.Errorf("redis client not connected")
+	}
+	// 写入命令
+	cmdArr := strings.Fields(cmd)
+	if len(cmdArr) == 0 {
+		return "", fmt.Errorf("redis command is empty")
+	}
+	conn := *c.Client
+	args := make([]any, 0)
+	for _, v := range cmdArr {
+		args = append(args, v)
+	}
+	return conn.Do(context.Background(), args...).Result()
+}

src/modules/network_element/controller/ne_host.go

@@ -79,6 +79,13 @@ func (s *NeHostController) Add(c *gin.Context) {
 		return
 	}
 
+	if body.GroupID == "1" {
+		// 主机信息操作【%s】失败，禁止操作网元
+		msg := i18n.TKey(language, "neHost.banNE")
+		c.JSON(200, result.ErrMsg(msg))
+		return
+	}
+
 	// 检查属性值唯一
 	uniqueHost := s.neHostService.CheckUniqueHostTitle(body.GroupID, body.Title, body.HostType, "")
 	if !uniqueHost {

src/modules/network_element/repository/ne_host.go

@@ -111,8 +111,28 @@ func (r *NeHost) SelectPage(query map[string]any) map[string]any {
 	params = append(params, pageNum*pageSize)
 	params = append(params, pageSize)
 
+	// 排序
+	orderSql := ""
+	if sv, ok := query["sortField"]; ok && sv != "" {
+		sortSql := fmt.Sprint(sv)
+		if sortSql == "updateTime" {
+			sortSql = "update_time"
+		}
+		if sortSql == "createTime" {
+			sortSql = "create_time"
+		}
+		if ov, ok := query["sortOrder"]; ok && ov != "" {
+			if fmt.Sprint(ov) == "desc" {
+				sortSql += " desc "
+			} else {
+				sortSql += " asc "
+			}
+		}
+		orderSql = fmt.Sprintf(" order by %s ", sortSql)
+	}
+
 	// 查询数据
-	querySql := r.selectSql + whereSql + pageSql
+	querySql := r.selectSql + whereSql + orderSql + pageSql
 	results, err := datasource.RawDB("", querySql, params)
 	if err != nil {
 		logger.Errorf("query err => %v", err)

src/modules/network_element/repository/ne_info.go

@@ -18,6 +18,7 @@ var neListSort = []string{
 	"IMS",
 	"AMF",
 	"AUSF",
+	"UDR",
 	"UDM",
 	"SMF",
 	"PCF",

src/modules/network_element/service/ne_host.go

@@ -149,6 +149,13 @@ func (r *NeHost) DeleteByIds(hostIds []string) (int64, error) {
 		return 0, fmt.Errorf("neHost.noData")
 	}
 
+	for _, v := range ids {
+		if v.GroupID == "1" {
+			// 主机信息操作【%s】失败，禁止操作网元
+			return 0, fmt.Errorf("neHost.banNE")
+		}
+	}
+
 	if len(ids) == len(hostIds) {
 		rows := r.neHostRepository.DeleteByIds(hostIds)
 		return rows, nil

src/modules/ws/controller/ws_redis.go

@@ -0,0 +1,69 @@
+package controller
+
+import (
+	"be.ems/src/framework/i18n"
+	"be.ems/src/framework/logger"
+	"be.ems/src/framework/redis"
+	"be.ems/src/framework/utils/ctx"
+	"be.ems/src/framework/vo/result"
+	neService "be.ems/src/modules/network_element/service"
+
+	"github.com/gin-gonic/gin"
+)
+
+// Redis 终端
+//
+// GET /redis?hostId=1
+func (s *WSController) Redis(c *gin.Context) {
+	language := ctx.AcceptLanguage(c)
+	var query struct {
+		HostId string `form:"hostId" binding:"required"` // 连接主机ID
+	}
+	if err := c.ShouldBindQuery(&query); err != nil {
+		c.JSON(400, result.CodeMsg(400, i18n.TKey(language, "app.common.err400")))
+		return
+	}
+
+	// 登录用户信息
+	loginUser, err := ctx.LoginUser(c)
+	if err != nil {
+		c.JSON(401, result.CodeMsg(401, i18n.TKey(language, err.Error())))
+		return
+	}
+
+	neHost := neService.NewNeHost.SelectById(query.HostId)
+	if neHost.HostID != query.HostId || neHost.HostType != "redis" {
+		// 没有可访问主机信息数据！
+		c.JSON(200, result.ErrMsg(i18n.TKey(language, "neHost.noData")))
+		return
+	}
+
+	// 创建链接Redis客户端
+	var connRedis redis.ConnRedis
+	neHost.CopyTo(&connRedis)
+	client, err := connRedis.NewClient()
+	if err != nil {
+		// 连接主机失败，请检查连接参数后重试
+		c.JSON(200, result.ErrMsg(i18n.TKey(language, "neHost.errByHostInfo")))
+		return
+	}
+	defer client.Close()
+
+	// 将 HTTP 连接升级为 WebSocket 连接
+	wsConn := s.wsService.UpgraderWs(c.Writer, c.Request)
+	if wsConn == nil {
+		return
+	}
+	defer wsConn.Close()
+
+	wsClient := s.wsService.ClientCreate(loginUser.UserID, nil, wsConn, client)
+	go s.wsService.ClientWriteListen(wsClient)
+	go s.wsService.ClientReadListen(wsClient, s.wsReceiveService.Redis)
+
+	// 等待停止信号
+	for value := range wsClient.StopChan {
+		s.wsService.ClientClose(wsClient.ID)
+		logger.Infof("ws Stop Client UID %s %s", wsClient.BindUid, value)
+		return
+	}
+}

src/modules/ws/service/ws_receive.go

@@ -4,9 +4,12 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"reflect"
+	"strings"
 	"time"
 
 	"be.ems/src/framework/logger"
+	"be.ems/src/framework/redis"
 	"be.ems/src/framework/telnet"
 	"be.ems/src/framework/utils/ssh"
 	"be.ems/src/framework/vo/result"
@@ -104,7 +107,7 @@ func (s *WSReceive) Shell(client *model.WSClient, reqMsg model.WSRequest) {
 		command := reqMsg.Data.(string)
 		sshClientSession := client.ChildConn.(*ssh.SSHClientSession)
 		_, err = sshClientSession.Write(command)
-	case "ssh_resize":
+	case "resize":
 		// SSH会话窗口重置
 		msgByte, _ := json.Marshal(reqMsg.Data)
 		var data struct {
@@ -225,7 +228,7 @@ func (s *WSReceive) Telnet(client *model.WSClient, reqMsg model.WSRequest) {
 		command := reqMsg.Data.(string)
 		telnetClientSession := client.ChildConn.(*telnet.TelnetClientSession)
 		_, err = telnetClientSession.Write(command)
-	case "telnet_resize":
+	case "resize":
 		// Telnet会话窗口重置
 		msgByte, _ := json.Marshal(reqMsg.Data)
 		var data struct {
@@ -256,3 +259,76 @@ func (s *WSReceive) Telnet(client *model.WSClient, reqMsg model.WSRequest) {
 		client.MsgChan <- resByte
 	}
 }
+
+// Redis 接收终端交互业务处理
+func (s *WSReceive) Redis(client *model.WSClient, reqMsg model.WSRequest) {
+	// 必传requestId确认消息
+	if reqMsg.RequestID == "" {
+		msg := "message requestId is required"
+		logger.Infof("ws Shell UID %s err: %s", client.BindUid, msg)
+		msgByte, _ := json.Marshal(result.ErrMsg(msg))
+		client.MsgChan <- msgByte
+		return
+	}
+
+	var resByte []byte
+	var err error
+
+	switch reqMsg.Type {
+	case "close":
+		s.close(client)
+		return
+	case "redis":
+		// Redis会话消息接收写入会话
+		command := fmt.Sprint(reqMsg.Data)
+		redisClientSession := client.ChildConn.(*redis.ConnRedis)
+		output, err := redisClientSession.RunCMD(command)
+		dataStr := ""
+		if err != nil {
+			dataStr = fmt.Sprintf("%s \r\n", err.Error())
+		} else {
+			// 获取结果的反射类型
+			resultType := reflect.TypeOf(output)
+			switch resultType.Kind() {
+			case reflect.Slice:
+				// 如果是切片类型需要进一步判断是否是 []string 或 []interface{}
+				if resultType.Elem().Kind() == reflect.String {
+					dataStr = fmt.Sprintf("%s \r\n", strings.Join(output.([]string), "\r\n"))
+				} else if resultType.Elem().Kind() == reflect.Interface {
+					arr := []string{}
+					for _, v := range output.([]any) {
+						arr = append(arr, fmt.Sprintf("%s", v))
+					}
+					dataStr = fmt.Sprintf("%s \r\n", strings.Join(arr, "\r\n"))
+				}
+			case reflect.Ptr:
+				dataStr = "\r\n"
+			case reflect.String, reflect.Int64:
+				dataStr = fmt.Sprintf("%s \r\n", output)
+			default:
+				dataStr = fmt.Sprintf("%s \r\n", output)
+			}
+		}
+		resByte, _ = json.Marshal(result.Ok(map[string]any{
+			"requestId": reqMsg.RequestID,
+			"data":      dataStr,
+		}))
+	default:
+		err = fmt.Errorf("message type %s not supported", reqMsg.Type)
+	}
+
+	if err != nil {
+		logger.Warnf("ws Shell UID %s err: %s", client.BindUid, err.Error())
+		msgByte, _ := json.Marshal(result.ErrMsg(err.Error()))
+		client.MsgChan <- msgByte
+		if err == io.EOF {
+			// 等待1s后关闭连接
+			time.Sleep(1 * time.Second)
+			client.StopChan <- struct{}{}
+		}
+		return
+	}
+	if len(resByte) > 0 {
+		client.MsgChan <- resByte
+	}
+}

src/modules/ws/ws.go

@@ -35,6 +35,11 @@ func Setup(router *gin.Engine) {
 			collectlogs.OperateLog(collectlogs.OptionNew("log.operate.title.ws", collectlogs.BUSINESS_TYPE_OTHER)),
 			controller.NewWSController.Telnet,
 		)
+		wsGroup.GET("/redis",
+			middleware.PreAuthorize(nil),
+			collectlogs.OperateLog(collectlogs.OptionNew("log.operate.title.ws", collectlogs.BUSINESS_TYPE_OTHER)),
+			controller.NewWSController.Redis,
+		)
 		wsGroup.GET("/view",
 			middleware.PreAuthorize(nil),
 			collectlogs.OperateLog(collectlogs.OptionNew("log.operate.title.ws", collectlogs.BUSINESS_TYPE_OTHER)),