Control.Distributed.Process.Zookeeper

Provides service and node discovery for Cloud Haskell applications using a Zookeeper cluster for name registration, lookups and leader election. Uses the hzk bindings to the Zookeeper multi-threaded C library.

Objectives and features:

• Compatible with distributed-process-p2p API - can work as a drop-in replacement.
• No dependencies beyond those already included by distributed-process, hzk and network-transport-tcp.
• Automatic registration of local names to Zookeeper.
• Global singleton processes with leader election and re-elections on leader exit.

Initialization

bootstrap

Arguments

:: HostName

Hostname or IP this Cloud Haskell node will listen on.

-> ServiceName

Port or port name this node will listen on.

-> String

The Zookeeper endpoint(s) -- comma separated list of host:port

-> RemoteTable

Cloud Haskell RemoteTable to use in the new node.

-> Process ()

Process computation to run in the new node.

-> IO ()

Create a new Cloud Haskell node on the provided IP/Port and start a Zookeeper-backed controller process (zkController) with a default configuration connected to the provided Zookeeper server list. Finally execute the supplied Process computation.

bootstrap = bootstrapWith defaultConfig

bootstrapWith

Arguments

:: Config

controller configuration

-> HostName

Hostname or IP this Cloud Haskell node will listen on.

-> ServiceName

Port or port name this node will listen on.

-> String

The Zookeeper endpoint(s) -- comma separated list of host:port

-> RemoteTable

Cloud Haskell RemoteTable to use in the new node.

-> Process ()

Process computation to run in the new node.

-> IO ()

Create a new Cloud Haskell node on the provided IP/Port and start a Zookeeper-backed controller process (zkController) connected to the provided Zookeeper server list and finally execute the supplied Process computation.

zkController

Arguments

:: String

The Zookeeper endpoint(s) -- comma separated list of host:port

-> Process ()

Run a Zookeeper service process, and installs an MXAgent to automatically register all local names in Zookeeper using defaultConfig.

zkController = zkControllerWith defaultConfig

zkControllerWith

Arguments

:: Config

-> String

The Zookeeper endpoint(s) -- comma separated list of host:port

-> Process ()

As zkController but accept Config options rather than assuming defaults.

API

registerZK :: String -> ProcessId -> Process (Either String ())

Register a name and pid as a service in Zookeeper. The controller will monitor the pid and remove its child node from Zookeeper when it exits.

Names will be registered at "/distributed-process/services/<name>/<pid>"

Note: By default all locally registered names (using register) will be registered in Zookeeper under the same name by an MxAgent process. Use this function if you want to register an anonymous pid or use a different name than is registered with the local Process, or when you are using a registerPrefix to exclude the automatic registration (see Config).

getCapable :: String -> Process [ProcessId]

Returns list of pids registered with the service name.

Results are cached by the controller until they are invalidated by subsequent changes to the service node in Zookeeper, which is communicated through a Watcher. Data will be fetched from Zookeeper only when it is changed and then requested again.

nsendCapable :: Serializable a => String -> a -> Process ()

Broadcast a message to all pids registered with a particular service name.

registerCandidate :: String -> Process () -> Process (Either String ProcessId)

Register a candidate process for election of a single process associated to the given global name, and returns the ProcessId of the elected global (which may or may not be on the local node). The Process () argument is only evaluated if this node ends up being the elected host for the global. Calling this function subsequently on the same node for the same name will replace the current candidate computation with the new one.

whereisGlobal :: String -> Process (Maybe ProcessId)

Find a registered global by name - see registerCandidate.

Compatibility

getPeers :: Process [NodeId]

Get a list of nodes advertised in Zookeeper. These are registered when zkController starts in path "/distributed-process/controllers/<pid>".

Note: this is included for API compatibility with distributed-process-p2p but its usage would suggest discovery patterns that could be made more efficient when using Zookeeper - i.e. just use getCapable.

nsendPeers :: Serializable a => String -> a -> Process ()

Broadcast a message to a specific service on all registered nodes.

Note: this is included for API compatibility with distributed-process-p2p but its usage would suggest discovery patterns that could be made more efficient when using Zookeeper - i.e. just use nfSendCapable to nfSend a broadcast directly to the registered process on each node.

Config

data Config

Constructors

Config

Fields

registerPrefix :: String

Only register locally registered process names with zookeeper if the name begins with the given prefix. Default is "" which will register every locally registered process in the Zookeeper services node.

logTrace :: String -> Process ()

An operation that will be called for trace level logging. defaultConfig uses nolog.

logError :: String -> Process ()

An operation that will be called for error logging. defaultConfig uses say.

zLogLevel :: ZLogLevel

The log level for the C Zookeper library. defaultConfig uses ZLogWarn.

acl :: AclList

The ACL to use for every node - see hzk documentation for AclList. Note that if your nodes do not connect with the same identity, every node will need at least Read permission to all nodes created by this package.

credentials :: Maybe (Scheme, ByteString)

Credentials for Zookeeper, see hzk addAuth for details.

defaultConfig :: Config

By default all local names are registered with zookeeper, and only error messages are logged through say.

defaultConfig = Config {
      registerPrefix = ""
    , logTrace = nolog
    , logError = say . ("[C.D.P.Zookeeper: ERROR] - " ++)
    , zLogLevel = ZK.ZLogWarn
    , acl = OpenAclUnsafe
    , credentials = Nothing
    }

Utility

nolog :: String -> Process ()

A no-op that can be used for either of the loggers in Config. Because no actual I/O is performed, it fully evaluates the message so thunks do not build up.

sayTrace :: String -> Process ()

Simple formatter for trace output through say.

waitController :: Int -> Process (Maybe ())

Wait for zkController to startup and register iteself. This is only useful if you are *not* using a bootstrap function to start your node, but rather starting the node yourself and using one of the zkController functions.