Move etcd helpers to tools package so they can depend on api package. Add…

pkg/api/helper.go

@@ -49,6 +49,13 @@ func AddKnownTypes(types ...interface{}) {
 	}
 }
 
+// Takes an arbitary api type, returns pointer to its JSONBase field.
+// obj must be a pointer to an api type.
+func FindJSONBase(obj interface{}) (*JSONBase, error) {
+	_, jsonBase, err := nameAndJSONBase(obj)
+	return jsonBase, err
+}
+
 // Encode turns the given api object into an appropriate JSON string.
 // Will return an error if the object doesn't have an embedded JSONBase.
 // Obj may be a pointer to a struct, or a struct. If a struct, a copy

pkg/api/types.go

@@ -150,6 +150,7 @@ type JSONBase struct {
 	ID                string `json:"id,omitempty" yaml:"id,omitempty"`
 	CreationTimestamp string `json:"creationTimestamp,omitempty" yaml:"creationTimestamp,omitempty"`
 	SelfLink          string `json:"selfLink,omitempty" yaml:"selfLink,omitempty"`
+	ResourceVersion   uint64 `json:"resourceVersion,omitempty" yaml:"resourceVersion,omitempty"`
 }
 
 type PodStatus string

pkg/controller/replication_controller.go

@@ -25,6 +25,7 @@ import (
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/client"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/tools"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
 	"github.com/coreos/go-etcd/etcd"
 	"github.com/golang/glog"
@@ -34,7 +35,7 @@ import (
 // with actual running pods.
 // TODO: Remove the etcd dependency and re-factor in terms of a generic watch interface
 type ReplicationManager struct {
-	etcdClient util.EtcdClient
+	etcdClient tools.EtcdClient
 	kubeClient client.ClientInterface
 	podControl PodControlInterface
 	syncTime   <-chan time.Time
@@ -76,7 +77,7 @@ func (r RealPodControl) deletePod(podID string) error {
 	return r.kubeClient.DeletePod(podID)
 }
 
-func MakeReplicationManager(etcdClient util.EtcdClient, kubeClient client.ClientInterface) *ReplicationManager {
+func MakeReplicationManager(etcdClient tools.EtcdClient, kubeClient client.ClientInterface) *ReplicationManager {
 	rm := &ReplicationManager{
 		kubeClient: kubeClient,
 		etcdClient: etcdClient,
@@ -201,7 +202,7 @@ func (rm *ReplicationManager) syncReplicationController(controllerSpec api.Repli
 
 func (rm *ReplicationManager) synchronize() {
 	var controllerSpecs []api.ReplicationController
-	helper := util.EtcdHelper{rm.etcdClient}
+	helper := tools.EtcdHelper{rm.etcdClient}
 	err := helper.ExtractList("/registry/controllers", &controllerSpecs)
 	if err != nil {
 		glog.Errorf("Synchronization error: %v (%#v)", err, err)

pkg/controller/replication_controller_test.go

@@ -26,6 +26,7 @@ import (
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/client"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/tools"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
 	"github.com/coreos/go-etcd/etcd"
 )
@@ -377,8 +378,8 @@ func TestSyncronize(t *testing.T) {
 		},
 	}
 
-	fakeEtcd := util.MakeFakeEtcdClient(t)
-	fakeEtcd.Data["/registry/controllers"] = util.EtcdResponseWithError{
+	fakeEtcd := tools.MakeFakeEtcdClient(t)
+	fakeEtcd.Data["/registry/controllers"] = tools.EtcdResponseWithError{
 		R: &etcd.Response{
 			Node: &etcd.Node{
 				Nodes: []*etcd.Node{
@@ -432,7 +433,7 @@ func (a *asyncTimeout) done() {
 
 func TestWatchControllers(t *testing.T) {
 	defer beginTimeout(20 * time.Second).done()
-	fakeEtcd := util.MakeFakeEtcdClient(t)
+	fakeEtcd := tools.MakeFakeEtcdClient(t)
 	manager := MakeReplicationManager(fakeEtcd, nil)
 	var testControllerSpec api.ReplicationController
 	received := make(chan bool)

pkg/kubelet/kubelet.go

@@ -34,6 +34,7 @@ import (
 	"time"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/tools"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
 	"github.com/coreos/go-etcd/etcd"
 	"github.com/fsouza/go-dockerclient"
@@ -81,7 +82,7 @@ func New() *Kubelet {
 // The main kubelet implementation
 type Kubelet struct {
 	Hostname           string
-	EtcdClient         util.EtcdClient
+	EtcdClient         tools.EtcdClient
 	DockerClient       DockerInterface
 	DockerPuller       DockerPuller
 	CadvisorClient     CadvisorInterface
@@ -520,7 +521,7 @@ func (kl *Kubelet) ResponseToManifests(response *etcd.Response) ([]api.Container
 func (kl *Kubelet) getKubeletStateFromEtcd(key string, updateChannel chan<- manifestUpdate) error {
 	response, err := kl.EtcdClient.Get(key, true, false)
 	if err != nil {
-		if util.IsEtcdNotFound(err) {
+		if tools.IsEtcdNotFound(err) {
 			return nil
 		}
 		glog.Errorf("Error on etcd get of %s: %v", key, err)

pkg/kubelet/kubelet_test.go

@@ -26,6 +26,7 @@ import (
 	"testing"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/tools"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
 	"github.com/coreos/go-etcd/etcd"
 	"github.com/fsouza/go-dockerclient"
@@ -75,8 +76,8 @@ func verifyError(t *testing.T, e error) {
 	}
 }
 
-func makeTestKubelet(t *testing.T) (*Kubelet, *util.FakeEtcdClient, *FakeDockerClient) {
-	fakeEtcdClient := util.MakeFakeEtcdClient(t)
+func makeTestKubelet(t *testing.T) (*Kubelet, *tools.FakeEtcdClient, *FakeDockerClient) {
+	fakeEtcdClient := tools.MakeFakeEtcdClient(t)
 	fakeDocker := &FakeDockerClient{
 		err: nil,
 	}
@@ -279,7 +280,7 @@ func TestGetKubeletStateFromEtcdNoData(t *testing.T) {
 	kubelet, fakeClient, _ := makeTestKubelet(t)
 	channel := make(chan manifestUpdate)
 	reader := startReading(channel)
-	fakeClient.Data["/registry/hosts/machine/kubelet"] = util.EtcdResponseWithError{
+	fakeClient.Data["/registry/hosts/machine/kubelet"] = tools.EtcdResponseWithError{
 		R: &etcd.Response{},
 		E: nil,
 	}
@@ -298,7 +299,7 @@ func TestGetKubeletStateFromEtcd(t *testing.T) {
 	kubelet, fakeClient, _ := makeTestKubelet(t)
 	channel := make(chan manifestUpdate)
 	reader := startReading(channel)
-	fakeClient.Data["/registry/hosts/machine/kubelet"] = util.EtcdResponseWithError{
+	fakeClient.Data["/registry/hosts/machine/kubelet"] = tools.EtcdResponseWithError{
 		R: &etcd.Response{
 			Node: &etcd.Node{
 				Value: util.MakeJSONString([]api.Container{}),
@@ -319,7 +320,7 @@ func TestGetKubeletStateFromEtcdNotFound(t *testing.T) {
 	kubelet, fakeClient, _ := makeTestKubelet(t)
 	channel := make(chan manifestUpdate)
 	reader := startReading(channel)
-	fakeClient.Data["/registry/hosts/machine/kubelet"] = util.EtcdResponseWithError{
+	fakeClient.Data["/registry/hosts/machine/kubelet"] = tools.EtcdResponseWithError{
 		R: &etcd.Response{},
 		E: &etcd.EtcdError{
 			ErrorCode: 100,
@@ -338,7 +339,7 @@ func TestGetKubeletStateFromEtcdError(t *testing.T) {
 	kubelet, fakeClient, _ := makeTestKubelet(t)
 	channel := make(chan manifestUpdate)
 	reader := startReading(channel)
-	fakeClient.Data["/registry/hosts/machine/kubelet"] = util.EtcdResponseWithError{
+	fakeClient.Data["/registry/hosts/machine/kubelet"] = tools.EtcdResponseWithError{
 		R: &etcd.Response{},
 		E: &etcd.EtcdError{
 			ErrorCode: 200, // non not found error

pkg/registry/etcd_registry.go

@@ -21,7 +21,7 @@ import (
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/tools"
 	"github.com/golang/glog"
 )
 
@@ -30,7 +30,7 @@ import (
 
 // EtcdRegistry is an implementation of both ControllerRegistry and PodRegistry which is backed with etcd.
 type EtcdRegistry struct {
-	etcdClient      util.EtcdClient
+	etcdClient      tools.EtcdClient
 	machines        MinionRegistry
 	manifestFactory ManifestFactory
 }
@@ -39,7 +39,7 @@ type EtcdRegistry struct {
 // 'client' is the connection to etcd
 // 'machines' is the list of machines
 // 'scheduler' is the scheduling algorithm to use.
-func MakeEtcdRegistry(client util.EtcdClient, machines MinionRegistry) *EtcdRegistry {
+func MakeEtcdRegistry(client tools.EtcdClient, machines MinionRegistry) *EtcdRegistry {
 	registry := &EtcdRegistry{
 		etcdClient: client,
 		machines:   machines,
@@ -54,8 +54,8 @@ func makePodKey(machine, podID string) string {
 	return "/registry/hosts/" + machine + "/pods/" + podID
 }
 
-func (registry *EtcdRegistry) helper() *util.EtcdHelper {
-	return &util.EtcdHelper{registry.etcdClient}
+func (registry *EtcdRegistry) helper() *tools.EtcdHelper {
+	return &tools.EtcdHelper{registry.etcdClient}
 }
 
 func (registry *EtcdRegistry) ListPods(selector labels.Selector) ([]api.Pod, error) {

pkg/tools/doc.go

+/*
+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 tools implements general tools which depend on the api package.
+package tools

pkg/util/etcd_tools.go → pkg/tools/etcd_tools.go

@@ -14,13 +14,14 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package util
+package tools
 
 import (
 	"encoding/json"
 	"fmt"
 	"reflect"
 
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
 	"github.com/coreos/go-etcd/etcd"
 )
 
@@ -117,8 +118,14 @@ func (h *EtcdHelper) ExtractList(key string, slicePtr interface{}) error {
 // a zero object of the requested type, or an error, depending on ignoreNotFound. Treats
 // empty responses and nil response nodes exactly like a not found error.
 func (h *EtcdHelper) ExtractObj(key string, objPtr interface{}, ignoreNotFound bool) error {
-	_, _, err := h.bodyAndExtractObj(key, objPtr, ignoreNotFound)
-	return err
+	_, index, err := h.bodyAndExtractObj(key, objPtr, ignoreNotFound)
+	if err != nil {
+		return err
+	}
+	if jsonBase, err := api.FindJSONBase(objPtr); err == nil {
+		jsonBase.ResourceVersion = index
+	}
+	return nil
 }
 
 func (h *EtcdHelper) bodyAndExtractObj(key string, objPtr interface{}, ignoreNotFound bool) (body string, modifiedIndex uint64, err error) {
@@ -147,7 +154,11 @@ func (h *EtcdHelper) SetObj(key string, obj interface{}) error {
 	if err != nil {
 		return err
 	}
-	_, err = h.Client.Set(key, string(data), 0)
+	if jsonBase, err := api.FindJSONBase(obj); err == nil && jsonBase.ResourceVersion != 0 {
+		_, err = h.Client.CompareAndSwap(key, string(data), 0, "", jsonBase.ResourceVersion)
+	} else {
+		_, err = h.Client.Set(key, string(data), 0)
+	}
 	return err
 }
 

pkg/util/etcd_tools_test.go → pkg/tools/etcd_tools_test.go

@@ -14,13 +14,14 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package util
+package tools
 
 import (
 	"fmt"
 	"reflect"
 	"testing"
 
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
 	"github.com/coreos/go-etcd/etcd"
 )
 
@@ -83,7 +84,7 @@ func TestExtractList(t *testing.T) {
 func TestExtractObj(t *testing.T) {
 	fakeClient := MakeFakeEtcdClient(t)
 	expect := testMarshalType{ID: "foo"}
-	fakeClient.Set("/some/key", MakeJSONString(expect), 0)
+	fakeClient.Set("/some/key", util.MakeJSONString(expect), 0)
 	helper := EtcdHelper{fakeClient}
 	var got testMarshalType
 	err := helper.ExtractObj("/some/key", &got, false)
@@ -143,7 +144,7 @@ func TestSetObj(t *testing.T) {
 	if err != nil {
 		t.Errorf("Unexpected error %#v", err)
 	}
-	expect := MakeJSONString(obj)
+	expect := util.MakeJSONString(obj)
 	got := fakeClient.Data["/some/key"].R.Node.Value
 	if expect != got {
 		t.Errorf("Wanted %v, got %v", expect, got)

pkg/util/fake_etcd_client.go → pkg/tools/fake_etcd_client.go

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package util
+package tools
 
 import (
 	"fmt"

pkg/util/doc.go

@@ -15,5 +15,6 @@ limitations under the License.
 */
 
 // Package util implements various utility functions used in both testing and implementation
-// of Kubernetes
+// of Kubernetes. Package util may not depend on any other package in the Kubernetes
+// package tree.
 package util