zookeeper

README

Valkeyrie Zookeeper

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	Zookeeper
Put	🟢️
Get	🟢️
Delete	🟢️
Exists	🟢️
Watch	🟢️
WatchTree	🟢️
NewLock (Lock/Unlock)	🟢️
List	🟢️
DeleteTree	🟢️
AtomicPut	🟢️
AtomicDelete	🟢️

Supported Versions

Zookeeper versions >= 3.4.5.

Examples

package main

import (
	"context"
	"log"

	"github.com/kvtools/valkeyrie"
	"github.com/kvtools/zookeeper"
)

func main() {
	ctx := context.Background()

	config := &zookeeper.Config{
        Username: "example",
	}

	kv, err := valkeyrie.NewStore(ctx, zookeeper.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 zookeeper contains the ZooKeeper store implementation.

Index

• Constants
• func New(_ context.Context, endpoints []string, options *Config) (store.Store, error)
• type Config
• type Store
  • func (s *Store) AtomicDelete(_ context.Context, key string, previous *store.KVPair) (bool, error)
  • func (s *Store) AtomicPut(_ context.Context, key string, value []byte, previous *store.KVPair, ...) (bool, *store.KVPair, error)
  • func (s *Store) Close() error
  • func (s *Store) Delete(_ context.Context, key string) error
  • func (s *Store) DeleteTree(_ context.Context, directory string) error
  • func (s *Store) Exists(_ context.Context, key string, _ *store.ReadOptions) (bool, error)
  • func (s *Store) Get(_ context.Context, key string, _ *store.ReadOptions) (pair *store.KVPair, err error)
  • func (s *Store) List(ctx context.Context, directory string, opts *store.ReadOptions) ([]*store.KVPair, error)
  • func (s *Store) NewLock(_ context.Context, key string, opts *store.LockOptions) (lock store.Locker, err error)
  • func (s *Store) Put(ctx 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, opts *store.ReadOptions) (<-chan []*store.KVPair, error)

Constants

const (
	// SOH control character.
	SOH = "\x01"
)

const StoreName = "zookeeper"

StoreName the name of the store.

Functions

func New

func New(_ context.Context, endpoints []string, options *Config) (store.Store, error)

New creates a new Zookeeper client.

Types

type Config

type Config struct {
	ConnectionTimeout time.Duration
	Username          string
	Password          string
	MaxBufferSize     int
}

Config the Zookeeper configuration.

type Store

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

Store implements the store.Store interface.

func (*Store) AtomicDelete

func (s *Store) AtomicDelete(_ context.Context, key string, previous *store.KVPair) (bool, error)

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(_ 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) Close

func (s *Store) Close() error

Close closes the client connection.

func (*Store) Delete

func (s *Store) Delete(_ context.Context, key string) error

Delete a value at "key".

func (*Store) DeleteTree

func (s *Store) DeleteTree(_ context.Context, directory string) error

DeleteTree deletes a range of keys under a given directory.

func (*Store) Exists

func (s *Store) Exists(_ context.Context, key string, _ *store.ReadOptions) (bool, error)

Exists checks if the key exists inside the store.

func (*Store) Get

func (s *Store) Get(_ context.Context, key string, _ *store.ReadOptions) (pair *store.KVPair, err error)

Get the value at "key". Returns the last modified index to use in conjunction to Atomic calls.

func (*Store) List

func (s *Store) List(ctx context.Context, directory string, opts *store.ReadOptions) ([]*store.KVPair, error)

List child nodes of a given directory.

func (*Store) NewLock

func (s *Store) NewLock(_ context.Context, key string, opts *store.LockOptions) (lock store.Locker, err error)

NewLock returns a handle to a lock struct which can be used to provide mutual exclusion on a key.

func (*Store) Put

func (s *Store) Put(ctx context.Context, key string, value []byte, opts *store.WriteOptions) error

Put a value at "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, opts *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.