Merge pull request #1464 from hmrm/add-etcd-config

cmd/apiserver/apiserver.go

@@ -37,9 +37,12 @@ import (
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/cloudprovider"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/master"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/resources"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/tools"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/ui"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/version/verflag"
+
+	"github.com/coreos/go-etcd/etcd"
 	"github.com/golang/glog"
 )
 
@@ -56,6 +59,7 @@ var (
 	minionCacheTTL        = flag.Duration("minion_cache_ttl", 30*time.Second, "Duration of time to cache minion information. Default 30 seconds")
 	tokenAuthFile         = flag.String("token_auth_file", "", "If set, the file that will be used to secure the API server via token authentication")
 	etcdServerList        util.StringList
+	etcdConfigFile        = flag.String("etcd_config", "", "The config file for the etcd client. Mutually exclusive with -etcd_servers")
 	machineList           util.StringList
 	corsAllowedOriginList util.StringList
 	allowPrivileged       = flag.Bool("allow_privileged", false, "If true, allow privileged containers.")
@@ -66,7 +70,7 @@ var (
 
 func init() {
 	flag.Var(&address, "address", "The IP address on to serve on (set to 0.0.0.0 for all interfaces)")
-	flag.Var(&etcdServerList, "etcd_servers", "List of etcd servers to watch (http://ip:port), comma separated")
+	flag.Var(&etcdServerList, "etcd_servers", "List of etcd servers to watch (http://ip:port), comma separated. Mutually exclusive with -etcd_config")
 	flag.Var(&machineList, "machines", "List of machines to schedule onto, comma separated.")
 	flag.Var(&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.")
 }
@@ -114,6 +118,20 @@ func initCloudProvider(name string, configFilePath string) cloudprovider.Interfa
 	return cloud
 }
 
+func newEtcd(etcdConfigFile string, etcdServerList util.StringList) (helper tools.EtcdHelper, err error) {
+	var client tools.EtcdGetSet
+	if etcdConfigFile != "" {
+		client, err = etcd.NewClientFromFile(etcdConfigFile)
+		if err != nil {
+			return helper, err
+		}
+	} else {
+		client = etcd.NewClient(etcdServerList)
+	}
+
+	return master.NewEtcdHelper(client, *storageVersion)
+}
+
 func main() {
 	flag.Parse()
 	util.InitLogs()
@@ -122,8 +140,8 @@ func main() {
 	verflag.PrintAndExitIfRequested()
 	verifyMinionFlags()
 
-	if len(etcdServerList) == 0 {
-		glog.Fatalf("-etcd_servers flag is required.")
+	if (*etcdConfigFile != "" && len(etcdServerList) != 0) || (*etcdConfigFile == "" && len(etcdServerList) == 0) {
+		glog.Fatalf("specify either -etcd_servers or -etcd_config")
 	}
 
 	capabilities.Initialize(capabilities.Capabilities{
@@ -147,9 +165,9 @@ func main() {
 		glog.Fatalf("Invalid server address: %v", err)
 	}
 
-	helper, err := master.NewEtcdHelper(etcdServerList, *storageVersion)
+	helper, err := newEtcd(*etcdConfigFile, etcdServerList)
 	if err != nil {
-		glog.Fatalf("Invalid storage version: %v", err)
+		glog.Fatalf("Invalid storage version or misconfigured etcd: %v", err)
 	}
 
 	m := master.New(&master.Config{

cmd/integration/integration.go

@@ -117,7 +117,7 @@ func startComponents(manifestURL string) (apiServerURL string) {
 	cl.PollPeriod = time.Second * 1
 	cl.Sync = true
 
-	helper, err := master.NewEtcdHelper(servers, "")
+	helper, err := master.NewEtcdHelper(etcdClient, "")
 	if err != nil {
 		glog.Fatalf("Unable to get etcd helper: %v", err)
 	}

cmd/kubelet/kubelet.go

@@ -37,7 +37,6 @@ import (
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet"
 	kconfig "github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/config"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/master"
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/tools"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/version/verflag"
 	"github.com/coreos/go-etcd/etcd"
@@ -61,6 +60,7 @@ var (
 	networkContainerImage = flag.String("network_container_image", kubelet.NetworkContainerImage, "The image that network containers in each pod will use.")
 	dockerEndpoint        = flag.String("docker_endpoint", "", "If non-empty, use this for the docker endpoint to communicate with")
 	etcdServerList        util.StringList
+	etcdConfigFile        = flag.String("etcd_config", "", "The config file for the etcd client. Mutually exclusive with -etcd_servers")
 	rootDirectory         = flag.String("root_dir", defaultRootDir, "Directory path for managing kubelet files (volume mounts,etc).")
 	allowPrivileged       = flag.Bool("allow_privileged", false, "If true, allow containers to request privileged mode. [default=false]")
 	registryPullQPS       = flag.Float64("registry_qps", 0.0, "If > 0, limit registry pull QPS to this value.  If 0, unlimited. [default=0.0]")
@@ -69,7 +69,7 @@ var (
 )
 
 func init() {
-	flag.Var(&etcdServerList, "etcd_servers", "List of etcd servers to watch (http://ip:port), comma separated")
+	flag.Var(&etcdServerList, "etcd_servers", "List of etcd servers to watch (http://ip:port), comma separated. Mutually exclusive with -etcd_config")
 	flag.Var(&address, "address", "The IP address for the info server to serve on (set to 0.0.0.0 for all interfaces)")
 }
 
@@ -160,10 +160,19 @@ func main() {
 	}
 
 	// define etcd config source and initialize etcd client
-	var etcdClient tools.EtcdClient
+	var etcdClient *etcd.Client
 	if len(etcdServerList) > 0 {
-		glog.Infof("Watching for etcd configs at %v", etcdServerList)
 		etcdClient = etcd.NewClient(etcdServerList)
+	} else if *etcdConfigFile != "" {
+		var err error
+		etcdClient, err = etcd.NewClientFromFile(*etcdConfigFile)
+		if err != nil {
+			glog.Fatalf("Error with etcd config file: %v", err)
+		}
+	}
+
+	if etcdClient != nil {
+		glog.Infof("Watching for etcd configs at %v", etcdClient.GetCluster())
 		kconfig.NewSourceEtcd(kconfig.EtcdKeyForHost(hostname), etcdClient, cfg.Channel("etcd"))
 	}
 

cmd/proxy/proxy.go

@@ -33,13 +33,14 @@ import (
 var (
 	configFile     = flag.String("configfile", "/tmp/proxy_config", "Configuration file for the proxy")
 	etcdServerList util.StringList
+	etcdConfigFile = flag.String("etcd_config", "", "The config file for the etcd client. Mutually exclusive with -etcd_servers")
 	bindAddress    = util.IP(net.ParseIP("0.0.0.0"))
 	clientConfig   = &client.Config{}
 )
 
 func init() {
 	client.BindClientConfigFlags(flag.CommandLine, clientConfig)
-	flag.Var(&etcdServerList, "etcd_servers", "List of etcd servers to watch (http://ip:port), comma separated (optional)")
+	flag.Var(&etcdServerList, "etcd_servers", "List of etcd servers to watch (http://ip:port), comma separated (optional). Mutually exclusive with -etcd_config")
 	flag.Var(&bindAddress, "bind_address", "The address for the proxy server to serve on (set to 0.0.0.0 for all interfaces)")
 }
 
@@ -66,18 +67,34 @@ func main() {
 			serviceConfig.Channel("api"),
 			endpointsConfig.Channel("api"),
 		)
-	}
+	} else {
+
+		var etcdClient *etcd.Client
+
+		// Set up etcd client
+		if len(etcdServerList) > 0 {
+			// Set up logger for etcd client
+			etcd.SetLogger(util.NewLogger("etcd "))
+			etcdClient = etcd.NewClient(etcdServerList)
+		} else if *etcdConfigFile != "" {
+			// Set up logger for etcd client
+			etcd.SetLogger(util.NewLogger("etcd "))
+			var err error
+			etcdClient, err = etcd.NewClientFromFile(*etcdConfigFile)
+
+			if err != nil {
+				glog.Fatalf("Error with etcd config file: %v", err)
+			}
+		}
 
-	// Create a configuration source that handles configuration from etcd.
-	if len(etcdServerList) > 0 && clientConfig.Host == "" {
-		glog.Infof("Using etcd servers %v", etcdServerList)
+		// Create a configuration source that handles configuration from etcd.
+		if etcdClient != nil {
+			glog.Infof("Using etcd servers %v", etcdClient.GetCluster())
 
-		// Set up logger for etcd client
-		etcd.SetLogger(util.NewLogger("etcd "))
-		etcdClient := etcd.NewClient(etcdServerList)
-		config.NewConfigSourceEtcd(etcdClient,
-			serviceConfig.Channel("etcd"),
-			endpointsConfig.Channel("etcd"))
+			config.NewConfigSourceEtcd(etcdClient,
+				serviceConfig.Channel("etcd"),
+				endpointsConfig.Channel("etcd"))
+		}
 	}
 
 	// And create a configuration source that reads from a local file

pkg/master/master.go

@@ -38,7 +38,6 @@ import (
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/tools"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
 
-	goetcd "github.com/coreos/go-etcd/etcd"
 	"github.com/golang/glog"
 )
 
@@ -69,8 +68,7 @@ type Master struct {
 
 // NewEtcdHelper returns an EtcdHelper for the provided arguments or an error if the version
 // is incorrect.
-func NewEtcdHelper(etcdServers []string, version string) (helper tools.EtcdHelper, err error) {
-	client := goetcd.NewClient(etcdServers)
+func NewEtcdHelper(client tools.EtcdGetSet, version string) (helper tools.EtcdHelper, err error) {
 	if version == "" {
 		version = latest.Version
 	}

test/integration/client_test.go

@@ -38,7 +38,7 @@ func init() {
 }
 
 func TestClient(t *testing.T) {
-	helper, err := master.NewEtcdHelper(newEtcdClient().GetCluster(), "v1beta1")
+	helper, err := master.NewEtcdHelper(newEtcdClient(), "v1beta1")
 	if err != nil {
 		t.Fatalf("unexpected error: %v", err)
 	}