Compare commits

..

No commits in common. "v3" and "v3.10.24" have entirely different histories.
v3 ... v3.10.24

3 changed files with 35 additions and 66 deletions

View File

@ -2,7 +2,6 @@ package redis
import ( import (
"context" "context"
"errors"
"reflect" "reflect"
"strings" "strings"
"time" "time"
@ -119,8 +118,6 @@ func (r *Store) Exists(ctx context.Context, key string, opts ...store.ExistsOpti
b := r.pool.Get() b := r.pool.Get()
defer r.pool.Put(b) defer r.pool.Put(b)
options := store.NewExistsOptions(opts...) options := store.NewExistsOptions(opts...)
labels := make([]string, 0, 6)
labels = append(labels, "name", options.Name, "statement", "exists")
timeout := r.opts.Timeout timeout := r.opts.Timeout
if options.Timeout > 0 { 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) 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() ts := time.Now()
val, err := r.cli.Exists(ctx, rkey).Result() val, err := r.cli.Exists(ctx, rkey).Result()
setSpanError(ctx, err) setSpanError(ctx, err)
te := time.Since(ts) te := time.Since(ts)
r.opts.Meter.Counter(semconv.StoreRequestInflight, labels...).Dec() 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.Summary(semconv.StoreRequestLatencyMicroseconds, labels...).Update(te.Seconds()) r.opts.Meter.Histogram(semconv.StoreRequestDurationSeconds, "name", options.Name).Update(te.Seconds())
r.opts.Meter.Histogram(semconv.StoreRequestDurationSeconds, labels...).Update(te.Seconds()) if err == goredis.Nil || (err == nil && val == 0) {
if errors.Is(err, goredis.Nil) || (err == nil && val == 0) { r.opts.Meter.Counter(semconv.StoreRequestTotal, "name", options.Name, "status", "miss").Inc()
r.opts.Meter.Counter(semconv.StoreRequestTotal, append(labels, "status", "miss")...).Inc()
return store.ErrNotFound return store.ErrNotFound
} else if err == nil { } 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 { } 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 return err
} }
@ -162,8 +158,6 @@ func (r *Store) Read(ctx context.Context, key string, val interface{}, opts ...s
defer r.pool.Put(b) defer r.pool.Put(b)
options := store.NewReadOptions(opts...) options := store.NewReadOptions(opts...)
labels := make([]string, 0, 6)
labels = append(labels, "name", options.Name, "statement", "read")
timeout := r.opts.Timeout timeout := r.opts.Timeout
if options.Timeout > 0 { 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) 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() ts := time.Now()
buf, err := r.cli.Get(ctx, rkey).Bytes() buf, err := r.cli.Get(ctx, rkey).Bytes()
setSpanError(ctx, err) setSpanError(ctx, err)
te := time.Since(ts) te := time.Since(ts)
r.opts.Meter.Counter(semconv.StoreRequestInflight, labels...).Dec() r.opts.Meter.Counter(semconv.StoreRequestInflight, "name", options.Name).Dec()
r.opts.Meter.Summary(semconv.StoreRequestLatencyMicroseconds, labels...).Update(te.Seconds()) r.opts.Meter.Summary(semconv.StoreRequestLatencyMicroseconds, "name", options.Name).Update(te.Seconds())
r.opts.Meter.Histogram(semconv.StoreRequestDurationSeconds, labels...).Update(te.Seconds()) r.opts.Meter.Histogram(semconv.StoreRequestDurationSeconds, "name", options.Name).Update(te.Seconds())
if errors.Is(err, goredis.Nil) || (err == nil && buf == nil) { if err == goredis.Nil || (err == nil && buf == nil) {
r.opts.Meter.Counter(semconv.StoreRequestTotal, append(labels, "status", "miss")...).Inc() r.opts.Meter.Counter(semconv.StoreRequestTotal, "name", options.Name, "status", "miss").Inc()
return store.ErrNotFound return store.ErrNotFound
} else if err == nil { } 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 { } 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 return err
} }
@ -369,8 +363,6 @@ func (r *Store) Delete(ctx context.Context, key string, opts ...store.DeleteOpti
defer r.pool.Put(b) defer r.pool.Put(b)
options := store.NewDeleteOptions(opts...) options := store.NewDeleteOptions(opts...)
labels := make([]string, 0, 6)
labels = append(labels, "name", options.Name, "statement", "delete")
timeout := r.opts.Timeout timeout := r.opts.Timeout
if options.Timeout > 0 { if options.Timeout > 0 {
@ -383,22 +375,21 @@ func (r *Store) Delete(ctx context.Context, key string, opts ...store.DeleteOpti
defer cancel() defer cancel()
} }
r.opts.Meter.Counter(semconv.StoreRequestInflight, labels...).Inc() r.opts.Meter.Counter(semconv.StoreRequestInflight, "name", options.Name).Inc()
ts := time.Now() ts := time.Now()
err := r.cli.Del(ctx, r.getKey(b, r.opts.Namespace, options.Namespace, key)).Err() err := r.cli.Del(ctx, r.getKey(b, r.opts.Namespace, options.Namespace, key)).Err()
te := time.Since(ts) te := time.Since(ts)
setSpanError(ctx, err) setSpanError(ctx, err)
r.opts.Meter.Counter(semconv.StoreRequestInflight, "name", options.Name).Dec()
r.opts.Meter.Counter(semconv.StoreRequestInflight, labels...).Dec() r.opts.Meter.Summary(semconv.StoreRequestLatencyMicroseconds, "name", options.Name).Update(te.Seconds())
r.opts.Meter.Summary(semconv.StoreRequestLatencyMicroseconds, labels...).Update(te.Seconds()) r.opts.Meter.Histogram(semconv.StoreRequestDurationSeconds, "name", options.Name).Update(te.Seconds())
r.opts.Meter.Histogram(semconv.StoreRequestDurationSeconds, labels...).Update(te.Seconds()) if err == goredis.Nil {
if errors.Is(err, goredis.Nil) { r.opts.Meter.Counter(semconv.StoreRequestTotal, "name", options.Name, "status", "miss").Inc()
r.opts.Meter.Counter(semconv.StoreRequestTotal, append(labels, "status", "miss")...).Inc()
return store.ErrNotFound return store.ErrNotFound
} else if err == nil { } 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 { } 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 return err
} }
@ -493,8 +484,6 @@ func (r *Store) Write(ctx context.Context, key string, val interface{}, opts ...
defer r.pool.Put(b) defer r.pool.Put(b)
options := store.NewWriteOptions(opts...) options := store.NewWriteOptions(opts...)
labels := make([]string, 0, 6)
labels = append(labels, "name", options.Name, "statement", "write")
timeout := r.opts.Timeout timeout := r.opts.Timeout
if options.Timeout > 0 { 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() ts := time.Now()
err := r.cli.Set(ctx, rkey, buf, options.TTL).Err() err := r.cli.Set(ctx, rkey, buf, options.TTL).Err()
te := time.Since(ts) te := time.Since(ts)
setSpanError(ctx, err) setSpanError(ctx, err)
r.opts.Meter.Counter(semconv.StoreRequestInflight, labels...).Dec() r.opts.Meter.Counter(semconv.StoreRequestInflight, "name", options.Name).Dec()
r.opts.Meter.Summary(semconv.StoreRequestLatencyMicroseconds, labels...).Update(te.Seconds()) r.opts.Meter.Summary(semconv.StoreRequestLatencyMicroseconds, "name", options.Name).Update(te.Seconds())
r.opts.Meter.Histogram(semconv.StoreRequestDurationSeconds, labels...).Update(te.Seconds()) r.opts.Meter.Histogram(semconv.StoreRequestDurationSeconds, "name", options.Name).Update(te.Seconds())
if errors.Is(err, goredis.Nil) { if err == goredis.Nil {
r.opts.Meter.Counter(semconv.StoreRequestTotal, append(labels, "status", "miss")...).Inc() r.opts.Meter.Counter(semconv.StoreRequestTotal, "name", options.Name, "status", "miss").Inc()
return store.ErrNotFound return store.ErrNotFound
} else if err == nil { } 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 { } 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 return err
} }

View File

@ -12,26 +12,6 @@ import (
"go.unistack.org/micro/v3/tracer" "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) { func TestKeepTTL(t *testing.T) {
ctx := context.Background() ctx := context.Background()

View File

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