pharnexbase
/
kratos
11
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

use real ip in trace and metrics.

Browse Source
pull/265/head
felixhao 5 years ago
parent 3e92753a1b
commit ecb0c6104c
2 changed files with 65 additions and 38 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats Download Patch File Download Diff File
  1. 24
      pkg/database/sql/CHANGELOG.md
  2. 79
      pkg/database/sql/sql.go

24
pkg/database/sql/CHANGELOG.md
Unescape View File

@ -0,0 +1,24 @@
### database/sql
##### Version 1.3.1
> 1. trace、metric 区别不同数据库实例（主、从）
##### Version 1.3.0
> 1. add slow log
##### Version 1.2.1
> 1.支持上报熔断错误
##### Version 1.2.0
> 1.添加数据库读写分离
##### Version 1.1.1
> 1.修复部分函数中的context cancel leak
##### Version 1.1.0
> 1.添加context和timeout支持
> 2.添加breaker支持
> 3.更新driver，支持context/pool timeout
##### Version 1.0.0
> 1.支持trace/stats

79
pkg/database/sql/sql.go
Unescape View File

@ -8,10 +8,10 @@ import (
"sync/atomic" "sync/atomic"
"time" "time"
"github.com/bilibili/kratos/pkg/ecode" "go-common/library/ecode"
"github.com/bilibili/kratos/pkg/log" "go-common/library/log"
"github.com/bilibili/kratos/pkg/net/netutil/breaker" "go-common/library/net/netutil/breaker"
"github.com/bilibili/kratos/pkg/net/trace" "go-common/library/net/trace"
"github.com/pkg/errors" "github.com/pkg/errors"
) )
@ -38,7 +38,7 @@ var (
type DB struct { type DB struct {
write *conn write *conn
read []*conn read []*conn
idx uint64 idx int64
master *DB master *DB
} }
@ -47,6 +47,7 @@ type conn struct {
*sql.DB *sql.DB
breaker breaker.Breaker breaker breaker.Breaker
conf *Config conf *Config
addr string
} }
// Tx transaction. // Tx transaction.
@ -130,17 +131,19 @@ func Open(c *Config) (*DB, error) {
if err != nil { if err != nil {
return nil, err return nil, err
} }
addr := parseDSNAddr(c.DSN)
brkGroup := breaker.NewGroup(c.Breaker) brkGroup := breaker.NewGroup(c.Breaker)
brk := brkGroup.Get(c.Addr) brk := brkGroup.Get(addr)
w := &conn{DB: d, breaker: brk, conf: c} w := &conn{DB: d, breaker: brk, conf: c, addr: addr}
rs := make([]*conn, 0, len(c.ReadDSN)) rs := make([]*conn, 0, len(c.ReadDSN))
for _, rd := range c.ReadDSN { for _, rd := range c.ReadDSN {
d, err := connect(c, rd) d, err := connect(c, rd)
if err != nil { if err != nil {
return nil, err return nil, err
} }
brk := brkGroup.Get(parseDSNAddr(rd)) addr = parseDSNAddr(rd)
r := &conn{DB: d, breaker: brk, conf: c} brk := brkGroup.Get(addr)
r := &conn{DB: d, breaker: brk, conf: c, addr: addr}
rs = append(rs, r) rs = append(rs, r)
} }
db.write = w db.write = w
@ -217,8 +220,8 @@ func (db *DB) readIndex() int {
if len(db.read) == 0 { if len(db.read) == 0 {
return 0 return 0
} }
v := atomic.AddUint64(&db.idx, 1) v := atomic.AddInt64(&db.idx, 1)
return int(v % uint64(len(db.read))) return int(v) % len(db.read)
} }
// Close closes the write and read database, releasing any open resources. // Close closes the write and read database, releasing any open resources.
@ -271,7 +274,7 @@ func (db *conn) begin(c context.Context) (tx *Tx, err error) {
t, ok := trace.FromContext(c) t, ok := trace.FromContext(c)
if ok { if ok {
t = t.Fork(_family, "begin") t = t.Fork(_family, "begin")
t.SetTag(trace.String(trace.TagAddress, db.conf.Addr), trace.String(trace.TagComment, "")) t.SetTag(trace.String(trace.TagAddress, db.addr), trace.String(trace.TagComment, ""))
defer func() { defer func() {
if err != nil { if err != nil {
t.Finish(&err) t.Finish(&err)
@ -279,12 +282,12 @@ func (db *conn) begin(c context.Context) (tx *Tx, err error) {
}() }()
} }
if err = db.breaker.Allow(); err != nil { if err = db.breaker.Allow(); err != nil {
_metricReqErr.Inc(db.conf.Addr, db.conf.Addr, "begin", "breaker") _metricReqErr.Inc(db.addr, db.addr, "begin", "breaker")
return return
} }
_, c, cancel := db.conf.TranTimeout.Shrink(c) _, c, cancel := db.conf.TranTimeout.Shrink(c)
rtx, err := db.BeginTx(c, nil) rtx, err := db.BeginTx(c, nil)
_metricReqDur.Observe(int64(time.Since(now)/time.Millisecond), db.conf.Addr, db.conf.Addr, "begin") _metricReqDur.Observe(int64(time.Since(now)/time.Millisecond), db.addr, db.addr, "begin")
if err != nil { if err != nil {
err = errors.WithStack(err) err = errors.WithStack(err)
cancel() cancel()
@ -299,18 +302,18 @@ func (db *conn) exec(c context.Context, query string, args ...interface{}) (res
defer slowLog(fmt.Sprintf("Exec query(%s) args(%+v)", query, args), now) defer slowLog(fmt.Sprintf("Exec query(%s) args(%+v)", query, args), now)
if t, ok := trace.FromContext(c); ok { if t, ok := trace.FromContext(c); ok {
t = t.Fork(_family, "exec") t = t.Fork(_family, "exec")
t.SetTag(trace.String(trace.TagAddress, db.conf.Addr), trace.String(trace.TagComment, query)) t.SetTag(trace.String(trace.TagAddress, db.addr), trace.String(trace.TagComment, query))
defer t.Finish(&err) defer t.Finish(&err)
} }
if err = db.breaker.Allow(); err != nil { if err = db.breaker.Allow(); err != nil {
_metricReqErr.Inc(db.conf.Addr, db.conf.Addr, "exec", "breaker") _metricReqErr.Inc(db.addr, db.addr, "exec", "breaker")
return return
} }
_, c, cancel := db.conf.ExecTimeout.Shrink(c) _, c, cancel := db.conf.ExecTimeout.Shrink(c)
res, err = db.ExecContext(c, query, args...) res, err = db.ExecContext(c, query, args...)
cancel() cancel()
db.onBreaker(&err) db.onBreaker(&err)
_metricReqDur.Observe(int64(time.Since(now)/time.Millisecond), db.conf.Addr, db.conf.Addr, "exec") _metricReqDur.Observe(int64(time.Since(now)/time.Millisecond), db.addr, db.addr, "exec")
if err != nil { if err != nil {
err = errors.Wrapf(err, "exec:%s, args:%+v", query, args) err = errors.Wrapf(err, "exec:%s, args:%+v", query, args)
} }
@ -322,18 +325,18 @@ func (db *conn) ping(c context.Context) (err error) {
defer slowLog("Ping", now) defer slowLog("Ping", now)
if t, ok := trace.FromContext(c); ok { if t, ok := trace.FromContext(c); ok {
t = t.Fork(_family, "ping") t = t.Fork(_family, "ping")
t.SetTag(trace.String(trace.TagAddress, db.conf.Addr), trace.String(trace.TagComment, "")) t.SetTag(trace.String(trace.TagAddress, db.addr), trace.String(trace.TagComment, ""))
defer t.Finish(&err) defer t.Finish(&err)
} }
if err = db.breaker.Allow(); err != nil { if err = db.breaker.Allow(); err != nil {
_metricReqErr.Inc(db.conf.Addr, db.conf.Addr, "ping", "breaker") _metricReqErr.Inc(db.addr, db.addr, "ping", "breaker")
return return
} }
_, c, cancel := db.conf.ExecTimeout.Shrink(c) _, c, cancel := db.conf.ExecTimeout.Shrink(c)
err = db.PingContext(c) err = db.PingContext(c)
cancel() cancel()
db.onBreaker(&err) db.onBreaker(&err)
_metricReqDur.Observe(int64(time.Since(now)/time.Millisecond), db.conf.Addr, db.conf.Addr, "ping") _metricReqDur.Observe(int64(time.Since(now)/time.Millisecond), db.addr, db.addr, "ping")
if err != nil { if err != nil {
err = errors.WithStack(err) err = errors.WithStack(err)
} }
@ -379,17 +382,17 @@ func (db *conn) query(c context.Context, query string, args ...interface{}) (row
defer slowLog(fmt.Sprintf("Query query(%s) args(%+v)", query, args), now) defer slowLog(fmt.Sprintf("Query query(%s) args(%+v)", query, args), now)
if t, ok := trace.FromContext(c); ok { if t, ok := trace.FromContext(c); ok {
t = t.Fork(_family, "query") t = t.Fork(_family, "query")
t.SetTag(trace.String(trace.TagAddress, db.conf.Addr), trace.String(trace.TagComment, query)) t.SetTag(trace.String(trace.TagAddress, db.addr), trace.String(trace.TagComment, query))
defer t.Finish(&err) defer t.Finish(&err)
} }
if err = db.breaker.Allow(); err != nil { if err = db.breaker.Allow(); err != nil {
_metricReqErr.Inc(db.conf.Addr, db.conf.Addr, "query", "breaker") _metricReqErr.Inc(db.addr, db.addr, "query", "breaker")
return return
} }
_, c, cancel := db.conf.QueryTimeout.Shrink(c) _, c, cancel := db.conf.QueryTimeout.Shrink(c)
rs, err := db.DB.QueryContext(c, query, args...) rs, err := db.DB.QueryContext(c, query, args...)
db.onBreaker(&err) db.onBreaker(&err)
_metricReqDur.Observe(int64(time.Since(now)/time.Millisecond), db.conf.Addr, db.conf.Addr, "query") _metricReqDur.Observe(int64(time.Since(now)/time.Millisecond), db.addr, db.addr, "query")
if err != nil { if err != nil {
err = errors.Wrapf(err, "query:%s, args:%+v", query, args) err = errors.Wrapf(err, "query:%s, args:%+v", query, args)
cancel() cancel()
@ -405,15 +408,15 @@ func (db *conn) queryRow(c context.Context, query string, args ...interface{}) *
t, ok := trace.FromContext(c) t, ok := trace.FromContext(c)
if ok { if ok {
t = t.Fork(_family, "queryrow") t = t.Fork(_family, "queryrow")
t.SetTag(trace.String(trace.TagAddress, db.conf.Addr), trace.String(trace.TagComment, query)) t.SetTag(trace.String(trace.TagAddress, db.addr), trace.String(trace.TagComment, query))
} }
if err := db.breaker.Allow(); err != nil { if err := db.breaker.Allow(); err != nil {
_metricReqErr.Inc(db.conf.Addr, db.conf.Addr, "queryRow", "breaker") _metricReqErr.Inc(db.addr, db.addr, "queryRow", "breaker")
return &Row{db: db, t: t, err: err} return &Row{db: db, t: t, err: err}
} }
_, c, cancel := db.conf.QueryTimeout.Shrink(c) _, c, cancel := db.conf.QueryTimeout.Shrink(c)
r := db.DB.QueryRowContext(c, query, args...) r := db.DB.QueryRowContext(c, query, args...)
_metricReqDur.Observe(int64(time.Since(now)/time.Millisecond), db.conf.Addr, db.conf.Addr, "queryrow") _metricReqDur.Observe(int64(time.Since(now)/time.Millisecond), db.addr, db.addr, "queryrow")
return &Row{db: db, Row: r, query: query, args: args, t: t, cancel: cancel} return &Row{db: db, Row: r, query: query, args: args, t: t, cancel: cancel}
} }
@ -445,11 +448,11 @@ func (s *Stmt) Exec(c context.Context, args ...interface{}) (res sql.Result, err
} }
} else if t, ok := trace.FromContext(c); ok { } else if t, ok := trace.FromContext(c); ok {
t = t.Fork(_family, "exec") t = t.Fork(_family, "exec")
t.SetTag(trace.String(trace.TagAddress, s.db.conf.Addr), trace.String(trace.TagComment, s.query)) t.SetTag(trace.String(trace.TagAddress, s.db.addr), trace.String(trace.TagComment, s.query))
defer t.Finish(&err) defer t.Finish(&err)
} }
if err = s.db.breaker.Allow(); err != nil { if err = s.db.breaker.Allow(); err != nil {
_metricReqErr.Inc(s.db.conf.Addr, s.db.conf.Addr, "stmt:exec", "breaker") _metricReqErr.Inc(s.db.addr, s.db.addr, "stmt:exec", "breaker")
return return
} }
stmt, ok := s.stmt.Load().(*sql.Stmt) stmt, ok := s.stmt.Load().(*sql.Stmt)
@ -461,7 +464,7 @@ func (s *Stmt) Exec(c context.Context, args ...interface{}) (res sql.Result, err
res, err = stmt.ExecContext(c, args...) res, err = stmt.ExecContext(c, args...)
cancel() cancel()
s.db.onBreaker(&err) s.db.onBreaker(&err)
_metricReqDur.Observe(int64(time.Since(now)/time.Millisecond), s.db.conf.Addr, s.db.conf.Addr, "stmt:exec") _metricReqDur.Observe(int64(time.Since(now)/time.Millisecond), s.db.addr, s.db.addr, "stmt:exec")
if err != nil { if err != nil {
err = errors.Wrapf(err, "exec:%s, args:%+v", s.query, args) err = errors.Wrapf(err, "exec:%s, args:%+v", s.query, args)
} }
@ -483,11 +486,11 @@ func (s *Stmt) Query(c context.Context, args ...interface{}) (rows *Rows, err er
} }
} else if t, ok := trace.FromContext(c); ok { } else if t, ok := trace.FromContext(c); ok {
t = t.Fork(_family, "query") t = t.Fork(_family, "query")
t.SetTag(trace.String(trace.TagAddress, s.db.conf.Addr), trace.String(trace.TagComment, s.query)) t.SetTag(trace.String(trace.TagAddress, s.db.addr), trace.String(trace.TagComment, s.query))
defer t.Finish(&err) defer t.Finish(&err)
} }
if err = s.db.breaker.Allow(); err != nil { if err = s.db.breaker.Allow(); err != nil {
_metricReqErr.Inc(s.db.conf.Addr, s.db.conf.Addr, "stmt:query", "breaker") _metricReqErr.Inc(s.db.addr, s.db.addr, "stmt:query", "breaker")
return return
} }
stmt, ok := s.stmt.Load().(*sql.Stmt) stmt, ok := s.stmt.Load().(*sql.Stmt)
@ -498,7 +501,7 @@ func (s *Stmt) Query(c context.Context, args ...interface{}) (rows *Rows, err er
_, c, cancel := s.db.conf.QueryTimeout.Shrink(c) _, c, cancel := s.db.conf.QueryTimeout.Shrink(c)
rs, err := stmt.QueryContext(c, args...) rs, err := stmt.QueryContext(c, args...)
s.db.onBreaker(&err) s.db.onBreaker(&err)
_metricReqDur.Observe(int64(time.Since(now)/time.Millisecond), s.db.conf.Addr, s.db.conf.Addr, "stmt:query") _metricReqDur.Observe(int64(time.Since(now)/time.Millisecond), s.db.addr, s.db.addr, "stmt:query")
if err != nil { if err != nil {
err = errors.Wrapf(err, "query:%s, args:%+v", s.query, args) err = errors.Wrapf(err, "query:%s, args:%+v", s.query, args)
cancel() cancel()
@ -527,11 +530,11 @@ func (s *Stmt) QueryRow(c context.Context, args ...interface{}) (row *Row) {
} }
} else if t, ok := trace.FromContext(c); ok { } else if t, ok := trace.FromContext(c); ok {
t = t.Fork(_family, "queryrow") t = t.Fork(_family, "queryrow")
t.SetTag(trace.String(trace.TagAddress, s.db.conf.Addr), trace.String(trace.TagComment, s.query)) t.SetTag(trace.String(trace.TagAddress, s.db.addr), trace.String(trace.TagComment, s.query))
row.t = t row.t = t
} }
if row.err = s.db.breaker.Allow(); row.err != nil { if row.err = s.db.breaker.Allow(); row.err != nil {
_metricReqErr.Inc(s.db.conf.Addr, s.db.conf.Addr, "stmt:queryrow", "breaker") _metricReqErr.Inc(s.db.addr, s.db.addr, "stmt:queryrow", "breaker")
return return
} }
stmt, ok := s.stmt.Load().(*sql.Stmt) stmt, ok := s.stmt.Load().(*sql.Stmt)
@ -541,7 +544,7 @@ func (s *Stmt) QueryRow(c context.Context, args ...interface{}) (row *Row) {
_, c, cancel := s.db.conf.QueryTimeout.Shrink(c) _, c, cancel := s.db.conf.QueryTimeout.Shrink(c)
row.Row = stmt.QueryRowContext(c, args...) row.Row = stmt.QueryRowContext(c, args...)
row.cancel = cancel row.cancel = cancel
_metricReqDur.Observe(int64(time.Since(now)/time.Millisecond), s.db.conf.Addr, s.db.conf.Addr, "stmt:queryrow") _metricReqDur.Observe(int64(time.Since(now)/time.Millisecond), s.db.addr, s.db.addr, "stmt:queryrow")
return return
} }
@ -582,7 +585,7 @@ func (tx *Tx) Exec(query string, args ...interface{}) (res sql.Result, err error
tx.t.SetTag(trace.String(trace.TagAnnotation, fmt.Sprintf("exec %s", query))) tx.t.SetTag(trace.String(trace.TagAnnotation, fmt.Sprintf("exec %s", query)))
} }
res, err = tx.tx.ExecContext(tx.c, query, args...) res, err = tx.tx.ExecContext(tx.c, query, args...)
_metricReqDur.Observe(int64(time.Since(now)/time.Millisecond), tx.db.conf.Addr, tx.db.conf.Addr, "tx:exec") _metricReqDur.Observe(int64(time.Since(now)/time.Millisecond), tx.db.addr, tx.db.addr, "tx:exec")
if err != nil { if err != nil {
err = errors.Wrapf(err, "exec:%s, args:%+v", query, args) err = errors.Wrapf(err, "exec:%s, args:%+v", query, args)
} }
@ -597,7 +600,7 @@ func (tx *Tx) Query(query string, args ...interface{}) (rows *Rows, err error) {
now := time.Now() now := time.Now()
defer slowLog(fmt.Sprintf("Query query(%s) args(%+v)", query, args), now) defer slowLog(fmt.Sprintf("Query query(%s) args(%+v)", query, args), now)
defer func() { defer func() {
_metricReqDur.Observe(int64(time.Since(now)/time.Millisecond), tx.db.conf.Addr, tx.db.conf.Addr, "tx:query") _metricReqDur.Observe(int64(time.Since(now)/time.Millisecond), tx.db.addr, tx.db.addr, "tx:query")
}() }()
rs, err := tx.tx.QueryContext(tx.c, query, args...) rs, err := tx.tx.QueryContext(tx.c, query, args...)
if err == nil { if err == nil {
@ -618,7 +621,7 @@ func (tx *Tx) QueryRow(query string, args ...interface{}) *Row {
now := time.Now() now := time.Now()
defer slowLog(fmt.Sprintf("QueryRow query(%s) args(%+v)", query, args), now) defer slowLog(fmt.Sprintf("QueryRow query(%s) args(%+v)", query, args), now)
defer func() { defer func() {
_metricReqDur.Observe(int64(time.Since(now)/time.Millisecond), tx.db.conf.Addr, tx.db.conf.Addr, "tx:queryrow") _metricReqDur.Observe(int64(time.Since(now)/time.Millisecond), tx.db.addr, tx.db.addr, "tx:queryrow")
}() }()
r := tx.tx.QueryRowContext(tx.c, query, args...) r := tx.tx.QueryRowContext(tx.c, query, args...)
return &Row{Row: r, db: tx.db, query: query, args: args} return &Row{Row: r, db: tx.db, query: query, args: args}

Write Preview
Loading…
Cancel
Save