Skip to content

chores: refactor replica state #276

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
GrigoryPervakov merged 2 commits into from
Jul 22, 2026
Merged

chores: refactor replica state #276

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
64a08db
chores: refactor replica state
GrigoryPervakov Jul 21, 2026
1a23f9d
Potential fix for pull request finding
GrigoryPervakov Jul 22, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
59 changes: 32 additions & 27 deletions internal/controller/clickhouse/sync.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,6 @@ type replicaState struct {
chctrl.ReplicaState

ReloadError error
Error bool
}

func (r replicaState) Ready() bool {
Expand All @@ -51,28 +50,32 @@ func (r replicaState) Ready() bool {
}

func (r replicaState) HasDiff(rev chctrl.RevisionState) bool {
return rev.ReplicaHasDiff(r.ReplicaState)
return rev.ReplicaHasDiff(r.ReplicaResources)
}

func (r replicaState) UpdateStage(rev chctrl.RevisionState) chctrl.ReplicaUpdateStage {
if r.STS == nil {
return chctrl.StageNotExists
}

if r.Error {
if r.StartupError != nil {
return chctrl.StageError
}

if !r.Updated() {
if !r.StatefulSetUpdated() {
return chctrl.StageUpdating
}

if r.HasDiff(rev) {
return chctrl.StageHasDiff
}

if !r.Reloaded(rev.ReloadConfigRevision) && r.ReloadError == nil {
return chctrl.StageUpdating
}

// ReloadError is a retryable connectivity state, not a replica error:
if !r.Ready() || r.ReloadError != nil || !r.Reloaded(rev.ReloadConfigRevision) {
if !r.Ready() || r.ReloadError != nil {
return chctrl.StageNotReadyUpToDate
}

Expand Down Expand Up @@ -461,19 +464,19 @@ func (r *clickhouseReconciler) reconcileActiveReplicaStatus(ctx context.Context,
return v1.ClickHouseReplicaID{}, replicaState{}, fmt.Errorf("get replica ID from StatefulSet labels: %w", err)
}

hasError, err := chctrl.CheckPodError(ctx, log, r.GetClient(), &sts)
pod, err := chctrl.GetReplicaPod(ctx, log, r.GetClient(), &sts)
if err != nil {
log.Warn("failed to check replica pod error", "statefulset", sts.Name, "error", err)

hasError = true
return id, replicaState{}, fmt.Errorf("get replica pod: %w", err)
}

startupErr := chctrl.PodStartupError(pod)

var (
probe replicaProbe
reloadErr error
)

if !hasError && sts.Status.ReadyReplicas > 0 && r.commander != nil {
if startupErr == nil && sts.Status.ReadyReplicas > 0 && r.commander != nil {
ctx, cancel := context.WithTimeout(ctx, chctrl.LoadReplicaStateTimeout)
defer cancel()

Expand All @@ -498,20 +501,22 @@ func (r *clickhouseReconciler) reconcileActiveReplicaStatus(ctx context.Context,

return id, replicaState{
ReloadError: reloadErr,
Error: hasError,
replicaProbe: probe,
ReplicaState: chctrl.ReplicaState{
STS: &sts,
CFG: configMaps[id],
PVCs: pvcs,
ReplicaResources: chctrl.ReplicaResources{
STS: &sts,
CFG: configMaps[id],
PVCs: pvcs,
},
Pod: pod,
StartupError: startupErr,
},
}, nil
})

for id, res := range execResults {
if res.Err != nil {
log.Info("failed to load replica state", "error", res.Err, "replica_id", id)
continue
return chctrl.StepResult{}, fmt.Errorf("load replica state %s: %w", id, res.Err)
}

if _, ok := r.ReplicaState[id]; !ok {
Expand Down Expand Up @@ -542,7 +547,7 @@ func (r *clickhouseReconciler) reconcileWarnings(ctx context.Context, log ctrlut

results := ctrlutil.ExecuteParallel(ids, func(id v1.ClickHouseReplicaID) (v1.ClickHouseReplicaID, struct{}, error) {
replica := r.ReplicaState[id]
if replica.Error || replica.STS == nil || replica.STS.Status.ReadyReplicas == 0 {
if replica.StartupError != nil || replica.STS == nil || replica.STS.Status.ReadyReplicas == 0 {
return id, struct{}{}, nil
}

Expand Down Expand Up @@ -652,7 +657,7 @@ func (r *clickhouseReconciler) reconcileClusterRevisions(ctx context.Context, lo
id := v1.ClickHouseReplicaID{ShardID: shard, Index: index}
replica := r.ReplicaState[id]

if replica.HasDiff(r.revs) || !replica.Updated() {
if replica.HasDiff(r.revs) || !replica.StatefulSetUpdated() {
notUpdated = append(notUpdated, id.String())
}

Expand Down Expand Up @@ -875,7 +880,7 @@ func (r *clickhouseReconciler) reconcileDatabaseSync(ctx context.Context, log ct
}

func (r *clickhouseReconciler) reconcileCleanUp(ctx context.Context, log ctrlutil.Logger) (chctrl.StepResult, error) {
var replicasToRemove = map[v1.ClickHouseReplicaID]chctrl.ReplicaState{}
var replicasToRemove = map[v1.ClickHouseReplicaID]chctrl.ReplicaResources{}

configMaps, err := chctrl.ListReplicaResources[v1.ClickHouseReplicaID, *corev1.ConfigMap, *corev1.ConfigMapList](ctx, &r.ResourceManager, v1.ClickHouseIDFromLabels)
if err != nil {
Expand All @@ -887,7 +892,7 @@ func (r *clickhouseReconciler) reconcileCleanUp(ctx context.Context, log ctrluti
continue
}

replicasToRemove[id] = chctrl.ReplicaState{
replicasToRemove[id] = chctrl.ReplicaResources{
CFG: configMap,
}
}
Expand Down Expand Up @@ -941,8 +946,9 @@ func (r *clickhouseReconciler) reconcileCleanUp(ctx context.Context, log ctrluti

func (r *clickhouseReconciler) evaluateReplicaConditions() {
var (
errorIDs, notReadyIDs []string
notReadyShards []int32
startupErrors = map[string]string{}
notReadyIDs []string
notReadyShards []int32
)

for shard := range r.Cluster.Shards() {
Expand All @@ -951,8 +957,8 @@ func (r *clickhouseReconciler) evaluateReplicaConditions() {
id := v1.ClickHouseReplicaID{ShardID: shard, Index: index}
replica := r.ReplicaState[id]

if replica.Error {
errorIDs = append(errorIDs, id.String())
if replica.StartupError != nil {
startupErrors[id.String()] = *replica.StartupError
}

if !replica.Ready() {
Expand All @@ -974,7 +980,7 @@ func (r *clickhouseReconciler) evaluateReplicaConditions() {
exists := len(r.ReplicaState)
expected := int(r.Cluster.Replicas() * r.Cluster.Shards())

r.SetCondition(chctrl.ReplicaStartupCondition(errorIDs))
r.SetCondition(chctrl.ReplicaStartupCondition(startupErrors))
r.SetCondition(chctrl.ClusterSizeCondition(exists, expected))

if r.commander == nil {
Expand Down Expand Up @@ -1028,8 +1034,7 @@ func (r *clickhouseReconciler) updateReplica(ctx context.Context, log ctrlutil.L
result, err := r.ReconcileReplicaResources(ctx, log, chctrl.ReplicaUpdateInput{
Revisions: r.revs,
Existing: replica.ReplicaState,
HasError: replica.Error,
Desired: chctrl.ReplicaState{
Desired: chctrl.ReplicaResources{
CFG: configMap,
STS: statefulSet,
PVCs: chctrl.DesiredPVCs(r.Cluster.Spec.DataVolumeClaimSpec, r.Cluster.Spec.AdditionalVolumeClaimTemplates),
Expand Down
18 changes: 15 additions & 3 deletions internal/controller/clickhouse/sync_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,9 @@ var _ = Describe("UpdateStage", func() {
ReloadConfigRevision: rev.ReloadConfigRevision,
UsersReloadConfigRevision: rev.ReloadConfigRevision,
},
ReplicaState: chctrl.ReplicaState{STS: sts, CFG: cfg},
ReplicaState: chctrl.ReplicaState{
ReplicaResources: chctrl.ReplicaResources{STS: sts, CFG: cfg},
},
}
}

Expand All @@ -62,6 +64,8 @@ var _ = Describe("UpdateStage", func() {

It("should treat a reload failure as NotReadyUpToDate, not Error", func() {
replica := settledReplica()
replica.ReloadConfigRevision = ""
replica.UsersReloadConfigRevision = ""
replica.ReloadError = errors.New("dial tcp: connect: connection refused")

Expect(replica.UpdateStage(rev)).To(Equal(chctrl.StageNotReadyUpToDate))
Expand All @@ -76,10 +80,18 @@ var _ = Describe("UpdateStage", func() {
Expect(replica.UpdateStage(rev)).To(Equal(chctrl.StageHasDiff))
})

It("should report NotReadyUpToDate when the probe returned nothing", func() {
It("should report Updating while waiting for a reload result", func() {
replica := settledReplica()
replica.replicaProbe = replicaProbe{}

Expect(replica.UpdateStage(rev)).To(Equal(chctrl.StageNotReadyUpToDate))
Expect(replica.UpdateStage(rev)).To(Equal(chctrl.StageUpdating))
})

It("should wait for an existing StatefulSet update before a new diff", func() {
replica := settledReplica()
replica.STS.Status.ObservedGeneration = 0
ctrlutil.AddSpecHashToObject(replica.STS, "next-sts-rev")

Expect(replica.UpdateStage(rev)).To(Equal(chctrl.StageUpdating))
})
})
57 changes: 29 additions & 28 deletions internal/controller/keeper/sync.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,6 @@ import (
type replicaState struct {
chctrl.ReplicaState

Error bool
Status serverStatus
}

Expand All @@ -45,19 +44,19 @@ func (r replicaState) Ready(replicaCount int) bool {
}

func (r replicaState) HasDiff(rev chctrl.RevisionState) bool {
return rev.ReplicaHasDiff(r.ReplicaState)
return rev.ReplicaHasDiff(r.ReplicaResources)
}

func (r replicaState) UpdateStage(rev chctrl.RevisionState, replicaCount int) chctrl.ReplicaUpdateStage {
if r.STS == nil {
return chctrl.StageNotExists
}

if r.Error {
if r.StartupError != nil {
return chctrl.StageError
}

if !r.Updated() {
if !r.StatefulSetUpdated() {
return chctrl.StageUpdating
}

Expand Down Expand Up @@ -226,44 +225,46 @@ func (r *keeperReconciler) reconcileActiveReplicaStatus(ctx context.Context, log
return -1, replicaState{}, fmt.Errorf("get StatefulSet replica ID: %w", err)
}

hasError, err := chctrl.CheckPodError(ctx, log, r.GetClient(), &sts)
pod, err := chctrl.GetReplicaPod(ctx, log, r.GetClient(), &sts)
if err != nil {
log.Warn("failed to check replica pod error", "statefulset", sts.Name, "error", err)

hasError = true
return id, replicaState{}, fmt.Errorf("get replica pod: %w", err)
}

startupErr := chctrl.PodStartupError(pod)

var status serverStatus
if !hasError && sts.Status.ReadyReplicas > 0 {
if startupErr == nil && sts.Status.ReadyReplicas > 0 {
ctx, cancel := context.WithTimeout(ctx, chctrl.LoadReplicaStateTimeout)
defer cancel()

status = getServerStatus(ctx, log.With("replica_id", id), r.Dialer, r.Cluster.HostnameByID(id),
r.Cluster.Spec.Settings.TLS.Required)
}

pvcs, pvcErr := r.GetReplicaPVCs(ctx, &sts)
if pvcErr != nil {
log.Error(pvcErr, "failed to get PVCs for replica", "replica_id", id)
pvcs, err := r.GetReplicaPVCs(ctx, &sts)
if err != nil {
log.Error(err, "failed to get PVCs for replica", "replica_id", id)
}

log.Debug("load replica state done", "replica_id", id, "statefulset", sts.Name)

return id, replicaState{
Error: hasError,
Status: status,

ReplicaState: chctrl.ReplicaState{
STS: &sts,
CFG: configMaps[id],
PVCs: pvcs,
ReplicaResources: chctrl.ReplicaResources{
STS: &sts,
CFG: configMaps[id],
PVCs: pvcs,
},
Pod: pod,
StartupError: startupErr,
},
}, nil
})
for id, res := range execResults {
if res.Err != nil {
log.Info("failed to load replica state", "error", res.Err, "replica_id", id)
continue
return chctrl.StepResult{}, fmt.Errorf("load replica state %d: %w", id, res.Err)
}

if _, ok := r.ReplicaState[id]; !ok {
Expand All @@ -285,7 +286,6 @@ func (r *keeperReconciler) reconcileActiveReplicaStatus(ctx context.Context, log
if _, exists := r.ReplicaState[id]; !exists {
log.Info("adding missing replica from quorum config", "replica_id", id)
r.ReplicaState[id] = replicaState{
Error: false,
Status: serverStatus{},
}
}
Expand Down Expand Up @@ -556,16 +556,18 @@ func (r *keeperReconciler) reconcileCleanUp(ctx context.Context, log ctrlutil.Lo
}

func (r *keeperReconciler) evaluateReplicaConditions() {
var errorIDs, notReadyIDs, notUpdatedIDs []string

replicasByMode := map[string][]v1.KeeperReplicaID{}
var (
notReadyIDs, notUpdatedIDs []string
startupErrors = map[string]string{}
replicasByMode = map[string][]v1.KeeperReplicaID{}
)

r.Cluster.Status.ReadyReplicas = 0
for id, replica := range r.ReplicaState {
idStr := fmt.Sprintf("%d", id)

if replica.Error {
errorIDs = append(errorIDs, idStr)
if replica.StartupError != nil {
startupErrors[idStr] = *replica.StartupError
}

if !replica.Ready(len(r.ReplicaState)) {
Expand All @@ -575,12 +577,12 @@ func (r *keeperReconciler) evaluateReplicaConditions() {
replicasByMode[replica.Status.ServerState] = append(replicasByMode[replica.Status.ServerState], id)
}

if replica.HasDiff(r.revs) || !replica.Updated() {
if replica.HasDiff(r.revs) || !replica.StatefulSetUpdated() {
notUpdatedIDs = append(notUpdatedIDs, idStr)
}
}

r.SetCondition(chctrl.ReplicaStartupCondition(errorIDs))
r.SetCondition(chctrl.ReplicaStartupCondition(startupErrors))
r.SetCondition(chctrl.HealthyCondition(notReadyIDs))
r.SetCondition(chctrl.ConfigSyncCondition(nil, notUpdatedIDs, nil))
r.evaluateVersionConditions(len(notUpdatedIDs) > 0)
Expand Down Expand Up @@ -731,8 +733,7 @@ func (r *keeperReconciler) updateReplica(ctx context.Context, log ctrlutil.Logge
result, err := r.ReconcileReplicaResources(ctx, log, chctrl.ReplicaUpdateInput{
Revisions: r.revs,
Existing: replica.ReplicaState,
HasError: replica.Error,
Desired: chctrl.ReplicaState{
Desired: chctrl.ReplicaResources{
CFG: configMap,
STS: statefulSet,
PVCs: chctrl.DesiredPVCs(r.Cluster.Spec.DataVolumeClaimSpec, nil),
Expand Down
Loading
Loading