Compare commits

...

1 Commits

Author SHA1 Message Date
Vladimir Mihailenco fd8b14e651 Remove otel instrumentation 2021-03-19 13:47:44 +02:00
4 changed files with 57 additions and 117 deletions

View File

@ -9,7 +9,6 @@ import (
"github.com/go-redis/redis/v8/internal" "github.com/go-redis/redis/v8/internal"
"github.com/go-redis/redis/v8/internal/proto" "github.com/go-redis/redis/v8/internal/proto"
"go.opentelemetry.io/otel/trace"
) )
var noDeadline = time.Time{} var noDeadline = time.Time{}
@ -66,41 +65,37 @@ func (cn *Conn) RemoteAddr() net.Addr {
} }
func (cn *Conn) WithReader(ctx context.Context, timeout time.Duration, fn func(rd *proto.Reader) error) error { func (cn *Conn) WithReader(ctx context.Context, timeout time.Duration, fn func(rd *proto.Reader) error) error {
return internal.WithSpan(ctx, "redis.with_reader", func(ctx context.Context, span trace.Span) error { if err := cn.netConn.SetReadDeadline(cn.deadline(ctx, timeout)); err != nil {
if err := cn.netConn.SetReadDeadline(cn.deadline(ctx, timeout)); err != nil { return err
return internal.RecordError(ctx, span, err) }
} if err := fn(cn.rd); err != nil {
if err := fn(cn.rd); err != nil { return err
return internal.RecordError(ctx, span, err) }
} return nil
return nil
})
} }
func (cn *Conn) WithWriter( func (cn *Conn) WithWriter(
ctx context.Context, timeout time.Duration, fn func(wr *proto.Writer) error, ctx context.Context, timeout time.Duration, fn func(wr *proto.Writer) error,
) error { ) error {
return internal.WithSpan(ctx, "redis.with_writer", func(ctx context.Context, span trace.Span) error { if err := cn.netConn.SetWriteDeadline(cn.deadline(ctx, timeout)); err != nil {
if err := cn.netConn.SetWriteDeadline(cn.deadline(ctx, timeout)); err != nil { return err
return internal.RecordError(ctx, span, err) }
}
if cn.bw.Buffered() > 0 { if cn.bw.Buffered() > 0 {
cn.bw.Reset(cn.netConn) cn.bw.Reset(cn.netConn)
} }
if err := fn(cn.wr); err != nil { if err := fn(cn.wr); err != nil {
return internal.RecordError(ctx, span, err) return err
} }
if err := cn.bw.Flush(); err != nil { if err := cn.bw.Flush(); err != nil {
return internal.RecordError(ctx, span, err) return err
} }
internal.WritesCounter.Add(ctx, 1) internal.WritesCounter.Add(ctx, 1)
return nil return nil
})
} }
func (cn *Conn) Close() error { func (cn *Conn) Close() error {

View File

@ -4,24 +4,19 @@ import (
"context" "context"
"time" "time"
"github.com/go-redis/redis/v8/internal/proto"
"github.com/go-redis/redis/v8/internal/util" "github.com/go-redis/redis/v8/internal/util"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/trace"
) )
func Sleep(ctx context.Context, dur time.Duration) error { func Sleep(ctx context.Context, dur time.Duration) error {
return WithSpan(ctx, "time.Sleep", func(ctx context.Context, span trace.Span) error { t := time.NewTimer(dur)
t := time.NewTimer(dur) defer t.Stop()
defer t.Stop()
select { select {
case <-t.C: case <-t.C:
return nil return nil
case <-ctx.Done(): case <-ctx.Done():
return ctx.Err() return ctx.Err()
} }
})
} }
func ToLower(s string) string { func ToLower(s string) string {
@ -49,25 +44,3 @@ func isLower(s string) bool {
} }
return true return true
} }
//------------------------------------------------------------------------------
var tracer = otel.Tracer("github.com/go-redis/redis")
func WithSpan(ctx context.Context, name string, fn func(context.Context, trace.Span) error) error {
if span := trace.SpanFromContext(ctx); !span.IsRecording() {
return fn(ctx, span)
}
ctx, span := tracer.Start(ctx, name)
defer span.End()
return fn(ctx, span)
}
func RecordError(ctx context.Context, span trace.Span, err error) error {
if err != proto.Nil {
span.RecordError(err)
}
return err
}

View File

@ -12,10 +12,7 @@ import (
"strings" "strings"
"time" "time"
"github.com/go-redis/redis/v8/internal"
"github.com/go-redis/redis/v8/internal/pool" "github.com/go-redis/redis/v8/internal/pool"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/trace"
) )
// Limiter is the interface of a rate limiter or a circuit breaker. // Limiter is the interface of a rate limiter or a circuit breaker.
@ -292,20 +289,7 @@ func getUserPassword(u *url.URL) (string, string) {
func newConnPool(opt *Options) *pool.ConnPool { func newConnPool(opt *Options) *pool.ConnPool {
return pool.NewConnPool(&pool.Options{ return pool.NewConnPool(&pool.Options{
Dialer: func(ctx context.Context) (net.Conn, error) { Dialer: func(ctx context.Context) (net.Conn, error) {
var conn net.Conn return opt.Dialer(ctx, opt.Network, opt.Addr)
err := internal.WithSpan(ctx, "redis.dial", func(ctx context.Context, span trace.Span) error {
span.SetAttributes(
attribute.String("db.connection_string", opt.Addr),
)
var err error
conn, err = opt.Dialer(ctx, opt.Network, opt.Addr)
if err != nil {
_ = internal.RecordError(ctx, span, err)
}
return err
})
return conn, err
}, },
PoolSize: opt.PoolSize, PoolSize: opt.PoolSize,
MinIdleConns: opt.MinIdleConns, MinIdleConns: opt.MinIdleConns,

View File

@ -10,8 +10,6 @@ import (
"github.com/go-redis/redis/v8/internal" "github.com/go-redis/redis/v8/internal"
"github.com/go-redis/redis/v8/internal/pool" "github.com/go-redis/redis/v8/internal/pool"
"github.com/go-redis/redis/v8/internal/proto" "github.com/go-redis/redis/v8/internal/proto"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/trace"
) )
// Nil reply returned by Redis when key does not exist. // Nil reply returned by Redis when key does not exist.
@ -214,9 +212,7 @@ func (c *baseClient) _getConn(ctx context.Context) (*pool.Conn, error) {
return cn, nil return cn, nil
} }
err = internal.WithSpan(ctx, "redis.init_conn", func(ctx context.Context, span trace.Span) error { err = c.initConn(ctx, cn)
return c.initConn(ctx, cn)
})
if err != nil { if err != nil {
c.connPool.Remove(ctx, cn, err) c.connPool.Remove(ctx, cn, err)
if err := errors.Unwrap(err); err != nil { if err := errors.Unwrap(err); err != nil {
@ -288,43 +284,35 @@ func (c *baseClient) releaseConn(ctx context.Context, cn *pool.Conn, err error)
func (c *baseClient) withConn( func (c *baseClient) withConn(
ctx context.Context, fn func(context.Context, *pool.Conn) error, ctx context.Context, fn func(context.Context, *pool.Conn) error,
) error { ) error {
return internal.WithSpan(ctx, "redis.with_conn", func(ctx context.Context, span trace.Span) error { cn, err := c.getConn(ctx)
cn, err := c.getConn(ctx) if err != nil {
if err != nil { return err
return err }
}
if span.IsRecording() { defer func() {
if remoteAddr := cn.RemoteAddr(); remoteAddr != nil { c.releaseConn(ctx, cn, err)
span.SetAttributes(attribute.String("net.peer.ip", remoteAddr.String())) }()
}
}
defer func() { done := ctx.Done()
c.releaseConn(ctx, cn, err) if done == nil {
}() err = fn(ctx, cn)
return err
}
done := ctx.Done() errc := make(chan error, 1)
if done == nil { go func() { errc <- fn(ctx, cn) }()
err = fn(ctx, cn)
return err
}
errc := make(chan error, 1) select {
go func() { errc <- fn(ctx, cn) }() case <-done:
_ = cn.Close()
// Wait for the goroutine to finish and send something.
<-errc
select { err = ctx.Err()
case <-done: return err
_ = cn.Close() case err = <-errc:
// Wait for the goroutine to finish and send something. return err
<-errc }
err = ctx.Err()
return err
case err = <-errc:
return err
}
})
} }
func (c *baseClient) process(ctx context.Context, cmd Cmder) error { func (c *baseClient) process(ctx context.Context, cmd Cmder) error {
@ -333,7 +321,7 @@ func (c *baseClient) process(ctx context.Context, cmd Cmder) error {
attempt := attempt attempt := attempt
var retry bool var retry bool
err := internal.WithSpan(ctx, "redis.process", func(ctx context.Context, span trace.Span) error { err := func() error {
if attempt > 0 { if attempt > 0 {
if err := internal.Sleep(ctx, c.retryBackoff(attempt)); err != nil { if err := internal.Sleep(ctx, c.retryBackoff(attempt)); err != nil {
return err return err
@ -364,7 +352,7 @@ func (c *baseClient) process(ctx context.Context, cmd Cmder) error {
} }
retry = shouldRetry(err, atomic.LoadUint32(&retryTimeout) == 1) retry = shouldRetry(err, atomic.LoadUint32(&retryTimeout) == 1)
return err return err
}) }()
if err == nil || !retry { if err == nil || !retry {
return err return err
} }