TRKJ/mayfly-go
2
0
Fork 0
mirror of https://gitee.com/dromara/mayfly-go synced 2025-11-04 00:10:25 +08:00
Code Issues Packages Projects Releases Wiki Activity

!134 feat: 新增支持es和连接池

Browse Source 
* feat: 各连接，支持连接池
* feat：支持es
This commit is contained in:
zongyangleo
2025-05-21 04:42:30 +00:00
committed by Coder慌
parent f676ec9e7b
commit 142bbd265d
89 changed files with 5734 additions and 575 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
server/.gitignore vendored
View File

@@ -8,4 +8,7 @@ mayfly_rsa.pub
# mysql 程序目录
/db/mysql/
# mariadb 程序目录
/db/mariadb/
/db/mariadb/
*.sqlite
file

8
server/internal/db/api/db.go
View File

@@ -8,6 +8,7 @@ import (
"mayfly-go/internal/db/application"
"mayfly-go/internal/db/application/dto"
"mayfly-go/internal/db/config"
"mayfly-go/internal/db/dbm"
"mayfly-go/internal/db/dbm/dbi"
"mayfly-go/internal/db/domain/entity"
"mayfly-go/internal/db/imsg"
@@ -142,6 +143,8 @@ func (d *Db) ExecSql(rc *req.Ctx) {
dbId := getDbId(rc)
dbConn, err := d.dbApp.GetDbConn(dbId, form.Db)
biz.ErrIsNil(err)
defer dbm.PutDbConn(dbConn)
biz.ErrIsNilAppendErr(d.tagApp.CanAccess(rc.GetLoginAccount().Id, dbConn.Info.CodePath...), "%s")
global.EventBus.Publish(rc.MetaCtx, event.EventTopicResourceOp, dbConn.Info.CodePath[0])
@@ -193,6 +196,7 @@ func (d *Db) ExecSqlFile(rc *req.Ctx) {
dbConn, err := d.dbApp.GetDbConn(dbId, dbName)
biz.ErrIsNil(err)
defer dbm.PutDbConn(dbConn)
biz.ErrIsNilAppendErr(d.tagApp.CanAccess(rc.GetLoginAccount().Id, dbConn.Info.CodePath...), "%s")
rc.ReqParam = fmt.Sprintf("filename: %s -> %s", filename, dbConn.Info.GetLogDesc())
@@ -226,6 +230,8 @@ func (d *Db) DumpSql(rc *req.Ctx) {
la := rc.GetLoginAccount()
dbConn, err := d.dbApp.GetDbConn(dbId, dbName)
biz.ErrIsNil(err)
defer dbm.PutDbConn(dbConn)
biz.ErrIsNilAppendErr(d.tagApp.CanAccess(la.Id, dbConn.Info.CodePath...), "%s")
now := time.Now()
@@ -354,6 +360,7 @@ func (d *Db) CopyTable(rc *req.Ctx) {
conn, err := d.dbApp.GetDbConn(form.Id, form.Db)
biz.ErrIsNilAppendErr(err, "copy table error: %s")
defer dbm.PutDbConn(conn)
err = conn.GetDialect().CopyTable(copy)
if err != nil {
@@ -377,5 +384,6 @@ func getDbName(rc *req.Ctx) string {
func (d *Db) getDbConn(rc *req.Ctx) *dbi.DbConn {
dc, err := d.dbApp.GetDbConn(getDbId(rc), getDbName(rc))
biz.ErrIsNil(err)
defer dbm.PutDbConn(dc)
return dc
}

3
server/internal/db/api/db_transfer.go
View File

@@ -6,6 +6,7 @@ import (
"mayfly-go/internal/db/api/vo"
"mayfly-go/internal/db/application"
"mayfly-go/internal/db/application/dto"
"mayfly-go/internal/db/dbm"
"mayfly-go/internal/db/domain/entity"
"mayfly-go/internal/db/imsg"
fileapp "mayfly-go/internal/file/application"
@@ -151,6 +152,8 @@ func (d *DbTransferTask) FileRun(rc *req.Ctx) {
targetDbConn, err := d.dbApp.GetDbConn(fm.TargetDbId, fm.TargetDbName)
biz.ErrIsNilAppendErr(err, "failed to connect to the target database: %s")
defer dbm.PutDbConn(targetDbConn)
biz.ErrIsNilAppendErr(d.tagApp.CanAccess(rc.GetLoginAccount().Id, targetDbConn.Info.CodePath...), "%s")
filename, reader, err := d.fileApp.GetReader(context.TODO(), tFile.FileKey)

5
server/internal/db/application/db_data_sync.go
View File

@@ -5,6 +5,7 @@ import (
"context"
"encoding/json"
"fmt"
"mayfly-go/internal/db/dbm"
"mayfly-go/internal/db/dbm/dbi"
"mayfly-go/internal/db/domain/entity"
"mayfly-go/internal/db/domain/repository"
@@ -150,6 +151,7 @@ func (app *dataSyncAppImpl) RunCronJob(ctx context.Context, id uint64) error {
return
}
srcConn, err := app.dbApp.GetDbConn(uint64(task.SrcDbId), task.SrcDbName)
defer dbm.PutDbConn(srcConn)
if err != nil {
logx.ErrorfContext(ctx, "failed to connect to the source database: %s", err.Error())
return
@@ -204,12 +206,15 @@ func (app *dataSyncAppImpl) doDataSync(ctx context.Context, sql string, task *en
// 获取源数据库连接
srcConn, err := app.dbApp.GetDbConn(uint64(task.SrcDbId), task.SrcDbName)
defer dbm.PutDbConn(srcConn)
if err != nil {
return syncLog, errorx.NewBiz("failed to connect to the source database: %s", err.Error())
}
// 获取目标数据库连接
targetConn, err := app.dbApp.GetDbConn(uint64(task.TargetDbId), task.TargetDbName)
defer dbm.PutDbConn(targetConn)
if err != nil {
return syncLog, errorx.NewBiz("failed to connect to the target database: %s", err.Error())
}

2
server/internal/db/application/db_sql_exec.go
View File

@@ -5,6 +5,7 @@ import (
"fmt"
"mayfly-go/internal/db/application/dto"
"mayfly-go/internal/db/config"
"mayfly-go/internal/db/dbm"
"mayfly-go/internal/db/dbm/dbi"
"mayfly-go/internal/db/dbm/sqlparser"
"mayfly-go/internal/db/dbm/sqlparser/sqlstmt"
@@ -283,6 +284,7 @@ func (d *dbSqlExecAppImpl) FlowBizHandle(ctx context.Context, bizHandleParam *fl
}
dbConn, err := d.dbApp.GetDbConn(execSqlBizForm.DbId, execSqlBizForm.DbName)
defer dbm.PutDbConn(dbConn)
if err != nil {
return nil, err
}

3
server/internal/db/application/db_transfer.go
View File

@@ -6,6 +6,7 @@ import (
"fmt"
"io"
"mayfly-go/internal/db/application/dto"
"mayfly-go/internal/db/dbm"
"mayfly-go/internal/db/dbm/dbi"
"mayfly-go/internal/db/dbm/sqlparser"
"mayfly-go/internal/db/domain/entity"
@@ -209,6 +210,7 @@ func (app *dbTransferAppImpl) Run(ctx context.Context, taskId uint64, logId uint
// 获取源库连接、目标库连接，判断连接可用性，否则记录日志：xx连接不可用
// 获取源库表信息
srcConn, err := app.dbApp.GetDbConn(uint64(task.SrcDbId), task.SrcDbName)
defer dbm.PutDbConn(srcConn)
if err != nil {
app.EndTransfer(ctx, logId, taskId, "failed to obtain source db connection", err, nil)
return
@@ -247,6 +249,7 @@ func (app *dbTransferAppImpl) Run(ctx context.Context, taskId uint64, logId uint
func (app *dbTransferAppImpl) transfer2Db(ctx context.Context, taskId uint64, logId uint64, task *entity.DbTransferTask, srcConn *dbi.DbConn, start time.Time, tables []dbi.Table) {
// 获取目标库表信息
targetConn, err := app.dbApp.GetDbConn(uint64(task.TargetDbId), task.TargetDbName)
defer dbm.PutDbConn(targetConn)
if err != nil {
app.EndTransfer(ctx, logId, taskId, "failed to get target db connection", err, nil)
return

18
server/internal/db/dbm/dbi/conn.go
View File

@@ -3,9 +3,9 @@ package dbi
import (
"context"
"database/sql"
"errors"
"fmt"
"mayfly-go/internal/machine/mcm"
"mayfly-go/pkg/errorx"
"mayfly-go/pkg/logx"
)
@@ -173,14 +173,18 @@ func (d *DbConn) Close() {
if err := d.db.Close(); err != nil {
logx.Errorf("关闭数据库实例[%s]连接失败: %s", d.Id, err.Error())
}
// 如果是使用了自己实现的ssh隧道转发，则需要手动将其关闭
if d.Info.useSshTunnel {
mcm.CloseSshTunnelMachine(d.Info.SshTunnelMachineId, fmt.Sprintf("db:%d", d.Info.Id))
}
// TODO 关闭实例隧道会影响其他正在使用的连接，所以暂时不关闭
//if d.Info.useSshTunnel {
// mcm.CloseSshTunnelMachine(d.Info.SshTunnelMachineId, fmt.Sprintf("db:%d", d.Info.Id))
//}
d.db = nil
}
}
func (d *DbConn) Ping() error {
return d.db.Ping()
}
// 游标方式遍历查询rows, walkFn error不为nil, 则跳出遍历
func (d *DbConn) walkQueryRows(ctx context.Context, selectSql string, walkFn WalkQueryRowsFunc, args ...any) ([]*QueryColumn, error) {
cancelCtx, cancelFunc := context.WithCancel(ctx)
@@ -238,10 +242,10 @@ func (d *DbConn) walkQueryRows(ctx context.Context, selectSql string, walkFn Wal
// 包装sql执行相关错误
func wrapSqlError(err error) error {
if err == context.Canceled {
if errors.Is(err, context.Canceled) {
return errorx.NewBiz("execution cancel")
}
if err == context.DeadlineExceeded {
if errors.Is(err, context.DeadlineExceeded) {
return errorx.NewBiz("execution timeout")
}
return err

109
server/internal/db/dbm/dbm.go
View File

@@ -9,69 +9,83 @@ import (
_ "mayfly-go/internal/db/dbm/oracle"
_ "mayfly-go/internal/db/dbm/postgres"
_ "mayfly-go/internal/db/dbm/sqlite"
"mayfly-go/internal/machine/mcm"
"mayfly-go/internal/pkg/consts"
"mayfly-go/pkg/cache"
"mayfly-go/pkg/logx"
"sync"
"mayfly-go/pkg/pool"
"time"
)
// 客户端连接缓存，指定时间内没有访问则会被关闭, key为数据库连接id
var connCache = cache.NewTimedCache(consts.DbConnExpireTime, 5*time.Second).
WithUpdateAccessTime(true).
OnEvicted(func(key any, value any) {
logx.Info(fmt.Sprintf("delete db conn cache, id = %s", key))
value.(*dbi.DbConn).Close()
})
var connPool = make(map[string]pool.Pool)
var instPool = make(map[uint64]pool.Pool)
func init() {
mcm.AddCheckSshTunnelMachineUseFunc(func(machineId int) bool {
// 遍历所有db连接实例，若存在db实例使用该ssh隧道机器，则返回true，表示还在使用中...
items := connCache.Items()
for _, v := range items {
if v.Value.(*dbi.DbConn).Info.SshTunnelMachineId == machineId {
return true
}
}
return false
})
}
var mutex sync.Mutex
// PutDbConn 释放连接
func PutDbConn(c *dbi.DbConn) {
if nil == c {
return
}
connId := dbi.GetDbConnId(c.Info.Id, c.Info.Database)
if p, ok := connPool[connId]; ok {
p.Put(c)
}
}
// 从缓存中获取数据库连接信息，若缓存中不存在则会使用回调函数获取dbInfo进行连接并缓存
func GetDbConn(dbId uint64, database string, getDbInfo func() (*dbi.DbInfo, error)) (*dbi.DbConn, error) {
func getPool(dbId uint64, database string, getDbInfo func() (*dbi.DbInfo, error)) (pool.Pool, error) {
connId := dbi.GetDbConnId(dbId, database)
// connId不为空，则为需要缓存
needCache := connId != ""
if needCache {
load, ok := connCache.Get(connId)
if ok {
return load.(*dbi.DbConn), nil
// 获取连接池，如果没有，则创建一个
if p, ok := connPool[connId]; !ok {
var err error
p, err = pool.NewChannelPool(&pool.Config{
InitialCap: 1, //资源池初始连接数
MaxCap: 10, //最大空闲连接数
MaxIdle: 10, //最大并发连接数
IdleTimeout: 10 * time.Minute, // 连接最大空闲时间，过期则失效
Factory: func() (interface{}, error) {
// 若缓存中不存在，则从回调函数中获取DbInfo
dbInfo, err := getDbInfo()
if err != nil {
return nil, err
}
// 连接数据库
return Conn(dbInfo)
},
Close: func(v interface{}) error {
v.(*dbi.DbConn).Close()
return nil
},
Ping: func(v interface{}) error {
return v.(*dbi.DbConn).Ping()
},
})
if err != nil {
return nil, err
}
connPool[connId] = p
instPool[dbId] = p
return p, nil
} else {
return p, nil
}
}
mutex.Lock()
defer mutex.Unlock()
// GetDbConn 从连接池中获取连接信息，记的用完连接后必须调用 PutDbConn 还回池
func GetDbConn(dbId uint64, database string, getDbInfo func() (*dbi.DbInfo, error)) (*dbi.DbConn, error) {
// 若缓存中不存在，则从回调函数中获取DbInfo
dbInfo, err := getDbInfo()
p, err := getPool(dbId, database, getDbInfo)
if err != nil {
return nil, err
}
// 连接数据库
dbConn, err := Conn(dbInfo)
// 从连接池中获取一个可用的连接
c, err := p.Get()
if err != nil {
return nil, err
}
ec := c.(*dbi.DbConn)
return ec, nil
if needCache {
connCache.Put(connId, dbConn)
}
return dbConn, nil
}
// 使用指定dbInfo信息进行连接
@@ -81,16 +95,19 @@ func Conn(di *dbi.DbInfo) (*dbi.DbConn, error) {
// 根据实例id获取连接
func GetDbConnByInstanceId(instanceId uint64) *dbi.DbConn {
for _, connItem := range connCache.Items() {
conn := connItem.Value.(*dbi.DbConn)
if conn.Info.InstanceId == instanceId {
return conn
if p, ok := instPool[instanceId]; ok {
c, err := p.Get()
if err != nil {
logx.Error(fmt.Sprintf("实例id[%d]连接获取失败：%s", instanceId, err))
return nil
}
return c.(*dbi.DbConn)
}
return nil
}
// 删除db缓存并关闭该数据库所有连接
func CloseDb(dbId uint64, db string) {
connCache.Delete(dbi.GetDbConnId(dbId, db))
delete(connPool, dbi.GetDbConnId(dbId, db))
delete(instPool, dbId)
}

7
server/internal/es/api/api.go Normal file
View File

@@ -0,0 +1,7 @@
package api
import "mayfly-go/pkg/ioc"
func InitIoc() {
ioc.Register(new(Instance))
}

170
server/internal/es/api/es_instance.go Normal file
View File

@@ -0,0 +1,170 @@
package api
import (
"github.com/may-fly/cast"
"mayfly-go/internal/es/api/form"
"mayfly-go/internal/es/api/vo"
"mayfly-go/internal/es/application"
"mayfly-go/internal/es/application/dto"
"mayfly-go/internal/es/domain/entity"
"mayfly-go/internal/es/esm/esi"
"mayfly-go/internal/es/imsg"
"mayfly-go/internal/pkg/consts"
tagapp "mayfly-go/internal/tag/application"
tagentity "mayfly-go/internal/tag/domain/entity"
"mayfly-go/pkg/biz"
"mayfly-go/pkg/model"
"mayfly-go/pkg/req"
"mayfly-go/pkg/utils/collx"
"net/http"
"net/url"
"strings"
)
type Instance struct {
inst application.Instance `inject:"T"`
tagApp tagapp.TagTree `inject:"T"`
resourceAuthCertApp tagapp.ResourceAuthCert `inject:"T"`
}
func (d *Instance) ReqConfs() *req.Confs {
reqs := [...]*req.Conf{
// /es/instance 获取实例列表
req.NewGet("", d.Instances),
// /es/instance/test-conn 测试连接
req.NewPost("/test-conn", d.TestConn),
// /es/instance 添加实例
req.NewPost("", d.SaveInstance).Log(req.NewLogSaveI(imsg.LogEsInstSave)),
// /es/instance/:id 删除实例
req.NewDelete(":instanceId", d.DeleteInstance).Log(req.NewLogSaveI(imsg.LogEsInstDelete)),
// /es/instance/proxy 反向代理es接口请求
req.NewAny("/proxy/:instanceId/*path", d.Proxy),
}
return req.NewConfs("/es/instance", reqs[:]...)
}
func (d *Instance) Instances(rc *req.Ctx) {
queryCond := req.BindQuery(rc, new(entity.InstanceQuery))
// 只查询实例，兼容没有录入密码的实例
instTags := d.tagApp.GetAccountTags(rc.GetLoginAccount().Id, &tagentity.TagTreeQuery{
TypePaths: collx.AsArray(tagentity.NewTypePaths(tagentity.TagTypeEsInstance)),
CodePathLikes: collx.AsArray(queryCond.TagPath),
})
// 不存在可操作的数据库，即没有可操作数据
if len(instTags) == 0 {
rc.ResData = model.NewEmptyPageResult[any]()
return
}
dbInstCodes := tagentity.GetCodesByCodePaths(tagentity.TagTypeEsInstance, instTags.GetCodePaths()...)
queryCond.Codes = dbInstCodes
res, err := d.inst.GetPageList(queryCond)
biz.ErrIsNil(err)
resVo := model.PageResultConv[*entity.EsInstance, *vo.InstanceListVO](res)
instvos := resVo.List
// 只查询标签
certTags := d.tagApp.GetAccountTags(rc.GetLoginAccount().Id, &tagentity.TagTreeQuery{
TypePaths: collx.AsArray(tagentity.NewTypePaths(tagentity.TagTypeEsInstance, tagentity.TagTypeAuthCert)),
CodePathLikes: collx.AsArray(queryCond.TagPath),
})
// 填充授权凭证信息
d.resourceAuthCertApp.FillAuthCertByAcNames(tagentity.GetCodesByCodePaths(tagentity.TagTypeAuthCert, certTags.GetCodePaths()...), collx.ArrayMap(instvos, func(vos *vo.InstanceListVO) tagentity.IAuthCert {
return vos
})...)
// 填充标签信息
d.tagApp.FillTagInfo(tagentity.TagType(consts.ResourceTypeEsInstance), collx.ArrayMap(instvos, func(insvo *vo.InstanceListVO) tagentity.ITagResource {
return insvo
})...)
rc.ResData = res
}
func (d *Instance) TestConn(rc *req.Ctx) {
fm := &form.InstanceForm{}
instance := req.BindJsonAndCopyTo[*entity.EsInstance](rc, fm, new(entity.EsInstance))
var ac *tagentity.ResourceAuthCert
if len(fm.AuthCerts) > 0 {
ac = fm.AuthCerts[0]
}
res, err := d.inst.TestConn(instance, ac)
biz.ErrIsNil(err)
rc.ResData = res
}
func (d *Instance) SaveInstance(rc *req.Ctx) {
fm := &form.InstanceForm{}
instance := req.BindJsonAndCopyTo[*entity.EsInstance](rc, fm, new(entity.EsInstance))
rc.ReqParam = fm
id, err := d.inst.SaveInst(rc.MetaCtx, &dto.SaveEsInstance{
EsInstance: instance,
AuthCerts: fm.AuthCerts,
TagCodePaths: fm.TagCodePaths,
})
biz.ErrIsNil(err)
rc.ResData = id
}
func (d *Instance) DeleteInstance(rc *req.Ctx) {
idsStr := rc.PathParam("instanceId")
rc.ReqParam = idsStr
ids := strings.Split(idsStr, ",")
for _, v := range ids {
biz.ErrIsNilAppendErr(d.inst.Delete(rc.MetaCtx, cast.ToUint64(v)), "delete db instance failed: %s")
}
}
func (d *Instance) Proxy(rc *req.Ctx) {
path := rc.PathParam("path")
instanceId := getInstanceId(rc)
// 去掉request中的 id 和 path参数，否则es会报错
r := rc.GetRequest()
_ = RemoveQueryParam(r, "id", "path")
err := d.inst.DoConn(instanceId, func(conn *esi.EsConn) error {
conn.Proxy(rc.GetWriter(), r, path)
return nil
})
biz.ErrIsNil(err)
}
func RemoveQueryParam(req *http.Request, paramNames ...string) error {
parsedURL, err := url.ParseRequestURI(req.RequestURI)
if err != nil {
return err
}
// Get the query parameters
queryParams, err := url.ParseQuery(parsedURL.RawQuery)
if err != nil {
return err
}
// Remove the specified query parameter
for i := range paramNames {
delete(queryParams, paramNames[i])
}
// Reconstruct the query string
parsedURL.RawQuery = queryParams.Encode()
// Update the request URL
req.URL = parsedURL
req.RequestURI = parsedURL.String()
return nil
}
func getInstanceId(rc *req.Ctx) uint64 {
instanceId := rc.PathParamInt("instanceId")
biz.IsTrue(instanceId > 0, "instanceId error")
return uint64(instanceId)
}

18
server/internal/es/api/form/instance.go Normal file
View File

@@ -0,0 +1,18 @@
package form
import (
tagentity "mayfly-go/internal/tag/domain/entity"
)
type InstanceForm struct {
Id uint64 `json:"id"`
Name string `binding:"required" json:"name"`
Host string `binding:"required" json:"host"`
Port int `binding:"required" json:"port"`
Version string `json:"version"`
Remark string `json:"remark"`
SshTunnelMachineId int `json:"sshTunnelMachineId"`
AuthCerts []*tagentity.ResourceAuthCert `json:"authCerts"` // 资产授权凭证信息列表
TagCodePaths []string `binding:"required" json:"tagCodePaths"`
}

31
server/internal/es/api/vo/instance.go Normal file
View File

@@ -0,0 +1,31 @@
package vo
import (
tagentity "mayfly-go/internal/tag/domain/entity"
"time"
)
type InstanceListVO struct {
tagentity.AuthCerts // 授权凭证信息
tagentity.ResourceTags
Id *int64 `json:"id"`
Code string `json:"code"`
Name *string `json:"name"`
Host *string `json:"host"`
Port *int `json:"port"`
Version *string `json:"version"`
CreateTime *time.Time `json:"createTime"`
Creator *string `json:"creator"`
CreatorId *int64 `json:"creatorId"`
UpdateTime *time.Time `json:"updateTime"`
Modifier *string `json:"modifier"`
ModifierId *int64 `json:"modifierId"`
SshTunnelMachineId int `json:"sshTunnelMachineId"`
}
func (i *InstanceListVO) GetCode() string {
return i.Code
}

15
server/internal/es/application/application.go Normal file
View File

@@ -0,0 +1,15 @@
package application
import (
"mayfly-go/pkg/ioc"
"sync"
)
func InitIoc() {
ioc.Register(new(instanceAppImpl), ioc.WithComponentName("EsInstanceApp"))
}
func Init() {
sync.OnceFunc(func() {
})()
}

12
server/internal/es/application/dto/dto.go Normal file
View File

@@ -0,0 +1,12 @@
package dto
import (
"mayfly-go/internal/es/domain/entity"
tagentity "mayfly-go/internal/tag/domain/entity"
)
type SaveEsInstance struct {
EsInstance *entity.EsInstance
AuthCerts []*tagentity.ResourceAuthCert
TagCodePaths []string
}

284
server/internal/es/application/es_instance.go Normal file
View File

@@ -0,0 +1,284 @@
package application
import (
"context"
"fmt"
"mayfly-go/internal/es/application/dto"
"mayfly-go/internal/es/domain/entity"
"mayfly-go/internal/es/domain/repository"
"mayfly-go/internal/es/esm/esi"
"mayfly-go/internal/es/imsg"
"mayfly-go/internal/pkg/consts"
tagapp "mayfly-go/internal/tag/application"
tagdto "mayfly-go/internal/tag/application/dto"
tagentity "mayfly-go/internal/tag/domain/entity"
"mayfly-go/pkg/base"
"mayfly-go/pkg/errorx"
"mayfly-go/pkg/model"
"mayfly-go/pkg/pool"
"mayfly-go/pkg/utils/collx"
"mayfly-go/pkg/utils/stringx"
"mayfly-go/pkg/utils/structx"
"time"
)
type Instance interface {
base.App[*entity.EsInstance]
// GetPageList 分页获取数据库实例
GetPageList(condition *entity.InstanceQuery, orderBy ...string) (*model.PageResult[*entity.EsInstance], error)
// DoConn 获取连接并执行函数
DoConn(instanceId uint64, fn func(*esi.EsConn) error) error
TestConn(instance *entity.EsInstance, ac *tagentity.ResourceAuthCert) (map[string]any, error)
SaveInst(ctx context.Context, d *dto.SaveEsInstance) (uint64, error)
Delete(ctx context.Context, instanceId uint64) error
}
var _ Instance = &instanceAppImpl{}
var connPool = make(map[uint64]pool.Pool)
type instanceAppImpl struct {
base.AppImpl[*entity.EsInstance, repository.EsInstance]
tagApp tagapp.TagTree `inject:"T"`
resourceAuthCertApp tagapp.ResourceAuthCert `inject:"T"`
}
// GetPageList 分页获取数据库实例
func (app *instanceAppImpl) GetPageList(condition *entity.InstanceQuery, orderBy ...string) (*model.PageResult[*entity.EsInstance], error) {
return app.GetRepo().GetInstanceList(condition, orderBy...)
}
func (app *instanceAppImpl) DoConn(instanceId uint64, fn func(*esi.EsConn) error) error {
// 通过实例id获取实例连接信息
p, err := app.getPool(instanceId)
if err != nil {
return err
}
// 从连接池中获取一个可用的连接
c, err := p.Get()
if err != nil {
return err
}
ec := c.(*esi.EsConn)
// 用完后放回连接池
defer p.Put(c)
return fn(ec)
}
func (app *instanceAppImpl) getPool(instanceId uint64) (pool.Pool, error) {
// 获取连接池，如果没有，则创建一个
if p, ok := connPool[instanceId]; !ok {
var err error
p, err = pool.NewChannelPool(&pool.Config{
InitialCap: 1, //资源池初始连接数
MaxCap: 10, //最大空闲连接数
MaxIdle: 10, //最大并发连接数
IdleTimeout: 10 * time.Minute, // 连接最大空闲时间，过期则失效
Factory: func() (interface{}, error) {
return app.createConn(instanceId)
},
Close: func(v interface{}) error {
return v.(*esi.EsConn).Close()
},
Ping: func(v interface{}) error {
return v.(*esi.EsConn).Ping()
},
})
if err != nil {
return nil, err
}
connPool[instanceId] = p
return p, nil
} else {
return p, nil
}
}
func (app *instanceAppImpl) createConn(instanceId uint64) (*esi.EsConn, error) {
// 缓存不存在，则重新连接
instance, err := app.GetById(instanceId)
if err != nil {
return nil, errorx.NewBiz("es instance not found")
}
ei, err := app.ToEsInfo(instance, nil)
if err != nil {
return nil, err
}
ei.CodePath = app.tagApp.ListTagPathByTypeAndCode(int8(tagentity.TagTypeEsInstance), instance.Code)
conn, _, err := ei.Conn()
if err != nil {
return nil, err
}
// 缓存连接信息
return conn, nil
}
func (app *instanceAppImpl) ToEsInfo(instance *entity.EsInstance, ac *tagentity.ResourceAuthCert) (*esi.EsInfo, error) {
ei := new(esi.EsInfo)
ei.InstanceId = instance.Id
structx.Copy(ei, instance)
ei.OriginUrl = fmt.Sprintf("http://%s:%d", instance.Host, instance.Port)
if ac != nil {
if ac.Ciphertext == "" && ac.Name != "" {
ac1, err := app.resourceAuthCertApp.GetAuthCert(ac.Name)
if err == nil {
ac = ac1
}
}
} else {
if instance.Code != "" {
ac2, err := app.resourceAuthCertApp.GetResourceAuthCert(tagentity.TagTypeEsInstance, instance.Code)
if err == nil {
ac = ac2
}
}
}
if ac != nil && ac.Ciphertext != "" {
ei.Username = ac.Username
ei.Password = ac.Ciphertext
}
return ei, nil
}
func (app *instanceAppImpl) TestConn(instance *entity.EsInstance, ac *tagentity.ResourceAuthCert) (map[string]any, error) {
instance.Network = instance.GetNetwork()
ei, err := app.ToEsInfo(instance, ac)
if err != nil {
return nil, err
}
_, res, err := ei.Conn()
if err != nil {
return nil, err
}
return res, nil
}
func (app *instanceAppImpl) SaveInst(ctx context.Context, instance *dto.SaveEsInstance) (uint64, error) {
instanceEntity := instance.EsInstance
// 默认tcp连接
instanceEntity.Network = instanceEntity.GetNetwork()
resourceType := consts.ResourceTypeEsInstance
authCerts := instance.AuthCerts
tagCodePaths := instance.TagCodePaths
// 查找是否存在该库
oldInstance := &entity.EsInstance{
Host: instanceEntity.Host,
Port: instanceEntity.Port,
SshTunnelMachineId: instanceEntity.SshTunnelMachineId,
}
err := app.GetByCond(oldInstance)
if instanceEntity.Id == 0 {
if err == nil {
return 0, errorx.NewBizI(ctx, imsg.ErrEsInstExist)
}
instanceEntity.Code = stringx.Rand(10)
return instanceEntity.Id, app.Tx(ctx, func(ctx context.Context) error {
return app.Insert(ctx, instanceEntity)
}, func(ctx context.Context) error {
return app.resourceAuthCertApp.RelateAuthCert(ctx, &tagdto.RelateAuthCert{
ResourceCode: instanceEntity.Code,
ResourceType: tagentity.TagType(resourceType),
AuthCerts: authCerts,
})
}, func(ctx context.Context) error {
return app.tagApp.SaveResourceTag(ctx, &tagdto.SaveResourceTag{
ResourceTag: app.genEsInstanceResourceTag(instanceEntity, authCerts),
ParentTagCodePaths: tagCodePaths,
})
})
}
// 如果存在该库，则校验修改的库是否为该库
if err == nil {
if oldInstance.Id != instanceEntity.Id {
return 0, errorx.NewBizI(ctx, imsg.ErrEsInstExist)
}
} else {
// 根据host等未查到旧数据，则需要根据id重新获取，因为后续需要使用到code
oldInstance, err = app.GetById(instanceEntity.Id)
if err != nil {
return 0, errorx.NewBiz("db instance not found")
}
}
return oldInstance.Id, app.Tx(ctx, func(ctx context.Context) error {
return app.UpdateById(ctx, instanceEntity)
}, func(ctx context.Context) error {
return app.resourceAuthCertApp.RelateAuthCert(ctx, &tagdto.RelateAuthCert{
ResourceCode: oldInstance.Code,
ResourceType: tagentity.TagType(resourceType),
AuthCerts: authCerts,
})
}, func(ctx context.Context) error {
if instanceEntity.Name != oldInstance.Name {
if err := app.tagApp.UpdateTagName(ctx, tagentity.TagTypeDbInstance, oldInstance.Code, instanceEntity.Name); err != nil {
return err
}
}
return app.tagApp.SaveResourceTag(ctx, &tagdto.SaveResourceTag{
ResourceTag: app.genEsInstanceResourceTag(oldInstance, authCerts),
ParentTagCodePaths: tagCodePaths,
})
})
}
func (app *instanceAppImpl) genEsInstanceResourceTag(ei *entity.EsInstance, authCerts []*tagentity.ResourceAuthCert) *tagdto.ResourceTag {
// 授权证书对应的tag
authCertTags := collx.ArrayMap[*tagentity.ResourceAuthCert, *tagdto.ResourceTag](authCerts, func(val *tagentity.ResourceAuthCert) *tagdto.ResourceTag {
return &tagdto.ResourceTag{
Code: val.Name,
Name: val.Username,
Type: tagentity.TagTypeAuthCert,
}
})
// es实例
return &tagdto.ResourceTag{
Code: ei.Code,
Name: ei.Name,
Type: tagentity.TagTypeEsInstance,
Children: authCertTags,
}
}
func (app *instanceAppImpl) Delete(ctx context.Context, instanceId uint64) error {
instance, err := app.GetById(instanceId)
if err != nil {
return errorx.NewBiz("db instnace not found")
}
return app.Tx(ctx, func(ctx context.Context) error {
// 删除该实例
return app.DeleteById(ctx, instanceId)
}, func(ctx context.Context) error {
// 删除该实例关联的授权凭证信息
return app.resourceAuthCertApp.RelateAuthCert(ctx, &tagdto.RelateAuthCert{
ResourceCode: instance.Code,
ResourceType: tagentity.TagType(consts.ResourceTypeEsInstance),
})
}, func(ctx context.Context) error {
// 删除该实例关联的tag信息
return app.tagApp.DeleteTagByParam(ctx, &tagdto.DelResourceTag{
ResourceCode: instance.Code,
ResourceType: tagentity.TagType(consts.ResourceTypeEsInstance),
})
})
}

36
server/internal/es/domain/entity/es_instance.go Normal file
View File

@@ -0,0 +1,36 @@
package entity
import (
"fmt"
"mayfly-go/pkg/model"
)
type EsInstance struct {
model.Model
Code string `json:"code" gorm:"size:32;not null;"`
Name string `json:"name" gorm:"size:32;not null;"`
Host string `json:"host" gorm:"size:255;not null;"`
Port int `json:"port"`
Network string `json:"network" gorm:"size:20;"`
Version string `json:"version" gorm:"size:50;"`
AuthCertName string `json:"authCertName" gorm:"size:255;"`
SshTunnelMachineId int `json:"sshTunnelMachineId"` // ssh隧道机器id
}
func (d *EsInstance) TableName() string {
return "t_es_instance"
}
// 获取es连接网络, 若没有使用ssh隧道，则直接返回。否则返回拼接的网络需要注册至指定dial
func (d *EsInstance) GetNetwork() string {
network := d.Network
if d.SshTunnelMachineId <= 0 {
if network == "" {
return "tcp"
} else {
return network
}
}
return fmt.Sprintf("es+ssh:%d", d.SshTunnelMachineId)
}

16
server/internal/es/domain/entity/query.go Normal file
View File

@@ -0,0 +1,16 @@
package entity
import "mayfly-go/pkg/model"
// InstanceQuery 数据库实例查询
type InstanceQuery struct {
model.PageParam
Id uint64 `json:"id" form:"id"`
Name string `json:"name" form:"name"`
Code string `json:"code" form:"code"`
Host string `json:"host" form:"host"`
TagPath string `json:"tagPath" form:"tagPath"`
Keyword string `json:"keyword" form:"keyword"`
Codes []string
}

14
server/internal/es/domain/repository/es_instance.go Normal file
View File

@@ -0,0 +1,14 @@
package repository
import (
"mayfly-go/internal/es/domain/entity"
"mayfly-go/pkg/base"
"mayfly-go/pkg/model"
)
type EsInstance interface {
base.Repo[*entity.EsInstance]
// 分页获取数据库实例信息列表
GetInstanceList(condition *entity.InstanceQuery, orderBy ...string) (*model.PageResult[*entity.EsInstance], error)
}

29
server/internal/es/esm/esi/buffer_pool.go Normal file
View File

@@ -0,0 +1,29 @@
package esi
import (
"net/http/httputil"
"sync"
)
type BufferPool struct {
pool *sync.Pool
}
// 需要实现 httputil.BufferPool
var _ httputil.BufferPool = (*BufferPool)(nil)
func NewBufferPool() *BufferPool {
return &BufferPool{&sync.Pool{
New: func() interface{} {
return make([]byte, 32*1024)
},
}}
}
func (b *BufferPool) Get() []byte {
return b.pool.Get().([]byte)
}
func (b *BufferPool) Put(buf []byte) {
b.pool.Put(buf)
}

55
server/internal/es/esm/esi/conn.go Normal file
View File

@@ -0,0 +1,55 @@
package esi
import (
"fmt"
"mayfly-go/internal/machine/mcm"
"mayfly-go/pkg/logx"
"net/http"
"net/http/httputil"
"net/url"
)
type EsConn struct {
Id uint64
Info *EsInfo
proxy *httputil.ReverseProxy
}
// StartProxy 开始代理
func (d *EsConn) StartProxy() error {
// 目标 URL
targetURL, err := url.Parse(d.Info.baseUrl)
if err != nil {
logx.Errorf("Error parsing URL: %v", err)
return err
}
// 创建反向代理
d.proxy = httputil.NewSingleHostReverseProxy(targetURL)
// 设置 proxy buffer pool
d.proxy.BufferPool = NewBufferPool()
return nil
}
func (d *EsConn) Proxy(w http.ResponseWriter, r *http.Request, path string) {
r.URL.Path = path
if d.Info.authorization != "" {
r.Header.Set("Authorization", d.Info.authorization)
}
r.Header.Set("connection", "keep-alive")
r.Header.Set("Accept", "application/json")
d.proxy.ServeHTTP(w, r)
}
func (d *EsConn) Close() error {
// 如果是使用了ssh隧道转发，则需要手动将其关闭
if d.Info.useSshTunnel {
mcm.CloseSshTunnelMachine(uint64(d.Info.SshTunnelMachineId), fmt.Sprintf("es:%d", d.Id))
}
return nil
}
func (d *EsConn) Ping() error {
_, err := d.Info.Ping()
return err
}

142
server/internal/es/esm/esi/es_info.go Normal file
View File

@@ -0,0 +1,142 @@
package esi
import (
"encoding/base64"
"fmt"
machineapp "mayfly-go/internal/machine/application"
"mayfly-go/internal/machine/mcm"
"mayfly-go/pkg/errorx"
"mayfly-go/pkg/httpx"
"mayfly-go/pkg/logx"
"mayfly-go/pkg/model"
"mayfly-go/pkg/utils/structx"
"net/http"
"strings"
)
type EsVersion string
type EsInfo struct {
model.ExtraData // 连接需要的其他额外参数（json字符串），如oracle数据库需要指定sid等
InstanceId uint64 // 实例id
Name string
Host string
Port int
Network string
Username string
Password string
Version EsVersion // 数据库版本信息，用于语法兼容
DefaultVersion bool // 经过查询数据库版本信息后，是否仍然使用默认版本
CodePath []string
SshTunnelMachineId int
useSshTunnel bool // 是否使用系统自己实现的ssh隧道连接,而非库自带的
OriginUrl string // 原始url
baseUrl string // 发起http请求的基本url
authorization string // 发起http请求携带的认证信息
}
// 获取记录日志的描述
func (di *EsInfo) GetLogDesc() string {
return fmt.Sprintf("ES[id=%d, tag=%s, name=%s, ip=%s:%d]", di.InstanceId, di.CodePath, di.Name, di.Host, di.Port)
}
// 连接数据库
func (di *EsInfo) Conn() (*EsConn, map[string]any, error) {
// 使用basic加密用户名和密码
if di.Username != "" && di.Password != "" {
encodeString := base64.StdEncoding.EncodeToString([]byte(fmt.Sprintf("%s:%s", di.Username, di.Password)))
di.authorization = fmt.Sprintf("Basic %s", encodeString)
}
// 使用ssh隧道
err := di.IfUseSshTunnelChangeIpPort()
if err != nil {
logx.Errorf("es ssh failed: %s, err:%s", di.baseUrl, err.Error())
return nil, nil, errorx.NewBiz("es ssh failed: %s", err.Error())
}
// 尝试获取es版本信息，调用接口：get /
res, err := di.Ping()
if err != nil {
logx.Errorf("es ping failed: %s, err:%s", di.baseUrl, err.Error())
return nil, nil, errorx.NewBiz("es ping failed: %s", err.Error())
}
esc := &EsConn{Id: di.InstanceId, Info: di}
err = esc.StartProxy()
if err != nil {
logx.Errorf("es porxy failed: %s, err:%s", di.baseUrl, err.Error())
return nil, nil, err
}
if di.OriginUrl != di.baseUrl {
logx.Infof("es porxy success: %s => %s", di.baseUrl, di.OriginUrl)
} else {
logx.Infof("es porxy success: %s", di.baseUrl)
}
return esc, res, nil
}
func (di *EsInfo) Ping() (map[string]any, error) {
return di.ExecApi("get", "", nil)
}
// ExecApi 执行api
func (di *EsInfo) ExecApi(method, path string, data any, timeoutSecond ...int) (map[string]any, error) {
request := httpx.NewReq(di.baseUrl + path)
if di.authorization != "" {
request.Header("Authorization", di.authorization)
}
if len(timeoutSecond) > 0 { // 设置超时时间
request.Timeout(timeoutSecond[0])
}
switch strings.ToUpper(method) {
case http.MethodGet:
if data != nil {
return request.GetByQuery(structx.ToMap(data)).BodyToMap()
}
return request.Get().BodyToMap()
case http.MethodPost:
return request.PostObj(data).BodyToMap()
case http.MethodPut:
return request.PutObj(data).BodyToMap()
}
return nil, errorx.NewBiz("不支持的请求方法: %s", method)
}
// 如果使用了ssh隧道，将其host port改变其本地映射host port
func (di *EsInfo) IfUseSshTunnelChangeIpPort() error {
// 开启ssh隧道
if di.SshTunnelMachineId > 0 {
stm, err := GetSshTunnel(di.SshTunnelMachineId)
if err != nil {
return err
}
exposedIp, exposedPort, err := stm.OpenSshTunnel(fmt.Sprintf("es:%d", di.InstanceId), di.Host, di.Port)
if err != nil {
return err
}
di.Host = exposedIp
di.Port = exposedPort
di.useSshTunnel = true
di.baseUrl = fmt.Sprintf("http://%s:%d", exposedIp, exposedPort)
} else {
di.baseUrl = fmt.Sprintf("http://%s:%d", di.Host, di.Port)
}
return nil
}
// 根据ssh tunnel机器id返回ssh tunnel
func GetSshTunnel(sshTunnelMachineId int) (*mcm.SshTunnelMachine, error) {
return machineapp.GetMachineApp().GetSshTunnelMachine(sshTunnelMachineId)
}

9
server/internal/es/imsg/en.go Normal file
View File

@@ -0,0 +1,9 @@
package imsg
import "mayfly-go/pkg/i18n"
var En = map[i18n.MsgId]string{
LogEsInstSave: "Es - Save Instance",
LogEsInstDelete: "Es - Delete Instance",
ErrEsInstExist: "The es instance already exists",
}

19
server/internal/es/imsg/imsg.go Normal file
View File

@@ -0,0 +1,19 @@
package imsg
import (
"mayfly-go/internal/pkg/consts"
"mayfly-go/pkg/i18n"
)
func init() {
i18n.AppendLangMsg(i18n.Zh_CN, Zh_CN)
i18n.AppendLangMsg(i18n.En, En)
}
const (
// es inst
LogEsInstDelete = iota + consts.ImsgNumEs
LogEsInstSave
ErrEsInstExist
)

9
server/internal/es/imsg/zh_cn.go Normal file
View File

@@ -0,0 +1,9 @@
package imsg
import "mayfly-go/pkg/i18n"
var Zh_CN = map[i18n.MsgId]string{
LogEsInstSave: "ES-保存实例",
LogEsInstDelete: "ES-删除实例",
ErrEsInstExist: "ES实例已存在",
}

34
server/internal/es/infrastructure/persistence/es_instance.go Normal file
View File

@@ -0,0 +1,34 @@
package persistence
import (
"mayfly-go/internal/es/domain/entity"
"mayfly-go/internal/es/domain/repository"
"mayfly-go/pkg/base"
"mayfly-go/pkg/model"
)
type instanceRepoImpl struct {
base.RepoImpl[*entity.EsInstance]
}
func NewInstanceRepo() repository.EsInstance {
return &instanceRepoImpl{}
}
// 分页获取数据库信息列表
func (d *instanceRepoImpl) GetInstanceList(condition *entity.InstanceQuery, orderBy ...string) (*model.PageResult[*entity.EsInstance], error) {
qd := model.NewCond().
Eq("id", condition.Id).
Eq("host", condition.Host).
Like("name", condition.Name).
Like("code", condition.Code).
In("code", condition.Codes)
keyword := condition.Keyword
if keyword != "" {
keyword = "%" + keyword + "%"
qd.And("host like ? or name like ? or code like ?", keyword, keyword, keyword)
}
return d.PageByCond(qd, condition.PageParam)
}

9
server/internal/es/infrastructure/persistence/persistence.go Normal file
View File

@@ -0,0 +1,9 @@
package persistence
import (
"mayfly-go/pkg/ioc"
)
func InitIoc() {
ioc.Register(NewInstanceRepo(), ioc.WithComponentName("EsInstanceRepo"))
}

18
server/internal/es/init/init.go Normal file
View File

@@ -0,0 +1,18 @@
package init
import (
"mayfly-go/initialize"
"mayfly-go/internal/es/api"
"mayfly-go/internal/es/application"
"mayfly-go/internal/es/infrastructure/persistence"
)
func init() {
initialize.AddInitIocFunc(func() {
persistence.InitIoc()
application.InitIoc()
api.InitIoc()
})
initialize.AddInitFunc(application.Init)
}

41
server/internal/es/readme.md Normal file
View File

@@ -0,0 +1,41 @@
# es 模块开发步骤
## 1、模块设计：
### es实例
- 支持录入es实例：所属标签、ip、端口、账号、密码、ssh跳板机、
### es操作
- 参照db操作，右侧标签树，实例列表，实例下子菜单：
- 索引管理：支持右键菜单：刷新、添加索引、显示系统索引（以.开头的索引名）
- 索引设置：过滤索引名^\..*
- 索引列表：展开索引名列表，以索引名排序，支持右键菜单：复制名字、添加别名、索引迁移、关闭/打开索引、删除索引
- 索引详情：
- 索引增删改查
- 索引迁移：
- 如果 Mapping 中字段已经定义就不能修改其字段的类型等属性了，同时也不能改变分片的数量， 可以使用 Reindex API 来解决这个问题。
- 支持迁移到其他实例的指定索引，默认选中当前实例
- 数据浏览：
- 跳转到：基础搜索、高级搜索
- 基础搜索:
- 保存es查询条件，指定查询名，关联：实例id、索引名
- 可视化组装查询条件
- 加载保存的查询条件列表、删除、修改、应用
- 高级搜索：自己拼接查询json，返回并展示查询结果json
- 仪表盘：一些指标数据：基本信息、节点信息、插件信息、集群状态、集群健康值
- 设置：一些公共设置
## 开发路线
1、后端封装所需接口
参考 src/components/es/api/ClusterApi.ts
- 实例管理接口设计：/es/instance/:实例id/:index/具体接口
- 实例代理接口设计：/es/instance/proxy/:实例id/:官方api接口
2、前端参考es-client相关页面逻辑
参照： https://gitee.com/liuzongyang/es-client

17
server/internal/machine/api/machine.go
View File

@@ -137,6 +137,8 @@ func (m *Machine) SimpleMachieInfo(rc *req.Ctx) {
func (m *Machine) MachineStats(rc *req.Ctx) {
cli, err := m.machineApp.GetCli(GetMachineId(rc))
biz.ErrIsNilAppendErr(err, "connection error: %s")
defer mcm.PutMachineCli(cli)
rc.ResData = cli.GetAllStats()
}
@@ -198,6 +200,8 @@ func (m *Machine) GetProcess(rc *req.Ctx) {
cli, err := m.machineApp.GetCli(GetMachineId(rc))
biz.ErrIsNilAppendErr(err, "connection error: %s")
defer mcm.PutMachineCli(cli)
biz.ErrIsNilAppendErr(m.tagTreeApp.CanAccess(rc.GetLoginAccount().Id, cli.Info.CodePath...), "%s")
res, err := cli.Run(cmd)
@@ -212,6 +216,8 @@ func (m *Machine) KillProcess(rc *req.Ctx) {
cli, err := m.machineApp.GetCli(GetMachineId(rc))
biz.ErrIsNilAppendErr(err, "connection error: %s")
defer mcm.PutMachineCli(cli)
biz.ErrIsNilAppendErr(m.tagTreeApp.CanAccess(rc.GetLoginAccount().Id, cli.Info.CodePath...), "%s")
res, err := cli.Run("sudo kill -9 " + pid)
@@ -221,6 +227,8 @@ func (m *Machine) KillProcess(rc *req.Ctx) {
func (m *Machine) GetUsers(rc *req.Ctx) {
cli, err := m.machineApp.GetCli(GetMachineId(rc))
biz.ErrIsNilAppendErr(err, "connection error: %s")
defer mcm.PutMachineCli(cli)
res, err := cli.GetUsers()
biz.ErrIsNil(err)
rc.ResData = res
@@ -229,6 +237,8 @@ func (m *Machine) GetUsers(rc *req.Ctx) {
func (m *Machine) GetGroups(rc *req.Ctx) {
cli, err := m.machineApp.GetCli(GetMachineId(rc))
biz.ErrIsNilAppendErr(err, "connection error: %s")
defer mcm.PutMachineCli(cli)
res, err := cli.GetGroups()
biz.ErrIsNil(err)
rc.ResData = res
@@ -252,9 +262,12 @@ func (m *Machine) WsSSH(rc *req.Ctx) {
err = req.PermissionHandler(rc)
biz.ErrIsNil(err, mcm.GetErrorContentRn("You do not have permission to operate the machine terminal, please log in again and try again ~"))
cli, err := m.machineApp.NewCli(GetMachineAc(rc))
cli, err := m.machineApp.GetCliByAc(GetMachineAc(rc))
biz.ErrIsNilAppendErr(err, mcm.GetErrorContentRn("connection error: %s"))
defer cli.Close()
defer func() {
cli.Close()
mcm.PutMachineCli(cli)
}()
biz.ErrIsNilAppendErr(m.tagTreeApp.CanAccess(rc.GetLoginAccount().Id, cli.Info.CodePath...), mcm.GetErrorContentRn("%s"))
global.EventBus.Publish(rc.MetaCtx, event.EventTopicResourceOp, cli.Info.CodePath[0])

2
server/internal/machine/api/machine_file.go
View File

@@ -328,6 +328,8 @@ func (m *MachineFile) UploadFolder(rc *req.Ctx) {
folderName := filepath.Dir(paths[0])
mcli, err := m.machineFileApp.GetMachineCli(authCertName)
biz.ErrIsNil(err)
defer mcm.PutMachineCli(mcli)
mi := mcli.Info
sftpCli, err := mcli.GetSftpCli()

3
server/internal/machine/api/machine_script.go
View File

@@ -5,6 +5,7 @@ import (
"mayfly-go/internal/machine/api/vo"
"mayfly-go/internal/machine/application"
"mayfly-go/internal/machine/domain/entity"
"mayfly-go/internal/machine/mcm"
tagapp "mayfly-go/internal/tag/application"
"mayfly-go/pkg/biz"
"mayfly-go/pkg/model"
@@ -79,6 +80,8 @@ func (m *MachineScript) RunMachineScript(rc *req.Ctx) {
}
cli, err := m.machineApp.GetCliByAc(ac)
biz.ErrIsNilAppendErr(err, "connection error: %s")
defer mcm.PutMachineCli(cli)
biz.ErrIsNilAppendErr(m.tagApp.CanAccess(rc.GetLoginAccount().Id, cli.Info.CodePath...), "%s")
res, err := cli.Run(script)

5
server/internal/machine/application/machine.go
View File

@@ -239,11 +239,6 @@ func (m *machineAppImpl) GetCliByAc(authCertName string) (*mcm.Cli, error) {
}
func (m *machineAppImpl) GetCli(machineId uint64) (*mcm.Cli, error) {
cli, err := mcm.GetMachineCliById(machineId)
if err == nil {
return cli, nil
}
_, authCert, err := m.getMachineAndAuthCert(machineId)
if err != nil {
return nil, err

5
server/internal/machine/application/machine_cronjob.go
View File

@@ -5,6 +5,7 @@ import (
"mayfly-go/internal/machine/application/dto"
"mayfly-go/internal/machine/domain/entity"
"mayfly-go/internal/machine/domain/repository"
"mayfly-go/internal/machine/mcm"
tagapp "mayfly-go/internal/tag/application"
tagentity "mayfly-go/internal/tag/domain/entity"
"mayfly-go/pkg/base"
@@ -178,12 +179,14 @@ func (m *machineCronJobAppImpl) runCronJob0(mid uint64, cronJob *entity.MachineC
ExecTime: time.Now(),
}
machineCli, err := m.machineApp.GetCli(uint64(mid))
machineCli, err := m.machineApp.GetCli(mid)
res := ""
if err != nil {
machine, _ := m.machineApp.GetById(mid)
execRes.MachineCode = machine.Code
} else {
defer mcm.PutMachineCli(machineCli)
execRes.MachineCode = machineCli.Info.Code
res, err = machineCli.Run(cronJob.Script)
if err != nil {

9
server/internal/machine/application/machine_file.go
View File

@@ -170,6 +170,8 @@ func (m *machineFileAppImpl) GetDirSize(ctx context.Context, opParam *dto.Machin
if err != nil {
return "", err
}
defer mcm.PutMachineCli(mcli)
res, err := mcli.Run(fmt.Sprintf("du -sh %s", path))
if err != nil {
// 若存在目录为空，则可能会返回如下内容。最后一行即为真正目录内容所占磁盘空间大小
@@ -202,6 +204,8 @@ func (m *machineFileAppImpl) FileStat(ctx context.Context, opParam *dto.MachineF
if err != nil {
return "", err
}
defer mcm.PutMachineCli(mcli)
return mcli.Run(fmt.Sprintf("stat -L %s", path))
}
@@ -379,6 +383,8 @@ func (m *machineFileAppImpl) RemoveFile(ctx context.Context, opParam *dto.Machin
if err != nil {
return nil, err
}
defer mcm.PutMachineCli(mcli)
minfo := mcli.Info
// 优先使用命令删除（速度快），sftp需要递归遍历删除子文件等
@@ -429,6 +435,7 @@ func (m *machineFileAppImpl) Copy(ctx context.Context, opParam *dto.MachineFileO
if err != nil {
return nil, err
}
defer mcm.PutMachineCli(mcli)
mi := mcli.Info
res, err := mcli.Run(fmt.Sprintf("cp -r %s %s", strings.Join(path, " "), toPath))
@@ -458,6 +465,7 @@ func (m *machineFileAppImpl) Mv(ctx context.Context, opParam *dto.MachineFileOp,
if err != nil {
return nil, err
}
defer mcm.PutMachineCli(mcli)
mi := mcli.Info
res, err := mcli.Run(fmt.Sprintf("mv %s %s", strings.Join(path, " "), toPath))
@@ -493,6 +501,7 @@ func (m *machineFileAppImpl) GetMachineSftpCli(opParam *dto.MachineFileOp) (*mcm
if err != nil {
return nil, nil, err
}
defer mcm.PutMachineCli(mcli)
sftpCli, err := mcli.GetSftpCli()
if err != nil {

7
server/internal/machine/mcm/client.go
View File

@@ -18,6 +18,11 @@ type Cli struct {
sftpClient *sftp.Client // sftp客户端
}
func (c *Cli) Ping() error {
_, _, err := c.sshClient.Conn.SendRequest("ping", true, nil)
return err
}
// GetSftpCli 获取sftp client
func (c *Cli) GetSftpCli() (*sftp.Client, error) {
if c.sshClient == nil {
@@ -89,7 +94,7 @@ func (c *Cli) Close() {
}
if sshTunnelMachineId != 0 {
logx.Debugf("close machine ssh tunnel -> machineId=%d, sshTunnelMachineId=%d", m.Id, sshTunnelMachineId)
CloseSshTunnelMachine(int(sshTunnelMachineId), m.GetTunnelId())
CloseSshTunnelMachine(sshTunnelMachineId, m.GetTunnelId())
}
}

147
server/internal/machine/mcm/client_cache.go
View File

@@ -1,122 +1,77 @@
package mcm
import (
"errors"
"mayfly-go/internal/pkg/consts"
tagentity "mayfly-go/internal/tag/domain/entity"
"mayfly-go/pkg/cache"
"mayfly-go/pkg/logx"
"mayfly-go/pkg/pool"
"time"
)
// 机器客户端连接缓存，指定时间内没有访问则会被关闭
var cliCache = cache.NewTimedCache(consts.MachineConnExpireTime, 5*time.Second).
WithUpdateAccessTime(true).
OnEvicted(func(_, value any) {
value.(*Cli).Close()
})
var mcConnPool = make(map[string]pool.Pool)
var mcIdPool = make(map[uint64]pool.Pool)
func init() {
AddCheckSshTunnelMachineUseFunc(func(machineId int) bool {
// 遍历所有机器连接实例，若存在机器连接实例使用该ssh隧道机器，则返回true，表示还在使用中...
items := cliCache.Items()
for _, v := range items {
sshTunnelMachine := v.Value.(*Cli).Info.SshTunnelMachine
if sshTunnelMachine != nil && int(sshTunnelMachine.Id) == machineId {
return true
}
}
func getMcPool(authCertName string, getMachine func(string) (*MachineInfo, error)) (pool.Pool, error) {
// 获取连接池，如果没有，则创建一个
if p, ok := mcConnPool[authCertName]; !ok {
var err error
p, err = pool.NewChannelPool(&pool.Config{
InitialCap: 1, //资源池初始连接数
MaxCap: 10, //最大空闲连接数
MaxIdle: 10, //最大并发连接数
IdleTimeout: 10 * time.Minute, // 连接最大空闲时间，过期则失效
Factory: func() (interface{}, error) {
mi, err := getMachine(authCertName)
if err != nil {
return nil, err
}
mi.Key = authCertName
return mi.Conn()
},
Close: func(v interface{}) error {
v.(*Cli).Close()
return nil
},
Ping: func(v interface{}) error {
return v.(*Cli).Ping()
},
})
if err != nil {
return nil, err
}
return false
})
go checkClientAvailability(3 * time.Minute)
mcConnPool[authCertName] = p
return p, nil
} else {
return p, nil
}
}
func PutMachineCli(c *Cli) {
if nil == c {
return
}
if p, ok := mcConnPool[c.Info.AuthCertName]; ok {
p.Put(c)
}
}
// 从缓存中获取客户端信息，不存在则回调获取机器信息函数，并新建。
// @param 机器的授权凭证名
func GetMachineCli(authCertName string, getMachine func(string) (*MachineInfo, error)) (*Cli, error) {
if load, ok := cliCache.Get(authCertName); ok {
return load.(*Cli), nil
}
mi, err := getMachine(authCertName)
p, err := getMcPool(authCertName, getMachine)
if err != nil {
return nil, err
}
mi.Key = authCertName
c, err := mi.Conn()
// 从连接池中获取一个可用的连接
c, err := p.Get()
if err != nil {
return nil, err
}
cliCache.Put(authCertName, c)
return c, nil
}
// 根据机器id从已连接的机器客户端中获取特权账号连接, 若不存在特权账号，则随机返回一个
func GetMachineCliById(machineId uint64) (*Cli, error) {
// 遍历所有机器连接实例，删除指定机器id关联的连接...
items := cliCache.Items()
var machineCli *Cli
for _, v := range items {
cli := v.Value.(*Cli)
mi := cli.Info
if mi.Id != machineId {
continue
}
machineCli = cli
// 如果是特权账号，则跳出
if mi.AuthCertType == tagentity.AuthCertTypePrivileged {
break
}
}
if machineCli != nil {
return machineCli, nil
}
return nil, errors.New("no connection exists for this machine id")
return c.(*Cli), nil
}
// 删除指定机器缓存客户端，并关闭客户端连接
func DeleteCli(id uint64) {
// 遍历所有机器连接实例，删除指定机器id关联的连接...
items := cliCache.Items()
for _, v := range items {
mi := v.Value.(*Cli).Info
if mi.Id == id {
cliCache.Delete(mi.Key)
}
}
}
// 检查缓存中的客户端是否可用，不可用则关闭客户端连接
func checkClientAvailability(interval time.Duration) {
ticker := time.NewTicker(interval)
defer ticker.Stop()
for range ticker.C {
// 遍历所有机器连接实例，若存在机器连接实例使用该ssh隧道机器，则返回true，表示还在使用中...
items := cliCache.Items()
for _, v := range items {
if v == nil {
continue
}
cli := v.Value.(*Cli)
if cli.Info == nil {
continue
}
if cli.sshClient == nil {
continue
}
if cli.sshClient.Conn == nil {
continue
}
if _, _, err := cli.sshClient.Conn.SendRequest("ping", true, nil); err != nil {
logx.Errorf("machine[%s] cache client is not available: %s", cli.Info.Name, err.Error())
DeleteCli(cli.Info.Id)
}
logx.Debugf("machine[%s] cache client is available", cli.Info.Name)
}
}
delete(mcIdPool, id)
}

2
server/internal/machine/mcm/machine.go
View File

@@ -62,7 +62,7 @@ func (mi *MachineInfo) Conn() (*Cli, error) {
sshClient, err := GetSshClient(mi, nil)
if err != nil {
if mi.UseSshTunnel() {
CloseSshTunnelMachine(int(mi.TempSshMachineId), mi.GetTunnelId())
CloseSshTunnelMachine(mi.TempSshMachineId, mi.GetTunnelId())
}
return nil, err
}

156
server/internal/machine/mcm/sshtunnel.go
View File

@@ -1,59 +1,31 @@
package mcm
import (
"errors"
"fmt"
"io"
"mayfly-go/pkg/logx"
"mayfly-go/pkg/scheduler"
"mayfly-go/pkg/pool"
"mayfly-go/pkg/utils/netx"
"net"
"sync"
"time"
"golang.org/x/crypto/ssh"
)
var (
sshTunnelMachines map[int]*SshTunnelMachine = make(map[int]*SshTunnelMachine)
mutex sync.Mutex
// 所有检测ssh隧道机器是否被使用的函数
checkSshTunnelMachineHasUseFuncs []CheckSshTunnelMachineHasUseFunc
// 是否开启检查ssh隧道机器是否被使用，只有使用到了隧道机器才启用
startCheckSshTunnelHasUse bool = false
tunnelPool = make(map[int]pool.Pool)
)
// 检查ssh隧道机器是否有被使用
type CheckSshTunnelMachineHasUseFunc func(int) bool
func startCheckUse() {
logx.Info("start periodically checking if the ssh tunnel machine is still in use")
// 每十分钟检查一次隧道机器是否还有被使用
scheduler.AddFun("@every 10m", func() {
if !mutex.TryLock() {
return
}
defer mutex.Unlock()
// 遍历隧道机器，都未被使用将会被关闭
for mid, sshTunnelMachine := range sshTunnelMachines {
logx.Debugf("periodically check if the ssh tunnel machine [%d] is still in use...", mid)
hasUse := false
for _, checkUseFunc := range checkSshTunnelMachineHasUseFuncs {
// 如果一个在使用则返回不关闭，不继续后续检查
if checkUseFunc(mid) {
hasUse = true
break
}
}
if !hasUse {
// 都未被使用，则关闭
sshTunnelMachine.Close()
}
}
})
}
// 添加ssh隧道机器检测是否使用函数
func AddCheckSshTunnelMachineUseFunc(checkFunc CheckSshTunnelMachineHasUseFunc) {
if checkSshTunnelMachineHasUseFuncs == nil {
@@ -64,12 +36,18 @@ func AddCheckSshTunnelMachineUseFunc(checkFunc CheckSshTunnelMachineHasUseFunc)
// ssh隧道机器
type SshTunnelMachine struct {
mi *MachineInfo
machineId int // 隧道机器id
SshClient *ssh.Client
mutex sync.Mutex
tunnels map[string]*Tunnel // 隧道id -> 隧道
}
func (stm *SshTunnelMachine) Ping() error {
_, _, err := stm.SshClient.Conn.SendRequest("ping", true, nil)
return err
}
func (stm *SshTunnelMachine) OpenSshTunnel(id string, ip string, port int) (exposedIp string, exposedPort int, err error) {
stm.mutex.Lock()
defer stm.mutex.Unlock()
@@ -77,6 +55,7 @@ func (stm *SshTunnelMachine) OpenSshTunnel(id string, ip string, port int) (expo
tunnel := stm.tunnels[id]
// 已存在该id隧道，则直接返回
if tunnel != nil {
// FIXME 后期改成池化连接，定时60秒检查连接可用性
return tunnel.localHost, tunnel.localPort, nil
}
@@ -85,7 +64,7 @@ func (stm *SshTunnelMachine) OpenSshTunnel(id string, ip string, port int) (expo
return "", 0, err
}
localHost := "0.0.0.0"
localHost := "127.0.0.1"
localAddr := fmt.Sprintf("%s:%d", localHost, localPort)
listener, err := net.Listen("tcp", localAddr)
if err != nil {
@@ -104,13 +83,13 @@ func (stm *SshTunnelMachine) OpenSshTunnel(id string, ip string, port int) (expo
go tunnel.Open(stm.SshClient)
stm.tunnels[tunnel.id] = tunnel
return tunnel.localHost, tunnel.localPort, nil
return localHost, localPort, nil
}
func (st *SshTunnelMachine) GetDialConn(network string, addr string) (net.Conn, error) {
st.mutex.Lock()
defer st.mutex.Unlock()
return st.SshClient.Dial(network, addr)
func (stm *SshTunnelMachine) GetDialConn(network string, addr string) (net.Conn, error) {
stm.mutex.Lock()
defer stm.mutex.Unlock()
return stm.SshClient.Dial(network, addr)
}
func (stm *SshTunnelMachine) Close() {
@@ -131,55 +110,82 @@ func (stm *SshTunnelMachine) Close() {
logx.Errorf("error in closing ssh tunnel machine [%d]: %s", stm.machineId, err.Error())
}
}
delete(sshTunnelMachines, stm.machineId)
delete(tunnelPool, stm.machineId)
}
func getTunnelPool(machineId int, getMachine func(uint64) (*MachineInfo, error)) (pool.Pool, error) {
// 获取连接池，如果没有，则创建一个
if p, ok := tunnelPool[machineId]; !ok {
var err error
p, err = pool.NewChannelPool(&pool.Config{
InitialCap: 1, //资源池初始连接数
MaxCap: 10, //最大空闲连接数
MaxIdle: 10, //最大并发连接数
IdleTimeout: 10 * time.Minute, // 连接最大空闲时间，过期则失效
Factory: func() (interface{}, error) {
mi, err := getMachine(uint64(machineId))
if err != nil {
return nil, err
}
if mi == nil {
return nil, errors.New("error get machine info")
}
sshClient, err := GetSshClient(mi, nil)
if err != nil {
return nil, err
}
stm := &SshTunnelMachine{SshClient: sshClient, machineId: machineId, tunnels: map[string]*Tunnel{}, mi: mi}
logx.Infof("connect to the ssh tunnel machine for the first time[%d][%s:%d]", machineId, mi.Ip, mi.Port)
return stm, err
},
Close: func(v interface{}) error {
v.(*SshTunnelMachine).Close()
return nil
},
Ping: func(v interface{}) error {
return v.(*SshTunnelMachine).Ping()
},
})
if err != nil {
return nil, err
}
tunnelPool[machineId] = p
return p, nil
} else {
return p, nil
}
}
// 获取ssh隧道机器，方便统一管理充当ssh隧道的机器，避免创建多个ssh client
func GetSshTunnelMachine(machineId int, getMachine func(uint64) (*MachineInfo, error)) (*SshTunnelMachine, error) {
mutex.Lock()
defer mutex.Unlock()
sshTunnelMachine := sshTunnelMachines[machineId]
if sshTunnelMachine != nil {
return sshTunnelMachine, nil
p, err := getTunnelPool(machineId, getMachine)
if err != nil {
return nil, err
}
me, err := getMachine(uint64(machineId))
// 从连接池中获取一个可用的连接
c, err := p.Get()
if err != nil {
return nil, err
}
sshClient, err := GetSshClient(me, nil)
if err != nil {
return nil, err
}
sshTunnelMachine = &SshTunnelMachine{SshClient: sshClient, machineId: machineId, tunnels: map[string]*Tunnel{}}
logx.Infof("connect to the ssh tunnel machine for the first time[%d][%s:%d]", machineId, me.Ip, me.Port)
sshTunnelMachines[machineId] = sshTunnelMachine
// 如果实用了隧道机器且还没开始定时检查是否还被实用，则执行定时任务检测隧道是否还被使用
if !startCheckSshTunnelHasUse {
startCheckUse()
startCheckSshTunnelHasUse = true
}
return sshTunnelMachine, nil
return c.(*SshTunnelMachine), nil
}
// 关闭ssh隧道机器的指定隧道
func CloseSshTunnelMachine(machineId int, tunnelId string) {
sshTunnelMachine := sshTunnelMachines[machineId]
if sshTunnelMachine == nil {
return
}
sshTunnelMachine.mutex.Lock()
defer sshTunnelMachine.mutex.Unlock()
t := sshTunnelMachine.tunnels[tunnelId]
if t != nil {
t.Close()
delete(sshTunnelMachine.tunnels, tunnelId)
}
func CloseSshTunnelMachine(machineId uint64, tunnelId string) {
//sshTunnelMachine := mcIdPool[machineId]
//if sshTunnelMachine == nil {
// return
//}
//
//sshTunnelMachine.mutex.Lock()
//defer sshTunnelMachine.mutex.Unlock()
//t := sshTunnelMachine.tunnels[tunnelId]
//if t != nil {
// t.Close()
// delete(sshTunnelMachine.tunnels, tunnelId)
//}
}
type Tunnel struct {

14
server/internal/mongo/api/mongo.go
View File

@@ -8,6 +8,7 @@ import (
"mayfly-go/internal/mongo/application"
"mayfly-go/internal/mongo/domain/entity"
"mayfly-go/internal/mongo/imsg"
"mayfly-go/internal/mongo/mgm"
"mayfly-go/internal/pkg/consts"
tagapp "mayfly-go/internal/tag/application"
tagentity "mayfly-go/internal/tag/domain/entity"
@@ -127,6 +128,8 @@ func (m *Mongo) DeleteMongo(rc *req.Ctx) {
func (m *Mongo) Databases(rc *req.Ctx) {
conn, err := m.mongoApp.GetMongoConn(m.GetMongoId(rc))
biz.ErrIsNil(err)
defer mgm.PutMongoConn(conn)
res, err := conn.Cli.ListDatabases(context.TODO(), bson.D{})
biz.ErrIsNilAppendErr(err, "get mongo dbs error: %s")
rc.ResData = res
@@ -135,6 +138,7 @@ func (m *Mongo) Databases(rc *req.Ctx) {
func (m *Mongo) Collections(rc *req.Ctx) {
conn, err := m.mongoApp.GetMongoConn(m.GetMongoId(rc))
biz.ErrIsNil(err)
defer mgm.PutMongoConn(conn)
global.EventBus.Publish(rc.MetaCtx, event.EventTopicResourceOp, conn.Info.CodePath[0])
@@ -152,6 +156,8 @@ func (m *Mongo) RunCommand(rc *req.Ctx) {
conn, err := m.mongoApp.GetMongoConn(m.GetMongoId(rc))
biz.ErrIsNil(err)
defer mgm.PutMongoConn(conn)
rc.ReqParam = collx.Kvs("mongo", conn.Info, "cmd", commandForm)
// 顺序执行
@@ -181,6 +187,8 @@ func (m *Mongo) FindCommand(rc *req.Ctx) {
conn, err := m.mongoApp.GetMongoConn(m.GetMongoId(rc))
biz.ErrIsNil(err)
defer mgm.PutMongoConn(conn)
cli := conn.Cli
limit := commandForm.Limit
@@ -215,6 +223,8 @@ func (m *Mongo) UpdateByIdCommand(rc *req.Ctx) {
conn, err := m.mongoApp.GetMongoConn(m.GetMongoId(rc))
biz.ErrIsNil(err)
defer mgm.PutMongoConn(conn)
rc.ReqParam = collx.Kvs("mongo", conn.Info, "cmd", commandForm)
// 解析docId文档id，如果为string类型则使用ObjectId解析，解析失败则为普通字符串
@@ -238,6 +248,8 @@ func (m *Mongo) DeleteByIdCommand(rc *req.Ctx) {
conn, err := m.mongoApp.GetMongoConn(m.GetMongoId(rc))
biz.ErrIsNil(err)
defer mgm.PutMongoConn(conn)
rc.ReqParam = collx.Kvs("mongo", conn.Info, "cmd", commandForm)
// 解析docId文档id，如果为string类型则使用ObjectId解析，解析失败则为普通字符串
@@ -260,6 +272,8 @@ func (m *Mongo) InsertOneCommand(rc *req.Ctx) {
conn, err := m.mongoApp.GetMongoConn(m.GetMongoId(rc))
biz.ErrIsNil(err)
defer mgm.PutMongoConn(conn)
rc.ReqParam = collx.Kvs("mongo", conn.Info, "cmd", commandForm)
res, err := conn.Cli.Database(commandForm.Database).Collection(commandForm.Collection).InsertOne(context.TODO(), commandForm.Doc)

104
server/internal/mongo/mgm/conn_cache.go
View File

@@ -1,72 +1,80 @@
package mgm
import (
"mayfly-go/internal/machine/mcm"
"mayfly-go/internal/pkg/consts"
"mayfly-go/pkg/cache"
"mayfly-go/pkg/logx"
"sync"
"context"
"mayfly-go/pkg/pool"
"time"
)
// mongo客户端连接缓存，指定时间内没有访问则会被关闭
var connCache = cache.NewTimedCache(consts.MongoConnExpireTime, 5*time.Second).
WithUpdateAccessTime(true).
OnEvicted(func(key any, value any) {
logx.Infof("删除mongo连接缓存: id = %v", key)
value.(*MongoConn).Close()
})
var connPool = make(map[string]pool.Pool)
func init() {
mcm.AddCheckSshTunnelMachineUseFunc(func(machineId int) bool {
// 遍历所有mongo连接实例，若存在redis实例使用该ssh隧道机器，则返回true，表示还在使用中...
items := connCache.Items()
for _, v := range items {
if v.Value.(*MongoConn).Info.SshTunnelMachineId == machineId {
return true
}
}
return false
})
}
var mutex sync.Mutex
func getPool(mongoId uint64, getMongoInfo func() (*MongoInfo, error)) (pool.Pool, error) {
connId := getConnId(mongoId)
// 获取连接池，如果没有，则创建一个
if p, ok := connPool[connId]; !ok {
var err error
p, err = pool.NewChannelPool(&pool.Config{
InitialCap: 1, //资源池初始连接数
MaxCap: 10, //最大空闲连接数
MaxIdle: 10, //最大并发连接数
IdleTimeout: 10 * time.Minute, // 连接最大空闲时间，过期则失效
Factory: func() (interface{}, error) {
// 若缓存中不存在，则从回调函数中获取MongoInfo
mi, err := getMongoInfo()
if err != nil {
return nil, err
}
// 连接mongo
return mi.Conn()
},
Close: func(v interface{}) error {
v.(*MongoConn).Close()
return nil
},
Ping: func(v interface{}) error {
return v.(*MongoConn).Cli.Ping(context.Background(), nil)
},
})
if err != nil {
return nil, err
}
connPool[connId] = p
return p, nil
} else {
return p, nil
}
}
func PutMongoConn(c *MongoConn) {
if nil == c {
return
}
if p, ok := connPool[getConnId(c.Info.Id)]; ok {
p.Put(c)
}
}
// 从缓存中获取mongo连接信息, 若缓存中不存在则会使用回调函数获取mongoInfo进行连接并缓存
func GetMongoConn(mongoId uint64, getMongoInfo func() (*MongoInfo, error)) (*MongoConn, error) {
connId := getConnId(mongoId)
// connId不为空，则为需要缓存
needCache := connId != ""
if needCache {
load, ok := connCache.Get(connId)
if ok {
return load.(*MongoConn), nil
}
}
mutex.Lock()
defer mutex.Unlock()
// 若缓存中不存在，则从回调函数中获取MongoInfo
mi, err := getMongoInfo()
p, err := getPool(mongoId, getMongoInfo)
if err != nil {
return nil, err
}
// 连接mongo
mc, err := mi.Conn()
// 从连接池中获取一个可用的连接
c, err := p.Get()
if err != nil {
return nil, err
}
if needCache {
connCache.Put(connId, mc)
}
return mc, nil
return c.(*MongoConn), nil
}
// 关闭连接，并移除缓存连接
func CloseConn(mongoId uint64) {
connCache.Delete(mongoId)
connId := getConnId(mongoId)
delete(connPool, connId)
}

2
server/internal/mongo/mgm/info.go
View File

@@ -75,5 +75,5 @@ func getConnId(id uint64) string {
if id == 0 {
return ""
}
return fmt.Sprintf("%d", id)
return fmt.Sprintf("mongo:%d", id)
}

4
server/internal/pkg/consts/consts.go
View File

@@ -9,6 +9,7 @@ const (
DbConnExpireTime = 120 * time.Minute
RedisConnExpireTime = 30 * time.Minute
MongoConnExpireTime = 30 * time.Minute
EsConnExpireTime = 30 * time.Minute
/**** 开发测试使用 ****/
// MachineConnExpireTime = 4 * time.Minute
@@ -20,6 +21,8 @@ const (
ResourceTypeDbInstance int8 = 2
ResourceTypeRedis int8 = 3
ResourceTypeMongo int8 = 4
ResourceTypeAuthCert int8 = 5
ResourceTypeEsInstance int8 = 6
// imsg起始编号
ImsgNumSys = 10000
@@ -31,4 +34,5 @@ const (
ImsgNumRedis = 70000
ImsgNumMongo = 80000
ImsgNumMsg = 90000
ImsgNumEs = 100000
)

5
server/internal/pkg/starter/gorm.go
View File

@@ -46,6 +46,11 @@ func initMysql(m config.Mysql) *gorm.DB {
sqlDB, _ := db.DB()
sqlDB.SetMaxIdleConns(m.MaxIdleConns)
sqlDB.SetMaxOpenConns(m.MaxOpenConns)
// 如果是开发环境时，打印sql语句
if logx.GetConfig().IsDebug() {
db = db.Debug()
}
return db
}
}

5
server/internal/redis/api/redis.go
View File

@@ -152,6 +152,7 @@ func (r *Redis) DeleteRedis(rc *req.Ctx) {
func (r *Redis) RedisInfo(rc *req.Ctx) {
ri, err := r.redisApp.GetRedisConn(uint64(rc.PathParamInt("id")), 0)
biz.ErrIsNil(err)
defer rdm.PutRedisConn(ri)
section := rc.Query("section")
mode := ri.Info.Mode
@@ -229,6 +230,8 @@ func (r *Redis) RedisInfo(rc *req.Ctx) {
func (r *Redis) ClusterInfo(rc *req.Ctx) {
ri, err := r.redisApp.GetRedisConn(uint64(rc.PathParamInt("id")), 0)
biz.ErrIsNil(err)
defer rdm.PutRedisConn(ri)
biz.IsEquals(ri.Info.Mode, rdm.ClusterMode, "non-cluster mode")
info, _ := ri.ClusterCli.ClusterInfo(context.Background()).Result()
nodesStr, _ := ri.ClusterCli.ClusterNodes(context.Background()).Result()
@@ -280,6 +283,8 @@ func (r *Redis) checkKeyAndGetRedisConn(rc *req.Ctx) (*rdm.RedisConn, string) {
func (r *Redis) getRedisConn(rc *req.Ctx) *rdm.RedisConn {
ri, err := r.redisApp.GetRedisConn(getIdAndDbNum(rc))
biz.ErrIsNil(err)
defer rdm.PutRedisConn(ri)
biz.ErrIsNilAppendErr(r.tagApp.CanAccess(rc.GetLoginAccount().Id, ri.Info.CodePath...), "%s")
return ri
}

106
server/internal/redis/rdm/conn_cache.go
View File

@@ -1,73 +1,81 @@
package rdm
import (
"fmt"
"mayfly-go/internal/machine/mcm"
"mayfly-go/internal/pkg/consts"
"mayfly-go/pkg/cache"
"mayfly-go/pkg/logx"
"sync"
"context"
"mayfly-go/pkg/pool"
"time"
)
// redis客户端连接缓存，指定时间内没有访问则会被关闭
var connCache = cache.NewTimedCache(consts.RedisConnExpireTime, 5*time.Second).
WithUpdateAccessTime(true).
OnEvicted(func(key any, value any) {
logx.Info(fmt.Sprintf("remove the redis connection cache id = %s", key))
value.(*RedisConn).Close()
})
func init() {
mcm.AddCheckSshTunnelMachineUseFunc(func(machineId int) bool {
// 遍历所有redis连接实例，若存在redis实例使用该ssh隧道机器，则返回true，表示还在使用中...
items := connCache.Items()
for _, v := range items {
if v.Value.(*RedisConn).Info.SshTunnelMachineId == machineId {
return true
}
}
return false
})
}
var mutex sync.Mutex
var connPool = make(map[string]pool.Pool)
func getPool(redisId uint64, db int, getRedisInfo func() (*RedisInfo, error)) (pool.Pool, error) {
connId := getConnId(redisId, db)
// 获取连接池，如果没有，则创建一个
if p, ok := connPool[connId]; !ok {
var err error
p, err = pool.NewChannelPool(&pool.Config{
InitialCap: 1, //资源池初始连接数
MaxCap: 10, //最大空闲连接数
MaxIdle: 10, //最大并发连接数
IdleTimeout: 10 * time.Minute, // 连接最大空闲时间，过期则失效
Factory: func() (interface{}, error) {
// 若缓存中不存在，则从回调函数中获取RedisInfo
ri, err := getRedisInfo()
if err != nil {
return nil, err
}
// 连接数据库
return ri.Conn()
},
Close: func(v interface{}) error {
v.(*RedisConn).Close()
return nil
},
Ping: func(v interface{}) error {
_, err := v.(*RedisConn).Cli.Ping(context.Background()).Result()
return err
},
})
if err != nil {
return nil, err
}
connPool[connId] = p
return p, nil
} else {
return p, nil
}
}
func PutRedisConn(c *RedisConn) {
if nil == c {
return
}
if p, ok := connPool[getConnId(c.Info.Id, c.Info.Db)]; ok {
p.Put(c)
}
}
// 从缓存中获取redis连接信息, 若缓存中不存在则会使用回调函数获取redisInfo进行连接并缓存
func GetRedisConn(redisId uint64, db int, getRedisInfo func() (*RedisInfo, error)) (*RedisConn, error) {
connId := getConnId(redisId, db)
// connId不为空，则为需要缓存
needCache := connId != ""
if needCache {
load, ok := connCache.Get(connId)
if ok {
return load.(*RedisConn), nil
}
}
mutex.Lock()
defer mutex.Unlock()
// 若缓存中不存在，则从回调函数中获取RedisInfo
ri, err := getRedisInfo()
p, err := getPool(redisId, db, getRedisInfo)
if err != nil {
return nil, err
}
// 连接数据库
rc, err := ri.Conn()
// 从连接池中获取一个可用的连接
c, err := p.Get()
if err != nil {
return nil, err
}
if needCache {
connCache.Put(connId, rc)
}
return rc, nil
// 用完后记的放回连接池
return c.(*RedisConn), nil
}
// 移除redis连接缓存并关闭redis连接
func CloseConn(id uint64, db int) {
connCache.Delete(getConnId(id, db))
delete(connPool, getConnId(id, db))
}

6
server/internal/tag/api/tag_tree.go
View File

@@ -168,9 +168,15 @@ func (p *TagTree) CountTagResource(rc *req.Ctx) {
CodePathLikes: collx.AsArray(tagPath),
}).GetCodePaths()...)
esCodes := entity.GetCodesByCodePaths(entity.TagTypeEsInstance, p.tagTreeApp.GetAccountTags(accountId, &entity.TagTreeQuery{
Types: collx.AsArray(entity.TagTypeEsInstance),
CodePathLikes: collx.AsArray(tagPath),
}).GetCodePaths()...)
rc.ResData = collx.M{
"machine": len(machineCodes),
"db": len(dbCodes),
"es": len(esCodes),
"redis": len(p.tagTreeApp.GetAccountTags(accountId, &entity.TagTreeQuery{
Types: collx.AsArray(entity.TagTypeRedis),
CodePathLikes: collx.AsArray(tagPath),

3
server/internal/tag/domain/entity/tag_tree.go
View File

@@ -32,9 +32,10 @@ const (
TagTypeTag TagType = -1
TagTypeMachine TagType = TagType(consts.ResourceTypeMachine)
TagTypeDbInstance TagType = TagType(consts.ResourceTypeDbInstance) // 数据库实例
TagTypeEsInstance TagType = TagType(consts.ResourceTypeEsInstance) // es实例
TagTypeRedis TagType = TagType(consts.ResourceTypeRedis)
TagTypeMongo TagType = TagType(consts.ResourceTypeMongo)
TagTypeAuthCert TagType = 5 // 授权凭证类型
TagTypeAuthCert TagType = TagType(consts.ResourceTypeAuthCert) // 授权凭证类型
TagTypeDb TagType = 22 // 数据库名
)

1
server/main.go
View File

@@ -4,6 +4,7 @@ import (
_ "mayfly-go/internal/auth/init"
_ "mayfly-go/internal/common/init"
_ "mayfly-go/internal/db/init"
_ "mayfly-go/internal/es/init"
_ "mayfly-go/internal/file/init"
_ "mayfly-go/internal/flow/init"
_ "mayfly-go/internal/machine/init"

115
server/migration/migrations/v1_10.go
View File

@@ -1,15 +1,19 @@
package migrations
import (
flowentity "mayfly-go/internal/flow/domain/entity"
"github.com/go-gormigrate/gormigrate/v2"
"gorm.io/gorm"
esentity "mayfly-go/internal/es/domain/entity"
flowentity "mayfly-go/internal/flow/domain/entity"
sysentity "mayfly-go/internal/sys/domain/entity"
"mayfly-go/pkg/model"
"time"
)
func V1_10() []*gormigrate.Migration {
var migrations []*gormigrate.Migration
migrations = append(migrations, V1_10_0()...)
migrations = append(migrations, V1_10_1()...)
return migrations
}
@@ -22,7 +26,7 @@ func V1_10_0() []*gormigrate.Migration {
&flowentity.Procinst{},
&flowentity.Execution{},
&flowentity.ProcinstTask{},
flowentity.ProcinstTaskCandidate{},
&flowentity.ProcinstTaskCandidate{},
&flowentity.HisProcinstOp{})
if err != nil {
return err
@@ -36,3 +40,108 @@ func V1_10_0() []*gormigrate.Migration {
},
}
}
func V1_10_1() []*gormigrate.Migration {
return []*gormigrate.Migration{
{
ID: "20250422-v1.10.1-es",
Migrate: func(tx *gorm.DB) error {
// 添加实例表
entities := [...]any{
new(esentity.EsInstance),
}
for _, e := range entities {
if err := tx.AutoMigrate(e); err != nil {
return err
}
}
// 添加菜单资源
resources := []*sysentity.Resource{
{
Model: model.Model{CreateModel: model.CreateModel{DeletedModel: model.DeletedModel{IdModel: model.IdModel{Id: 1745292787}}}},
Pid: 0,
UiPath: "lbOU73qg/",
Name: "Elasticsearch",
Code: "/es",
Type: 1,
Meta: `{"icon":"icon es/es-color","isKeepAlive":true,"routeName":"ES"}`,
Weight: 7,
},
{
Model: model.Model{CreateModel: model.CreateModel{DeletedModel: model.DeletedModel{IdModel: model.IdModel{Id: 1745319348}}}},
Pid: 1745292787,
UiPath: "lbOU73qg/gZ2MHF0b/",
Name: "es.instance",
Code: "EsInstance ",
Type: 1,
Meta: `{"component":"ops/es/InstanceList","icon":"icon es/es-color","isKeepAlive":true,"routeName":"EsInstanceList"}`,
Weight: 1745319348,
},
{
Model: model.Model{CreateModel: model.CreateModel{DeletedModel: model.DeletedModel{IdModel: model.IdModel{Id: 1745319410}}}},
Pid: 1745319348,
UiPath: "lbOU73qg/gZ2MHF0b/rcKBdxB5/",
Name: "es.instanceSave",
Code: "es:instance:save",
Type: 2,
Weight: 1745319410,
},
{
Model: model.Model{CreateModel: model.CreateModel{DeletedModel: model.DeletedModel{IdModel: model.IdModel{Id: 1745319424}}}},
Pid: 1745319348,
UiPath: "lbOU73qg/gZ2MHF0b/IMGhLSJK/",
Name: "es.instanceDel",
Code: "es:instance:del",
Type: 2,
Weight: 1745319424,
},
{
Model: model.Model{CreateModel: model.CreateModel{DeletedModel: model.DeletedModel{IdModel: model.IdModel{Id: 1745494931}}}},
Pid: 1745292787,
UiPath: "lbOU73qg/2sDi4isw/",
Name: "es.operation",
Code: "EsOperation",
Type: 1,
Meta: `{"component":"ops/es/EsOperation","icon":"icon es/es-color","isKeepAlive":true,"routeName":"EsOperation"}`,
Weight: 1745319347,
},
{
Model: model.Model{CreateModel: model.CreateModel{DeletedModel: model.DeletedModel{IdModel: model.IdModel{Id: 1745659240}}}},
Pid: 1745494931,
UiPath: "lbOU73qg/2sDi4isw/SQNFhhhn/",
Name: "es.dataSave",
Code: "es:data:save",
Type: 2,
Weight: 1745659240,
},
{
Model: model.Model{CreateModel: model.CreateModel{DeletedModel: model.DeletedModel{IdModel: model.IdModel{Id: 1745659315}}}},
Pid: 1745494931,
UiPath: "lbOU73qg/2sDi4isw/XAgy5Uvp/",
Name: "es.dataDel",
Code: "es:data:del",
Type: 2,
Weight: 1745659315,
},
}
now := time.Now()
for _, res := range resources {
res.Status = 1
res.CreateTime = &now
res.CreatorId = 1
res.Creator = "admin"
res.UpdateTime = &now
res.ModifierId = 1
res.Modifier = "admin"
tx.Create(res)
}
// 给超管授权
return nil
},
Rollback: func(tx *gorm.DB) error {
return nil
},
},
}
}

23
server/pkg/httpx/httpx.go
View File

@@ -54,8 +54,8 @@ func (r *Req) Header(name, value string) *Req {
return r
}
func (r *Req) Timeout(timeout int) *Req {
r.timeout = timeout
func (r *Req) Timeout(second int) *Req {
r.timeout = second
return r
}
@@ -107,6 +107,25 @@ func (r *Req) PostForm(params string) *Resp {
return sendRequest(r)
}
func (r *Req) PutJson(body string) *Resp {
buf := bytes.NewBufferString(body)
r.method = "PUT"
r.body = buf
if r.header == nil {
r.header = make(map[string]string)
}
r.header["Content-type"] = "application/json"
return sendRequest(r)
}
func (r *Req) PutObj(body any) *Resp {
marshal, err := json.Marshal(body)
if err != nil {
return &Resp{err: errors.New("解析json obj错误")}
}
return r.PutJson(string(marshal))
}
func (r *Req) PostMulipart(files []MultipartFile, reqParams collx.M) *Resp {
buf := &bytes.Buffer{}
// 文件写入 buf

216
server/pkg/pool/channel.go Normal file
View File

@@ -0,0 +1,216 @@
package pool
import (
"errors"
"fmt"
"mayfly-go/pkg/logx"
"sync"
"time"
//"reflect"
)
var (
//ErrMaxActiveConnReached 连接池超限
ErrMaxActiveConnReached = errors.New("MaxActiveConnReached")
)
// Config 连接池相关配置
type Config struct {
//连接池中拥有的最小连接数
InitialCap int
//最大并发存活连接数
MaxCap int
//最大空闲连接
MaxIdle int
//生成连接的方法
Factory func() (interface{}, error)
//关闭连接的方法
Close func(interface{}) error
//检查连接是否有效的方法
Ping func(interface{}) error
//连接最大空闲时间，超过该事件则将失效
IdleTimeout time.Duration
}
// channelPool 存放连接信息
type channelPool struct {
mu sync.RWMutex
conns chan *idleConn
factory func() (interface{}, error)
close func(interface{}) error
ping func(interface{}) error
idleTimeout, waitTimeOut time.Duration
maxActive int
openingConns int
}
type idleConn struct {
conn interface{}
t time.Time
}
// NewChannelPool 初始化连接
func NewChannelPool(poolConfig *Config) (Pool, error) {
if !(poolConfig.InitialCap <= poolConfig.MaxIdle && poolConfig.MaxCap >= poolConfig.MaxIdle && poolConfig.InitialCap >= 0) {
return nil, errors.New("invalid capacity settings")
}
if poolConfig.Factory == nil {
return nil, errors.New("invalid factory func settings")
}
if poolConfig.Close == nil {
return nil, errors.New("invalid close func settings")
}
c := &channelPool{
conns: make(chan *idleConn, poolConfig.MaxIdle),
factory: poolConfig.Factory,
close: poolConfig.Close,
idleTimeout: poolConfig.IdleTimeout,
maxActive: poolConfig.MaxCap,
openingConns: poolConfig.InitialCap,
}
if poolConfig.Ping != nil {
c.ping = poolConfig.Ping
}
for i := 0; i < poolConfig.InitialCap; i++ {
conn, err := c.factory()
if err != nil {
c.Release()
return nil, fmt.Errorf("factory is not able to fill the pool: %s", err)
}
c.conns <- &idleConn{conn: conn, t: time.Now()}
}
return c, nil
}
// getConns 获取所有连接
func (c *channelPool) getConns() chan *idleConn {
c.mu.Lock()
conns := c.conns
c.mu.Unlock()
return conns
}
// Get 从pool中取一个连接
func (c *channelPool) Get() (interface{}, error) {
conns := c.getConns()
if conns == nil {
return nil, ErrClosed
}
for {
select {
case wrapConn := <-conns:
if wrapConn == nil {
return nil, ErrClosed
}
//判断是否超时，超时则丢弃
if timeout := c.idleTimeout; timeout > 0 {
if wrapConn.t.Add(timeout).Before(time.Now()) {
//丢弃并关闭该连接
c.Close(wrapConn.conn)
continue
}
}
//判断是否失效，失效则丢弃，如果用户没有设定 ping 方法，就不检查
if c.ping != nil {
if err := c.Ping(wrapConn.conn); err != nil {
c.Close(wrapConn.conn)
continue
}
}
return wrapConn.conn, nil
default:
c.mu.Lock()
logx.Debugf("openConn %v %v", c.openingConns, c.maxActive)
defer c.mu.Unlock()
if c.openingConns >= c.maxActive {
return nil, ErrMaxActiveConnReached
}
if c.factory == nil {
return nil, ErrClosed
}
conn, err := c.factory()
if err != nil {
return nil, err
}
c.openingConns++
return conn, nil
}
}
}
// Put 将连接放回pool中
func (c *channelPool) Put(conn interface{}) error {
if conn == nil {
return errors.New("connection is nil. rejecting")
}
c.mu.Lock()
if c.conns == nil {
c.mu.Unlock()
return c.Close(conn)
}
select {
case c.conns <- &idleConn{conn: conn, t: time.Now()}:
c.mu.Unlock()
return nil
default:
c.mu.Unlock()
//连接池已满，直接关闭该连接
return c.Close(conn)
}
}
// Close 关闭单条连接
func (c *channelPool) Close(conn interface{}) error {
if conn == nil {
return errors.New("connection is nil. rejecting")
}
c.mu.Lock()
defer c.mu.Unlock()
if c.close == nil {
return nil
}
c.openingConns--
return c.close(conn)
}
// Ping 检查单条连接是否有效
func (c *channelPool) Ping(conn interface{}) error {
if conn == nil {
return errors.New("connection is nil. rejecting")
}
return c.ping(conn)
}
// Release 释放连接池中所有连接
func (c *channelPool) Release() {
c.mu.Lock()
conns := c.conns
c.conns = nil
c.factory = nil
c.ping = nil
closeFun := c.close
c.close = nil
c.mu.Unlock()
if conns == nil {
return
}
close(conns)
for wrapConn := range conns {
//log.Printf("Type %v\n",reflect.TypeOf(wrapConn.conn))
_ = closeFun(wrapConn.conn)
}
}
// Len 连接池中已有的连接
func (c *channelPool) Len() int {
return len(c.getConns())
}

21
server/pkg/pool/pool.go Normal file
View File

@@ -0,0 +1,21 @@
package pool
import "errors"
var (
//ErrClosed 连接池已经关闭Error
ErrClosed = errors.New("pool is closed")
)
// Pool 基本方法
type Pool interface {
Get() (interface{}, error)
Put(interface{}) error
Close(interface{}) error
Release()
Len() int
}

9
server/pkg/req/conf.go
View File

@@ -43,6 +43,9 @@ func NewPut(path string, handler HandlerFunc) *Conf {
func NewDelete(path string, handler HandlerFunc) *Conf {
return New("DELETE", path, handler)
}
func NewAny(path string, handler HandlerFunc) *Conf {
return New("any", path, handler)
}
func (r *Conf) ToGinHFunc() gin.HandlerFunc {
return func(c *gin.Context) {
@@ -82,7 +85,11 @@ func (r *Conf) NoRes() *Conf {
// 注册至group
func (r *Conf) Group(gr *gin.RouterGroup) *Conf {
gr.Handle(r.method, r.path, r.ToGinHFunc())
if r.method == "any" {
gr.Any(r.path, r.ToGinHFunc())
} else {
gr.Handle(r.method, r.path, r.ToGinHFunc())
}
return r
}