Breakup the registry package into separate packages.

Currently all registry implementations live in a single package,
which makes it bit harder to maintain. The different registry
implementations do not follow the same coding style and naming
conventions, which makes the code harder to read.

Breakup the registry package into smaller packages based on
the registry implementation. Refactor the registry packages
to follow a similar coding style and naming convention.

This patch does not introduce any changes in behavior.

pkg/registry/etcd/etcd.go

@@ -0,0 +1,330 @@
+/*
+Copyright 2014 Google Inc. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package etcd
+
+import (
+	"fmt"
+
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/apiserver"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/registry/minion"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/tools"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
+
+	"github.com/golang/glog"
+)
+
+// TODO: Need to add a reconciler loop that makes sure that things in pods are reflected into
+//       kubelet (and vice versa)
+
+// Registry implements PodRegistry, ControllerRegistry and ServiceRegistry
+// with backed by etcd.
+type Registry struct {
+	tools.EtcdHelper
+	manifestFactory ManifestFactory
+}
+
+// NewRegistry creates an etcd registry.
+func NewRegistry(client tools.EtcdClient, machines minion.Registry) *Registry {
+	registry := &Registry{
+		EtcdHelper: tools.EtcdHelper{
+			client,
+			api.Codec,
+			api.ResourceVersioner,
+		},
+	}
+	registry.manifestFactory = &BasicManifestFactory{
+		serviceRegistry: registry,
+	}
+	return registry
+}
+
+func makePodKey(podID string) string {
+	return "/registry/pods/" + podID
+}
+
+// ListPods obtains a list of pods that match selector.
+func (r *Registry) ListPods(selector labels.Selector) ([]api.Pod, error) {
+	allPods := []api.Pod{}
+	filteredPods := []api.Pod{}
+	if err := r.ExtractList("/registry/pods", &allPods); err != nil {
+		return nil, err
+	}
+	for _, pod := range allPods {
+		if selector.Matches(labels.Set(pod.Labels)) {
+			// TODO: Currently nothing sets CurrentState.Host. We need a feedback loop that sets
+			// the CurrentState.Host and Status fields. Here we pretend that reality perfectly
+			// matches our desires.
+			pod.CurrentState.Host = pod.DesiredState.Host
+			filteredPods = append(filteredPods, pod)
+		}
+	}
+	return filteredPods, nil
+}
+
+// GetPod gets a specific pod specified by its ID.
+func (r *Registry) GetPod(podID string) (*api.Pod, error) {
+	var pod api.Pod
+	if err := r.ExtractObj(makePodKey(podID), &pod, false); err != nil {
+		return nil, err
+	}
+	// TODO: Currently nothing sets CurrentState.Host. We need a feedback loop that sets
+	// the CurrentState.Host and Status fields. Here we pretend that reality perfectly
+	// matches our desires.
+	pod.CurrentState.Host = pod.DesiredState.Host
+	return &pod, nil
+}
+
+func makeContainerKey(machine string) string {
+	return "/registry/hosts/" + machine + "/kubelet"
+}
+
+// CreatePod creates a pod based on a specification, schedule it onto a specific machine.
+func (r *Registry) CreatePod(machine string, pod api.Pod) error {
+	// Set current status to "Waiting".
+	pod.CurrentState.Status = api.PodWaiting
+	pod.CurrentState.Host = ""
+	// DesiredState.Host == "" is a signal to the scheduler that this pod needs scheduling.
+	pod.DesiredState.Status = api.PodRunning
+	pod.DesiredState.Host = ""
+	if err := r.CreateObj(makePodKey(pod.ID), &pod); err != nil {
+		return err
+	}
+	// TODO: Until scheduler separation is completed, just assign here.
+	return r.AssignPod(pod.ID, machine)
+}
+
+// AssignPod assigns the given pod to the given machine.
+// TODO: hook this up via apiserver, not by calling it from CreatePod().
+func (r *Registry) AssignPod(podID string, machine string) error {
+	podKey := makePodKey(podID)
+	var finalPod *api.Pod
+	err := r.AtomicUpdate(podKey, &api.Pod{}, func(obj interface{}) (interface{}, error) {
+		pod, ok := obj.(*api.Pod)
+		if !ok {
+			return nil, fmt.Errorf("unexpected object: %#v", obj)
+		}
+		pod.DesiredState.Host = machine
+		finalPod = pod
+		return pod, nil
+	})
+	if err != nil {
+		return err
+	}
+	// TODO: move this to a watch/rectification loop.
+	manifest, err := r.manifestFactory.MakeManifest(machine, *finalPod)
+	if err != nil {
+		return err
+	}
+	contKey := makeContainerKey(machine)
+	err = r.AtomicUpdate(contKey, &api.ContainerManifestList{}, func(in interface{}) (interface{}, error) {
+		manifests := *in.(*api.ContainerManifestList)
+		manifests.Items = append(manifests.Items, manifest)
+		return manifests, nil
+	})
+	if err != nil {
+		// Don't strand stuff. This is a terrible hack that won't be needed
+		// when the above TODO is fixed.
+		err2 := r.Delete(podKey, false)
+		if err2 != nil {
+			glog.Errorf("Probably stranding a pod, couldn't delete %v: %#v", podKey, err2)
+		}
+	}
+	return err
+}
+
+func (r *Registry) UpdatePod(pod api.Pod) error {
+	return fmt.Errorf("unimplemented!")
+}
+
+// DeletePod deletes an existing pod specified by its ID.
+func (r *Registry) DeletePod(podID string) error {
+	var pod api.Pod
+	podKey := makePodKey(podID)
+	err := r.ExtractObj(podKey, &pod, false)
+	if tools.IsEtcdNotFound(err) {
+		return apiserver.NewNotFoundErr("pod", podID)
+	}
+	if err != nil {
+		return err
+	}
+	// First delete the pod, so a scheduler doesn't notice it getting removed from the
+	// machine and attempt to put it somewhere.
+	err = r.Delete(podKey, true)
+	if tools.IsEtcdNotFound(err) {
+		return apiserver.NewNotFoundErr("pod", podID)
+	}
+	if err != nil {
+		return err
+	}
+	machine := pod.DesiredState.Host
+	if machine == "" {
+		// Pod was never scheduled anywhere, just return.
+		return nil
+	}
+	// Next, remove the pod from the machine atomically.
+	contKey := makeContainerKey(machine)
+	return r.AtomicUpdate(contKey, &api.ContainerManifestList{}, func(in interface{}) (interface{}, error) {
+		manifests := in.(*api.ContainerManifestList)
+		newManifests := make([]api.ContainerManifest, 0, len(manifests.Items))
+		found := false
+		for _, manifest := range manifests.Items {
+			if manifest.ID != podID {
+				newManifests = append(newManifests, manifest)
+			} else {
+				found = true
+			}
+		}
+		if !found {
+			// This really shouldn't happen, it indicates something is broken, and likely
+			// there is a lost pod somewhere.
+			// However it is "deleted" so log it and move on
+			glog.Infof("Couldn't find: %s in %#v", podID, manifests)
+		}
+		manifests.Items = newManifests
+		return manifests, nil
+	})
+}
+
+// ListControllers obtains a list of ReplicationControllers.
+func (r *Registry) ListControllers() ([]api.ReplicationController, error) {
+	var controllers []api.ReplicationController
+	err := r.ExtractList("/registry/controllers", &controllers)
+	return controllers, err
+}
+
+// WatchControllers begins watching for new, changed, or deleted controllers.
+func (r *Registry) WatchControllers(label, field labels.Selector, resourceVersion uint64) (watch.Interface, error) {
+	if !field.Empty() {
+		return nil, fmt.Errorf("no field selector implemented for controllers")
+	}
+	return r.WatchList("/registry/controllers", resourceVersion, func(obj interface{}) bool {
+		return label.Matches(labels.Set(obj.(*api.ReplicationController).Labels))
+	})
+}
+
+func makeControllerKey(id string) string {
+	return "/registry/controllers/" + id
+}
+
+// GetController gets a specific ReplicationController specified by its ID.
+func (r *Registry) GetController(controllerID string) (*api.ReplicationController, error) {
+	var controller api.ReplicationController
+	key := makeControllerKey(controllerID)
+	err := r.ExtractObj(key, &controller, false)
+	if tools.IsEtcdNotFound(err) {
+		return nil, apiserver.NewNotFoundErr("replicationController", controllerID)
+	}
+	if err != nil {
+		return nil, err
+	}
+	return &controller, nil
+}
+
+// CreateController creates a new ReplicationController.
+func (r *Registry) CreateController(controller api.ReplicationController) error {
+	err := r.CreateObj(makeControllerKey(controller.ID), controller)
+	if tools.IsEtcdNodeExist(err) {
+		return apiserver.NewAlreadyExistsErr("replicationController", controller.ID)
+	}
+	return err
+}
+
+// UpdateController replaces an existing ReplicationController.
+func (r *Registry) UpdateController(controller api.ReplicationController) error {
+	return r.SetObj(makeControllerKey(controller.ID), controller)
+}
+
+// DeleteController deletes a ReplicationController specified by its ID.
+func (r *Registry) DeleteController(controllerID string) error {
+	key := makeControllerKey(controllerID)
+	err := r.Delete(key, false)
+	if tools.IsEtcdNotFound(err) {
+		return apiserver.NewNotFoundErr("replicationController", controllerID)
+	}
+	return err
+}
+
+func makeServiceKey(name string) string {
+	return "/registry/services/specs/" + name
+}
+
+// ListServices obtains a list of Services.
+func (r *Registry) ListServices() (api.ServiceList, error) {
+	var list api.ServiceList
+	err := r.ExtractList("/registry/services/specs", &list.Items)
+	return list, err
+}
+
+// CreateService creates a new Service.
+func (r *Registry) CreateService(svc api.Service) error {
+	err := r.CreateObj(makeServiceKey(svc.ID), svc)
+	if tools.IsEtcdNodeExist(err) {
+		return apiserver.NewAlreadyExistsErr("service", svc.ID)
+	}
+	return err
+}
+
+// GetService obtains a Service specified by its name.
+func (r *Registry) GetService(name string) (*api.Service, error) {
+	key := makeServiceKey(name)
+	var svc api.Service
+	err := r.ExtractObj(key, &svc, false)
+	if tools.IsEtcdNotFound(err) {
+		return nil, apiserver.NewNotFoundErr("service", name)
+	}
+	if err != nil {
+		return nil, err
+	}
+	return &svc, nil
+}
+
+func makeServiceEndpointsKey(name string) string {
+	return "/registry/services/endpoints/" + name
+}
+
+// DeleteService deletes a Service specified by its name.
+func (r *Registry) DeleteService(name string) error {
+	key := makeServiceKey(name)
+	err := r.Delete(key, true)
+	if tools.IsEtcdNotFound(err) {
+		return apiserver.NewNotFoundErr("service", name)
+	}
+	if err != nil {
+		return err
+	}
+	key = makeServiceEndpointsKey(name)
+	err = r.Delete(key, true)
+	if !tools.IsEtcdNotFound(err) {
+		return err
+	}
+	return nil
+}
+
+// UpdateService replaces an existing Service.
+func (r *Registry) UpdateService(svc api.Service) error {
+	return r.SetObj(makeServiceKey(svc.ID), svc)
+}
+
+// UpdateEndpoints update Endpoints of a Service.
+func (r *Registry) UpdateEndpoints(e api.Endpoints) error {
+	return r.AtomicUpdate(makeServiceEndpointsKey(e.ID), &api.Endpoints{},
+		func(interface{}) (interface{}, error) {
+			return e, nil
+		})
+}

pkg/registry/etcd/etcd_test.go

@@ -0,0 +1,821 @@
+/*
+Copyright 2014 Google Inc. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package etcd
+
+import (
+	"reflect"
+	"testing"
+
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/apiserver"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/registry/minion"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/registry/registrytest"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/tools"
+
+	"github.com/coreos/go-etcd/etcd"
+)
+
+func MakeTestEtcdRegistry(client tools.EtcdClient, machines []string) *Registry {
+	registry := NewRegistry(client, minion.NewRegistry(machines))
+	registry.manifestFactory = &BasicManifestFactory{
+		serviceRegistry: &registrytest.ServiceRegistry{},
+	}
+	return registry
+}
+
+func TestEtcdGetPod(t *testing.T) {
+	fakeClient := tools.MakeFakeEtcdClient(t)
+	fakeClient.Set("/registry/pods/foo", api.EncodeOrDie(api.Pod{JSONBase: api.JSONBase{ID: "foo"}}), 0)
+	registry := MakeTestEtcdRegistry(fakeClient, []string{"machine"})
+	pod, err := registry.GetPod("foo")
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	if pod.ID != "foo" {
+		t.Errorf("Unexpected pod: %#v", pod)
+	}
+}
+
+func TestEtcdGetPodNotFound(t *testing.T) {
+	fakeClient := tools.MakeFakeEtcdClient(t)
+	fakeClient.Data["/registry/pods/foo"] = tools.EtcdResponseWithError{
+		R: &etcd.Response{
+			Node: nil,
+		},
+		E: tools.EtcdErrorNotFound,
+	}
+	registry := MakeTestEtcdRegistry(fakeClient, []string{"machine"})
+	_, err := registry.GetPod("foo")
+	if err == nil {
+		t.Errorf("Unexpected non-error.")
+	}
+}
+
+func TestEtcdCreatePod(t *testing.T) {
+	fakeClient := tools.MakeFakeEtcdClient(t)
+	fakeClient.TestIndex = true
+	fakeClient.Data["/registry/pods/foo"] = tools.EtcdResponseWithError{
+		R: &etcd.Response{
+			Node: nil,
+		},
+		E: tools.EtcdErrorNotFound,
+	}
+	fakeClient.Set("/registry/hosts/machine/kubelet", api.EncodeOrDie(&api.ContainerManifestList{}), 0)
+	registry := MakeTestEtcdRegistry(fakeClient, []string{"machine"})
+	err := registry.CreatePod("machine", api.Pod{
+		JSONBase: api.JSONBase{
+			ID: "foo",
+		},
+		DesiredState: api.PodState{
+			Manifest: api.ContainerManifest{
+				Containers: []api.Container{
+					{
+						Name: "foo",
+					},
+				},
+			},
+		},
+	})
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	resp, err := fakeClient.Get("/registry/pods/foo", false, false)
+	if err != nil {
+		t.Fatalf("Unexpected error %v", err)
+	}
+	var pod api.Pod
+	err = api.DecodeInto([]byte(resp.Node.Value), &pod)
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	if pod.ID != "foo" {
+		t.Errorf("Unexpected pod: %#v %s", pod, resp.Node.Value)
+	}
+	var manifests api.ContainerManifestList
+	resp, err = fakeClient.Get("/registry/hosts/machine/kubelet", false, false)
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	err = api.DecodeInto([]byte(resp.Node.Value), &manifests)
+	if len(manifests.Items) != 1 || manifests.Items[0].ID != "foo" {
+		t.Errorf("Unexpected manifest list: %#v", manifests)
+	}
+}
+
+func TestEtcdCreatePodAlreadyExisting(t *testing.T) {
+	fakeClient := tools.MakeFakeEtcdClient(t)
+	fakeClient.Data["/registry/pods/foo"] = tools.EtcdResponseWithError{
+		R: &etcd.Response{
+			Node: &etcd.Node{
+				Value: api.EncodeOrDie(api.Pod{JSONBase: api.JSONBase{ID: "foo"}}),
+			},
+		},
+		E: nil,
+	}
+	registry := MakeTestEtcdRegistry(fakeClient, []string{"machine"})
+	err := registry.CreatePod("machine", api.Pod{
+		JSONBase: api.JSONBase{
+			ID: "foo",
+		},
+	})
+	if err == nil {
+		t.Error("Unexpected non-error")
+	}
+}
+
+func TestEtcdCreatePodWithContainersError(t *testing.T) {
+	fakeClient := tools.MakeFakeEtcdClient(t)
+	fakeClient.TestIndex = true
+	fakeClient.Data["/registry/pods/foo"] = tools.EtcdResponseWithError{
+		R: &etcd.Response{
+			Node: nil,
+		},
+		E: tools.EtcdErrorNotFound,
+	}
+	fakeClient.Data["/registry/hosts/machine/kubelet"] = tools.EtcdResponseWithError{
+		R: &etcd.Response{
+			Node: nil,
+		},
+		E: tools.EtcdErrorValueRequired,
+	}
+	registry := MakeTestEtcdRegistry(fakeClient, []string{"machine"})
+	err := registry.CreatePod("machine", api.Pod{
+		JSONBase: api.JSONBase{
+			ID: "foo",
+		},
+	})
+	if err == nil {
+		t.Fatalf("Unexpected non-error")
+	}
+	_, err = fakeClient.Get("/registry/pods/foo", false, false)
+	if err == nil {
+		t.Error("Unexpected non-error")
+	}
+	if !tools.IsEtcdNotFound(err) {
+		t.Errorf("Unexpected error: %#v", err)
+	}
+}
+
+func TestEtcdCreatePodWithContainersNotFound(t *testing.T) {
+	fakeClient := tools.MakeFakeEtcdClient(t)
+	fakeClient.TestIndex = true
+	fakeClient.Data["/registry/pods/foo"] = tools.EtcdResponseWithError{
+		R: &etcd.Response{
+			Node: nil,
+		},
+		E: tools.EtcdErrorNotFound,
+	}
+	fakeClient.Data["/registry/hosts/machine/kubelet"] = tools.EtcdResponseWithError{
+		R: &etcd.Response{
+			Node: nil,
+		},
+		E: tools.EtcdErrorNotFound,
+	}
+	registry := MakeTestEtcdRegistry(fakeClient, []string{"machine"})
+	err := registry.CreatePod("machine", api.Pod{
+		JSONBase: api.JSONBase{
+			ID: "foo",
+		},
+		DesiredState: api.PodState{
+			Manifest: api.ContainerManifest{
+				ID: "foo",
+				Containers: []api.Container{
+					{
+						Name: "foo",
+					},
+				},
+			},
+		},
+	})
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	resp, err := fakeClient.Get("/registry/pods/foo", false, false)
+	if err != nil {
+		t.Fatalf("Unexpected error %v", err)
+	}
+	var pod api.Pod
+	err = api.DecodeInto([]byte(resp.Node.Value), &pod)
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	if pod.ID != "foo" {
+		t.Errorf("Unexpected pod: %#v %s", pod, resp.Node.Value)
+	}
+	var manifests api.ContainerManifestList
+	resp, err = fakeClient.Get("/registry/hosts/machine/kubelet", false, false)
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	err = api.DecodeInto([]byte(resp.Node.Value), &manifests)
+	if len(manifests.Items) != 1 || manifests.Items[0].ID != "foo" {
+		t.Errorf("Unexpected manifest list: %#v", manifests)
+	}
+}
+
+func TestEtcdCreatePodWithExistingContainers(t *testing.T) {
+	fakeClient := tools.MakeFakeEtcdClient(t)
+	fakeClient.TestIndex = true
+	fakeClient.Data["/registry/pods/foo"] = tools.EtcdResponseWithError{
+		R: &etcd.Response{
+			Node: nil,
+		},
+		E: tools.EtcdErrorNotFound,
+	}
+	fakeClient.Set("/registry/hosts/machine/kubelet", api.EncodeOrDie(api.ContainerManifestList{
+		Items: []api.ContainerManifest{
+			{ID: "bar"},
+		},
+	}), 0)
+	registry := MakeTestEtcdRegistry(fakeClient, []string{"machine"})
+	err := registry.CreatePod("machine", api.Pod{
+		JSONBase: api.JSONBase{
+			ID: "foo",
+		},
+		DesiredState: api.PodState{
+			Manifest: api.ContainerManifest{
+				ID: "foo",
+				Containers: []api.Container{
+					{
+						Name: "foo",
+					},
+				},
+			},
+		},
+	})
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	resp, err := fakeClient.Get("/registry/pods/foo", false, false)
+	if err != nil {
+		t.Fatalf("Unexpected error %v", err)
+	}
+	var pod api.Pod
+	err = api.DecodeInto([]byte(resp.Node.Value), &pod)
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	if pod.ID != "foo" {
+		t.Errorf("Unexpected pod: %#v %s", pod, resp.Node.Value)
+	}
+	var manifests api.ContainerManifestList
+	resp, err = fakeClient.Get("/registry/hosts/machine/kubelet", false, false)
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	err = api.DecodeInto([]byte(resp.Node.Value), &manifests)
+	if len(manifests.Items) != 2 || manifests.Items[1].ID != "foo" {
+		t.Errorf("Unexpected manifest list: %#v", manifests)
+	}
+}
+
+func TestEtcdDeletePod(t *testing.T) {
+	fakeClient := tools.MakeFakeEtcdClient(t)
+	fakeClient.TestIndex = true
+
+	key := "/registry/pods/foo"
+	fakeClient.Set(key, api.EncodeOrDie(api.Pod{
+		JSONBase:     api.JSONBase{ID: "foo"},
+		DesiredState: api.PodState{Host: "machine"},
+	}), 0)
+	fakeClient.Set("/registry/hosts/machine/kubelet", api.EncodeOrDie(&api.ContainerManifestList{
+		Items: []api.ContainerManifest{
+			{ID: "foo"},
+		},
+	}), 0)
+	registry := MakeTestEtcdRegistry(fakeClient, []string{"machine"})
+	err := registry.DeletePod("foo")
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	if len(fakeClient.DeletedKeys) != 1 {
+		t.Errorf("Expected 1 delete, found %#v", fakeClient.DeletedKeys)
+	} else if fakeClient.DeletedKeys[0] != key {
+		t.Errorf("Unexpected key: %s, expected %s", fakeClient.DeletedKeys[0], key)
+	}
+	response, err := fakeClient.Get("/registry/hosts/machine/kubelet", false, false)
+	if err != nil {
+		t.Fatalf("Unexpected error %v", err)
+	}
+	var manifests api.ContainerManifestList
+	api.DecodeInto([]byte(response.Node.Value), &manifests)
+	if len(manifests.Items) != 0 {
+		t.Errorf("Unexpected container set: %s, expected empty", response.Node.Value)
+	}
+}
+
+func TestEtcdDeletePodMultipleContainers(t *testing.T) {
+	fakeClient := tools.MakeFakeEtcdClient(t)
+	fakeClient.TestIndex = true
+
+	key := "/registry/pods/foo"
+	fakeClient.Set(key, api.EncodeOrDie(api.Pod{
+		JSONBase:     api.JSONBase{ID: "foo"},
+		DesiredState: api.PodState{Host: "machine"},
+	}), 0)
+	fakeClient.Set("/registry/hosts/machine/kubelet", api.EncodeOrDie(&api.ContainerManifestList{
+		Items: []api.ContainerManifest{
+			{ID: "foo"},
+			{ID: "bar"},
+		},
+	}), 0)
+	registry := MakeTestEtcdRegistry(fakeClient, []string{"machine"})
+	err := registry.DeletePod("foo")
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	if len(fakeClient.DeletedKeys) != 1 {
+		t.Errorf("Expected 1 delete, found %#v", fakeClient.DeletedKeys)
+	}
+	if fakeClient.DeletedKeys[0] != key {
+		t.Errorf("Unexpected key: %s, expected %s", fakeClient.DeletedKeys[0], key)
+	}
+	response, err := fakeClient.Get("/registry/hosts/machine/kubelet", false, false)
+	if err != nil {
+		t.Fatalf("Unexpected error %v", err)
+	}
+	var manifests api.ContainerManifestList
+	api.DecodeInto([]byte(response.Node.Value), &manifests)
+	if len(manifests.Items) != 1 {
+		t.Fatalf("Unexpected manifest set: %#v, expected empty", manifests)
+	}
+	if manifests.Items[0].ID != "bar" {
+		t.Errorf("Deleted wrong manifest: %#v", manifests)
+	}
+}
+
+func TestEtcdEmptyListPods(t *testing.T) {
+	fakeClient := tools.MakeFakeEtcdClient(t)
+	key := "/registry/pods"
+	fakeClient.Data[key] = tools.EtcdResponseWithError{
+		R: &etcd.Response{
+			Node: &etcd.Node{
+				Nodes: []*etcd.Node{},
+			},
+		},
+		E: nil,
+	}
+	registry := MakeTestEtcdRegistry(fakeClient, []string{"machine"})
+	pods, err := registry.ListPods(labels.Everything())
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	if len(pods) != 0 {
+		t.Errorf("Unexpected pod list: %#v", pods)
+	}
+}
+
+func TestEtcdListPodsNotFound(t *testing.T) {
+	fakeClient := tools.MakeFakeEtcdClient(t)
+	key := "/registry/pods"
+	fakeClient.Data[key] = tools.EtcdResponseWithError{
+		R: &etcd.Response{},
+		E: tools.EtcdErrorNotFound,
+	}
+	registry := MakeTestEtcdRegistry(fakeClient, []string{"machine"})
+	pods, err := registry.ListPods(labels.Everything())
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	if len(pods) != 0 {
+		t.Errorf("Unexpected pod list: %#v", pods)
+	}
+}
+
+func TestEtcdListPods(t *testing.T) {
+	fakeClient := tools.MakeFakeEtcdClient(t)
+	key := "/registry/pods"
+	fakeClient.Data[key] = tools.EtcdResponseWithError{
+		R: &etcd.Response{
+			Node: &etcd.Node{
+				Nodes: []*etcd.Node{
+					{
+						Value: api.EncodeOrDie(api.Pod{
+							JSONBase:     api.JSONBase{ID: "foo"},
+							DesiredState: api.PodState{Host: "machine"},
+						}),
+					},
+					{
+						Value: api.EncodeOrDie(api.Pod{
+							JSONBase:     api.JSONBase{ID: "bar"},
+							DesiredState: api.PodState{Host: "machine"},
+						}),
+					},
+				},
+			},
+		},
+		E: nil,
+	}
+	registry := MakeTestEtcdRegistry(fakeClient, []string{"machine"})
+	pods, err := registry.ListPods(labels.Everything())
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	if len(pods) != 2 || pods[0].ID != "foo" || pods[1].ID != "bar" {
+		t.Errorf("Unexpected pod list: %#v", pods)
+	}
+	if pods[0].CurrentState.Host != "machine" ||
+		pods[1].CurrentState.Host != "machine" {
+		t.Errorf("Failed to populate host name.")
+	}
+}
+
+func TestEtcdListControllersNotFound(t *testing.T) {
+	fakeClient := tools.MakeFakeEtcdClient(t)
+	key := "/registry/controllers"
+	fakeClient.Data[key] = tools.EtcdResponseWithError{
+		R: &etcd.Response{},
+		E: tools.EtcdErrorNotFound,
+	}
+	registry := MakeTestEtcdRegistry(fakeClient, []string{"machine"})
+	controllers, err := registry.ListControllers()
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	if len(controllers) != 0 {
+		t.Errorf("Unexpected controller list: %#v", controllers)
+	}
+}
+
+func TestEtcdListServicesNotFound(t *testing.T) {
+	fakeClient := tools.MakeFakeEtcdClient(t)
+	key := "/registry/services/specs"
+	fakeClient.Data[key] = tools.EtcdResponseWithError{
+		R: &etcd.Response{},
+		E: tools.EtcdErrorNotFound,
+	}
+	registry := MakeTestEtcdRegistry(fakeClient, []string{"machine"})
+	services, err := registry.ListServices()
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	if len(services.Items) != 0 {
+		t.Errorf("Unexpected controller list: %#v", services)
+	}
+}
+
+func TestEtcdListControllers(t *testing.T) {
+	fakeClient := tools.MakeFakeEtcdClient(t)
+	key := "/registry/controllers"
+	fakeClient.Data[key] = tools.EtcdResponseWithError{
+		R: &etcd.Response{
+			Node: &etcd.Node{
+				Nodes: []*etcd.Node{
+					{
+						Value: api.EncodeOrDie(api.ReplicationController{JSONBase: api.JSONBase{ID: "foo"}}),
+					},
+					{
+						Value: api.EncodeOrDie(api.ReplicationController{JSONBase: api.JSONBase{ID: "bar"}}),
+					},
+				},
+			},
+		},
+		E: nil,
+	}
+	registry := MakeTestEtcdRegistry(fakeClient, []string{"machine"})
+	controllers, err := registry.ListControllers()
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	if len(controllers) != 2 || controllers[0].ID != "foo" || controllers[1].ID != "bar" {
+		t.Errorf("Unexpected controller list: %#v", controllers)
+	}
+}
+
+func TestEtcdGetController(t *testing.T) {
+	fakeClient := tools.MakeFakeEtcdClient(t)
+	fakeClient.Set("/registry/controllers/foo", api.EncodeOrDie(api.ReplicationController{JSONBase: api.JSONBase{ID: "foo"}}), 0)
+	registry := MakeTestEtcdRegistry(fakeClient, []string{"machine"})
+	ctrl, err := registry.GetController("foo")
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	if ctrl.ID != "foo" {
+		t.Errorf("Unexpected controller: %#v", ctrl)
+	}
+}
+
+func TestEtcdGetControllerNotFound(t *testing.T) {
+	fakeClient := tools.MakeFakeEtcdClient(t)
+	fakeClient.Data["/registry/controllers/foo"] = tools.EtcdResponseWithError{
+		R: &etcd.Response{
+			Node: nil,
+		},
+		E: tools.EtcdErrorNotFound,
+	}
+	registry := MakeTestEtcdRegistry(fakeClient, []string{"machine"})
+	ctrl, err := registry.GetController("foo")
+	if ctrl != nil {
+		t.Errorf("Unexpected non-nil controller: %#v", ctrl)
+	}
+	if err == nil {
+		t.Error("Unexpected non-error.")
+	}
+}
+
+func TestEtcdDeleteController(t *testing.T) {
+	fakeClient := tools.MakeFakeEtcdClient(t)
+	registry := MakeTestEtcdRegistry(fakeClient, []string{"machine"})
+	err := registry.DeleteController("foo")
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	if len(fakeClient.DeletedKeys) != 1 {
+		t.Errorf("Expected 1 delete, found %#v", fakeClient.DeletedKeys)
+	}
+	key := "/registry/controllers/foo"
+	if fakeClient.DeletedKeys[0] != key {
+		t.Errorf("Unexpected key: %s, expected %s", fakeClient.DeletedKeys[0], key)
+	}
+}
+
+func TestEtcdCreateController(t *testing.T) {
+	fakeClient := tools.MakeFakeEtcdClient(t)
+	registry := MakeTestEtcdRegistry(fakeClient, []string{"machine"})
+	err := registry.CreateController(api.ReplicationController{
+		JSONBase: api.JSONBase{
+			ID: "foo",
+		},
+	})
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	resp, err := fakeClient.Get("/registry/controllers/foo", false, false)
+	if err != nil {
+		t.Fatalf("Unexpected error %v", err)
+	}
+	var ctrl api.ReplicationController
+	err = api.DecodeInto([]byte(resp.Node.Value), &ctrl)
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	if ctrl.ID != "foo" {
+		t.Errorf("Unexpected pod: %#v %s", ctrl, resp.Node.Value)
+	}
+}
+
+func TestEtcdCreateControllerAlreadyExisting(t *testing.T) {
+	fakeClient := tools.MakeFakeEtcdClient(t)
+	fakeClient.Set("/registry/controllers/foo", api.EncodeOrDie(api.ReplicationController{JSONBase: api.JSONBase{ID: "foo"}}), 0)
+
+	registry := MakeTestEtcdRegistry(fakeClient, []string{"machine"})
+	err := registry.CreateController(api.ReplicationController{
+		JSONBase: api.JSONBase{
+			ID: "foo",
+		},
+	})
+	if !apiserver.IsAlreadyExists(err) {
+		t.Errorf("expected already exists err, got %#v", err)
+	}
+}
+
+func TestEtcdUpdateController(t *testing.T) {
+	fakeClient := tools.MakeFakeEtcdClient(t)
+	fakeClient.TestIndex = true
+
+	resp, _ := fakeClient.Set("/registry/controllers/foo", api.EncodeOrDie(api.ReplicationController{JSONBase: api.JSONBase{ID: "foo"}}), 0)
+	registry := MakeTestEtcdRegistry(fakeClient, []string{"machine"})
+	err := registry.UpdateController(api.ReplicationController{
+		JSONBase: api.JSONBase{ID: "foo", ResourceVersion: resp.Node.ModifiedIndex},
+		DesiredState: api.ReplicationControllerState{
+			Replicas: 2,
+		},
+	})
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	ctrl, err := registry.GetController("foo")
+	if ctrl.DesiredState.Replicas != 2 {
+		t.Errorf("Unexpected controller: %#v", ctrl)
+	}
+}
+
+func TestEtcdListServices(t *testing.T) {
+	fakeClient := tools.MakeFakeEtcdClient(t)
+	key := "/registry/services/specs"
+	fakeClient.Data[key] = tools.EtcdResponseWithError{
+		R: &etcd.Response{
+			Node: &etcd.Node{
+				Nodes: []*etcd.Node{
+					{
+						Value: api.EncodeOrDie(api.Service{JSONBase: api.JSONBase{ID: "foo"}}),
+					},
+					{
+						Value: api.EncodeOrDie(api.Service{JSONBase: api.JSONBase{ID: "bar"}}),
+					},
+				},
+			},
+		},
+		E: nil,
+	}
+	registry := MakeTestEtcdRegistry(fakeClient, []string{"machine"})
+	services, err := registry.ListServices()
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	if len(services.Items) != 2 || services.Items[0].ID != "foo" || services.Items[1].ID != "bar" {
+		t.Errorf("Unexpected pod list: %#v", services)
+	}
+}
+
+func TestEtcdCreateService(t *testing.T) {
+	fakeClient := tools.MakeFakeEtcdClient(t)
+	registry := MakeTestEtcdRegistry(fakeClient, []string{"machine"})
+	err := registry.CreateService(api.Service{
+		JSONBase: api.JSONBase{ID: "foo"},
+	})
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	resp, err := fakeClient.Get("/registry/services/specs/foo", false, false)
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	var service api.Service
+	err = api.DecodeInto([]byte(resp.Node.Value), &service)
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	if service.ID != "foo" {
+		t.Errorf("Unexpected service: %#v %s", service, resp.Node.Value)
+	}
+}
+
+func TestEtcdCreateServiceAlreadyExisting(t *testing.T) {
+	fakeClient := tools.MakeFakeEtcdClient(t)
+	fakeClient.Set("/registry/services/specs/foo", api.EncodeOrDie(api.Service{JSONBase: api.JSONBase{ID: "foo"}}), 0)
+	registry := MakeTestEtcdRegistry(fakeClient, []string{"machine"})
+	err := registry.CreateService(api.Service{
+		JSONBase: api.JSONBase{ID: "foo"},
+	})
+	if !apiserver.IsAlreadyExists(err) {
+		t.Errorf("expected already exists err, got %#v", err)
+	}
+}
+
+func TestEtcdGetService(t *testing.T) {
+	fakeClient := tools.MakeFakeEtcdClient(t)
+	fakeClient.Set("/registry/services/specs/foo", api.EncodeOrDie(api.Service{JSONBase: api.JSONBase{ID: "foo"}}), 0)
+	registry := MakeTestEtcdRegistry(fakeClient, []string{"machine"})
+	service, err := registry.GetService("foo")
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	if service.ID != "foo" {
+		t.Errorf("Unexpected pod: %#v", service)
+	}
+}
+
+func TestEtcdGetServiceNotFound(t *testing.T) {
+	fakeClient := tools.MakeFakeEtcdClient(t)
+	fakeClient.Data["/registry/services/specs/foo"] = tools.EtcdResponseWithError{
+		R: &etcd.Response{
+			Node: nil,
+		},
+		E: tools.EtcdErrorNotFound,
+	}
+	registry := MakeTestEtcdRegistry(fakeClient, []string{"machine"})
+	_, err := registry.GetService("foo")
+	if err == nil {
+		t.Errorf("Unexpected non-error.")
+	}
+}
+
+func TestEtcdDeleteService(t *testing.T) {
+	fakeClient := tools.MakeFakeEtcdClient(t)
+	registry := MakeTestEtcdRegistry(fakeClient, []string{"machine"})
+	err := registry.DeleteService("foo")
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	if len(fakeClient.DeletedKeys) != 2 {
+		t.Errorf("Expected 2 delete, found %#v", fakeClient.DeletedKeys)
+	}
+	key := "/registry/services/specs/foo"
+	if fakeClient.DeletedKeys[0] != key {
+		t.Errorf("Unexpected key: %s, expected %s", fakeClient.DeletedKeys[0], key)
+	}
+	key = "/registry/services/endpoints/foo"
+	if fakeClient.DeletedKeys[1] != key {
+		t.Errorf("Unexpected key: %s, expected %s", fakeClient.DeletedKeys[1], key)
+	}
+}
+
+func TestEtcdUpdateService(t *testing.T) {
+	fakeClient := tools.MakeFakeEtcdClient(t)
+	fakeClient.TestIndex = true
+
+	resp, _ := fakeClient.Set("/registry/services/specs/foo", api.EncodeOrDie(api.Service{JSONBase: api.JSONBase{ID: "foo"}}), 0)
+	registry := MakeTestEtcdRegistry(fakeClient, []string{"machine"})
+	testService := api.Service{
+		JSONBase: api.JSONBase{ID: "foo", ResourceVersion: resp.Node.ModifiedIndex},
+		Labels: map[string]string{
+			"baz": "bar",
+		},
+		Selector: map[string]string{
+			"baz": "bar",
+		},
+	}
+	err := registry.UpdateService(testService)
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	svc, err := registry.GetService("foo")
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	// Clear modified indices before the equality test.
+	svc.ResourceVersion = 0
+	testService.ResourceVersion = 0
+	if !reflect.DeepEqual(*svc, testService) {
+		t.Errorf("Unexpected service: got\n %#v\n, wanted\n %#v", svc, testService)
+	}
+}
+
+func TestEtcdUpdateEndpoints(t *testing.T) {
+	fakeClient := tools.MakeFakeEtcdClient(t)
+	fakeClient.TestIndex = true
+	registry := MakeTestEtcdRegistry(fakeClient, []string{"machine"})
+	endpoints := api.Endpoints{
+		JSONBase:  api.JSONBase{ID: "foo"},
+		Endpoints: []string{"baz", "bar"},
+	}
+
+	fakeClient.Set("/registry/services/endpoints/foo", api.EncodeOrDie(api.Endpoints{}), 0)
+
+	err := registry.UpdateEndpoints(endpoints)
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	response, err := fakeClient.Get("/registry/services/endpoints/foo", false, false)
+	if err != nil {
+		t.Fatalf("Unexpected error %v", err)
+	}
+	var endpointsOut api.Endpoints
+	err = api.DecodeInto([]byte(response.Node.Value), &endpointsOut)
+	if !reflect.DeepEqual(endpoints, endpointsOut) {
+		t.Errorf("Unexpected endpoints: %#v, expected %#v", endpointsOut, endpoints)
+	}
+}
+
+// TODO We need a test for the compare and swap behavior.  This basically requires two things:
+//   1) Add a per-operation synchronization channel to the fake etcd client, such that any operation waits on that
+//      channel, this will enable us to orchestrate the flow of etcd requests in the test.
+//   2) We need to make the map from key to (response, error) actually be a [](response, error) and pop
+//      our way through the responses.  That will enable us to hand back multiple different responses for
+//      the same key.
+//   Once that infrastructure is in place, the test looks something like:
+//      Routine #1                               Routine #2
+//         Read
+//         Wait for sync on update               Read
+//                                               Update
+//         Update
+//   In the buggy case, this will result in lost data.  In the correct case, the second update should fail
+//   and be retried.

pkg/registry/etcd/manifest_factory.go

@@ -0,0 +1,43 @@
+/*
+Copyright 2014 Google Inc. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package etcd
+
+import (
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/registry/service"
+)
+
+type ManifestFactory interface {
+	// Make a container object for a given pod, given the machine that the pod is running on.
+	MakeManifest(machine string, pod api.Pod) (api.ContainerManifest, error)
+}
+
+type BasicManifestFactory struct {
+	serviceRegistry service.Registry
+}
+
+func (b *BasicManifestFactory) MakeManifest(machine string, pod api.Pod) (api.ContainerManifest, error) {
+	envVars, err := service.GetServiceEnvironmentVariables(b.serviceRegistry, machine)
+	if err != nil {
+		return api.ContainerManifest{}, err
+	}
+	for ix, container := range pod.DesiredState.Manifest.Containers {
+		pod.DesiredState.Manifest.ID = pod.ID
+		pod.DesiredState.Manifest.Containers[ix].Env = append(container.Env, envVars...)
+	}
+	return pod.DesiredState.Manifest, nil
+}

pkg/registry/etcd/manifest_factory_test.go

@@ -0,0 +1,221 @@
+/*
+Copyright 2014 Google Inc. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package etcd
+
+import (
+	"reflect"
+	"testing"
+
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/registry/registrytest"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
+)
+
+func TestMakeManifestNoServices(t *testing.T) {
+	registry := registrytest.ServiceRegistry{}
+	factory := &BasicManifestFactory{
+		serviceRegistry: &registry,
+	}
+
+	manifest, err := factory.MakeManifest("machine", api.Pod{
+		JSONBase: api.JSONBase{ID: "foobar"},
+		DesiredState: api.PodState{
+			Manifest: api.ContainerManifest{
+				Containers: []api.Container{
+					{
+						Name: "foo",
+					},
+				},
+			},
+		},
+	})
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	container := manifest.Containers[0]
+	if len(container.Env) != 1 ||
+		container.Env[0].Name != "SERVICE_HOST" ||
+		container.Env[0].Value != "machine" {
+		t.Errorf("Expected one env vars, got: %#v", manifest)
+	}
+	if manifest.ID != "foobar" {
+		t.Errorf("Failed to assign ID to manifest: %#v", manifest.ID)
+	}
+}
+
+func TestMakeManifestServices(t *testing.T) {
+	registry := registrytest.ServiceRegistry{
+		List: api.ServiceList{
+			Items: []api.Service{
+				{
+					JSONBase: api.JSONBase{ID: "test"},
+					Port:     8080,
+					ContainerPort: util.IntOrString{
+						Kind:   util.IntstrInt,
+						IntVal: 900,
+					},
+				},
+			},
+		},
+	}
+	factory := &BasicManifestFactory{
+		serviceRegistry: &registry,
+	}
+
+	manifest, err := factory.MakeManifest("machine", api.Pod{
+		DesiredState: api.PodState{
+			Manifest: api.ContainerManifest{
+				Containers: []api.Container{
+					{
+						Name: "foo",
+					},
+				},
+			},
+		},
+	})
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	container := manifest.Containers[0]
+	envs := []api.EnvVar{
+		{
+			Name:  "TEST_SERVICE_PORT",
+			Value: "8080",
+		},
+		{
+			Name:  "TEST_PORT",
+			Value: "tcp://machine:8080",
+		},
+		{
+			Name:  "TEST_PORT_900_TCP",
+			Value: "tcp://machine:8080",
+		},
+		{
+			Name:  "TEST_PORT_900_TCP_PROTO",
+			Value: "tcp",
+		},
+		{
+			Name:  "TEST_PORT_900_TCP_PORT",
+			Value: "8080",
+		},
+		{
+			Name:  "TEST_PORT_900_TCP_ADDR",
+			Value: "machine",
+		},
+		{
+			Name:  "SERVICE_HOST",
+			Value: "machine",
+		},
+	}
+	if len(container.Env) != 7 {
+		t.Errorf("Expected 7 env vars, got %d: %#v", len(container.Env), manifest)
+		return
+	}
+	for ix := range container.Env {
+		if !reflect.DeepEqual(envs[ix], container.Env[ix]) {
+			t.Errorf("expected %#v, got %#v", envs[ix], container.Env[ix])
+		}
+	}
+}
+
+func TestMakeManifestServicesExistingEnvVar(t *testing.T) {
+	registry := registrytest.ServiceRegistry{
+		List: api.ServiceList{
+			Items: []api.Service{
+				{
+					JSONBase: api.JSONBase{ID: "test"},
+					Port:     8080,
+					ContainerPort: util.IntOrString{
+						Kind:   util.IntstrInt,
+						IntVal: 900,
+					},
+				},
+			},
+		},
+	}
+	factory := &BasicManifestFactory{
+		serviceRegistry: &registry,
+	}
+
+	manifest, err := factory.MakeManifest("machine", api.Pod{
+		DesiredState: api.PodState{
+			Manifest: api.ContainerManifest{
+				Containers: []api.Container{
+					{
+						Env: []api.EnvVar{
+							{
+								Name:  "foo",
+								Value: "bar",
+							},
+						},
+					},
+				},
+			},
+		},
+	})
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	container := manifest.Containers[0]
+
+	envs := []api.EnvVar{
+		{
+			Name:  "foo",
+			Value: "bar",
+		},
+		{
+			Name:  "TEST_SERVICE_PORT",
+			Value: "8080",
+		},
+		{
+			Name:  "TEST_PORT",
+			Value: "tcp://machine:8080",
+		},
+		{
+			Name:  "TEST_PORT_900_TCP",
+			Value: "tcp://machine:8080",
+		},
+		{
+			Name:  "TEST_PORT_900_TCP_PROTO",
+			Value: "tcp",
+		},
+		{
+			Name:  "TEST_PORT_900_TCP_PORT",
+			Value: "8080",
+		},
+		{
+			Name:  "TEST_PORT_900_TCP_ADDR",
+			Value: "machine",
+		},
+		{
+			Name:  "SERVICE_HOST",
+			Value: "machine",
+		},
+	}
+	if len(container.Env) != 8 {
+		t.Errorf("Expected 8 env vars, got: %#v", manifest)
+		return
+	}
+	for ix := range container.Env {
+		if !reflect.DeepEqual(envs[ix], container.Env[ix]) {
+			t.Errorf("expected %#v, got %#v", envs[ix], container.Env[ix])
+		}
+	}
+}