Compare commits
5 Commits
Author | SHA1 | Date | |
---|---|---|---|
256e61a437 | |||
f9cdd41c94 | |||
ecad15fe17 | |||
fa3d18b353 | |||
2f3951773f |
1
.gitignore
vendored
Normal file
1
.gitignore
vendored
Normal file
@@ -0,0 +1 @@
|
||||
.idea
|
@@ -27,6 +27,9 @@ var (
|
||||
DefaultMeterStatsInterval = 5 * time.Second
|
||||
// DefaultMeterMetricPrefix holds default metric prefix
|
||||
DefaultMeterMetricPrefix = "micro_store_"
|
||||
|
||||
labelHost = "redis_host"
|
||||
labelName = "redis_name"
|
||||
)
|
||||
|
||||
// Options struct holds wrapper options
|
||||
@@ -36,6 +39,8 @@ type Options struct {
|
||||
Tracer tracer.Tracer
|
||||
MeterMetricPrefix string
|
||||
MeterStatsInterval time.Duration
|
||||
RedisHost string
|
||||
RedisName string
|
||||
}
|
||||
|
||||
// Option func signature
|
||||
@@ -56,7 +61,9 @@ func NewOptions(opts ...Option) Options {
|
||||
}
|
||||
|
||||
options.Meter = options.Meter.Clone(
|
||||
meter.MetricPrefix(options.MeterMetricPrefix),
|
||||
meter.Labels(
|
||||
labelHost, options.RedisHost,
|
||||
labelName, options.RedisName),
|
||||
)
|
||||
|
||||
options.Logger = options.Logger.Clone(logger.WithCallerSkipCount(1))
|
||||
|
36
redis.go
36
redis.go
@@ -218,9 +218,11 @@ func (r *Store) MRead(ctx context.Context, keys []string, vals interface{}, opts
|
||||
defer cancel()
|
||||
}
|
||||
|
||||
var rkeys []string
|
||||
if r.opts.Namespace != "" || options.Namespace != "" {
|
||||
rkeys = make([]string, len(keys))
|
||||
for idx, key := range keys {
|
||||
keys[idx] = r.getKey(r.opts.Namespace, options.Namespace, key)
|
||||
rkeys[idx] = r.getKey(r.opts.Namespace, options.Namespace, key)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -228,10 +230,18 @@ func (r *Store) MRead(ctx context.Context, keys []string, vals interface{}, opts
|
||||
ts := time.Now()
|
||||
var rvals []interface{}
|
||||
var err error
|
||||
if r.cli.Client != nil {
|
||||
rvals, err = r.cli.Client.MGet(ctx, keys...).Result()
|
||||
if r.opts.Namespace != "" || options.Namespace != "" {
|
||||
if r.cli.Client != nil {
|
||||
rvals, err = r.cli.Client.MGet(ctx, rkeys...).Result()
|
||||
} else {
|
||||
rvals, err = r.cli.ClusterClient.MGet(ctx, rkeys...).Result()
|
||||
}
|
||||
} else {
|
||||
rvals, err = r.cli.ClusterClient.MGet(ctx, keys...).Result()
|
||||
if r.cli.Client != nil {
|
||||
rvals, err = r.cli.Client.MGet(ctx, keys...).Result()
|
||||
} else {
|
||||
rvals, err = r.cli.ClusterClient.MGet(ctx, keys...).Result()
|
||||
}
|
||||
}
|
||||
setSpanError(ctx, err)
|
||||
te := time.Since(ts)
|
||||
@@ -308,19 +318,29 @@ func (r *Store) MDelete(ctx context.Context, keys []string, opts ...store.Delete
|
||||
defer cancel()
|
||||
}
|
||||
|
||||
var rkeys []string
|
||||
if r.opts.Namespace != "" || options.Namespace != "" {
|
||||
rkeys = make([]string, len(keys))
|
||||
for idx, key := range keys {
|
||||
keys[idx] = r.getKey(r.opts.Namespace, options.Namespace, key)
|
||||
rkeys[idx] = r.getKey(r.opts.Namespace, options.Namespace, key)
|
||||
}
|
||||
}
|
||||
|
||||
r.opts.Meter.Counter(semconv.StoreRequestInflight, "name", options.Name).Inc()
|
||||
ts := time.Now()
|
||||
var err error
|
||||
if r.cli.Client != nil {
|
||||
err = r.cli.Client.Del(ctx, keys...).Err()
|
||||
if r.opts.Namespace != "" || options.Namespace != "" {
|
||||
if r.cli.Client != nil {
|
||||
err = r.cli.Client.Del(ctx, rkeys...).Err()
|
||||
} else {
|
||||
err = r.cli.ClusterClient.Del(ctx, rkeys...).Err()
|
||||
}
|
||||
} else {
|
||||
err = r.cli.ClusterClient.Del(ctx, keys...).Err()
|
||||
if r.cli.Client != nil {
|
||||
err = r.cli.Client.Del(ctx, keys...).Err()
|
||||
} else {
|
||||
err = r.cli.ClusterClient.Del(ctx, keys...).Err()
|
||||
}
|
||||
}
|
||||
setSpanError(ctx, err)
|
||||
te := time.Since(ts)
|
||||
|
13
tracer.go
13
tracer.go
@@ -6,7 +6,7 @@ import (
|
||||
"net"
|
||||
"strconv"
|
||||
|
||||
"github.com/redis/go-redis/extra/rediscmd/v9"
|
||||
rediscmd "github.com/redis/go-redis/extra/rediscmd/v9"
|
||||
"github.com/redis/go-redis/v9"
|
||||
"go.unistack.org/micro/v3/tracer"
|
||||
)
|
||||
@@ -18,14 +18,12 @@ func setTracing(rdb redis.UniversalClient, tr tracer.Tracer, opts ...tracer.Span
|
||||
connString := formatDBConnString(opt.Network, opt.Addr)
|
||||
rdb.AddHook(newTracingHook(connString, tr))
|
||||
case *redis.ClusterClient:
|
||||
rdb.AddHook(newTracingHook("", tr, opts...))
|
||||
rdb.OnNewNode(func(rdb *redis.Client) {
|
||||
opt := rdb.Options()
|
||||
connString := formatDBConnString(opt.Network, opt.Addr)
|
||||
rdb.AddHook(newTracingHook(connString, tr))
|
||||
})
|
||||
case *redis.Ring:
|
||||
rdb.AddHook(newTracingHook("", tr, opts...))
|
||||
rdb.OnNewNode(func(rdb *redis.Client) {
|
||||
opt := rdb.Options()
|
||||
connString := formatDBConnString(opt.Network, opt.Addr)
|
||||
@@ -55,11 +53,12 @@ func newTracingHook(connString string, tr tracer.Tracer, opts ...tracer.SpanOpti
|
||||
|
||||
func (h *tracingHook) DialHook(hook redis.DialHook) redis.DialHook {
|
||||
return func(ctx context.Context, network, addr string) (net.Conn, error) {
|
||||
_, span := h.tr.Start(ctx, "redis.dial", h.opts...)
|
||||
defer span.Finish()
|
||||
|
||||
/*
|
||||
_, span := h.tr.Start(ctx, "redis.dial", h.opts...)
|
||||
defer span.Finish()
|
||||
*/
|
||||
conn, err := hook(ctx, network, addr)
|
||||
recordError(span, err)
|
||||
// recordError(span, err)
|
||||
|
||||
return conn, err
|
||||
}
|
||||
|
Reference in New Issue
Block a user