feat: add optional GORM pool and SQL logging adapter
This commit is contained in:
@@ -2,6 +2,8 @@
|
||||
|
||||
结构化 MySQL 配置、独立 TLS/mTLS、驱动日志和生成项目接入见 [官方 MySQL 适配器](mysql/README.md)。已有 Open/OpenConnector 接口继续支持。
|
||||
|
||||
使用 GORM 的项目可显式接入 [gormx 可选适配器](gormx/README.md),复用本包连接池,并将 SQL 日志接入 logx;普通 database 包不导入 GORM。
|
||||
|
||||
`database` 包提供基于标准库 `database/sql` 的连接池生命周期,不绑定 ORM 或驱动:
|
||||
|
||||
- `Config` / `ConfigFromEnv`:连接池参数和 `GO_OMNI_DB_*` 环境变量。
|
||||
|
||||
104
database/gormx/README.md
Normal file
104
database/gormx/README.md
Normal file
@@ -0,0 +1,104 @@
|
||||
# 可选 GORM 适配器
|
||||
|
||||
从 `v0.14.1-preview` 提供。`gormx` 把 GORM 接到 go-omni 已初始化的 `*sql.DB`,保留一份连接池和一处资源所有权。使用 GORM v2 API(本版验收版本为 `gorm.io/gorm v1.31.2`、MySQL dialect `v1.6.0`)。
|
||||
|
||||
核心 `omni`、`database`、`database/mysql` 不导入此包;普通项目的编译依赖链不包含 GORM。仓库同一 Go module 的 go.mod 会记录可选适配器依赖,使用方显式导入 `database/gormx` 才编译它。本版没有新增默认 ORM 或 `--orm` 脚手架开关。
|
||||
|
||||
## 生成项目接入
|
||||
|
||||
```sh
|
||||
go install git.yuntuo.hk/k2safe/go-omni/cmd/go-omni@v0.14.1-preview
|
||||
go-omni new --database mysql --module example.com/billing ./billing
|
||||
```
|
||||
|
||||
按 [MySQL 配置说明](../mysql/README.md) 设置连接信息、TLS 和池大小。在开发者维护的 `internal/svc/serviceContext.go` 中增加 `ORM *gorm.DB` 字段,并填写现有 Init:
|
||||
|
||||
```go
|
||||
import (
|
||||
"context"
|
||||
"git.yuntuo.hk/k2safe/go-omni/database/gormx"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
// 向现有 ServiceContext 添加 ORM *gorm.DB 字段。
|
||||
// 保留 Runtime/Config/Logger/DB 等现有字段。
|
||||
func (c *ServiceContext) Init(ctx context.Context) error {
|
||||
orm, err := gormx.MySQL(ctx, c.DB, gormx.WithLogger(c.Logger))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
c.ORM = orm
|
||||
return nil
|
||||
}
|
||||
```
|
||||
|
||||
执行 `go mod tidy`。main、runtimeGen、Handler 和路由无需修改;重新生成保留 svc 的字段与 Init。已有项目先 `go get git.yuntuo.hk/k2safe/go-omni@v0.14.1-preview`,同样接入;v0.14.0 生成项目的目录和生命周期契约不变。
|
||||
|
||||
Repository 接收 `*gorm.DB`,查询使用请求 context:
|
||||
|
||||
```go
|
||||
err := repo.db.WithContext(ctx).Where("id = ?", id).First(&item).Error
|
||||
|
||||
err = repo.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||||
// 同一事务内始终使用 tx。
|
||||
return tx.Create(&item).Error
|
||||
})
|
||||
```
|
||||
|
||||
现有模型、Hooks、Scopes、事务、命名策略和错误处理继续由项目维护。共享连接池不会把分别发起的 SQL/GORM 事务自动合并;不要在一个事务内切回根 ORM 或 c.DB 执行本应属于该事务的写入。
|
||||
|
||||
## 连接池与失败处理
|
||||
|
||||
- MySQL 配置、TLS、字符集、会话参数、池大小及健康检查由底层 `database/mysql` 管理,适配器不再设置另一套 DSN 或池参数。
|
||||
- GORM 方言的版本协商继续执行,自动 Ping 关闭以复用已完成的池初始化。初始化查询使用传入 context,并受池的 PingTimeout 限制;可用 `WithInitTimeout(3*time.Second)` 单独指定。
|
||||
- 返回成功后不保留启动 context,每次查询用 `WithContext(ctx)`。查询取消、错误、事务提交/回滚保持 GORM 原生语义。
|
||||
- 对方言初始化失败时 GORM 自动关闭连接池的路径提供保护;错误仍保留 errors.Is/As,原池继续归调用方管理。生成项目的 Init 返回错误时,再由 Runtime 统一回滚该项目资源。
|
||||
- 服务排空后由 go-omni 关闭原池;`orm.DB()` 返回的仍是那一个池。不要给它重复登记 Close,也不要让 Repository 关闭它。
|
||||
|
||||
适配器自身不创建后台 worker、不自动建表、不执行 AutoMigrate、不改写默认事务或全表更新保护,也不替业务重试写入。
|
||||
|
||||
## SQL 日志
|
||||
|
||||
默认借用当前 context 的 logx,也可显式 `WithLogger(c.Logger)`。每个 ORM 可以使用独立日志实例;不会改动进程全局 Logger。
|
||||
|
||||
```go
|
||||
sqlLog := gormx.DefaultLogConfig()
|
||||
sqlLog.Level = "info" // silent / error / warn / info
|
||||
sqlLog.SlowThreshold = 300*time.Millisecond
|
||||
|
||||
orm, err := gormx.MySQL(ctx, c.DB,
|
||||
gormx.WithLogger(c.Logger),
|
||||
gormx.WithLogConfig(sqlLog),
|
||||
gormx.WithConfig(gorm.Config{
|
||||
TranslateError: true,
|
||||
// NamingStrategy 等原生选项可按项目添加。
|
||||
}),
|
||||
)
|
||||
```
|
||||
|
||||
默认记录错误和超过 200 ms 的慢查询,忽略 ErrRecordNotFound 的错误事件;`SlowThreshold: 0` 关闭慢查询事件。事件为 `gorm.query`、`gorm.slowQuery`、`gorm.queryFailed`,携带 durationMs、rowsAffected、sql 和 context 中的 requestId/traceId。GORM 的 `Debug()`/`LogMode` 使用副本,不改变其他请求的日志级别;最终输出还受 logx 自身级别控制。
|
||||
|
||||
默认不插入绑定参数、不记录驱动完整错误文案,只记录错误类别;SQL 正文最多保留 4096 字节,截断后附加省略号。明确开启 `IncludeParameters` / `IncludeErrorDetails` 才输出这些内容。日志不会解析和脱敏直接写在原始 SQL 里的字面量,查询仍应使用占位符。仅需要日志适配的既有 GORM 实例,也可使用 `NewLogger(logger, config)` 返回的原生 `logger.Interface`。
|
||||
|
||||
`WithConfig` 中若显式设置 `gorm.Config.Logger`,则使用项目自己的 Logger,该 Logger 自行负责脱敏和上下文。配置和 Plugins/ClauseBuilders 容器会复制;用户自定义插件、NamingStrategy 等对象仍由用户保证并发安全。
|
||||
|
||||
## 其他方言与高级配置
|
||||
|
||||
MySQL 便捷入口封装了下面的通用入口:
|
||||
|
||||
```go
|
||||
orm, err := gormx.Open(ctx, pool,
|
||||
func(conn gorm.ConnPool) gorm.Dialector {
|
||||
return mysql.New(mysql.Config{Conn: conn})
|
||||
},
|
||||
gormx.WithLogger(logger),
|
||||
)
|
||||
```
|
||||
|
||||
其他方言可由项目提供相同形式的工厂,必须使用传入的 Conn,不能创建或替换连接池。本版真实数据库验收覆盖 MySQL 8.4,不代表其他数据库已经联调。
|
||||
|
||||
不能通过 WithConfig 的 ConnPool/Dialector 覆盖此约定。初始化期间不暴露裸 sql.DB;需要它的自定义插件应在 Open 成功后显式注册。创建额外池或替换 ConnPool 的路由插件不属于本适配器的一池所有权范围。
|
||||
|
||||
PrepareStmt 默认关闭;若显式开启,保留 GORM 的缓存行为,容量、TTL 和缓存清理由项目按 GORM 接口管理。当前固定的 GORM 版本,PreparedStmtDB.Close 只清理语句,不提供停止其缓存清理 goroutine 的公开接口;因此应在服务初始化时创建一次 ORM,避免按请求反复创建。适配器不修改第三方缓存实现,也不宣称解决其生命周期限制。
|
||||
|
||||
参考:[GORM 现有连接池](https://gorm.io/docs/connecting_to_the_database.html#Existing-database-connection)、[GORM Logger](https://gorm.io/docs/logger.html)。本版验收见 [GORM 验收报告](../../reports/gormAcceptance.md)。
|
||||
121
database/gormx/gorm.go
Normal file
121
database/gormx/gorm.go
Normal file
@@ -0,0 +1,121 @@
|
||||
// Package gormx optionally attaches GORM to a pool owned by GoOmni. It never
|
||||
// creates a pool, changes pool limits, migrates tables or closes the owner.
|
||||
package gormx
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"maps"
|
||||
"reflect"
|
||||
"time"
|
||||
|
||||
"git.yuntuo.hk/k2safe/go-omni/database"
|
||||
"git.yuntuo.hk/k2safe/go-omni/logx"
|
||||
gormmysql "gorm.io/driver/mysql"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
// DialectorFactory must use the supplied connection pool. It must not create a
|
||||
// pool or retain/use the ORM before Open succeeds. For example, return
|
||||
// mysql.New(mysql.Config{Conn: conn}). Other dialects can use the same contract.
|
||||
type DialectorFactory func(conn gorm.ConnPool) gorm.Dialector
|
||||
|
||||
type options struct {
|
||||
config gorm.Config
|
||||
logger *logx.Logger
|
||||
logConfig LogConfig
|
||||
timeout time.Duration
|
||||
}
|
||||
type Option func(*options)
|
||||
|
||||
// WithConfig preserves native GORM behavior, including naming, hooks, prepared
|
||||
// statements and transactions. ConnPool/Dialector are supplied by this adapter;
|
||||
// automatic Ping is disabled because the owning pool already passed startup Ping.
|
||||
// A custom Config.Logger takes precedence over the logx adapter.
|
||||
func WithConfig(config gorm.Config) Option { return func(o *options) { o.config = config } }
|
||||
func WithLogger(logger *logx.Logger) Option { return func(o *options) { o.logger = logger } }
|
||||
func WithLogConfig(config LogConfig) Option { return func(o *options) { o.logConfig = config } }
|
||||
func WithInitTimeout(timeout time.Duration) Option { return func(o *options) { o.timeout = timeout } }
|
||||
|
||||
// Error exposes a safe initialization summary while preserving errors.Is/As.
|
||||
type Error struct{ cause error }
|
||||
|
||||
func (e *Error) Error() string { return "gormx: initialization failed" }
|
||||
func (e *Error) Unwrap() error { return e.cause }
|
||||
|
||||
// MySQL attaches GORM's MySQL dialect, including version negotiation, to the
|
||||
// existing pool. Configure TLS, timeouts and connection limits on that pool.
|
||||
func MySQL(ctx context.Context, pool *database.DB, opts ...Option) (*gorm.DB, error) {
|
||||
return Open(ctx, pool, func(conn gorm.ConnPool) gorm.Dialector {
|
||||
return gormmysql.New(gormmysql.Config{Conn: conn})
|
||||
}, opts...)
|
||||
}
|
||||
|
||||
// Open borrows pool. On failure (including dialect initialization failure),
|
||||
// the original pool remains owned by the caller. No ORM Close is required.
|
||||
// The returned ORM does not retain the startup context: each request should use
|
||||
// WithContext(ctx). Never close the sql.DB returned by orm.DB() independently.
|
||||
func Open(ctx context.Context, pool *database.DB, factory DialectorFactory, opts ...Option) (*gorm.DB, error) {
|
||||
if ctx == nil {
|
||||
ctx = context.Background()
|
||||
}
|
||||
if err := ctx.Err(); err != nil {
|
||||
return nil, &Error{cause: err}
|
||||
}
|
||||
if pool == nil || pool.DB == nil || factory == nil {
|
||||
return nil, errors.New("gormx: pool and dialector factory are required")
|
||||
}
|
||||
timeout := pool.PoolConfig().PingTimeout
|
||||
if timeout == 0 {
|
||||
timeout = 5 * time.Second
|
||||
}
|
||||
o := options{timeout: timeout, logConfig: DefaultLogConfig(), logger: logx.FromContext(ctx)}
|
||||
for _, option := range opts {
|
||||
if option != nil {
|
||||
option(&o)
|
||||
}
|
||||
}
|
||||
if o.timeout <= 0 {
|
||||
return nil, errors.New("gormx: initialization timeout must be positive")
|
||||
}
|
||||
if o.config.ConnPool != nil || o.config.Dialector != nil {
|
||||
return nil, errors.New("gormx: configure the existing pool through the dialector factory")
|
||||
}
|
||||
logger, err := NewLogger(o.logger, o.logConfig)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
config := o.config
|
||||
config.Plugins = maps.Clone(config.Plugins)
|
||||
config.ClauseBuilders = maps.Clone(config.ClauseBuilders)
|
||||
if config.Logger == nil {
|
||||
config.Logger = logger
|
||||
}
|
||||
config.DisableAutomaticPing = true
|
||||
startup, cancel := context.WithTimeout(ctx, o.timeout)
|
||||
defer cancel()
|
||||
borrowed := &borrowedPool{db: pool.DB}
|
||||
borrowed.startup.Store(&startupContext{startup})
|
||||
config.ConnPool = borrowed
|
||||
dialect := factory(borrowed)
|
||||
if dialect == nil || (reflect.ValueOf(dialect).Kind() == reflect.Pointer && reflect.ValueOf(dialect).IsNil()) {
|
||||
return nil, errors.New("gormx: dialector factory returned nil")
|
||||
}
|
||||
orm, err := gorm.Open(dialect, &config)
|
||||
if err != nil {
|
||||
return nil, &Error{cause: err}
|
||||
}
|
||||
if err = startup.Err(); err != nil {
|
||||
return nil, &Error{cause: err}
|
||||
}
|
||||
// Prepared statements are GORM's own wrapper over the very same pool.
|
||||
conn := orm.ConnPool
|
||||
if prepared, ok := conn.(*gorm.PreparedStmtDB); ok {
|
||||
conn = prepared.ConnPool
|
||||
}
|
||||
if actual, ok := conn.(*borrowedPool); !ok || actual != borrowed {
|
||||
return nil, errors.New("gormx: dialector or plugin replaced the supplied pool")
|
||||
}
|
||||
borrowed.startup.Store(nil)
|
||||
return orm, nil
|
||||
}
|
||||
173
database/gormx/gorm_test.go
Normal file
173
database/gormx/gorm_test.go
Normal file
@@ -0,0 +1,173 @@
|
||||
package gormx
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql/driver"
|
||||
"errors"
|
||||
"io"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.yuntuo.hk/k2safe/go-omni/database"
|
||||
"git.yuntuo.hk/k2safe/go-omni/logx"
|
||||
"gorm.io/gorm"
|
||||
"gorm.io/gorm/clause"
|
||||
"gorm.io/gorm/schema"
|
||||
)
|
||||
|
||||
type testConnector struct {
|
||||
closed atomic.Int32
|
||||
query func(context.Context) error
|
||||
}
|
||||
|
||||
func (c *testConnector) Connect(context.Context) (driver.Conn, error) {
|
||||
return &testConn{owner: c}, nil
|
||||
}
|
||||
func (c *testConnector) Driver() driver.Driver { return testDriver{} }
|
||||
func (c *testConnector) Close() error { c.closed.Add(1); return nil }
|
||||
|
||||
type testDriver struct{}
|
||||
|
||||
func (testDriver) Open(string) (driver.Conn, error) { return nil, errors.New("unexpected new pool") }
|
||||
|
||||
type testConn struct{ owner *testConnector }
|
||||
|
||||
func (c *testConn) Prepare(string) (driver.Stmt, error) { return nil, errors.New("unused") }
|
||||
func (c *testConn) Close() error { return nil }
|
||||
func (c *testConn) Begin() (driver.Tx, error) { return nil, errors.New("unused") }
|
||||
func (c *testConn) Ping(context.Context) error { return nil }
|
||||
func (c *testConn) QueryContext(ctx context.Context, _ string, _ []driver.NamedValue) (driver.Rows, error) {
|
||||
if c.owner.query != nil {
|
||||
if err := c.owner.query(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
return &testRows{}, nil
|
||||
}
|
||||
|
||||
type testRows struct{ done bool }
|
||||
|
||||
func (r *testRows) Columns() []string { return []string{"version"} }
|
||||
func (r *testRows) Close() error { return nil }
|
||||
func (r *testRows) Next(values []driver.Value) error {
|
||||
if r.done {
|
||||
return io.EOF
|
||||
}
|
||||
r.done = true
|
||||
values[0] = "8.4.0"
|
||||
return nil
|
||||
}
|
||||
func poolFixture(t *testing.T) (*database.DB, *testConnector) {
|
||||
t.Helper()
|
||||
connector := &testConnector{}
|
||||
pool, err := database.OpenConnector(context.Background(), connector, database.DefaultPoolConfig())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = pool.Close() })
|
||||
return pool, connector
|
||||
}
|
||||
|
||||
func TestBorrowedPoolFailureAndStartupDeadline(t *testing.T) {
|
||||
for _, mode := range []string{"driverError", "deadline", "canceled"} {
|
||||
t.Run(mode, func(t *testing.T) {
|
||||
pool, connector := poolFixture(t)
|
||||
cause := errors.New("private-driver-error")
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
connector.query = func(ctx context.Context) error {
|
||||
if mode == "driverError" {
|
||||
return cause
|
||||
}
|
||||
<-ctx.Done()
|
||||
return ctx.Err()
|
||||
}
|
||||
if mode == "canceled" {
|
||||
cancel()
|
||||
cause = context.Canceled
|
||||
} else if mode == "deadline" {
|
||||
cause = context.DeadlineExceeded
|
||||
}
|
||||
start := time.Now()
|
||||
orm, err := MySQL(ctx, pool, WithInitTimeout(20*time.Millisecond))
|
||||
if orm != nil || !errors.Is(err, cause) {
|
||||
t.Fatal("lost cause", err)
|
||||
}
|
||||
if time.Since(start) > time.Second {
|
||||
t.Fatal("startup deadline ignored")
|
||||
}
|
||||
if connector.closed.Load() != 0 {
|
||||
t.Fatal("borrowed pool closed after GORM failure")
|
||||
}
|
||||
if healthErr := pool.Health(context.Background()); healthErr != nil {
|
||||
t.Fatal("owner pool unusable", healthErr)
|
||||
}
|
||||
if err.Error() == cause.Error() {
|
||||
t.Fatal("unsafe initialization error")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestPoolIdentityNativeOptionsAndRequestContext(t *testing.T) {
|
||||
pool, connector := poolFixture(t)
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cfg := gorm.Config{NamingStrategy: schema.NamingStrategy{TablePrefix: "business_"}, ClauseBuilders: map[string]clause.ClauseBuilder{}}
|
||||
logger, _ := logx.New(logx.Config{Output: "discard"})
|
||||
orm, err := MySQL(ctx, pool, WithConfig(cfg), WithLogger(logger))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
cancel() // The startup context must not poison future queries.
|
||||
got, err := orm.DB()
|
||||
if err != nil || got != pool.DB {
|
||||
t.Fatal("pool identity changed", err)
|
||||
}
|
||||
if len(cfg.ClauseBuilders) != 0 || cfg.ConnPool != nil || cfg.Logger != nil {
|
||||
t.Fatal("native input config mutated")
|
||||
}
|
||||
if orm.NamingStrategy.TableName("Item") != "business_items" {
|
||||
t.Fatal("native option lost")
|
||||
}
|
||||
if orm.SkipDefaultTransaction || orm.AllowGlobalUpdate {
|
||||
t.Fatal("transaction safety defaults changed")
|
||||
}
|
||||
ctx, cancel = context.WithCancel(context.Background())
|
||||
cancel()
|
||||
var v string
|
||||
if err = orm.WithContext(ctx).Raw("SELECT VERSION()").Scan(&v).Error; !errors.Is(err, context.Canceled) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err = orm.WithContext(context.Background()).Raw("SELECT VERSION()").Scan(&v).Error; err != nil || v != "8.4.0" {
|
||||
t.Fatal(v, err)
|
||||
}
|
||||
if connector.closed.Load() != 0 {
|
||||
t.Fatal("pool unexpectedly closed")
|
||||
}
|
||||
if err = pool.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err = orm.Raw("SELECT VERSION()").Scan(&v).Error; err == nil {
|
||||
t.Fatal("owner close did not affect GORM")
|
||||
}
|
||||
}
|
||||
|
||||
func TestRejectedOptionsDoNotTouchPool(t *testing.T) {
|
||||
for _, option := range []Option{WithInitTimeout(0), WithLogConfig(LogConfig{Level: "bad"}), WithConfig(gorm.Config{ConnPool: &borrowedPool{}})} {
|
||||
pool, connector := poolFixture(t)
|
||||
if _, err := MySQL(context.Background(), pool, option); err == nil {
|
||||
t.Fatal("invalid option accepted")
|
||||
}
|
||||
if connector.closed.Load() != 0 {
|
||||
t.Fatal("pool closed")
|
||||
}
|
||||
}
|
||||
if _, err := MySQL(nil, nil); err == nil {
|
||||
t.Fatal("nil pool accepted")
|
||||
}
|
||||
pool, _ := poolFixture(t)
|
||||
if _, err := Open(nil, pool, func(gorm.ConnPool) gorm.Dialector { return nil }); err == nil {
|
||||
t.Fatal("nil dialect accepted")
|
||||
}
|
||||
}
|
||||
157
database/gormx/integration_test.go
Normal file
157
database/gormx/integration_test.go
Normal file
@@ -0,0 +1,157 @@
|
||||
package gormx_test
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"net"
|
||||
"os"
|
||||
"strconv"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.yuntuo.hk/k2safe/go-omni/database/gormx"
|
||||
adapter "git.yuntuo.hk/k2safe/go-omni/database/mysql"
|
||||
"git.yuntuo.hk/k2safe/go-omni/logx"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
type record struct {
|
||||
ID int64 `gorm:"primaryKey;autoIncrement:false"`
|
||||
Name string
|
||||
}
|
||||
|
||||
func (record) TableName() string { return "gorm_adapter_records" }
|
||||
|
||||
func TestGORMMySQLTransactionsCancellationAndLogs(t *testing.T) {
|
||||
if os.Getenv("GO_OMNI_CONNECTOR_MYSQL_TEST") != "1" {
|
||||
t.Skip("use scripts/verifyConnector.py --gorm with isolated MySQL")
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
|
||||
defer cancel()
|
||||
cfg := adapter.DefaultConfig()
|
||||
host, port, err := net.SplitHostPort(os.Getenv("GO_OMNI_CONNECTOR_ADDR"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
cfg.Host = host
|
||||
cfg.Port, _ = strconv.Atoi(port)
|
||||
cfg.User = os.Getenv("GO_OMNI_CONNECTOR_USER")
|
||||
cfg.Password = os.Getenv("GO_OMNI_CONNECTOR_PASSWORD")
|
||||
cfg.Database = os.Getenv("GO_OMNI_CONNECTOR_DATABASE")
|
||||
cfg.TLS.CAFile = os.Getenv("GO_OMNI_CONNECTOR_CA_FILE")
|
||||
cfg.TLS.ServerName = "localhost"
|
||||
cfg.Pool.MaxOpenConns = 2
|
||||
cfg.Pool.MaxIdleConns = 2
|
||||
pool, err := adapter.Open(ctx, cfg)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer pool.Close()
|
||||
var logs bytes.Buffer
|
||||
logger, err := logx.New(logx.Config{Writer: &logs})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer logger.Close()
|
||||
logConfig := gormx.DefaultLogConfig()
|
||||
logConfig.Level = "info"
|
||||
orm, err := gormx.MySQL(ctx, pool, gormx.WithLogger(logger), gormx.WithLogConfig(logConfig), gormx.WithConfig(gorm.Config{PrepareStmt: true, TranslateError: true}))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
sqlDB, err := orm.DB()
|
||||
if err != nil || sqlDB != pool.DB {
|
||||
t.Fatal("pool not shared", err)
|
||||
}
|
||||
if sqlDB.Stats().MaxOpenConnections != 2 {
|
||||
t.Fatal("pool limits changed")
|
||||
}
|
||||
var tables int
|
||||
if err = pool.QueryRowContext(ctx, "SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = DATABASE() AND table_name = 'gorm_adapter_records'").Scan(&tables); err != nil || tables != 0 {
|
||||
t.Fatal("adapter must not migrate schema", err)
|
||||
}
|
||||
if _, err = pool.ExecContext(ctx, "CREATE TABLE gorm_adapter_records (id BIGINT PRIMARY KEY, name VARCHAR(100) UNIQUE NOT NULL)"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer pool.ExecContext(context.Background(), "DROP TABLE gorm_adapter_records")
|
||||
requestCtx := logx.WithRequestID(ctx, "gorm-request")
|
||||
orm = orm.WithContext(requestCtx)
|
||||
secret := "private-bound-name"
|
||||
if err = orm.Transaction(func(tx *gorm.DB) error { return tx.Create(&record{ID: 1, Name: secret}).Error }); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
wanted := errors.New("rollback sentinel")
|
||||
if err = orm.Transaction(func(tx *gorm.DB) error {
|
||||
if e := tx.Create(&record{ID: 2, Name: "rollback"}).Error; e != nil {
|
||||
return e
|
||||
}
|
||||
return wanted
|
||||
}); !errors.Is(err, wanted) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
func() {
|
||||
defer func() {
|
||||
if recover() != "rollback panic" {
|
||||
t.Error("panic changed")
|
||||
}
|
||||
}()
|
||||
_ = orm.Transaction(func(tx *gorm.DB) error {
|
||||
if e := tx.Create(&record{ID: 3, Name: "panic"}).Error; e != nil {
|
||||
t.Fatal(e)
|
||||
}
|
||||
panic("rollback panic")
|
||||
})
|
||||
}()
|
||||
var count int64
|
||||
if err = orm.Model(&record{}).Count(&count).Error; err != nil || count != 1 {
|
||||
t.Fatal("commit/rollback differs", count, err)
|
||||
}
|
||||
if err = orm.Transaction(func(tx *gorm.DB) error {
|
||||
if e := tx.Create(&record{ID: 4, Name: "outer"}).Error; e != nil {
|
||||
return e
|
||||
}
|
||||
if e := tx.Transaction(func(inner *gorm.DB) error {
|
||||
if e := inner.Create(&record{ID: 5, Name: "inner"}).Error; e != nil {
|
||||
return e
|
||||
}
|
||||
return wanted
|
||||
}); !errors.Is(e, wanted) {
|
||||
return errors.New("nested rollback failed")
|
||||
}
|
||||
return nil
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err = orm.Model(&record{}).Count(&count).Error; err != nil || count != 2 {
|
||||
t.Fatal(count, err)
|
||||
}
|
||||
if err = orm.Create(&record{ID: 6, Name: secret}).Error; !errors.Is(err, gorm.ErrDuplicatedKey) {
|
||||
t.Fatal("native error translation lost", err)
|
||||
}
|
||||
var missing record
|
||||
if err = orm.First(&missing, "id = ?", 999).Error; !errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
t.Fatal("not-found semantics changed", err)
|
||||
}
|
||||
deadline, stop := context.WithTimeout(requestCtx, 30*time.Millisecond)
|
||||
start := time.Now()
|
||||
err = orm.WithContext(deadline).Exec("SELECT SLEEP(3)").Error
|
||||
stop()
|
||||
if !errors.Is(err, context.DeadlineExceeded) || time.Since(start) > time.Second {
|
||||
t.Fatal("query cancellation lost", err)
|
||||
}
|
||||
if err = pool.Health(ctx); err != nil {
|
||||
t.Fatal("pool did not recover", err)
|
||||
}
|
||||
if strings.Contains(logs.String(), secret) || !strings.Contains(logs.String(), `"requestId":"gorm-request"`) || !strings.Contains(logs.String(), "gorm.queryFailed") {
|
||||
t.Fatal("SQL log privacy/correlation differs")
|
||||
}
|
||||
var cipher struct {
|
||||
VariableName string
|
||||
Value string
|
||||
}
|
||||
if err = orm.Raw("SHOW SESSION STATUS LIKE 'Ssl_cipher'").Scan(&cipher).Error; err != nil || cipher.Value == "" {
|
||||
t.Fatal("GORM bypassed TLS pool", err)
|
||||
}
|
||||
}
|
||||
171
database/gormx/logger.go
Normal file
171
database/gormx/logger.go
Normal file
@@ -0,0 +1,171 @@
|
||||
package gormx
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
"unicode/utf8"
|
||||
|
||||
"git.yuntuo.hk/k2safe/go-omni/logx"
|
||||
"gorm.io/gorm"
|
||||
gormlog "gorm.io/gorm/logger"
|
||||
)
|
||||
|
||||
type LogConfig struct {
|
||||
Level string `json:"level"` // silent, error, warn (default), info
|
||||
SlowThreshold time.Duration `json:"slowThreshold"` // zero disables slow-query events
|
||||
IgnoreRecordNotFound bool `json:"ignoreRecordNotFound"`
|
||||
IncludeParameters bool `json:"includeParameters"`
|
||||
IncludeErrorDetails bool `json:"includeErrorDetails"`
|
||||
MaxSQLBytes int `json:"maxSqlBytes"` // zero uses 4096; bounds emitted SQL text
|
||||
}
|
||||
|
||||
func DefaultLogConfig() LogConfig {
|
||||
return LogConfig{Level: "warn", SlowThreshold: 200 * time.Millisecond, IgnoreRecordNotFound: true, MaxSQLBytes: 4096}
|
||||
}
|
||||
func (c LogConfig) Validate() error {
|
||||
switch c.Level {
|
||||
case "", "silent", "error", "warn", "info":
|
||||
default:
|
||||
return errors.New("gormx: invalid SQL log level")
|
||||
}
|
||||
if c.SlowThreshold < 0 || c.MaxSQLBytes < 0 {
|
||||
return errors.New("gormx: SQL log limits cannot be negative")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
type sqlLogger struct {
|
||||
logger *logx.Logger
|
||||
config LogConfig
|
||||
level gormlog.LogLevel
|
||||
}
|
||||
|
||||
// NewLogger adapts logx without changing its level, globals or ownership.
|
||||
// Bound SQL arguments and raw driver error text are omitted by default. SQL
|
||||
// literals written directly into a query are not parsed or redacted.
|
||||
func NewLogger(logger *logx.Logger, config LogConfig) (gormlog.Interface, error) {
|
||||
if err := config.Validate(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if config.MaxSQLBytes == 0 {
|
||||
config.MaxSQLBytes = 4096
|
||||
}
|
||||
level := gormlog.Warn
|
||||
switch config.Level {
|
||||
case "silent":
|
||||
level = gormlog.Silent
|
||||
case "error":
|
||||
level = gormlog.Error
|
||||
case "info":
|
||||
level = gormlog.Info
|
||||
}
|
||||
return &sqlLogger{logger: logger, config: config, level: level}, nil
|
||||
}
|
||||
func (l *sqlLogger) LogMode(level gormlog.LogLevel) gormlog.Interface {
|
||||
copy := *l
|
||||
if level < gormlog.Silent || level > gormlog.Info {
|
||||
level = gormlog.Silent
|
||||
}
|
||||
copy.level = level
|
||||
return ©
|
||||
}
|
||||
func (l *sqlLogger) output(ctx context.Context) *logx.Logger {
|
||||
if l.logger != nil {
|
||||
return l.logger
|
||||
}
|
||||
return logx.FromContext(ctx)
|
||||
}
|
||||
func (l *sqlLogger) message(msg string, args []any) string {
|
||||
if !l.config.IncludeErrorDetails {
|
||||
copy := append([]any(nil), args...)
|
||||
for i, v := range copy {
|
||||
if e, ok := v.(error); ok {
|
||||
copy[i] = errorKind(e)
|
||||
}
|
||||
}
|
||||
args = copy
|
||||
}
|
||||
return clip(fmt.Sprintf(msg, args...), l.config.MaxSQLBytes)
|
||||
}
|
||||
func (l *sqlLogger) Info(ctx context.Context, msg string, args ...any) {
|
||||
if l.level >= gormlog.Info {
|
||||
l.output(ctx).InfoContext(ctx, "gorm.message", "message", l.message(msg, args))
|
||||
}
|
||||
}
|
||||
func (l *sqlLogger) Warn(ctx context.Context, msg string, args ...any) {
|
||||
if l.level >= gormlog.Warn {
|
||||
l.output(ctx).WarnContext(ctx, "gorm.message", "message", l.message(msg, args))
|
||||
}
|
||||
}
|
||||
func (l *sqlLogger) Error(ctx context.Context, msg string, args ...any) {
|
||||
if l.level >= gormlog.Error {
|
||||
l.output(ctx).ErrorContext(ctx, "gorm.message", "message", l.message(msg, args))
|
||||
}
|
||||
}
|
||||
|
||||
func (l *sqlLogger) ParamsFilter(_ context.Context, query string, args ...any) (string, []any) {
|
||||
if !l.config.IncludeParameters {
|
||||
return query, nil
|
||||
}
|
||||
return query, args
|
||||
}
|
||||
func (l *sqlLogger) Trace(ctx context.Context, begin time.Time, fc func() (string, int64), err error) {
|
||||
if l.level == gormlog.Silent {
|
||||
return
|
||||
}
|
||||
elapsed := time.Since(begin)
|
||||
failure := err != nil && !(l.config.IgnoreRecordNotFound && errors.Is(err, gorm.ErrRecordNotFound))
|
||||
slow := l.config.SlowThreshold > 0 && elapsed > l.config.SlowThreshold
|
||||
level := gormlog.Info
|
||||
if failure && l.level >= gormlog.Error {
|
||||
level = gormlog.Error
|
||||
} else if slow && l.level >= gormlog.Warn {
|
||||
level = gormlog.Warn
|
||||
} else if l.level < gormlog.Info {
|
||||
return
|
||||
}
|
||||
query, rows := fc()
|
||||
attrs := []any{"durationMs", float64(elapsed) / float64(time.Millisecond), "rowsAffected", rows, "sql", clip(query, l.config.MaxSQLBytes)}
|
||||
if level == gormlog.Error {
|
||||
attrs = append(attrs, "errorKind", errorKind(err))
|
||||
if l.config.IncludeErrorDetails {
|
||||
attrs = append(attrs, "error", clip(err.Error(), l.config.MaxSQLBytes))
|
||||
}
|
||||
}
|
||||
logger := l.output(ctx)
|
||||
switch level {
|
||||
case gormlog.Error:
|
||||
logger.ErrorContext(ctx, "gorm.queryFailed", attrs...)
|
||||
case gormlog.Warn:
|
||||
logger.WarnContext(ctx, "gorm.slowQuery", attrs...)
|
||||
default:
|
||||
logger.InfoContext(ctx, "gorm.query", attrs...)
|
||||
}
|
||||
}
|
||||
func errorKind(err error) string {
|
||||
switch {
|
||||
case errors.Is(err, context.Canceled):
|
||||
return "canceled"
|
||||
case errors.Is(err, context.DeadlineExceeded):
|
||||
return "deadlineExceeded"
|
||||
case errors.Is(err, gorm.ErrRecordNotFound):
|
||||
return "recordNotFound"
|
||||
default:
|
||||
return fmt.Sprintf("%T", err)
|
||||
}
|
||||
}
|
||||
func clip(value string, limit int) string {
|
||||
if len(value) <= limit {
|
||||
return value
|
||||
}
|
||||
value = value[:limit]
|
||||
for !utf8.ValidString(value) && len(value) > 0 {
|
||||
value = value[:len(value)-1]
|
||||
}
|
||||
return value + "…"
|
||||
}
|
||||
|
||||
var _ gormlog.Interface = (*sqlLogger)(nil)
|
||||
var _ gorm.ParamsFilter = (*sqlLogger)(nil)
|
||||
92
database/gormx/logger_test.go
Normal file
92
database/gormx/logger_test.go
Normal file
@@ -0,0 +1,92 @@
|
||||
package gormx
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.yuntuo.hk/k2safe/go-omni/logx"
|
||||
"gorm.io/gorm"
|
||||
gormlog "gorm.io/gorm/logger"
|
||||
)
|
||||
|
||||
func TestSQLLoggerLevelsPrivacyAndCorrelation(t *testing.T) {
|
||||
var out bytes.Buffer
|
||||
base, _ := logx.New(logx.Config{Writer: &out})
|
||||
l, _ := NewLogger(base, DefaultLogConfig())
|
||||
ctx := logx.WithRequestID(logx.WithTraceID(context.Background(), "trace-1"), "request-1")
|
||||
count := 0
|
||||
fc := func() (string, int64) { count++; return "SELECT * FROM records WHERE token = ?", 1 }
|
||||
l.Trace(ctx, time.Now(), fc, nil)
|
||||
l.Trace(ctx, time.Now(), fc, fmt.Errorf("wrapped: %w", gorm.ErrRecordNotFound))
|
||||
l.LogMode(gormlog.Silent).Trace(ctx, time.Now(), fc, errors.New("password-value"))
|
||||
if count != 0 {
|
||||
t.Fatal("ignored event evaluated SQL")
|
||||
}
|
||||
l.Trace(ctx, time.Now().Add(-time.Second), fc, nil)
|
||||
l.Trace(ctx, time.Now(), fc, errors.New("password-value"))
|
||||
l.Error(ctx, "failed: %v", errors.New("password-value"))
|
||||
l.LogMode(gormlog.Info).Trace(ctx, time.Now(), fc, nil)
|
||||
if count != 3 {
|
||||
t.Fatal(count)
|
||||
}
|
||||
if strings.Contains(out.String(), "password-value") {
|
||||
t.Fatal("driver detail leaked", out.String())
|
||||
}
|
||||
for _, text := range []string{"gorm.slowQuery", "gorm.queryFailed", "gorm.query", `"requestId":"request-1"`, `"traceId":"trace-1"`} {
|
||||
if !strings.Contains(out.String(), text) {
|
||||
t.Fatal(text, out.String())
|
||||
}
|
||||
}
|
||||
q, args := l.(gorm.ParamsFilter).ParamsFilter(ctx, "SELECT ?", "private")
|
||||
if q != "SELECT ?" || len(args) != 0 {
|
||||
t.Fatal("default SQL arguments exposed")
|
||||
}
|
||||
// LogMode clones rather than mutating the shared instance.
|
||||
l.Trace(ctx, time.Now(), fc, nil)
|
||||
if count != 3 {
|
||||
t.Fatal("session mode changed root logger")
|
||||
}
|
||||
}
|
||||
|
||||
func TestSQLLoggerExplicitDetailsAndConcurrentModes(t *testing.T) {
|
||||
var out bytes.Buffer
|
||||
base, _ := logx.New(logx.Config{Writer: &out})
|
||||
cfg := DefaultLogConfig()
|
||||
cfg.IncludeParameters = true
|
||||
cfg.IncludeErrorDetails = true
|
||||
cfg.Level = "info"
|
||||
cfg.MaxSQLBytes = 8
|
||||
l, err := NewLogger(base, cfg)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_, args := l.(gorm.ParamsFilter).ParamsFilter(context.Background(), "SELECT ?", "value")
|
||||
if len(args) != 1 {
|
||||
t.Fatal("explicit arguments dropped")
|
||||
}
|
||||
var wg sync.WaitGroup
|
||||
for i := 0; i < 20; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
l.LogMode(gormlog.Error).Trace(context.Background(), time.Now(), func() (string, int64) { return "SELECT 中文标识符", -1 }, errors.New("failure"))
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
for _, line := range strings.Split(strings.TrimSpace(out.String()), "\n") {
|
||||
var event map[string]any
|
||||
if err = json.Unmarshal([]byte(line), &event); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if event["error"] != "failure" || strings.Contains(event["sql"].(string), "<22>") {
|
||||
t.Fatal(event)
|
||||
}
|
||||
}
|
||||
}
|
||||
52
database/gormx/pool.go
Normal file
52
database/gormx/pool.go
Normal file
@@ -0,0 +1,52 @@
|
||||
package gormx
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"sync/atomic"
|
||||
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
// GORM can call DB().Close on failed initialization. Do not expose the owner's
|
||||
// sql.DB until initialization has succeeded; SQL methods remain fully usable.
|
||||
// Dialects using context.Background during version probing are bounded by the
|
||||
// startup context, which stops affecting queries as soon as Open succeeds.
|
||||
type borrowedPool struct {
|
||||
db *sql.DB
|
||||
startup atomic.Pointer[startupContext]
|
||||
}
|
||||
|
||||
type startupContext struct{ context.Context }
|
||||
|
||||
func (p *borrowedPool) operationContext(ctx context.Context) context.Context {
|
||||
if startup := p.startup.Load(); startup != nil {
|
||||
return startup.Context
|
||||
}
|
||||
return ctx
|
||||
}
|
||||
func (p *borrowedPool) GetDBConn() (*sql.DB, error) {
|
||||
if p.startup.Load() != nil {
|
||||
return nil, gorm.ErrInvalidDB
|
||||
}
|
||||
return p.db, nil
|
||||
}
|
||||
func (p *borrowedPool) PrepareContext(ctx context.Context, query string) (*sql.Stmt, error) {
|
||||
return p.db.PrepareContext(p.operationContext(ctx), query)
|
||||
}
|
||||
func (p *borrowedPool) ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error) {
|
||||
return p.db.ExecContext(p.operationContext(ctx), query, args...)
|
||||
}
|
||||
func (p *borrowedPool) QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error) {
|
||||
return p.db.QueryContext(p.operationContext(ctx), query, args...)
|
||||
}
|
||||
func (p *borrowedPool) QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row {
|
||||
return p.db.QueryRowContext(p.operationContext(ctx), query, args...)
|
||||
}
|
||||
func (p *borrowedPool) BeginTx(ctx context.Context, opts *sql.TxOptions) (*sql.Tx, error) {
|
||||
return p.db.BeginTx(p.operationContext(ctx), opts)
|
||||
}
|
||||
|
||||
var _ gorm.ConnPool = (*borrowedPool)(nil)
|
||||
var _ gorm.TxBeginner = (*borrowedPool)(nil)
|
||||
var _ gorm.GetDBConnector = (*borrowedPool)(nil)
|
||||
@@ -2,6 +2,8 @@
|
||||
|
||||
本包可选,使用 `database/sql` 和 `go-sql-driver/mysql`,复用框架的唯一连接池、Ping、事务与健康检查。不自动建库/迁移、不重试写操作,也不强制某个 MySQL 版本或业务表。
|
||||
|
||||
项目使用 GORM 时,在 svc.Init 中调用 [gormx.MySQL](../gormx/README.md) 即可复用这个池、TLS 和清理流程。
|
||||
|
||||
生成项目选择 `--database mysql` 即接入:
|
||||
|
||||
```sh
|
||||
|
||||
Reference in New Issue
Block a user