zk

README

Go Zookeeper client

Implements the Zookeeper protocol in Go.

Status

Early Alpha - WIP

Docs: http://godoc.org/github.com/streadway/zk

Code: http://github.com/streadway/zk

Contributing

Fork, write tests, fail tests, fix stuff, pass tests, go fmt, go vet, open a pull request.

A zookeeper server should be running on 127.0.0.1:2181 with a node /test.

TODO

• Reconnection or other kind of connection management
• SetAuth during connection establishment
• SASL (blocked on a kerberos domain setup to test)
• Multi transactions (blocked on finding a good API)
• Acceptance tests for flakey connections
• Acceptance tests for server partitions

License

BSD 2 clause - see LICENSE for more details.

Documentation

Overview

Native client for the Zookeeper protocol.

Stability

Alpha. This package is still under development, the APIs may change.

Sessions

A Session represents an active connection to a Zookeeper server and an identity in a Zookeeper ensemble. Sessions can be moved between connections within an ensemble by using Dial with a previous Session's config.

// Move or reconnect a session
next, err := Dial(prev.Config)

Watches from the previous connection will be moved to the next connection. After a successful Dial, previous sessions should be considered disconnected.

Watches

The Zookeeper ensemble ensures that watches are totally ordered against all responses. Since this library presents a synchronous interface to a Zookeeper server, it maintains order of watches relative to other watches.

A change in data or children is observable before the watch fires. This means that when a watch channel receives an event, the change can be observed with a separate call.

Errors

Each API method can return one of many standard errors. These errors communicate the state of the session in the ensemble. It is up to the application to respond to these errors specifically to the application's needs.

ErrConnection --> Disconnected
ErrExpired -----> Expired
ErrAuth --------> Unauthenticated

When disconnected, the outstanding requests may or may not have been committed to the ensemble. Commands are only committed when no error is returned so it is up to the application to determine the strategy for repeating the command. Applications should Dial a new connection with the same session config to retry the command that caused the disconnect.

When expired, a new session must be established. Any ephemeral nodes will be deleted and must be re-created by the new session in a new connection.

Dial may return ErrAuth. ErrAuth indicates the session is expired and the connection closed. The application must use different authentication credentials and Dial again.

Index

• Constants
• Variables
• func DialRandom(conf *Config) (net.Conn, error)
• type ACL
• type Access
• type Config
• type CreateType
• type Event
• type EventType
• type Permission
• type Session
  • func Dial(conf *Config) (*Session, error)
  • func (zk *Session) Children(path string, watch chan<- Event) ([]string, Stat, error)
  • func (zk *Session) Close() error
  • func (zk *Session) Create(path string, data []byte, types CreateType, acl ACL) (string, error)
  • func (zk *Session) Delete(path string, version int32) error
  • func (zk *Session) Exists(path string, watch chan<- Event) (bool, Stat, error)
  • func (zk *Session) Get(path string, watch chan<- Event) ([]byte, Stat, error)
  • func (zk *Session) GetAcl(path string) (ACL, Stat, error)
  • func (zk *Session) Set(path string, data []byte, version int32) (Stat, error)
  • func (zk *Session) SetAcl(path string, acl ACL, version int32) (Stat, error)
  • func (zk *Session) Sync(path string) (string, error)
• type Stat
• type Watches

Constants

const (
	// Default session timeout
	Timeout = 4 * time.Second

	// Default address when Config.Addrs is empty
	Address = "127.0.0.1:2181"
)

Variables

var (
	AclOpen     = ACL{Access{PermAll, "world", "anyone"}}
	AclReadOnly = ACL{Access{PermRead, "world", "anyone"}}
)

Commonly used ACLs for nodes

var (
	ErrExists     = errors.New("zk: node already exists")   // Create
	ErrInvalidAcl = errors.New("zk: ACL is not supported")  // Create
	ErrVersion    = errors.New("zk: node version differs")  // Set/Delete
	ErrNoNode     = errors.New("zk: no node found at path") // Exists/Get

	ErrExpired    = errors.New("zk: session expired")         // All
	ErrConnection = errors.New("zk: connection lost")         // All
	ErrAuth       = errors.New("zk: connection unauthorized") // All

	ErrProtocol = errors.New("zk: protocol error")
)

Functions

func DialRandom

func DialRandom(conf *Config) (net.Conn, error)

DialRandom dials a normally distributed random selection from the Addrs with a connection timeout such that every address could be attempted before the session is expired.

Types

type ACL

type ACL []Access

ACL is an Access Control List used in Create, SetAcl and GetAcl

type Access

type Access struct {
	Perms    Permission // Bitmask of Permissions
	Scheme   string     // one of "world" "auth" "digest" "host" "ip"
	Identity string     // Scheme specific identity like 127.0.0.1/32
}

Access represents permissions bitmask for an identity under a scheme

type Config

type Config struct {
	// Host:Port pairs of a subset of servers in the ensemble
	Addrs []string

	// net.Conn factory - default is DialRandom
	Dial func(conf *Config) (net.Conn, error)

	Id       int64  // Id is the session identity useful for server correlation
	Password []byte // Password authenticates this connection to the session id

	Zxid    int64         // Zxid is the last transaction the client has seen
	Timeout time.Duration // Duration until a disconnect expires

	DataWatches   Watches
	ChildWatches  Watches
	ExistsWatches Watches
}

Config captures the connection and session state for dialing or re-dialing a Zookeeper server.

type CreateType

type CreateType int32

CreateType specifies the type of the node to Create.

const (
	CreatePersistent CreateType = 0 // Node remains after session expires
	CreateEphemeral  CreateType = 1 // Node deletes when the creator session expires
	CreateSequence   CreateType = 2 // Sequence number is appended to the node path
)

type Event

type Event struct {
	Type EventType
	Path string
}

Event contains the type of change, and path of that change. It will be delivered once per watch.

type EventType

type EventType int32

EventType is one of 4 node events emitted by the server

const (
	Created EventType = 1 // From Exists, Get
	Deleted EventType = 2 // From Exists, Get
	Changed EventType = 3 // From Exists, Get
	Child   EventType = 4 // From Children
)

type Permission

type Permission int32

Permission is a bitmask of permissions that apply to a node for a scheme's identity.

const (
	PermRead Permission = 1 << iota
	PermWrite
	PermCreate
	PermDelete
	PermAdmin

	PermAll Permission = (1 << iota) - 1
)

type Session

type Session struct {
	*Config
	// contains filtered or unexported fields
}

func Dial

func Dial(conf *Config) (*Session, error)

Dial connects then either transfers or establishes a new session to a Zookeeper ensemble.

func (*Session) Children

func (zk *Session) Children(path string, watch chan<- Event) ([]string, Stat, error)

Children returns the paths of the children and node statistics for the path given. If watch is provided, it will be fired once when the number of children at this path changes.

Children returns ErrNoNode and does not set a watch when the node does not exist.

func (*Session) Close

func (zk *Session) Close() error

Close tells the server it should expire this session and closes the connection.

func (*Session) Create

func (zk *Session) Create(path string, data []byte, types CreateType, acl ACL) (string, error)

Create makes a node if it doesn't already exist, setting the data and ACL.

Create returns the path created which will differ from the path requested for Sequence nodes.

Create returns ErrExists if the node already exists.

func (*Session) Delete

func (zk *Session) Delete(path string, version int32) error

Delete removes the node if the version matches.

func (*Session) Exists

func (zk *Session) Exists(path string, watch chan<- Event) (bool, Stat, error)

Exists returns the stat of a node if it exists and optionally sets a watch that will fire when node changes.

func (*Session) Get

func (zk *Session) Get(path string, watch chan<- Event) ([]byte, Stat, error)

Get returns the data and node statistics for the path. If watch is provided, it will be fired once when the data changes at this path.

Get returns ErrNoNode and does not set a watch when the node does not exist.

func (*Session) GetAcl

func (zk *Session) GetAcl(path string) (ACL, Stat, error)

GetAcl returns the last seen access control list and the stat for the given path.

func (*Session) Set

func (zk *Session) Set(path string, data []byte, version int32) (Stat, error)

Set changes the data and updates the stat attributes at the provided path.

Set returns ErrNoNode and does not set a watch when the node does not exist.

func (*Session) SetAcl

func (zk *Session) SetAcl(path string, acl ACL, version int32) (Stat, error)

SetAcl replaces the access control list for the given path and version.

func (*Session) Sync

func (zk *Session) Sync(path string) (string, error)

Sync reaches consensus and commits the data of the node at the path to disk.

type Stat

type Stat struct {
	proto.Stat // FIXME(streadway) don't expose proto package
}

type Watches

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

Watches synchronizes access to an internal map of paths to event chans. Event chans receive a single Event and are removed from this map when the watch at that path is triggered.