1 Star 0 Fork 2

who7708/etcd

加入 Gitee
与超过 1200万 开发者一起发现、参与优秀开源项目,私有仓库也完全免费 :)
免费加入
克隆/下载
store.go 16.79 KB
一键复制 编辑 原始数据 按行查看 历史
Gyu-Ho Lee 提交于 2017-10-03 02:14 . *: fix 'ineffassign' issues
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620
// Copyright 2017 The etcd Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package v2v3
import (
"context"
"fmt"
"path"
"strings"
"time"
"github.com/coreos/etcd/clientv3"
"github.com/coreos/etcd/clientv3/concurrency"
etcdErr "github.com/coreos/etcd/error"
"github.com/coreos/etcd/mvcc/mvccpb"
"github.com/coreos/etcd/store"
)
// store implements the Store interface for V2 using
// a v3 client.
type v2v3Store struct {
c *clientv3.Client
// pfx is the v3 prefix where keys should be stored.
pfx string
ctx context.Context
}
const maxPathDepth = 63
var errUnsupported = fmt.Errorf("TTLs are unsupported")
func NewStore(c *clientv3.Client, pfx string) store.Store { return newStore(c, pfx) }
func newStore(c *clientv3.Client, pfx string) *v2v3Store { return &v2v3Store{c, pfx, c.Ctx()} }
func (s *v2v3Store) Index() uint64 { panic("STUB") }
func (s *v2v3Store) Get(nodePath string, recursive, sorted bool) (*store.Event, error) {
key := s.mkPath(nodePath)
resp, err := s.c.Txn(s.ctx).Then(
clientv3.OpGet(key+"/"),
clientv3.OpGet(key),
).Commit()
if err != nil {
return nil, err
}
if kvs := resp.Responses[0].GetResponseRange().Kvs; len(kvs) != 0 || isRoot(nodePath) {
nodes, err := s.getDir(nodePath, recursive, sorted, resp.Header.Revision)
if err != nil {
return nil, err
}
cidx, midx := uint64(0), uint64(0)
if len(kvs) > 0 {
cidx, midx = mkV2Rev(kvs[0].CreateRevision), mkV2Rev(kvs[0].ModRevision)
}
return &store.Event{
Action: store.Get,
Node: &store.NodeExtern{
Key: nodePath,
Dir: true,
Nodes: nodes,
CreatedIndex: cidx,
ModifiedIndex: midx,
},
EtcdIndex: mkV2Rev(resp.Header.Revision),
}, nil
}
kvs := resp.Responses[1].GetResponseRange().Kvs
if len(kvs) == 0 {
return nil, etcdErr.NewError(etcdErr.EcodeKeyNotFound, nodePath, mkV2Rev(resp.Header.Revision))
}
return &store.Event{
Action: store.Get,
Node: s.mkV2Node(kvs[0]),
EtcdIndex: mkV2Rev(resp.Header.Revision),
}, nil
}
func (s *v2v3Store) getDir(nodePath string, recursive, sorted bool, rev int64) ([]*store.NodeExtern, error) {
rootNodes, err := s.getDirDepth(nodePath, 1, rev)
if err != nil || !recursive {
return rootNodes, err
}
nextNodes := rootNodes
nodes := make(map[string]*store.NodeExtern)
// Breadth walk the subdirectories
for i := 2; len(nextNodes) > 0; i++ {
for _, n := range nextNodes {
nodes[n.Key] = n
if parent := nodes[path.Dir(n.Key)]; parent != nil {
parent.Nodes = append(parent.Nodes, n)
}
}
if nextNodes, err = s.getDirDepth(nodePath, i, rev); err != nil {
return nil, err
}
}
return rootNodes, nil
}
func (s *v2v3Store) getDirDepth(nodePath string, depth int, rev int64) ([]*store.NodeExtern, error) {
pd := s.mkPathDepth(nodePath, depth)
resp, err := s.c.Get(s.ctx, pd, clientv3.WithPrefix(), clientv3.WithRev(rev))
if err != nil {
return nil, err
}
nodes := make([]*store.NodeExtern, len(resp.Kvs))
for i, kv := range resp.Kvs {
nodes[i] = s.mkV2Node(kv)
}
return nodes, nil
}
func (s *v2v3Store) Set(
nodePath string,
dir bool,
value string,
expireOpts store.TTLOptionSet,
) (*store.Event, error) {
if expireOpts.Refresh || !expireOpts.ExpireTime.IsZero() {
return nil, errUnsupported
}
if isRoot(nodePath) {
return nil, etcdErr.NewError(etcdErr.EcodeRootROnly, nodePath, 0)
}
ecode := 0
applyf := func(stm concurrency.STM) error {
parent := path.Dir(nodePath)
if !isRoot(parent) && stm.Rev(s.mkPath(parent)+"/") == 0 {
ecode = etcdErr.EcodeKeyNotFound
return nil
}
key := s.mkPath(nodePath)
if dir {
if stm.Rev(key) != 0 {
// exists as non-dir
ecode = etcdErr.EcodeNotDir
return nil
}
key = key + "/"
} else if stm.Rev(key+"/") != 0 {
ecode = etcdErr.EcodeNotFile
return nil
}
stm.Put(key, value, clientv3.WithPrevKV())
stm.Put(s.mkActionKey(), store.Set)
return nil
}
resp, err := s.newSTM(applyf)
if err != nil {
return nil, err
}
if ecode != 0 {
return nil, etcdErr.NewError(ecode, nodePath, mkV2Rev(resp.Header.Revision))
}
createRev := resp.Header.Revision
var pn *store.NodeExtern
if pkv := prevKeyFromPuts(resp); pkv != nil {
pn = s.mkV2Node(pkv)
createRev = pkv.CreateRevision
}
vp := &value
if dir {
vp = nil
}
return &store.Event{
Action: store.Set,
Node: &store.NodeExtern{
Key: nodePath,
Value: vp,
Dir: dir,
ModifiedIndex: mkV2Rev(resp.Header.Revision),
CreatedIndex: mkV2Rev(createRev),
},
PrevNode: pn,
EtcdIndex: mkV2Rev(resp.Header.Revision),
}, nil
}
func (s *v2v3Store) Update(nodePath, newValue string, expireOpts store.TTLOptionSet) (*store.Event, error) {
if isRoot(nodePath) {
return nil, etcdErr.NewError(etcdErr.EcodeRootROnly, nodePath, 0)
}
if expireOpts.Refresh || !expireOpts.ExpireTime.IsZero() {
return nil, errUnsupported
}
key := s.mkPath(nodePath)
ecode := 0
applyf := func(stm concurrency.STM) error {
if rev := stm.Rev(key + "/"); rev != 0 {
ecode = etcdErr.EcodeNotFile
return nil
}
if rev := stm.Rev(key); rev == 0 {
ecode = etcdErr.EcodeKeyNotFound
return nil
}
stm.Put(key, newValue, clientv3.WithPrevKV())
stm.Put(s.mkActionKey(), store.Update)
return nil
}
resp, err := s.newSTM(applyf)
if err != nil {
return nil, err
}
if ecode != 0 {
return nil, etcdErr.NewError(etcdErr.EcodeNotFile, nodePath, mkV2Rev(resp.Header.Revision))
}
pkv := prevKeyFromPuts(resp)
return &store.Event{
Action: store.Update,
Node: &store.NodeExtern{
Key: nodePath,
Value: &newValue,
ModifiedIndex: mkV2Rev(resp.Header.Revision),
CreatedIndex: mkV2Rev(pkv.CreateRevision),
},
PrevNode: s.mkV2Node(pkv),
EtcdIndex: mkV2Rev(resp.Header.Revision),
}, nil
}
func (s *v2v3Store) Create(
nodePath string,
dir bool,
value string,
unique bool,
expireOpts store.TTLOptionSet,
) (*store.Event, error) {
if isRoot(nodePath) {
return nil, etcdErr.NewError(etcdErr.EcodeRootROnly, nodePath, 0)
}
if expireOpts.Refresh || !expireOpts.ExpireTime.IsZero() {
return nil, errUnsupported
}
ecode := 0
applyf := func(stm concurrency.STM) error {
ecode = 0
key := s.mkPath(nodePath)
if unique {
// append unique item under the node path
for {
key = nodePath + "/" + fmt.Sprintf("%020s", time.Now())
key = path.Clean(path.Join("/", key))
key = s.mkPath(key)
if stm.Rev(key) == 0 {
break
}
}
}
if stm.Rev(key) > 0 || stm.Rev(key+"/") > 0 {
ecode = etcdErr.EcodeNodeExist
return nil
}
// build path if any directories in path do not exist
dirs := []string{}
for p := path.Dir(nodePath); !isRoot(p); p = path.Dir(p) {
pp := s.mkPath(p)
if stm.Rev(pp) > 0 {
ecode = etcdErr.EcodeNotDir
return nil
}
if stm.Rev(pp+"/") == 0 {
dirs = append(dirs, pp+"/")
}
}
for _, d := range dirs {
stm.Put(d, "")
}
if dir {
// directories marked with extra slash in key name
key += "/"
}
stm.Put(key, value)
stm.Put(s.mkActionKey(), store.Create)
return nil
}
resp, err := s.newSTM(applyf)
if err != nil {
return nil, err
}
if ecode != 0 {
return nil, etcdErr.NewError(ecode, nodePath, mkV2Rev(resp.Header.Revision))
}
var v *string
if !dir {
v = &value
}
return &store.Event{
Action: store.Create,
Node: &store.NodeExtern{
Key: nodePath,
Value: v,
Dir: dir,
ModifiedIndex: mkV2Rev(resp.Header.Revision),
CreatedIndex: mkV2Rev(resp.Header.Revision),
},
EtcdIndex: mkV2Rev(resp.Header.Revision),
}, nil
}
func (s *v2v3Store) CompareAndSwap(
nodePath string,
prevValue string,
prevIndex uint64,
value string,
expireOpts store.TTLOptionSet,
) (*store.Event, error) {
if isRoot(nodePath) {
return nil, etcdErr.NewError(etcdErr.EcodeRootROnly, nodePath, 0)
}
if expireOpts.Refresh || !expireOpts.ExpireTime.IsZero() {
return nil, errUnsupported
}
key := s.mkPath(nodePath)
resp, err := s.c.Txn(s.ctx).If(
s.mkCompare(nodePath, prevValue, prevIndex)...,
).Then(
clientv3.OpPut(key, value, clientv3.WithPrevKV()),
clientv3.OpPut(s.mkActionKey(), store.CompareAndSwap),
).Else(
clientv3.OpGet(key),
clientv3.OpGet(key+"/"),
).Commit()
if err != nil {
return nil, err
}
if !resp.Succeeded {
return nil, compareFail(nodePath, prevValue, prevIndex, resp)
}
pkv := resp.Responses[0].GetResponsePut().PrevKv
return &store.Event{
Action: store.CompareAndSwap,
Node: &store.NodeExtern{
Key: nodePath,
Value: &value,
CreatedIndex: mkV2Rev(pkv.CreateRevision),
ModifiedIndex: mkV2Rev(resp.Header.Revision),
},
PrevNode: s.mkV2Node(pkv),
EtcdIndex: mkV2Rev(resp.Header.Revision),
}, nil
}
func (s *v2v3Store) Delete(nodePath string, dir, recursive bool) (*store.Event, error) {
if isRoot(nodePath) {
return nil, etcdErr.NewError(etcdErr.EcodeRootROnly, nodePath, 0)
}
if !dir && !recursive {
return s.deleteNode(nodePath)
}
if !recursive {
return s.deleteEmptyDir(nodePath)
}
dels := make([]clientv3.Op, maxPathDepth+1)
dels[0] = clientv3.OpDelete(s.mkPath(nodePath)+"/", clientv3.WithPrevKV())
for i := 1; i < maxPathDepth; i++ {
dels[i] = clientv3.OpDelete(s.mkPathDepth(nodePath, i), clientv3.WithPrefix())
}
dels[maxPathDepth] = clientv3.OpPut(s.mkActionKey(), store.Delete)
resp, err := s.c.Txn(s.ctx).If(
clientv3.Compare(clientv3.Version(s.mkPath(nodePath)+"/"), ">", 0),
clientv3.Compare(clientv3.Version(s.mkPathDepth(nodePath, maxPathDepth)+"/"), "=", 0),
).Then(
dels...,
).Commit()
if err != nil {
return nil, err
}
if !resp.Succeeded {
return nil, etcdErr.NewError(etcdErr.EcodeNodeExist, nodePath, mkV2Rev(resp.Header.Revision))
}
dresp := resp.Responses[0].GetResponseDeleteRange()
return &store.Event{
Action: store.Delete,
PrevNode: s.mkV2Node(dresp.PrevKvs[0]),
EtcdIndex: mkV2Rev(resp.Header.Revision),
}, nil
}
func (s *v2v3Store) deleteEmptyDir(nodePath string) (*store.Event, error) {
resp, err := s.c.Txn(s.ctx).If(
clientv3.Compare(clientv3.Version(s.mkPathDepth(nodePath, 1)), "=", 0).WithPrefix(),
).Then(
clientv3.OpDelete(s.mkPath(nodePath)+"/", clientv3.WithPrevKV()),
clientv3.OpPut(s.mkActionKey(), store.Delete),
).Commit()
if err != nil {
return nil, err
}
if !resp.Succeeded {
return nil, etcdErr.NewError(etcdErr.EcodeDirNotEmpty, nodePath, mkV2Rev(resp.Header.Revision))
}
dresp := resp.Responses[0].GetResponseDeleteRange()
if len(dresp.PrevKvs) == 0 {
return nil, etcdErr.NewError(etcdErr.EcodeNodeExist, nodePath, mkV2Rev(resp.Header.Revision))
}
return &store.Event{
Action: store.Delete,
PrevNode: s.mkV2Node(dresp.PrevKvs[0]),
EtcdIndex: mkV2Rev(resp.Header.Revision),
}, nil
}
func (s *v2v3Store) deleteNode(nodePath string) (*store.Event, error) {
resp, err := s.c.Txn(s.ctx).If(
clientv3.Compare(clientv3.Version(s.mkPath(nodePath)+"/"), "=", 0),
).Then(
clientv3.OpDelete(s.mkPath(nodePath), clientv3.WithPrevKV()),
clientv3.OpPut(s.mkActionKey(), store.Delete),
).Commit()
if err != nil {
return nil, err
}
if !resp.Succeeded {
return nil, etcdErr.NewError(etcdErr.EcodeNotFile, nodePath, mkV2Rev(resp.Header.Revision))
}
pkvs := resp.Responses[0].GetResponseDeleteRange().PrevKvs
if len(pkvs) == 0 {
return nil, etcdErr.NewError(etcdErr.EcodeKeyNotFound, nodePath, mkV2Rev(resp.Header.Revision))
}
pkv := pkvs[0]
return &store.Event{
Action: store.Delete,
Node: &store.NodeExtern{
Key: nodePath,
CreatedIndex: mkV2Rev(pkv.CreateRevision),
ModifiedIndex: mkV2Rev(resp.Header.Revision),
},
PrevNode: s.mkV2Node(pkv),
EtcdIndex: mkV2Rev(resp.Header.Revision),
}, nil
}
func (s *v2v3Store) CompareAndDelete(nodePath, prevValue string, prevIndex uint64) (*store.Event, error) {
if isRoot(nodePath) {
return nil, etcdErr.NewError(etcdErr.EcodeRootROnly, nodePath, 0)
}
key := s.mkPath(nodePath)
resp, err := s.c.Txn(s.ctx).If(
s.mkCompare(nodePath, prevValue, prevIndex)...,
).Then(
clientv3.OpDelete(key, clientv3.WithPrevKV()),
clientv3.OpPut(s.mkActionKey(), store.CompareAndDelete),
).Else(
clientv3.OpGet(key),
clientv3.OpGet(key+"/"),
).Commit()
if err != nil {
return nil, err
}
if !resp.Succeeded {
return nil, compareFail(nodePath, prevValue, prevIndex, resp)
}
// len(pkvs) > 1 since txn only succeeds when key exists
pkv := resp.Responses[0].GetResponseDeleteRange().PrevKvs[0]
return &store.Event{
Action: store.CompareAndDelete,
Node: &store.NodeExtern{
Key: nodePath,
CreatedIndex: mkV2Rev(pkv.CreateRevision),
ModifiedIndex: mkV2Rev(resp.Header.Revision),
},
PrevNode: s.mkV2Node(pkv),
EtcdIndex: mkV2Rev(resp.Header.Revision),
}, nil
}
func compareFail(nodePath, prevValue string, prevIndex uint64, resp *clientv3.TxnResponse) error {
if dkvs := resp.Responses[1].GetResponseRange().Kvs; len(dkvs) > 0 {
return etcdErr.NewError(etcdErr.EcodeNotFile, nodePath, mkV2Rev(resp.Header.Revision))
}
kvs := resp.Responses[0].GetResponseRange().Kvs
if len(kvs) == 0 {
return etcdErr.NewError(etcdErr.EcodeKeyNotFound, nodePath, mkV2Rev(resp.Header.Revision))
}
kv := kvs[0]
indexMatch := (prevIndex == 0 || kv.ModRevision == int64(prevIndex))
valueMatch := (prevValue == "" || string(kv.Value) == prevValue)
var cause string
switch {
case indexMatch && !valueMatch:
cause = fmt.Sprintf("[%v != %v]", prevValue, string(kv.Value))
case valueMatch && !indexMatch:
cause = fmt.Sprintf("[%v != %v]", prevIndex, kv.ModRevision)
default:
cause = fmt.Sprintf("[%v != %v] [%v != %v]", prevValue, string(kv.Value), prevIndex, kv.ModRevision)
}
return etcdErr.NewError(etcdErr.EcodeTestFailed, cause, mkV2Rev(resp.Header.Revision))
}
func (s *v2v3Store) mkCompare(nodePath, prevValue string, prevIndex uint64) []clientv3.Cmp {
key := s.mkPath(nodePath)
cmps := []clientv3.Cmp{clientv3.Compare(clientv3.Version(key), ">", 0)}
if prevIndex != 0 {
cmps = append(cmps, clientv3.Compare(clientv3.ModRevision(key), "=", mkV3Rev(prevIndex)))
}
if prevValue != "" {
cmps = append(cmps, clientv3.Compare(clientv3.Value(key), "=", prevValue))
}
return cmps
}
func (s *v2v3Store) JsonStats() []byte { panic("STUB") }
func (s *v2v3Store) DeleteExpiredKeys(cutoff time.Time) { panic("STUB") }
func (s *v2v3Store) Version() int { return 2 }
// TODO: move this out of the Store interface?
func (s *v2v3Store) Save() ([]byte, error) { panic("STUB") }
func (s *v2v3Store) Recovery(state []byte) error { panic("STUB") }
func (s *v2v3Store) Clone() store.Store { panic("STUB") }
func (s *v2v3Store) SaveNoCopy() ([]byte, error) { panic("STUB") }
func (s *v2v3Store) HasTTLKeys() bool { panic("STUB") }
func (s *v2v3Store) mkPath(nodePath string) string { return s.mkPathDepth(nodePath, 0) }
func (s *v2v3Store) mkNodePath(p string) string {
return path.Clean(p[len(s.pfx)+len("/k/000/"):])
}
// mkPathDepth makes a path to a key that encodes its directory depth
// for fast directory listing. If a depth is provided, it is added
// to the computed depth.
func (s *v2v3Store) mkPathDepth(nodePath string, depth int) string {
normalForm := path.Clean(path.Join("/", nodePath))
n := strings.Count(normalForm, "/") + depth
return fmt.Sprintf("%s/%03d/k/%s", s.pfx, n, normalForm)
}
func (s *v2v3Store) mkActionKey() string { return s.pfx + "/act" }
func isRoot(s string) bool { return len(s) == 0 || s == "/" || s == "/0" || s == "/1" }
func mkV2Rev(v3Rev int64) uint64 {
if v3Rev == 0 {
return 0
}
return uint64(v3Rev - 1)
}
func mkV3Rev(v2Rev uint64) int64 {
if v2Rev == 0 {
return 0
}
return int64(v2Rev + 1)
}
// mkV2Node creates a V2 NodeExtern from a V3 KeyValue
func (s *v2v3Store) mkV2Node(kv *mvccpb.KeyValue) *store.NodeExtern {
if kv == nil {
return nil
}
n := &store.NodeExtern{
Key: string(s.mkNodePath(string(kv.Key))),
Dir: kv.Key[len(kv.Key)-1] == '/',
CreatedIndex: mkV2Rev(kv.CreateRevision),
ModifiedIndex: mkV2Rev(kv.ModRevision),
}
if !n.Dir {
v := string(kv.Value)
n.Value = &v
}
return n
}
// prevKeyFromPuts gets the prev key that is being put; ignores
// the put action response.
func prevKeyFromPuts(resp *clientv3.TxnResponse) *mvccpb.KeyValue {
for _, r := range resp.Responses {
pkv := r.GetResponsePut().PrevKv
if pkv != nil && pkv.CreateRevision > 0 {
return pkv
}
}
return nil
}
func (s *v2v3Store) newSTM(applyf func(concurrency.STM) error) (*clientv3.TxnResponse, error) {
return concurrency.NewSTM(s.c, applyf, concurrency.WithIsolation(concurrency.Serializable))
}
马建仓 AI 助手
尝试更多
代码解读
代码找茬
代码优化
1
https://gitee.com/who7708/etcd.git
git@gitee.com:who7708/etcd.git
who7708
etcd
etcd
v3.3.7

搜索帮助