序
本文次要钻研一下 cortex 的 kv.Client
kv.Client
github.com/cortexproject/cortex/pkg/ring/kv/client.go
// Client is a high-level client for key-value stores (such as Etcd and
// Consul) that exposes operations such as CAS and Watch which take callbacks.
// It also deals with serialisation by using a Codec and having a instance of
// the the desired type passed in to methods ala json.Unmarshal.
type Client interface {
// List returns a list of keys under the given prefix. Returned keys will
// include the prefix.
List(ctx context.Context, prefix string) ([]string, error)
// Get a specific key. Will use a codec to deserialise key to appropriate type.
// If the key does not exist, Get will return nil and no error.
Get(ctx context.Context, key string) (interface{}, error)
// Delete a specific key. Deletions are best-effort and no error will
// be returned if the key does not exist.
Delete(ctx context.Context, key string) error
// CAS stands for Compare-And-Swap. Will call provided callback f with the
// current value of the key and allow callback to return a different value.
// Will then attempt to atomically swap the current value for the new value.
// If that doesn't succeed will try again - callback will be called again
// with new value etc. Guarantees that only a single concurrent CAS
// succeeds. Callback can return nil to indicate it is happy with existing
// value.
CAS(ctx context.Context, key string, f func(in interface{}) (out interface{}, retry bool, err error)) error
// WatchKey calls f whenever the value stored under key changes.
WatchKey(ctx context.Context, key string, f func(interface{}) bool)
// WatchPrefix calls f whenever any value stored under prefix changes.
WatchPrefix(ctx context.Context, prefix string, f func(string, interface{}) bool)
}
kv.Client 接口定义了 List、Get、Delete、CAS、WatchKey、WatchPrefix 办法
Client
github.com/cortexproject/cortex/pkg/ring/kv/memberlist/memberlist_client.go
// Client implements kv.Client interface, by using memberlist.KV
type Client struct {
kv *KV // reference to singleton memberlist-based KV
codec codec.Codec
}
// List is part of kv.Client interface.
func (c *Client) List(ctx context.Context, prefix string) ([]string, error) {err := c.awaitKVRunningOrStopping(ctx)
if err != nil {return nil, err}
return c.kv.List(prefix), nil
}
// Get is part of kv.Client interface.
func (c *Client) Get(ctx context.Context, key string) (interface{}, error) {err := c.awaitKVRunningOrStopping(ctx)
if err != nil {return nil, err}
return c.kv.Get(key, c.codec)
}
// Delete is part of kv.Client interface.
func (c *Client) Delete(ctx context.Context, key string) error {return errors.New("memberlist does not support Delete")
}
// CAS is part of kv.Client interface
func (c *Client) CAS(ctx context.Context, key string, f func(in interface{}) (out interface{}, retry bool, err error)) error {err := c.awaitKVRunningOrStopping(ctx)
if err != nil {return err}
return c.kv.CAS(ctx, key, c.codec, f)
}
// WatchKey is part of kv.Client interface.
func (c *Client) WatchKey(ctx context.Context, key string, f func(interface{}) bool) {err := c.awaitKVRunningOrStopping(ctx)
if err != nil {return}
c.kv.WatchKey(ctx, key, c.codec, f)
}
// WatchPrefix calls f whenever any value stored under prefix changes.
// Part of kv.Client interface.
func (c *Client) WatchPrefix(ctx context.Context, prefix string, f func(string, interface{}) bool) {err := c.awaitKVRunningOrStopping(ctx)
if err != nil {return}
c.kv.WatchPrefix(ctx, prefix, c.codec, f)
}
Client 实现了 kv.Client 接口,其 List、Get、CAS、WatchKey、WatchPrefix 办法均代理给 kv,其 Delete 办法返回 error
KV.List
github.com/cortexproject/cortex/pkg/ring/kv/memberlist/memberlist_client.go
// List returns all known keys under a given prefix.
// No communication with other nodes in the cluster is done here.
func (m *KV) List(prefix string) []string {m.storeMu.Lock()
defer m.storeMu.Unlock()
var keys []string
for k := range m.store {if strings.HasPrefix(k, prefix) {keys = append(keys, k)
}
}
return keys
}
KV.List 办法遍历 m.store,查找是否有指定 prefix 的 key
KV.Get
github.com/cortexproject/cortex/pkg/ring/kv/memberlist/memberlist_client.go
// Get returns current value associated with given key.
// No communication with other nodes in the cluster is done here.
func (m *KV) Get(key string, codec codec.Codec) (interface{}, error) {val, _, err := m.get(key, codec)
return val, err
}
// Returns current value with removed tombstones.
func (m *KV) get(key string, codec codec.Codec) (out interface{}, version uint, err error) {m.storeMu.Lock()
v := m.store[key]
m.storeMu.Unlock()
out = nil
if v.value != nil {out, err = codec.Decode(v.value)
if err != nil {return nil, 0, err}
if mr, ok := out.(Mergeable); ok {
// remove ALL tombstones before returning to client.
// No need for clients to see them.
mr.RemoveTombstones(time.Time{})
}
}
return out, v.version, nil
}
KV.Get 办法次要是从 m.store[key] 获取数据
KV.CAS
github.com/cortexproject/cortex/pkg/ring/kv/memberlist/memberlist_client.go
func (m *KV) CAS(ctx context.Context, key string, codec codec.Codec, f func(in interface{}) (out interface{}, retry bool, err error)) error {
var lastError error = nil
outer:
for retries := m.maxCasRetries; retries > 0; retries-- {m.casAttempts.Inc()
if lastError == errNoChangeDetected {
// We only get here, if 'f' reports some change, but Merge function reports no change. This can happen
// with Ring's merge function, which depends on timestamps (and not the tokens) with 1-second resolution.
// By waiting for one second, we hope that Merge will be able to detect change from 'f' function.
select {case <-time.After(noChangeDetectedRetrySleep):
// ok
case <-ctx.Done():
lastError = ctx.Err()
break outer
}
}
change, newver, retry, err := m.trySingleCas(key, codec, f)
if err != nil {level.Debug(m.logger).Log("msg", "CAS attempt failed", "err", err, "retry", retry)
lastError = err
if !retry {break}
continue
}
if change != nil {m.casSuccesses.Inc()
m.notifyWatchers(key)
if m.State() == services.Running {m.broadcastNewValue(key, change, newver, codec)
} else {level.Warn(m.logger).Log("msg", "skipped broadcasting CAS update because memberlist KV is shutting down", "key", key)
}
}
return nil
}
if lastError == errVersionMismatch {
// this is more likely error than version mismatch.
lastError = errTooManyRetries
}
m.casFailures.Inc()
return fmt.Errorf("failed to CAS-update key %s: %v", key, lastError)
}
KV.CAS 通过 for 循环 m.maxCasRetries 执行 m.trySingleCas 操作
KV.WatchKey
github.com/cortexproject/cortex/pkg/ring/kv/memberlist/memberlist_client.go
func (m *KV) WatchKey(ctx context.Context, key string, codec codec.Codec, f func(interface{}) bool) {
// keep one extra notification, to avoid missing notification if we're busy running the function
w := make(chan string, 1)
// register watcher
m.watchersMu.Lock()
m.watchers[key] = append(m.watchers[key], w)
m.watchersMu.Unlock()
defer func() {
// unregister watcher on exit
m.watchersMu.Lock()
defer m.watchersMu.Unlock()
removeWatcherChannel(key, w, m.watchers)
}()
for {
select {
case <-w:
// value changed
val, _, err := m.get(key, codec)
if err != nil {level.Warn(m.logger).Log("msg", "failed to decode value while watching for changes", "key", key, "err", err)
continue
}
if !f(val) {return}
case <-m.shutdown:
// stop watching on shutdown
return
case <-ctx.Done():
return
}
}
}
KV.WatchKey 办法会往 m.watchers[key] 追加 channel,而后 for 循环 select 期待 channel 的写入
KV.WatchPrefix
github.com/cortexproject/cortex/pkg/ring/kv/memberlist/memberlist_client.go
func (m *KV) WatchPrefix(ctx context.Context, prefix string, codec codec.Codec, f func(string, interface{}) bool) {
// we use bigger buffer here, since keys are interesting and we don't want to lose them.
w := make(chan string, 16)
// register watcher
m.watchersMu.Lock()
m.prefixWatchers[prefix] = append(m.prefixWatchers[prefix], w)
m.watchersMu.Unlock()
defer func() {
// unregister watcher on exit
m.watchersMu.Lock()
defer m.watchersMu.Unlock()
removeWatcherChannel(prefix, w, m.prefixWatchers)
}()
for {
select {
case key := <-w:
val, _, err := m.get(key, codec)
if err != nil {level.Warn(m.logger).Log("msg", "failed to decode value while watching for changes", "key", key, "err", err)
continue
}
if !f(key, val) {return}
case <-m.shutdown:
// stop watching on shutdown
return
case <-ctx.Done():
return
}
}
}
KV.WatchPrefix 办法与 WatchKey 类型,不过它 channel 的长度为 16,追加到的是 m.prefixWatchers[prefix]
小结
cortex 的 kv.Client 接口定义了 List、Get、Delete、CAS、WatchKey、WatchPrefix 办法;Client 实现了 kv.Client 接口,其 List、Get、CAS、WatchKey、WatchPrefix 办法均代理给 kv,其 Delete 办法返回 error。
doc
- cortex