Merge pull request #587 from hongchaodeng/f

SHE: auto-detect TLS and create secure etcd client

Author: hongchaodeng

cmd/bootkube/recover.go

@@ -87,7 +87,8 @@ func runCmdRecover(cmd *cobra.Command, args []string) error {
 
 		bootkube.UserOutput("Waiting for etcd server to start...\n")
 
-		err = etcdutil.WaitClusterReady(recovery.RecoveryEtcdClientAddr)
+		// TODO: add self-hosted etcd+TLS support?
+		err = etcdutil.WaitClusterReady(recovery.RecoveryEtcdClientAddr, nil)
 		if err != nil {
 			return err
 		}

glide.lock

@@ -9,6 +9,7 @@ import:
   subpackages:
   - clientv3
   - pkg/fileutil
+  - pkg/transport
 - package: github.com/ghodss/yaml
   version: 0ca9ea5df5451ffdf184b4428c902747c2c11cd7
 - package: github.com/golang/glog

glide.yaml

@@ -6,10 +6,10 @@ import (
 	"path/filepath"
 	"time"
 
-	"k8s.io/client-go/tools/clientcmd"
-
 	"github.com/kubernetes-incubator/bootkube/pkg/asset"
 	"github.com/kubernetes-incubator/bootkube/pkg/util/etcdutil"
+
+	"k8s.io/client-go/tools/clientcmd"
 )
 
 const assetTimeout = 20 * time.Minute
@@ -87,7 +87,11 @@ func (b *bootkube) Run() error {
 
 	if selfHostedEtcd {
 		UserOutput("Migrating to self-hosted etcd cluster...\n")
-		if err = etcdutil.Migrate(kubeConfig, filepath.Join(b.assetDir, asset.AssetPathBootstrapEtcdService), filepath.Join(b.assetDir, asset.AssetPathMigrateEtcdCluster)); err != nil {
+
+		svcPath := filepath.Join(b.assetDir, asset.AssetPathBootstrapEtcdService)
+		tprPath := filepath.Join(b.assetDir, asset.AssetPathMigrateEtcdCluster)
+		err = etcdutil.Migrate(kubeConfig, b.assetDir, svcPath, tprPath)
+		if err != nil {
 			return err
 		}
 	}

pkg/bootkube/bootkube.go

@@ -2,10 +2,13 @@ package etcdutil
 
 import (
 	"context"
+	"crypto/tls"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"io/ioutil"
+	"os"
+	"path/filepath"
 	"time"
 
 	"github.com/kubernetes-incubator/bootkube/pkg/asset"
@@ -32,7 +35,19 @@ var (
 	pollTimeout  = 300 * time.Second
 )
 
-func Migrate(kubeConfig clientcmd.ClientConfig, svcPath, tprPath string) error {
+func Migrate(kubeConfig clientcmd.ClientConfig, assetDir, svcPath, tprPath string) error {
+	useEtcdTLS, err := detectEtcdTLS(assetDir)
+	if err != nil {
+		return err
+	}
+	var etcdTLS *tls.Config
+	if useEtcdTLS {
+		etcdTLS, err = makeTLSConfig(assetDir)
+		if err != nil {
+			return err
+		}
+	}
+
 	config, err := kubeConfig.ClientConfig()
 	if err != nil {
 		return fmt.Errorf("failed to create kube client config: %v", err)
@@ -49,7 +64,7 @@ func Migrate(kubeConfig clientcmd.ClientConfig, svcPath, tprPath string) error {
 	}
 	glog.Infof("created etcd cluster TPR")
 
-	if err := createBootstrapEtcdService(kubecli, svcPath); err != nil {
+	if err := createBootstrapEtcdService(kubecli, etcdTLS, svcPath); err != nil {
 		return fmt.Errorf("failed to create bootstrap-etcd-service: %v", err)
 	}
 	defer cleanupBootstrapEtcdService(kubecli)
@@ -70,7 +85,7 @@ func Migrate(kubeConfig clientcmd.ClientConfig, svcPath, tprPath string) error {
 	}
 	glog.Info("etcd cluster for migration is now running")
 
-	if err := waitBootEtcdRemoved(etcdServiceIP); err != nil {
+	if err := waitBootEtcdRemoved(etcdServiceIP, etcdTLS); err != nil {
 		return fmt.Errorf("failed to wait for boot-etcd to be removed: %v", err)
 	}
 	glog.Info("removed boot-etcd from the etcd cluster")
@@ -99,7 +114,7 @@ func waitEtcdTPRReady(restClient restclient.Interface, ns string) error {
 	return nil
 }
 
-func createBootstrapEtcdService(kubecli kubernetes.Interface, svcPath string) error {
+func createBootstrapEtcdService(kubecli kubernetes.Interface, etcdTLS *tls.Config, svcPath string) error {
 	// Create the service.
 	svcb, err := ioutil.ReadFile(svcPath)
 	if err != nil {
@@ -115,8 +130,12 @@ func createBootstrapEtcdService(kubecli kubernetes.Interface, svcPath string) er
 		return err
 	}
 
+	scheme := "http://"
+	if etcdTLS != nil {
+		scheme = "https://"
+	}
 	// Wait for the service to be reachable (sometimes this takes a little while).
-	if err := WaitClusterReady("http://" + svc.Spec.ClusterIP + ":12379"); err != nil {
+	if err := WaitClusterReady(scheme+svc.Spec.ClusterIP+":12379", etcdTLS); err != nil {
 		return fmt.Errorf("timed out waiting for bootstrap etcd service: %s", err)
 	}
 	return nil
@@ -165,12 +184,18 @@ func getServiceIP(kubecli kubernetes.Interface, ns, svcName string) (string, err
 	return svc.Spec.ClusterIP, nil
 }
 
-func waitBootEtcdRemoved(etcdServiceIP string) error {
+func waitBootEtcdRemoved(etcdServiceIP string, etcdTLS *tls.Config) error {
+	scheme := "http"
+	if etcdTLS != nil {
+		scheme = "https"
+	}
+	cfg := clientv3.Config{
+		Endpoints:   []string{fmt.Sprintf("%s://%s:2379", scheme, etcdServiceIP)},
+		TLS:         etcdTLS,
+		DialTimeout: 5 * time.Second,
+	}
+
 	err := wait.Poll(pollInterval, pollTimeout, func() (bool, error) {
-		cfg := clientv3.Config{
-			Endpoints:   []string{fmt.Sprintf("http://%s:2379", etcdServiceIP)},
-			DialTimeout: 5 * time.Second,
-		}
 		etcdcli, err := clientv3.New(cfg)
 		if err != nil {
 			glog.Errorf("failed to create etcd client, will retry: %v", err)
@@ -210,3 +235,15 @@ func cleanupBootstrapEtcdService(kubecli kubernetes.Interface) {
 		glog.Errorf("timed out removing bootstrap-etcd-service: %v", err)
 	}
 }
+
+func detectEtcdTLS(assetDir string) (bool, error) {
+	etcdCAAssetPath := filepath.Join(assetDir, asset.AssetPathEtcdCA)
+	_, err := os.Stat(etcdCAAssetPath)
+	if err == nil {
+		return true, nil
+	}
+	if os.IsNotExist(err) {
+		return false, nil
+	}
+	return false, err
+}

pkg/util/etcdutil/migrate.go

@@ -1,30 +1,48 @@
 package etcdutil
 
 import (
-	"fmt"
-	"io/ioutil"
-	"net/http"
+	"context"
+	"crypto/tls"
+	"path/filepath"
 
+	"github.com/coreos/etcd/clientv3"
+	"github.com/coreos/etcd/pkg/transport"
 	"github.com/golang/glog"
+	"github.com/kubernetes-incubator/bootkube/pkg/asset"
 	"k8s.io/apimachinery/pkg/util/wait"
 )
 
 // WaitClusterReady waits the etcd server ready to serve client requests.
-func WaitClusterReady(endpoint string) error {
+func WaitClusterReady(endpoint string, etcdTLS *tls.Config) error {
+	cfg := clientv3.Config{
+		Endpoints:   []string{endpoint},
+		TLS:         etcdTLS,
+		DialTimeout: pollInterval,
+	}
 	err := wait.Poll(pollInterval, pollTimeout, func() (bool, error) {
-		resp, err := http.Get(fmt.Sprintf("%s/version", endpoint))
+		etcdcli, err := clientv3.New(cfg)
 		if err != nil {
-			glog.Infof("could not read from etcd: %v", err)
+			glog.Infof("could not create etcd client, retrying later: %v", err)
 			return false, nil
 		}
-		defer resp.Body.Close()
-		body, err := ioutil.ReadAll(resp.Body)
-		if len(body) == 0 || err != nil {
-			glog.Infof("could not read from etcd: %v", err)
+		ctx, cancel := context.WithTimeout(context.Background(), pollInterval)
+		_, err = etcdcli.Get(ctx, "/")
+		cancel()
+		if err != nil {
+			glog.Infof("could not read from etcd, retrying later: %v", err)
 			return false, nil
 		}
 		return true, nil
 	})
 
 	return err
 }
+
+func makeTLSConfig(assetDir string) (*tls.Config, error) {
+	tlsInfo := transport.TLSInfo{
+		TrustedCAFile: filepath.Join(assetDir, asset.AssetPathEtcdCA),
+		CertFile:      filepath.Join(assetDir, asset.AssetPathEtcdClientCert),
+		KeyFile:       filepath.Join(assetDir, asset.AssetPathEtcdClientKey),
+	}
+	return tlsInfo.ClientConfig()
+}