Add etcd settings to nodeup and protokube

nodeup/pkg/model/convenience.go

@@ -28,6 +28,7 @@ import (
 	"k8s.io/kops/upup/pkg/fi/nodeup/nodetasks"
 
 	"github.com/golang/glog"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
 // s is a helper that builds a *string from a string value
@@ -171,3 +172,8 @@ func addHostPathMapping(pod *v1.Pod, container *v1.Container, name, path string)
 
 	return &container.VolumeMounts[len(container.VolumeMounts)-1]
 }
+
+// convEtcdSettingsToMs converts etcd settings to a string rep of int milliseconds
+func convEtcdSettingsToMs(dur *metav1.Duration) string {
+	return strconv.FormatInt(dur.Nanoseconds()/1000000, 10)
+}

nodeup/pkg/model/protokube.go

@@ -192,22 +192,24 @@ type ProtokubeFlags struct {
 	Channels    []string `json:"channels,omitempty" flag:"channels"`
 	Cloud       *string  `json:"cloud,omitempty" flag:"cloud"`
 	// ClusterID flag is required only for vSphere cloud type, to pass cluster id information to protokube. AWS and GCE workflows ignore this flag.
-	ClusterID         *string  `json:"cluster-id,omitempty" flag:"cluster-id"`
-	Containerized     *bool    `json:"containerized,omitempty" flag:"containerized"`
-	DNSInternalSuffix *string  `json:"dnsInternalSuffix,omitempty" flag:"dns-internal-suffix"`
-	DNSProvider       *string  `json:"dnsProvider,omitempty" flag:"dns"`
-	DNSServer         *string  `json:"dns-server,omitempty" flag:"dns-server"`
-	EtcdImage         *string  `json:"etcd-image,omitempty" flag:"etcd-image"`
-	InitializeRBAC    *bool    `json:"initializeRBAC,omitempty" flag:"initialize-rbac"`
-	LogLevel          *int32   `json:"logLevel,omitempty" flag:"v"`
-	Master            *bool    `json:"master,omitempty" flag:"master"`
-	PeerTLSCaFile     *string  `json:"peer-ca,omitempty" flag:"peer-ca"`
-	PeerTLSCertFile   *string  `json:"peer-cert,omitempty" flag:"peer-cert"`
-	PeerTLSKeyFile    *string  `json:"peer-key,omitempty" flag:"peer-key"`
-	TLSCAFile         *string  `json:"tls-ca,omitempty" flag:"tls-ca"`
-	TLSCertFile       *string  `json:"tls-cert,omitempty" flag:"tls-cert"`
-	TLSKeyFile        *string  `json:"tls-key,omitempty" flag:"tls-key"`
-	Zone              []string `json:"zone,omitempty" flag:"zone"`
+	ClusterID                 *string  `json:"cluster-id,omitempty" flag:"cluster-id"`
+	Containerized             *bool    `json:"containerized,omitempty" flag:"containerized"`
+	DNSInternalSuffix         *string  `json:"dnsInternalSuffix,omitempty" flag:"dns-internal-suffix"`
+	DNSProvider               *string  `json:"dnsProvider,omitempty" flag:"dns"`
+	DNSServer                 *string  `json:"dns-server,omitempty" flag:"dns-server"`
+	EtcdImage                 *string  `json:"etcd-image,omitempty" flag:"etcd-image"`
+	EtcdLeaderElectionTimeout *string  `json:"etcd-election-timeout,omitempty" flag:"etcd-election-timeout"`
+	EtcdHearbeatInterval      *string  `json:"etcd-heartbeat-interval,omitempty" flag:"etcd-heartbeat-interval"`
+	InitializeRBAC            *bool    `json:"initializeRBAC,omitempty" flag:"initialize-rbac"`
+	LogLevel                  *int32   `json:"logLevel,omitempty" flag:"v"`
+	Master                    *bool    `json:"master,omitempty" flag:"master"`
+	PeerTLSCaFile             *string  `json:"peer-ca,omitempty" flag:"peer-ca"`
+	PeerTLSCertFile           *string  `json:"peer-cert,omitempty" flag:"peer-cert"`
+	PeerTLSKeyFile            *string  `json:"peer-key,omitempty" flag:"peer-key"`
+	TLSCAFile                 *string  `json:"tls-ca,omitempty" flag:"tls-ca"`
+	TLSCertFile               *string  `json:"tls-cert,omitempty" flag:"tls-cert"`
+	TLSKeyFile                *string  `json:"tls-key,omitempty" flag:"tls-key"`
+	Zone                      []string `json:"zone,omitempty" flag:"zone"`
 }
 
 // ProtokubeFlags is responsible for building the command line flags for protokube
@@ -216,12 +218,25 @@ func (t *ProtokubeBuilder) ProtokubeFlags(k8sVersion semver.Version) *ProtokubeF
 	// lets keep that for another PR and allow the version change
 	imageVersion := t.Cluster.Spec.EtcdClusters[0].Version
 
+	var leaderElectionTimeout string
+	var heartbeatInterval string
+
+	if v := t.Cluster.Spec.EtcdClusters[0].LeaderElectionTimeout; v != nil {
+		leaderElectionTimeout = convEtcdSettingsToMs(v)
+	}
+
+	if v := t.Cluster.Spec.EtcdClusters[0].HeartbeatInterval; v != nil {
+		heartbeatInterval = convEtcdSettingsToMs(v)
+	}
+
 	f := &ProtokubeFlags{
-		Channels:      t.NodeupConfig.Channels,
-		Containerized: fi.Bool(true),
-		EtcdImage:     s(fmt.Sprintf("gcr.io/google_containers/etcd:%s", imageVersion)),
-		LogLevel:      fi.Int32(4),
-		Master:        b(t.IsMaster),
+		Channels:                  t.NodeupConfig.Channels,
+		Containerized:             fi.Bool(true),
+		EtcdImage:                 s(fmt.Sprintf("gcr.io/google_containers/etcd:%s", imageVersion)),
+		EtcdLeaderElectionTimeout: s(leaderElectionTimeout),
+		EtcdHearbeatInterval:      s(heartbeatInterval),
+		LogLevel:                  fi.Int32(4),
+		Master:                    b(t.IsMaster),
 	}
 
 	// initialize rbac on Kubernetes >= 1.6 and master

pkg/apis/kops/cluster.go

@@ -292,6 +292,10 @@ type EtcdClusterSpec struct {
 	EnableEtcdTLS bool `json:"enableEtcdTLS,omitempty"`
 	// Version is the version of etcd to run i.e. 2.1.2, 3.0.17 etcd
 	Version string `json:"version,omitempty"`
+	// LeaderElectionTimeout is the time (in milliseconds) for an etcd leader election timeout
+	LeaderElectionTimeout *metav1.Duration `json:"leaderElectionTimeout,omitempty"`
+	// HeartbeatInterval is the time (in milliseconds) for an etcd heartbeat interval
+	HeartbeatInterval *metav1.Duration `json:"heartbeatInterval,omitempty"`
 }
 
 // EtcdMemberSpec is a specification for a etcd member

pkg/apis/kops/v1alpha1/cluster.go

@@ -291,6 +291,10 @@ type EtcdClusterSpec struct {
 	EnableEtcdTLS bool `json:"enableEtcdTLS,omitempty"`
 	// Version is the version of etcd to run i.e. 2.1.2, 3.0.17 etcd
 	Version string `json:"version,omitempty"`
+	// LeaderElectionTimeout is the time (in milliseconds) for an etcd leader election timeout
+	LeaderElectionTimeout *metav1.Duration `json:"leaderElectionTimeout,omitempty"`
+	// HeartbeatInterval is the time (in milliseconds) for an etcd heartbeat interval
+	HeartbeatInterval *metav1.Duration `json:"heartbeatInterval,omitempty"`
 }
 
 // EtcdMemberSpec is a specification for a etcd member

pkg/apis/kops/v1alpha1/zz_generated.conversion.go

@@ -1179,6 +1179,8 @@ func autoConvert_v1alpha1_EtcdClusterSpec_To_kops_EtcdClusterSpec(in *EtcdCluste
 	}
 	out.EnableEtcdTLS = in.EnableEtcdTLS
 	out.Version = in.Version
+	out.LeaderElectionTimeout = in.LeaderElectionTimeout
+	out.HeartbeatInterval = in.HeartbeatInterval
 	return nil
 }
 
@@ -1203,6 +1205,8 @@ func autoConvert_kops_EtcdClusterSpec_To_v1alpha1_EtcdClusterSpec(in *kops.EtcdC
 	}
 	out.EnableEtcdTLS = in.EnableEtcdTLS
 	out.Version = in.Version
+	out.LeaderElectionTimeout = in.LeaderElectionTimeout
+	out.HeartbeatInterval = in.HeartbeatInterval
 	return nil
 }
 

pkg/apis/kops/v1alpha1/zz_generated.deepcopy.go

@@ -1233,6 +1233,24 @@ func (in *EtcdClusterSpec) DeepCopyInto(out *EtcdClusterSpec) {
 			}
 		}
 	}
+	if in.LeaderElectionTimeout != nil {
+		in, out := &in.LeaderElectionTimeout, &out.LeaderElectionTimeout
+		if *in == nil {
+			*out = nil
+		} else {
+			*out = new(v1.Duration)
+			**out = **in
+		}
+	}
+	if in.HeartbeatInterval != nil {
+		in, out := &in.HeartbeatInterval, &out.HeartbeatInterval
+		if *in == nil {
+			*out = nil
+		} else {
+			*out = new(v1.Duration)
+			**out = **in
+		}
+	}
 	return
 }
 

pkg/apis/kops/v1alpha2/cluster.go

@@ -289,6 +289,10 @@ type EtcdClusterSpec struct {
 	EnableEtcdTLS bool `json:"enableEtcdTLS,omitempty"`
 	// Version is the version of etcd to run i.e. 2.1.2, 3.0.17 etcd
 	Version string `json:"version,omitempty"`
+	// LeaderElectionTimeout is the time (in milliseconds) for an etcd leader election timeout
+	LeaderElectionTimeout *metav1.Duration `json:"leaderElectionTimeout,omitempty"`
+	// HeartbeatInterval is the time (in milliseconds) for an etcd heartbeat interval
+	HeartbeatInterval *metav1.Duration `json:"heartbeatInterval,omitempty"`
 }
 
 // EtcdMemberSpec is a specification for a etcd member

pkg/apis/kops/v1alpha2/zz_generated.conversion.go

@@ -1278,6 +1278,8 @@ func autoConvert_v1alpha2_EtcdClusterSpec_To_kops_EtcdClusterSpec(in *EtcdCluste
 	}
 	out.EnableEtcdTLS = in.EnableEtcdTLS
 	out.Version = in.Version
+	out.LeaderElectionTimeout = in.LeaderElectionTimeout
+	out.HeartbeatInterval = in.HeartbeatInterval
 	return nil
 }
 
@@ -1302,6 +1304,8 @@ func autoConvert_kops_EtcdClusterSpec_To_v1alpha2_EtcdClusterSpec(in *kops.EtcdC
 	}
 	out.EnableEtcdTLS = in.EnableEtcdTLS
 	out.Version = in.Version
+	out.LeaderElectionTimeout = in.LeaderElectionTimeout
+	out.HeartbeatInterval = in.HeartbeatInterval
 	return nil
 }
 

pkg/apis/kops/v1alpha2/zz_generated.deepcopy.go

@@ -1243,6 +1243,24 @@ func (in *EtcdClusterSpec) DeepCopyInto(out *EtcdClusterSpec) {
 			}
 		}
 	}
+	if in.LeaderElectionTimeout != nil {
+		in, out := &in.LeaderElectionTimeout, &out.LeaderElectionTimeout
+		if *in == nil {
+			*out = nil
+		} else {
+			*out = new(v1.Duration)
+			**out = **in
+		}
+	}
+	if in.HeartbeatInterval != nil {
+		in, out := &in.HeartbeatInterval, &out.HeartbeatInterval
+		if *in == nil {
+			*out = nil
+		} else {
+			*out = new(v1.Duration)
+			**out = **in
+		}
+	}
 	return
 }
 

pkg/apis/kops/zz_generated.deepcopy.go

@@ -1402,6 +1402,24 @@ func (in *EtcdClusterSpec) DeepCopyInto(out *EtcdClusterSpec) {
 			}
 		}
 	}
+	if in.LeaderElectionTimeout != nil {
+		in, out := &in.LeaderElectionTimeout, &out.LeaderElectionTimeout
+		if *in == nil {
+			*out = nil
+		} else {
+			*out = new(v1.Duration)
+			**out = **in
+		}
+	}
+	if in.HeartbeatInterval != nil {
+		in, out := &in.HeartbeatInterval, &out.HeartbeatInterval
+		if *in == nil {
+			*out = nil
+		} else {
+			*out = new(v1.Duration)
+			**out = **in
+		}
+	}
 	return
 }
 

protokube/cmd/protokube/main.go

@@ -61,7 +61,8 @@ func run() error {
 	var zones []string
 	var applyTaints, initializeRBAC, containerized, master bool
 	var cloud, clusterID, dnsServer, dnsProviderID, dnsInternalSuffix, gossipSecret, gossipListen string
-	var flagChannels, tlsCert, tlsKey, tlsCA, peerCert, peerKey, peerCA, etcdImageSource string
+	var flagChannels, tlsCert, tlsKey, tlsCA, peerCert, peerKey, peerCA string
+	var etcdImageSource, etcdElectionTimeout, etcdHeartbeatInterval string
 
 	flag.BoolVar(&applyTaints, "apply-taints", applyTaints, "Apply taints to nodes based on the role")
 	flag.BoolVar(&containerized, "containerized", containerized, "Set if we are running containerized.")
@@ -82,6 +83,8 @@ func run() error {
 	flags.StringSliceVarP(&zones, "zone", "z", []string{}, "Configure permitted zones and their mappings")
 	flags.StringVar(&dnsProviderID, "dns", "aws-route53", "DNS provider we should use (aws-route53, google-clouddns, coredns)")
 	flags.StringVar(&etcdImageSource, "etcd-image", "gcr.io/google_containers/etcd:2.2.1", "Etcd Source Container Registry")
+	flags.StringVar(&etcdElectionTimeout, "etcd-election-timeout", etcdElectionTimeout, "time in ms for an election to timeout")
+	flags.StringVar(&etcdHeartbeatInterval, "etcd-heartbeat-interval", etcdHeartbeatInterval, "time in ms of a heartbeat interval")
 	flags.StringVar(&gossipSecret, "gossip-secret", gossipSecret, "Secret to use to secure gossip")
 
 	// Trick to avoid 'logging before flag.Parse' warning
@@ -282,22 +285,24 @@ func run() error {
 	}
 
 	k := &protokube.KubeBoot{
-		ApplyTaints:       applyTaints,
-		Channels:          channels,
-		DNS:               dnsProvider,
-		EtcdImageSource:   etcdImageSource,
-		InitializeRBAC:    initializeRBAC,
-		InternalDNSSuffix: dnsInternalSuffix,
-		InternalIP:        internalIP,
-		Kubernetes:        protokube.NewKubernetesContext(),
-		Master:            master,
-		ModelDir:          modelDir,
-		PeerCA:            peerCA,
-		PeerCert:          peerCert,
-		PeerKey:           peerKey,
-		TLSCA:             tlsCA,
-		TLSCert:           tlsCert,
-		TLSKey:            tlsKey,
+		ApplyTaints:           applyTaints,
+		Channels:              channels,
+		DNS:                   dnsProvider,
+		EtcdImageSource:       etcdImageSource,
+		EtcdElectionTimeout:   etcdElectionTimeout,
+		EtcdHeartbeatInterval: etcdHeartbeatInterval,
+		InitializeRBAC:        initializeRBAC,
+		InternalDNSSuffix:     dnsInternalSuffix,
+		InternalIP:            internalIP,
+		Kubernetes:            protokube.NewKubernetesContext(),
+		Master:                master,
+		ModelDir:              modelDir,
+		PeerCA:                peerCA,
+		PeerCert:              peerCert,
+		PeerKey:               peerKey,
+		TLSCA:                 tlsCA,
+		TLSCert:               tlsCert,
+		TLSKey:                tlsKey,
 	}
 
 	k.Init(volumes)

protokube/pkg/protokube/etcd_cluster.go

@@ -73,6 +73,10 @@ type EtcdCluster struct {
 	PeerCert string
 	// PeerKey is the path to a peer ca for etcd
 	PeerKey string
+	// ElectionTimeout is the leader election timeout
+	ElectionTimeout string
+	// HeartbeatInterval is the heartbeat interval
+	HeartbeatInterval string
 }
 
 // EtcdNode is a definition for the etcd node
@@ -97,21 +101,23 @@ func newEtcdController(kubeBoot *KubeBoot, v *Volume, spec *etcd.EtcdClusterSpec
 
 	cluster := &EtcdCluster{
 		// @TODO we need to deprecate this port and use 2379, but that would be a breaking change
-		ClientPort:      4001,
-		ClusterName:     "etcd-" + spec.ClusterKey,
-		CPURequest:      resource.MustParse("200m"),
-		DataDirName:     "data-" + spec.ClusterKey,
-		ImageSource:     kubeBoot.EtcdImageSource,
-		TLSCA:           kubeBoot.TLSCA,
-		TLSCert:         kubeBoot.TLSCert,
-		TLSKey:          kubeBoot.TLSKey,
-		PeerCA:          kubeBoot.PeerCA,
-		PeerCert:        kubeBoot.PeerCert,
-		PeerKey:         kubeBoot.PeerKey,
-		PeerPort:        2380,
-		PodName:         "etcd-server-" + spec.ClusterKey,
-		Spec:            spec,
-		VolumeMountPath: v.Mountpoint,
+		ClientPort:        4001,
+		ClusterName:       "etcd-" + spec.ClusterKey,
+		CPURequest:        resource.MustParse("200m"),
+		DataDirName:       "data-" + spec.ClusterKey,
+		ImageSource:       kubeBoot.EtcdImageSource,
+		TLSCA:             kubeBoot.TLSCA,
+		TLSCert:           kubeBoot.TLSCert,
+		TLSKey:            kubeBoot.TLSKey,
+		PeerCA:            kubeBoot.PeerCA,
+		PeerCert:          kubeBoot.PeerCert,
+		PeerKey:           kubeBoot.PeerKey,
+		PeerPort:          2380,
+		PodName:           "etcd-server-" + spec.ClusterKey,
+		Spec:              spec,
+		VolumeMountPath:   v.Mountpoint,
+		ElectionTimeout:   kubeBoot.EtcdElectionTimeout,
+		HeartbeatInterval: kubeBoot.EtcdHeartbeatInterval,
 	}
 
 	// We used to build this through text files ... it turns out to just be more complicated than code!

protokube/pkg/protokube/etcd_manifest.go

@@ -48,7 +48,7 @@ func BuildEtcdManifest(c *EtcdCluster) *v1.Pod {
 				"/bin/sh", "-c", "/usr/local/bin/etcd 2>&1 | /bin/tee -a /var/log/etcd.log",
 			},
 		}
-		// build the the environment variables for etcd service
+		// build the environment variables for etcd service
 		container.Env = buildEtcdEnvironmentOptions(c)
 
 		container.LivenessProbe = &v1.Probe{
@@ -170,6 +170,14 @@ func buildEtcdEnvironmentOptions(c *EtcdCluster) []v1.EnvVar {
 		{Name: "ETCD_INITIAL_CLUSTER_STATE", Value: "new"},
 		{Name: "ETCD_INITIAL_CLUSTER_TOKEN", Value: c.ClusterToken}}...)
 
+	// add timeout/hearbeat settings
+	if notEmpty(c.ElectionTimeout) {
+		options = append(options, v1.EnvVar{Name: "ETCD_ELECTION_TIMEOUT", Value: c.ElectionTimeout})
+	}
+	if notEmpty(c.HeartbeatInterval) {
+		options = append(options, v1.EnvVar{Name: "ETCD_HEARTBEAT_INTERVAL", Value: c.HeartbeatInterval})
+	}
+
 	// @check if we are using peer certificates
 	if notEmpty(c.PeerCA) {
 		options = append(options, []v1.EnvVar{

protokube/pkg/protokube/kube_boot.go

@@ -50,6 +50,10 @@ type KubeBoot struct {
 	ModelDir string
 	// Etcd container registry location.
 	EtcdImageSource string
+	// EtcdElectionTimeout is is the leader election timeout
+	EtcdElectionTimeout string
+	// EtcdHeartbeatInterval is the heartbeat interval
+	EtcdHeartbeatInterval string
 	// TLSCA is the path to a client ca for etcd
 	TLSCA string
 	// TLSCert is the path to a tls certificate for etcd

protokube/tests/integration/build_etcd_manifest/integration_test.go

@@ -34,6 +34,7 @@ func TestBuildEtcdManifest(t *testing.T) {
 	}{
 		{TestFile: "non_tls.yaml"},
 		{TestFile: "tls.yaml"},
+		{TestFile: "etcd_env_vars.yaml"},
 	}
 	for i, x := range cs {
 		cluster, expected := loadTestIntegration(t, path.Join("main", x.TestFile))