zkhelper

README

Coordinator

Support both zookeeper and etcd

Documentation

Overview

Package fakezk is a pretty complete mock implementation of a Zookeper connection (see go/zk/zk.Conn). All operations work as expected with the exceptions of zk.Conn.ACL and zk.Conn.SetACL. zk.Conn.SetACL will succeed, but it is a noop (and the ACLs won't be respected). zk.Conn.ACL will panic. It is OK to access the connection from multiple goroutines, but the locking is very naive (every operation locks the whole connection).

Index

• Constants
• Variables
• func ChildrenRecursive(zconn Conn, zkPath string) ([]string, error)
• func CreateOrUpdate(zconn Conn, zkPath, value string, flags int, aclv []zk.ACL, recursive bool) (pathCreated string, err error)
• func CreatePidNode(zconn Conn, zkPath string, contents string, done chan struct{}) error
• func CreateRecursive(zconn Conn, zkPath, value string, flags int, aclv []zk.ACL) (pathCreated string, err error)
• func DefaultACLs() []zk.ACL
• func DefaultDirACLs() []zk.ACL
• func DefaultFileACLs() []zk.ACL
• func DeleteRecursive(zconn Conn, zkPath string, version int) error
• func HasWildcard(path string) bool
• func IsDirectory(aclv []zk.ACL) bool
• func NodeExists(zconn Conn, zkPath string) (bool, error)
• func ObtainQueueLock(zconn Conn, zkPath string, wait time.Duration, interrupted chan struct{}) error
• func ResolveWildcards(zconn Conn, zkPaths []string) ([]string, error)
• func ZkErrorEqual(a, b error) bool
• func ZkEventOk(e zk.Event) bool
• type Conn
  • func ConnectToZk(zkAddr string) (Conn, error)
  • func NewConn() Conn
  • func NewConnFromFile(filename string) Conn
  • func NewEtcdConn(zkAddr string) (Conn, error)
• type ElectorTask
• type MyZkConn
  • func (conn *MyZkConn) Seq2Str(seq int64) string
• type PooledEtcdClient
  • func (c *PooledEtcdClient) Close()
• type ZElector
  • func CreateElection(zconn Conn, zkPath string) ZElector
  • func (zm ZElector) Interrupt()
  • func (zm ZElector) Lock(desc string) error
  • func (zm ZElector) LockWithTimeout(wait time.Duration, desc string) (err error)
  • func (ze *ZElector) RunTask(task ElectorTask) error
  • func (zm ZElector) Unlock() error
• type ZLocker
  • func CreateMutex(zconn Conn, zkPath string) ZLocker
• type ZkError
  • func (ze *ZkError) Error() string

Constants

const (
	// PERM_DIRECTORY are default permissions for a node.
	PERM_DIRECTORY = zk.PermAdmin | zk.PermCreate | zk.PermDelete | zk.PermRead | zk.PermWrite
	// PERM_FILE allows a zk node to emulate file behavior by disallowing child nodes.
	PERM_FILE   = zk.PermAdmin | zk.PermRead | zk.PermWrite
	MagicPrefix = "zk"
)

const MAX_TTL = 365 * 24 * 60 * 60

Variables

var (
	// This error is returned by functions that wait for a result
	// when they are interrupted.
	ErrInterrupted = errors.New("zkutil: obtaining lock was interrupted")

	// This error is returned by functions that wait for a result
	// when the timeout value is reached.
	ErrTimeout = errors.New("zkutil: obtaining lock timed out")
)

Functions

func ChildrenRecursive

func ChildrenRecursive(zconn Conn, zkPath string) ([]string, error)

func CreateOrUpdate

func CreateOrUpdate(zconn Conn, zkPath, value string, flags int, aclv []zk.ACL, recursive bool) (pathCreated string, err error)

func CreatePidNode

func CreatePidNode(zconn Conn, zkPath string, contents string, done chan struct{}) error

Close the release channel when you want to clean up nicely.

func CreateRecursive

func CreateRecursive(zconn Conn, zkPath, value string, flags int, aclv []zk.ACL) (pathCreated string, err error)

Create a path and any pieces required, think mkdir -p. Intermediate znodes are always created empty.

func DefaultACLs

func DefaultACLs() []zk.ACL

func DefaultDirACLs

func DefaultDirACLs() []zk.ACL

func DefaultFileACLs

func DefaultFileACLs() []zk.ACL

func DeleteRecursive

func DeleteRecursive(zconn Conn, zkPath string, version int) error

func HasWildcard

func HasWildcard(path string) bool

func IsDirectory

func IsDirectory(aclv []zk.ACL) bool

IsDirectory returns if this node should be treated as a directory.

func NodeExists

func NodeExists(zconn Conn, zkPath string) (bool, error)

func ObtainQueueLock

func ObtainQueueLock(zconn Conn, zkPath string, wait time.Duration, interrupted chan struct{}) error

The lexically lowest node is the lock holder - verify that this path holds the lock. Call this queue-lock because the semantics are a hybrid. Normal zk locks make assumptions about sequential numbering that don't hold when the data in a lock is modified. if the provided 'interrupted' chan is closed, we'll just stop waiting and return an interruption error

func ResolveWildcards

func ResolveWildcards(zconn Conn, zkPaths []string) ([]string, error)

resolve paths like: /zk/nyc/vt/tablets/*/action /zk/global/vt/keyspaces/*/shards/*/action /zk/*/vt/tablets/*/action into real existing paths

If you send paths that don't contain any wildcard and don't exist, this function will return an empty array.

func ZkErrorEqual

func ZkErrorEqual(a, b error) bool

func ZkEventOk

func ZkEventOk(e zk.Event) bool

Types

type Conn

type Conn interface {
	Get(path string) (data []byte, stat zk.Stat, err error)
	GetW(path string) (data []byte, stat zk.Stat, watch <-chan zk.Event, err error)

	Children(path string) (children []string, stat zk.Stat, err error)
	ChildrenW(path string) (children []string, stat zk.Stat, watch <-chan zk.Event, err error)

	Exists(path string) (exist bool, stat zk.Stat, err error)
	ExistsW(path string) (exist bool, stat zk.Stat, watch <-chan zk.Event, err error)

	Create(path string, value []byte, flags int32, aclv []zk.ACL) (pathCreated string, err error)

	Set(path string, value []byte, version int32) (stat zk.Stat, err error)

	Delete(path string, version int32) (err error)

	Close()

	GetACL(path string) ([]zk.ACL, zk.Stat, error)
	SetACL(path string, aclv []zk.ACL, version int32) (zk.Stat, error)

	Seq2Str(seq int64) string
}

This interface is really close to the zookeeper connection interface. It uses the Stat interface defined here instead of the zookeeper.Stat structure for stats. Everything else is the same as in zookeeper. So refer to the zookeeper docs for the conventions used here (for instance, using -1 as version to specify any version)

func ConnectToZk

func ConnectToZk(zkAddr string) (Conn, error)

func NewConn

func NewConn() Conn

NewConn returns a fake zk.Conn implementation. Data is stored in memory, and there's a global connection lock for concurrent access.

func NewConnFromFile

func NewConnFromFile(filename string) Conn

NewConnFromFile returns a fake zk.Conn implementation, that is seeded with the json data extracted from the input file.

func NewEtcdConn

func NewEtcdConn(zkAddr string) (Conn, error)

type ElectorTask

type ElectorTask interface {
	Run() error
	Stop()
	// Return true if interrupted, false if it died of natural causes.
	// An interrupted task indicates that the election should stop.
	Interrupted() bool
}

ElectorTask is the interface for a task that runs essentially forever or until something bad happens. If a task must be stopped, it should be handled promptly - no second notification will be sent.

type MyZkConn

type MyZkConn struct {
	*zk.Conn
}

func (*MyZkConn) Seq2Str

func (conn *MyZkConn) Seq2Str(seq int64) string

type PooledEtcdClient

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

func (*PooledEtcdClient) Close

func (c *PooledEtcdClient) Close()

type ZElector

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

ZElector stores basic state for running an election.

func CreateElection

func CreateElection(zconn Conn, zkPath string) ZElector

CreateElection returns an initialized elector. An election is really a cycle of events. You are flip-flopping between leader and candidate. It's better to think of this as a stream of events that one needs to react to.

func (ZElector) Interrupt

func (zm ZElector) Interrupt()

Interrupt releases a lock that's held.

func (ZElector) Lock

func (zm ZElector) Lock(desc string) error

Lock returns nil when the lock is acquired.

func (ZElector) LockWithTimeout

func (zm ZElector) LockWithTimeout(wait time.Duration, desc string) (err error)

LockWithTimeout returns nil when the lock is acquired. A lock is held if the file exists and you are the creator. Setting the wait to zero makes this a nonblocking lock check.

FIXME(msolo) Disallow non-super users from removing the lock?

func (*ZElector) RunTask

func (ze *ZElector) RunTask(task ElectorTask) error

RunTask returns nil when the underlyingtask ends or the error it generated.

func (ZElector) Unlock

func (zm ZElector) Unlock() error

Unlock returns nil if the lock was successfully released. Otherwise, it is most likely a zk related error.

type ZLocker

type ZLocker interface {
	Lock(desc string) error
	LockWithTimeout(wait time.Duration, desc string) error
	Unlock() error
	Interrupt()
}

ZLocker is an interface for a lock that can fail.

func CreateMutex

func CreateMutex(zconn Conn, zkPath string) ZLocker

CreateMutex initializes an unaquired mutex. A mutex is released only by Unlock. You can clean up a mutex with delete, but you should be careful doing so.

type ZkError

type ZkError struct {
	Code error
	Op   string
	Path string
}

func (*ZkError) Error

func (ze *ZkError) Error() string