unistack-org/micro-store-redis
4
0
Fork 2
Code Issues Pull Requests Actions Packages Projects Releases 36 Wiki Activity

add using lazy connect #122

Merged
vtolstov merged 4 commits from devstigneev/micro-store-redis:lazy_conn into v3 2024-11-27 01:57:41 +03:00
Conversation 0 Commits 4 Files Changed 5 +139 -58
5 changed files with 139 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

55
event.go Normal file
View File

@ -0,0 +1,55 @@
package redis
import (
"context"
"errors"
"net"
"sync/atomic"
goredis "github.com/redis/go-redis/v9"
)
type eventHook struct {
connected *atomic.Bool
}
var _ goredis.Hook = (*eventHook)(nil)
func newEventHook(connected *atomic.Bool) *eventHook {
return &eventHook{connected: connected}
}
func (h *eventHook) DialHook(hook goredis.DialHook) goredis.DialHook {
return func(ctx context.Context, network, addr string) (net.Conn, error) {
conn, err := hook(ctx, network, addr)
if err != nil && !isRedisError(err) {
h.connected.Store(false)
}
return conn, err
}
}
func (h *eventHook) ProcessHook(hook goredis.ProcessHook) goredis.ProcessHook {
return func(ctx context.Context, cmd goredis.Cmder) error {
err := hook(ctx, cmd)
if err != nil && !isRedisError(err) {
h.connected.Store(false)
}
return err
}
}
func (h *eventHook) ProcessPipelineHook(hook goredis.ProcessPipelineHook) goredis.ProcessPipelineHook {
return func(ctx context.Context, cmds []goredis.Cmder) error {
err := hook(ctx, cmds)
if err != nil && !isRedisError(err) {
h.connected.Store(false)
}
return err
}
}
func isRedisError(err error) bool {
var rerr goredis.Error
return errors.As(err, &rerr)
}

8
go.mod
View File

@ -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.7.0
github.com/redis/go-redis/v9 v9.7.0
go.unistack.org/micro/v3 v3.10.106
)
require (
@ -15,5 +15,5 @@ require (
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
github.com/google/go-cmp v0.6.0 // indirect
go.unistack.org/micro-proto/v3 v3.4.1 // indirect
google.golang.org/protobuf v1.35.1 // indirect
google.golang.org/protobuf v1.35.2 // indirect
)

16
go.sum
View File

@ -8,13 +8,13 @@ github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/r
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.7.0 h1:BIx9TNZH/Jsr4l1i7VVxnV0JPiwYj8qyrHyuL0fGZrk=
github.com/redis/go-redis/extra/rediscmd/v9 v9.7.0/go.mod h1:eTg/YQtGYAZD5r3DlGlJptJ45AHA+/G+2NPn30PKzik=
github.com/redis/go-redis/v9 v9.7.0 h1:HhLSs+B6O021gwzl+locl0zEDnyNkxMtf/Z3NNBMa9E=
github.com/redis/go-redis/v9 v9.7.0/go.mod h1:f6zhXITC7JUJIlPEiBOTXxJgPLdZcA93GewI7inzyWw=
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=
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=
go.unistack.org/micro/v3 v3.10.106 h1:ya4+n58l4PImtrIKrJi1GgkUuJ1gmzLYa9WKYI1JFLs=
go.unistack.org/micro/v3 v3.10.106/go.mod h1:YzMldzHN9Ei+zy5t/Psu7RUWDZwUfrNYiStSQtTz90g=
google.golang.org/protobuf v1.35.2 h1:8Ar7bF+apOIoThw1EdZl0p1oWvMqTHmpA2fRTyZO8io=
google.golang.org/protobuf v1.35.2/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=

97
redis.go
View File

@ -5,6 +5,7 @@ import (
"errors"
"reflect"
"strings"
"sync/atomic"
"time"
goredis "github.com/redis/go-redis/v9"
@ -55,20 +56,29 @@ var (
)
type Store struct {
opts store.Options
cli goredis.UniversalClient
done chan struct{}
pool *pool.StringsPool
connected *atomic.Bool
opts store.Options
}
func (r *Store) Connect(ctx context.Context) error {
if r.connected.Load() {
return nil
}
if r.cli == nil {
return store.ErrNotConnected
}
err := r.cli.Ping(ctx).Err()
if r.opts.LazyConnect {
return nil
}
if err := r.cli.Ping(ctx).Err(); err != nil {
setSpanError(ctx, err)
return err
}
r.connected.Store(true)
return nil
}
func (r *Store) Init(opts ...store.Option) error {
for _, o := range opts {
@ -102,19 +112,20 @@ func (r *Store) ClusterClient() *goredis.ClusterClient {
}
func (r *Store) Disconnect(ctx context.Context) error {
var err error
select {
case <-r.done:
return err
default:
if !r.connected.Load() {
return nil
}
if r.cli != nil {
err = r.cli.Close()
}
close(r.done)
if err := r.cli.Close(); err != nil {
return err
}
}
r.connected.Store(false)
return nil
}
func (r *Store) Exists(ctx context.Context, key string, opts ...store.ExistsOption) error {
b := r.pool.Get()
defer r.pool.Put(b)
@ -149,7 +160,7 @@ func (r *Store) Exists(ctx context.Context, key string, opts ...store.ExistsOpti
return store.ErrNotFound
} else if err == nil {
r.opts.Meter.Counter(semconv.StoreRequestTotal, append(labels, "status", "hit")...).Inc()
} else if err != nil {
} else {
r.opts.Meter.Counter(semconv.StoreRequestTotal, append(labels, "status", "failure")...).Inc()
return err
}
@ -311,6 +322,8 @@ func (r *Store) MRead(ctx context.Context, keys []string, vals interface{}, opts
func (r *Store) MDelete(ctx context.Context, keys []string, opts ...store.DeleteOption) error {
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 {
@ -335,7 +348,7 @@ func (r *Store) MDelete(ctx context.Context, keys []string, opts ...store.Delete
}
}
r.opts.Meter.Counter(semconv.StoreRequestInflight, "name", options.Name).Inc()
r.opts.Meter.Counter(semconv.StoreRequestInflight, labels...).Inc()
ts := time.Now()
var err error
if r.opts.Namespace != "" || options.Namespace != "" {
@ -348,16 +361,16 @@ func (r *Store) MDelete(ctx context.Context, keys []string, opts ...store.Delete
}
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())
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 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
} 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
}
@ -407,6 +420,8 @@ func (r *Store) Delete(ctx context.Context, key string, opts ...store.DeleteOpti
func (r *Store) MWrite(ctx context.Context, keys []string, vals []interface{}, opts ...store.WriteOption) error {
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 {
@ -440,7 +455,7 @@ func (r *Store) MWrite(ctx context.Context, keys []string, vals []interface{}, o
}
}
r.opts.Meter.Counter(semconv.StoreRequestInflight, "name", options.Name).Inc()
r.opts.Meter.Counter(semconv.StoreRequestInflight, labels...).Inc()
pipeliner := func(pipe goredis.Pipeliner) error {
for idx := 0; idx < len(kvs); idx += 2 {
@ -460,27 +475,27 @@ func (r *Store) MWrite(ctx context.Context, keys []string, vals []interface{}, o
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())
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 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
} 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
}
for _, cmd := range cmds {
if err = cmd.Err(); err != nil {
if 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
}
setSpanError(ctx, err)
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
}
}
@ -537,7 +552,7 @@ func (r *Store) Write(ctx context.Context, key string, val interface{}, opts ...
return store.ErrNotFound
} else if err == nil {
r.opts.Meter.Counter(semconv.StoreRequestTotal, append(labels, "status", "hit")...).Inc()
} else if err != nil {
} else {
r.opts.Meter.Counter(semconv.StoreRequestTotal, append(labels, "status", "failure")...).Inc()
return err
}
@ -550,6 +565,9 @@ func (r *Store) List(ctx context.Context, opts ...store.ListOption) ([]string, e
defer r.pool.Put(b)
options := store.NewListOptions(opts...)
labels := make([]string, 0, 6)
labels = append(labels, "name", options.Name, "statement", "list")
if len(options.Namespace) == 0 {
options.Namespace = r.opts.Namespace
}
@ -571,7 +589,7 @@ func (r *Store) List(ctx context.Context, opts ...store.ListOption) ([]string, e
}
// TODO: add support for prefix/suffix/limit
r.opts.Meter.Counter(semconv.StoreRequestInflight, "name", options.Name).Inc()
r.opts.Meter.Counter(semconv.StoreRequestInflight, labels...).Inc()
ts := time.Now()
var keys []string
var err error
@ -591,16 +609,16 @@ func (r *Store) List(ctx context.Context, opts ...store.ListOption) ([]string, e
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())
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 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 nil, 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", "git")...).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 nil, err
}
@ -632,7 +650,11 @@ func (r *Store) String() string {
}
func NewStore(opts ...store.Option) *Store {
return &Store{done: make(chan struct{}), opts: store.NewOptions(opts...)}
b := atomic.Bool{}
return &Store{
opts: store.NewOptions(opts...),
connected: &b,
}
}
func (r *Store) configure() error {
@ -723,6 +745,7 @@ func (r *Store) configure() error {
r.cli = goredis.NewUniversalClient(universalOptions)
setTracing(r.cli, r.opts.Tracer)
r.cli.AddHook(newEventHook(r.connected))
r.pool = pool.NewStringsPool(50)

5
redis_test.go
View File

@ -4,6 +4,7 @@ import (
"bytes"
"context"
"os"
"sync/atomic"
"testing"
"time"
@ -40,7 +41,7 @@ func TestKeepTTL(t *testing.T) {
}
r := NewStore(store.Addrs(os.Getenv("STORE_NODES")))
if err := r.Init(); err != nil {
if err := r.Init(store.LazyConnect(true)); err != nil {
t.Fatal(err)
}
if err := r.Connect(ctx); err != nil {
@ -119,9 +120,11 @@ func Test_rkv_configure(t *testing.T) {
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
b := atomic.Bool{}
rc := &Store{
opts: tt.fields.options,
cli: tt.fields.Client,
connected: &b,
}
err := rc.configure()
if (err != nil) != tt.wantErr {