etcdadpt

package module
v0.5.2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Feb 8, 2023 License: Apache-2.0 Imports: 17 Imported by: 4

README

etcd-adapter

Expose a standard KV operation API, adapt to embeded etcd and etcd client

How to use?

Step 1. Import the module and it's all plugins.

import (
	github.com/little-cui/etcdadpt
	_ "github.com/little-cui/etcdadpt/embedded"
	_ "github.com/little-cui/etcdadpt/remote"
)

Step 2. Select one mode and do initialization.

With embedded etcd mode:

etcdadpt.Init(etcdadpt.Config{
	Kind:             "embedded_etcd",
	ClusterName:      "c-0",
	ClusterAddresses: "c-0=http://127.0.0.1:2379",
})

This mode will start an embedded etcd server.

With remote etcd mode:

startup etcd server.

docker run -d  -p 2379:2379 --name etcd quay.io/coreos/etcd:v3.2.13 etcd \
  --listen-client-urls http://0.0.0.0:2379 --advertise-client-urls http://0.0.0.0:2379

write the following code.

etcdadpt.Init(etcdadpt.Config{
	Kind:             "etcd",
	ClusterAddresses: "127.0.0.1:2379",
})

Step 3. call the API and enjoy it!

// put a key
_ := etcdadpt.Put(context.Background(), "/key", "abc")
// get a key
kv, _ := etcdadpt.Get(context.Background(), "/key")
log.Println(fmt.Sprintf("%v", kv))

and you will see log print below:

key:"/key" create_revision:4 mod_revision:4 version:1 value:"abc"

Distributed Etcd lock

example
lock, _ := etcdadpt.Lock("/test", -1)
defer lock.Unlock()
//do something
g += 1
fmt.Println(g)

// lock a key for a period of time, and then renew
dLock, err := etcdadpt.Lock("renewKey", 5)
time.Sleep(3 * time.Second)
err = dLock.Refresh()

Examples

Also see the full demo HERE!

Documentation

Overview

Package etcdadpt is the abstraction of kv database operator

Index

Constants

View Source 
const (
	// MaxTxnNumberOneTime the same as v3rpc.MaxOpsPerTxn = 128
	MaxTxnNumberOneTime = 128
	// DefaultPageCount grpc does not allow to transport a large body more then 4MB in a request
	DefaultPageCount = 4096
	// DefaultDialTimeout the timeout dial to etcd
	DefaultDialTimeout     = 10 * time.Second
	DefaultRequestTimeout  = 30 * time.Second
	DefaultCompactInterval = time.Hour
	DefaultClusterName     = "default"
)
View Source 
const (
	DefaultLockTTL    = 60
	DefaultRetryTimes = 3

	DefaultLock         = "/lock"
	OperationGlobalLock = "GLOBAL_LOCK"
)

Variables

View Source 
var ErrLeaseIDNotExists = errors.New("leaseID is nil")
View Source 
var (
	ErrLeaseNotFound = errors.New(rpctypes.ErrLeaseNotFound.Error())
)
View Source 
var ErrLockKeyFail = errors.New("fail to lock key")
View Source 
var (
	ErrNoPlugin = errors.New("required etcd adapter implement, please import the pkg")
)
View Source 
var (
	IsDebug bool
)

Functions

func Delete 

func Delete(ctx context.Context, key string, opts ...OpOption) (bool, error)

func DeleteMany 

func DeleteMany(ctx context.Context, opts ...OpOptions) (bool, error)

func Exist 

func Exist(ctx context.Context, key string) (bool, error)

Exist get one kv, if can not get return false

func Get 

func Get(ctx context.Context, key string) (*mvccpb.KeyValue, error)

Get get one kv

func GetClusterURL 

func GetClusterURL(clusterName, clusterURLs, managerURLs string) []string

func Init 

func Init(cfg Config) error

Init construct storage plugin instance invoked by sc main process

func Insert 

func Insert(ctx context.Context, key, value string, opts ...OpOption) (bool, error)

func InsertBytes 

func InsertBytes(ctx context.Context, key string, value []byte, opts ...OpOption) (bool, error)

InsertBytes insert a new kv, return false if the key exist

func Install 

func Install(pluginImplName string, newFunc newClientFunc)

Install load plugins configuration into plugins

func List 

func List(ctx context.Context, key string, opts ...OpOption) ([]*mvccpb.KeyValue, int64, error)

List get kv list

func Put 

func Put(ctx context.Context, key string, value string, opts ...OpOption) error

Put insert or update kv

func PutBytes 

func PutBytes(ctx context.Context, key string, value []byte, opts ...OpOption) error

PutBytes insert or update kv

func Txn 

func Txn(ctx context.Context, opts []OpOptions) error

Types

type Action 

type Action int
const (
	ActionGet Action = iota
	ActionPut
	ActionDelete
)

func (Action) String 

func (at Action) String() string

type CacheMode 

type CacheMode int
const (
	ModeBoth CacheMode = iota
	ModeCache
	ModeNoCache
)

func (CacheMode) String 

func (cm CacheMode) String() string

type Client 

type Client interface {
	Err() <-chan error
	Ready() <-chan struct{}
	Do(ctx context.Context, opts ...OpOption) (*Response, error)
	Txn(ctx context.Context, ops []OpOptions) (*Response, error)
	TxnWithCmp(ctx context.Context, success []OpOptions, cmp []CmpOptions, fail []OpOptions) (*Response, error)
	LeaseGrant(ctx context.Context, TTL int64) (leaseID int64, err error)
	LeaseRenew(ctx context.Context, leaseID int64) (TTL int64, err error)
	LeaseRevoke(ctx context.Context, leaseID int64) error
	// Watch block util:
	// 1. connection error
	// 2. call send function failed
	// 3. response.Err()
	// 4. time out to watch, but return nil
	Watch(ctx context.Context, opts ...OpOption) error
	Compact(ctx context.Context, reserve int64) error
	Close()

	ListCluster(ctx context.Context) (Clusters, error)

	Status(ctx context.Context) (*StatusResponse, error)
}

Client is an abstraction of kv database operator Support etcd by default

func Instance 

func Instance() Client

Instance is the instance of Etcd client

func NewInstance 

func NewInstance(cfg Config) (Client, error)

type Clusters 

type Clusters map[string][]string

func ListCluster 

func ListCluster(ctx context.Context) (Clusters, error)

func ParseClusters 

func ParseClusters(clusterName, clusterURLs, managerURLs string) Clusters

ParseClusters convert the cluster url string to Clusters type. The clusterURLs format like 'sc-0=http(s)://host1:port1,http(s)://host2:port2,sc-1=http(s)://host3:port3', managerURLs is optional, set the result value with the key clusterName

type CmpOption 

type CmpOption func(op *CmpOptions)

type CmpOptions 

type CmpOptions struct {
	Key    []byte
	Type   CmpType
	Result CmpResult
	Value  interface{}
}

func EqualCreateRev 

func EqualCreateRev(key string, v interface{}) CmpOptions

func EqualModRev 

func EqualModRev(key string, v interface{}) CmpOptions

func EqualVal 

func EqualVal(key string, v interface{}) CmpOptions

func EqualVer 

func EqualVer(key string, v interface{}) CmpOptions

func ExistKey added in v0.3.1 

func ExistKey(key string) CmpOptions

func GreaterCreateRev 

func GreaterCreateRev(key string, v interface{}) CmpOptions

func GreaterModRev 

func GreaterModRev(key string, v interface{}) CmpOptions

func If 

func If(opts ...CmpOptions) []CmpOptions

utils

func LessCreateRev 

func LessCreateRev(key string, v interface{}) CmpOptions

func LessModRev 

func LessModRev(key string, v interface{}) CmpOptions

func NotEqualCreateRev 

func NotEqualCreateRev(key string, v interface{}) CmpOptions

func NotEqualModRev 

func NotEqualModRev(key string, v interface{}) CmpOptions

func NotEqualVal 

func NotEqualVal(key string, v interface{}) CmpOptions

func NotEqualVer 

func NotEqualVer(key string, v interface{}) CmpOptions

func NotExistKey added in v0.3.1 

func NotExistKey(key string) CmpOptions

func (CmpOptions) String 

func (op CmpOptions) String() string

type CmpResult 

type CmpResult int
const (
	CmpEqual CmpResult = iota
	CmpGreater
	CmpLess
	CmpNotEqual
)

func (CmpResult) String 

func (cr CmpResult) String() string

type CmpType 

type CmpType int
const (
	CmpVersion CmpType = iota
	CmpCreate
	CmpMod
	CmpValue
)

func (CmpType) String 

func (ct CmpType) String() string

type Config 

type Config struct {
	// Kind plugin kind, can be 'etcd' or 'embedded_etcd'
	Kind string `json:"-"`
	// Logger logger for adapter, by default use openlog.GetLogger()
	Logger     openlog.Logger `json:"-"`
	SslEnabled bool           `json:"-"`
	TLSConfig  *tls.Config    `json:"-"`
	// ErrorFunc called when connection error occurs
	ErrorFunc func(err error) `json:"-"`
	// ConnectedFunc called when connected
	ConnectedFunc func() `json:"-"`
	// ManagerAddress optional, the list of cluster manager endpoints
	ManagerAddress string `json:"manageAddress,omitempty"`
	// ClusterName required when Kind = 'embedded_etcd'
	ClusterName string `json:"manageName,omitempty"`
	// ClusterAddresses required, the list of cluster client endpoints
	ClusterAddresses string        `json:"manageClusters,omitempty"` // the raw string of cluster configuration
	DialTimeout      time.Duration `json:"connectTimeout"`
	RequestTimeOut   time.Duration `json:"registryTimeout"`
	// AutoSyncInterval optional, then duration of auto sync the cluster members and check them health
	AutoSyncInterval time.Duration `json:"autoSyncInterval"`
	// CompactInterval optional, set DefaultCompactInterval if value equal to 0
	CompactInterval   time.Duration `json:"-"`
	CompactIndexDelta int64         `json:"-"`
}

func (*Config) Init 

func (c *Config) Init()

type DLock added in v0.3.0 

type DLock struct {
	// contains filtered or unexported fields
}

func Lock added in v0.3.0 

func Lock(key string, ttl int64) (*DLock, error)

Lock func will lock the key, and retry three times if it fails. ttl unit is second.

func TryLock added in v0.3.0 

func TryLock(key string, ttl int64) (*DLock, error)

TryLock func will try to lock the key. ttl unit is second.

func (*DLock) ID added in v0.3.0 

func (m *DLock) ID() string

func (*DLock) Refresh added in v0.3.0 

func (m *DLock) Refresh() error

func (*DLock) Unlock added in v0.3.0 

func (m *DLock) Unlock() (err error)

type OpOption 

type OpOption func(*OpOptions)
var DEL OpOption = func(op *OpOptions) { op.Action = ActionDelete }
var GET OpOption = func(op *OpOptions) { op.Action = ActionGet }
var PUT OpOption = func(op *OpOptions) { op.Action = ActionPut }

func WatchPrefixOpOptions 

func WatchPrefixOpOptions(key string) []OpOption

func WithAscendOrder 

func WithAscendOrder() OpOption

func WithCacheOnly 

func WithCacheOnly() OpOption

func WithCountOnly 

func WithCountOnly() OpOption

func WithDescendOrder 

func WithDescendOrder() OpOption

func WithEndKey 

func WithEndKey(key []byte) OpOption

func WithGlobal 

func WithGlobal() OpOption

func WithIgnoreLease 

func WithIgnoreLease() OpOption

func WithKey 

func WithKey(key []byte) OpOption

func WithKeyOnly 

func WithKeyOnly() OpOption

func WithLease 

func WithLease(leaseID int64) OpOption

func WithLimit 

func WithLimit(i int64) OpOption

func WithNoCache 

func WithNoCache() OpOption

func WithNoneOrder 

func WithNoneOrder() OpOption

func WithOffset 

func WithOffset(i int64) OpOption

func WithPrefix 

func WithPrefix() OpOption

func WithPrevKv 

func WithPrevKv() OpOption

func WithRev 

func WithRev(revision int64) OpOption

func WithStrEndKey 

func WithStrEndKey(key string) OpOption

func WithStrKey 

func WithStrKey(key string) OpOption

func WithStrValue 

func WithStrValue(value string) OpOption

func WithValue 

func WithValue(value []byte) OpOption

func WithWatchCallback 

func WithWatchCallback(f WatchCallback) OpOption

type OpOptions 

type OpOptions struct {
	Action Action
	Key    []byte
	// EndKey must be lexicographically greater than Key.
	EndKey        []byte
	Value         []byte
	Prefix        bool
	PrevKV        bool
	Lease         int64
	KeyOnly       bool
	CountOnly     bool
	OrderBy       SortTarget
	SortOrder     SortOrder
	Revision      int64
	IgnoreLease   bool
	Mode          CacheMode
	WatchCallback WatchCallback
	Offset        int64
	Limit         int64
	Global        bool
}

func OpDel 

func OpDel(opts ...OpOption) (op OpOptions)

func OpGet 

func OpGet(opts ...OpOption) (op OpOptions)

func OpPut 

func OpPut(opts ...OpOption) (op OpOptions)

func Ops 

func Ops(ops ...OpOptions) []OpOptions

func OptionsToOp 

func OptionsToOp(opts ...OpOption) (op OpOptions)

func (OpOptions) CacheOnly 

func (op OpOptions) CacheOnly() bool

func (OpOptions) LargeRequestPaging added in v0.1.3 

func (op OpOptions) LargeRequestPaging() bool

func (OpOptions) NoCache 

func (op OpOptions) NoCache() bool

func (OpOptions) String 

func (op OpOptions) String() string

func (OpOptions) URI 

func (op OpOptions) URI() string

type Operation 

type Operation func(...OpOption) (op OpOptions)

type Response 

type Response struct {
	Action    Action
	Kvs       []*mvccpb.KeyValue
	Count     int64
	Revision  int64
	Succeeded bool
}

func ListAndDelete 

func ListAndDelete(ctx context.Context, key string, opts ...OpOption) (*Response, error)

ListAndDelete delete key and return the deleted key

func ListAndDeleteMany 

func ListAndDeleteMany(ctx context.Context, opts ...OpOptions) (*Response, error)

ListAndDeleteMany delete key and return the deleted key

func PutBytesAndGet 

func PutBytesAndGet(ctx context.Context, key string, value []byte, opts ...OpOption) (*Response, error)

PutBytesAndGet insert/update kv and return it

func TxnWithCmp 

func TxnWithCmp(ctx context.Context, opts []OpOptions,
	cmp []CmpOptions, fail []OpOptions) (resp *Response, err error)

func (*Response) MaxModRevision 

func (pr *Response) MaxModRevision() (max int64)

func (*Response) String 

func (pr *Response) String() string

type SortOrder 

type SortOrder int
const (
	SortNone SortOrder = iota
	SortAscend
	SortDescend
)

func (SortOrder) String 

func (so SortOrder) String() string

type SortTarget 

type SortTarget int
const (
	OrderByKey SortTarget = iota
	OrderByCreate
	OrderByMod
	OrderByVer
)

func (SortTarget) String 

func (st SortTarget) String() string

type StatusResponse added in v0.5.0 

type StatusResponse struct {
	DBSize int64
}

type WatchCallback 

type WatchCallback func(message string, evt *Response) error

Source Files

View all Source files

Directories

Path Synopsis
examples
dev command
middleware
log
metrics
tracing

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.