Implement etcd connection

.golangci.yml

@@ -89,6 +89,7 @@ linters-settings:
           - github.com/cluster-api-provider-k3s/cluster-api-k3s
 
           - github.com/google/uuid
+          - github.com/pkg/errors
   gci:
     sections:
       - standard

bootstrap/controllers/kthreesconfig_controller.go

@@ -43,6 +43,7 @@ import (
 
 	bootstrapv1 "github.com/cluster-api-provider-k3s/cluster-api-k3s/bootstrap/api/v1beta1"
 	"github.com/cluster-api-provider-k3s/cluster-api-k3s/pkg/cloudinit"
+	"github.com/cluster-api-provider-k3s/cluster-api-k3s/pkg/etcd"
 	"github.com/cluster-api-provider-k3s/cluster-api-k3s/pkg/k3s"
 	"github.com/cluster-api-provider-k3s/cluster-api-k3s/pkg/kubeconfig"
 	"github.com/cluster-api-provider-k3s/cluster-api-k3s/pkg/locking"
@@ -251,6 +252,16 @@ func (r *KThreesConfigReconciler) joinControlplane(ctx context.Context, scope *S
 		return err
 	}
 
+	if scope.Config.Spec.IsEtcdEmbedded() {
+		etcdProxyFile := bootstrapv1.File{
+			Path:        etcd.EtcdProxyDaemonsetYamlLocation,
+			Content:     etcd.EtcdProxyDaemonsetYaml,
+			Owner:       "root:root",
+			Permissions: "0640",
+		}
+		files = append(files, etcdProxyFile)
+	}
+
 	cpInput := &cloudinit.ControlPlaneInput{
 		BaseUserData: cloudinit.BaseUserData{
 			PreK3sCommands:  scope.Config.Spec.PreK3sCommands,
@@ -455,6 +466,16 @@ func (r *KThreesConfigReconciler) handleClusterNotInitialized(ctx context.Contex
 		return ctrl.Result{}, err
 	}
 
+	if scope.Config.Spec.IsEtcdEmbedded() {
+		etcdProxyFile := bootstrapv1.File{
+			Path:        etcd.EtcdProxyDaemonsetYamlLocation,
+			Content:     etcd.EtcdProxyDaemonsetYaml,
+			Owner:       "root:root",
+			Permissions: "0640",
+		}
+		files = append(files, etcdProxyFile)
+	}
+
 	cpinput := &cloudinit.ControlPlaneInput{
 		BaseUserData: cloudinit.BaseUserData{
 			PreK3sCommands:  scope.Config.Spec.PreK3sCommands,

controlplane/controllers/kthreescontrolplane_controller.go

@@ -61,6 +61,9 @@ type KThreesControlPlaneReconciler struct {
 	controller controller.Controller
 	recorder   record.EventRecorder
 
+	EtcdDialTimeout time.Duration
+	EtcdCallTimeout time.Duration
+
 	managementCluster         k3s.ManagementCluster
 	managementClusterUncached k3s.ManagementCluster
 }
@@ -290,11 +293,19 @@ func (r *KThreesControlPlaneReconciler) SetupWithManager(mgr ctrl.Manager) error
 	r.recorder = mgr.GetEventRecorderFor("k3s-control-plane-controller")
 
 	if r.managementCluster == nil {
-		r.managementCluster = &k3s.Management{Client: r.Client}
+		r.managementCluster = &k3s.Management{
+			Client:          r.Client,
+			EtcdDialTimeout: r.EtcdDialTimeout,
+			EtcdCallTimeout: r.EtcdCallTimeout,
+		}
 	}
 
 	if r.managementClusterUncached == nil {
-		r.managementClusterUncached = &k3s.Management{Client: mgr.GetAPIReader()}
+		r.managementClusterUncached = &k3s.Management{
+			Client:          mgr.GetAPIReader(),
+			EtcdDialTimeout: r.EtcdDialTimeout,
+			EtcdCallTimeout: r.EtcdCallTimeout,
+		}
 	}
 
 	return nil

controlplane/controllers/scale.go

@@ -114,32 +114,31 @@ func (r *KThreesControlPlaneReconciler) scaleDownControlPlane(
 		return result, err
 	}
 
-	// workloadCluster, err := r.managementCluster.GetWorkloadCluster(ctx, util.ObjectKey(cluster))
-	// if err != nil {
-	//	logger.Error(err, "Failed to create client to workload cluster")
-	//	return ctrl.Result{}, fmt.Errorf(err, "failed to create client to workload cluster")
-	// }
-
 	if machineToDelete == nil {
 		logger.Info("Failed to pick control plane Machine to delete")
 		return ctrl.Result{}, fmt.Errorf("failed to pick control plane Machine to delete: %w", err)
 	}
 
-	// TODO figure out etcd complexities
 	// If KCP should manage etcd, If etcd leadership is on machine that is about to be deleted, move it to the newest member available.
-	/**
 	if controlPlane.IsEtcdManaged() {
+		workloadCluster, err := r.managementCluster.GetWorkloadCluster(ctx, util.ObjectKey(cluster))
+		if err != nil {
+			logger.Error(err, "Failed to create client to workload cluster")
+			return ctrl.Result{}, fmt.Errorf("failed to create client to workload cluster: %w", err)
+		}
+
 		etcdLeaderCandidate := controlPlane.Machines.Newest()
 		if err := workloadCluster.ForwardEtcdLeadership(ctx, machineToDelete, etcdLeaderCandidate); err != nil {
 			logger.Error(err, "Failed to move leadership to candidate machine", "candidate", etcdLeaderCandidate.Name)
 			return ctrl.Result{}, err
 		}
+		logger.Info("etcd move etcd leader succeed, node to delete %s", machineToDelete.Status.NodeRef.Name)
 		if err := workloadCluster.RemoveEtcdMemberForMachine(ctx, machineToDelete); err != nil {
 			logger.Error(err, "Failed to remove etcd member for machine")
 			return ctrl.Result{}, err
 		}
+		logger.Info("etcd remove etcd member succeed, node to delete %s", machineToDelete.Status.NodeRef.Name)
 	}
-	**/
 
 	logger = logger.WithValues("machine", machineToDelete)
 	if err := r.Client.Delete(ctx, machineToDelete); err != nil && !apierrors.IsNotFound(err) {

controlplane/main.go

@@ -32,6 +32,7 @@ import (
 	bootstrapv1beta1 "github.com/cluster-api-provider-k3s/cluster-api-k3s/bootstrap/api/v1beta1"
 	controlplanev1beta1 "github.com/cluster-api-provider-k3s/cluster-api-k3s/controlplane/api/v1beta1"
 	"github.com/cluster-api-provider-k3s/cluster-api-k3s/controlplane/controllers"
+	"github.com/cluster-api-provider-k3s/cluster-api-k3s/pkg/etcd"
 )
 
 var (
@@ -53,6 +54,8 @@ func main() {
 	var metricsAddr string
 	var enableLeaderElection bool
 	var syncPeriod time.Duration
+	var etcdDialTimeout time.Duration
+	var etcdCallTimeout time.Duration
 
 	flag.StringVar(&metricsAddr, "metrics-addr", ":8080", "The address the metric endpoint binds to.")
 	flag.BoolVar(&enableLeaderElection, "enable-leader-election", false,
@@ -61,6 +64,13 @@ func main() {
 
 	flag.DurationVar(&syncPeriod, "sync-period", 10*time.Minute,
 		"The minimum interval at which watched resources are reconciled (e.g. 15m)")
+
+	flag.DurationVar(&etcdDialTimeout, "etcd-dial-timeout-duration", 10*time.Second,
+		"Duration that the etcd client waits at most to establish a connection with etcd")
+
+	flag.DurationVar(&etcdCallTimeout, "etcd-call-timeout-duration", etcd.DefaultCallTimeout,
+		"Duration that the etcd client waits at most for read and write operations to etcd.")
+
 	flag.Parse()
 
 	ctrl.SetLogger(zap.New(zap.UseDevMode(true)))
@@ -79,9 +89,11 @@ func main() {
 	}
 
 	if err = (&controllers.KThreesControlPlaneReconciler{
-		Client: mgr.GetClient(),
-		Log:    ctrl.Log.WithName("controllers").WithName("KThreesControlPlane"),
-		Scheme: mgr.GetScheme(),
+		Client:          mgr.GetClient(),
+		Log:             ctrl.Log.WithName("controllers").WithName("KThreesControlPlane"),
+		Scheme:          mgr.GetScheme(),
+		EtcdDialTimeout: etcdDialTimeout,
+		EtcdCallTimeout: etcdCallTimeout,
 	}).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "KThreesControlPlane")
 		os.Exit(1)

pkg/etcd/doc.go

@@ -0,0 +1,20 @@
+/*
+Copyright 2020 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+/*
+Package etcd provides a connection to an etcd member.
+*/
+package etcd

pkg/etcd/etcd-proxy.go

@@ -0,0 +1,10 @@
+package etcd
+
+import (
+	_ "embed"
+)
+
+const EtcdProxyDaemonsetYamlLocation = "/var/lib/rancher/k3s/server/manifests/etcd-proxy.yaml"
+
+//go:embed etcd-proxy.yaml
+var EtcdProxyDaemonsetYaml string

pkg/etcd/etcd-proxy.yaml

@@ -0,0 +1,50 @@
+apiVersion: apps/v1
+kind: DaemonSet
+metadata:
+  name: etcd-proxy
+  namespace: kube-system
+  labels:
+    app: etcd-proxy
+spec:
+  selector:
+    matchLabels:
+      app: etcd-proxy
+  template:
+    metadata:
+      labels:
+        app: etcd-proxy
+    spec:
+      nodeSelector:
+        node-role.kubernetes.io/etcd: "true"
+      tolerations:
+      - key: node-role.kubernetes.io/control-plane
+        operator: Exists
+        effect: NoSchedule
+      - key: node-role.kubernetes.io/master
+        operator: Exists
+        effect: NoSchedule
+      containers:
+      - name: etcd-proxy
+        image: alpine/socat
+        env:
+        - name: HOSTIP
+          valueFrom:
+            fieldRef:
+              fieldPath: status.hostIP
+        args: 
+        - TCP4-LISTEN:2379,fork,reuseaddr
+        - TCP4:$(HOSTIP):2379
+        resources:
+          limits:
+            memory: 200Mi
+          requests:
+            cpu: 100m
+            memory: 200Mi
+        volumeMounts:
+        - name: varlog
+          mountPath: /var/log
+      terminationGracePeriodSeconds: 30
+      volumes:
+      - name: varlog
+        hostPath:
+          path: /var/log