feat: 新增pgsql数据操作&redis集群操作

2026-02-12 15:55:36 +08:00 · 2022-07-10 12:14:06 +08:00
parent 729a3d7028
commit 1c18a01bf6
30 changed files with 584 additions and 185 deletions
--- a/server/internal/devops/application/db_app.go
+++ b/server/internal/devops/application/db_app.go
@@ -17,6 +17,8 @@ import (
 	"strings"
 	"sync"
 	"time"
+
+	_ "github.com/lib/pq"
 )

 type Db interface {
@@ -228,10 +230,6 @@ func (d *DbInstance) SelectData(execSql string) ([]string, []map[string]interfac
 	if !isSelect && !isShow && !isExplain {
 		return nil, nil, errors.New("该sql非查询语句")
 	}
-	// 没加limit，则默认限制50条
-	if isSelect && !strings.Contains(execSql, "limit") && !strings.Contains(execSql, "LIMIT") {
-		execSql = execSql + " LIMIT 50"
-	}

 	rows, err := d.db.Query(execSql)
 	if err != nil {
@@ -335,8 +333,21 @@ func (d *DbInstance) Close() {

 // 获取dataSourceName
 func getDsn(d *entity.Db) string {
+	var dsn string
 	if d.Type == "mysql" {
-		return fmt.Sprintf("%s:%s@%s(%s:%d)/%s?timeout=8s", d.Username, d.Password, d.Network, d.Host, d.Port, d.Database)
+		dsn = fmt.Sprintf("%s:%s@%s(%s:%d)/%s?timeout=8s", d.Username, d.Password, d.Network, d.Host, d.Port, d.Database)
+		if d.Params != "" {
+			dsn = fmt.Sprintf("%s&%s", dsn, d.Params)
+		}
+		return dsn
+	}
+
+	if d.Type == "postgres" {
+		dsn = fmt.Sprintf("host=%s port=%d user=%s password=%s dbname=%s sslmode=disable", d.Host, d.Port, d.Username, d.Password, d.Database)
+		if d.Params != "" {
+			dsn = fmt.Sprintf("%s %s", dsn, strings.Join(strings.Split(d.Params, "&"), " "))
+		}
+		return dsn
 	}
 	return ""
 }
@@ -370,7 +381,7 @@ const (
 	DEFAULT_COLUMN_SIZE = 2000

 	// mysql 列信息元数据
-	MYSQL_COLOUMN_MA = `SELECT table_name tableName, column_name columnName, column_type columnType,
+	MYSQL_COLUMN_MA = `SELECT table_name tableName, column_name columnName, column_type columnType,
 	column_comment columnComment, column_key columnKey, extra, is_nullable nullable from information_schema.columns
 	WHERE table_name in (%s) AND table_schema = (SELECT database()) ORDER BY tableName, ordinal_position LIMIT %d, %d`

@@ -379,10 +390,58 @@ const (
 	WHERE table_name in (%s) AND table_schema = (SELECT database())`
 )

+const (
+	// postgres 表信息元数据
+	PGSQL_TABLE_MA = `SELECT obj_description(c.oid) AS "tableComment", c.relname AS "tableName" FROM pg_class c 
+	JOIN pg_namespace n ON c.relnamespace = n.oid WHERE n.nspname = (select current_schema()) AND c.reltype > 0`
+
+	PGSQL_TABLE_INFO = `SELECT obj_description(c.oid) AS "tableComment", c.relname AS "tableName" FROM pg_class c 
+	JOIN pg_namespace n ON c.relnamespace = n.oid WHERE n.nspname = (select current_schema()) AND c.reltype > 0`
+
+	PGSQL_INDEX_INFO = `SELECT indexname AS "indexName", indexdef AS "indexComment"
+	FROM pg_indexes WHERE schemaname =  (select current_schema()) AND tablename = '%s'`
+
+	PGSQL_COLUMN_MA = `SELECT
+		C.relname AS "tableName",
+		A.attname AS "columnName",
+		concat_ws ( '', t.typname, SUBSTRING ( format_type ( a.atttypid, a.atttypmod ) FROM '\(.*\)' ) ) AS "columnType",
+		d.description AS "columnComment" 
+	FROM
+		pg_attribute a LEFT JOIN pg_description d ON d.objoid = a.attrelid 
+		AND d.objsubid = A.attnum
+		LEFT JOIN pg_class c ON A.attrelid = c.oid
+		LEFT JOIN pg_namespace pn ON c.relnamespace = pn.oid
+		LEFT JOIN pg_type t ON a.atttypid = t.oid 
+	WHERE
+		A.attnum >= 0 
+		AND pn.nspname = (select current_schema())
+		AND C.relname in (%s)
+	ORDER BY
+		C.relname DESC,
+		A.attnum ASC
+	OFFSET %d LIMIT %d	
+	`
+
+	PGSQL_COLUMN_MA_COUNT = `SELECT COUNT(*) "maNum"
+	FROM
+		pg_attribute a LEFT JOIN pg_description d ON d.objoid = a.attrelid 
+		AND d.objsubid = A.attnum
+		LEFT JOIN pg_class c ON A.attrelid = c.oid
+		LEFT JOIN pg_namespace pn ON c.relnamespace = pn.oid
+		LEFT JOIN pg_type t ON a.atttypid = t.oid 
+	WHERE
+		A.attnum >= 0 
+		AND pn.nspname = (select current_schema())
+		AND C.relname in (%s)
+	`
+)
+
 func (d *DbInstance) GetTableMetedatas() []map[string]interface{} {
 	var sql string
 	if d.Type == "mysql" {
 		sql = MYSQL_TABLE_MA
+	} else if d.Type == "postgres" {
+		sql = PGSQL_TABLE_MA
 	}
 	_, res, _ := d.SelectData(sql)
 	return res
@@ -401,7 +460,10 @@ func (d *DbInstance) GetColumnMetadatas(tableNames ...string) []map[string]inter
 	var sqlTmp string
 	if d.Type == "mysql" {
 		countSqlTmp = MYSQL_COLOUMN_MA_COUNT
-		sqlTmp = MYSQL_COLOUMN_MA
+		sqlTmp = MYSQL_COLUMN_MA
+	} else if d.Type == "postgres" {
+		countSqlTmp = PGSQL_COLUMN_MA_COUNT
+		sqlTmp = PGSQL_COLUMN_MA
 	}

 	countSql := fmt.Sprintf(countSqlTmp, tableName)
@@ -433,6 +495,8 @@ func (d *DbInstance) GetTableInfos() []map[string]interface{} {
 	var sql string
 	if d.Type == "mysql" {
 		sql = MYSQL_TABLE_INFO
+	} else if d.Type == "postgres" {
+		sql = PGSQL_TABLE_INFO
 	}
 	_, res, _ := d.SelectData(sql)
 	return res
@@ -442,6 +506,8 @@ func (d *DbInstance) GetTableIndex(tableName string) []map[string]interface{} {
 	var sql string
 	if d.Type == "mysql" {
 		sql = fmt.Sprintf(MYSQL_INDEX_INFO, tableName)
+	} else if d.Type == "postgres" {
+		sql = fmt.Sprintf(PGSQL_INDEX_INFO, tableName)
 	}
 	_, res, _ := d.SelectData(sql)
 	return res
--- a/server/internal/devops/application/redis_app.go
+++ b/server/internal/devops/application/redis_app.go
@@ -1,6 +1,7 @@
 package application

 import (
+	"context"
 	"fmt"
 	"mayfly-go/internal/devops/domain/entity"
 	"mayfly-go/internal/devops/domain/repository"
@@ -9,9 +10,10 @@ import (
 	"mayfly-go/pkg/cache"
 	"mayfly-go/pkg/global"
 	"mayfly-go/pkg/model"
+	"strings"
 	"time"

-	"github.com/go-redis/redis"
+	"github.com/go-redis/redis/v8"
 )

 type Redis interface {
@@ -102,24 +104,53 @@ func (r *redisAppImpl) GetRedisInstance(id uint64) *RedisInstance {
 	// 缓存不存在，则回调获取redis信息
 	re := r.GetById(id)
 	biz.NotNil(re, "redis信息不存在")
+
+	redisMode := re.Mode
+	ri := &RedisInstance{Id: id, ProjectId: re.ProjectId, Mode: redisMode}
+	if redisMode == "" || redisMode == entity.RedisModeStandalone {
+		rcli := getRedisCient(re)
+		// 测试连接
+		_, e := rcli.Ping(context.Background()).Result()
+		if e != nil {
+			rcli.Close()
+			panic(biz.NewBizErr(fmt.Sprintf("redis连接失败: %s", e.Error())))
+		}
+		ri.Cli = rcli
+	} else if redisMode == entity.RedisModeCluster {
+		ccli := getRedisClusterClient(re)
+		// 测试连接
+		_, e := ccli.Ping(context.Background()).Result()
+		if e != nil {
+			ccli.Close()
+			panic(biz.NewBizErr(fmt.Sprintf("redis集群连接失败: %s", e.Error())))
+		}
+		ri.ClusterCli = ccli
+	}
+
 	global.Log.Infof("连接redis: %s", re.Host)
-
-	rcli := redis.NewClient(&redis.Options{
-		Addr:     re.Host,
-		Password: re.Password, // no password set
-		DB:       re.Db,       // use default DB
-	})
-	// 测试连接
-	_, e := rcli.Ping().Result()
-	biz.ErrIsNilAppendErr(e, "redis连接失败: %s")
-
-	ri := &RedisInstance{Id: id, ProjectId: re.ProjectId, Cli: rcli}
 	if needCache {
 		redisCache.Put(re.Id, ri)
 	}
 	return ri
 }

+func getRedisCient(re *entity.Redis) *redis.Client {
+	return redis.NewClient(&redis.Options{
+		Addr:        re.Host,
+		Password:    re.Password, // no password set
+		DB:          re.Db,       // use default DB
+		DialTimeout: 8 * time.Second,
+	})
+}
+
+func getRedisClusterClient(re *entity.Redis) *redis.ClusterClient {
+	return redis.NewClusterClient(&redis.ClusterOptions{
+		Addrs:       strings.Split(re.Host, ","),
+		Password:    re.Password,
+		DialTimeout: 8 * time.Second,
+	})
+}
+
 //------------------------------------------------------------------------------

 // redis客户端连接缓存，30分钟内没有访问则会被关闭
@@ -127,35 +158,64 @@ var redisCache = cache.NewTimedCache(30*time.Minute, 5*time.Second).
 	WithUpdateAccessTime(true).
 	OnEvicted(func(key interface{}, value interface{}) {
 		global.Log.Info(fmt.Sprintf("删除redis连接缓存 id = %d", key))
-		value.(*RedisInstance).Cli.Close()
+		value.(*RedisInstance).Close()
 	})

-// redis实例
-type RedisInstance struct {
-	Id        uint64
-	ProjectId uint64
-	Cli       *redis.Client
-}
-
 // 移除redis连接缓存并关闭redis连接
 func CloseRedis(id uint64) {
 	redisCache.Delete(id)
 }

 func TestRedisConnection(re *entity.Redis) {
-	rcli := redis.NewClient(&redis.Options{
-		Addr:     re.Host,
-		Password: re.Password, // no password set
-		DB:       re.Db,       // use default DB
-	})
-	defer rcli.Close()
+	var cmd redis.Cmdable
+	if re.Mode == "" || re.Mode == entity.RedisModeStandalone {
+		rcli := getRedisCient(re)
+		defer rcli.Close()
+		cmd = rcli
+	} else if re.Mode == entity.RedisModeCluster {
+		ccli := getRedisClusterClient(re)
+		defer ccli.Close()
+		cmd = ccli
+	}
+
 	// 测试连接
-	_, e := rcli.Ping().Result()
+	_, e := cmd.Ping(context.Background()).Result()
 	biz.ErrIsNilAppendErr(e, "Redis连接失败: %s")
 }

+// redis实例
+type RedisInstance struct {
+	Id         uint64
+	ProjectId  uint64
+	Mode       string
+	Cli        *redis.Client
+	ClusterCli *redis.ClusterClient
+}
+
+// 获取命令执行接口的具体实现
+func (r *RedisInstance) GetCmdable() redis.Cmdable {
+	redisMode := r.Mode
+	if redisMode == "" || redisMode == entity.RedisModeStandalone {
+		return r.Cli
+	}
+	if r.Mode == entity.RedisModeCluster {
+		return r.ClusterCli
+	}
+	return nil
+}
+
 func (r *RedisInstance) Scan(cursor uint64, match string, count int64) ([]string, uint64) {
-	keys, newcursor, err := r.Cli.Scan(cursor, match, count).Result()
+	keys, newcursor, err := r.GetCmdable().Scan(context.Background(), cursor, match, count).Result()
 	biz.ErrIsNilAppendErr(err, "scan失败: %s")
 	return keys, newcursor
 }
+
+func (r *RedisInstance) Close() {
+	if r.Mode == entity.RedisModeStandalone {
+		r.Cli.Close()
+		return
+	}
+	if r.Mode == entity.RedisModeCluster {
+		r.ClusterCli.Close()
+	}
+}