README
¶
Valkeyrie Consul
valkeyrie provides a Go native library to store metadata using Distributed Key/Value stores (or common databases).
Compatibility
A storage backend in valkeyrie implements (fully or partially) the Store interface.
| Calls | Consul |
|---|---|
| Put | 🟢️ |
| Get | 🟢️ |
| Delete | 🟢️ |
| Exists | 🟢️ |
| Watch | 🟢️ |
| WatchTree | 🟢️ |
| NewLock (Lock/Unlock) | 🟢️ |
| List | 🟢️ |
| DeleteTree | 🟢️ |
| AtomicPut | 🟢️ |
| AtomicDelete | 🟢️ |
Supported Versions
Consul versions >= 0.5.1 because it uses Sessions with Delete behavior for the use of TTLs (mimics zookeeper's Ephemeral node support).
If you don't plan to use TTLs: you can use Consul version 0.4.0+.
Examples
package main
import (
"context"
"log"
"time"
"github.com/kvtools/consul"
"github.com/kvtools/valkeyrie"
)
func main() {
ctx := context.Background()
config := &consul.Config{
ConnectionTimeout: 10 * time.Second,
}
kv, err := valkeyrie.NewStore(ctx, consul.StoreName, []string{"localhost:8500"}, config)
if err != nil {
log.Fatal("Cannot create store")
}
key := "foo"
err = kv.Put(ctx, key, []byte("bar"), nil)
if err != nil {
log.Fatalf("Error trying to put value at key: %v", key)
}
pair, err := kv.Get(ctx, key, nil)
if err != nil {
log.Fatalf("Error trying accessing value at key: %v", key)
}
log.Printf("value: %s", string(pair.Value))
err = kv.Delete(ctx, key)
if err != nil {
log.Fatalf("Error trying to delete key %v", key)
}
}
Documentation
¶
Overview ¶
Package consul contains the Consul store implementation.
Index ¶
- Constants
- Variables
- type Config
- type Store
- func (s *Store) AtomicDelete(ctx context.Context, key string, previous *store.KVPair) (bool, error)
- func (s *Store) AtomicPut(ctx context.Context, key string, value []byte, previous *store.KVPair, ...) (bool, *store.KVPair, error)
- func (s *Store) Close() error
- func (s *Store) Delete(ctx context.Context, key string) error
- func (s *Store) DeleteTree(ctx context.Context, directory string) error
- func (s *Store) Exists(ctx context.Context, key string, opts *store.ReadOptions) (bool, error)
- func (s *Store) Get(_ context.Context, key string, opts *store.ReadOptions) (*store.KVPair, error)
- func (s *Store) List(_ context.Context, directory string, opts *store.ReadOptions) ([]*store.KVPair, error)
- func (s *Store) NewLock(ctx context.Context, key string, opts *store.LockOptions) (store.Locker, error)
- func (s *Store) Put(_ context.Context, key string, value []byte, opts *store.WriteOptions) error
- func (s *Store) Watch(ctx context.Context, key string, _ *store.ReadOptions) (<-chan *store.KVPair, error)
- func (s *Store) WatchTree(ctx context.Context, directory string, _ *store.ReadOptions) (<-chan []*store.KVPair, error)
Constants ¶
const ( // DefaultWatchWaitTime is how long we block for at a time to check if the watched key has changed. // This affects the minimum time it takes to cancel a watch. DefaultWatchWaitTime = 15 * time.Second // RenewSessionRetryMax is the number of time we should try to renew the session before giving up and throwing an error. RenewSessionRetryMax = 5 // MaxSessionDestroyAttempts is the maximum times we will try // to explicitly destroy the session attached to a lock after // the connectivity to the store has been lost. MaxSessionDestroyAttempts = 5 )
const StoreName = "consul"
StoreName the name of the store.
Variables ¶
var ( // ErrMultipleEndpointsUnsupported is thrown when there are multiple endpoints specified for Consul. ErrMultipleEndpointsUnsupported = errors.New("consul does not support multiple endpoints") // ErrSessionRenew is thrown when the session can't be renewed because the Consul version does not support sessions. ErrSessionRenew = errors.New("cannot set or renew session for ttl, unable to operate on sessions") )
Functions ¶
This section is empty.
Types ¶
type Config ¶
type Config struct {
TLS *tls.Config
ConnectionTimeout time.Duration
Token string
Namespace string
}
Config the Consul configuration.
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store implements the store.Store interface.
func (*Store) AtomicDelete ¶
AtomicDelete deletes a value at "key" if the key has not been modified in the meantime, throws an error if this is the case.
func (*Store) AtomicPut ¶
func (s *Store) AtomicPut(ctx context.Context, key string, value []byte, previous *store.KVPair, _ *store.WriteOptions) (bool, *store.KVPair, error)
AtomicPut puts a value at "key" if the key has not been modified in the meantime, throws an error if this is the case.
func (*Store) DeleteTree ¶
DeleteTree deletes a range of keys under a given directory.
func (*Store) Get ¶
Get the value at "key". Returns the last modified index to use in conjunction to CAS calls.
func (*Store) List ¶
func (s *Store) List(_ context.Context, directory string, opts *store.ReadOptions) ([]*store.KVPair, error)
List child nodes of a given directory.
func (*Store) NewLock ¶
func (s *Store) NewLock(ctx context.Context, key string, opts *store.LockOptions) (store.Locker, error)
NewLock returns a handle to a lock struct which can be used to provide mutual exclusion on a key.
func (*Store) Watch ¶
func (s *Store) Watch(ctx context.Context, key string, _ *store.ReadOptions) (<-chan *store.KVPair, error)
Watch for changes on a "key". It returns a channel that will receive changes or pass on errors. Upon creation, the current value will first be sent to the channel. Providing a non-nil stopCh can be used to stop watching.
func (*Store) WatchTree ¶
func (s *Store) WatchTree(ctx context.Context, directory string, _ *store.ReadOptions) (<-chan []*store.KVPair, error)
WatchTree watches for changes on a "directory". It returns a channel that will receive changes or pass on errors. Upon creating a watch, the current children values will be sent to the channel. Providing a non-nil stopCh can be used to stop watching.
Source Files