forked from mirror/redis
otel: add RemoteAddr
This commit is contained in:
parent
9e4063ab5a
commit
566840fe27
|
@ -9,6 +9,7 @@ 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/api/trace"
|
||||||
)
|
)
|
||||||
|
|
||||||
var noDeadline = time.Time{}
|
var noDeadline = time.Time{}
|
||||||
|
@ -58,11 +59,14 @@ func (cn *Conn) Write(b []byte) (int, error) {
|
||||||
}
|
}
|
||||||
|
|
||||||
func (cn *Conn) RemoteAddr() net.Addr {
|
func (cn *Conn) RemoteAddr() net.Addr {
|
||||||
return cn.netConn.RemoteAddr()
|
if cn.netConn != nil {
|
||||||
|
return cn.netConn.RemoteAddr()
|
||||||
|
}
|
||||||
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
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, "with_reader", func(ctx context.Context) error {
|
return internal.WithSpan(ctx, "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 internal.RecordError(ctx, err)
|
return internal.RecordError(ctx, err)
|
||||||
}
|
}
|
||||||
|
@ -76,7 +80,7 @@ func (cn *Conn) WithReader(ctx context.Context, timeout time.Duration, fn func(r
|
||||||
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, "with_writer", func(ctx context.Context) error {
|
return internal.WithSpan(ctx, "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 internal.RecordError(ctx, err)
|
return internal.RecordError(ctx, err)
|
||||||
}
|
}
|
||||||
|
|
|
@ -11,7 +11,7 @@ import (
|
||||||
)
|
)
|
||||||
|
|
||||||
func Sleep(ctx context.Context, dur time.Duration) error {
|
func Sleep(ctx context.Context, dur time.Duration) error {
|
||||||
return WithSpan(ctx, "sleep", func(ctx context.Context) error {
|
return WithSpan(ctx, "sleep", func(ctx context.Context, span trace.Span) error {
|
||||||
t := time.NewTimer(dur)
|
t := time.NewTimer(dur)
|
||||||
defer t.Stop()
|
defer t.Stop()
|
||||||
|
|
||||||
|
@ -62,15 +62,15 @@ func Unwrap(err error) error {
|
||||||
|
|
||||||
//------------------------------------------------------------------------------
|
//------------------------------------------------------------------------------
|
||||||
|
|
||||||
func WithSpan(ctx context.Context, name string, fn func(context.Context) error) error {
|
func WithSpan(ctx context.Context, name string, fn func(context.Context, trace.Span) error) error {
|
||||||
if !trace.SpanFromContext(ctx).IsRecording() {
|
if span := trace.SpanFromContext(ctx); !span.IsRecording() {
|
||||||
return fn(ctx)
|
return fn(ctx, span)
|
||||||
}
|
}
|
||||||
|
|
||||||
ctx, span := global.Tracer("github.com/go-redis/redis").Start(ctx, name)
|
ctx, span := global.Tracer("github.com/go-redis/redis").Start(ctx, name)
|
||||||
defer span.End()
|
defer span.End()
|
||||||
|
|
||||||
return fn(ctx)
|
return fn(ctx, span)
|
||||||
}
|
}
|
||||||
|
|
||||||
func RecordError(ctx context.Context, err error) error {
|
func RecordError(ctx context.Context, err error) error {
|
||||||
|
|
|
@ -293,9 +293,9 @@ 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
|
var conn net.Conn
|
||||||
err := internal.WithSpan(ctx, "dialer", func(ctx context.Context) error {
|
err := internal.WithSpan(ctx, "dialer", func(ctx context.Context, span trace.Span) error {
|
||||||
var err error
|
var err error
|
||||||
trace.SpanFromContext(ctx).SetAttributes(
|
span.SetAttributes(
|
||||||
label.String("redis.network", opt.Network),
|
label.String("redis.network", opt.Network),
|
||||||
label.String("redis.addr", opt.Addr),
|
label.String("redis.addr", opt.Addr),
|
||||||
)
|
)
|
||||||
|
|
15
redis.go
15
redis.go
|
@ -8,6 +8,8 @@ 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/api/trace"
|
||||||
|
"go.opentelemetry.io/otel/label"
|
||||||
)
|
)
|
||||||
|
|
||||||
// Nil reply returned by Redis when key does not exist.
|
// Nil reply returned by Redis when key does not exist.
|
||||||
|
@ -223,7 +225,7 @@ func (c *baseClient) _getConn(ctx context.Context) (*pool.Conn, error) {
|
||||||
return cn, nil
|
return cn, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
err = internal.WithSpan(ctx, "init_conn", func(ctx context.Context) error {
|
err = internal.WithSpan(ctx, "init_conn", func(ctx context.Context, span trace.Span) error {
|
||||||
return c.initConn(ctx, cn)
|
return c.initConn(ctx, cn)
|
||||||
})
|
})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
@ -297,11 +299,18 @@ 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, "with_conn", func(ctx context.Context) error {
|
return internal.WithSpan(ctx, "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() {
|
||||||
|
if remoteAddr := cn.RemoteAddr(); remoteAddr != nil {
|
||||||
|
span.SetAttributes(label.String("net.peer.ip", remoteAddr.String()))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
defer func() {
|
defer func() {
|
||||||
c.releaseConn(ctx, cn, err)
|
c.releaseConn(ctx, cn, err)
|
||||||
}()
|
}()
|
||||||
|
@ -317,7 +326,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, "process", func(ctx context.Context) error {
|
err := internal.WithSpan(ctx, "process", func(ctx context.Context, span trace.Span) 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
|
||||||
|
|
Loading…
Reference in New Issue