go.mod

@@ -5,9 +5,9 @@ go 1.22
 toolchain go1.22.4
 
 require (
-	github.com/redis/go-redis/extra/rediscmd/v9 v9.6.2
-	github.com/redis/go-redis/v9 v9.6.2
-	go.unistack.org/micro/v3 v3.10.97
+	github.com/redis/go-redis/extra/rediscmd/v9 v9.0.5
+	github.com/redis/go-redis/v9 v9.6.1
+	go.unistack.org/micro/v3 v3.10.96
 )
 
 require (

go.sum

@@ -1,20 +1,24 @@
+github.com/bsm/ginkgo/v2 v2.7.0/go.mod h1:AiKlXPm7ItEHNc/2+OkrNG4E0ITzojb9/xWzvQ9XZ9w=
 github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
 github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
+github.com/bsm/gomega v1.26.0/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
 github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
 github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
+github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
 github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
 github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
 github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
-github.com/redis/go-redis/extra/rediscmd/v9 v9.6.2 h1:oBlErygFka9FAVfowzW7sRUfB7n31YX0aFzfaOflM3w=
-github.com/redis/go-redis/extra/rediscmd/v9 v9.6.2/go.mod h1:jYLUE5tC8UsFnpSclhEjZlFLMMtHH4jTfFMnshNWkoo=
-github.com/redis/go-redis/v9 v9.6.2 h1:w0uvkRbc9KpgD98zcvo5IrVUsn0lXpRMuhNgiHDJzdk=
-github.com/redis/go-redis/v9 v9.6.2/go.mod h1:0C0c6ycQsdpVNQpxb1njEQIqkx5UcsM8FJCQLgE9+RA=
+github.com/redis/go-redis/extra/rediscmd/v9 v9.0.5 h1:EaDatTxkdHG+U3Bk4EUr+DZ7fOGwTfezUiUJMaIcaho=
+github.com/redis/go-redis/extra/rediscmd/v9 v9.0.5/go.mod h1:fyalQWdtzDBECAQFBJuQe5bzQ02jGd5Qcbgb97Flm7U=
+github.com/redis/go-redis/v9 v9.0.5/go.mod h1:WqMKv5vnQbRuZstUwxQI195wHy+t4PuXDOjzMvcuQHk=
+github.com/redis/go-redis/v9 v9.6.1 h1:HHDteefn6ZkTtY5fGUE8tj8uy85AHk6zP7CpzIAM0y4=
+github.com/redis/go-redis/v9 v9.6.1/go.mod h1:0C0c6ycQsdpVNQpxb1njEQIqkx5UcsM8FJCQLgE9+RA=
 go.unistack.org/micro-proto/v3 v3.4.1 h1:UTjLSRz2YZuaHk9iSlVqqsA50JQNAEK2ZFboGqtEa9Q=
 go.unistack.org/micro-proto/v3 v3.4.1/go.mod h1:okx/cnOhzuCX0ggl/vToatbCupi0O44diiiLLsZ93Zo=
-go.unistack.org/micro/v3 v3.10.97 h1:8l7fv+i06/PjPrBBhRC/ZQkWGIOuHPg3jJN0vktYE78=
-go.unistack.org/micro/v3 v3.10.97/go.mod h1:YzMldzHN9Ei+zy5t/Psu7RUWDZwUfrNYiStSQtTz90g=
+go.unistack.org/micro/v3 v3.10.96 h1:njukb0DlXOJspX5mLSwcP7+cVSg/8Fy+6++mYkOpY2A=
+go.unistack.org/micro/v3 v3.10.96/go.mod h1:YzMldzHN9Ei+zy5t/Psu7RUWDZwUfrNYiStSQtTz90g=
 google.golang.org/protobuf v1.35.1 h1:m3LfL6/Ca+fqnjnlqQXNpFPABW1UD7mjh8KO2mKFytA=
 google.golang.org/protobuf v1.35.1/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=

options.go

@@ -61,7 +61,7 @@ func NewOptions(opts ...Option) Options {
 			labelName, options.RedisName),
 	)
 
-	options.Logger = options.Logger.Clone(logger.WithAddCallerSkipCount(1))
+	options.Logger = options.Logger.Clone(logger.WithCallerSkipCount(1))
 
 	return options
 }

redis.go

@@ -2,7 +2,6 @@ package redis
 
 import (
 	"context"
-	"errors"
 	"reflect"
 	"strings"
 	"time"
@@ -119,8 +118,6 @@ func (r *Store) Exists(ctx context.Context, key string, opts ...store.ExistsOpti
 	b := r.pool.Get()
 	defer r.pool.Put(b)
 	options := store.NewExistsOptions(opts...)
-	labels := make([]string, 0, 6)
-	labels = append(labels, "name", options.Name, "statement", "exists")
 
 	timeout := r.opts.Timeout
 	if options.Timeout > 0 {
@@ -135,22 +132,21 @@ func (r *Store) Exists(ctx context.Context, key string, opts ...store.ExistsOpti
 
 	rkey := r.getKey(b, r.opts.Namespace, options.Namespace, key)
 
-	r.opts.Meter.Counter(semconv.StoreRequestInflight, labels...).Inc()
+	r.opts.Meter.Counter(semconv.StoreRequestInflight, "name", options.Name).Inc()
 	ts := time.Now()
 	val, err := r.cli.Exists(ctx, rkey).Result()
 	setSpanError(ctx, err)
 	te := time.Since(ts)
-	r.opts.Meter.Counter(semconv.StoreRequestInflight, labels...).Dec()
-
-	r.opts.Meter.Summary(semconv.StoreRequestLatencyMicroseconds, labels...).Update(te.Seconds())
-	r.opts.Meter.Histogram(semconv.StoreRequestDurationSeconds, labels...).Update(te.Seconds())
-	if errors.Is(err, goredis.Nil) || (err == nil && val == 0) {
-		r.opts.Meter.Counter(semconv.StoreRequestTotal, append(labels, "status", "miss")...).Inc()
+	r.opts.Meter.Counter(semconv.StoreRequestInflight, "name", options.Name).Dec()
+	r.opts.Meter.Summary(semconv.StoreRequestLatencyMicroseconds, "name", options.Name).Update(te.Seconds())
+	r.opts.Meter.Histogram(semconv.StoreRequestDurationSeconds, "name", options.Name).Update(te.Seconds())
+	if err == goredis.Nil || (err == nil && val == 0) {
+		r.opts.Meter.Counter(semconv.StoreRequestTotal, "name", options.Name, "status", "miss").Inc()
 		return store.ErrNotFound
 	} else if err == nil {
-		r.opts.Meter.Counter(semconv.StoreRequestTotal, append(labels, "status", "hit")...).Inc()
+		r.opts.Meter.Counter(semconv.StoreRequestTotal, "name", options.Name, "status", "hit").Inc()
 	} else if err != nil {
-		r.opts.Meter.Counter(semconv.StoreRequestTotal, append(labels, "status", "failure")...).Inc()
+		r.opts.Meter.Counter(semconv.StoreRequestTotal, "name", options.Name, "status", "failure").Inc()
 		return err
 	}
 
@@ -162,8 +158,6 @@ func (r *Store) Read(ctx context.Context, key string, val interface{}, opts ...s
 	defer r.pool.Put(b)
 
 	options := store.NewReadOptions(opts...)
-	labels := make([]string, 0, 6)
-	labels = append(labels, "name", options.Name, "statement", "read")
 
 	timeout := r.opts.Timeout
 	if options.Timeout > 0 {
@@ -178,21 +172,21 @@ func (r *Store) Read(ctx context.Context, key string, val interface{}, opts ...s
 
 	rkey := r.getKey(b, r.opts.Namespace, options.Namespace, key)
 
-	r.opts.Meter.Counter(semconv.StoreRequestInflight, labels...).Inc()
+	r.opts.Meter.Counter(semconv.StoreRequestInflight, "name", options.Name).Inc()
 	ts := time.Now()
 	buf, err := r.cli.Get(ctx, rkey).Bytes()
 	setSpanError(ctx, err)
 	te := time.Since(ts)
-	r.opts.Meter.Counter(semconv.StoreRequestInflight, labels...).Dec()
-	r.opts.Meter.Summary(semconv.StoreRequestLatencyMicroseconds, labels...).Update(te.Seconds())
-	r.opts.Meter.Histogram(semconv.StoreRequestDurationSeconds, labels...).Update(te.Seconds())
-	if errors.Is(err, goredis.Nil) || (err == nil && buf == nil) {
-		r.opts.Meter.Counter(semconv.StoreRequestTotal, append(labels, "status", "miss")...).Inc()
+	r.opts.Meter.Counter(semconv.StoreRequestInflight, "name", options.Name).Dec()
+	r.opts.Meter.Summary(semconv.StoreRequestLatencyMicroseconds, "name", options.Name).Update(te.Seconds())
+	r.opts.Meter.Histogram(semconv.StoreRequestDurationSeconds, "name", options.Name).Update(te.Seconds())
+	if err == goredis.Nil || (err == nil && buf == nil) {
+		r.opts.Meter.Counter(semconv.StoreRequestTotal, "name", options.Name, "status", "miss").Inc()
 		return store.ErrNotFound
 	} else if err == nil {
-		r.opts.Meter.Counter(semconv.StoreRequestTotal, append(labels, "status", "hit")...).Inc()
+		r.opts.Meter.Counter(semconv.StoreRequestTotal, "name", options.Name, "status", "hit").Inc()
 	} else if err != nil {
-		r.opts.Meter.Counter(semconv.StoreRequestTotal, append(labels, "status", "failure")...).Inc()
+		r.opts.Meter.Counter(semconv.StoreRequestTotal, "name", options.Name, "status", "failure").Inc()
 		return err
 	}
 
@@ -369,8 +363,6 @@ func (r *Store) Delete(ctx context.Context, key string, opts ...store.DeleteOpti
 	defer r.pool.Put(b)
 
 	options := store.NewDeleteOptions(opts...)
-	labels := make([]string, 0, 6)
-	labels = append(labels, "name", options.Name, "statement", "delete")
 
 	timeout := r.opts.Timeout
 	if options.Timeout > 0 {
@@ -383,22 +375,21 @@ func (r *Store) Delete(ctx context.Context, key string, opts ...store.DeleteOpti
 		defer cancel()
 	}
 
-	r.opts.Meter.Counter(semconv.StoreRequestInflight, labels...).Inc()
+	r.opts.Meter.Counter(semconv.StoreRequestInflight, "name", options.Name).Inc()
 	ts := time.Now()
 	err := r.cli.Del(ctx, r.getKey(b, r.opts.Namespace, options.Namespace, key)).Err()
 	te := time.Since(ts)
 	setSpanError(ctx, err)
-
-	r.opts.Meter.Counter(semconv.StoreRequestInflight, labels...).Dec()
-	r.opts.Meter.Summary(semconv.StoreRequestLatencyMicroseconds, labels...).Update(te.Seconds())
-	r.opts.Meter.Histogram(semconv.StoreRequestDurationSeconds, labels...).Update(te.Seconds())
-	if errors.Is(err, goredis.Nil) {
-		r.opts.Meter.Counter(semconv.StoreRequestTotal, append(labels, "status", "miss")...).Inc()
+	r.opts.Meter.Counter(semconv.StoreRequestInflight, "name", options.Name).Dec()
+	r.opts.Meter.Summary(semconv.StoreRequestLatencyMicroseconds, "name", options.Name).Update(te.Seconds())
+	r.opts.Meter.Histogram(semconv.StoreRequestDurationSeconds, "name", options.Name).Update(te.Seconds())
+	if err == goredis.Nil {
+		r.opts.Meter.Counter(semconv.StoreRequestTotal, "name", options.Name, "status", "miss").Inc()
 		return store.ErrNotFound
 	} else if err == nil {
-		r.opts.Meter.Counter(semconv.StoreRequestTotal, append(labels, "status", "hit")...).Inc()
+		r.opts.Meter.Counter(semconv.StoreRequestTotal, "name", options.Name, "status", "hit").Inc()
 	} else if err != nil {
-		r.opts.Meter.Counter(semconv.StoreRequestTotal, append(labels, "status", "failure")...).Inc()
+		r.opts.Meter.Counter(semconv.StoreRequestTotal, "name", options.Name, "status", "failure").Inc()
 		return err
 	}
 
@@ -493,8 +484,6 @@ func (r *Store) Write(ctx context.Context, key string, val interface{}, opts ...
 	defer r.pool.Put(b)
 
 	options := store.NewWriteOptions(opts...)
-	labels := make([]string, 0, 6)
-	labels = append(labels, "name", options.Name, "statement", "write")
 
 	timeout := r.opts.Timeout
 	if options.Timeout > 0 {
@@ -523,22 +512,22 @@ func (r *Store) Write(ctx context.Context, key string, val interface{}, opts ...
 		}
 	}
 
-	r.opts.Meter.Counter(semconv.StoreRequestInflight, labels...).Inc()
+	r.opts.Meter.Counter(semconv.StoreRequestInflight, "name", options.Name).Inc()
 	ts := time.Now()
 	err := r.cli.Set(ctx, rkey, buf, options.TTL).Err()
 	te := time.Since(ts)
 	setSpanError(ctx, err)
 
-	r.opts.Meter.Counter(semconv.StoreRequestInflight, labels...).Dec()
-	r.opts.Meter.Summary(semconv.StoreRequestLatencyMicroseconds, labels...).Update(te.Seconds())
-	r.opts.Meter.Histogram(semconv.StoreRequestDurationSeconds, labels...).Update(te.Seconds())
-	if errors.Is(err, goredis.Nil) {
-		r.opts.Meter.Counter(semconv.StoreRequestTotal, append(labels, "status", "miss")...).Inc()
+	r.opts.Meter.Counter(semconv.StoreRequestInflight, "name", options.Name).Dec()
+	r.opts.Meter.Summary(semconv.StoreRequestLatencyMicroseconds, "name", options.Name).Update(te.Seconds())
+	r.opts.Meter.Histogram(semconv.StoreRequestDurationSeconds, "name", options.Name).Update(te.Seconds())
+	if err == goredis.Nil {
+		r.opts.Meter.Counter(semconv.StoreRequestTotal, "name", options.Name, "status", "miss").Inc()
 		return store.ErrNotFound
 	} else if err == nil {
-		r.opts.Meter.Counter(semconv.StoreRequestTotal, append(labels, "status", "hit")...).Inc()
+		r.opts.Meter.Counter(semconv.StoreRequestTotal, "name", options.Name, "status", "hit").Inc()
 	} else if err != nil {
-		r.opts.Meter.Counter(semconv.StoreRequestTotal, append(labels, "status", "failure")...).Inc()
+		r.opts.Meter.Counter(semconv.StoreRequestTotal, "name", options.Name, "status", "failure").Inc()
 		return err
 	}
 
@@ -636,12 +625,12 @@ func NewStore(opts ...store.Option) *Store {
 }
 
 func (r *Store) configure() error {
+	universalOptions := &goredis.UniversalOptions{}
+
 	if r.cli != nil && r.opts.Context == nil {
 		return nil
 	}
 
-	universalOptions := DefaultUniversalOptions
-
 	if r.opts.Context != nil {
 		if o, ok := r.opts.Context.Value(configKey{}).(*goredis.Options); ok {
 			universalOptions.Addrs = []string{o.Addr}
@@ -715,6 +704,10 @@ func (r *Store) configure() error {
 		}
 	}
 
+	if universalOptions == nil {
+		universalOptions = DefaultUniversalOptions
+	}
+
 	if len(r.opts.Addrs) > 0 {
 		universalOptions.Addrs = r.opts.Addrs
 	} else {

redis_test.go

@@ -12,26 +12,6 @@ import (
 	"go.unistack.org/micro/v3/tracer"
 )
 
-func TestLazyConnect(t *testing.T) {
-	ctx := context.Background()
-	var err error
-
-	r := NewStore()
-
-	if err = r.Init(); err != nil {
-		t.Fatal(err)
-	}
-	if err = r.Connect(ctx); err != nil {
-		t.Logf("connect failed %v", err)
-	}
-
-	for {
-		if err = r.Write(ctx, "mykey", "myval"); err != nil {
-			t.Logf("failed to write %v", err)
-		}
-	}
-}
-
 func TestKeepTTL(t *testing.T) {
 	ctx := context.Background()
 

stats.go

@@ -33,17 +33,20 @@ func (r *Store) statsMeter() {
 		ticker := time.NewTicker(meter.DefaultMeterStatsInterval)
 		defer ticker.Stop()
 
-		for _ = range ticker.C {
-			if st == nil {
-				return
+		for {
+			select {
+			case <-ticker.C:
+				if st == nil {
+					return
+				}
+				stats := st.PoolStats()
+				r.opts.Meter.Counter(PoolHitsTotal).Set(uint64(stats.Hits))
+				r.opts.Meter.Counter(PoolMissesTotal).Set(uint64(stats.Misses))
+				r.opts.Meter.Counter(PoolTimeoutTotal).Set(uint64(stats.Timeouts))
+				r.opts.Meter.Counter(PoolConnTotalCurrent).Set(uint64(stats.TotalConns))
+				r.opts.Meter.Counter(PoolConnIdleCurrent).Set(uint64(stats.IdleConns))
+				r.opts.Meter.Counter(PoolConnStaleTotal).Set(uint64(stats.StaleConns))
 			}
-			stats := st.PoolStats()
-			r.opts.Meter.Counter(PoolHitsTotal).Set(uint64(stats.Hits))
-			r.opts.Meter.Counter(PoolMissesTotal).Set(uint64(stats.Misses))
-			r.opts.Meter.Counter(PoolTimeoutTotal).Set(uint64(stats.Timeouts))
-			r.opts.Meter.Counter(PoolConnTotalCurrent).Set(uint64(stats.TotalConns))
-			r.opts.Meter.Counter(PoolConnIdleCurrent).Set(uint64(stats.IdleConns))
-			r.opts.Meter.Counter(PoolConnStaleTotal).Set(uint64(stats.StaleConns))
 		}
 	}()
 }

tracer.go

@@ -73,7 +73,7 @@ func (h *tracingHook) ProcessHook(hook goredis.ProcessHook) goredis.ProcessHook
 		case "cluster slots":
 			break
 		default:
-			_, span := h.tr.Start(ctx, "sdk.database", append(h.opts, tracer.WithSpanLabels("db.statement", cmdString))...)
+			_, span := h.tr.Start(ctx, "goredis.process", append(h.opts, tracer.WithSpanLabels("db.statement", cmdString))...)
 			defer func() {
 				recordError(span, err)
 				span.Finish()
@@ -91,11 +91,11 @@ func (h *tracingHook) ProcessPipelineHook(hook goredis.ProcessPipelineHook) gore
 		_, cmdsString := rediscmd.CmdsString(cmds)
 
 		opts := append(h.opts, tracer.WithSpanLabels(
-			"db.database.num_cmd", strconv.Itoa(len(cmds)),
+			"db.goredis.num_cmd", strconv.Itoa(len(cmds)),
 			"db.statement", cmdsString,
 		))
 
-		_, span := h.tr.Start(ctx, "sdk.database", opts...)
+		_, span := h.tr.Start(ctx, "goredis.process_pipeline", opts...)
 		defer span.Finish()
 
 		err := hook(ctx, cmds)