Compare commits
No commits in common. "v3" and "v3.10.24" have entirely different histories.
75
redis.go
75
redis.go
@ -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
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -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()
|
||||||
|
|
||||||
|
@ -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)
|
||||||
|
Loading…
Reference in New Issue
Block a user