9 Commits

Author SHA1 Message Date
bab2833cf7 Merge pull request 'add err skip' (#81) from devstigneev/micro-wrapper-sql:issue_73 into master
Some checks failed
build / test (push) Successful in 1m54s
build / lint (push) Successful in 9m17s
codeql / analyze (go) (push) Failing after 1m28s
Reviewed-on: #81
2024-04-11 11:40:28 +03:00
564efef60b moved compare error to LoggerObserver
Some checks failed
automerge / automerge (pull_request) Has been skipped
dependabot-automerge / automerge (pull_request) Has been skipped
autoapprove / autoapprove (pull_request) Successful in 8s
codeql / analyze (go) (pull_request) Has been cancelled
prbuild / test (pull_request) Has been cancelled
prbuild / lint (pull_request) Has been cancelled
2024-04-11 11:38:09 +03:00
cb9d32bf1c add err skip
Some checks failed
automerge / automerge (pull_request) Has been skipped
dependabot-automerge / automerge (pull_request) Has been skipped
autoapprove / autoapprove (pull_request) Successful in 8s
codeql / analyze (go) (pull_request) Has been cancelled
prbuild / test (pull_request) Has been cancelled
prbuild / lint (pull_request) Has been cancelled
2024-04-07 22:12:39 +03:00
c363033abd fix label names
Some checks failed
build / test (push) Failing after 1m29s
build / lint (push) Failing after 2m34s
codeql / analyze (go) (push) Failing after 3m22s
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2023-09-05 07:08:08 +03:00
dd914ed9fb Merge pull request 'updatev4' (#80) from updatev4 into master
Some checks failed
build / test (push) Failing after 1m28s
build / lint (push) Failing after 2m38s
codeql / analyze (go) (push) Failing after 3m21s
Reviewed-on: #80
2023-09-01 14:49:52 +03:00
7d2787c3a6 updates from testing
Some checks failed
automerge / automerge (pull_request) Failing after 4s
codeql / analyze (go) (pull_request) Failing after 3m20s
dependabot-automerge / automerge (pull_request) Has been skipped
prbuild / test (pull_request) Failing after 1m28s
prbuild / lint (pull_request) Failing after 2m34s
autoapprove / autoapprove (pull_request) Failing after 1m25s
2023-09-01 14:48:57 +03:00
a81da5bd2d fixup metrics and tracing
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2023-09-01 00:23:26 +03:00
44cd4a1e16 Merge pull request 'use generated interface' (#76) from generator into master
Some checks failed
build / test (push) Failing after 5s
build / lint (push) Failing after 5s
codeql / analyze (go) (push) Failing after 6s
Reviewed-on: #76
2023-06-10 13:10:15 +03:00
dbce4a1d9e use generated interface
Some checks failed
dependabot-automerge / automerge (pull_request) Has been skipped
autoapprove / autoapprove (pull_request) Failing after 4s
automerge / automerge (pull_request) Failing after 4s
codeql / analyze (go) (pull_request) Failing after 6s
prbuild / test (pull_request) Failing after 5s
prbuild / lint (pull_request) Failing after 6s
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2023-06-10 13:09:25 +03:00
14 changed files with 21291 additions and 350 deletions

9
.gitignore vendored
View File

@@ -4,6 +4,7 @@
*.dll
*.so
*.dylib
bin
# Test binary, built with `go test -c`
*.test
@@ -13,3 +14,11 @@
# Dependency directories (remove the comment below to include it)
# vendor/
# Go workspace file
go.work
# General
.DS_Store
.idea
.vscode

View File

@@ -4,8 +4,11 @@ import (
"database/sql/driver"
"errors"
"fmt"
"runtime"
)
//go:generate sh -c "go run gen.go > wrap_gen.go"
// namedValueToValue converts driver arguments of NamedValue format to Value format. Implemented in the same way as in
// database/sql ctxutil.go.
func namedValueToValue(named []driver.NamedValue) ([]driver.Value, error) {
@@ -29,7 +32,20 @@ func namedValueToLabels(named []driver.NamedValue) []interface{} {
} else {
name = fmt.Sprintf("$%d", param.Ordinal)
}
largs = append(largs, fmt.Sprintf("%s=%s", name, param.Value))
largs = append(largs, fmt.Sprintf("%s=%v", name, param.Value))
}
return largs
}
// getCallerName get the name of the function A where A() -> B() -> GetFunctionCallerName()
func getCallerName() string {
pc, _, _, ok := runtime.Caller(3)
details := runtime.FuncForPC(pc)
var callerName string
if ok && details != nil {
callerName = details.Name()
} else {
callerName = labelUnknown
}
return callerName
}

415
conn.go
View File

@@ -6,10 +6,24 @@ import (
"fmt"
"time"
"go.unistack.org/micro/v3/tracer"
requestid "go.unistack.org/micro-wrapper-requestid/v4"
"go.unistack.org/micro/v4/tracer"
)
var _ driver.Conn = &wrapperConn{}
var (
_ driver.Conn = (*wrapperConn)(nil)
_ driver.ConnBeginTx = (*wrapperConn)(nil)
_ driver.ConnPrepareContext = (*wrapperConn)(nil)
_ driver.Pinger = (*wrapperConn)(nil)
_ driver.Validator = (*wrapperConn)(nil)
_ driver.Queryer = (*wrapperConn)(nil) // nolint:staticcheck
_ driver.QueryerContext = (*wrapperConn)(nil)
_ driver.Execer = (*wrapperConn)(nil) // nolint:staticcheck
_ driver.ExecerContext = (*wrapperConn)(nil)
// _ driver.Connector
// _ driver.Driver
// _ driver.DriverContext
)
// wrapperConn defines a wrapper for driver.Conn
type wrapperConn struct {
@@ -18,41 +32,7 @@ type wrapperConn struct {
conn driver.Conn
opts Options
ctx context.Context
}
// Prepare implements driver.Conn Prepare
func (w *wrapperConn) Prepare(query string) (driver.Stmt, error) {
var ctx context.Context
if w.ctx != nil {
ctx = w.ctx
} else {
ctx = context.Background()
}
labels := []string{labelMethod, "Prepare", labelQuery, labelUnknown}
ts := time.Now()
stmt, err := w.conn.Prepare(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)
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(ctx, "Prepare", labelUnknown, td, err)...).Log(ctx, 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(ctx, "Prepare", labelUnknown, td, err)...).Log(ctx, w.opts.LoggerLevel)
}
return &wrapperStmt{stmt: stmt, opts: w.opts, ctx: ctx}, nil
span tracer.Span
}
// Close implements driver.Conn Close
@@ -77,8 +57,8 @@ 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(ctx, "Close", labelUnknown, td, err)...).Log(ctx, w.opts.LoggerLevel)
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Fields(w.opts.LoggerObserver(ctx, "Close", getCallerName(), td, err)...).Log(ctx, w.opts.LoggerLevel)
}
return err
@@ -95,125 +75,75 @@ func (w *wrapperConn) Begin() (driver.Tx, error) {
labels := []string{labelMethod, "Begin"}
ts := time.Now()
// nolint:staticcheck
tx, err := w.conn.Begin()
tx, err := w.conn.Begin() // nolint:staticcheck
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(ctx, "Begin", labelUnknown, td, err)...).Log(ctx, w.opts.LoggerLevel)
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Fields(w.opts.LoggerObserver(ctx, "Begin", getCallerName(), td, err)...).Log(ctx, 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(ctx, "Begin", labelUnknown, td, err)...).Log(ctx, w.opts.LoggerLevel)
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Fields(w.opts.LoggerObserver(ctx, "Begin", getCallerName(), td, err)...).Log(ctx, w.opts.LoggerLevel)
}
return &wrapperTx{tx: tx, opts: w.opts, ctx: ctx}, nil
}
// BeginTx implements driver.ConnBeginTx BeginTx
func (w *wrapperConn) BeginTx(ctx context.Context, opts driver.TxOptions) (driver.Tx, error) {
nctx, span := w.opts.Tracer.Start(ctx, "Transaction")
span.AddLabels("method", "BeginTx")
nctx, span := w.opts.Tracer.Start(ctx, "sdk.database", tracer.WithSpanKind(tracer.SpanKindClient))
span.AddLabels("db.method", "BeginTx")
name := getQueryName(ctx)
if name != "" {
span.AddLabels("query", name)
} else {
name = labelUnknown
span.AddLabels("db.statement", name)
if id, ok := ctx.Value(requestid.XRequestIDKey).(string); ok {
span.AddLabels("x-request-id", id)
}
labels := []string{labelMethod, "BeginTx", labelQuery, name}
if connBeginTx, ok := w.conn.(driver.ConnBeginTx); ok {
ts := time.Now()
tx, err := connBeginTx.BeginTx(nctx, opts)
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(ctx, "BeginTx", name, td, err)...).Log(ctx, 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(ctx, "BeginTx", name, td, err)...).Log(ctx, w.opts.LoggerLevel)
}
w.ctx = nctx
return &wrapperTx{ctx: ctx, tx: tx, opts: w.opts, span: span, conn: w}, nil
connBeginTx, ok := w.conn.(driver.ConnBeginTx)
if !ok {
return w.Begin()
}
ts := time.Now()
// nolint:staticcheck
tx, err := w.conn.Begin()
tx, err := connBeginTx.BeginTx(nctx, opts)
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())
span.SetStatus(tracer.SpanStatusError, err.Error())
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Fields(w.opts.LoggerObserver(ctx, "BeginTx", getCallerName(), td, err)...).Log(ctx, 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(ctx, "BeginTx", name, td, err)...).Log(ctx, w.opts.LoggerLevel)
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Fields(w.opts.LoggerObserver(ctx, "BeginTx", getCallerName(), td, err)...).Log(ctx, w.opts.LoggerLevel)
}
w.ctx = nctx
return &wrapperTx{ctx: ctx, tx: tx, opts: w.opts, span: span, conn: w}, nil
return &wrapperTx{tx: tx, opts: w.opts, ctx: ctx, span: span}, nil
}
// PrepareContext implements driver.ConnPrepareContext PrepareContext
func (w *wrapperConn) PrepareContext(ctx context.Context, query string) (driver.Stmt, error) {
var nctx context.Context
var span tracer.Span
// Prepare implements driver.Conn Prepare
func (w *wrapperConn) Prepare(query string) (driver.Stmt, error) {
var ctx context.Context
if w.ctx != nil {
nctx, span = w.opts.Tracer.Start(w.ctx, "PrepareContext")
ctx = w.ctx
} else {
nctx, span = w.opts.Tracer.Start(ctx, "PrepareContext")
}
span.AddLabels("method", "PrepareContext")
name := getQueryName(ctx)
if name != "" {
span.AddLabels("query", name)
} else {
name = labelUnknown
}
labels := []string{labelMethod, "PrepareContext", labelQuery, name}
if conn, ok := w.conn.(driver.ConnPrepareContext); ok {
ts := time.Now()
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(ctx, "PrepareContext", name, td, err)...).Log(ctx, 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(ctx, "PrepareContext", name, td, err)...).Log(ctx, w.opts.LoggerLevel)
}
return &wrapperStmt{stmt: stmt, opts: w.opts, ctx: nctx}, nil
ctx = context.Background()
}
labels := []string{labelMethod, "Prepare", labelQuery, getCallerName()}
ts := time.Now()
stmt, err := w.conn.Prepare(query)
td := time.Since(ts)
@@ -222,16 +152,66 @@ func (w *wrapperConn) PrepareContext(ctx context.Context, query string) (driver.
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.V(w.opts.LoggerLevel) {
w.opts.Logger.Fields(w.opts.LoggerObserver(ctx, "Prepare", getCallerName(), td, err)...).Log(ctx, 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(ctx, "PrepareContext", name, td, err)...).Log(ctx, w.opts.LoggerLevel)
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Fields(w.opts.LoggerObserver(ctx, "Prepare", getCallerName(), td, err)...).Log(ctx, w.opts.LoggerLevel)
}
return stmt, nil
return wrapStmt(stmt, query, w.opts), nil
}
// PrepareContext implements driver.ConnPrepareContext PrepareContext
func (w *wrapperConn) PrepareContext(ctx context.Context, query string) (driver.Stmt, error) {
var nctx context.Context
var span tracer.Span
if w.ctx != nil {
nctx, span = w.opts.Tracer.Start(w.ctx, "sdk.database", tracer.WithSpanKind(tracer.SpanKindClient))
} else {
nctx, span = w.opts.Tracer.Start(ctx, "sdk.database", tracer.WithSpanKind(tracer.SpanKindClient))
}
span.AddLabels("db.method", "PrepareContext")
name := getQueryName(ctx)
span.AddLabels("db.statement", name)
if id, ok := ctx.Value(requestid.XRequestIDKey).(string); ok {
span.AddLabels("x-request-id", id)
}
labels := []string{labelMethod, "PrepareContext", labelQuery, name}
conn, ok := w.conn.(driver.ConnPrepareContext)
if !ok {
return w.Prepare(query)
}
ts := time.Now()
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.SetStatus(tracer.SpanStatusError, err.Error())
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Fields(w.opts.LoggerObserver(ctx, "PrepareContext", getCallerName(), td, err)...).Log(ctx, 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.V(w.opts.LoggerLevel) {
w.opts.Logger.Fields(w.opts.LoggerObserver(ctx, "PrepareContext", getCallerName(), td, err)...).Log(ctx, w.opts.LoggerLevel)
}
return wrapStmt(stmt, query, w.opts), nil
}
// Exec implements driver.Execer Exec
@@ -242,15 +222,16 @@ func (w *wrapperConn) Exec(query string, args []driver.Value) (driver.Result, er
} else {
ctx = context.Background()
}
labels := []string{labelMethod, "Exec", labelQuery, getCallerName()}
// nolint:staticcheck
execer, ok := w.conn.(driver.Execer)
conn, ok := w.conn.(driver.Execer)
if !ok {
return nil, driver.ErrSkip
}
labels := []string{labelMethod, "Exec", labelQuery, labelUnknown}
ts := time.Now()
res, err := execer.Exec(query, args)
res, err := conn.Exec(query, args)
td := time.Since(ts)
te := td.Seconds()
if err != nil {
@@ -260,8 +241,8 @@ 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(ctx, "Exec", labelUnknown, td, err)...).Log(ctx, w.opts.LoggerLevel)
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Fields(w.opts.LoggerObserver(ctx, "Exec", getCallerName(), td, err)...).Log(ctx, w.opts.LoggerLevel)
}
return res, err
}
@@ -270,89 +251,69 @@ func (w *wrapperConn) Exec(query string, args []driver.Value) (driver.Result, er
func (w *wrapperConn) ExecContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Result, error) {
var nctx context.Context
var span tracer.Span
if w.ctx != nil {
nctx, span = w.opts.Tracer.Start(w.ctx, "ExecContext")
nctx, span = w.opts.Tracer.Start(w.ctx, "sdk.database", tracer.WithSpanKind(tracer.SpanKindClient))
} else {
nctx, span = w.opts.Tracer.Start(ctx, "ExecContext")
nctx, span = w.opts.Tracer.Start(ctx, "sdk.database", tracer.WithSpanKind(tracer.SpanKindClient))
}
span.AddLabels("method", "ExecContext")
span.AddLabels("db.method", "ExecContext")
name := getQueryName(ctx)
if name != "" {
span.AddLabels("query", name)
} else {
name = labelUnknown
span.AddLabels("db.statement", name)
if id, ok := ctx.Value(requestid.XRequestIDKey).(string); ok {
span.AddLabels("x-request-id", id)
}
defer span.Finish()
if len(args) > 0 {
span.AddLabels("args", fmt.Sprintf("%v", namedValueToLabels(args)))
span.AddLabels("db.args", fmt.Sprintf("%v", namedValueToLabels(args)))
}
labels := []string{labelMethod, "ExecContext", labelQuery, name}
if conn, ok := w.conn.(driver.ExecerContext); ok {
ts := time.Now()
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)
span.AddLabels("err", err.Error())
} else {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelSuccess)...).Inc()
}
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(ctx, "ExecContext", name, td, err)...).Log(ctx, 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(ctx, "ExecContext", labelUnknown, 0, err)...).Log(ctx, w.opts.LoggerLevel)
}
return nil, err
conn, ok := w.conn.(driver.ExecerContext)
if !ok {
// nolint:staticcheck
return nil, driver.ErrSkip
}
ts := time.Now()
// nolint:staticcheck
res, err := w.Exec(query, values)
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)
span.AddLabels("err", err.Error())
span.SetStatus(tracer.SpanStatusError, err.Error())
} else {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelSuccess)...).Inc()
}
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(ctx, "ExecContext", name, td, err)...).Log(ctx, w.opts.LoggerLevel)
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Fields(w.opts.LoggerObserver(ctx, "ExecContext", getCallerName(), td, err)...).Log(ctx, w.opts.LoggerLevel)
}
return res, err
}
// Ping implements driver.Pinger Ping
func (w *wrapperConn) Ping(ctx context.Context) error {
conn, ok := w.conn.(driver.Pinger)
if !ok {
wc, err := w.d.Open(w.dname)
// fallback path to check db alive
pc, err := w.d.Open(w.dname)
if err != nil {
return err
}
return wc.Close()
return pc.Close()
}
var nctx context.Context
var span tracer.Span
if w.ctx != nil {
nctx, span = w.opts.Tracer.Start(w.ctx, "Ping")
nctx, span = w.opts.Tracer.Start(w.ctx, "sdk.database", tracer.WithSpanKind(tracer.SpanKindClient))
} else {
nctx, span = w.opts.Tracer.Start(ctx, "Ping")
nctx, span = w.opts.Tracer.Start(ctx, "sdk.database", tracer.WithSpanKind(tracer.SpanKindClient))
}
defer span.Finish()
labels := []string{labelMethod, "Ping"}
@@ -362,10 +323,9 @@ func (w *wrapperConn) Ping(ctx context.Context) error {
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(ctx, "Ping", labelUnknown, td, err)...).Log(ctx, w.opts.LoggerLevel)
span.SetStatus(tracer.SpanStatusError, err.Error())
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Fields(w.opts.LoggerObserver(ctx, "Ping", getCallerName(), td, err)...).Log(ctx, w.opts.LoggerLevel)
}
return err
} else {
@@ -385,14 +345,13 @@ func (w *wrapperConn) Query(query string, args []driver.Value) (driver.Rows, err
} else {
ctx = context.Background()
}
//nolint:staticcheck
// nolint:staticcheck
conn, ok := w.conn.(driver.Queryer)
if !ok {
return nil, driver.ErrSkip
}
labels := []string{labelMethod, "Query", labelQuery, labelUnknown}
labels := []string{labelMethod, "Query", labelQuery, getCallerName()}
ts := time.Now()
rows, err := conn.Query(query, args)
td := time.Since(ts)
@@ -404,8 +363,8 @@ 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(ctx, "Query", labelUnknown, td, err)...).Log(ctx, w.opts.LoggerLevel)
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Fields(w.opts.LoggerObserver(ctx, "Query", getCallerName(), td, err)...).Log(ctx, w.opts.LoggerLevel)
}
return rows, err
}
@@ -414,67 +373,69 @@ func (w *wrapperConn) Query(query string, args []driver.Value) (driver.Rows, err
func (w *wrapperConn) QueryContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Rows, error) {
var nctx context.Context
var span tracer.Span
if w.ctx != nil {
nctx, span = w.opts.Tracer.Start(w.ctx, "QueryContext")
nctx, span = w.opts.Tracer.Start(w.ctx, "sdk.database", tracer.WithSpanKind(tracer.SpanKindClient))
} else {
nctx, span = w.opts.Tracer.Start(ctx, "QueryContext")
nctx, span = w.opts.Tracer.Start(ctx, "sdk.database", tracer.WithSpanKind(tracer.SpanKindClient))
}
span.AddLabels("method", "QueryContext")
span.AddLabels("db.method", "QueryContext")
name := getQueryName(ctx)
if name != "" {
span.AddLabels("query", name)
} else {
name = labelUnknown
span.AddLabels("db.statement", name)
if id, ok := ctx.Value(requestid.XRequestIDKey).(string); ok {
span.AddLabels("x-request-id", id)
}
defer span.Finish()
if len(args) > 0 {
span.AddLabels("args", fmt.Sprintf("%v", namedValueToLabels(args)))
span.AddLabels("db.args", fmt.Sprintf("%v", namedValueToLabels(args)))
}
labels := []string{labelMethod, "QueryContext", labelQuery, name}
if conn, ok := w.conn.(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(ctx, "QueryContext", name, td, err)...).Log(ctx, 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(ctx, "QueryContext", name, 0, err)...).Log(ctx, w.opts.LoggerLevel)
}
return nil, err
conn, ok := w.conn.(driver.QueryerContext)
if !ok {
return nil, driver.ErrSkip
}
ts := time.Now()
// nolint:staticcheck
rows, err := w.Query(query, values)
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())
span.SetStatus(tracer.SpanStatusError, 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(ctx, "QueryContext", name, td, err)...).Log(ctx, w.opts.LoggerLevel)
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Fields(w.opts.LoggerObserver(ctx, "QueryContext", getCallerName(), td, err)...).Log(ctx, w.opts.LoggerLevel)
}
return rows, err
}
// CheckNamedValue implements driver.NamedValueChecker
func (w *wrapperConn) CheckNamedValue(v *driver.NamedValue) error {
s, ok := w.conn.(driver.NamedValueChecker)
if !ok {
return driver.ErrSkip
}
return s.CheckNamedValue(v)
}
// IsValid implements driver.Validator
func (w *wrapperConn) IsValid() bool {
v, ok := w.conn.(driver.Validator)
if !ok {
return w.conn != nil
}
return v.IsValid()
}
func (w *wrapperConn) ResetSession(ctx context.Context) error {
s, ok := w.conn.(driver.SessionResetter)
if !ok {
return driver.ErrSkip
}
return s.ResetSession(ctx)
}

2
doc.go
View File

@@ -1,2 +1,2 @@
// package wrapper provides SQL driver wrapper with micro tracing, logging, metering capabilities
package wrapper // import "go.unistack.org/micro-wrapper-sql/v3"
package wrapper

View File

@@ -6,6 +6,22 @@ import (
"time"
)
var (
// _ driver.DriverContext = (*wrapperDriver)(nil)
// _ driver.Connector = (*wrapperDriver)(nil)
)
type conn interface {
driver.Pinger
driver.Execer
driver.ExecerContext
driver.Queryer
driver.QueryerContext
driver.Conn
driver.ConnPrepareContext
driver.ConnBeginTx
}
// wrapperDriver defines a wrapper for driver.Driver
type wrapperDriver struct {
driver driver.Driver
@@ -18,19 +34,57 @@ func NewWrapper(d driver.Driver, opts ...Option) driver.Driver {
return &wrapperDriver{driver: d, opts: NewOptions(opts...), ctx: context.Background()}
}
type wrappedConnector struct {
connector driver.Connector
name string
opts Options
ctx context.Context
}
func NewWrapperConnector(c driver.Connector, opts ...Option) driver.Connector {
return &wrappedConnector{connector: c, opts: NewOptions(opts...), ctx: context.Background()}
}
// Connect implements driver.Driver Connect
func (w *wrappedConnector) Connect(ctx context.Context) (driver.Conn, error) {
return w.connector.Connect(ctx)
}
// Driver implements driver.Driver Driver
func (w *wrappedConnector) Driver() driver.Driver {
return w.connector.Driver()
}
/*
// Connect implements driver.Driver OpenConnector
func (w *wrapperDriver) OpenConnector(name string) (driver.Conn, error) {
return &wrapperConnector{driver: w.driver, name: name, opts: w.opts}, nil
}
*/
// Open implements driver.Driver Open
func (w *wrapperDriver) Open(name string) (driver.Conn, error) {
// ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) // Ensure eventual timeout
// defer cancel()
/*
connector, err := w.OpenConnector(name)
if err != nil {
return nil, err
}
return connector.Connect(ctx)
*/
ts := time.Now()
c, err := w.driver.Open(name)
td := time.Since(ts)
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(w.ctx, "Open", labelUnknown, td, err)...).Log(w.ctx, w.opts.LoggerLevel)
w.opts.Logger.Fields(w.opts.LoggerObserver(w.ctx, "Open", getCallerName(), td, err)...).Log(w.ctx, w.opts.LoggerLevel)
}
if err != nil {
return nil, err
}
return &wrapperConn{d: w, dname: name, conn: c, opts: w.opts, ctx: w.ctx}, nil
return wrapConn(c, w.opts), nil
}

167
gen.go Normal file
View File

@@ -0,0 +1,167 @@
//go:build ignore
package main
import (
"bytes"
"crypto/md5"
"fmt"
"io"
"sort"
"strings"
)
var connIfaces = []string{
"driver.ConnBeginTx",
"driver.ConnPrepareContext",
"driver.Execer",
"driver.ExecerContext",
"driver.NamedValueChecker",
"driver.Pinger",
"driver.Queryer",
"driver.QueryerContext",
"driver.SessionResetter",
"driver.Validator",
}
var stmtIfaces = []string{
"driver.StmtExecContext",
"driver.StmtQueryContext",
"driver.ColumnConverter",
"driver.NamedValueChecker",
}
func getHash(s []string) string {
h := md5.New()
io.WriteString(h, strings.Join(s, "|"))
return fmt.Sprintf("%x", h.Sum(nil))
}
func main() {
comboConn := all(connIfaces)
sort.Slice(comboConn, func(i, j int) bool {
return len(comboConn[i]) < len(comboConn[j])
})
comboStmt := all(stmtIfaces)
sort.Slice(comboStmt, func(i, j int) bool {
return len(comboStmt[i]) < len(comboStmt[j])
})
b := bytes.NewBuffer(nil)
b.WriteString("// Code generated. DO NOT EDIT.\n\n")
b.WriteString("package wrapper\n\n")
b.WriteString(`import "database/sql/driver"`)
b.WriteString("\n\n")
b.WriteString("func wrapConn(dc driver.Conn, opts Options) driver.Conn {\n")
b.WriteString("\tc := &wrapperConn{conn: dc, opts: opts}\n")
for idx := len(comboConn) - 1; idx >= 0; idx-- {
ifaces := comboConn[idx]
n := len(ifaces)
if n == 0 {
continue
}
h := getHash(ifaces)
b.WriteString(fmt.Sprintf("\tif _, ok := dc.(wrapConn%04d_%s); ok {\n", n, h))
b.WriteString("\treturn struct {\n")
b.WriteString("\t\tdriver.Conn\n")
b.WriteString(fmt.Sprintf("\t\t\t%s", strings.Join(ifaces, "\n\t\t\t")))
b.WriteString("\t\t\n}{")
for idx := range ifaces {
if idx > 0 {
b.WriteString(", ")
b.WriteString("c")
} else if idx == 0 {
b.WriteString("c")
} else {
b.WriteString("c")
}
}
b.WriteString(", c}\n")
b.WriteString("}\n\n")
}
b.WriteString("return c\n")
b.WriteString("}\n")
for idx := len(comboConn) - 1; idx >= 0; idx-- {
ifaces := comboConn[idx]
n := len(ifaces)
if n == 0 {
continue
}
h := getHash(ifaces)
b.WriteString(fmt.Sprintf("// %s\n", strings.Join(ifaces, "|")))
b.WriteString(fmt.Sprintf("type wrapConn%04d_%s interface {\n", n, h))
for _, iface := range ifaces {
b.WriteString(fmt.Sprintf("\t%s\n", iface))
}
b.WriteString("}\n\n")
}
b.WriteString("func wrapStmt(stmt driver.Stmt, query string, opts Options) driver.Stmt {\n")
b.WriteString("\tc := &wrapperStmt{stmt: stmt, query: query, opts: opts}\n")
for idx := len(comboStmt) - 1; idx >= 0; idx-- {
ifaces := comboStmt[idx]
n := len(ifaces)
if n == 0 {
continue
}
h := getHash(ifaces)
b.WriteString(fmt.Sprintf("\tif _, ok := stmt.(wrapStmt%04d_%s); ok {\n", n, h))
b.WriteString("\treturn struct {\n")
b.WriteString("\t\tdriver.Stmt\n")
b.WriteString(fmt.Sprintf("\t\t\t%s", strings.Join(ifaces, "\n\t\t\t")))
b.WriteString("\t\t\n}{")
for idx := range ifaces {
if idx > 0 {
b.WriteString(", ")
b.WriteString("c")
} else if idx == 0 {
b.WriteString("c")
} else {
b.WriteString("c")
}
}
b.WriteString(", c}\n")
b.WriteString("}\n\n")
}
b.WriteString("return c\n")
b.WriteString("}\n")
for idx := len(comboStmt) - 1; idx >= 0; idx-- {
ifaces := comboStmt[idx]
n := len(ifaces)
if n == 0 {
continue
}
h := getHash(ifaces)
b.WriteString(fmt.Sprintf("// %s\n", strings.Join(ifaces, "|")))
b.WriteString(fmt.Sprintf("type wrapStmt%04d_%s interface {\n", n, h))
for _, iface := range ifaces {
b.WriteString(fmt.Sprintf("\t%s\n", iface))
}
b.WriteString("}\n\n")
}
fmt.Printf("%s\n", b.String())
}
// all returns all combinations for a given string array.
func all[T any](set []T) (subsets [][]T) {
length := uint(len(set))
for subsetBits := 1; subsetBits < (1 << length); subsetBits++ {
var subset []T
for object := uint(0); object < length; object++ {
if (subsetBits>>object)&1 == 1 {
subset = append(subset, set[object])
}
}
subsets = append(subsets, subset)
}
return subsets
}

9
go.mod
View File

@@ -1,5 +1,8 @@
module go.unistack.org/micro-wrapper-sql/v3
module go.unistack.org/micro-wrapper-sql/v4
go 1.16
go 1.20
require go.unistack.org/micro/v3 v3.10.14
require (
go.unistack.org/micro-wrapper-requestid/v4 v4.0.3
go.unistack.org/micro/v4 v4.0.8
)

12
go.sum
View File

@@ -1,8 +1,4 @@
github.com/imdario/mergo v0.3.13/go.mod h1:4lJ1jqUDcsbIECGy0RUJAXNIhg+6ocWgb1ALK2O4oXg=
github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
github.com/silas/dag v0.0.0-20211117232152-9d50aa809f35/go.mod h1:7RTUFBdIRC9nZ7/3RyRNH1bdqIShrDejd1YbLwgPS+I=
go.unistack.org/micro/v3 v3.10.14 h1:7fgLpwGlCN67twhwtngJDEQvrMkUBDSA5vzZqxIDqNE=
go.unistack.org/micro/v3 v3.10.14/go.mod h1:uMAc0U/x7dmtICCrblGf0ZLgYegu3VwQAquu+OFCw1Q=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/yaml.v3 v3.0.0/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
go.unistack.org/micro-wrapper-requestid/v4 v4.0.3 h1:LhloK8yLLsov4XiKw9i20wdBUfvWCG1TaBNgNVtNa8Q=
go.unistack.org/micro-wrapper-requestid/v4 v4.0.3/go.mod h1:vRIkCMyhbBVus1xQlr/Fef65n07znrBGkpSCpjkEiqk=
go.unistack.org/micro/v4 v4.0.8 h1:D5n18+fhyRjJkKSPdIcW1ratWUuzy9TzrS5ud/qImWg=
go.unistack.org/micro/v4 v4.0.8/go.mod h1:QT3gOIE4qGgBiQGm2Pad/62Sl5R53QfrgYHD448aX14=

View File

@@ -2,12 +2,14 @@ package wrapper
import (
"context"
"database/sql"
"errors"
"fmt"
"time"
"go.unistack.org/micro/v3/logger"
"go.unistack.org/micro/v3/meter"
"go.unistack.org/micro/v3/tracer"
"go.unistack.org/micro/v4/logger"
"go.unistack.org/micro/v4/meter"
"go.unistack.org/micro/v4/tracer"
)
var (
@@ -17,8 +19,8 @@ var (
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 := []interface{}{"db.method", method, "took", fmt.Sprintf("%v", td)}
if err != nil && !errors.Is(err, sql.ErrNoRows) {
labels = append(labels, "error", err.Error())
}
if query != labelUnknown {
@@ -29,23 +31,23 @@ var (
)
var (
MaxOpenConnections = "max_open_connections"
OpenConnections = "open_connections"
InuseConnections = "inuse_connections"
IdleConnections = "idle_connections"
WaitConnections = "wait_connections"
MaxOpenConnections = "max_open_conn"
OpenConnections = "open_conn"
InuseConnections = "inuse_conn"
IdleConnections = "idle_conn"
WaitConnections = "waited_conn"
BlockedSeconds = "blocked_seconds"
MaxIdleClosed = "max_idle_closed"
MaxIdletimeClosed = "max_idletime_closed"
MaxLifetimeClosed = "max_lifetime_closed"
MaxIdletimeClosed = "closed_max_idle"
MaxLifetimeClosed = "closed_max_lifetime"
meterRequestTotal = "request_total"
meterRequestLatencyMicroseconds = "request_latency_microseconds"
meterRequestLatencyMicroseconds = "latency_microseconds"
meterRequestDurationSeconds = "request_duration_seconds"
labelUnknown = "unknown"
labelQuery = "query"
labelMethod = "method"
labelQuery = "db_statement"
labelMethod = "db_method"
labelStatus = "status"
labelSuccess = "success"
labelFailure = "failure"
@@ -78,7 +80,7 @@ func NewOptions(opts ...Option) Options {
Tracer: tracer.DefaultTracer,
MeterStatsInterval: DefaultMeterStatsInterval,
MeterMetricPrefix: DefaultMeterMetricPrefix,
LoggerLevel: logger.DebugLevel,
LoggerLevel: logger.ErrorLevel,
LoggerObserver: DefaultLoggerObserver,
}
for _, o := range opts {
@@ -177,8 +179,8 @@ func QueryName(ctx context.Context, name string) context.Context {
}
func getQueryName(ctx context.Context) string {
if v, ok := ctx.Value(queryNameKey{}).(string); ok {
if v, ok := ctx.Value(queryNameKey{}).(string); ok && v != labelUnknown {
return v
}
return ""
return getCallerName()
}

View File

@@ -1,4 +1,4 @@
package wrapper // import "go.unistack.org/micro-wrapper-sql/v3"
package wrapper // import "go.unistack.org/micro-wrapper-sql/v4"
import (
"context"

156
stmt.go
View File

@@ -6,16 +6,23 @@ import (
"fmt"
"time"
"go.unistack.org/micro/v3/tracer"
requestid "go.unistack.org/micro-wrapper-requestid/v4"
"go.unistack.org/micro/v4/tracer"
)
var _ driver.Stmt = &wrapperStmt{}
var (
_ driver.Stmt = (*wrapperStmt)(nil)
_ driver.StmtQueryContext = (*wrapperStmt)(nil)
_ driver.StmtExecContext = (*wrapperStmt)(nil)
_ driver.NamedValueChecker = (*wrapperStmt)(nil)
)
// wrapperStmt defines a wrapper for driver.Stmt
type wrapperStmt struct {
stmt driver.Stmt
opts Options
ctx context.Context
stmt driver.Stmt
opts Options
query string
ctx context.Context
}
// Close implements driver.Stmt Close
@@ -39,8 +46,8 @@ 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(ctx, "Close", labelUnknown, td, err)...).Log(ctx, w.opts.LoggerLevel)
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Fields(w.opts.LoggerObserver(ctx, "Close", getCallerName(), td, err)).Log(ctx, w.opts.LoggerLevel)
}
return err
}
@@ -50,6 +57,15 @@ func (w *wrapperStmt) NumInput() int {
return w.stmt.NumInput()
}
// CheckNamedValue implements driver.NamedValueChecker
func (w *wrapperStmt) CheckNamedValue(v *driver.NamedValue) error {
s, ok := w.stmt.(driver.NamedValueChecker)
if !ok {
return driver.ErrSkip
}
return s.CheckNamedValue(v)
}
// Exec implements driver.Stmt Exec
func (w *wrapperStmt) Exec(args []driver.Value) (driver.Result, error) {
var ctx context.Context
@@ -60,8 +76,7 @@ func (w *wrapperStmt) Exec(args []driver.Value) (driver.Result, error) {
}
labels := []string{labelMethod, "Exec"}
ts := time.Now()
// nolint:staticcheck
res, err := w.stmt.Exec(args)
res, err := w.stmt.Exec(args) // nolint:staticcheck
td := time.Since(ts)
te := td.Seconds()
if err != nil {
@@ -72,9 +87,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(ctx, "Exec", labelUnknown, td, err)...).Log(ctx, w.opts.LoggerLevel)
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Fields(w.opts.LoggerObserver(ctx, "Exec", getCallerName(), td, err)).Log(ctx, w.opts.LoggerLevel)
}
return res, err
}
@@ -88,8 +104,7 @@ func (w *wrapperStmt) Query(args []driver.Value) (driver.Rows, error) {
}
labels := []string{labelMethod, "Query"}
ts := time.Now()
// nolint:staticcheck
rows, err := w.stmt.Query(args)
rows, err := w.stmt.Query(args) // nolint:staticcheck
td := time.Since(ts)
te := td.Seconds()
if err != nil {
@@ -100,73 +115,81 @@ 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(ctx, "Query", labelUnknown, td, err)...).Log(ctx, w.opts.LoggerLevel)
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Fields(w.opts.LoggerObserver(ctx, "Query", getCallerName(), td, err)).Log(ctx, w.opts.LoggerLevel)
}
return rows, err
}
// ExecContext implements driver.ExecerContext ExecContext
func (w *wrapperStmt) ExecContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Result, error) {
// ColumnConverter implements driver.ColumnConverter
func (w *wrapperStmt) ColumnConverter(idx int) driver.ValueConverter {
s, ok := w.stmt.(driver.ColumnConverter) // nolint:staticcheck
if !ok {
return nil
}
return s.ColumnConverter(idx)
}
// ExecContext implements driver.StmtExecContext ExecContext
func (w *wrapperStmt) ExecContext(ctx context.Context, args []driver.NamedValue) (driver.Result, error) {
var nctx context.Context
var span tracer.Span
if w.ctx != nil {
nctx, span = w.opts.Tracer.Start(w.ctx, "ExecContext")
nctx, span = w.opts.Tracer.Start(w.ctx, "sdk.database", tracer.WithSpanKind(tracer.SpanKindClient))
} else {
nctx, span = w.opts.Tracer.Start(ctx, "ExecContext")
nctx, span = w.opts.Tracer.Start(ctx, "sdk.database", tracer.WithSpanKind(tracer.SpanKindClient))
}
span.AddLabels("method", "ExecContext")
span.AddLabels("db.method", "ExecContext")
name := getQueryName(ctx)
if name != "" {
span.AddLabels("query", name)
} else {
name = labelUnknown
}
span.AddLabels("db.statement", name)
defer span.Finish()
if len(args) > 0 {
span.AddLabels("args", fmt.Sprintf("%v", namedValueToLabels(args)))
span.AddLabels("db.args", fmt.Sprintf("%v", namedValueToLabels(args)))
}
if id, ok := ctx.Value(requestid.XRequestIDKey).(string); ok {
span.AddLabels("x-request-id", id)
}
labels := []string{labelMethod, "ExecContext", labelQuery, name}
if conn, ok := w.stmt.(driver.ExecerContext); ok {
if conn, ok := w.stmt.(driver.StmtExecContext); ok {
ts := time.Now()
res, err := conn.ExecContext(nctx, query, args)
res, err := conn.ExecContext(nctx, 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())
span.SetStatus(tracer.SpanStatusError, 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(ctx, "ExecContext", name, td, err)...).Log(ctx, w.opts.LoggerLevel)
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Fields(w.opts.LoggerObserver(ctx, "ExecContext", name, td, err)).Log(ctx, w.opts.LoggerLevel)
}
return res, 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())
span.SetStatus(tracer.SpanStatusError, err.Error())
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(ctx, "ExecContext", name, 0, err)...).Log(ctx, w.opts.LoggerLevel)
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Fields(w.opts.LoggerObserver(ctx, "ExecContext", name, 0, err)).Log(ctx, w.opts.LoggerLevel)
}
return nil, err
}
ts := time.Now()
// nolint:staticcheck
res, err := w.Exec(values)
res, err := w.Exec(values) // nolint:staticcheck
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())
span.SetStatus(tracer.SpanStatusError, err.Error())
} else {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelSuccess)...).Inc()
}
@@ -174,42 +197,42 @@ 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 {
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Fields(w.opts.LoggerObserver(ctx, "ExecContext", name, td, err)).Log(ctx, w.opts.LoggerLevel)
}
return res, err
}
// QueryContext implements Driver.QueryerContext QueryContext
func (w *wrapperStmt) QueryContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Rows, error) {
// QueryContext implements driver.StmtQueryContext StmtQueryContext
func (w *wrapperStmt) QueryContext(ctx context.Context, args []driver.NamedValue) (driver.Rows, error) {
var nctx context.Context
var span tracer.Span
if w.ctx != nil {
nctx, span = w.opts.Tracer.Start(w.ctx, "QueryContext")
nctx, span = w.opts.Tracer.Start(w.ctx, "sdk.database", tracer.WithSpanKind(tracer.SpanKindClient))
} else {
nctx, span = w.opts.Tracer.Start(ctx, "QueryContext")
nctx, span = w.opts.Tracer.Start(ctx, "sdk.database", tracer.WithSpanKind(tracer.SpanKindClient))
}
span.AddLabels("method", "QueryContext")
span.AddLabels("db.method", "QueryContext")
name := getQueryName(ctx)
if name != "" {
span.AddLabels("query", name)
} else {
name = labelUnknown
}
span.AddLabels("db.statement", name)
defer span.Finish()
if len(args) > 0 {
span.AddLabels("args", fmt.Sprintf("%v", namedValueToLabels(args)))
span.AddLabels("db.args", fmt.Sprintf("%v", namedValueToLabels(args)))
}
if id, ok := ctx.Value(requestid.XRequestIDKey).(string); ok {
span.AddLabels("x-request-id", id)
}
labels := []string{labelMethod, "QueryContext", labelQuery, name}
if conn, ok := w.stmt.(driver.QueryerContext); ok {
if conn, ok := w.stmt.(driver.StmtQueryContext); ok {
ts := time.Now()
rows, err := conn.QueryContext(nctx, query, args)
rows, err := conn.QueryContext(nctx, 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())
span.SetStatus(tracer.SpanStatusError, err.Error())
} else {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelSuccess)...).Inc()
}
@@ -217,32 +240,30 @@ func (w *wrapperStmt) 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 {
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Fields(w.opts.LoggerObserver(ctx, "QueryContext", name, td, err)).Log(ctx, 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())
span.SetStatus(tracer.SpanStatusError, err.Error())
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(ctx, "QueryContext", name, 0, err)...).Log(ctx, w.opts.LoggerLevel)
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Fields(w.opts.LoggerObserver(ctx, "QueryContext", name, 0, err)).Log(ctx, w.opts.LoggerLevel)
}
return nil, err
}
ts := time.Now()
// nolint:staticcheck
rows, err := w.Query(values)
rows, err := w.Query(values) // nolint:staticcheck
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())
span.SetStatus(tracer.SpanStatusError, err.Error())
} else {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelSuccess)...).Inc()
}
@@ -250,8 +271,9 @@ func (w *wrapperStmt) 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(ctx, "QueryContext", name, td, err)...).Log(ctx, w.opts.LoggerLevel)
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Fields(w.opts.LoggerObserver(ctx, "QueryContext", name, td, err)).Log(ctx, w.opts.LoggerLevel)
}
return rows, err
}

37
tx.go
View File

@@ -5,64 +5,59 @@ import (
"database/sql/driver"
"time"
"go.unistack.org/micro/v3/tracer"
"go.unistack.org/micro/v4/tracer"
)
var _ driver.Tx = &wrapperTx{}
var _ driver.Tx = (*wrapperTx)(nil)
// wrapperTx defines a wrapper for driver.Tx
type wrapperTx struct {
tx driver.Tx
span tracer.Span
opts Options
conn *wrapperConn
ctx context.Context
}
// Commit implements driver.Tx Commit
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 err != nil {
w.span.AddLabels("error", true)
w.span.AddLabels("err", err.Error())
if w.span != nil {
if err != nil {
w.span.SetStatus(tracer.SpanStatusError, err.Error())
}
w.span.Finish()
}
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(w.ctx, "Commit", labelUnknown, td, err)...).Log(w.ctx, w.opts.LoggerLevel)
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Fields(w.opts.LoggerObserver(w.ctx, "Commit", getCallerName(), td, err)).Log(w.ctx, w.opts.LoggerLevel)
}
w.ctx = nil
w.conn.ctx = nil
return err
}
// Rollback implements driver.Tx Rollback
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 err != nil {
w.span.AddLabels("error", true)
w.span.AddLabels("err", err.Error())
if w.span != nil {
if err != nil {
w.span.SetStatus(tracer.SpanStatusError, err.Error())
}
w.span.Finish()
}
if w.opts.LoggerEnabled {
w.opts.Logger.Fields(w.opts.LoggerObserver(w.ctx, "Rollback", labelUnknown, td, err)...).Log(w.ctx, w.opts.LoggerLevel)
if w.opts.LoggerEnabled && w.opts.Logger.V(w.opts.LoggerLevel) {
w.opts.Logger.Fields(w.opts.LoggerObserver(w.ctx, "Rollback", getCallerName(), td, err)).Log(w.ctx, w.opts.LoggerLevel)
}
w.ctx = nil
w.conn.ctx = nil
return err
}

17
wrap.go Normal file
View File

@@ -0,0 +1,17 @@
package wrapper
import (
"database/sql/driver"
)
func wrapDriver(d driver.Driver, opts Options) driver.Driver {
if _, ok := d.(driver.DriverContext); ok {
return &wrapperDriver{driver: d, opts: opts}
}
return struct{ driver.Driver }{&wrapperDriver{driver: d, opts: opts}}
}
// WrapConn allows an existing driver.Conn to be wrapped.
func WrapConn(c driver.Conn, opts ...Option) driver.Conn {
return wrapConn(c, NewOptions(opts...))
}

20699
wrap_gen.go Normal file

File diff suppressed because it is too large Load Diff