forked from mirror/redis
Remove otel instrumentation
This commit is contained in:
parent
a3cd07dab0
commit
fd8b14e651
|
@ -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 {
|
||||||
|
|
|
@ -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
|
|
||||||
}
|
|
||||||
|
|
18
options.go
18
options.go
|
@ -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,
|
||||||
|
|
66
redis.go
66
redis.go
|
@ -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
|
||||||
}
|
}
|
||||||
|
|
Loading…
Reference in New Issue