Skip to content

Commit 1b77706

Browse files
vmihailencovmihailenco
committed
Optimize Otel instrumentation
1 parent 7c5bbc3 commit 1b77706

File tree

4 files changed

+123
-122
lines changed

4 files changed

+123
-122
lines changed

‎internal/pool/conn.go‎

Lines changed: 27 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,6 @@ import (
99

1010
"github.com/go-redis/redis/v8/internal"
1111
"github.com/go-redis/redis/v8/internal/proto"
12-
"go.opentelemetry.io/otel/trace"
1312
)
1413

1514
var noDeadline = time.Time{}
@@ -66,41 +65,43 @@ func (cn *Conn) RemoteAddr() net.Addr {
6665
}
6766

6867
func (cn *Conn) WithReader(ctx context.Context, timeout time.Duration, fn func(rd *proto.Reader) error) error {
69-
return internal.WithSpan(ctx, "redis.with_reader", func(ctx context.Context, span trace.Span) error {
70-
if err := cn.netConn.SetReadDeadline(cn.deadline(ctx, timeout)); err != nil {
71-
return internal.RecordError(ctx, span, err)
72-
}
73-
if err := fn(cn.rd); err != nil {
74-
return internal.RecordError(ctx, span, err)
75-
}
76-
return nil
77-
})
68+
ctx, span := internal.StartSpan(ctx, "redis.with_reader")
69+
defer span.End()
70+
71+
if err := cn.netConn.SetReadDeadline(cn.deadline(ctx, timeout)); err != nil {
72+
return internal.RecordError(ctx, span, err)
73+
}
74+
if err := fn(cn.rd); err != nil {
75+
return internal.RecordError(ctx, span, err)
76+
}
77+
return nil
7878
}
7979

8080
func (cn *Conn) WithWriter(
8181
ctx context.Context, timeout time.Duration, fn func(wr *proto.Writer) error,
8282
) error {
83-
return internal.WithSpan(ctx, "redis.with_writer", func(ctx context.Context, span trace.Span) error {
84-
if err := cn.netConn.SetWriteDeadline(cn.deadline(ctx, timeout)); err != nil {
85-
return internal.RecordError(ctx, span, err)
86-
}
83+
ctx, span := internal.StartSpan(ctx, "redis.with_writer")
84+
defer span.End()
8785

88-
if cn.bw.Buffered() > 0 {
89-
cn.bw.Reset(cn.netConn)
90-
}
86+
if err := cn.netConn.SetWriteDeadline(cn.deadline(ctx, timeout)); err != nil {
87+
return internal.RecordError(ctx, span, err)
88+
}
9189

92-
if err := fn(cn.wr); err != nil {
93-
return internal.RecordError(ctx, span, err)
94-
}
90+
if cn.bw.Buffered() > 0 {
91+
cn.bw.Reset(cn.netConn)
92+
}
9593

96-
if err := cn.bw.Flush(); err != nil {
97-
return internal.RecordError(ctx, span, err)
98-
}
94+
if err := fn(cn.wr); err != nil {
95+
return internal.RecordError(ctx, span, err)
96+
}
97+
98+
if err := cn.bw.Flush(); err != nil {
99+
return internal.RecordError(ctx, span, err)
100+
}
99101

100-
internal.WritesCounter.Add(ctx, 1)
102+
internal.WritesCounter.Add(ctx, 1)
101103

102-
return nil
103-
})
104+
return nil
104105
}
105106

106107
func (cn *Conn) Close() error {

‎internal/util.go‎

Lines changed: 14 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -11,17 +11,18 @@ import (
1111
)
1212

1313
func Sleep(ctx context.Context, dur time.Duration) error {
14-
return WithSpan(ctx, "time.Sleep", func(ctx context.Context, span trace.Span) error {
15-
t := time.NewTimer(dur)
16-
defer t.Stop()
14+
_, span := StartSpan(ctx, "time.Sleep")
15+
defer span.End()
1716

18-
select {
19-
case <-t.C:
20-
return nil
21-
case <-ctx.Done():
22-
return ctx.Err()
23-
}
24-
})
17+
t := time.NewTimer(dur)
18+
defer t.Stop()
19+
20+
select {
21+
case <-t.C:
22+
return nil
23+
case <-ctx.Done():
24+
return ctx.Err()
25+
}
2526
}
2627

2728
func ToLower(s string) string {
@@ -54,15 +55,11 @@ func isLower(s string) bool {
5455

5556
var tracer = otel.Tracer("github.com/go-redis/redis")
5657

57-
func WithSpan(ctx context.Context, name string, fn func(context.Context, trace.Span) error) error {
58+
func StartSpan(ctx context.Context, name string) (context.Context, trace.Span) {
5859
if span := trace.SpanFromContext(ctx); !span.IsRecording() {
59-
return fn(ctx, span)
60+
return ctx, span
6061
}
61-
62-
ctx, span := tracer.Start(ctx, name)
63-
defer span.End()
64-
65-
return fn(ctx, span)
62+
return tracer.Start(ctx, name)
6663
}
6764

6865
func RecordError(ctx context.Context, span trace.Span, err error) error {

‎options.go‎

Lines changed: 11 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,6 @@ import (
1515
"github.com/go-redis/redis/v8/internal"
1616
"github.com/go-redis/redis/v8/internal/pool"
1717
"go.opentelemetry.io/otel/attribute"
18-
"go.opentelemetry.io/otel/trace"
1918
)
2019

2120
// Limiter is the interface of a rate limiter or a circuit breaker.
@@ -292,20 +291,21 @@ func getUserPassword(u *url.URL) (string, string) {
292291
func newConnPool(opt *Options) *pool.ConnPool {
293292
return pool.NewConnPool(&pool.Options{
294293
Dialer: func(ctx context.Context) (net.Conn, error) {
295-
var conn net.Conn
296-
err := internal.WithSpan(ctx, "redis.dial", func(ctx context.Context, span trace.Span) error {
294+
ctx, span := internal.StartSpan(ctx, "redis.dial")
295+
defer span.End()
296+
297+
if span.IsRecording() {
297298
span.SetAttributes(
298299
attribute.String("db.connection_string", opt.Addr),
299300
)
301+
}
302+
303+
cn, err := opt.Dialer(ctx, opt.Network, opt.Addr)
304+
if err != nil {
305+
return nil, internal.RecordError(ctx, span, err)
306+
}
300307

301-
var err error
302-
conn, err = opt.Dialer(ctx, opt.Network, opt.Addr)
303-
if err != nil {
304-
_ = internal.RecordError(ctx, span, err)
305-
}
306-
return err
307-
})
308-
return conn, err
308+
return cn, nil
309309
},
310310
PoolSize: opt.PoolSize,
311311
MinIdleConns: opt.MinIdleConns,

‎redis.go‎

Lines changed: 71 additions & 68 deletions
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,6 @@ import (
1111
"github.com/go-redis/redis/v8/internal/pool"
1212
"github.com/go-redis/redis/v8/internal/proto"
1313
"go.opentelemetry.io/otel/attribute"
14-
"go.opentelemetry.io/otel/trace"
1514
)
1615

1716
// Nil reply returned by Redis when key does not exist.
@@ -214,10 +213,7 @@ func (c *baseClient) _getConn(ctx context.Context) (*pool.Conn, error) {
214213
return cn, nil
215214
}
216215

217-
err = internal.WithSpan(ctx, "redis.init_conn", func(ctx context.Context, span trace.Span) error {
218-
return c.initConn(ctx, cn)
219-
})
220-
if err != nil {
216+
if err := c.initConn(ctx, cn); err != nil {
221217
c.connPool.Remove(ctx, cn, err)
222218
if err := errors.Unwrap(err); err != nil {
223219
return nil, err
@@ -241,6 +237,9 @@ func (c *baseClient) initConn(ctx context.Context, cn *pool.Conn) error {
241237
return nil
242238
}
243239

240+
ctx, span := internal.StartSpan(ctx, "redis.init_conn")
241+
defer span.End()
242+
244243
connPool := pool.NewSingleConnPool(c.connPool, cn)
245244
conn := newConn(ctx, c.opt, connPool)
246245

@@ -288,91 +287,95 @@ func (c *baseClient) releaseConn(ctx context.Context, cn *pool.Conn, err error)
288287
func (c *baseClient) withConn(
289288
ctx context.Context, fn func(context.Context, *pool.Conn) error,
290289
) error {
291-
return internal.WithSpan(ctx, "redis.with_conn", func(ctx context.Context, span trace.Span) error {
292-
cn, err := c.getConn(ctx)
293-
if err != nil {
294-
return err
295-
}
290+
ctx, span := internal.StartSpan(ctx, "redis.with_conn")
291+
defer span.End()
296292

297-
if span.IsRecording() {
298-
if remoteAddr := cn.RemoteAddr(); remoteAddr != nil {
299-
span.SetAttributes(attribute.String("net.peer.ip", remoteAddr.String()))
300-
}
293+
cn, err := c.getConn(ctx)
294+
if err != nil {
295+
return err
296+
}
297+
298+
if span.IsRecording() {
299+
if remoteAddr := cn.RemoteAddr(); remoteAddr != nil {
300+
span.SetAttributes(attribute.String("net.peer.ip", remoteAddr.String()))
301301
}
302+
}
302303

303-
defer func() {
304-
c.releaseConn(ctx, cn, err)
305-
}()
304+
defer func() {
305+
c.releaseConn(ctx, cn, err)
306+
}()
306307

307-
done := ctx.Done()
308-
if done == nil {
309-
err = fn(ctx, cn)
310-
return err
311-
}
308+
done := ctx.Done()
309+
if done == nil {
310+
err = fn(ctx, cn)
311+
return err
312+
}
312313

313-
errc := make(chan error, 1)
314-
go func() { errc <- fn(ctx, cn) }()
314+
errc := make(chan error, 1)
315+
go func() { errc <- fn(ctx, cn) }()
315316

316-
select {
317-
case <-done:
318-
_ = cn.Close()
319-
// Wait for the goroutine to finish and send something.
320-
<-errc
317+
select {
318+
case <-done:
319+
_ = cn.Close()
320+
// Wait for the goroutine to finish and send something.
321+
<-errc
321322

322-
err = ctx.Err()
323-
return err
324-
case err = <-errc:
325-
return err
326-
}
327-
})
323+
err = ctx.Err()
324+
return err
325+
case err = <-errc:
326+
return err
327+
}
328328
}
329329

330330
func (c *baseClient) process(ctx context.Context, cmd Cmder) error {
331331
var lastErr error
332332
for attempt := 0; attempt <= c.opt.MaxRetries; attempt++ {
333333
attempt := attempt
334334

335-
var retry bool
336-
err := internal.WithSpan(ctx, "redis.process", func(ctx context.Context, span trace.Span) error {
337-
if attempt > 0 {
338-
if err := internal.Sleep(ctx, c.retryBackoff(attempt)); err != nil {
339-
return err
340-
}
341-
}
342-
343-
retryTimeout := uint32(1)
344-
err := c.withConn(ctx, func(ctx context.Context, cn *pool.Conn) error {
345-
err := cn.WithWriter(ctx, c.opt.WriteTimeout, func(wr *proto.Writer) error {
346-
return writeCmd(wr, cmd)
347-
})
348-
if err != nil {
349-
return err
350-
}
351-
352-
err = cn.WithReader(ctx, c.cmdTimeout(cmd), cmd.readReply)
353-
if err != nil {
354-
if cmd.readTimeout() == nil {
355-
atomic.StoreUint32(&retryTimeout, 1)
356-
}
357-
return err
358-
}
359-
360-
return nil
361-
})
362-
if err == nil {
363-
return nil
364-
}
365-
retry = shouldRetry(err, atomic.LoadUint32(&retryTimeout) == 1)
366-
return err
367-
})
335+
retry, err := c._process(ctx, cmd, attempt)
368336
if err == nil || !retry {
369337
return err
370338
}
339+
371340
lastErr = err
372341
}
373342
return lastErr
374343
}
375344

345+
func (c *baseClient) _process(ctx context.Context, cmd Cmder, attempt int) (bool, error) {
346+
if attempt > 0 {
347+
if err := internal.Sleep(ctx, c.retryBackoff(attempt)); err != nil {
348+
return false, err
349+
}
350+
}
351+
352+
retryTimeout := uint32(1)
353+
err := c.withConn(ctx, func(ctx context.Context, cn *pool.Conn) error {
354+
err := cn.WithWriter(ctx, c.opt.WriteTimeout, func(wr *proto.Writer) error {
355+
return writeCmd(wr, cmd)
356+
})
357+
if err != nil {
358+
return err
359+
}
360+
361+
err = cn.WithReader(ctx, c.cmdTimeout(cmd), cmd.readReply)
362+
if err != nil {
363+
if cmd.readTimeout() == nil {
364+
atomic.StoreUint32(&retryTimeout, 1)
365+
}
366+
return err
367+
}
368+
369+
return nil
370+
})
371+
if err == nil {
372+
return false, nil
373+
}
374+
375+
retry := shouldRetry(err, atomic.LoadUint32(&retryTimeout) == 1)
376+
return retry, err
377+
}
378+
376379
func (c *baseClient) retryBackoff(attempt int) time.Duration {
377380
return internal.RetryBackoff(attempt, c.opt.MinRetryBackoff, c.opt.MaxRetryBackoff)
378381
}

0 commit comments

Comments
 (0)