func (f *FakePodStore) ListPods(podPrefix kp.PodPrefix, hostname types.NodeName) ([]kp.ManifestResult, time.Duration, error) {
	f.podLock.Lock()
	defer f.podLock.Unlock()
	res := make([]kp.ManifestResult, 0)
	for key, manifest := range f.podResults {
		if key.podPrefix == podPrefix && key.hostname == hostname {
			// TODO(mpuncel) make ManifestResult not contain the path, it's silly to have to do things like this
			path := path.Join(string(podPrefix), hostname.String(), string(manifest.ID()))
			uniqueKey, err := kp.PodUniqueKeyFromConsulPath(path)
			if err != nil {
				return nil, 0, err
			}

			res = append(res, kp.ManifestResult{
				Manifest: manifest,
				PodLocation: types.PodLocation{
					Node:  hostname,
					PodID: manifest.ID(),
				},
				PodUniqueKey: uniqueKey,
			})
		}
	}
	return res, 0, nil
}

func (s *fakeKpStore) Pod(podPrefix kp.PodPrefix, nodeName types.NodeName, podID types.PodID) (
	manifest.Manifest, time.Duration, error) {
	key := path.Join(string(podPrefix), nodeName.String(), podID.String())
	if manifest, ok := s.manifests[key]; ok {
		return manifest, 0, nil
	}
	return nil, 0, pods.NoCurrentManifest
}

// NewLegacyP2RM is a constructor for the P2RM type which configures it to
// remove a "legacy" pod. It will generate the storage types based on its
// api.Client argument
func NewLegacyP2RM(client consulutil.ConsulClient, podName types.PodID, nodeName types.NodeName, labeler labels.ApplicatorWithoutWatches) *P2RM {
	rm := &P2RM{}
	rm.LabelID = path.Join(nodeName.String(), podName.String())
	rm.PodID = podName
	rm.NodeName = nodeName
	rm.PodUniqueKey = ""
	rm.configureStorage(client, labeler)
	return rm
}

// NewP2RM is a constructor for the P2RM type. It will generate the necessary
// storage types based on its api.Client argument
func NewP2RM(client consulutil.ConsulClient, podName string, nodeName types.NodeName) *P2RM {
	rm := &P2RM{}
	rm.Client = client
	rm.Store = kp.NewConsulStore(client)
	rm.RCStore = rcstore.NewConsul(client, 5)
	rm.Labeler = labels.NewConsulApplicator(client, 3)
	rm.LabelID = path.Join(nodeName.String(), podName)
	rm.PodName = podName
	rm.NodeName = nodeName

	return rm
}

func nodePath(podPrefix PodPrefix, nodeName types.NodeName) (string, error) {
	// hook tree is an exception to the rule because they are not scheduled
	// by host, and it is valid to want to watch for them agnostic to pod
	// id. There are plans to deploy hooks by host at which time this
	// exception can be removed
	if podPrefix == HOOK_TREE {
		nodeName = ""
	} else {
		if nodeName == "" {
			return "", util.Errorf("nodeName not specified when computing host path")
		}
	}

	return path.Join(string(podPrefix), nodeName.String()), nil
}

func (r *replication) queryReality(node types.NodeName) (manifest.Manifest, error) {
	for {
		select {
		case r.concurrentRealityRequests <- struct{}{}:
			man, _, err := r.store.Pod(kp.REALITY_TREE, node, r.manifest.ID())
			<-r.concurrentRealityRequests
			return man, err
		case <-time.After(5 * time.Second):
			r.logger.Infof("Waiting on concurrentRealityRequests for pod: %s/%s", node.String(), r.manifest.ID())
		case <-time.After(1 * time.Minute):
			err := util.Errorf("Timed out while waiting for reality query rate limit")
			r.logger.Error(err)
			return nil, err
		}
	}
}

// matches podstore.consulStore signature
func (c Client) Schedule(manifest manifest.Manifest, node types.NodeName) (types.PodUniqueKey, error) {
	manifestBytes, err := manifest.Marshal()
	if err != nil {
		return "", util.Errorf("Could not marshal manifest: %s", err)
	}

	req := &podstore_protos.SchedulePodRequest{
		NodeName: node.String(),
		Manifest: string(manifestBytes),
	}

	resp, err := c.client.SchedulePod(context.Background(), req)
	if err != nil {
		return "", util.Errorf("Could not schedule pod: %s", err)
	}

	return types.PodUniqueKey(resp.PodUniqueKey), nil
}

func (s *fakeKpStore) DeletePod(podPrefix kp.PodPrefix, nodeName types.NodeName, podID types.PodID) (time.Duration, error) {
	key := path.Join(string(podPrefix), nodeName.String(), podID.String())
	delete(s.manifests, key)
	return 0, nil
}

func (s *fakeKpStore) SetPod(podPrefix kp.PodPrefix, nodeName types.NodeName, manifest manifest.Manifest) (time.Duration, error) {
	key := path.Join(string(podPrefix), nodeName.String(), string(manifest.ID()))
	s.manifests[key] = manifest
	return 0, nil
}

func (f fakeConsulStore) GetHealth(service string, node types.NodeName) (kp.WatchResult, error) {
	return f.results[node.String()], nil
}

// these utility functions are used primarily while we exist in a mutable
// deployment world. We will need to figure out how to replace these with
// different datasources to allow RCs and DSs to continue to function correctly
// in the future.
func MakePodLabelKey(node types.NodeName, podID types.PodID) string {
	return node.String() + "/" + podID.String()
}

func computeRealityIndexPath(key types.PodUniqueKey, node types.NodeName) string {
	return path.Join("reality", node.String(), key.String())
}

func computeIntentIndexPath(key types.PodUniqueKey, node types.NodeName) string {
	return path.Join("intent", node.String(), key.String())
}