bingyi/gf
1
0
Fork 0
mirror of https://gitee.com/johng/gf synced 2026-07-08 14:39:50 +08:00
Code Issues Packages Projects Releases Wiki Activity

add tracing feature for package gdb

Browse Source
This commit is contained in:
jianchenma
2021-01-25 21:17:32 +08:00
parent cc1e340585
commit b89d8d2740
12 changed files with 449 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
database/gdb/gdb.go
View File

@ -55,7 +55,7 @@ type DB interface {
Query(sql string, args ...interface{}) (*sql.Rows, error)
Exec(sql string, args ...interface{}) (sql.Result, error)
Prepare(sql string, execOnMaster ...bool) (*sql.Stmt, error)
Prepare(sql string, execOnMaster ...bool) (*Stmt, error)
// ===========================================================================
// Common APIs for CURD.
@ -80,7 +80,7 @@ type DB interface {
DoQuery(link Link, sql string, args ...interface{}) (rows *sql.Rows, err error)
DoGetAll(link Link, sql string, args ...interface{}) (result Result, err error)
DoExec(link Link, sql string, args ...interface{}) (result sql.Result, err error)
DoPrepare(link Link, sql string) (*sql.Stmt, error)
DoPrepare(link Link, sql string) (*Stmt, error)
DoInsert(link Link, table string, data interface{}, option int, batch ...int) (result sql.Result, err error)
DoBatchInsert(link Link, table string, list interface{}, option int, batch ...int) (result sql.Result, err error)
DoUpdate(link Link, table string, data interface{}, condition string, args ...interface{}) (result sql.Result, err error)
@ -154,6 +154,7 @@ type DB interface {
Tables(schema ...string) (tables []string, err error)
TableFields(table string, schema ...string) (map[string]*TableField, error)
HasTable(name string) (bool, error)
FilteredLinkInfo() string
// HandleSqlBeforeCommit is a hook function, which deals with the sql string before
// it's committed to underlying driver. The parameter <link> specifies the current
@ -191,6 +192,7 @@ type Driver interface {
// Sql is the sql recording struct.
type Sql struct {
Sql string // SQL string(may contain reserved char '?').
Type string // SQL operation type.
Args []interface{} // Arguments for this sql.
Format string // Formatted sql which contains arguments in the sql.
Error error // Execution result.
@ -216,9 +218,9 @@ type Link interface {
Query(sql string, args ...interface{}) (*sql.Rows, error)
Exec(sql string, args ...interface{}) (sql.Result, error)
Prepare(sql string) (*sql.Stmt, error)
QueryContext(ctx context.Context, query string, args ...interface{}) (*sql.Rows, error)
ExecContext(ctx context.Context, query string, args ...interface{}) (sql.Result, error)
PrepareContext(ctx context.Context, query string) (*sql.Stmt, error)
QueryContext(ctx context.Context, sql string, args ...interface{}) (*sql.Rows, error)
ExecContext(ctx context.Context, sql string, args ...interface{}) (sql.Result, error)
PrepareContext(ctx context.Context, sql string) (*sql.Stmt, error)
}
// Counter is the type for update count.

168
database/gdb/gdb_core.go
View File

@ -11,8 +11,13 @@ import (
"context"
"database/sql"
"fmt"
"github.com/gogf/gf"
"github.com/gogf/gf/errors/gerror"
"github.com/gogf/gf/text/gstr"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/codes"
"go.opentelemetry.io/otel/label"
"go.opentelemetry.io/otel/trace"
"reflect"
"strings"
@ -85,24 +90,27 @@ func (c *Core) DoQuery(link Link, sql string, args ...interface{}) (rows *sql.Ro
sql, args = c.DB.HandleSqlBeforeCommit(link, sql, args)
ctx := c.DB.GetCtx()
if c.GetConfig().QueryTimeout > 0 {
ctx, _ = context.WithTimeout(ctx, c.GetConfig().QueryTimeout)
var cancelFunc context.CancelFunc
ctx, cancelFunc = context.WithTimeout(ctx, c.GetConfig().QueryTimeout)
defer cancelFunc()
}
mTime1 := gtime.TimestampMilli()
rows, err = link.QueryContext(ctx, sql, args...)
mTime2 := gtime.TimestampMilli()
sqlObj := &Sql{
Sql: sql,
Type: "DB.QueryContext",
Args: args,
Format: FormatSqlWithArgs(sql, args),
Error: err,
Start: mTime1,
End: mTime2,
Group: c.DB.GetGroup(),
}
c.addSqlToTracing(ctx, sqlObj)
if c.DB.GetDebug() {
mTime1 := gtime.TimestampMilli()
rows, err = link.QueryContext(ctx, sql, args...)
mTime2 := gtime.TimestampMilli()
s := &Sql{
Sql: sql,
Args: args,
Format: FormatSqlWithArgs(sql, args),
Error: err,
Start: mTime1,
End: mTime2,
Group: c.DB.GetGroup(),
}
c.writeSqlToLogger(s)
} else {
rows, err = link.QueryContext(ctx, sql, args...)
c.writeSqlToLogger(sqlObj)
}
if err == nil {
return rows, nil
@ -112,6 +120,48 @@ func (c *Core) DoQuery(link Link, sql string, args ...interface{}) (rows *sql.Ro
return nil, err
}
func (c *Core) addSqlToTracing(ctx context.Context, sql *Sql) {
if !c.DB.GetConfig().Tracing {
return
}
tr := otel.GetTracerProvider().Tracer(
"github.com/gogf/gf/database/gdb",
trace.WithInstrumentationVersion(fmt.Sprintf(`%s`, gf.VERSION)),
)
ctx, span := tr.Start(ctx, sql.Type)
defer span.End()
if sql.Error != nil {
span.SetStatus(codes.Error, fmt.Sprintf(`%+v`, sql.Error))
}
labels := make([]label.KeyValue, 0)
labels = append(labels, label.String("db.type", c.DB.GetConfig().Type))
if c.DB.GetConfig().Host != "" {
labels = append(labels, label.String("db.host", c.DB.GetConfig().Host))
}
if c.DB.GetConfig().Port != "" {
labels = append(labels, label.String("db.port", c.DB.GetConfig().Port))
}
if c.DB.GetConfig().Name != "" {
labels = append(labels, label.String("db.name", c.DB.GetConfig().Name))
}
if c.DB.GetConfig().User != "" {
labels = append(labels, label.String("db.user", c.DB.GetConfig().User))
}
if filteredLinkInfo := c.DB.FilteredLinkInfo(); filteredLinkInfo != "" {
labels = append(labels, label.String("db.link", c.DB.FilteredLinkInfo()))
}
if group := c.DB.GetGroup(); group != "" {
labels = append(labels, label.String("db.group", group))
}
span.SetAttributes(labels...)
span.AddEvent("db.execution", trace.WithAttributes(
label.String(`db.execution.sql`, sql.Format),
label.String(`db.execution.cost`, fmt.Sprintf(`%d ms`, sql.End-sql.Start)),
label.String(`db.execution.type`, sql.Type),
))
}
// Exec commits one query SQL to underlying driver and returns the execution result.
// It is most commonly used for data inserting and updating.
func (c *Core) Exec(sql string, args ...interface{}) (result sql.Result, err error) {
@ -129,32 +179,31 @@ func (c *Core) DoExec(link Link, sql string, args ...interface{}) (result sql.Re
sql, args = c.DB.HandleSqlBeforeCommit(link, sql, args)
ctx := c.DB.GetCtx()
if c.GetConfig().ExecTimeout > 0 {
ctx, _ = context.WithTimeout(ctx, c.GetConfig().ExecTimeout)
var cancelFunc context.CancelFunc
ctx, cancelFunc = context.WithTimeout(ctx, c.GetConfig().ExecTimeout)
defer cancelFunc()
}
if c.DB.GetDebug() {
mTime1 := gtime.TimestampMilli()
if !c.DB.GetDryRun() {
result, err = link.ExecContext(ctx, sql, args...)
} else {
result = new(SqlResult)
}
mTime2 := gtime.TimestampMilli()
s := &Sql{
Sql: sql,
Args: args,
Format: FormatSqlWithArgs(sql, args),
Error: err,
Start: mTime1,
End: mTime2,
Group: c.DB.GetGroup(),
}
c.writeSqlToLogger(s)
mTime1 := gtime.TimestampMilli()
if !c.DB.GetDryRun() {
result, err = link.ExecContext(ctx, sql, args...)
} else {
if !c.DB.GetDryRun() {
result, err = link.ExecContext(ctx, sql, args...)
} else {
result = new(SqlResult)
}
result = new(SqlResult)
}
mTime2 := gtime.TimestampMilli()
sqlObj := &Sql{
Sql: sql,
Type: "DB.ExecContext",
Args: args,
Format: FormatSqlWithArgs(sql, args),
Error: err,
Start: mTime1,
End: mTime2,
Group: c.DB.GetGroup(),
}
c.addSqlToTracing(ctx, sqlObj)
if c.DB.GetDebug() {
c.writeSqlToLogger(sqlObj)
}
return result, formatError(err, sql, args...)
}
@ -167,7 +216,7 @@ func (c *Core) DoExec(link Link, sql string, args ...interface{}) (result sql.Re
//
// The parameter <execOnMaster> specifies whether executing the sql on master node,
// or else it executes the sql on slave node if master-slave configured.
func (c *Core) Prepare(sql string, execOnMaster ...bool) (*sql.Stmt, error) {
func (c *Core) Prepare(sql string, execOnMaster ...bool) (*Stmt, error) {
var (
err error
link Link
@ -185,12 +234,37 @@ func (c *Core) Prepare(sql string, execOnMaster ...bool) (*sql.Stmt, error) {
}
// doPrepare calls prepare function on given link object and returns the statement object.
func (c *Core) DoPrepare(link Link, sql string) (*sql.Stmt, error) {
func (c *Core) DoPrepare(link Link, sql string) (*Stmt, error) {
ctx := c.DB.GetCtx()
if c.GetConfig().QueryTimeout > 0 {
ctx, _ = context.WithTimeout(ctx, c.GetConfig().QueryTimeout)
if c.GetConfig().PrepareTimeout > 0 {
var cancelFunc context.CancelFunc
ctx, cancelFunc = context.WithTimeout(ctx, c.GetConfig().PrepareTimeout)
defer cancelFunc()
}
return link.PrepareContext(ctx, sql)
var (
mTime1 = gtime.TimestampMilli()
stmt, err = link.PrepareContext(ctx, sql)
mTime2 = gtime.TimestampMilli()
sqlObj = &Sql{
Sql: sql,
Type: "DB.PrepareContext",
Args: nil,
Format: FormatSqlWithArgs(sql, nil),
Error: err,
Start: mTime1,
End: mTime2,
Group: c.DB.GetGroup(),
}
)
c.addSqlToTracing(ctx, sqlObj)
if c.DB.GetDebug() {
c.writeSqlToLogger(sqlObj)
}
return &Stmt{
Stmt: stmt,
core: c,
sql: sql,
}, err
}
// GetAll queries and returns data records from database.
@ -334,7 +408,9 @@ func (c *Core) Begin() (*TX, error) {
} else {
ctx := c.DB.GetCtx()
if c.GetConfig().TranTimeout > 0 {
ctx, _ = context.WithTimeout(ctx, c.GetConfig().TranTimeout)
var cancelFunc context.CancelFunc
ctx, cancelFunc = context.WithTimeout(ctx, c.GetConfig().TranTimeout)
defer cancelFunc()
}
if tx, err := master.BeginTx(ctx, nil); err == nil {
return &TX{

1
database/gdb/gdb_core_config.go
View File

@ -51,6 +51,7 @@ type ConfigNode struct {
UpdatedAt string `json:"updatedAt"` // (Optional) The filed name of table for automatic-filled updated datetime.
DeletedAt string `json:"deletedAt"` // (Optional) The filed name of table for automatic-filled updated datetime.
TimeMaintainDisabled bool `json:"timeMaintainDisabled"` // (Optional) Disable the automatic time maintaining feature.
Tracing bool `json:"tracing"` // (Optional) Tracing enable the tracing feature of database.
}
// configs is internal used configuration object.

15
database/gdb/gdb_driver_mssql.go
View File

@ -56,6 +56,21 @@ func (d *DriverMssql) Open(config *ConfigNode) (*sql.DB, error) {
}
}
// FilteredLinkInfo retrieves and returns filtered `linkInfo` that can be using for
// logging or tracing purpose.
func (d *DriverMssql) FilteredLinkInfo() string {
linkInfo := d.GetConfig().LinkInfo
if linkInfo == "" {
return ""
}
s, _ := gregex.ReplaceString(
`(.+);\s*password=(.+);\s*server=(.+)`,
`$1;password=xxx;server=$3`,
d.GetConfig().LinkInfo,
)
return s
}
// GetChars returns the security char for this type of database.
func (d *DriverMssql) GetChars() (charLeft string, charRight string) {
return "\"", "\""

15
database/gdb/gdb_driver_mysql.go
View File

@ -54,6 +54,21 @@ func (d *DriverMysql) Open(config *ConfigNode) (*sql.DB, error) {
}
}
// FilteredLinkInfo retrieves and returns filtered `linkInfo` that can be using for
// logging or tracing purpose.
func (d *DriverMysql) FilteredLinkInfo() string {
linkInfo := d.GetConfig().LinkInfo
if linkInfo == "" {
return ""
}
s, _ := gregex.ReplaceString(
`(.+?):(.+)@tcp(.+)`,
`$1:xxx@tcp$3`,
linkInfo,
)
return s
}
// GetChars returns the security char for this type of database.
func (d *DriverMysql) GetChars() (charLeft string, charRight string) {
return "`", "`"

21
database/gdb/gdb_driver_oracle.go
View File

@ -49,7 +49,11 @@ func (d *DriverOracle) Open(config *ConfigNode) (*sql.DB, error) {
if config.LinkInfo != "" {
source = config.LinkInfo
} else {
source = fmt.Sprintf("%s/%s@%s", config.User, config.Pass, config.Name)
// 账号/密码@地址:端口/数据库名称
source = fmt.Sprintf(
"%s/%s@%s:%s/%s",
config.User, config.Pass, config.Host, config.Port, config.Name,
)
}
intlog.Printf("Open: %s", source)
if db, err := sql.Open("oci8", source); err == nil {
@ -59,6 +63,21 @@ func (d *DriverOracle) Open(config *ConfigNode) (*sql.DB, error) {
}
}
// FilteredLinkInfo retrieves and returns filtered `linkInfo` that can be using for
// logging or tracing purpose.
func (d *DriverOracle) FilteredLinkInfo() string {
linkInfo := d.GetConfig().LinkInfo
if linkInfo == "" {
return ""
}
s, _ := gregex.ReplaceString(
`(.+?)\s*/\s*(.+)\s*@\s*(.+)\s*:\s*(\d+)\s*/\s*(.+)`,
`$1/xxx@$3:$4/$5`,
linkInfo,
)
return s
}
// GetChars returns the security char for this type of database.
func (d *DriverOracle) GetChars() (charLeft string, charRight string) {
return "\"", "\""

15
database/gdb/gdb_driver_pgsql.go
View File

@ -54,6 +54,21 @@ func (d *DriverPgsql) Open(config *ConfigNode) (*sql.DB, error) {
}
}
// FilteredLinkInfo retrieves and returns filtered `linkInfo` that can be using for
// logging or tracing purpose.
func (d *DriverPgsql) FilteredLinkInfo() string {
linkInfo := d.GetConfig().LinkInfo
if linkInfo == "" {
return ""
}
s, _ := gregex.ReplaceString(
`(.+?)\s*password=(.+)\s*host=(.+)`,
`$1 password=xxx host=$3`,
linkInfo,
)
return s
}
// GetChars returns the security char for this type of database.
func (d *DriverPgsql) GetChars() (charLeft string, charRight string) {
return "\"", "\""

6
database/gdb/gdb_driver_sqlite.go
View File

@ -53,6 +53,12 @@ func (d *DriverSqlite) Open(config *ConfigNode) (*sql.DB, error) {
}
}
// FilteredLinkInfo retrieves and returns filtered `linkInfo` that can be using for
// logging or tracing purpose.
func (d *DriverSqlite) FilteredLinkInfo() string {
return d.GetConfig().LinkInfo
}
// GetChars returns the security char for this type of database.
func (d *DriverSqlite) GetChars() (charLeft string, charRight string) {
return "`", "`"

163
database/gdb/gdb_statement.go Normal file
View File

@ -0,0 +1,163 @@
// Copyright GoFrame Author(https://goframe.org). All Rights Reserved.
//
// This Source Code Form is subject to the terms of the MIT License.
// If a copy of the MIT was not distributed with this file,
// You can obtain one at https://github.com/gogf/gf.
package gdb
import (
"context"
"database/sql"
"github.com/gogf/gf/os/gtime"
)
// Stmt is a prepared statement.
// A Stmt is safe for concurrent use by multiple goroutines.
//
// If a Stmt is prepared on a Tx or Conn, it will be bound to a single
// underlying connection forever. If the Tx or Conn closes, the Stmt will
// become unusable and all operations will return an error.
// If a Stmt is prepared on a DB, it will remain usable for the lifetime of the
// DB. When the Stmt needs to execute on a new underlying connection, it will
// prepare itself on the new connection automatically.
type Stmt struct {
*sql.Stmt
core *Core
sql string
}
// ExecContext executes a prepared statement with the given arguments and
// returns a Result summarizing the effect of the statement.
func (s *Stmt) ExecContext(ctx context.Context, args ...interface{}) (sql.Result, error) {
var cancelFunc context.CancelFunc
ctx, cancelFunc = context.WithCancel(ctx)
defer cancelFunc()
if s.core.DB.GetConfig().ExecTimeout > 0 {
var cancelFuncForTimeout context.CancelFunc
ctx, cancelFuncForTimeout = context.WithTimeout(ctx, s.core.DB.GetConfig().ExecTimeout)
defer cancelFuncForTimeout()
}
var (
mTime1 = gtime.TimestampMilli()
result, err = s.Stmt.ExecContext(ctx, args...)
mTime2 = gtime.TimestampMilli()
sqlObj = &Sql{
Sql: s.sql,
Type: "Statement.ExecContext",
Args: args,
Format: FormatSqlWithArgs(s.sql, args),
Error: err,
Start: mTime1,
End: mTime2,
Group: s.core.DB.GetGroup(),
}
)
s.core.addSqlToTracing(ctx, sqlObj)
if s.core.DB.GetDebug() {
s.core.writeSqlToLogger(sqlObj)
}
return result, err
}
// QueryContext executes a prepared query statement with the given arguments
// and returns the query results as a *Rows.
func (s *Stmt) QueryContext(ctx context.Context, args ...interface{}) (*sql.Rows, error) {
var cancelFunc context.CancelFunc
ctx, cancelFunc = context.WithCancel(ctx)
defer cancelFunc()
if s.core.DB.GetConfig().QueryTimeout > 0 {
var cancelFuncForTimeout context.CancelFunc
ctx, cancelFuncForTimeout = context.WithTimeout(ctx, s.core.DB.GetConfig().QueryTimeout)
defer cancelFuncForTimeout()
}
var (
mTime1 = gtime.TimestampMilli()
rows, err = s.Stmt.QueryContext(ctx, args...)
mTime2 = gtime.TimestampMilli()
sqlObj = &Sql{
Sql: s.sql,
Type: "Statement.QueryContext",
Args: args,
Format: FormatSqlWithArgs(s.sql, args),
Error: err,
Start: mTime1,
End: mTime2,
Group: s.core.DB.GetGroup(),
}
)
s.core.addSqlToTracing(ctx, sqlObj)
if s.core.DB.GetDebug() {
s.core.writeSqlToLogger(sqlObj)
}
return rows, err
}
// QueryRowContext executes a prepared query statement with the given arguments.
// If an error occurs during the execution of the statement, that error will
// be returned by a call to Scan on the returned *Row, which is always non-nil.
// If the query selects no rows, the *Row's Scan will return ErrNoRows.
// Otherwise, the *Row's Scan scans the first selected row and discards
// the rest.
func (s *Stmt) QueryRowContext(ctx context.Context, args ...interface{}) *sql.Row {
var cancelFunc context.CancelFunc
ctx, cancelFunc = context.WithCancel(ctx)
defer cancelFunc()
if s.core.DB.GetConfig().QueryTimeout > 0 {
var cancelFuncForTimeout context.CancelFunc
ctx, cancelFuncForTimeout = context.WithTimeout(ctx, s.core.DB.GetConfig().QueryTimeout)
defer cancelFuncForTimeout()
}
var (
mTime1 = gtime.TimestampMilli()
row = s.Stmt.QueryRowContext(ctx, args...)
mTime2 = gtime.TimestampMilli()
sqlObj = &Sql{
Sql: s.sql,
Type: "Statement.QueryRowContext",
Args: args,
Format: FormatSqlWithArgs(s.sql, args),
Error: nil,
Start: mTime1,
End: mTime2,
Group: s.core.DB.GetGroup(),
}
)
s.core.addSqlToTracing(ctx, sqlObj)
if s.core.DB.GetDebug() {
s.core.writeSqlToLogger(sqlObj)
}
return row
}
// Exec executes a prepared statement with the given arguments and
// returns a Result summarizing the effect of the statement.
func (s *Stmt) Exec(args ...interface{}) (sql.Result, error) {
return s.ExecContext(context.Background(), args)
}
// Query executes a prepared query statement with the given arguments
// and returns the query results as a *Rows.
func (s *Stmt) Query(args ...interface{}) (*sql.Rows, error) {
return s.QueryContext(context.Background(), args...)
}
// QueryRow executes a prepared query statement with the given arguments.
// If an error occurs during the execution of the statement, that error will
// be returned by a call to Scan on the returned *Row, which is always non-nil.
// If the query selects no rows, the *Row's Scan will return ErrNoRows.
// Otherwise, the *Row's Scan scans the first selected row and discards
// the rest.
//
// Example usage:
//
// var name string
// err := nameByUseridStmt.QueryRow(id).Scan(&name)
func (s *Stmt) QueryRow(args ...interface{}) *sql.Row {
return s.QueryRowContext(context.Background(), args...)
}
// Close closes the statement.
func (s *Stmt) Close() error {
return s.Stmt.Close()
}

2
database/gdb/gdb_transaction.go
View File

@ -48,7 +48,7 @@ func (tx *TX) Exec(sql string, args ...interface{}) (sql.Result, error) {
// returned statement.
// The caller must call the statement's Close method
// when the statement is no longer needed.
func (tx *TX) Prepare(sql string) (*sql.Stmt, error) {
func (tx *TX) Prepare(sql string) (*Stmt, error) {
return tx.db.DoPrepare(tx.tx, sql)
}