2020-05-15 17:59:37 +03:00
|
|
|
// +build kvdb_etcd
|
|
|
|
|
2020-02-18 21:26:57 +03:00
|
|
|
package etcd
|
|
|
|
|
|
|
|
import (
|
2020-02-18 21:35:53 +03:00
|
|
|
"context"
|
2020-03-13 18:59:26 +03:00
|
|
|
"fmt"
|
2020-02-18 21:35:53 +03:00
|
|
|
"io"
|
2020-03-13 18:59:26 +03:00
|
|
|
"runtime"
|
|
|
|
"sync"
|
2020-02-18 21:26:57 +03:00
|
|
|
"time"
|
|
|
|
|
2020-02-18 21:35:53 +03:00
|
|
|
"github.com/btcsuite/btcwallet/walletdb"
|
2020-02-18 21:26:57 +03:00
|
|
|
"github.com/coreos/etcd/clientv3"
|
2020-12-16 20:23:06 +03:00
|
|
|
"github.com/coreos/etcd/clientv3/namespace"
|
2020-05-08 19:38:46 +03:00
|
|
|
"github.com/coreos/etcd/pkg/transport"
|
2020-02-18 21:26:57 +03:00
|
|
|
)
|
|
|
|
|
|
|
|
const (
|
2020-08-10 17:50:06 +03:00
|
|
|
// etcdConnectionTimeout is the timeout until successful connection to
|
|
|
|
// the etcd instance.
|
2020-02-18 21:26:57 +03:00
|
|
|
etcdConnectionTimeout = 10 * time.Second
|
2020-02-18 21:35:53 +03:00
|
|
|
|
|
|
|
// etcdLongTimeout is a timeout for longer taking etcd operatons.
|
|
|
|
etcdLongTimeout = 30 * time.Second
|
2020-02-18 21:26:57 +03:00
|
|
|
)
|
|
|
|
|
2020-03-13 18:59:26 +03:00
|
|
|
// callerStats holds commit stats for a specific caller. Currently it only
|
|
|
|
// holds the max stat, meaning that for a particular caller the largest
|
|
|
|
// commit set is recorded.
|
|
|
|
type callerStats struct {
|
|
|
|
count int
|
|
|
|
commitStats CommitStats
|
|
|
|
}
|
|
|
|
|
|
|
|
func (s callerStats) String() string {
|
|
|
|
return fmt.Sprintf("count: %d, retries: %d, rset: %d, wset: %d",
|
2020-08-10 17:50:06 +03:00
|
|
|
s.count, s.commitStats.Retries, s.commitStats.Rset,
|
|
|
|
s.commitStats.Wset)
|
2020-03-13 18:59:26 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
// commitStatsCollector collects commit stats for commits succeeding
|
|
|
|
// and also for commits failing.
|
|
|
|
type commitStatsCollector struct {
|
|
|
|
sync.RWMutex
|
|
|
|
succ map[string]*callerStats
|
|
|
|
fail map[string]*callerStats
|
|
|
|
}
|
|
|
|
|
|
|
|
// newCommitStatsColletor creates a new commitStatsCollector instance.
|
|
|
|
func newCommitStatsColletor() *commitStatsCollector {
|
|
|
|
return &commitStatsCollector{
|
|
|
|
succ: make(map[string]*callerStats),
|
|
|
|
fail: make(map[string]*callerStats),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// PrintStats returns collected stats pretty printed into a string.
|
|
|
|
func (c *commitStatsCollector) PrintStats() string {
|
|
|
|
c.RLock()
|
|
|
|
defer c.RUnlock()
|
|
|
|
|
|
|
|
s := "\nFailure:\n"
|
|
|
|
for k, v := range c.fail {
|
|
|
|
s += fmt.Sprintf("%s\t%s\n", k, v)
|
|
|
|
}
|
|
|
|
|
|
|
|
s += "\nSuccess:\n"
|
|
|
|
for k, v := range c.succ {
|
|
|
|
s += fmt.Sprintf("%s\t%s\n", k, v)
|
|
|
|
}
|
|
|
|
|
|
|
|
return s
|
|
|
|
}
|
|
|
|
|
|
|
|
// updateStatsMap updatess commit stats map for a caller.
|
|
|
|
func updateStatMap(
|
|
|
|
caller string, stats CommitStats, m map[string]*callerStats) {
|
|
|
|
|
|
|
|
if _, ok := m[caller]; !ok {
|
|
|
|
m[caller] = &callerStats{}
|
|
|
|
}
|
|
|
|
|
|
|
|
curr := m[caller]
|
|
|
|
curr.count++
|
|
|
|
|
|
|
|
// Update only if the total commit set is greater or equal.
|
|
|
|
currTotal := curr.commitStats.Rset + curr.commitStats.Wset
|
|
|
|
if currTotal <= (stats.Rset + stats.Wset) {
|
|
|
|
curr.commitStats = stats
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// callback is an STM commit stats callback passed which can be passed
|
|
|
|
// using a WithCommitStatsCallback to the STM upon construction.
|
|
|
|
func (c *commitStatsCollector) callback(succ bool, stats CommitStats) {
|
|
|
|
caller := "unknown"
|
|
|
|
|
|
|
|
// Get the caller. As this callback is called from
|
|
|
|
// the backend interface that means we need to ascend
|
|
|
|
// 4 frames in the callstack.
|
|
|
|
_, file, no, ok := runtime.Caller(4)
|
|
|
|
if ok {
|
|
|
|
caller = fmt.Sprintf("%s#%d", file, no)
|
|
|
|
}
|
|
|
|
|
|
|
|
c.Lock()
|
|
|
|
defer c.Unlock()
|
|
|
|
|
|
|
|
if succ {
|
|
|
|
updateStatMap(caller, stats, c.succ)
|
|
|
|
} else {
|
|
|
|
updateStatMap(caller, stats, c.fail)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2020-02-18 21:26:57 +03:00
|
|
|
// db holds a reference to the etcd client connection.
|
|
|
|
type db struct {
|
2020-03-13 18:59:26 +03:00
|
|
|
config BackendConfig
|
|
|
|
cli *clientv3.Client
|
|
|
|
commitStatsCollector *commitStatsCollector
|
2020-08-10 17:50:06 +03:00
|
|
|
txQueue *commitQueue
|
2020-02-18 21:26:57 +03:00
|
|
|
}
|
|
|
|
|
2020-02-18 21:35:53 +03:00
|
|
|
// Enforce db implements the walletdb.DB interface.
|
|
|
|
var _ walletdb.DB = (*db)(nil)
|
|
|
|
|
2020-02-18 21:26:57 +03:00
|
|
|
// BackendConfig holds and etcd backend config and connection parameters.
|
|
|
|
type BackendConfig struct {
|
2020-05-25 19:00:22 +03:00
|
|
|
// Ctx is the context we use to cancel operations upon exit.
|
|
|
|
Ctx context.Context
|
|
|
|
|
2020-02-18 21:26:57 +03:00
|
|
|
// Host holds the peer url of the etcd instance.
|
|
|
|
Host string
|
|
|
|
|
|
|
|
// User is the username for the etcd peer.
|
|
|
|
User string
|
|
|
|
|
|
|
|
// Pass is the password for the etcd peer.
|
|
|
|
Pass string
|
2020-03-13 18:59:26 +03:00
|
|
|
|
2020-12-21 18:18:13 +03:00
|
|
|
// DisableTLS disables the use of TLS for etcd connections.
|
|
|
|
DisableTLS bool
|
|
|
|
|
2020-05-08 19:38:46 +03:00
|
|
|
// CertFile holds the path to the TLS certificate for etcd RPC.
|
|
|
|
CertFile string
|
|
|
|
|
|
|
|
// KeyFile holds the path to the TLS private key for etcd RPC.
|
|
|
|
KeyFile string
|
|
|
|
|
|
|
|
// InsecureSkipVerify should be set to true if we intend to
|
|
|
|
// skip TLS verification.
|
|
|
|
InsecureSkipVerify bool
|
|
|
|
|
2020-05-12 17:02:46 +03:00
|
|
|
// Prefix the hash of the prefix will be used as the root
|
|
|
|
// bucket id. This enables key space separation similar to
|
|
|
|
// name spaces.
|
|
|
|
Prefix string
|
|
|
|
|
2020-12-16 20:23:06 +03:00
|
|
|
// Namespace is the etcd namespace that we'll use for all keys.
|
|
|
|
Namespace string
|
|
|
|
|
2020-03-13 18:59:26 +03:00
|
|
|
// CollectCommitStats indicates wheter to commit commit stats.
|
|
|
|
CollectCommitStats bool
|
2020-02-18 21:26:57 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
// newEtcdBackend returns a db object initialized with the passed backend
|
|
|
|
// config. If etcd connection cannot be estabished, then returns error.
|
|
|
|
func newEtcdBackend(config BackendConfig) (*db, error) {
|
2020-05-25 19:00:22 +03:00
|
|
|
if config.Ctx == nil {
|
|
|
|
config.Ctx = context.Background()
|
|
|
|
}
|
|
|
|
|
2020-12-21 18:18:13 +03:00
|
|
|
clientCfg := clientv3.Config{
|
2020-07-10 16:29:48 +03:00
|
|
|
Context: config.Ctx,
|
|
|
|
Endpoints: []string{config.Host},
|
|
|
|
DialTimeout: etcdConnectionTimeout,
|
|
|
|
Username: config.User,
|
|
|
|
Password: config.Pass,
|
|
|
|
MaxCallSendMsgSize: 16384*1024 - 1,
|
2020-12-21 18:18:13 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
if !config.DisableTLS {
|
|
|
|
tlsInfo := transport.TLSInfo{
|
|
|
|
CertFile: config.CertFile,
|
|
|
|
KeyFile: config.KeyFile,
|
|
|
|
InsecureSkipVerify: config.InsecureSkipVerify,
|
|
|
|
}
|
|
|
|
|
|
|
|
tlsConfig, err := tlsInfo.ClientConfig()
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
clientCfg.TLS = tlsConfig
|
|
|
|
}
|
|
|
|
|
|
|
|
cli, err := clientv3.New(clientCfg)
|
2020-02-18 21:26:57 +03:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
2020-12-16 20:23:06 +03:00
|
|
|
// Apply the namespace.
|
|
|
|
cli.KV = namespace.NewKV(cli.KV, config.Namespace)
|
|
|
|
cli.Watcher = namespace.NewWatcher(cli.Watcher, config.Namespace)
|
|
|
|
cli.Lease = namespace.NewLease(cli.Lease, config.Namespace)
|
|
|
|
|
2020-02-18 21:26:57 +03:00
|
|
|
backend := &db{
|
2020-08-10 17:50:06 +03:00
|
|
|
cli: cli,
|
|
|
|
config: config,
|
|
|
|
txQueue: NewCommitQueue(config.Ctx),
|
2020-03-13 18:59:26 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
if config.CollectCommitStats {
|
|
|
|
backend.commitStatsCollector = newCommitStatsColletor()
|
2020-02-18 21:26:57 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
return backend, nil
|
|
|
|
}
|
|
|
|
|
2020-03-13 18:59:26 +03:00
|
|
|
// getSTMOptions creats all STM options based on the backend config.
|
|
|
|
func (db *db) getSTMOptions() []STMOptionFunc {
|
2020-08-10 17:50:06 +03:00
|
|
|
opts := []STMOptionFunc{
|
|
|
|
WithAbortContext(db.config.Ctx),
|
|
|
|
}
|
2020-05-25 19:00:22 +03:00
|
|
|
|
2020-03-13 18:59:26 +03:00
|
|
|
if db.config.CollectCommitStats {
|
|
|
|
opts = append(opts,
|
|
|
|
WithCommitStatsCallback(db.commitStatsCollector.callback),
|
|
|
|
)
|
|
|
|
}
|
|
|
|
|
|
|
|
return opts
|
|
|
|
}
|
|
|
|
|
2020-02-18 21:35:53 +03:00
|
|
|
// View opens a database read transaction and executes the function f with the
|
2020-10-20 17:18:40 +03:00
|
|
|
// transaction passed as a parameter. After f exits, the transaction is rolled
|
|
|
|
// back. If f errors, its error is returned, not a rollback error (if any
|
|
|
|
// occur). The passed reset function is called before the start of the
|
|
|
|
// transaction and can be used to reset intermediate state. As callers may
|
|
|
|
// expect retries of the f closure (depending on the database backend used), the
|
|
|
|
// reset function will be called before each retry respectively.
|
|
|
|
func (db *db) View(f func(tx walletdb.ReadTx) error, reset func()) error {
|
2020-02-18 21:35:53 +03:00
|
|
|
apply := func(stm STM) error {
|
2020-10-20 17:18:40 +03:00
|
|
|
reset()
|
2020-05-12 17:02:46 +03:00
|
|
|
return f(newReadWriteTx(stm, db.config.Prefix))
|
2020-02-18 21:35:53 +03:00
|
|
|
}
|
|
|
|
|
2020-08-10 17:50:06 +03:00
|
|
|
return RunSTM(db.cli, apply, db.txQueue, db.getSTMOptions()...)
|
2020-02-18 21:35:53 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
// Update opens a database read/write transaction and executes the function f
|
2020-10-26 16:06:32 +03:00
|
|
|
// with the transaction passed as a parameter. After f exits, if f did not
|
|
|
|
// error, the transaction is committed. Otherwise, if f did error, the
|
|
|
|
// transaction is rolled back. If the rollback fails, the original error
|
|
|
|
// returned by f is still returned. If the commit fails, the commit error is
|
|
|
|
// returned. As callers may expect retries of the f closure, the reset function
|
|
|
|
// will be called before each retry respectively.
|
|
|
|
func (db *db) Update(f func(tx walletdb.ReadWriteTx) error, reset func()) error {
|
2020-02-18 21:35:53 +03:00
|
|
|
apply := func(stm STM) error {
|
2020-10-26 16:06:32 +03:00
|
|
|
reset()
|
2020-05-12 17:02:46 +03:00
|
|
|
return f(newReadWriteTx(stm, db.config.Prefix))
|
2020-02-18 21:35:53 +03:00
|
|
|
}
|
|
|
|
|
2020-08-10 17:50:06 +03:00
|
|
|
return RunSTM(db.cli, apply, db.txQueue, db.getSTMOptions()...)
|
2020-03-13 18:59:26 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
// PrintStats returns all collected stats pretty printed into a string.
|
|
|
|
func (db *db) PrintStats() string {
|
|
|
|
if db.commitStatsCollector != nil {
|
|
|
|
return db.commitStatsCollector.PrintStats()
|
|
|
|
}
|
|
|
|
|
|
|
|
return ""
|
2020-02-18 21:35:53 +03:00
|
|
|
}
|
|
|
|
|
2020-08-10 17:50:06 +03:00
|
|
|
// BeginReadWriteTx opens a database read+write transaction.
|
2020-02-18 21:35:53 +03:00
|
|
|
func (db *db) BeginReadWriteTx() (walletdb.ReadWriteTx, error) {
|
2020-05-12 17:02:46 +03:00
|
|
|
return newReadWriteTx(
|
2020-08-10 17:50:06 +03:00
|
|
|
NewSTM(db.cli, db.txQueue, db.getSTMOptions()...),
|
2020-05-12 17:02:46 +03:00
|
|
|
db.config.Prefix,
|
|
|
|
), nil
|
2020-02-18 21:35:53 +03:00
|
|
|
}
|
|
|
|
|
2020-08-10 17:50:06 +03:00
|
|
|
// BeginReadTx opens a database read transaction.
|
2020-02-18 21:35:53 +03:00
|
|
|
func (db *db) BeginReadTx() (walletdb.ReadTx, error) {
|
2020-05-12 17:02:46 +03:00
|
|
|
return newReadWriteTx(
|
2020-08-10 17:50:06 +03:00
|
|
|
NewSTM(db.cli, db.txQueue, db.getSTMOptions()...),
|
2020-05-12 17:02:46 +03:00
|
|
|
db.config.Prefix,
|
|
|
|
), nil
|
2020-02-18 21:35:53 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
// Copy writes a copy of the database to the provided writer. This call will
|
|
|
|
// start a read-only transaction to perform all operations.
|
|
|
|
// This function is part of the walletdb.Db interface implementation.
|
|
|
|
func (db *db) Copy(w io.Writer) error {
|
2020-05-25 19:00:22 +03:00
|
|
|
ctx, cancel := context.WithTimeout(db.config.Ctx, etcdLongTimeout)
|
2020-02-18 21:35:53 +03:00
|
|
|
defer cancel()
|
|
|
|
|
|
|
|
readCloser, err := db.cli.Snapshot(ctx)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
_, err = io.Copy(w, readCloser)
|
|
|
|
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
// Close cleanly shuts down the database and syncs all data.
|
|
|
|
// This function is part of the walletdb.Db interface implementation.
|
2020-02-18 21:26:57 +03:00
|
|
|
func (db *db) Close() error {
|
|
|
|
return db.cli.Close()
|
|
|
|
}
|
2020-02-18 21:35:53 +03:00
|
|
|
|
|
|
|
// Batch opens a database read/write transaction and executes the function f
|
|
|
|
// with the transaction passed as a parameter. After f exits, if f did not
|
|
|
|
// error, the transaction is committed. Otherwise, if f did error, the
|
|
|
|
// transaction is rolled back. If the rollback fails, the original error
|
|
|
|
// returned by f is still returned. If the commit fails, the commit error is
|
|
|
|
// returned.
|
|
|
|
//
|
|
|
|
// Batch is only useful when there are multiple goroutines calling it.
|
|
|
|
func (db *db) Batch(apply func(tx walletdb.ReadWriteTx) error) error {
|
2020-10-26 16:06:32 +03:00
|
|
|
return db.Update(apply, func() {})
|
2020-02-18 21:35:53 +03:00
|
|
|
}
|