Adds ability to define a prefix for etcd paths

The API server can be supplied (via a command line flag) with a custom
prefix that is prepended to etcd resources paths.

Refs: #3476

cmd/integration/integration.go

@@ -51,6 +51,7 @@ import (
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/master"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/probe"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/service"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/tools/etcdtest"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/util/wait"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/volume/empty_dir"
@@ -160,7 +161,7 @@ func startComponents(firstManifestURL, secondManifestURL, apiVersion string) (st
 
 	cl := client.NewOrDie(&client.Config{Host: apiServer.URL, Version: apiVersion})
 
-	helper, err := master.NewEtcdHelper(etcdClient, "")
+	helper, err := master.NewEtcdHelper(etcdClient, "", etcdtest.PathPrefix())
 	if err != nil {
 		glog.Fatalf("Unable to get etcd helper: %v", err)
 	}

cmd/kube-apiserver/app/server.go

@@ -71,6 +71,7 @@ type APIServer struct {
 	AdmissionControlConfigFile string
 	EtcdServerList             util.StringList
 	EtcdConfigFile             string
+	EtcdPathPrefix             string
 	CorsAllowedOriginList      util.StringList
 	AllowPrivileged            bool
 	PortalNet                  util.IPNet // TODO: make this a list
@@ -98,6 +99,7 @@ func NewAPIServer() *APIServer {
 		EventTTL:               1 * time.Hour,
 		AuthorizationMode:      "AlwaysAllow",
 		AdmissionControl:       "AlwaysAdmit",
+		EtcdPathPrefix:         master.DefaultEtcdPathPrefix,
 		EnableLogsSupport:      true,
 		MasterServiceNamespace: api.NamespaceDefault,
 		ClusterName:            "kubernetes",
@@ -161,6 +163,7 @@ func (s *APIServer) AddFlags(fs *pflag.FlagSet) {
 	fs.StringVar(&s.AdmissionControlConfigFile, "admission_control_config_file", s.AdmissionControlConfigFile, "File with admission control configuration.")
 	fs.Var(&s.EtcdServerList, "etcd_servers", "List of etcd servers to watch (http://ip:port), comma separated. Mutually exclusive with -etcd_config")
 	fs.StringVar(&s.EtcdConfigFile, "etcd_config", s.EtcdConfigFile, "The config file for the etcd client. Mutually exclusive with -etcd_servers.")
+	fs.StringVar(&s.EtcdPathPrefix, "etcd_prefix", s.EtcdPathPrefix, "The prefix for all resource paths in etcd.")
 	fs.Var(&s.CorsAllowedOriginList, "cors_allowed_origins", "List of allowed origins for CORS, comma separated.  An allowed origin can be a regular expression to support subdomain matching.  If this list is empty CORS will not be enabled.")
 	fs.BoolVar(&s.AllowPrivileged, "allow_privileged", s.AllowPrivileged, "If true, allow privileged containers.")
 	fs.Var(&s.PortalNet, "portal_net", "A CIDR notation IP range from which to assign portal IPs. This must not overlap with any IP ranges assigned to nodes for pods.")
@@ -181,7 +184,7 @@ func (s *APIServer) verifyPortalFlags() {
 	}
 }
 
-func newEtcd(etcdConfigFile string, etcdServerList util.StringList, storageVersion string) (helper tools.EtcdHelper, err error) {
+func newEtcd(etcdConfigFile string, etcdServerList util.StringList, storageVersion string, pathPrefix string) (helper tools.EtcdHelper, err error) {
 	var client tools.EtcdGetSet
 	if etcdConfigFile != "" {
 		client, err = etcd.NewClientFromFile(etcdConfigFile)
@@ -192,7 +195,7 @@ func newEtcd(etcdConfigFile string, etcdServerList util.StringList, storageVersi
 		client = etcd.NewClient(etcdServerList)
 	}
 
-	return master.NewEtcdHelper(client, storageVersion)
+	return master.NewEtcdHelper(client, storageVersion, pathPrefix)
 }
 
 // Run runs the specified APIServer.  This should never exit.
@@ -232,7 +235,7 @@ func (s *APIServer) Run(_ []string) error {
 		glog.Fatalf("Invalid server address: %v", err)
 	}
 
-	helper, err := newEtcd(s.EtcdConfigFile, s.EtcdServerList, s.StorageVersion)
+	helper, err := newEtcd(s.EtcdConfigFile, s.EtcdServerList, s.StorageVersion, s.EtcdPathPrefix)
 	if err != nil {
 		glog.Fatalf("Invalid storage version or misconfigured etcd: %v", err)
 	}

cmd/kubernetes/kubernetes.go

@@ -81,7 +81,7 @@ func (h *delegateHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 func runApiServer(etcdClient tools.EtcdClient, addr net.IP, port int, masterServiceNamespace string) {
 	handler := delegateHandler{}
 
-	helper, err := master.NewEtcdHelper(etcdClient, "")
+	helper, err := master.NewEtcdHelper(etcdClient, "", master.DefaultEtcdPathPrefix)
 	if err != nil {
 		glog.Fatalf("Unable to get etcd helper: %v", err)
 	}

hack/test-go.sh

@@ -53,6 +53,9 @@ KUBE_RACE=${KUBE_RACE:-}   # use KUBE_RACE="-race" to enable race testing
 KUBE_GOVERALLS_BIN=${KUBE_GOVERALLS_BIN:-}
 # Comma separated list of API Versions that should be tested.
 KUBE_TEST_API_VERSIONS=${KUBE_TEST_API_VERSIONS:-"v1beta1,v1beta3"}
+# Prefixes for etcd paths (standard and customized)
+ETCD_STANDARD_PREFIX="registry"
+ETCD_CUSTOM_PREFIX="kubernetes.io/registry"
 
 kube::test::usage() {
   kube::log::usage_from_stdin <<EOF
@@ -203,10 +206,13 @@ reportCoverageToCoveralls() {
 
 # Convert the CSV to an array of API versions to test
 IFS=',' read -a apiVersions <<< "${KUBE_TEST_API_VERSIONS}"
+ETCD_PREFIX=${ETCD_STANDARD_PREFIX}
 for apiVersion in "${apiVersions[@]}"; do
   echo "Running tests for APIVersion: $apiVersion"
-  KUBE_API_VERSION="${apiVersion}" runTests "$@"
+  KUBE_API_VERSION="${apiVersion}" ETCD_PREFIX=${ETCD_STANDARD_PREFIX} runTests "$@"
 done
+echo "Using custom etcd path prefix: ${ETCD_CUSTOM_PREFIX}"
+KUBE_API_VERSION="${apiVersions[-1]}" ETCD_PREFIX=${ETCD_CUSTOM_PREFIX} runTests "$@"
 
 # We might run the tests for multiple versions, but we want to report only
 # one of them to coveralls. Here we report coverage from the last run.

pkg/api/rest/resttest/resttest.go

@@ -271,7 +271,6 @@ func (t *Tester) TestDeleteNoGraceful(createFn func() runtime.Object, wasGracefu
 	if err != nil {
 		t.Fatalf("object does not have ObjectMeta: %v\n%#v", err, existing)
 	}
-
 	ctx := api.WithNamespace(api.NewContext(), objectMeta.Namespace)
 	_, err = t.storage.(rest.GracefulDeleter).Delete(ctx, objectMeta.Name, api.NewDeleteOptions(10))
 	if err != nil {

pkg/master/master.go

@@ -71,6 +71,10 @@ import (
 	"github.com/golang/glog"
 )
 
+const (
+	DefaultEtcdPathPrefix = "/registry"
+)
+
 // Config is a structure used to configure a Master.
 type Config struct {
 	EtcdHelper        tools.EtcdHelper
@@ -178,15 +182,15 @@ type Master struct {
 
 // NewEtcdHelper returns an EtcdHelper for the provided arguments or an error if the version
 // is incorrect.
-func NewEtcdHelper(client tools.EtcdGetSet, version string) (helper tools.EtcdHelper, err error) {
+func NewEtcdHelper(client tools.EtcdGetSet, version string, prefix string) (helper tools.EtcdHelper, err error) {
 	if version == "" {
 		version = latest.Version
 	}
 	versionInterfaces, err := latest.InterfacesFor(version)
 	if err != nil {
 		return helper, err
 	}
-	return tools.NewEtcdHelper(client, versionInterfaces.Codec), nil
+	return tools.NewEtcdHelper(client, versionInterfaces.Codec, prefix), nil
 }
 
 // setDefaults fills in any fields not set that are required to have valid data.
@@ -363,7 +367,7 @@ func logStackOnRecover(panicReason interface{}, httpWriter http.ResponseWriter)
 func (m *Master) init(c *Config) {
 	// TODO: make initialization of the helper part of the Master, and allow some storage
 	// objects to have a newer storage version than the user's default.
-	newerHelper, err := NewEtcdHelper(c.EtcdHelper.Client, "v1beta3")
+	newerHelper, err := NewEtcdHelper(c.EtcdHelper.Client, "v1beta3", DefaultEtcdPathPrefix)
 	if err != nil {
 		glog.Fatalf("Unable to setup storage for v1beta3: %v", err)
 	}

pkg/master/master_test.go

@@ -23,14 +23,15 @@ import (
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/latest"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/registry/registrytest"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/tools"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/tools/etcdtest"
 )
 
 func TestGetServersToValidate(t *testing.T) {
 	master := Master{}
 	config := Config{}
 	fakeClient := tools.NewFakeEtcdClient(t)
 	fakeClient.Machines = []string{"http://machine1:4001", "http://machine2", "http://machine3:4003"}
-	config.EtcdHelper = tools.EtcdHelper{fakeClient, latest.Codec, nil}
+	config.EtcdHelper = tools.EtcdHelper{fakeClient, latest.Codec, nil, etcdtest.PathPrefix()}
 
 	master.nodeRegistry = registrytest.NewMinionRegistry([]string{"node1", "node2"}, api.NodeResources{})
 

pkg/registry/controller/etcd/etcd.go

@@ -34,7 +34,7 @@ type REST struct {
 
 // controllerPrefix is the location for controllers in etcd, only exposed
 // for testing
-var controllerPrefix = "/registry/controllers"
+var controllerPrefix = "/controllers"
 
 // NewREST returns a RESTStorage object that will work against replication controllers.
 func NewREST(h tools.EtcdHelper) *REST {

pkg/registry/controller/etcd/etcd_test.go

@@ -31,6 +31,7 @@ import (
 	etcdgeneric "github.com/GoogleCloudPlatform/kubernetes/pkg/registry/generic/etcd"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/tools"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/tools/etcdtest"
 	"github.com/coreos/go-etcd/etcd"
 )
 
@@ -39,11 +40,16 @@ const (
 	FAIL
 )
 
-// newStorage creates a REST storage backed by etcd helpers
-func newStorage(t *testing.T) (*REST, *tools.FakeEtcdClient) {
+func newHelper(t *testing.T) (*tools.FakeEtcdClient, tools.EtcdHelper) {
 	fakeEtcdClient := tools.NewFakeEtcdClient(t)
 	fakeEtcdClient.TestIndex = true
-	h := tools.NewEtcdHelper(fakeEtcdClient, latest.Codec)
+	helper := tools.NewEtcdHelper(fakeEtcdClient, latest.Codec, etcdtest.PathPrefix())
+	return fakeEtcdClient, helper
+}
+
+// newStorage creates a REST storage backed by etcd helpers
+func newStorage(t *testing.T) (*REST, *tools.FakeEtcdClient) {
+	fakeEtcdClient, h := newHelper(t)
 	storage := NewREST(h)
 	return storage, fakeEtcdClient
 }
@@ -107,6 +113,7 @@ func TestEtcdCreateController(t *testing.T) {
 		t.Errorf("unexpected error: %v", err)
 	}
 	key, _ := makeControllerKey(ctx, validController.Name)
+	key = etcdtest.AddPrefix(key)
 	resp, err := fakeClient.Get(key, false, false)
 	if err != nil {
 		t.Fatalf("Unexpected error %v", err)
@@ -126,6 +133,7 @@ func TestEtcdCreateControllerAlreadyExisting(t *testing.T) {
 	ctx := api.NewDefaultContext()
 	storage, fakeClient := newStorage(t)
 	key, _ := makeControllerKey(ctx, validController.Name)
+	key = etcdtest.AddPrefix(key)
 	fakeClient.Set(key, runtime.EncodeOrDie(latest.Codec, &validController), 0)
 
 	_, err := storage.Create(ctx, &validController)
@@ -199,6 +207,7 @@ func TestEtcdGetController(t *testing.T) {
 	ctx := api.NewDefaultContext()
 	storage, fakeClient := newStorage(t)
 	key, _ := makeControllerKey(ctx, validController.Name)
+	key = etcdtest.AddPrefix(key)
 	fakeClient.Set(key, runtime.EncodeOrDie(latest.Codec, &validController), 0)
 	ctrl, err := storage.Get(ctx, validController.Name)
 	if err != nil {
@@ -282,6 +291,9 @@ func TestEtcdGetControllerDifferentNamespace(t *testing.T) {
 	key1, _ := makeControllerKey(ctx1, validController.Name)
 	key2, _ := makeControllerKey(ctx2, validController.Name)
 
+	key1 = etcdtest.AddPrefix(key1)
+	key2 = etcdtest.AddPrefix(key2)
+
 	fakeClient.Set(key1, runtime.EncodeOrDie(latest.Codec, &validController), 0)
 	otherNsController := validController
 	otherNsController.Namespace = otherNs
@@ -317,6 +329,8 @@ func TestEtcdGetControllerNotFound(t *testing.T) {
 	ctx := api.NewDefaultContext()
 	storage, fakeClient := newStorage(t)
 	key, _ := makeControllerKey(ctx, validController.Name)
+	key = etcdtest.AddPrefix(key)
+
 	fakeClient.Data[key] = tools.EtcdResponseWithError{
 		R: &etcd.Response{
 			Node: nil,
@@ -336,6 +350,7 @@ func TestEtcdUpdateController(t *testing.T) {
 	ctx := api.NewDefaultContext()
 	storage, fakeClient := newStorage(t)
 	key, _ := makeControllerKey(ctx, validController.Name)
+	key = etcdtest.AddPrefix(key)
 
 	// set a key, then retrieve the current resource version and try updating it
 	resp, _ := fakeClient.Set(key, runtime.EncodeOrDie(latest.Codec, &validController), 0)
@@ -360,6 +375,8 @@ func TestEtcdDeleteController(t *testing.T) {
 	ctx := api.NewDefaultContext()
 	storage, fakeClient := newStorage(t)
 	key, _ := makeControllerKey(ctx, validController.Name)
+	key = etcdtest.AddPrefix(key)
+
 	fakeClient.Set(key, runtime.EncodeOrDie(latest.Codec, &validController), 0)
 	obj, err := storage.Delete(ctx, validController.Name, nil)
 	if err != nil {
@@ -382,6 +399,7 @@ func TestEtcdListControllers(t *testing.T) {
 	storage, fakeClient := newStorage(t)
 	ctx := api.NewDefaultContext()
 	key := makeControllerListKey(ctx)
+	key = etcdtest.AddPrefix(key)
 	controller := validController
 	controller.Name = "bar"
 	fakeClient.Data[key] = tools.EtcdResponseWithError{
@@ -413,6 +431,8 @@ func TestEtcdListControllersNotFound(t *testing.T) {
 	storage, fakeClient := newStorage(t)
 	ctx := api.NewDefaultContext()
 	key := makeControllerListKey(ctx)
+	key = etcdtest.AddPrefix(key)
+
 	fakeClient.Data[key] = tools.EtcdResponseWithError{
 		R: &etcd.Response{},
 		E: tools.EtcdErrorNotFound,
@@ -431,6 +451,7 @@ func TestEtcdListControllersLabelsMatch(t *testing.T) {
 	storage, fakeClient := newStorage(t)
 	ctx := api.NewDefaultContext()
 	key := makeControllerListKey(ctx)
+	key = etcdtest.AddPrefix(key)
 
 	controller := validController
 	controller.Labels = map[string]string{"k": "v"}
@@ -678,13 +699,16 @@ func TestCreate(t *testing.T) {
 }
 
 func TestDelete(t *testing.T) {
+	ctx := api.NewDefaultContext()
 	storage, fakeClient := newStorage(t)
 	test := resttest.New(t, storage, fakeClient.SetError)
+	key, _ := makeControllerKey(ctx, validController.Name)
+	key = etcdtest.AddPrefix(key)
 
 	createFn := func() runtime.Object {
 		rc := validController
 		rc.ResourceVersion = "1"
-		fakeClient.Data["/registry/controllers/default/foo"] = tools.EtcdResponseWithError{
+		fakeClient.Data[key] = tools.EtcdResponseWithError{
 			R: &etcd.Response{
 				Node: &etcd.Node{
 					Value:         runtime.EncodeOrDie(latest.Codec, &rc),
@@ -697,7 +721,7 @@ func TestDelete(t *testing.T) {
 	gracefulSetFn := func() bool {
 		// If the controller is still around after trying to delete either the delete
 		// failed, or we're deleting it gracefully.
-		if fakeClient.Data["/registry/controllers/default/foo"].R.Node != nil {
+		if fakeClient.Data[key].R.Node != nil {
 			return true
 		}
 		return false

pkg/registry/endpoint/etcd/etcd.go

@@ -34,7 +34,7 @@ type REST struct {
 
 // NewStorage returns a RESTStorage object that will work against endpoints.
 func NewStorage(h tools.EtcdHelper) *REST {
-	prefix := "/registry/services/endpoints"
+	prefix := "/services/endpoints"
 	return &REST{
 		&etcdgeneric.Etcd{
 			NewFunc:     func() runtime.Object { return &api.Endpoints{} },

pkg/registry/endpoint/etcd/etcd_test.go

@@ -26,6 +26,7 @@ import (
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/tools"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/tools/etcdtest"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
 
 	"github.com/coreos/go-etcd/etcd"
@@ -34,7 +35,7 @@ import (
 func newHelper(t *testing.T) (*tools.FakeEtcdClient, tools.EtcdHelper) {
 	fakeEtcdClient := tools.NewFakeEtcdClient(t)
 	fakeEtcdClient.TestIndex = true
-	helper := tools.NewEtcdHelper(fakeEtcdClient, latest.Codec)
+	helper := tools.NewEtcdHelper(fakeEtcdClient, latest.Codec, etcdtest.PathPrefix())
 	return fakeEtcdClient, helper
 }
 
@@ -89,6 +90,7 @@ func TestDelete(t *testing.T) {
 
 	endpoints := validChangedEndpoints()
 	key, _ := storage.KeyFunc(ctx, endpoints.Name)
+	key = etcdtest.AddPrefix(key)
 	createFn := func() runtime.Object {
 		fakeEtcdClient.Data[key] = tools.EtcdResponseWithError{
 			R: &etcd.Response{
@@ -113,6 +115,7 @@ func TestEtcdListEndpoints(t *testing.T) {
 	ctx := api.NewDefaultContext()
 	storage, fakeClient := newStorage(t)
 	key := storage.KeyRootFunc(ctx)
+	key = etcdtest.AddPrefix(key)
 	fakeClient.Data[key] = tools.EtcdResponseWithError{
 		R: &etcd.Response{
 			Node: &etcd.Node{
@@ -154,6 +157,7 @@ func TestEtcdGetEndpoints(t *testing.T) {
 	endpoints := validNewEndpoints()
 	name := endpoints.Name
 	key, _ := storage.KeyFunc(ctx, name)
+	key = etcdtest.AddPrefix(key)
 	fakeClient.Set(key, runtime.EncodeOrDie(latest.Codec, endpoints), 0)
 
 	response, err := fakeClient.Get(key, false, false)
@@ -183,6 +187,7 @@ func TestListEmptyEndpointsList(t *testing.T) {
 	storage, fakeClient := newStorage(t)
 	fakeClient.ChangeIndex = 1
 	key := storage.KeyRootFunc(ctx)
+	key = etcdtest.AddPrefix(key)
 	fakeClient.Data[key] = tools.EtcdResponseWithError{
 		R: &etcd.Response{},
 		E: fakeClient.NewError(tools.EtcdErrorCodeNotFound),
@@ -206,6 +211,7 @@ func TestListEndpointsList(t *testing.T) {
 	storage, fakeClient := newStorage(t)
 	fakeClient.ChangeIndex = 1
 	key := storage.KeyRootFunc(ctx)
+	key = etcdtest.AddPrefix(key)
 	fakeClient.Data[key] = tools.EtcdResponseWithError{
 		R: &etcd.Response{
 			Node: &etcd.Node{
@@ -266,6 +272,7 @@ func TestEtcdUpdateEndpoints(t *testing.T) {
 	endpoints := validChangedEndpoints()
 
 	key, _ := storage.KeyFunc(ctx, "foo")
+	key = etcdtest.AddPrefix(key)
 	fakeClient.Set(key, runtime.EncodeOrDie(latest.Codec, validNewEndpoints()), 0)
 
 	_, _, err := storage.Update(ctx, endpoints)
@@ -295,6 +302,7 @@ func TestDeleteEndpoints(t *testing.T) {
 	endpoints := validNewEndpoints()
 	name := endpoints.Name
 	key, _ := storage.KeyFunc(ctx, name)
+	key = etcdtest.AddPrefix(key)
 	fakeClient.ChangeIndex = 1
 	fakeClient.Data[key] = tools.EtcdResponseWithError{
 		R: &etcd.Response{