func (zk *zkf) RemoveService(service *service.Service) error {
	// acquire the service lock to prevent that service from being scheduled
	// as it is being deleted
	conn, err := zzk.GetLocalConnection(zzk.GeneratePoolPath(service.PoolID))
	if err != nil {
		return err
	}

	// remove the global list of all vhosts deployed
	if rootconn, err := zzk.GetLocalConnection("/"); err != nil {
		return err
	} else if err := zkservice.RemoveServiceVhosts(rootconn, service); err != nil {
		return err
	}

	// Ensure that the service's pool is locked for the duration
	finish := make(chan interface{})
	defer close(finish)
	if err := zkservice.EnsureServiceLock(nil, finish, conn); err != nil {
		return err
	}

	// FIXME: this may be a long-running operation, should we institute a timeout?
	return zkservice.RemoveService(conn, service.ID)
}

func getTestConn(c *C, path string) client.Connection {
	root, err := zzk.GetLocalConnection("/")
	c.Assert(err, IsNil)
	err = root.CreateDir(path)
	c.Assert(err, IsNil)
	conn, err := zzk.GetLocalConnection(path)
	c.Assert(err, IsNil)
	return conn
}

func (this *ControlPlaneDao) GetRunningServicesForHost(hostID string, services *[]dao.RunningService) error {
	// we initialize the data container to something here in case it has not been initialized yet
	*services = make([]dao.RunningService, 0)
	myHost, err := this.facade.GetHost(datastore.Get(), hostID)
	if err != nil {
		glog.Errorf("Unable to get host %v: %v", hostID, err)
		return err
	} else if myHost == nil {
		return nil
	}

	poolBasedConn, err := zzk.GetLocalConnection(zzk.GeneratePoolPath(myHost.PoolID))
	if err != nil {
		glog.Errorf("Error in getting a connection based on pool %v: %v", myHost.PoolID, err)
		return err
	}

	*services, err = zkservice.LoadRunningServicesByHost(poolBasedConn, hostID)
	if err != nil {
		glog.Errorf("zkservice.LoadRunningServicesByHost (conn: %+v host: %v) failed: %v", poolBasedConn, hostID, err)
		return err
	}

	return nil
}

func (this *ControlPlaneDao) GetRunningServices(request dao.EntityRequest, allRunningServices *[]dao.RunningService) error {
	// we initialize the data container to something here in case it has not been initialized yet
	*allRunningServices = make([]dao.RunningService, 0)
	allPools, err := this.facade.GetResourcePools(datastore.Get())
	if err != nil {
		glog.Error("runningservice.go failed to get resource pool")
		return err
	} else if allPools == nil || len(allPools) == 0 {
		return fmt.Errorf("no resource pools found")
	}

	for _, aPool := range allPools {
		poolBasedConn, err := zzk.GetLocalConnection(zzk.GeneratePoolPath(aPool.ID))
		if err != nil {
			glog.Error("runningservice.go Failed to get connection based on pool: %v", aPool.ID)
			return err
		}

		singlePoolRunningServices := []dao.RunningService{}
		singlePoolRunningServices, err = zkservice.LoadRunningServices(poolBasedConn)
		if err != nil {
			glog.Errorf("Failed GetAllRunningServices: %v", err)
			return err
		}

		for _, rs := range singlePoolRunningServices {
			*allRunningServices = append(*allRunningServices, rs)
		}
	}

	return nil
}

func (z *zkf) CheckRunningVHost(vhostName, serviceID string) error {
	rootBasedConnection, err := zzk.GetLocalConnection("/")
	if err != nil {
		return err
	}

	vr, err := zkregistry.VHostRegistry(rootBasedConnection)
	if err != nil {
		glog.Errorf("Error getting vhost registry: %v", err)
		return err
	}

	vhostEphemeralNodes, err := vr.GetVHostKeyChildren(rootBasedConnection, vhostName)
	if err != nil {
		glog.Errorf("GetVHostKeyChildren failed %v: %v", vhostName, err)
		return err
	}

	if len(vhostEphemeralNodes) > 0 {
		if vhost := vhostEphemeralNodes[0]; vhost.ServiceID != serviceID {
			err := fmt.Errorf("virtual host %s is already running under service %s", vhostName, vhost.ServiceID)
			return err
		}
	}

	return nil
}

func (this *ControlPlaneDao) GetRunningService(request dao.ServiceStateRequest, running *dao.RunningService) error {
	glog.V(3).Infof("ControlPlaneDao.GetRunningService: request=%v", request)
	*running = dao.RunningService{}

	serviceID := request.ServiceID
	poolID, err := this.facade.GetPoolForService(datastore.Get(), serviceID)
	if err != nil {
		glog.Errorf("Unable to get service %v: %v", serviceID, err)
		return err
	}

	poolBasedConn, err := zzk.GetLocalConnection(zzk.GeneratePoolPath(poolID))
	if err != nil {
		glog.Errorf("Error in getting a connection based on pool %v: %v", poolID, err)
		return err
	}

	if thisRunning, err := zkservice.LoadRunningService(poolBasedConn, request.ServiceID, request.ServiceStateID); err != nil {
		glog.Errorf("zkservice.LoadRunningService failed (conn: %+v serviceID: %v): %v", poolBasedConn, request.ServiceID, err)
		return err
	} else {
		if thisRunning != nil {
			*running = *thisRunning
		}
	}

	return nil
}

func (z *zkf) UpdateResourcePool(pool *pool.ResourcePool) error {
	conn, err := zzk.GetLocalConnection("/")
	if err != nil {
		return err
	}
	return zkservice.UpdateResourcePool(conn, pool)
}

func (z *zkf) UpdateHost(host *host.Host) error {
	conn, err := zzk.GetLocalConnection(zzk.GeneratePoolPath(host.PoolID))
	if err != nil {
		return err
	}
	return zkhost.UpdateHost(conn, host)
}

func (dfs *DistributedFilesystem) IsLocked() (bool, error) {
	conn, err := zzk.GetLocalConnection("/")
	if err != nil {
		return false, err
	}
	return zkservice.IsServiceLocked(conn)
}

func (z *zkf) RemoveResourcePool(poolID string) error {
	conn, err := zzk.GetLocalConnection("/")
	if err != nil {
		return err
	}
	return zkservice.RemoveResourcePool(conn, poolID)
}

func (z *zkf) RemoveVirtualIP(virtualIP *pool.VirtualIP) error {
	conn, err := zzk.GetLocalConnection(zzk.GeneratePoolPath(virtualIP.PoolID))
	if err != nil {
		return err
	}
	return zkvirtualip.RemoveVirtualIP(conn, virtualIP.IP)
}

func (this *ControlPlaneDao) GetRunningServicesForService(serviceID string, services *[]dao.RunningService) error {
	// we initialize the data container to something here in case it has not been initialized yet
	*services = make([]dao.RunningService, 0)

	poolID, err := this.facade.GetPoolForService(datastore.Get(), serviceID)
	if err != nil {
		glog.Errorf("Unable to get service %v: %v", serviceID, err)
		return err
	}

	poolBasedConn, err := zzk.GetLocalConnection(zzk.GeneratePoolPath(poolID))
	if err != nil {
		glog.Errorf("Error in getting a connection based on pool %v: %v", poolID, err)
		return err
	}

	svcs, err := zkservice.LoadRunningServicesByService(poolBasedConn, serviceID)
	if err != nil {
		glog.Errorf("LoadRunningServicesByService failed (conn: %+v serviceID: %v): %v", poolBasedConn, serviceID, err)
		return err
	}

	for _, svc := range svcs {
		*services = append(*services, svc)
	}

	return nil
}

func (zk *zkf) StopServiceInstance(poolID, hostID, stateID string) error {
	conn, err := zzk.GetLocalConnection(zzk.GeneratePoolPath(poolID))
	if err != nil {
		return err
	}

	return zkservice.StopServiceInstance(conn, hostID, stateID)
}

func (z *zkf) GetActiveHosts(poolID string, hosts *[]string) error {
	conn, err := zzk.GetLocalConnection(zzk.GeneratePoolPath(poolID))
	if err != nil {
		return err
	}
	*hosts, err = zkhost.GetActiveHosts(conn)
	return err
}

func (zk *zkf) UpdateService(service *service.Service) error {
	conn, err := zzk.GetLocalConnection(zzk.GeneratePoolPath(service.PoolID))
	if err != nil {
		return err
	}

	if err := zkservice.UpdateService(conn, service); err != nil {
		return err
	}

	rootconn, err := zzk.GetLocalConnection("/")
	if err != nil {
		return err
	}

	return zkservice.UpdateServiceVhosts(rootconn, service)
}

func (zk *zkf) WaitService(service *service.Service, state service.DesiredState, cancel <-chan interface{}) error {
	conn, err := zzk.GetLocalConnection(zzk.GeneratePoolPath(service.PoolID))
	if err != nil {
		return err
	}

	return zkservice.WaitService(cancel, conn, service.ID, state)
}

func TestClient(t *testing.T) {
	t.Skipf("Test cluster is not set up properly")
	zookeeper.EnsureZkFatjar()
	basePath := ""
	tc, err := zklib.StartTestCluster(1, nil, nil)
	if err != nil {
		t.Fatalf("could not start test zk cluster: %s", err)
	}
	defer os.RemoveAll(tc.Path)
	defer tc.Stop()
	time.Sleep(time.Second)

	servers := []string{fmt.Sprintf("127.0.0.1:%d", tc.Servers[0].Port)}

	dsnBytes, err := json.Marshal(zookeeper.DSN{Servers: servers, Timeout: time.Second * 15})
	if err != nil {
		t.Fatal("unexpected error creating zk DSN: %s", err)
	}
	dsn := string(dsnBytes)
	zClient, err := client.New("zookeeper", dsn, basePath, nil)

	zzk.InitializeLocalClient(zClient)

	conn, err := zzk.GetLocalConnection("/")
	if err != nil {
		t.Fatal("unexpected error getting connection")
	}

	h := host.New()
	h.ID = "nodeID"
	h.IPAddr = "192.168.1.5"
	h.PoolID = "default1"
	defer func(old func(string, os.FileMode) error) {
		mkdirAll = old
	}(mkdirAll)
	dir, err := ioutil.TempDir("", "serviced_var_")
	if err != nil {
		t.Fatalf("could not create tempdir: %s", err)
	}
	defer os.RemoveAll(dir)
	c, err := NewClient(h, dir)
	if err != nil {
		t.Fatalf("unexpected error creating client: %s", err)
	}
	defer c.Close()
	time.Sleep(time.Second * 5)

	// therefore, we need to check that the client was added under the pool from root
	nodePath := fmt.Sprintf("/storage/clients/%s", h.IPAddr)
	glog.Infof("about to check for %s", nodePath)
	if exists, err := conn.Exists(nodePath); err != nil {
		t.Fatalf("did not expect error checking for existence of %s: %s", nodePath, err)
	} else {
		if !exists {
			t.Fatalf("could not find %s", nodePath)
		}
	}
}

func (zk *zkf) GetServiceStates(poolID string, states *[]servicestate.ServiceState, serviceIDs ...string) error {
	conn, err := zzk.GetLocalConnection(zzk.GeneratePoolPath(poolID))
	if err != nil {
		return err
	}

	*states, err = zkservice.GetServiceStates(conn, serviceIDs...)
	return err
}

func (c *Controller) unregisterVhosts() {
	conn, err := zzk.GetLocalConnection("/")
	if err != nil {
		return
	}
	for _, path := range c.vhostZKPaths {
		glog.V(1).Infof("controller shutdown deleting vhost %v", path)
		conn.Delete(path)
	}
}

func (t *ZZKTest) TestActionListener_Listen(c *C) {
	conn, err := zzk.GetLocalConnection("/")
	c.Assert(err, IsNil)

	handler := &TestActionHandler{
		ResultMap: map[string]ActionResult{
			"success": ActionResult{2 * time.Second, []byte("success"), nil},
			"failure": ActionResult{time.Second, []byte("message failure"), fmt.Errorf("failure")},
		},
	}

	c.Log("Start actions and shutdown")
	shutdown := make(chan interface{})
	done := make(chan interface{})

	listener := NewActionListener(handler, "test-host-1")
	go func() {
		zzk.Listen(shutdown, make(chan error, 1), conn, listener)
		close(done)
	}()

	// send actions
	var wg sync.WaitGroup

	sendAction := func(dockerID string, command []string) {
		id, err := SendAction(conn, &Action{
			HostID:   listener.hostID,
			DockerID: dockerID,
			Command:  command,
		})
		c.Assert(err, IsNil)

		// There *might* be a race condition here if the node is processed before
		// we acquire the event data (see duration timeouts above)
		event, err := conn.GetW(actionPath(listener.hostID, id), &Action{})
		c.Assert(err, IsNil)

		wg.Add(1)
		go func() {
			defer wg.Done()
			<-event
		}()
		return
	}

	sendAction("success", []string{"do", "some", "command"})
	sendAction("failure", []string{"do", "some", "bad", "command"})

	c.Log("Waiting for actions to complete")
	wg.Wait()
	c.Log("Actions completed")
	close(shutdown)
	<-done
}