func MonitorResourcePool(shutdown <-chan interface{}, conn client.Connection, poolID string) <-chan *pool.ResourcePool {
	monitor := make(chan *pool.ResourcePool)
	go func() {
		defer close(monitor)
		if err := zzk.Ready(shutdown, conn, poolpath(poolID)); err != nil {
			glog.V(2).Infof("Could not watch pool %s: %s", poolID, err)
			return
		}
		for {
			var node PoolNode
			event, err := conn.GetW(poolpath(poolID), &node)
			if err != nil {
				glog.V(2).Infof("Could not get pool %s: %s", poolID, err)
				return
			}

			select {
			case monitor <- node.ResourcePool:
			case <-shutdown:
				return
			}

			select {
			case <-event:
			case <-shutdown:
				return
			}
		}
	}()
	return monitor
}

// wait waits for an individual service state to reach its desired state
func wait(shutdown <-chan interface{}, conn client.Connection, serviceID, stateID string, dstate service.DesiredState) error {
	for {
		var node ServiceStateNode
		event, err := conn.GetW(servicepath(serviceID, stateID), &node)
		if err == client.ErrNoNode {
			// if the node no longer exists, then there is nothing to watch, so we are done
			return nil
		} else if err != nil {
			glog.Errorf("Got an error while looking for %s (%s): %s", stateID, serviceID, err)
			return err
		}

		switch dstate {
		case service.SVCStop:
			// pass through, because the node needs to be deleted to be considered Stopped
		case service.SVCRun, service.SVCRestart:
			if node.IsRunning() {
				// instance reached the desired state
				return nil
			}
		case service.SVCPause:
			if node.IsPaused() {
				// instance reached the desired state
				return nil
			}
		}

		// wait for something to change on the node or shutdown
		select {
		case <-event:
		case <-shutdown:
			return zzk.ErrShutdown
		}
	}
}

// Ready waits for a node to be available for watching
func Ready(shutdown <-chan interface{}, conn client.Connection, p string) error {
	if exists, err := PathExists(conn, p); err != nil {
		return err
	} else if exists {
		return nil
	} else if p == "/" || p == "." {
		return fmt.Errorf("base path not found")
	}

	for {
		if err := Ready(shutdown, conn, path.Dir(p)); err != nil {
			return err
		}

		_, event, err := conn.ChildrenW(path.Dir(p))
		if err != nil {
			return err
		}

		if exists, err := PathExists(conn, p); err != nil {
			return err
		} else if exists {
			return nil
		}

		select {
		case <-event:
			// pass
		case <-shutdown:
			return ErrShutdown
		}
	}
}

func (r *registryType) watchItem(conn client.Connection, path string, nodeType client.Node, cancel <-chan bool, processNode func(conn client.Connection,
	node client.Node), errorHandler WatchError) error {
	exists, err := zzk.PathExists(conn, path)
	if err != nil {
		return err
	}
	if !exists {
		return client.ErrNoNode
	}
	for {
		event, err := conn.GetW(path, nodeType)
		if err != nil {
			glog.Errorf("Could not watch %s: %s", path, err)
			defer errorHandler(path, err)
			return err
		}
		processNode(conn, nodeType)
		//This blocks until a change happens under the key
		select {
		case ev := <-event:
			glog.V(2).Infof("watch event %+v at path: %s", ev, path)
		case <-cancel:
			return nil
		}

	}
	return nil
}

// IsServiceLocked verifies whether services are locked
func IsServiceLocked(conn client.Connection) (bool, error) {
	locks, err := conn.Children(zkServiceLock)
	if err == client.ErrNoNode {
		return false, nil
	}
	return len(locks) > 0, err
}

// PathExists verifies if a path exists and does not raise an exception if the
// path does not exist
func PathExists(conn client.Connection, p string) (bool, error) {
	exists, err := conn.Exists(p)
	if err == client.ErrNoNode {
		return false, nil
	}
	return exists, err
}

// WatchServiceLock waits for a service lock to be enabled/disabled
func WaitServiceLock(shutdown <-chan interface{}, conn client.Connection, enabled bool) error {
	for {
		// make sure the parent exists
		if err := zzk.Ready(shutdown, conn, zkServiceLock); err != nil {
			return err
		}

		// check if the lock is enabled
		nodes, event, err := conn.ChildrenW(zkServiceLock)
		if err != nil {
			return err
		}

		// check that the states match
		if locked := len(nodes) > 0; locked == enabled {
			return nil
		}

		// wait to reach the desired state or shutdown
		select {
		case <-event:
			// pass
		case <-shutdown:
			return zzk.ErrTimeout
		}
	}
}

// LoadRunningServicesByHost returns a slice of RunningServices given a host(s)
func LoadRunningServicesByHost(conn client.Connection, hostIDs ...string) ([]dao.RunningService, error) {
	var rss []dao.RunningService = make([]dao.RunningService, 0)
	for _, hostID := range hostIDs {
		if exists, err := zzk.PathExists(conn, hostpath(hostID)); err != nil {
			return nil, err
		} else if !exists {
			continue
		}

		stateIDs, err := conn.Children(hostpath(hostID))
		if err != nil {
			return nil, err
		}
		for _, ssID := range stateIDs {
			var hs HostState
			if err := conn.Get(hostpath(hostID, ssID), &hs); err != nil {
				return nil, err
			}

			rs, err := LoadRunningService(conn, hs.ServiceID, hs.ServiceStateID)
			if err != nil {
				return nil, err
			}

			rss = append(rss, *rs)
		}
	}
	return rss, nil
}

// StopServiceInstance stops a host state instance
func StopServiceInstance(conn client.Connection, hostID, stateID string) error {
	// verify that the host is active
	var isActive bool
	hostIDs, err := GetActiveHosts(conn)
	if err != nil {
		glog.Warningf("Could not verify if host %s is active: %s", hostID, err)
		isActive = false
	} else {
		for _, hid := range hostIDs {
			if isActive = hid == hostID; isActive {
				break
			}
		}
	}
	if isActive {
		// try to stop the instance nicely
		return updateInstance(conn, hostID, stateID, func(hsdata *HostState, _ *ss.ServiceState) {
			glog.V(2).Infof("Stopping service instance via %s host %s", stateID, hostID)
			hsdata.DesiredState = int(service.SVCStop)
		})
	} else {
		// if the host isn't active, then remove the instance
		var hs HostState
		if err := conn.Get(hostpath(hostID, stateID), &hs); err != nil {
			glog.Errorf("Could not look up host instance %s on host %s: %s", stateID, hostID, err)
			return err
		}
		return removeInstance(conn, hs.ServiceID, hs.HostID, hs.ServiceStateID)
	}
}

//Set node to the key in registry.  Returns the path of the node in the registry
func (r *registryType) setItem(conn client.Connection, key string, nodeID string, node client.Node) (string, error) {
	if err := r.ensureKey(conn, key); err != nil {
		return "", err
	}

	//TODO: make ephemeral
	path := r.getPath(key, nodeID)

	exists, err := zzk.PathExists(conn, path)
	if err != nil {
		return "", err
	}

	if exists {
		glog.V(3).Infof("Set to %s: %#v", path, node)
		epn := EndpointNode{}
		if err := conn.Get(path, &epn); err != nil {
			return "", err
		}
		node.SetVersion(epn.Version())
		if err := conn.Set(path, node); err != nil {
			return "", err
		}
	} else {
		if addPath, err := r.addItem(conn, key, nodeID, node); err != nil {
			return "", err
		} else {
			path = addPath
		}
		glog.V(3).Infof("Add to %s: %#v", path, node)
	}
	return path, nil
}

func watch(conn client.Connection, path string, cancel <-chan bool, processChildren ProcessChildrenFunc, errorHandler WatchError) error {
	exists, err := zzk.PathExists(conn, path)
	if err != nil {
		return err
	}
	if !exists {
		return client.ErrNoNode
	}
	for {
		glog.V(1).Infof("watching children at path: %s", path)
		nodeIDs, event, err := conn.ChildrenW(path)
		glog.V(1).Infof("child watch for path %s returned: %#v", path, nodeIDs)
		if err != nil {
			glog.Errorf("Could not watch %s: %s", path, err)
			defer errorHandler(path, err)
			return err
		}
		processChildren(conn, path, nodeIDs...)
		select {
		case ev := <-event:
			glog.V(1).Infof("watch event %+v at path: %s", ev, path)
		case <-cancel:
			glog.V(1).Infof("watch cancel at path: %s", path)
			return nil
		}
	}
	glog.V(1).Infof("no longer watching children at path: %s", path)
	return nil
}

func UpdateResourcePool(conn client.Connection, pool *pool.ResourcePool) error {
	var node PoolNode
	if err := conn.Get(poolpath(pool.ID), &node); err != nil {
		return err
	}
	node.ResourcePool = pool
	return conn.Set(poolpath(pool.ID), &node)
}

// GetItem gets EndpointNode at the given path.
func (ar *EndpointRegistry) GetItem(conn client.Connection, path string) (*EndpointNode, error) {
	var ep EndpointNode
	if err := conn.Get(path, &ep); err != nil {
		glog.Errorf("Could not get EndpointNode at %s: %s", path, err)
		return nil, err
	}
	return &ep, nil
}

func RemoveVirtualIP(conn client.Connection, ip string) error {
	glog.V(1).Infof("Removing virtual ip from zookeeper: %s", vippath(ip))
	err := conn.Delete(vippath(ip))
	if err == nil || err == client.ErrNoNode {
		return nil
	}
	return err
}

//GetItem gets VhostEndpoint at the given path.
func (vr *VhostRegistry) GetItem(conn client.Connection, path string) (*VhostEndpoint, error) {
	var vep VhostEndpoint
	if err := conn.Get(path, &vep); err != nil {
		glog.Infof("Could not get vhost endpoint at %s: %s", path, err)
		return nil, err
	}
	return &vep, nil
}

// getAllRemoteHostsFromZookeeper retrieves a list of remote storage clients from zookeeper
func getAllRemoteHostsFromZookeeper(conn client.Connection, storageClientsPath string) ([]string, error) {
	clients, err := conn.Children(storageClientsPath)
	if err != nil {
		glog.Errorf("unable to retrieve list of DFS remote hosts: %s", err)
		return []string{}, err
	}

	glog.V(4).Infof("DFS remote IPs: %+v", clients)
	return clients, nil
}

func AddVirtualIP(conn client.Connection, virtualIP *pool.VirtualIP) error {
	var node VirtualIPNode
	path := vippath(virtualIP.IP)

	glog.V(1).Infof("Adding virtual ip to zookeeper: %s", path)
	if err := conn.Create(path, &node); err != nil {
		return err
	}
	node.VirtualIP = virtualIP
	return conn.Set(path, &node)
}

// StopService schedules a service to stop
func StopService(conn client.Connection, serviceID string) error {
	glog.Infof("Scheduling service %s to stop", serviceID)
	var node ServiceNode
	path := servicepath(serviceID)

	if err := conn.Get(path, &node); err != nil {
		return err
	}
	node.Service.DesiredState = int(service.SVCStop)
	return conn.Set(path, &node)
}

// getStatus computes the status of a service state
func getStatus(conn client.Connection, state *servicestate.ServiceState) (dao.Status, error) {
	var status dao.Status

	// Set the state based on the service state object
	if !state.IsRunning() {
		status = dao.Stopped
	} else if state.IsPaused() {
		status = dao.Paused
	} else {
		status = dao.Running
	}

	// Set the state based on the host state object
	var hostState HostState
	if err := conn.Get(hostpath(state.HostID, state.ID), &hostState); err != nil && err != client.ErrNoNode {
		return dao.Status{}, err
	}

	if hostState.DesiredState == int(service.SVCStop) {
		switch status {
		case dao.Running, dao.Paused:
			status = dao.Stopping
		case dao.Stopped:
			// pass
		default:
			return dao.Status{}, ErrUnknownState
		}
	} else if hostState.DesiredState == int(service.SVCRun) {
		switch status {
		case dao.Stopped:
			status = dao.Starting
		case dao.Paused:
			status = dao.Resuming
		case dao.Running:
			// pass
		default:
			return dao.Status{}, ErrUnknownState
		}
	} else if hostState.DesiredState == int(service.SVCPause) {
		switch status {
		case dao.Running:
			status = dao.Pausing
		case dao.Paused, dao.Stopped:
			// pass
		default:
			return dao.Status{}, ErrUnknownState
		}
	} else {
		return dao.Status{}, ErrUnknownState
	}

	return status, nil
}

// CreateEndpointRegistry creates the endpoint registry and returns the EndpointRegistry type
// This is created in the leader, most other calls will just get that one
func CreateEndpointRegistry(conn client.Connection) (*EndpointRegistry, error) {
	path := zkEndpointsPath()
	if exists, err := zzk.PathExists(conn, path); err != nil {
		return nil, err
	} else if !exists {
		if err := conn.CreateDir(path); err != nil {
			glog.Errorf("error with CreateDir(%s) %+v", path, err)
			return nil, err
		}
	}
	return &EndpointRegistry{registryType{getPath: zkEndpointsPath, ephemeral: true}}, nil
}