Merge branch 'master' of github.com:redis/go-redis into vv-client-kil…

…l-max-age

CHANGELOG.md

@@ -1,3 +1,12 @@
+## Unreleased
+
+### Changed
+
+* `go-redis` won't skip span creation if the parent spans is not recording. ([#2980](https://github.com/redis/go-redis/issues/2980))
+  Users can use the OpenTelemetry sampler to control the sampling behavior.
+  For instance, you can use the `ParentBased(NeverSample())` sampler from `go.opentelemetry.io/otel/sdk/trace` to keep
+  a similar behavior (drop orphan spans) of `go-redis` as before.
+
 ## [9.0.5](https://github.com/redis/go-redis/compare/v9.0.4...v9.0.5) (2023-05-29)
 
 

bitmap_commands.go

@@ -16,6 +16,7 @@ type BitMapCmdable interface {
 	BitPos(ctx context.Context, key string, bit int64, pos ...int64) *IntCmd
 	BitPosSpan(ctx context.Context, key string, bit int8, start, end int64, span string) *IntCmd
 	BitField(ctx context.Context, key string, values ...interface{}) *IntSliceCmd
+	BitFieldRO(ctx context.Context, key string, values ...interface{}) *IntSliceCmd
 }
 
 func (c cmdable) GetBit(ctx context.Context, key string, offset int64) *IntCmd {

error.go

@@ -7,6 +7,7 @@ import (
 	"net"
 	"strings"
 
+	"github.com/redis/go-redis/v9/internal"
 	"github.com/redis/go-redis/v9/internal/pool"
 	"github.com/redis/go-redis/v9/internal/proto"
 )
@@ -129,7 +130,9 @@ func isMovedError(err error) (moved bool, ask bool, addr string) {
 	if ind == -1 {
 		return false, false, ""
 	}
+
 	addr = s[ind+1:]
+	addr = internal.GetAddr(addr)
 	return
 }
 

example_test.go

@@ -482,7 +482,7 @@ func ExampleClient_Watch() {
 				return err
 			}
 
-			// Actual opperation (local in optimistic lock).
+			// Actual operation (local in optimistic lock).
 			n++
 
 			// Operation is committed only if the watched keys remain unchanged.

extra/redisotel/tracing.go

@@ -91,10 +91,6 @@ func newTracingHook(connString string, opts ...TracingOption) *tracingHook {
 
 func (th *tracingHook) DialHook(hook redis.DialHook) redis.DialHook {
 	return func(ctx context.Context, network, addr string) (net.Conn, error) {
-		if !trace.SpanFromContext(ctx).IsRecording() {
-			return hook(ctx, network, addr)
-		}
-
 		ctx, span := th.conf.tracer.Start(ctx, "redis.dial", th.spanOpts...)
 		defer span.End()
 
@@ -109,10 +105,6 @@ func (th *tracingHook) DialHook(hook redis.DialHook) redis.DialHook {
 
 func (th *tracingHook) ProcessHook(hook redis.ProcessHook) redis.ProcessHook {
 	return func(ctx context.Context, cmd redis.Cmder) error {
-		if !trace.SpanFromContext(ctx).IsRecording() {
-			return hook(ctx, cmd)
-		}
-
 		fn, file, line := funcFileLine("github.com/redis/go-redis")
 
 		attrs := make([]attribute.KeyValue, 0, 8)
@@ -145,10 +137,6 @@ func (th *tracingHook) ProcessPipelineHook(
 	hook redis.ProcessPipelineHook,
 ) redis.ProcessPipelineHook {
 	return func(ctx context.Context, cmds []redis.Cmder) error {
-		if !trace.SpanFromContext(ctx).IsRecording() {
-			return hook(ctx, cmds)
-		}
-
 		fn, file, line := funcFileLine("github.com/redis/go-redis")
 
 		attrs := make([]attribute.KeyValue, 0, 8)

internal/util.go

@@ -2,6 +2,7 @@ package internal
 
 import (
 	"context"
+	"net"
 	"strings"
 	"time"
 
@@ -64,3 +65,19 @@ func ReplaceSpaces(s string) string {
 
 	return builder.String()
 }
+
+func GetAddr(addr string) string {
+	ind := strings.LastIndexByte(addr, ':')
+	if ind == -1 {
+		return ""
+	}
+
+	if strings.IndexByte(addr, '.') != -1 {
+		return addr
+	}
+
+	if addr[0] == '[' {
+		return addr
+	}
+	return net.JoinHostPort(addr[:ind], addr[ind+1:])
+}

internal/util_test.go

@@ -51,3 +51,24 @@ func TestIsLower(t *testing.T) {
 		Expect(isLower(str)).To(BeTrue())
 	})
 }
+
+func TestGetAddr(t *testing.T) {
+	It("getAddr", func() {
+		str := "127.0.0.1:1234"
+		Expect(GetAddr(str)).To(Equal(str))
+
+		str = "[::1]:1234"
+		Expect(GetAddr(str)).To(Equal(str))
+
+		str = "[fd01:abcd::7d03]:6379"
+		Expect(GetAddr(str)).To(Equal(str))
+
+		Expect(GetAddr("::1:1234")).To(Equal("[::1]:1234"))
+
+		Expect(GetAddr("fd01:abcd::7d03:6379")).To(Equal("[fd01:abcd::7d03]:6379"))
+
+		Expect(GetAddr("127.0.0.1")).To(Equal(""))
+
+		Expect(GetAddr("127")).To(Equal(""))
+	})
+}

stream_commands.go

@@ -178,36 +178,42 @@ func (c cmdable) XReadStreams(ctx context.Context, streams ...string) *XStreamSl
 
 func (c cmdable) XGroupCreate(ctx context.Context, stream, group, start string) *StatusCmd {
 	cmd := NewStatusCmd(ctx, "xgroup", "create", stream, group, start)
+	cmd.SetFirstKeyPos(2)
 	_ = c(ctx, cmd)
 	return cmd
 }
 
 func (c cmdable) XGroupCreateMkStream(ctx context.Context, stream, group, start string) *StatusCmd {
 	cmd := NewStatusCmd(ctx, "xgroup", "create", stream, group, start, "mkstream")
+	cmd.SetFirstKeyPos(2)
 	_ = c(ctx, cmd)
 	return cmd
 }
 
 func (c cmdable) XGroupSetID(ctx context.Context, stream, group, start string) *StatusCmd {
 	cmd := NewStatusCmd(ctx, "xgroup", "setid", stream, group, start)
+	cmd.SetFirstKeyPos(2)
 	_ = c(ctx, cmd)
 	return cmd
 }
 
 func (c cmdable) XGroupDestroy(ctx context.Context, stream, group string) *IntCmd {
 	cmd := NewIntCmd(ctx, "xgroup", "destroy", stream, group)
+	cmd.SetFirstKeyPos(2)
 	_ = c(ctx, cmd)
 	return cmd
 }
 
 func (c cmdable) XGroupCreateConsumer(ctx context.Context, stream, group, consumer string) *IntCmd {
 	cmd := NewIntCmd(ctx, "xgroup", "createconsumer", stream, group, consumer)
+	cmd.SetFirstKeyPos(2)
 	_ = c(ctx, cmd)
 	return cmd
 }
 
 func (c cmdable) XGroupDelConsumer(ctx context.Context, stream, group, consumer string) *IntCmd {
 	cmd := NewIntCmd(ctx, "xgroup", "delconsumer", stream, group, consumer)
+	cmd.SetFirstKeyPos(2)
 	_ = c(ctx, cmd)
 	return cmd
 }