Skip to content
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

fix selector path for cg scale subresource #3666

Merged
merged 4 commits into from
Feb 9, 2024
Merged

fix selector path for cg scale subresource #3666

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
aa8b739
fix selector path for cg scale subresource
Cali0707 Feb 2, 2024
f14b958
set .status.selector properly
Cali0707 Feb 2, 2024
5978864
fix: unit tests expect correct cg status
Cali0707 Feb 2, 2024
7420dd9
fix: selector is omitted from struct when empty
Cali0707 Feb 6, 2024
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
4 changes: 4 additions & 0 deletions control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -141,6 +141,10 @@ type ConsumerGroupStatus struct {
// same Template, but individual replicas also have a consistent identity.
// +optional
Replicas *int32 `json:"replicas,omitempty"`

// Selector is the string serialized label selector needed for the scale subresource.
// Defaults to ""
Selector string `json:"selector"`
Cali0707 marked this conversation as resolved.
Show resolved Hide resolved
}

// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
Expand Down
6 changes: 6 additions & 0 deletions control-plane/pkg/reconciler/consumergroup/consumergroup.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -178,6 +178,8 @@ type Reconciler struct {
func (r *Reconciler) ReconcileKind(ctx context.Context, cg *kafkainternals.ConsumerGroup) reconciler.Event {
recordExpectedReplicasMetric(ctx, cg)

r.reconcileStatusSelector(cg)

if err := r.reconcileInitialOffset(ctx, cg); err != nil {
return cg.MarkInitializeOffsetFailed("InitializeOffset", err)
}
Expand Down Expand Up @@ -259,6 +261,10 @@ func (r *Reconciler) FinalizeKind(ctx context.Context, cg *kafkainternals.Consum
return nil
}

func (r *Reconciler) reconcileStatusSelector(cg *kafkainternals.ConsumerGroup) {
cg.Status.Selector = labels.SelectorFromValidatedSet(cg.Spec.Selector).String()
}

func (r *Reconciler) deleteConsumerGroupMetadata(ctx context.Context, cg *kafkainternals.ConsumerGroup) error {
saramaSecurityOption, err := r.newAuthConfigOption(ctx, cg)
if err != nil {
Expand Down
15 changes: 15 additions & 0 deletions control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -148,6 +148,7 @@ func TestReconcileKind(t *testing.T) {
ConsumerGroupReplicas(2),
ConsumerGroupStatusReplicas(0),
ConsumerForTrigger(),
ConsumerGroupStatusSelector(ConsumerLabels),
)
cg.Status.Placements = []eventingduckv1alpha1.Placement{
{PodName: "p1", VReplicas: 1},
Expand Down Expand Up @@ -234,6 +235,7 @@ func TestReconcileKind(t *testing.T) {
),
)),
ConsumerGroupReplicas(2),
ConsumerGroupStatusSelector(ConsumerLabels),
ConsumerGroupStatusReplicas(0),
ConsumerForTrigger(),
)
Expand Down Expand Up @@ -340,6 +342,7 @@ func TestReconcileKind(t *testing.T) {
}),
)),
ConsumerGroupReplicas(2),
ConsumerGroupStatusSelector(ConsumerLabels),
ConsumerForTrigger(),
)
cg.InitializeConditions()
Expand Down Expand Up @@ -448,6 +451,7 @@ func TestReconcileKind(t *testing.T) {
}),
)),
ConsumerGroupReplicas(2),
ConsumerGroupStatusSelector(ConsumerLabels),
ConsumerGroupStatusReplicas(0),
ConsumerForTrigger(),
)
Expand Down Expand Up @@ -572,6 +576,7 @@ func TestReconcileKind(t *testing.T) {
}),
)),
ConsumerGroupReplicas(2),
ConsumerGroupStatusSelector(ConsumerLabels),
ConsumerGroupStatusReplicas(1),
ConsumerForTrigger(),
)
Expand Down Expand Up @@ -758,6 +763,7 @@ func TestReconcileKind(t *testing.T) {
}),
)),
ConsumerGroupReplicas(2),
ConsumerGroupStatusSelector(ConsumerLabels),
ConsumerForTrigger(),
)
cg.InitializeConditions()
Expand Down Expand Up @@ -974,6 +980,7 @@ func TestReconcileKind(t *testing.T) {
)),
ConsumerGroupReplicas(2),
ConsumerGroupStatusReplicas(1),
ConsumerGroupStatusSelector(ConsumerLabels),
ConsumerForTrigger(),
)
cg.Status.Placements = []eventingduckv1alpha1.Placement{
Expand Down Expand Up @@ -1063,6 +1070,7 @@ func TestReconcileKind(t *testing.T) {
ConsumerGroupReplicas(1),
ConsumerGroupStatusReplicas(0),
ConsumerForTrigger(),
ConsumerGroupStatusSelector(ConsumerLabels),
)
cg.Status.Placements = []eventingduckv1alpha1.Placement{
{PodName: "p1", VReplicas: 1},
Expand Down Expand Up @@ -1144,6 +1152,7 @@ func TestReconcileKind(t *testing.T) {
ConsumerGroupIdConfig("my.group.id"),
),
)),
ConsumerGroupStatusSelector(ConsumerLabels),
ConsumerGroupReplicas(2),
ConsumerGroupStatusReplicas(0),
ConsumerForTrigger(),
Expand Down Expand Up @@ -1230,6 +1239,7 @@ func TestReconcileKind(t *testing.T) {
ConsumerGroupIdConfig("my.group.id"),
),
)),
ConsumerGroupStatusSelector(ConsumerLabels),
ConsumerGroupReplicas(2),
ConsumerGroupStatusReplicas(1),
ConsumerForTrigger(),
Expand Down Expand Up @@ -1358,6 +1368,7 @@ func TestReconcileKind(t *testing.T) {
ConsumerInitialOffset(sources.OffsetLatest),
)),
)),
ConsumerGroupStatusSelector(ConsumerLabels),
ConsumerGroupReplicas(3),
ConsumerForTrigger(),
)
Expand Down Expand Up @@ -1468,6 +1479,7 @@ func TestReconcileKind(t *testing.T) {
ConsumerGroupIdConfig("my.group.id"),
),
)),
ConsumerGroupStatusSelector(ConsumerLabels),
ConsumerGroupReplicas(3),
ConsumerForTrigger(),
)
Expand Down Expand Up @@ -1577,6 +1589,7 @@ func TestReconcileKind(t *testing.T) {
ConsumerGroupIdConfig("my.group.id"),
),
)),
ConsumerGroupStatusSelector(ConsumerLabels),
ConsumerGroupReplicas(2),
ConsumerForTrigger(),
)
Expand Down Expand Up @@ -1640,6 +1653,7 @@ func TestReconcileKind(t *testing.T) {
),
)),
ConsumerGroupReplicas(2),
ConsumerGroupStatusSelector(ConsumerLabels),
ConsumerForTrigger(),
)
cg.GetConditionSet().Manage(cg.GetStatus()).InitializeConditions()
Expand Down Expand Up @@ -1776,6 +1790,7 @@ func TestReconcileKindNoAutoscaler(t *testing.T) {
),
)),
ConsumerGroupReplicas(2),
ConsumerGroupStatusSelector(ConsumerLabels),
ConsumerGroupStatusReplicas(1),
ConsumerForTrigger(),
)
Expand Down
7 changes: 7 additions & 0 deletions control-plane/pkg/reconciler/testing/objects_consumergroup.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ import (

corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/labels"
"k8s.io/utils/pointer"
"knative.dev/pkg/apis"
duckv1 "knative.dev/pkg/apis/duck/v1"
Expand Down Expand Up @@ -189,6 +190,12 @@ func ConsumerGroupStatusReplicas(replicas int32) ConsumerGroupOption {
}
}

func ConsumerGroupStatusSelector(label map[string]string) ConsumerGroupOption {
return func(cg *kafkainternals.ConsumerGroup) {
cg.Status.Selector = labels.SelectorFromSet(label).String()
}
}

func ConsumerGroupReplicasStatus(replicas int32) ConsumerGroupOption {
return func(cg *kafkainternals.ConsumerGroup) {
cg.Status.Replicas = pointer.Int32(replicas)
Expand Down
Loading