*: generate boot etcd manifest for self hosted etcd case

Author: xiang90

cmd/bootkube/recover.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/kubernetes-incubator/bootkube/pkg/bootkube"
 	"github.com/kubernetes-incubator/bootkube/pkg/recovery"
+	"github.com/kubernetes-incubator/bootkube/pkg/util/etcdutil"
 
 	"github.com/coreos/etcd/clientv3"
 	"github.com/spf13/cobra"
@@ -73,27 +74,30 @@ func runCmdRecover(cmd *cobra.Command, args []string) error {
 	case recoverOpts.etcdBackupFile != "":
 		bootkube.UserOutput("Attempting recovery using etcd backup file at %q...\n", recoverOpts.etcdBackupFile)
 
-		err := recovery.StartRecoveryEtcdForBackup(recoverOpts.podManifestPath, recoverOpts.etcdBackupFile)
+		err = recovery.StartRecoveryEtcdForBackup(recoverOpts.podManifestPath, recoverOpts.etcdBackupFile)
 		if err != nil {
 			return err
 		}
 		defer func() {
-			err := recovery.CleanRecoveryEtcd(recoverOpts.podManifestPath)
+			err = recovery.CleanRecoveryEtcd(recoverOpts.podManifestPath)
 			if err != nil {
 				bootkube.UserOutput("Failed to cleanup recovery etcd from the podManifestPath %v\n", recoverOpts.podManifestPath)
 			}
 		}()
 
 		bootkube.UserOutput("Waiting for etcd server to start...\n")
-		// TODO: better way to detect the startup...
-		time.Sleep(60 * time.Second)
 
-		recoverOpts.etcdServers = "http://localhost:32379"
+		err = etcdutil.WaitClusterReady(recovery.RecoveryEtcdClientAddr)
+		if err != nil {
+			return err
+		}
+
+		recoverOpts.etcdServers = recovery.RecoveryEtcdClientAddr
 		etcdClient, err := createEtcdClient()
 		if err != nil {
 			return err
 		}
-		backend = recovery.NewEtcdBackend(etcdClient, recoverOpts.etcdPrefix)
+		backend = recovery.NewSelfHostedEtcdBackend(etcdClient, recoverOpts.etcdPrefix, recoverOpts.etcdBackupFile)
 
 	default:
 		bootkube.UserOutput("Attempting recovery using apiserver at %q...\n", recoverOpts.kubeConfigPath)
@@ -123,6 +127,9 @@ func validateRecoverOpts(cmd *cobra.Command, args []string) error {
 	if recoverOpts.kubeConfigPath == "" {
 		return errors.New("missing required flag: --kubeconfig")
 	}
+	if recoverOpts.etcdBackupFile != "" && recoverOpts.podManifestPath == "" {
+		return errors.New("missing required flag: --pod-manifest-path (--etcd-backup-file flag is specified)")
+	}
 	return nil
 }
 

hack/multi-node/bootkube-test-recovery-self-hosted-etcd

@@ -0,0 +1,56 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+GLOG_v=${GLOG_v:-1}
+HOST=${HOST:-c1}
+
+if [ ! -d "cluster" ]; then
+  echo "Need some cluster assets to perform recovery; try running bootkube-up."
+fi
+
+if [ ! -f cluster/bootstrap-manifests/bootstrap-etcd.yaml ]; then
+  echo "ERROR: run bootkube-test-recovery for non self-hosted etcd Kubernetes cluster."
+  exit 1
+fi
+
+echo "Getting the etcd backup file"
+echo
+
+ssh -q -F ssh_config core@$HOST "sudo cp /var/etcd/kube-system-kube-etcd-0000/member/snap/db /home/core/etcdbackup"
+ssh -q -F ssh_config core@$HOST "sudo chown core:core /home/core/etcdbackup"
+scp -q -F ssh_config core@$HOST:/home/core/etcdbackup cluster/etcdbackup 
+
+echo
+echo "Destroying and re-creating the master node..."
+echo
+
+vagrant destroy -f $HOST
+vagrant up $HOST
+
+echo
+echo "As you can see, the cluster is now dead:"
+echo
+
+set -x
+! kubectl --kubeconfig=cluster/auth/kubeconfig get nodes
+{ set +x; } 2>/dev/null
+
+echo
+echo "Recovering the control plane from the etcd backup..."
+echo
+
+scp -q -F ssh_config ../../_output/bin/linux/bootkube cluster/auth/kubeconfig cluster/etcdbackup core@$HOST:/home/core
+ssh -q -F ssh_config core@$HOST "sudo GLOG_v=${GLOG_v} /home/core/bootkube recover \
+  --asset-dir=/home/core/recovered \
+  --etcd-backup-file=/home/core/etcdbackup \
+  --kubeconfig=/home/core/kubeconfig 2>> /home/core/recovery.log"
+
+echo
+echo "Running bootkube start..."
+echo
+
+ssh -q -F ssh_config core@$HOST "sudo GLOG_v=${GLOG_v} /home/core/bootkube start --asset-dir=/home/core/recovered 2>> /home/core/recovery.log"
+
+echo
+echo "The cluster should now be recovered. You should be able to access the cluster again using:"
+echo "kubectl --kubeconfig=cluster/auth/kubeconfig get nodes"

pkg/recovery/etcd.go

@@ -5,17 +5,22 @@ package recovery
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
+	"net/url"
 	"os"
 	"path"
 	"strings"
 
 	"github.com/kubernetes-incubator/bootkube/pkg/asset"
 
+	"github.com/coreos/etcd-operator/pkg/spec"
 	"github.com/coreos/etcd/clientv3"
+	"github.com/pborman/uuid"
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/pkg/api"
+	"k8s.io/client-go/pkg/apis/extensions/v1beta1"
 )
 
 // etcdBackend is a backend that extracts a controlPlane from an etcd instance.
@@ -57,6 +62,7 @@ func (s *etcdBackend) read(ctx context.Context) (*controlPlane, error) {
 			return nil, err
 		}
 	}
+
 	return cp, nil
 }
 
@@ -78,6 +84,19 @@ func (s *etcdBackend) get(ctx context.Context, key string, out runtime.Object, i
 	return decode(s.decoder, kv.Value, out)
 }
 
+func (s *etcdBackend) getBytes(ctx context.Context, key string) ([]byte, error) {
+	key = path.Join(s.pathPrefix, key)
+	getResp, err := s.client.KV.Get(ctx, key)
+	if err != nil {
+		return nil, err
+	}
+
+	if len(getResp.Kvs) == 0 {
+		return nil, fmt.Errorf("key not found: %s", key)
+	}
+	return getResp.Kvs[0].Value, nil
+}
+
 // list fetches a list runtime.Object from etcd located at key prefix `key`.
 func (s *etcdBackend) list(ctx context.Context, key string, listObj runtime.Object) error {
 	listPtr, err := meta.GetItemsPtr(listObj)
@@ -100,27 +119,173 @@ func (s *etcdBackend) list(ctx context.Context, key string, listObj runtime.Obje
 	return decodeList(elems, listPtr, s.decoder)
 }
 
-const assetPathRecoveryEtcd = "recovery-etcd.yaml"
+const (
+	assetPathRecoveryEtcd  = "recovery-etcd.yaml"
+	etcdTPRKey             = "ThirdPartyResourceData/etcd.coreos.com/clusters/kube-system/kube-etcd"
+	etcdMemberPodPrefix    = "pods/kube-system/kube-etcd-"
+	RecoveryEtcdClientAddr = "http://localhost:52379"
+)
+
+type etcdSelfhostedBackend struct {
+	*etcdBackend
+
+	backupPath string
+}
+
+// NewSelfHostedEtcdBackend constructs a new etcdBackend for the given client and pathPrefix, and backup file.
+func NewSelfHostedEtcdBackend(client *clientv3.Client, pathPrefix, backupPath string) Backend {
+	eb := &etcdBackend{
+		client:     client,
+		decoder:    api.Codecs.UniversalDecoder(),
+		pathPrefix: pathPrefix,
+	}
+
+	return &etcdSelfhostedBackend{
+		etcdBackend: eb,
+		backupPath:  backupPath,
+	}
+}
+
+// read implements Backend.read().
+func (s *etcdSelfhostedBackend) read(ctx context.Context) (*controlPlane, error) {
+	cp, err := s.etcdBackend.read(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	d, err := s.getBytes(ctx, etcdTPRKey)
+	if err != nil {
+		return nil, err
+	}
+
+	var tpr v1beta1.ThirdPartyResourceData
+	err = decode(s.decoder, d, &tpr)
+	if err != nil {
+		return nil, err
+	}
 
+	var kubeetcd spec.Cluster
+	err = json.Unmarshal(tpr.Data, &kubeetcd)
+	if err != nil {
+		return nil, err
+	}
+
+	etpr, err := createEtcdTPRAsset(kubeetcd)
+	if err != nil {
+		return nil, err
+	}
+	cp.tpr = etpr
+
+	serviceIP, err := getServiceIPFromClusterSpec(kubeetcd.Spec)
+	if err != nil {
+		return nil, err
+	}
+	eas := createBootEtcdAsset(s.pathPrefix, s.backupPath, serviceIP)
+	esas := createBootEtcdServiceAsset(serviceIP)
+	cp.bootEtcd = &eas
+	cp.bootEtcdService = &esas
+
+	return cp, nil
+}
+
+// StartRecoveryEtcdForBackup starts a recovery etcd container using given backup.
+// The started etcd server listens on RecoveryEtcdClientAddr.
 func StartRecoveryEtcdForBackup(p, backupPath string) error {
 	d, f := path.Split(backupPath)
 
 	config := struct {
 		Image      string
 		BackupFile string
 		BackupDir  string
+		ClientAddr string
 	}{
-		// TODO: this already exists in bootkube/cmd.
-		// do not duplicate this!
-		Image:      "quay.io/coreos/etcd:v3.1.6",
+		Image:      asset.DefaultImages.Etcd,
 		BackupFile: f,
 		BackupDir:  d,
+		ClientAddr: RecoveryEtcdClientAddr,
 	}
 
-	as := asset.MustCreateAssetFromTemplate(assetPathRecoveryEtcd, RecoveryEtcdTemplate, config)
+	as := asset.MustCreateAssetFromTemplate(assetPathRecoveryEtcd, recoveryEtcdTemplate, config)
 	return as.WriteFile(p)
 }
 
+// CleanRecoveryEtcd removes the recovery etcd static pod manifest and stops the recovery
+// etcd container.
 func CleanRecoveryEtcd(p string) error {
 	return os.Remove(path.Join(p, assetPathRecoveryEtcd))
 }
+
+func createBootEtcdAsset(pathPrefix, backupPath, serviceIP string) asset.Asset {
+	d, f := path.Split(backupPath)
+
+	config := struct {
+		Image             string
+		BackupFile        string
+		BackupDir         string
+		BootEtcdServiceIP string
+		TPRKey            string
+		MemberPodPrefix   string
+		ClusterToken      string
+	}{
+		Image:             asset.DefaultImages.Etcd,
+		BackupFile:        f,
+		BackupDir:         d,
+		BootEtcdServiceIP: serviceIP,
+		TPRKey:            path.Join(pathPrefix, etcdTPRKey),
+		MemberPodPrefix:   path.Join(pathPrefix, etcdMemberPodPrefix),
+		ClusterToken:      "bootkube-recovery-" + uuid.New(),
+	}
+
+	return asset.MustCreateAssetFromTemplate(asset.AssetPathBootstrapEtcd, bootFromBackupEtcdTemplate, config)
+}
+
+func createBootEtcdServiceAsset(serviceIP string) asset.Asset {
+	config := struct{ BootEtcdServiceIP string }{BootEtcdServiceIP: serviceIP}
+
+	return asset.MustCreateAssetFromTemplate(asset.AssetPathBootstrapEtcdService, recoveryEtcdSvcTemplate, config)
+}
+
+func createEtcdTPRAsset(s spec.Cluster) (*asset.Asset, error) {
+	clone := cloneEtcdClusterTPR(s)
+
+	data, err := json.Marshal(clone)
+	if err != nil {
+		return nil, err
+	}
+
+	return &asset.Asset{
+		Name: asset.AssetPathMigrateEtcdCluster,
+		Data: data,
+	}, nil
+}
+
+func getServiceIPFromClusterSpec(s spec.ClusterSpec) (string, error) {
+	ep := s.SelfHosted.BootMemberClientEndpoint
+	u, err := url.Parse(ep)
+	if err != nil {
+		return "", err
+	}
+	return stripPort(u.Host), nil
+}
+
+func cloneEtcdClusterTPR(s spec.Cluster) spec.Cluster {
+	var clone spec.Cluster
+	clone.Spec = s.Spec
+	clone.Metadata.SetName(s.Metadata.GetName())
+	clone.Metadata.SetNamespace(s.Metadata.GetNamespace())
+	clone.APIVersion = s.APIVersion
+	clone.Kind = s.Kind
+
+	return clone
+}
+
+func stripPort(hostport string) string {
+	colon := strings.IndexByte(hostport, ':')
+	if colon == -1 {
+		return hostport
+	}
+	if i := strings.IndexByte(hostport, ']'); i != -1 {
+		return strings.TrimPrefix(hostport[:i], "[")
+	}
+	return hostport[:colon]
+}