Skip to content

Commit c2dda98

Browse files
hvanHa Vanclaudedvaseekara
authored
feat: pvcRoles label for Kafka PVCs (#255)
* feature: PVC role labels * refactor: extract processRolesValue helper and sort roles for deterministic labels - Extract private processRolesValue() on BrokerConfig as a single source of truth for the role join logic used by both GetBrokerLabels (processRoles pod label) and GetPvcRolesLabelValue (pvcRoles PVC label) - Sort roles before joining so both labels are alphabetically ordered regardless of CR authoring order ([controller, broker] and [broker, controller] both produce broker_controller) - Add backfill log line in PVC reconcile path so operators can observe the one-time label backfill on existing PVCs after upgrade Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * Remove the log message * fix: update test expectation for sorted processRoles label Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> --------- Co-authored-by: Ha Van <red83362@adobe.com> Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com> Co-authored-by: dvaseekara <dvaseeka@redhat.com>
1 parent 02b274e commit c2dda98

5 files changed

Lines changed: 129 additions & 10 deletions

File tree

‎api/v1beta1/kafkacluster_types.go‎

Lines changed: 23 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ package v1beta1
1717

1818
import (
1919
"fmt"
20+
"sort"
2021
"strconv"
2122
"strings"
2223

@@ -44,6 +45,9 @@ const (
4445
// ProcessRolesKey is used to identify which process roles the Kafka pod has
4546
ProcessRolesKey = "processRoles"
4647

48+
// PvcRolesKey is used to identify which process roles a PVC serves (broker, controller, or broker_controller)
49+
PvcRolesKey = "pvcRoles"
50+
4751
// IsBrokerNodeKey is used to identify if the kafka pod is either a broker or a broker_controller
4852
IsBrokerNodeKey = "isBrokerNode"
4953

@@ -1025,12 +1029,23 @@ func (bConfig *BrokerConfig) GetBrokerAnnotations() map[string]string {
10251029
return util.CloneMap(bConfig.BrokerAnnotations)
10261030
}
10271031

1032+
// processRolesValue returns the joined role string used for both the processRoles pod label
1033+
// and the pvcRoles PVC label. In KRaft mode the roles are joined with "_"; in ZK mode always "broker".
1034+
func (bConfig *BrokerConfig) processRolesValue(kRaftMode bool) string {
1035+
if kRaftMode {
1036+
roles := append([]string(nil), bConfig.Roles...)
1037+
sort.Strings(roles)
1038+
return strings.Join(roles, "_")
1039+
}
1040+
return BrokerNodeProcessRole
1041+
}
1042+
10281043
// GetBrokerLabels returns the labels that are applied to broker pods
10291044
func (bConfig *BrokerConfig) GetBrokerLabels(kafkaClusterName string, brokerId int32, kRaftMode bool) map[string]string {
10301045
var kraftLabels map[string]string
10311046
if kRaftMode {
10321047
kraftLabels = map[string]string{
1033-
ProcessRolesKey: strings.Join(bConfig.Roles, "_"),
1048+
ProcessRolesKey: bConfig.processRolesValue(kRaftMode),
10341049
IsControllerNodeKey: fmt.Sprintf("%t", bConfig.IsControllerNode()),
10351050
IsBrokerNodeKey: fmt.Sprintf("%t", bConfig.IsBrokerNode()),
10361051
}
@@ -1137,6 +1152,13 @@ func (bConfig *BrokerConfig) IsCombinedNode() bool {
11371152
return bConfig.IsBrokerNode() && bConfig.IsControllerNode()
11381153
}
11391154

1155+
// GetPvcRolesLabelValue returns the value for the pvcRoles label on PVCs.
1156+
// In KRaft mode the value mirrors processRoles (e.g. "broker", "controller", "broker_controller").
1157+
// In ZooKeeper mode all nodes are brokers, so the value is always "broker".
1158+
func (bConfig *BrokerConfig) GetPvcRolesLabelValue(kRaftMode bool) string {
1159+
return bConfig.processRolesValue(kRaftMode)
1160+
}
1161+
11401162
// GetResources returns the broker specific Kubernetes resource
11411163
func (bConfig *BrokerConfig) GetResources() *corev1.ResourceRequirements {
11421164
if bConfig.Resources != nil {

‎api/v1beta1/kafkacluster_types_test.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -515,7 +515,7 @@ func TestGetBrokerLabels(t *testing.T) {
515515
BrokerIdLabelKey: strconv.Itoa(expectedBrokerId),
516516
KafkaCRLabelKey: expectedKafkaCRName,
517517
"test_label_key": "test_label_value",
518-
ProcessRolesKey: "controller_broker",
518+
ProcessRolesKey: "broker_controller",
519519
IsBrokerNodeKey: "true",
520520
IsControllerNodeKey: "true",
521521
},

‎pkg/resources/kafka/kafka.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -317,7 +317,7 @@ func (r *Reconciler) Reconcile(log logr.Logger) error {
317317
if storage.PvcSpec == nil {
318318
continue
319319
}
320-
o, err := r.pvc(broker.Id, index, storage)
320+
o, err := r.pvc(broker.Id, index, storage, brokerConfig, r.KafkaCluster.Spec.KRaftMode)
321321
if err != nil {
322322
return errors.WrapIfWithDetails(err, "failed to generate resource", "resources", "PersistentVolumeClaim")
323323
}

‎pkg/resources/kafka/pvc.go‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@ import (
3131
"github.com/Masterminds/sprig/v3"
3232
)
3333

34-
func (r *Reconciler) pvc(brokerId int32, storageIndex int, storage v1beta1.StorageConfig) (*corev1.PersistentVolumeClaim, error) {
34+
func (r *Reconciler) pvc(brokerId int32, storageIndex int, storage v1beta1.StorageConfig, brokerConfig *v1beta1.BrokerConfig, kRaftMode bool) (*corev1.PersistentVolumeClaim, error) {
3535
errCtx := []interface{}{v1beta1.BrokerIdLabelKey, brokerId, mountPathAnnotationKey, storage.MountPath}
3636

3737
pvcSpecYaml, err := yaml.Marshal(storage.PvcSpec)
@@ -67,7 +67,10 @@ func (r *Reconciler) pvc(brokerId int32, storageIndex int, storage v1beta1.Stora
6767
fmt.Sprintf(brokerStorageTemplate, r.KafkaCluster.Name, brokerId, storageIndex),
6868
apiutil.MergeLabels(
6969
apiutil.LabelsForKafka(r.KafkaCluster.Name),
70-
map[string]string{v1beta1.BrokerIdLabelKey: fmt.Sprintf("%d", brokerId)},
70+
map[string]string{
71+
v1beta1.BrokerIdLabelKey: fmt.Sprintf("%d", brokerId),
72+
v1beta1.PvcRolesKey: brokerConfig.GetPvcRolesLabelValue(kRaftMode),
73+
},
7174
),
7275
map[string]string{mountPathAnnotationKey: storage.MountPath}, r.KafkaCluster),
7376
Spec: pvcSpec,

‎pkg/resources/kafka/pvc_test.go‎

Lines changed: 99 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -47,13 +47,19 @@ func TestReconciler_pvc(t *testing.T) {
4747
},
4848
}
4949

50+
brokerConfig := &v1beta1.BrokerConfig{}
51+
5052
testCases := []struct {
5153
testName string
54+
brokerConfig *v1beta1.BrokerConfig
55+
kRaftMode bool
5256
storageConfig v1beta1.StorageConfig
5357
expectedPersistentVolumeClaim *corev1.PersistentVolumeClaim
5458
}{
5559
{
56-
testName: "storage config with no template",
60+
testName: "storage config with no template",
61+
brokerConfig: brokerConfig,
62+
kRaftMode: false,
5763
storageConfig: v1beta1.StorageConfig{
5864
MountPath: "/kafka-logs-1",
5965
PvcSpec: &corev1.PersistentVolumeClaimSpec{
@@ -71,6 +77,7 @@ func TestReconciler_pvc(t *testing.T) {
7177
v1beta1.AppLabelKey: "kafka",
7278
v1beta1.KafkaCRLabelKey: kafkaCluster.GetName(),
7379
v1beta1.BrokerIdLabelKey: "2",
80+
v1beta1.PvcRolesKey: "broker",
7481
},
7582
Annotations: map[string]string{
7683
"mountPath": "/kafka-logs-1",
@@ -84,7 +91,9 @@ func TestReconciler_pvc(t *testing.T) {
8491
},
8592
},
8693
{
87-
testName: "storage config with template",
94+
testName: "storage config with template",
95+
brokerConfig: brokerConfig,
96+
kRaftMode: false,
8897
storageConfig: v1beta1.StorageConfig{
8998
MountPath: "/kafka-logs-1",
9099
PvcSpec: &corev1.PersistentVolumeClaimSpec{
@@ -108,6 +117,7 @@ func TestReconciler_pvc(t *testing.T) {
108117
v1beta1.AppLabelKey: "kafka",
109118
v1beta1.KafkaCRLabelKey: kafkaCluster.GetName(),
110119
v1beta1.BrokerIdLabelKey: "2",
120+
v1beta1.PvcRolesKey: "broker",
111121
},
112122
Annotations: map[string]string{
113123
"mountPath": "/kafka-logs-1",
@@ -125,7 +135,9 @@ func TestReconciler_pvc(t *testing.T) {
125135
},
126136
},
127137
{
128-
testName: "storage config with template and very long mount path",
138+
testName: "storage config with template and very long mount path",
139+
brokerConfig: brokerConfig,
140+
kRaftMode: false,
129141
storageConfig: v1beta1.StorageConfig{
130142
MountPath: "/mountpath/that/exceeds63characters/kafka-logs-123456789123456789",
131143
PvcSpec: &corev1.PersistentVolumeClaimSpec{
@@ -147,6 +159,7 @@ func TestReconciler_pvc(t *testing.T) {
147159
v1beta1.AppLabelKey: "kafka",
148160
v1beta1.KafkaCRLabelKey: kafkaCluster.GetName(),
149161
v1beta1.BrokerIdLabelKey: "2",
162+
v1beta1.PvcRolesKey: "broker",
150163
},
151164
Annotations: map[string]string{
152165
"mountPath": "/mountpath/that/exceeds63characters/kafka-logs-123456789123456789",
@@ -164,7 +177,9 @@ func TestReconciler_pvc(t *testing.T) {
164177
},
165178
},
166179
{
167-
testName: "storage config with volume name template",
180+
testName: "storage config with volume name template",
181+
brokerConfig: brokerConfig,
182+
kRaftMode: false,
168183
storageConfig: v1beta1.StorageConfig{
169184
MountPath: "/kafka-logs-1",
170185
PvcSpec: &corev1.PersistentVolumeClaimSpec{
@@ -180,6 +195,7 @@ func TestReconciler_pvc(t *testing.T) {
180195
v1beta1.AppLabelKey: "kafka",
181196
v1beta1.KafkaCRLabelKey: kafkaCluster.GetName(),
182197
v1beta1.BrokerIdLabelKey: "2",
198+
v1beta1.PvcRolesKey: "broker",
183199
},
184200
Annotations: map[string]string{
185201
"mountPath": "/kafka-logs-1",
@@ -190,6 +206,84 @@ func TestReconciler_pvc(t *testing.T) {
190206
},
191207
},
192208
},
209+
{
210+
testName: "kraft controller-only node",
211+
brokerConfig: &v1beta1.BrokerConfig{Roles: []string{v1beta1.ControllerNodeProcessRole}},
212+
kRaftMode: true,
213+
storageConfig: v1beta1.StorageConfig{
214+
MountPath: "/kafka-logs-1",
215+
PvcSpec: &corev1.PersistentVolumeClaimSpec{},
216+
},
217+
expectedPersistentVolumeClaim: &corev1.PersistentVolumeClaim{
218+
ObjectMeta: metav1.ObjectMeta{
219+
Namespace: kafkaCluster.GetNamespace(),
220+
Name: "",
221+
GenerateName: fmt.Sprintf("%s-2-storage-1-", kafkaCluster.GetName()),
222+
Labels: map[string]string{
223+
v1beta1.AppLabelKey: "kafka",
224+
v1beta1.KafkaCRLabelKey: kafkaCluster.GetName(),
225+
v1beta1.BrokerIdLabelKey: "2",
226+
v1beta1.PvcRolesKey: "controller",
227+
},
228+
Annotations: map[string]string{
229+
"mountPath": "/kafka-logs-1",
230+
},
231+
},
232+
Spec: corev1.PersistentVolumeClaimSpec{},
233+
},
234+
},
235+
{
236+
testName: "kraft broker-only node",
237+
brokerConfig: &v1beta1.BrokerConfig{Roles: []string{v1beta1.BrokerNodeProcessRole}},
238+
kRaftMode: true,
239+
storageConfig: v1beta1.StorageConfig{
240+
MountPath: "/kafka-logs-1",
241+
PvcSpec: &corev1.PersistentVolumeClaimSpec{},
242+
},
243+
expectedPersistentVolumeClaim: &corev1.PersistentVolumeClaim{
244+
ObjectMeta: metav1.ObjectMeta{
245+
Namespace: kafkaCluster.GetNamespace(),
246+
Name: "",
247+
GenerateName: fmt.Sprintf("%s-2-storage-1-", kafkaCluster.GetName()),
248+
Labels: map[string]string{
249+
v1beta1.AppLabelKey: "kafka",
250+
v1beta1.KafkaCRLabelKey: kafkaCluster.GetName(),
251+
v1beta1.BrokerIdLabelKey: "2",
252+
v1beta1.PvcRolesKey: "broker",
253+
},
254+
Annotations: map[string]string{
255+
"mountPath": "/kafka-logs-1",
256+
},
257+
},
258+
Spec: corev1.PersistentVolumeClaimSpec{},
259+
},
260+
},
261+
{
262+
testName: "kraft combined broker+controller node",
263+
brokerConfig: &v1beta1.BrokerConfig{Roles: []string{v1beta1.BrokerNodeProcessRole, v1beta1.ControllerNodeProcessRole}},
264+
kRaftMode: true,
265+
storageConfig: v1beta1.StorageConfig{
266+
MountPath: "/kafka-logs-1",
267+
PvcSpec: &corev1.PersistentVolumeClaimSpec{},
268+
},
269+
expectedPersistentVolumeClaim: &corev1.PersistentVolumeClaim{
270+
ObjectMeta: metav1.ObjectMeta{
271+
Namespace: kafkaCluster.GetNamespace(),
272+
Name: "",
273+
GenerateName: fmt.Sprintf("%s-2-storage-1-", kafkaCluster.GetName()),
274+
Labels: map[string]string{
275+
v1beta1.AppLabelKey: "kafka",
276+
v1beta1.KafkaCRLabelKey: kafkaCluster.GetName(),
277+
v1beta1.BrokerIdLabelKey: "2",
278+
v1beta1.PvcRolesKey: "broker_controller",
279+
},
280+
Annotations: map[string]string{
281+
"mountPath": "/kafka-logs-1",
282+
},
283+
},
284+
Spec: corev1.PersistentVolumeClaimSpec{},
285+
},
286+
},
193287
}
194288

195289
t.Parallel()
@@ -198,7 +292,7 @@ func TestReconciler_pvc(t *testing.T) {
198292
test := test
199293

200294
t.Run(test.testName, func(t *testing.T) {
201-
pvc, err := r.pvc(2, 1, test.storageConfig)
295+
pvc, err := r.pvc(2, 1, test.storageConfig, test.brokerConfig, test.kRaftMode)
202296

203297
assert.NilError(t, err, "PVC creation should succeed")
204298

0 commit comments

Comments
 (0)