Compare commits
2 Commits
Author | SHA1 | Date | |
---|---|---|---|
77fe9cbef6 | |||
c59902201d |
75
redis.go
75
redis.go
@ -2,6 +2,7 @@ package redis
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"reflect"
|
||||
"strings"
|
||||
"time"
|
||||
@ -118,6 +119,8 @@ 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 {
|
||||
@ -132,21 +135,22 @@ 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, "name", options.Name).Inc()
|
||||
r.opts.Meter.Counter(semconv.StoreRequestInflight, labels...).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, "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()
|
||||
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()
|
||||
return store.ErrNotFound
|
||||
} else if err == nil {
|
||||
r.opts.Meter.Counter(semconv.StoreRequestTotal, "name", options.Name, "status", "hit").Inc()
|
||||
r.opts.Meter.Counter(semconv.StoreRequestTotal, append(labels, "status", "hit")...).Inc()
|
||||
} else if err != nil {
|
||||
r.opts.Meter.Counter(semconv.StoreRequestTotal, "name", options.Name, "status", "failure").Inc()
|
||||
r.opts.Meter.Counter(semconv.StoreRequestTotal, append(labels, "status", "failure")...).Inc()
|
||||
return err
|
||||
}
|
||||
|
||||
@ -158,6 +162,8 @@ 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 {
|
||||
@ -172,21 +178,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, "name", options.Name).Inc()
|
||||
r.opts.Meter.Counter(semconv.StoreRequestInflight, labels...).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, "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()
|
||||
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()
|
||||
return store.ErrNotFound
|
||||
} else if err == nil {
|
||||
r.opts.Meter.Counter(semconv.StoreRequestTotal, "name", options.Name, "status", "hit").Inc()
|
||||
r.opts.Meter.Counter(semconv.StoreRequestTotal, append(labels, "status", "hit")...).Inc()
|
||||
} else if err != nil {
|
||||
r.opts.Meter.Counter(semconv.StoreRequestTotal, "name", options.Name, "status", "failure").Inc()
|
||||
r.opts.Meter.Counter(semconv.StoreRequestTotal, append(labels, "status", "failure")...).Inc()
|
||||
return err
|
||||
}
|
||||
|
||||
@ -363,6 +369,8 @@ 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 {
|
||||
@ -375,21 +383,22 @@ func (r *Store) Delete(ctx context.Context, key string, opts ...store.DeleteOpti
|
||||
defer cancel()
|
||||
}
|
||||
|
||||
r.opts.Meter.Counter(semconv.StoreRequestInflight, "name", options.Name).Inc()
|
||||
r.opts.Meter.Counter(semconv.StoreRequestInflight, labels...).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, "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()
|
||||
|
||||
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()
|
||||
return store.ErrNotFound
|
||||
} else if err == nil {
|
||||
r.opts.Meter.Counter(semconv.StoreRequestTotal, "name", options.Name, "status", "hit").Inc()
|
||||
r.opts.Meter.Counter(semconv.StoreRequestTotal, append(labels, "status", "hit")...).Inc()
|
||||
} else if err != nil {
|
||||
r.opts.Meter.Counter(semconv.StoreRequestTotal, "name", options.Name, "status", "failure").Inc()
|
||||
r.opts.Meter.Counter(semconv.StoreRequestTotal, append(labels, "status", "failure")...).Inc()
|
||||
return err
|
||||
}
|
||||
|
||||
@ -484,6 +493,8 @@ 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 {
|
||||
@ -512,22 +523,22 @@ func (r *Store) Write(ctx context.Context, key string, val interface{}, opts ...
|
||||
}
|
||||
}
|
||||
|
||||
r.opts.Meter.Counter(semconv.StoreRequestInflight, "name", options.Name).Inc()
|
||||
r.opts.Meter.Counter(semconv.StoreRequestInflight, labels...).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, "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()
|
||||
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()
|
||||
return store.ErrNotFound
|
||||
} else if err == nil {
|
||||
r.opts.Meter.Counter(semconv.StoreRequestTotal, "name", options.Name, "status", "hit").Inc()
|
||||
r.opts.Meter.Counter(semconv.StoreRequestTotal, append(labels, "status", "hit")...).Inc()
|
||||
} else if err != nil {
|
||||
r.opts.Meter.Counter(semconv.StoreRequestTotal, "name", options.Name, "status", "failure").Inc()
|
||||
r.opts.Meter.Counter(semconv.StoreRequestTotal, append(labels, "status", "failure")...).Inc()
|
||||
return err
|
||||
}
|
||||
|
||||
|
@ -12,6 +12,26 @@ 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()
|
||||
|
||||
|
@ -73,7 +73,7 @@ func (h *tracingHook) ProcessHook(hook goredis.ProcessHook) goredis.ProcessHook
|
||||
case "cluster slots":
|
||||
break
|
||||
default:
|
||||
_, span := h.tr.Start(ctx, "goredis.process", append(h.opts, tracer.WithSpanLabels("db.statement", cmdString))...)
|
||||
_, span := h.tr.Start(ctx, "sdk.database", 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.goredis.num_cmd", strconv.Itoa(len(cmds)),
|
||||
"db.database.num_cmd", strconv.Itoa(len(cmds)),
|
||||
"db.statement", cmdsString,
|
||||
))
|
||||
|
||||
_, span := h.tr.Start(ctx, "goredis.process_pipeline", opts...)
|
||||
_, span := h.tr.Start(ctx, "sdk.database", opts...)
|
||||
defer span.Finish()
|
||||
|
||||
err := hook(ctx, cmds)
|
||||
|
Loading…
Reference in New Issue
Block a user