unistack-org/micro-wrapper-sql
4
0
Fork 1
Code Issues 3 Pull Requests Actions Packages Projects Releases 22 Wiki Activity

WIP: rewrite wrapper #53

Merged
vtolstov merged 5 commits from improvements into v3 2023-01-06 23:28:03 +03:00
Conversation 0 Commits 5 Files Changed 11 +251 -36
7 changed files with 251 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 43ee6ceb81 - Show all commits

136
conn.go
View File

@ -18,16 +18,26 @@ func (w *wrapperConn) Prepare(query string) (driver.Stmt, error) {
labels := []string{labelMethod, "Prepare", labelQuery, labelUnknown}
ts := time.Now()
stmt, err := w.conn.Prepare(query)
te := time.Since(ts).Seconds()
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(context.TODO(), "Prepare", labelUnknown, td, err)...).Log(context.TODO(), w.opts.LoggerLevel)
}
return nil, err
}
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelSuccess)...).Inc()
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(context.TODO(), "QueryContext", labelUnknown, td, err)...).Log(context.TODO(), w.opts.LoggerLevel)
}
return &wrapperStmt{stmt: stmt, opts: w.opts}, nil
}
@ -36,7 +46,8 @@ func (w *wrapperConn) Close() error {
labels := []string{labelMethod, "Close"}
ts := time.Now()
err := w.conn.Close()
te := time.Since(ts).Seconds()
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
} else {
@ -44,6 +55,11 @@ func (w *wrapperConn) Close() error {
}
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(context.TODO(), "QueryContext", labelUnknown, td, err)...).Log(context.TODO(), w.opts.LoggerLevel)
}
return err
}
@ -53,16 +69,23 @@ func (w *wrapperConn) Begin() (driver.Tx, error) {
ts := time.Now()
// nolint:staticcheck
tx, err := w.conn.Begin()
te := time.Since(ts).Seconds()
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(context.TODO(), "Begin", labelUnknown, td, err)...).Log(context.TODO(), w.opts.LoggerLevel)
}
return nil, err
}
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelSuccess)...).Inc()
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(context.TODO(), "Begin", labelUnknown, td, err)...).Log(context.TODO(), w.opts.LoggerLevel)
}
return &wrapperTx{tx: tx, opts: w.opts}, nil
}
@ -80,22 +103,30 @@ func (w *wrapperConn) BeginTx(ctx context.Context, opts driver.TxOptions) (drive
if connBeginTx, ok := w.conn.(driver.ConnBeginTx); ok {
ts := time.Now()
tx, err := connBeginTx.BeginTx(nctx, opts)
te := time.Since(ts).Seconds()
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
span.AddLabels("error", true)
span.AddLabels("err", err.Error())
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(context.TODO(), "BeginTx", name, td, err)...).Log(context.TODO(), w.opts.LoggerLevel)
}
return nil, err
}
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelSuccess)...).Inc()
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(context.TODO(), "BeginTx", name, td, err)...).Log(context.TODO(), w.opts.LoggerLevel)
}
return &wrapperTx{tx: tx, opts: w.opts, span: span}, nil
}
ts := time.Now()
// nolint:staticcheck
tx, err := w.conn.Begin()
te := time.Since(ts).Seconds()
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
@ -106,6 +137,9 @@ func (w *wrapperConn) BeginTx(ctx context.Context, opts driver.TxOptions) (drive
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelSuccess)...).Inc()
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(context.TODO(), "BeginTx", name, td, err)...).Log(context.TODO(), w.opts.LoggerLevel)
}
return tx, nil
}
@ -120,26 +154,34 @@ func (w *wrapperConn) PrepareContext(ctx context.Context, query string) (driver.
name = labelUnknown
}
labels := []string{labelMethod, "PrepareContext", labelQuery, name}
if connPrepareContext, ok := w.conn.(driver.ConnPrepareContext); ok {
if conn, ok := w.conn.(driver.ConnPrepareContext); ok {
ts := time.Now()
stmt, err := connPrepareContext.PrepareContext(nctx, query)
te := time.Since(ts).Seconds()
stmt, err := conn.PrepareContext(nctx, query)
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
span.AddLabels("error", true)
span.AddLabels("err", err.Error())
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(context.TODO(), "PrepareContext", name, td, err)...).Log(context.TODO(), w.opts.LoggerLevel)
}
return nil, err
}
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelSuccess)...).Inc()
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(context.TODO(), "PrepareContext", name, td, err)...).Log(context.TODO(), w.opts.LoggerLevel)
}
return &wrapperStmt{stmt: stmt, opts: w.opts}, nil
}
ts := time.Now()
stmt, err := w.conn.Prepare(query)
te := time.Since(ts).Seconds()
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
@ -150,6 +192,9 @@ func (w *wrapperConn) PrepareContext(ctx context.Context, query string) (driver.
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelSuccess)...).Inc()
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(context.TODO(), "PrepareContext", name, td, err)...).Log(context.TODO(), w.opts.LoggerLevel)
}
return stmt, nil
}
@ -160,7 +205,8 @@ func (w *wrapperConn) Exec(query string, args []driver.Value) (driver.Result, er
if execer, ok := w.conn.(driver.Execer); ok {
ts := time.Now()
res, err := execer.Exec(query, args)
te := time.Since(ts).Seconds()
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
} else {
@ -168,8 +214,14 @@ func (w *wrapperConn) Exec(query string, args []driver.Value) (driver.Result, er
}
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(context.TODO(), "Exec", labelUnknown, td, err)...).Log(context.TODO(), w.opts.LoggerLevel)
}
return res, err
}
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(context.TODO(), "Exec", labelUnknown, 0, ErrUnsupported)...).Log(context.TODO(), w.opts.LoggerLevel)
}
return nil, ErrUnsupported
}
@ -188,10 +240,11 @@ func (w *wrapperConn) ExecContext(ctx context.Context, query string, args []driv
span.AddLabels("args", fmt.Sprintf("%v", namedValueToLabels(args)))
}
labels := []string{labelMethod, "ExecContext", labelQuery, name}
if execerContext, ok := w.conn.(driver.ExecerContext); ok {
if conn, ok := w.conn.(driver.ExecerContext); ok {
ts := time.Now()
res, err := execerContext.ExecContext(nctx, query, args)
te := time.Since(ts).Seconds()
res, err := conn.ExecContext(nctx, query, args)
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
span.AddLabels("error", true)
@ -202,18 +255,25 @@ func (w *wrapperConn) ExecContext(ctx context.Context, query string, args []driv
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelSuccess)...).Inc()
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(context.TODO(), "ExecContext", name, td, err)...).Log(context.TODO(), w.opts.LoggerLevel)
}
return res, err
}
values, err := namedValueToValue(args)
if err != nil {
span.AddLabels("error", true)
span.AddLabels("err", err.Error())
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(context.TODO(), "ExecContext", labelUnknown, 0, err)...).Log(context.TODO(), w.opts.LoggerLevel)
}
return nil, err
}
ts := time.Now()
// nolint:staticcheck
res, err := w.Exec(query, values)
te := time.Since(ts).Seconds()
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
span.AddLabels("error", true)
@ -223,22 +283,29 @@ func (w *wrapperConn) ExecContext(ctx context.Context, query string, args []driv
}
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(context.TODO(), "ExecContext", name, td, err)...).Log(context.TODO(), w.opts.LoggerLevel)
}
return res, err
}
// Ping implements driver.Pinger Ping
func (w *wrapperConn) Ping(ctx context.Context) error {
if pinger, ok := w.conn.(driver.Pinger); ok {
if conn, ok := w.conn.(driver.Pinger); ok {
nctx, span := w.opts.Tracer.Start(ctx, "Ping")
defer span.Finish()
labels := []string{labelMethod, "Ping"}
ts := time.Now()
err := pinger.Ping(nctx)
te := time.Since(ts).Seconds()
err := conn.Ping(nctx)
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
span.AddLabels("error", true)
span.AddLabels("err", err.Error())
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(context.TODO(), "Ping", labelUnknown, td, err)...).Log(context.TODO(), w.opts.LoggerLevel)
}
return err
} else {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelSuccess)...).Inc()
@ -246,17 +313,21 @@ func (w *wrapperConn) Ping(ctx context.Context) error {
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
}
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(context.TODO(), "Ping", labelUnknown, 0, ErrUnsupported)...).Log(context.TODO(), w.opts.LoggerLevel)
}
return ErrUnsupported
}
// Query implements driver.Queryer Query
func (w *wrapperConn) Query(query string, args []driver.Value) (driver.Rows, error) {
// nolint:staticcheck
if queryer, ok := w.conn.(driver.Queryer); ok {
if conn, ok := w.conn.(driver.Queryer); ok {
labels := []string{labelMethod, "Query", labelQuery, labelUnknown}
ts := time.Now()
rows, err := queryer.Query(query, args)
te := time.Since(ts).Seconds()
rows, err := conn.Query(query, args)
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
} else {
@ -264,8 +335,14 @@ func (w *wrapperConn) Query(query string, args []driver.Value) (driver.Rows, err
}
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(context.TODO(), "Query", labelUnknown, td, err)...).Log(context.TODO(), w.opts.LoggerLevel)
}
return rows, err
}
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(context.TODO(), "Query", labelUnknown, 0, ErrUnsupported)...).Log(context.TODO(), w.opts.LoggerLevel)
}
return nil, ErrUnsupported
}
@ -284,10 +361,11 @@ func (w *wrapperConn) QueryContext(ctx context.Context, query string, args []dri
span.AddLabels("args", fmt.Sprintf("%v", namedValueToLabels(args)))
}
labels := []string{labelMethod, "QueryContext", labelQuery, name}
if queryerContext, ok := w.conn.(driver.QueryerContext); ok {
if conn, ok := w.conn.(driver.QueryerContext); ok {
ts := time.Now()
rows, err := queryerContext.QueryContext(nctx, query, args)
te := time.Since(ts).Seconds()
rows, err := conn.QueryContext(nctx, query, args)
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
span.AddLabels("error", true)
@ -297,18 +375,25 @@ func (w *wrapperConn) QueryContext(ctx context.Context, query string, args []dri
}
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(context.TODO(), "QueryContext", name, td, err)...).Log(context.TODO(), w.opts.LoggerLevel)
}
return rows, err
}
values, err := namedValueToValue(args)
if err != nil {
span.AddLabels("error", true)
span.AddLabels("err", err.Error())
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(context.TODO(), "QueryContext", name, 0, err)...).Log(context.TODO(), w.opts.LoggerLevel)
}
return nil, err
}
ts := time.Now()
// nolint:staticcheck
rows, err := w.Query(query, values)
te := time.Since(ts).Seconds()
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
span.AddLabels("error", true)
@ -318,5 +403,8 @@ func (w *wrapperConn) QueryContext(ctx context.Context, query string, args []dri
}
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(context.TODO(), "QueryContext", name, td, err)...).Log(context.TODO(), w.opts.LoggerLevel)
}
return rows, err
}

10
driver.go
View File

@ -1,7 +1,9 @@
package wrapper
import (
"context"
"database/sql/driver"
"time"
)
// wrapperDriver defines a wrapper for driver.Driver
@ -17,9 +19,17 @@ func NewWrapper(d driver.Driver, opts ...Option) driver.Driver {
// Open implements driver.Driver Open
func (w *wrapperDriver) Open(name string) (driver.Conn, error) {
ts := time.Now()
c, err := w.driver.Open(name)
td := time.Since(ts)
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(context.TODO(), "Open", labelUnknown, td, err)...).Log(context.TODO(), w.opts.LoggerLevel)
}
if err != nil {
return nil, err
}
return &wrapperConn{conn: c, opts: w.opts}, nil
}

2
go.mod
View File

@ -2,4 +2,4 @@ module go.unistack.org/micro-wrapper-sql/v3
go 1.16
require go.unistack.org/micro/v3 v3.9.17
require go.unistack.org/micro/v3 v3.9.19

4
go.sum
View File

@ -74,8 +74,8 @@ github.com/xeipuuv/gojsonschema v1.2.0/go.mod h1:anYRn/JVcOK2ZgGU+IjEV4nwlhoK5sQ
go.opentelemetry.io/proto/otlp v0.7.0/go.mod h1:PqfVotwruBrMGOCsRd/89rSnXhoiJIqeYNgFYFoEGnI=
go.unistack.org/micro-proto/v3 v3.3.1 h1:nQ0MtWvP2G3QrpOgawVOPhpZZYkq6umTGDqs8FxJYIo=
go.unistack.org/micro-proto/v3 v3.3.1/go.mod h1:cwRyv8uInM2I7EbU7O8Fx2Ls3N90Uw9UCCcq4olOdfE=
go.unistack.org/micro/v3 v3.9.17 h1:EJ9/XR9OTo/up/3aqWjaKS2YsWMA66b0dx+pc/0vIl8=
go.unistack.org/micro/v3 v3.9.17/go.mod h1:gI4RkJKHLPW7KV6h4+ZBOZD997MRvFRXMPQIHpozikI=
go.unistack.org/micro/v3 v3.9.19 h1:HFeHr20hnrs60EnlKWwrvQLTXILu3qhmHqs98jlqmqk=
go.unistack.org/micro/v3 v3.9.19/go.mod h1:gI4RkJKHLPW7KV6h4+ZBOZD997MRvFRXMPQIHpozikI=
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=

31
options.go
View File

@ -2,6 +2,7 @@ package wrapper
import (
"context"
"fmt"
"time"
"go.unistack.org/micro/v3/logger"
@ -14,6 +15,17 @@ var (
DefaultMeterStatsInterval = 5 * time.Second
// DefaultMeterMetricPrefix holds default metric prefix
DefaultMeterMetricPrefix = "micro_sql_"
// DefaultLoggerObserver used to prepare labels for logger
DefaultLoggerObserver = func(ctx context.Context, method string, query string, td time.Duration, err error) []interface{} {
labels := []interface{}{"method", method, "took", fmt.Sprintf("%v", td)}
if err != nil {
labels = append(labels, "error", err.Error())
}
if query != labelUnknown {
labels = append(labels, "query", query)
}
return labels
}
)
var (
@ -51,6 +63,8 @@ type Options struct {
MeterMetricPrefix string
MeterStatsInterval time.Duration
LoggerLevel logger.Level
LoggerEnabled bool
LoggerObserver func(ctx context.Context, method string, name string, td time.Duration, err error) []interface{}
}
// Option func signature
@ -65,6 +79,7 @@ func NewOptions(opts ...Option) Options {
MeterStatsInterval: DefaultMeterStatsInterval,
MeterMetricPrefix: DefaultMeterMetricPrefix,
LoggerLevel: logger.ErrorLevel,
LoggerObserver: DefaultLoggerObserver,
}
for _, o := range opts {
o(&options)
@ -78,6 +93,8 @@ func NewOptions(opts ...Option) Options {
),
)
options.Logger = options.Logger.Clone(logger.WithCallerSkipCount(1))
return options
}
@ -121,6 +138,13 @@ func Logger(l logger.Logger) Option {
}
}
// LoggerEnabled enable sql logging
func LoggerEnabled(b bool) Option {
return func(o *Options) {
o.LoggerEnabled = b
}
}
// LoggerLevel passes logger.Level option
func LoggerLevel(lvl logger.Level) Option {
return func(o *Options) {
@ -128,6 +152,13 @@ func LoggerLevel(lvl logger.Level) Option {
}
}
// LoggerObserver passes observer to fill logger fields
func LoggerObserver(obs func(context.Context, string, string, time.Duration, error) []interface{}) Option {
return func(o *Options) {
o.LoggerObserver = obs
}
}
// Tracer passes tracer.Tracer to wrapper
func Tracer(t tracer.Tracer) Option {
return func(o *Options) {

88
stmt.go
View File

@ -18,7 +18,8 @@ func (w *wrapperStmt) Close() error {
labels := []string{labelMethod, "Close"}
ts := time.Now()
err := w.stmt.Close()
te := time.Since(ts).Seconds()
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
} else {
@ -26,6 +27,10 @@ func (w *wrapperStmt) Close() error {
}
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(context.TODO(), "Close", labelUnknown, td, err)...).Log(context.TODO(), w.opts.LoggerLevel)
}
return err
}
@ -36,11 +41,12 @@ func (w *wrapperStmt) NumInput() int {
// Exec implements driver.Stmt Exec
func (w *wrapperStmt) Exec(args []driver.Value) (driver.Result, error) {
// nolint:staticcheck
labels := []string{labelMethod, "Exec"}
ts := time.Now()
// nolint:staticcheck
res, err := w.stmt.Exec(args)
te := time.Since(ts).Seconds()
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
} else {
@ -48,6 +54,10 @@ func (w *wrapperStmt) Exec(args []driver.Value) (driver.Result, error) {
}
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(context.TODO(), "Exec", labelUnknown, td, err)...).Log(context.TODO(), w.opts.LoggerLevel)
}
return res, err
}
@ -57,7 +67,8 @@ func (w *wrapperStmt) Query(args []driver.Value) (driver.Rows, error) {
ts := time.Now()
// nolint:staticcheck
rows, err := w.stmt.Query(args)
te := time.Since(ts).Seconds()
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
} else {
@ -65,6 +76,10 @@ func (w *wrapperStmt) Query(args []driver.Value) (driver.Rows, error) {
}
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(context.TODO(), "Query", labelUnknown, td, err)...).Log(context.TODO(), w.opts.LoggerLevel)
}
return rows, err
}
@ -83,10 +98,11 @@ func (w *wrapperStmt) ExecContext(ctx context.Context, query string, args []driv
span.AddLabels("args", fmt.Sprintf("%v", namedValueToLabels(args)))
}
labels := []string{labelMethod, "ExecContext", labelQuery, name}
if execerContext, ok := w.stmt.(driver.ExecerContext); ok {
if conn, ok := w.stmt.(driver.ExecerContext); ok {
ts := time.Now()
res, err := execerContext.ExecContext(nctx, query, args)
te := time.Since(ts).Seconds()
res, err := conn.ExecContext(nctx, query, args)
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
span.AddLabels("error", true)
@ -96,6 +112,10 @@ func (w *wrapperStmt) ExecContext(ctx context.Context, query string, args []driv
}
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(context.TODO(), "ExecContext", name, td, err)...).Log(context.TODO(), w.opts.LoggerLevel)
}
return res, err
}
values, err := namedValueToValue(args)
@ -103,13 +123,30 @@ func (w *wrapperStmt) ExecContext(ctx context.Context, query string, args []driv
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
span.AddLabels("error", true)
span.AddLabels("err", err.Error())
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(context.TODO(), "ExecContext", name, 0, err)...).Log(context.TODO(), w.opts.LoggerLevel)
}
return nil, err
}
ts := time.Now()
// nolint:staticcheck
res, err := w.Exec(values)
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
span.AddLabels("error", true)
span.AddLabels("err", err.Error())
} else {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelSuccess)...).Inc()
}
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(context.TODO(), "ExecContext", name, td, err)).Log(context.TODO(), w.opts.LoggerLevel)
}
return res, err
}
@ -128,25 +165,58 @@ func (w *wrapperStmt) QueryContext(ctx context.Context, query string, args []dri
if len(args) > 0 {
span.AddLabels("args", fmt.Sprintf("%v", namedValueToLabels(args)))
}
if queryerContext, ok := w.stmt.(driver.QueryerContext); ok {
rows, err := queryerContext.QueryContext(nctx, query, args)
labels := []string{labelMethod, "QueryContext", labelQuery, name}
if conn, ok := w.stmt.(driver.QueryerContext); ok {
ts := time.Now()
rows, err := conn.QueryContext(nctx, query, args)
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
span.AddLabels("error", true)
span.AddLabels("err", err.Error())
} else {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelSuccess)...).Inc()
}
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(context.TODO(), "QueryContext", name, td, err)).Log(context.TODO(), w.opts.LoggerLevel)
}
return rows, err
}
values, err := namedValueToValue(args)
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
span.AddLabels("error", true)
span.AddLabels("err", err.Error())
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(context.TODO(), "QueryContext", name, 0, err)...).Log(context.TODO(), w.opts.LoggerLevel)
}
return nil, err
}
ts := time.Now()
// nolint:staticcheck
rows, err := w.Query(values)
td := time.Since(ts)
te := td.Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
span.AddLabels("error", true)
span.AddLabels("err", err.Error())
} else {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelSuccess)...).Inc()
}
w.opts.Meter.Summary(meterRequestLatencyMicroseconds, labels...).Update(te)
w.opts.Meter.Histogram(meterRequestDurationSeconds, labels...).Update(te)
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(context.TODO(), "QueryContext", name, td, err)...).Log(context.TODO(), w.opts.LoggerLevel)
}
return rows, err
}

16
tx.go
View File

@ -1,7 +1,9 @@
package wrapper
import (
"context"
"database/sql/driver"
"time"
"go.unistack.org/micro/v3/tracer"
)
@ -18,7 +20,14 @@ func (w *wrapperTx) Commit() error {
if w.span != nil {
defer w.span.Finish()
}
ts := time.Now()
err := w.tx.Commit()
td := time.Since(ts)
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(context.TODO(), "Commit", labelUnknown, td, err)...).Log(context.TODO(), w.opts.LoggerLevel)
}
return err
}
@ -27,6 +36,13 @@ func (w *wrapperTx) Rollback() error {
if w.span != nil {
defer w.span.Finish()
}
ts := time.Now()
err := w.tx.Rollback()
td := time.Since(ts)
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(context.TODO(), "Rollback", labelUnknown, td, err)...).Log(context.TODO(), w.opts.LoggerLevel)
}
return err
}