Extended status check for reconciliation

internal/controller/etcdcluster_controller.go

@@ -48,6 +48,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 +73,57 @@ func (r *EtcdClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request)
  return reconcile.Result{}, nil
  }
 
+ sts := appsv1.StatefulSet{}
+ // create two services and the pdb, try fetching the sts
+ {
+ c := make(chan error)
+ 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 pod disruption budget: %w", err)
+ }
+ c <- err
+ }(c)
+ go func(chan<- error) {
+ err := r.Get(ctx, req.NamespacedName, &sts)
+ if client.IgnoreNotFound(err) != nil {
+ err = fmt.Errorf("couldn't get statefulset: %w", err)
+ }
+ c <- err
+ }(c)
+ for i := 0; i < 4; i++ {
+ if err := <-c; err != nil {
+ return ctrl.Result{}, err
+ }
+ }
+ }
+ /*
+ clusterClient, singleClients, err := factory.NewEtcdClientSet(ctx, instance, r.Client)
+ if err != nil {
+ return ctrl.Result{}, err
+ }
+ if clusterClient == nil || singleClients == nil {
+ // TODO: no endpoints case
+
+ }
+ if sts.UID != "" {
+ r.Patch()
+ }
+ */
  // fill conditions
  if len(instance.Status.Conditions) == 0 {
  factory.FillConditions(instance)

internal/controller/factory/etcd_client.go

@@ -0,0 +1,59 @@
+package factory
+
+import (
+ "context"
+ "fmt"
+
+ "github.com/aenix-io/etcd-operator/api/v1alpha1"
+ clientv3 "go.etcd.io/etcd/client/v3"
+ v1 "k8s.io/api/core/v1"
+ "k8s.io/apimachinery/pkg/types"
+ "sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+func NewEtcdClientSet(ctx context.Context, cluster *v1alpha1.EtcdCluster, client client.Client) (*clientv3.Client, []*clientv3.Client, error) {
+ cfg, err := configFromCluster(ctx, cluster, client)
+ if err != nil {
+ return nil, nil, err
+ }
+ if len(cfg.Endpoints) == 0 {
+ return nil, nil, nil
+ }
+ eps := cfg.Endpoints
+ clusterClient, err := clientv3.New(cfg)
+ if err != nil {
+ return nil, nil, fmt.Errorf("error building etcd cluster client: %w", err)
+ }
+ singleClients := make([]*clientv3.Client, len(eps))
+ for i, ep := range eps {
+ cfg.Endpoints = []string{ep}
+ singleClients[i], err = clientv3.New(cfg)
+ if err != nil {
+ return nil, nil, fmt.Errorf("error building etcd single-endpoint client for endpoint %s: %w", ep, err)
+ }
+ }
+ return clusterClient, singleClients, nil
+}
+
+func configFromCluster(ctx context.Context, cluster *v1alpha1.EtcdCluster, client client.Client) (clientv3.Config, error) {
+ ep := v1.Endpoints{}
+ err := client.Get(ctx, types.NamespacedName{Name: GetHeadlessServiceName(cluster), Namespace: cluster.Namespace}, &ep)
+ if err != nil {
+ return clientv3.Config{}, 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{}{}
+ }
+ }
+ for name := range names {
+ urls = append(urls, fmt.Sprintf("%s:%s", name, "2379"))
+ }
+
+ return clientv3.Config{Endpoints: urls}, nil
+}