Extended status check for reconciliation

internal/controller/etcdcluster_controller.go

@@ -21,6 +21,7 @@ import (
 	goerrors "errors"
 	"fmt"
 
+	clientv3 "go.etcd.io/etcd/client/v3"
 	policyv1 "k8s.io/api/policy/v1"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 
@@ -31,6 +32,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
@@ -48,6 +50,7 @@ type EtcdClusterReconciler struct {
 // +kubebuilder:rbac:groups=etcd.aenix.io,resources=etcdclusters,verbs=get;list;watch;create;update;patch;delete
 // +kubebuilder:rbac:groups=etcd.aenix.io,resources=etcdclusters/status,verbs=get;update;patch
 // +kubebuilder:rbac:groups=etcd.aenix.io,resources=etcdclusters/finalizers,verbs=update
+// +kubebuilder:rbac:groups="",resources=endpoints,verbs=get;list;watch
 // +kubebuilder:rbac:groups="",resources=configmaps,verbs=get;list;watch;create;update;watch;delete;patch
 // +kubebuilder:rbac:groups="",resources=services,verbs=get;create;delete;update;patch;list;watch
 // +kubebuilder:rbac:groups="apps",resources=statefulsets,verbs=get;create;delete;update;patch;list;watch
@@ -72,6 +75,60 @@ func (r *EtcdClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 		return reconcile.Result{}, nil
 	}
 
+	// create two services and the pdb
+	{
+		c := make(chan error, 3)
+		go func(chan<- error) {
+			err := factory.CreateOrUpdateClientService(ctx, instance, r.Client)
+			if err != nil {
+				err = fmt.Errorf("couldn't ensure client service: %w", err)
+			}
+			c <- err
+		}(c)
+		go func(chan<- error) {
+			err := factory.CreateOrUpdateHeadlessService(ctx, instance, r.Client)
+			if err != nil {
+				err = fmt.Errorf("couldn't ensure headless service: %w", err)
+			}
+			c <- err
+		}(c)
+		go func(chan<- error) {
+			err := factory.CreateOrUpdatePdb(ctx, instance, r.Client)
+			if err != nil {
+				err = fmt.Errorf("couldn't ensure headless service: %w", err)
+			}
+			c <- err
+		}(c)
+		for i := 0; i < 3; i++ {
+			if err := <-c; err != nil {
+				return ctrl.Result{}, err
+			}
+		}
+	}
+	var c clientv3.Client
+	{
+		var ep corev1.Endpoints
+		err := r.Get(ctx, types.NamespacedName{Name: factory.GetHeadlessServiceName(instance), Namespace: instance.Namespace}, &ep)
+		if err != nil {
+			return ctrl.Result{}, err
+		}
+		names := map[string]struct{}{}
+		urls := make([]string, 0, 8)
+		for _, v := range ep.Subsets {
+			for _, addr := range v.Addresses {
+				names[addr.Hostname] = struct{}{}
+			}
+			for _, addr := range v.NotReadyAddresses {
+				names[addr.Hostname] = struct{}{}
+			}
+		}
+		clientv3.New()
+		if instance.Spec.Security
+		for name := range names {
+			urls = append(urls, name)
+		}
+	}
+
 	// fill conditions
 	if len(instance.Status.Conditions) == 0 {
 		factory.FillConditions(instance)