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

refactor: 达梦ssh连接调整

Browse Source
This commit is contained in:
meilin.huang
2023-12-20 23:01:51 +08:00
parent f29a1560aa
commit 550631c03b
10 changed files with 53 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
mayfly_go_web/src/components/pagetable/PageTable.vue
View File

@@ -43,9 +43,14 @@
</div>
<el-form-item style="width: 200px" :key="nowSearchItem.prop">
<SearchFormItem v-if="!nowSearchItem.slot" :item="nowSearchItem" v-model="queryForm_[nowSearchItem.prop]" />
<SearchFormItem
@keyup.enter.native="searchFormItemKeyUpEnter"
v-if="!nowSearchItem.slot"
:item="nowSearchItem"
v-model="queryForm_[nowSearchItem.prop]"
/>
<slot v-else :name="nowSearchItem.slot"></slot>
<slot @keyup.enter.native="searchFormItemKeyUpEnter" v-else :name="nowSearchItem.slot"></slot>
</el-form-item>
</div>
@@ -314,6 +319,11 @@ const calcuTableHeight = () => {
state.tableMaxHeight = window.innerHeight - headerHeight + 'px';
};
const searchFormItemKeyUpEnter = (event: any) => {
event.preventDefault();
search();
};
const formatText = (data: any) => {
state.formatVal = '';
try {

4
mayfly_go_web/src/hooks/usePageTable.ts
View File

@@ -1,6 +1,6 @@
import Api from '@/common/Api';
import { ElMessage } from 'element-plus';
import { reactive, toRefs } from 'vue';
import { reactive, toRefs, toValue } from 'vue';
/**
* @description table 页面操作方法封装
@@ -39,7 +39,7 @@ export const usePageTable = (
if (!api) return;
try {
state.loading = true;
let sp = { ...state.searchParams };
let sp = toValue(state.searchParams);
if (beforeQueryFn) {
sp = beforeQueryFn(sp);
state.searchParams = sp;

5
server/internal/db/dbm/conn.go
View File

@@ -4,6 +4,7 @@ import (
"context"
"database/sql"
"fmt"
"mayfly-go/internal/machine/mcm"
"mayfly-go/pkg/errorx"
"mayfly-go/pkg/logx"
"reflect"
@@ -102,6 +103,10 @@ func (d *DbConn) Close() {
if err := d.db.Close(); err != nil {
logx.Errorf("关闭数据库实例[%s]连接失败: %s", d.Id, err.Error())
}
// 如果是达梦并且使用了ssh隧道，则需要手动将其关闭
if d.Info.Type == DM && d.Info.SshTunnelMachineId > 0 {
mcm.CloseSshTunnelMachine(d.Info.SshTunnelMachineId, fmt.Sprintf("db:%d", d.Info.Id))
}
d.db = nil
}
}

83
server/internal/db/dbm/dialect_dm.go
View File

@@ -4,93 +4,20 @@ import (
"context"
"database/sql"
"fmt"
"io"
machineapp "mayfly-go/internal/machine/application"
"mayfly-go/pkg/errorx"
"mayfly-go/pkg/logx"
"mayfly-go/pkg/utils/anyx"
"net"
"strings"
_ "gitee.com/chunanyong/dm"
)
type ConnectionInfo struct {
Port int
Listener net.Listener
remoteConn net.Conn
}
var connectionMap = make(map[string]ConnectionInfo)
func getLocalListener() (net.Listener, int, error) {
// Setup localListener (type net.Listener)
localListener, err := net.Listen("tcp", "localhost:0")
if err != nil {
return nil, 0, err
}
// 获取本地端口
localPort := localListener.Addr().(*net.TCPAddr).Port
return localListener, localPort, nil
}
func acceptConn(listener net.Listener, sshConn net.Conn) {
for {
localConn, err := listener.Accept()
if err != nil {
logx.Warn("端口转发出错", err)
return
}
go forward(localConn, sshConn)
}
}
func forward(localConn net.Conn, remoteConn net.Conn) {
copyConn := func(writer, reader net.Conn) {
_, err := io.Copy(writer, reader)
if err != nil {
logx.Warnf("io.Copy error: %s", err)
}
}
go copyConn(localConn, remoteConn)
go copyConn(remoteConn, localConn)
}
func openSsh(d *DbInfo) error {
sshTunnelMachine, err := machineapp.GetMachineApp().GetSshTunnelMachine(d.SshTunnelMachineId)
if err != nil {
return err
}
remoteConn, err := sshTunnelMachine.GetDialConn("tcp", fmt.Sprintf("%s:%d", d.Host, d.Port))
if err != nil {
return err
}
// 获取sshConn的本地端口
localLister, localPort, err := getLocalListener()
// defer localLister.Close()
go acceptConn(localLister, remoteConn)
connectionMap[d.Network] = ConnectionInfo{
Port: localPort,
Listener: localLister,
remoteConn: remoteConn,
}
d.Host = "127.0.0.1"
d.Port = localPort
return nil
}
// 创建一个成员变量存放ssh隧道转发对应的本地连接
func getDmDB(d *DbInfo) (*sql.DB, error) {
driverName := "dm"
// SSH Conect 暂时不支持隧道连接
db := d.Database
var dbParam string
if db != "" {
// postgres database可以使用db/schema表示，方便连接指定schema, 若不存在schema则使用默认schema
// dm database可以使用db/schema表示，方便连接指定schema, 若不存在schema则使用默认schema
ss := strings.Split(db, "/")
if len(ss) > 1 {
dbParam = fmt.Sprintf("%s?schema=%s", ss[0], ss[len(ss)-1])
@@ -101,10 +28,16 @@ func getDmDB(d *DbInfo) (*sql.DB, error) {
// 开启ssh隧道
if d.SshTunnelMachineId > 0 {
err := openSsh(d)
sshTunnelMachine, err := machineapp.GetMachineApp().GetSshTunnelMachine(d.SshTunnelMachineId)
if err != nil {
return nil, err
}
exposedIp, exposedPort, err := sshTunnelMachine.OpenSshTunnel(fmt.Sprintf("db:%d", d.Id), d.Host, d.Port)
if err != nil {
return nil, err
}
d.Host = exposedIp
d.Port = exposedPort
}
dsn := fmt.Sprintf("dm://%s:%s@%s:%d/%s", d.Username, d.Password, d.Host, d.Port, dbParam)

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

@@ -105,6 +105,10 @@ func (m *machineAppImpl) Save(ctx context.Context, me *entity.Machine, tagIds ..
if err == nil && oldMachine.Id != me.Id {
return errorx.NewBiz("该机器信息已存在")
}
// 如果调整了ssh username等会查不到旧数据，故需要根据id获取旧信息将code赋值给标签进行关联
if oldMachine.Code == "" {
oldMachine, _ = m.GetById(new(entity.Machine), me.Id)
}
// 关闭连接
mcm.DeleteCli(me.Id)

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

@@ -106,6 +106,6 @@ func (c *Cli) Close() {
}
if c.Info.SshTunnelMachine != nil {
logx.Infof("关闭机器的隧道信息: machineId=%d, sshTunnelMachineId=%d", c.Info.Id, c.Info.SshTunnelMachine.Id)
CloseSshTunnelMachine(int(c.Info.SshTunnelMachine.Id), c.Info.Id)
CloseSshTunnelMachine(int(c.Info.SshTunnelMachine.Id), c.Info.GetTunnelId())
}
}

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

@@ -32,6 +32,10 @@ func (m *MachineInfo) UseSshTunnel() bool {
return m.SshTunnelMachine != nil
}
func (m *MachineInfo) GetTunnelId() string {
return fmt.Sprintf("machine:%d", m.Id)
}
// 连接
func (mi *MachineInfo) Conn() (*Cli, error) {
logx.Infof("[%s]机器连接：%s:%d", mi.Name, mi.Ip, mi.Port)
@@ -46,7 +50,7 @@ func (mi *MachineInfo) Conn() (*Cli, error) {
sshClient, err := GetSshClient(mi)
if err != nil {
if mi.UseSshTunnel() {
CloseSshTunnelMachine(int(mi.SshTunnelMachine.Id), mi.Id)
CloseSshTunnelMachine(int(mi.SshTunnelMachine.Id), mi.GetTunnelId())
}
return nil, err
}
@@ -72,7 +76,7 @@ func (me *MachineInfo) IfUseSshTunnelChangeIpPort() error {
if err != nil {
return err
}
exposeIp, exposePort, err := sshTunnelMachine.OpenSshTunnel(me.Id, me.Ip, me.Port)
exposeIp, exposePort, err := sshTunnelMachine.OpenSshTunnel(me.GetTunnelId(), me.Ip, me.Port)
if err != nil {
return err
}

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

@@ -68,10 +68,10 @@ type SshTunnelMachine struct {
machineId int // 隧道机器id
SshClient *ssh.Client
mutex sync.Mutex
tunnels map[uint64]*Tunnel // 机器id -> 隧道
tunnels map[string]*Tunnel // 隧道id -> 隧道
}
func (stm *SshTunnelMachine) OpenSshTunnel(id uint64, ip string, port int) (exposedIp string, exposedPort int, err error) {
func (stm *SshTunnelMachine) OpenSshTunnel(id string, ip string, port int) (exposedIp string, exposedPort int, err error) {
stm.mutex.Lock()
defer stm.mutex.Unlock()
@@ -154,7 +154,7 @@ func GetSshTunnelMachine(machineId int, getMachine func(uint64) (*MachineInfo, e
if err != nil {
return nil, err
}
sshTunnelMachine = &SshTunnelMachine{SshClient: sshClient, machineId: machineId, tunnels: map[uint64]*Tunnel{}}
sshTunnelMachine = &SshTunnelMachine{SshClient: sshClient, machineId: machineId, tunnels: map[string]*Tunnel{}}
logx.Infof("初次连接ssh隧道机器[%d][%s:%d]", machineId, me.Ip, me.Port)
sshTunnelMachines[machineId] = sshTunnelMachine
@@ -168,7 +168,7 @@ func GetSshTunnelMachine(machineId int, getMachine func(uint64) (*MachineInfo, e
}
// 关闭ssh隧道机器的指定隧道
func CloseSshTunnelMachine(machineId int, tunnelId uint64) {
func CloseSshTunnelMachine(machineId int, tunnelId string) {
sshTunnelMachine := sshTunnelMachines[machineId]
if sshTunnelMachine == nil {
return
@@ -184,7 +184,7 @@ func CloseSshTunnelMachine(machineId int, tunnelId uint64) {
}
type Tunnel struct {
id uint64 // 唯一标识
id string // 唯一标识
machineId int // 隧道机器id
localHost string // 本地监听地址
localPort int // 本地端口
@@ -237,7 +237,7 @@ func (r *Tunnel) Close() {
}
r.remoteConnections = nil
_ = r.listener.Close()
logx.Debugf("隧道 %d 监听器关闭", r.id)
logx.Debugf("隧道 %s 监听器关闭", r.id)
}
func copyConn(writer, reader net.Conn) {

6
server/internal/mongo/application/mongo.go
View File

@@ -77,7 +77,7 @@ func (d *mongoAppImpl) TestConn(me *entity.Mongo) error {
}
func (d *mongoAppImpl) Save(ctx context.Context, m *entity.Mongo, tagIds ...uint64) error {
oldMongo := &entity.Mongo{Name: m.Name}
oldMongo := &entity.Mongo{Name: m.Name, SshTunnelMachineId: m.SshTunnelMachineId}
err := d.GetBy(oldMongo)
if m.Id == 0 {
@@ -99,6 +99,10 @@ func (d *mongoAppImpl) Save(ctx context.Context, m *entity.Mongo, tagIds ...uint
if err == nil && oldMongo.Id != m.Id {
return errorx.NewBiz("该名称已存在")
}
// 如果调整了ssh等会查不到旧数据，故需要根据id获取旧信息将code赋值给标签进行关联
if oldMongo.Code == "" {
oldMongo, _ = d.GetById(new(entity.Mongo), m.Id)
}
// 先关闭连接
mgm.CloseConn(m.Id)

4
server/internal/redis/application/redis.go
View File

@@ -103,6 +103,10 @@ func (r *redisAppImpl) Save(ctx context.Context, re *entity.Redis, tagIds ...uin
rdm.CloseConn(re.Id, db)
}
}
// 如果调整了ssh等会查不到旧数据，故需要根据id获取旧信息将code赋值给标签进行关联
if oldRedis.Code == "" {
oldRedis, _ = r.GetById(new(entity.Redis), re.Id)
}
re.PwdEncrypt()
return r.Tx(ctx, func(ctx context.Context) error {