Skip to content
This repository was archived by the owner on Mar 26, 2025. It is now read-only.

Commit b19ad62

Browse files
baluchickenbaluchicken
committed
Fix CC issues which happens when dowscaling multiple brokers or using not consecutive ids during upscale
1 parent 6acca4e commit b19ad62

File tree

6 files changed

+146
-108
lines changed

6 files changed

+146
-108
lines changed

‎api/v1beta1/common_types.go‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -66,8 +66,10 @@ const (
6666
Configured RackAwarenessState = "Configured"
6767
// WaitingForRackAwareness states the broker is waiting for the rack awareness config
6868
WaitingForRackAwareness RackAwarenessState = "WaitingForRackAwareness"
69-
// GracefulUpdateSucceeded states the broker is updated gracefully
70-
GracefulUpdateSucceeded CruiseControlState = "GracefulUpdateSucceeded"
69+
// GracefulUpscaleSucceeded states the broker is updated gracefully
70+
GracefulUpscaleSucceeded CruiseControlState = "GracefulUpscaleSucceeded"
71+
// GracefulUpscaleSucceeded states the broker is updated gracefully
72+
GracefulDownscaleSucceeded CruiseControlState = "GracefulDownscaleSucceeded"
7173
// GracefulUpdateRunning states the broker update task is still running in CC
7274
GracefulUpdateRunning CruiseControlState = "GracefulUpdateRunning"
7375
// GracefulUpdateFailed states the broker could not be updated gracefully

‎pkg/k8sutil/resource.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -128,7 +128,7 @@ func Reconcile(log logr.Logger, client runtimeClient.Client, desired runtime.Obj
128128
case *corev1.ConfigMap:
129129
// Only update status when configmap belongs to broker
130130
if id, ok := desired.(*corev1.ConfigMap).Labels["brokerId"]; ok {
131-
statusErr := UpdateBrokerStatus(client, id, cr, banzaicloudv1beta1.ConfigOutOfSync, log)
131+
statusErr := UpdateBrokerStatus(client, []string{id}, cr, banzaicloudv1beta1.ConfigOutOfSync, log)
132132
if statusErr != nil {
133133
return errors.WrapIfWithDetails(err, "updating status for resource failed", "kind", desiredType)
134134
}

‎pkg/k8sutil/status.go‎

Lines changed: 54 additions & 47 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ package k8sutil
1717
import (
1818
"context"
1919
"fmt"
20+
"strings"
2021
"time"
2122

2223
"emperror.dev/errors"
@@ -42,54 +43,10 @@ func IsMarkedForDeletion(m metav1.ObjectMeta) bool {
4243
}
4344

4445
// UpdateBrokerStatus updates the broker status with rack and configuration infos
45-
func UpdateBrokerStatus(c client.Client, brokerId string, cluster *v1beta1.KafkaCluster, state interface{}, logger logr.Logger) error {
46+
func UpdateBrokerStatus(c client.Client, brokerIds []string, cluster *v1beta1.KafkaCluster, state interface{}, logger logr.Logger) error {
4647
typeMeta := cluster.TypeMeta
4748

48-
if cluster.Status.BrokersState == nil {
49-
switch s := state.(type) {
50-
case banzaicloudv1beta1.RackAwarenessState:
51-
cluster.Status.BrokersState = map[string]banzaicloudv1beta1.BrokerState{brokerId: {RackAwarenessState: s}}
52-
case banzaicloudv1beta1.GracefulActionState:
53-
cluster.Status.BrokersState = map[string]banzaicloudv1beta1.BrokerState{brokerId: {GracefulActionState: s}}
54-
case banzaicloudv1beta1.ConfigurationState:
55-
cluster.Status.BrokersState = map[string]banzaicloudv1beta1.BrokerState{brokerId: {ConfigurationState: s}}
56-
}
57-
} else if val, ok := cluster.Status.BrokersState[brokerId]; ok {
58-
switch s := state.(type) {
59-
case banzaicloudv1beta1.RackAwarenessState:
60-
val.RackAwarenessState = s
61-
case banzaicloudv1beta1.GracefulActionState:
62-
val.GracefulActionState = s
63-
case banzaicloudv1beta1.ConfigurationState:
64-
val.ConfigurationState = s
65-
}
66-
cluster.Status.BrokersState[brokerId] = val
67-
} else {
68-
switch s := state.(type) {
69-
case banzaicloudv1beta1.RackAwarenessState:
70-
cluster.Status.BrokersState[brokerId] = banzaicloudv1beta1.BrokerState{RackAwarenessState: s}
71-
case banzaicloudv1beta1.GracefulActionState:
72-
cluster.Status.BrokersState[brokerId] = banzaicloudv1beta1.BrokerState{GracefulActionState: s}
73-
case banzaicloudv1beta1.ConfigurationState:
74-
cluster.Status.BrokersState[brokerId] = banzaicloudv1beta1.BrokerState{ConfigurationState: s}
75-
}
76-
}
77-
78-
err := c.Status().Update(context.Background(), cluster)
79-
if apierrors.IsNotFound(err) {
80-
err = c.Update(context.Background(), cluster)
81-
}
82-
if err != nil {
83-
if !apierrors.IsConflict(err) {
84-
return errors.WrapIff(err, "could not update Kafka broker %s state", brokerId)
85-
}
86-
err := c.Get(context.TODO(), types.NamespacedName{
87-
Namespace: cluster.Namespace,
88-
Name: cluster.Name,
89-
}, cluster)
90-
if err != nil {
91-
return errors.WrapIf(err, "could not get config for updating status")
92-
}
49+
for _, brokerId := range brokerIds {
9350

9451
if cluster.Status.BrokersState == nil {
9552
switch s := state.(type) {
@@ -120,13 +77,63 @@ func UpdateBrokerStatus(c client.Client, brokerId string, cluster *v1beta1.Kafka
12077
cluster.Status.BrokersState[brokerId] = banzaicloudv1beta1.BrokerState{ConfigurationState: s}
12178
}
12279
}
80+
}
81+
82+
err := c.Status().Update(context.Background(), cluster)
83+
if apierrors.IsNotFound(err) {
84+
err = c.Update(context.Background(), cluster)
85+
}
86+
if err != nil {
87+
if !apierrors.IsConflict(err) {
88+
return errors.WrapIff(err, "could not update Kafka broker(s) %s state", strings.Join(brokerIds, ","))
89+
}
90+
err := c.Get(context.TODO(), types.NamespacedName{
91+
Namespace: cluster.Namespace,
92+
Name: cluster.Name,
93+
}, cluster)
94+
if err != nil {
95+
return errors.WrapIf(err, "could not get config for updating status")
96+
}
97+
98+
for _, brokerId := range brokerIds {
99+
100+
if cluster.Status.BrokersState == nil {
101+
switch s := state.(type) {
102+
case banzaicloudv1beta1.RackAwarenessState:
103+
cluster.Status.BrokersState = map[string]banzaicloudv1beta1.BrokerState{brokerId: {RackAwarenessState: s}}
104+
case banzaicloudv1beta1.GracefulActionState:
105+
cluster.Status.BrokersState = map[string]banzaicloudv1beta1.BrokerState{brokerId: {GracefulActionState: s}}
106+
case banzaicloudv1beta1.ConfigurationState:
107+
cluster.Status.BrokersState = map[string]banzaicloudv1beta1.BrokerState{brokerId: {ConfigurationState: s}}
108+
}
109+
} else if val, ok := cluster.Status.BrokersState[brokerId]; ok {
110+
switch s := state.(type) {
111+
case banzaicloudv1beta1.RackAwarenessState:
112+
val.RackAwarenessState = s
113+
case banzaicloudv1beta1.GracefulActionState:
114+
val.GracefulActionState = s
115+
case banzaicloudv1beta1.ConfigurationState:
116+
val.ConfigurationState = s
117+
}
118+
cluster.Status.BrokersState[brokerId] = val
119+
} else {
120+
switch s := state.(type) {
121+
case banzaicloudv1beta1.RackAwarenessState:
122+
cluster.Status.BrokersState[brokerId] = banzaicloudv1beta1.BrokerState{RackAwarenessState: s}
123+
case banzaicloudv1beta1.GracefulActionState:
124+
cluster.Status.BrokersState[brokerId] = banzaicloudv1beta1.BrokerState{GracefulActionState: s}
125+
case banzaicloudv1beta1.ConfigurationState:
126+
cluster.Status.BrokersState[brokerId] = banzaicloudv1beta1.BrokerState{ConfigurationState: s}
127+
}
128+
}
129+
}
123130

124131
err = c.Status().Update(context.Background(), cluster)
125132
if apierrors.IsNotFound(err) {
126133
err = c.Update(context.Background(), cluster)
127134
}
128135
if err != nil {
129-
return errors.WrapIff(err, "could not update Kafka clusters broker %s state", brokerId)
136+
return errors.WrapIff(err, "could not update Kafka clusters broker(s) %s state", strings.Join(brokerIds, ","))
130137
}
131138
}
132139
// update loses the typeMeta of the config that's used later when setting ownerrefs

0 commit comments

Comments
 (0)