Add remaining changes to PR

This commit includes all changes for feature #181 that have not been split up into small stacked PRs. It should not be merged and will later be undone and split into smaller logical chunks of work.

internal/controller/etcdcluster_controller.go

@@ -91,6 +91,7 @@ func (r *EtcdClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 	}
 
 	state := observables{}
+	state.instance = instance
 
 	// create two services and the pdb
 	err = r.ensureUnconditionalObjects(ctx, instance)
@@ -112,11 +113,40 @@ func (r *EtcdClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 	}
 	state.endpointsFound = clusterClient != nil && singleClients != nil
 
+	if clusterClient != nil {
+		state.endpoints = clusterClient.Endpoints()
+	}
+
+	// fetch PVCs
+	state.pvcs, err = factory.PVCs(ctx, instance, r.Client)
+	if err != nil {
+		return ctrl.Result{}, err
+	}
+
 	if !state.endpointsFound {
 		if !state.stsExists {
-			// TODO: happy path for new cluster creation
-			log.Debug(ctx, "happy path for new cluster creation (not yet implemented)")
+			return r.createClusterFromScratch(ctx, &state) // TODO: needs implementing
 		}
+		// else try reconciling the sts
+		existingSts := state.statefulSet.DeepCopy()
+		desiredSts := factory.TemplateStatefulSet() // TODO: needs implementing
+		existingSts.Spec.Template.Spec = desiredSts.Spec.Template.Spec
+		err := r.patchOrCreateObject(ctx, existingSts)
+		if err != nil {
+			return ctrl.Result{}, err
+		}
+		state.statefulSet = *existingSts
+		if existingSts.Status.ReadyReplicas != *existingSts.Spec.Replicas { // TODO: this check might not be the best to check for a ready sts
+			return ctrl.Result{}, fmt.Errorf("waiting for statefulset to become ready")
+		}
+		if *existingSts.Spec.Replicas > 0 {
+			return ctrl.Result{}, fmt.Errorf("reached an impossible state (no endpoints, but active pods)")
+		}
+		if *instance.Spec.Replicas == 0 {
+			// cluster successfully scaled down to zero
+			return ctrl.Result{}, nil
+		}
+		return r.scaleUpFromZero(ctx, &state) // TODO: needs implementing
 	}
 
 	// get status of every endpoint and member list from every endpoint
@@ -661,3 +691,33 @@ func (r *EtcdClusterReconciler) ensureUnconditionalObjects(ctx context.Context,
 	}
 	return nil
 }
+
+func (r *EtcdClusterReconciler) patchOrCreateObject(ctx context.Context, obj client.Object) error {
+	err := r.Patch(ctx, obj, client.Apply, &client.PatchOptions{FieldManager: "etcd-operator"}, client.ForceOwnership)
+	if err == nil {
+		return nil
+	}
+	if client.IgnoreNotFound(err) == nil {
+		err = r.Create(ctx, obj)
+	}
+	return err
+}
+
+// TODO!
+func (r *EtcdClusterReconciler) createClusterFromScratch(ctx context.Context, state *observables) (ctrl.Result, error) {
+	cm := factory.TemplateClusterStateConfigMap(state.instance, "new", state.desiredReplicas())
+	err := ctrl.SetControllerReference(state.instance, cm, r.Scheme)
+	if err != nil {
+		return ctrl.Result{}, err
+	}
+	err = r.patchOrCreateObject(ctx, cm)
+	if err != nil {
+		return ctrl.Result{}, err
+	}
+	panic("not yet implemented")
+}
+
+// TODO!
+func (r *EtcdClusterReconciler) scaleUpFromZero(ctx context.Context, state *observables) (ctrl.Result, error) {
+	panic("not yet implemented")
+}

internal/controller/factory/pvc.go

@@ -30,6 +30,14 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 )
 
+func PVCLabels(cluster *etcdaenixiov1alpha1.EtcdCluster) map[string]string {
+	labels := PodLabels(cluster)
+	for key, value := range cluster.Spec.Storage.VolumeClaimTemplate.Labels {
+		labels[key] = value
+	}
+	return labels
+}
+
 func GetPVCName(cluster *etcdaenixiov1alpha1.EtcdCluster) string {
 	if len(cluster.Spec.Storage.VolumeClaimTemplate.Name) > 0 {
 		return cluster.Spec.Storage.VolumeClaimTemplate.Name
@@ -38,6 +46,16 @@ func GetPVCName(cluster *etcdaenixiov1alpha1.EtcdCluster) string {
 	return "data"
 }
 
+func PVCs(ctx context.Context, cluster *etcdaenixiov1alpha1.EtcdCluster, cli client.Client) ([]corev1.PersistentVolumeClaim, error) {
+	labels := PVCLabels(cluster)
+	pvcs := corev1.PersistentVolumeClaimList{}
+	err := cli.List(ctx, &pvcs, client.MatchingLabels(labels))
+	if err != nil {
+		return nil, err
+	}
+	return pvcs.Items, nil
+}
+
 // UpdatePersistentVolumeClaims checks and updates the sizes of PVCs in an EtcdCluster if the specified storage size is larger than the current.
 func UpdatePersistentVolumeClaims(ctx context.Context, cluster *etcdaenixiov1alpha1.EtcdCluster, rclient client.Client) error {
 	labelSelector := labels.SelectorFromSet(labels.Set{

internal/controller/factory/statefulset.go

@@ -41,19 +41,30 @@ const (
 	defaultBackendQuotaBytesFraction = 0.95
 )
 
+// TODO!
+func TemplateStatefulSet() *appsv1.StatefulSet {
+	panic("not yet implemented")
+}
+
+func PodLabels(cluster *etcdaenixiov1alpha1.EtcdCluster) map[string]string {
+	labels := NewLabelsBuilder().WithName().WithInstance(cluster.Name).WithManagedBy()
+
+	if cluster.Spec.PodTemplate.Labels != nil {
+		for key, value := range cluster.Spec.PodTemplate.Labels {
+			labels[key] = value
+		}
+	}
+
+	return labels
+}
+
 func CreateOrUpdateStatefulSet(
 	ctx context.Context,
 	cluster *etcdaenixiov1alpha1.EtcdCluster,
 	rclient client.Client,
 ) error {
 	podMetadata := metav1.ObjectMeta{
-		Labels: NewLabelsBuilder().WithName().WithInstance(cluster.Name).WithManagedBy(),
-	}
-
-	if cluster.Spec.PodTemplate.Labels != nil {
-		for key, value := range cluster.Spec.PodTemplate.Labels {
-			podMetadata.Labels[key] = value
-		}
+		Labels: PodLabels(cluster),
 	}
 
 	if cluster.Spec.PodTemplate.Annotations != nil {

internal/controller/observables.go

@@ -2,8 +2,12 @@ package controller
 
 import (
 	"context"
+	"strconv"
+	"strings"
 	"sync"
 
+	"github.com/aenix-io/etcd-operator/api/v1alpha1"
+	"github.com/aenix-io/etcd-operator/pkg/set"
 	clientv3 "go.etcd.io/etcd/client/v3"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
@@ -22,13 +26,14 @@ type etcdStatus struct {
 // observables stores observations that the operator can make about
 // states of objects in kubernetes
 type observables struct {
+	instance       *v1alpha1.EtcdCluster
 	statefulSet    appsv1.StatefulSet
 	stsExists      bool
+	endpoints      []string
 	endpointsFound bool
 	etcdStatuses   []etcdStatus
 	clusterID      uint64
-	_              int
-	_              []corev1.PersistentVolumeClaim
+	pvcs           []corev1.PersistentVolumeClaim
 }
 
 // setClusterID populates the clusterID field based on etcdStatuses
@@ -43,15 +48,43 @@ func (o *observables) setClusterID() {
 
 // inSplitbrain compares clusterID field with clusterIDs in etcdStatuses.
 // If more than one unique ID is reported, cluster is in splitbrain.
+// Also if members have different opinions on the list of members, this is
+// also a splitbrain.
 func (o *observables) inSplitbrain() bool {
+	return o.clusterIDsAllEqual() && o.memberListsAllEqual()
+}
+
+func (o *observables) clusterIDsAllEqual() bool {
+	ids := set.New[uint64]()
 	for i := range o.etcdStatuses {
 		if o.etcdStatuses[i].endpointStatus != nil {
-			if o.clusterID != o.etcdStatuses[i].endpointStatus.Header.ClusterId {
-				return true
+			ids.Add(o.etcdStatuses[i].endpointStatus.Header.ClusterId)
+		}
+	}
+	return len(ids) <= 1
+}
+
+func (o *observables) memberListsAllEqual() bool {
+	type m struct {
+		Name string
+		ID   uint64
+	}
+	memberLists := make([]set.Set[m], 0, len(o.etcdStatuses))
+	for i := range o.etcdStatuses {
+		if o.etcdStatuses[i].memberList != nil {
+			memberSet := set.New[m]()
+			for _, member := range o.etcdStatuses[i].memberList.Members {
+				memberSet.Add(m{member.Name, member.ID})
 			}
+			memberLists = append(memberLists, memberSet)
+		}
+	}
+	for i := range memberLists {
+		if !memberLists[0].Equals(memberLists[i]) {
+			return false
 		}
 	}
-	return false
+	return true
 }
 
 // fill takes a single-endpoint client and populates the fields of etcdStatus
@@ -67,14 +100,74 @@ func (s *etcdStatus) fill(ctx context.Context, c *clientv3.Client) {
 	wg.Wait()
 }
 
-// TODO: make a real function
-func (o *observables) _() int {
+func (o *observables) pvcMaxIndex() (max int) {
+	max = -1
+	for i := range o.pvcs {
+		tokens := strings.Split(o.pvcs[i].Name, "-")
+		index, err := strconv.Atoi(tokens[len(tokens)-1])
+		if err != nil {
+			continue
+		}
+		if index > max {
+			max = index
+		}
+	}
+	return max
+}
+
+func (o *observables) endpointMaxIndex() (max int) {
+	for i := range o.endpoints {
+		tokens := strings.Split(o.endpoints[i], ":")
+		if len(tokens) < 2 {
+			continue
+		}
+		tokens = strings.Split(tokens[len(tokens)-2], "-")
+		index, err := strconv.Atoi(tokens[len(tokens)-1])
+		if err != nil {
+			continue
+		}
+		if index > max {
+			max = index
+		}
+	}
+	return max
+}
+
+// TODO: make a real function to determine the right number of replicas.
+// Hint: if ClientURL in the member list is absent, the member has not yet
+// started, but if the name field is populated, this is a member of the
+// initial cluster. If the name field is empty, this member has just been
+// added with etcdctl member add (or equivalent API call).
+func (o *observables) desiredReplicas() (max int) {
+	max = -1
 	if o.etcdStatuses != nil {
 		for i := range o.etcdStatuses {
 			if o.etcdStatuses[i].memberList != nil {
-				return len(o.etcdStatuses[i].memberList.Members)
+				for j := range o.etcdStatuses[i].memberList.Members {
+					tokens := strings.Split(o.etcdStatuses[i].memberList.Members[j].Name, "-")
+					index, err := strconv.Atoi(tokens[len(tokens)-1])
+					if err != nil {
+						continue
+					}
+					if index > max {
+						max = index
+					}
+				}
 			}
 		}
 	}
-	return 0
+	if max > -1 {
+		return max + 1
+	}
+
+	if epMax := o.endpointMaxIndex(); epMax > max {
+		max = epMax
+	}
+	if pvcMax := o.pvcMaxIndex(); pvcMax > max {
+		max = pvcMax
+	}
+	if max == -1 {
+		return int(*o.instance.Spec.Replicas)
+	}
+	return max + 1
 }