unistack-org/micro-wrapper-sql
4
0
Fork 1
Code Issues 3 Pull Requests Actions Packages Projects Releases 23 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 +235 -32
5 changed files with 235 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit b2eb8e6c5c - Show all commits

1
common.go
View File

@ -22,6 +22,7 @@ func namedValueToValue(named []driver.NamedValue) ([]driver.Value, error) {
return dargs, nil
}
// namedValueToLabels convert driver arguments to interface{} slice
func namedValueToLabels(named []driver.NamedValue) []interface{} {
largs := make([]interface{}, len(named)*2)
var name string

167
conn.go
View File

@ -4,6 +4,7 @@ import (
"context"
"database/sql/driver"
"fmt"
"time"
)
// wrapperConn defines a wrapper for driver.Conn
@ -14,79 +15,160 @@ type wrapperConn struct {
// Prepare implements driver.Conn Prepare
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()
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)
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)
return &wrapperStmt{stmt: stmt, opts: w.opts}, nil
}
// Close implements driver.Conn Close
func (w *wrapperConn) Close() error {
return w.conn.Close()
labels := []string{labelMethod, "Close"}
ts := time.Now()
err := w.conn.Close()
te := time.Since(ts).Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
} 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)
return err
}
// Begin implements driver.Conn Begin
func (w *wrapperConn) Begin() (driver.Tx, error) {
labels := []string{labelMethod, "Begin"}
ts := time.Now()
// nolint:staticcheck
tx, err := w.conn.Begin()
te := time.Since(ts).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)
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)
return &wrapperTx{tx: tx, opts: w.opts}, 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, "BeginTx")
span.AddLabels("op", "BeginTx")
if name := getQueryName(ctx); name != "" {
span.AddLabels("method", "BeginTx")
name := getQueryName(ctx)
if name != "" {
span.AddLabels("query", name)
} else {
name = labelUnknown
}
labels := []string{labelMethod, "BeginTx", labelQuery, name}
if connBeginTx, ok := w.conn.(driver.ConnBeginTx); ok {
ts := time.Now()
tx, err := connBeginTx.BeginTx(nctx, opts)
te := time.Since(ts).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())
return nil, err
}
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelSuccess)...).Inc()
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()
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())
}
return tx, 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)
return tx, nil
}
// PrepareContext implements driver.ConnPrepareContext PrepareContext
func (w *wrapperConn) PrepareContext(ctx context.Context, query string) (driver.Stmt, error) {
nctx, span := w.opts.Tracer.Start(ctx, "PrepareContext")
span.AddLabels("op", "PrepareContext")
if name := getQueryName(ctx); name != "" {
span.AddLabels("method", "PrepareContext")
name := getQueryName(ctx)
if name != "" {
span.AddLabels("query", name)
} else {
name = labelUnknown
}
labels := []string{labelMethod, "PrepareContext", labelQuery, name}
if connPrepareContext, ok := w.conn.(driver.ConnPrepareContext); ok {
ts := time.Now()
stmt, err := connPrepareContext.PrepareContext(nctx, query)
te := time.Since(ts).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())
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)
return &wrapperStmt{stmt: stmt, opts: w.opts}, nil
}
ts := time.Now()
stmt, err := w.conn.Prepare(query)
te := time.Since(ts).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())
}
return stmt, 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)
return stmt, nil
}
// Exec implements driver.Execer Exec
func (w *wrapperConn) Exec(query string, args []driver.Value) (driver.Result, error) {
// nolint:staticcheck
labels := []string{labelMethod, "Exec", labelQuery, labelUnknown}
if execer, ok := w.conn.(driver.Execer); ok {
return execer.Exec(query, args)
ts := time.Now()
res, err := execer.Exec(query, args)
te := time.Since(ts).Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
} 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)
return res, err
}
return nil, ErrUnsupported
}
@ -94,20 +176,32 @@ func (w *wrapperConn) Exec(query string, args []driver.Value) (driver.Result, er
// Exec implements driver.StmtExecContext ExecContext
func (w *wrapperConn) ExecContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Result, error) {
nctx, span := w.opts.Tracer.Start(ctx, "ExecContext")
span.AddLabels("op", "ExecContext")
if name := getQueryName(ctx); name != "" {
span.AddLabels("method", "ExecContext")
name := getQueryName(ctx)
if name != "" {
span.AddLabels("query", name)
} else {
name = labelUnknown
}
defer span.Finish()
if len(args) > 0 {
span.AddLabels("args", fmt.Sprintf("%v", namedValueToLabels(args)))
}
labels := []string{labelMethod, "ExecContext", labelQuery, name}
if execerContext, ok := w.conn.(driver.ExecerContext); ok {
ts := time.Now()
res, err := execerContext.ExecContext(nctx, query, args)
te := time.Since(ts).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)
return res, err
}
values, err := namedValueToValue(args)
@ -116,11 +210,19 @@ func (w *wrapperConn) ExecContext(ctx context.Context, query string, args []driv
span.AddLabels("err", err.Error())
return nil, err
}
ts := time.Now()
// nolint:staticcheck
res, err := w.Exec(query, values)
te := time.Since(ts).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)
return res, err
}
@ -129,20 +231,40 @@ func (w *wrapperConn) Ping(ctx context.Context) error {
if pinger, 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()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
span.AddLabels("error", true)
span.AddLabels("err", err.Error())
return err
} 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)
}
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 {
return queryer.Query(query, args)
labels := []string{labelMethod, "Query", labelQuery, labelUnknown}
ts := time.Now()
rows, err := queryer.Query(query, args)
te := time.Since(ts).Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
} 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)
return rows, err
}
return nil, ErrUnsupported
}
@ -150,20 +272,31 @@ func (w *wrapperConn) Query(query string, args []driver.Value) (driver.Rows, err
// QueryContext implements Driver.QueryerContext QueryContext
func (w *wrapperConn) QueryContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Rows, error) {
nctx, span := w.opts.Tracer.Start(ctx, "QueryContext")
span.AddLabels("op", "QueryContext")
if name := getQueryName(ctx); name != "" {
span.AddLabels("method", "QueryContext")
name := getQueryName(ctx)
if name != "" {
span.AddLabels("query", name)
} else {
name = labelUnknown
}
defer span.Finish()
if len(args) > 0 {
span.AddLabels("args", fmt.Sprintf("%v", namedValueToLabels(args)))
}
labels := []string{labelMethod, "QueryContext", labelQuery, name}
if queryerContext, ok := w.conn.(driver.QueryerContext); ok {
ts := time.Now()
rows, err := queryerContext.QueryContext(nctx, query, args)
te := time.Since(ts).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)
return rows, err
}
values, err := namedValueToValue(args)
@ -172,10 +305,18 @@ func (w *wrapperConn) QueryContext(ctx context.Context, query string, args []dri
span.AddLabels("err", err.Error())
return nil, err
}
ts := time.Now()
// nolint:staticcheck
rows, err := w.Query(query, values)
te := time.Since(ts).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)
return rows, err
}

20
options.go
View File

@ -27,10 +27,14 @@ var (
MaxIdletimeClosed = "max_idletime_closed"
MaxLifetimeClosed = "max_lifetime_closed"
// RequestTotal = "request_total"
// RequestLatencyMicroseconds = "request_latency_microseconds"
// RequestDurationSeconds = "request_duration_seconds"
meterRequestTotal = "request_total"
meterRequestLatencyMicroseconds = "request_latency_microseconds"
meterRequestDurationSeconds = "request_duration_seconds"
labelUnknown = "unknown"
labelQuery = "query"
labelMethod = "method"
labelStatus = "status"
labelSuccess = "success"
labelFailure = "failure"
labelHost = "db_host"
@ -44,9 +48,6 @@ type Options struct {
Tracer tracer.Tracer
DatabaseHost string
DatabaseName string
ServiceName string
ServiceVersion string
ServiceID string
MeterMetricPrefix string
MeterStatsInterval time.Duration
LoggerLevel logger.Level
@ -120,6 +121,13 @@ func Logger(l logger.Logger) Option {
}
}
// LoggerLevel passes logger.Level option
func LoggerLevel(lvl logger.Level) Option {
return func(o *Options) {
o.LoggerLevel = lvl
}
}
// Tracer passes tracer.Tracer to wrapper
func Tracer(t tracer.Tracer) Option {
return func(o *Options) {

73
stmt.go
View File

@ -4,6 +4,7 @@ import (
"context"
"database/sql/driver"
"fmt"
"time"
)
// wrapperStmt defines a wrapper for driver.Stmt
@ -14,7 +15,18 @@ type wrapperStmt struct {
// Close implements driver.Stmt Close
func (w *wrapperStmt) Close() error {
return w.stmt.Close()
labels := []string{labelMethod, "Close"}
ts := time.Now()
err := w.stmt.Close()
te := time.Since(ts).Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
} 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)
return err
}
// NumInput implements driver.Stmt NumInput
@ -24,39 +36,76 @@ func (w *wrapperStmt) NumInput() int {
// Exec implements driver.Stmt Exec
func (w *wrapperStmt) Exec(args []driver.Value) (driver.Result, error) {
return w.stmt.Exec(args)
// nolint:staticcheck
labels := []string{labelMethod, "Exec"}
ts := time.Now()
res, err := w.stmt.Exec(args)
te := time.Since(ts).Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
} 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)
return res, err
}
// Query implements driver.Stmt Query
func (w *wrapperStmt) Query(args []driver.Value) (driver.Rows, error) {
return w.stmt.Query(args)
labels := []string{labelMethod, "Query"}
ts := time.Now()
// nolint:staticcheck
rows, err := w.stmt.Query(args)
te := time.Since(ts).Seconds()
if err != nil {
w.opts.Meter.Counter(meterRequestTotal, append(labels, labelStatus, labelFailure)...).Inc()
} 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)
return rows, err
}
// ExecContext implements driver.ExecerContext ExecContext
func (w *wrapperStmt) ExecContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Result, error) {
nctx, span := w.opts.Tracer.Start(ctx, "ExecContext")
span.AddLabels("op", "ExecContext")
if name := getQueryName(ctx); name != "" {
span.AddLabels("method", "ExecContext")
name := getQueryName(ctx)
if name != "" {
span.AddLabels("query", name)
} else {
name = labelUnknown
}
defer span.Finish()
if len(args) > 0 {
span.AddLabels("args", fmt.Sprintf("%v", namedValueToLabels(args)))
}
labels := []string{labelMethod, "ExecContext", labelQuery, name}
if execerContext, ok := w.stmt.(driver.ExecerContext); ok {
ts := time.Now()
res, err := execerContext.ExecContext(nctx, query, args)
te := time.Since(ts).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)
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())
return nil, err
}
// nolint:staticcheck
res, err := w.Exec(values)
if err != nil {
span.AddLabels("error", true)
@ -68,9 +117,12 @@ func (w *wrapperStmt) ExecContext(ctx context.Context, query string, args []driv
// QueryContext implements Driver.QueryerContext QueryContext
func (w *wrapperStmt) QueryContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Rows, error) {
nctx, span := w.opts.Tracer.Start(ctx, "QueryContext")
span.AddLabels("op", "QueryContext")
if name := getQueryName(ctx); name != "" {
span.AddLabels("method", "QueryContext")
name := getQueryName(ctx)
if name != "" {
span.AddLabels("query", name)
} else {
name = labelUnknown
}
defer span.Finish()
if len(args) > 0 {
@ -86,12 +138,11 @@ func (w *wrapperStmt) QueryContext(ctx context.Context, query string, args []dri
}
values, err := namedValueToValue(args)
if err != nil {
if err != nil {
span.AddLabels("error", true)
span.AddLabels("err", err.Error())
}
span.AddLabels("error", true)
span.AddLabels("err", err.Error())
return nil, err
}
// nolint:staticcheck
rows, err := w.Query(values)
if err != nil {
span.AddLabels("error", true)

6
tx.go
View File

@ -18,7 +18,8 @@ func (w *wrapperTx) Commit() error {
if w.span != nil {
defer w.span.Finish()
}
return w.tx.Commit()
err := w.tx.Commit()
return err
}
// Rollback implements driver.Tx Rollback
@ -26,5 +27,6 @@ func (w *wrapperTx) Rollback() error {
if w.span != nil {
defer w.span.Finish()
}
return w.tx.Rollback()
err := w.tx.Rollback()
return err
}