Skip to content

Commit 03c2c29

Browse files
twoGiantstwoGiants
committed
Fix ignored DISPATCHERS_MIN_REPLICAS env var.
Add MinReplicas to copied config in `createKafkaScheduler`. Cover bug with test and clean up the code a little. Extract duplicated code into function and replace strings with existing constants. Issue #4542. Signed-off-by: Stanislav Jakuschevskij <sjakusch@redhat.com>
1 parent 06d47e7 commit 03c2c29

3 files changed

Lines changed: 71 additions & 40 deletions

File tree

‎control-plane/pkg/reconciler/consumergroup/controller.go‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -373,6 +373,7 @@ func createKafkaScheduler(ctx context.Context, c SchedulerConfig, ssName string,
373373
StatefulSetName: ssName,
374374
RefreshPeriod: c.RefreshPeriod,
375375
Capacity: c.Capacity,
376+
MinReplicas: c.MinReplicas,
376377
},
377378
func() ([]scheduler.VPod, error) {
378379
consumerGroups, err := lister.List(labels.SelectorFromSet(getSelectorLabel(ssName)))
@@ -390,20 +391,19 @@ func createKafkaScheduler(ctx context.Context, c SchedulerConfig, ssName string,
390391
}
391392

392393
func getSelectorLabel(ssName string) map[string]string {
393-
//TODO remove hardcoded kinds
394394
var selectorLabel map[string]string
395395
switch ssName {
396396
case kafkainternals.SourceStatefulSetName:
397397
selectorLabel = map[string]string{
398-
kafkainternals.UserFacingResourceLabelSelector: "kafkasource",
398+
kafkainternals.UserFacingResourceLabelSelector: KafkaSourceScheduler,
399399
}
400400
case kafkainternals.BrokerStatefulSetName:
401401
selectorLabel = map[string]string{
402-
kafkainternals.UserFacingResourceLabelSelector: "trigger",
402+
kafkainternals.UserFacingResourceLabelSelector: KafkaTriggerScheduler,
403403
}
404404
case kafkainternals.ChannelStatefulSetName:
405405
selectorLabel = map[string]string{
406-
kafkainternals.UserFacingResourceLabelSelector: "kafkachannel",
406+
kafkainternals.UserFacingResourceLabelSelector: KafkaChannelScheduler,
407407
}
408408
}
409409
return selectorLabel

‎control-plane/pkg/reconciler/consumergroup/controller_test.go‎

Lines changed: 59 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,9 @@ package consumergroup
1818

1919
import (
2020
"context"
21+
"strconv"
2122
"testing"
23+
"time"
2224

2325
"github.com/stretchr/testify/require"
2426
corev1 "k8s.io/api/core/v1"
@@ -61,15 +63,20 @@ const (
6163
ConfigKafkaAutoscalerName = "config-kafka-autoscaler"
6264
)
6365

64-
func TestNewController(t *testing.T) {
66+
func fakeCtxWithDispatcher(t *testing.T) context.Context {
6567
ctx, _ := reconcilertesting.SetupFakeContext(t, func(ctx context.Context) context.Context {
6668
return filteredFactory.WithSelectors(ctx,
6769
internalsapi.DispatcherLabelSelectorStr,
6870
)
6971
})
70-
ctx, _ = kedaclient.With(ctx)
7172

72-
t.Setenv("SYSTEM_NAMESPACE", systemNamespace)
73+
return ctx
74+
}
75+
76+
func TestNewController(t *testing.T) {
77+
ctx := fakeCtxWithDispatcher(t)
78+
79+
ctx, _ = kedaclient.With(ctx)
7380

7481
ctx, _ = kubeclient.With(ctx,
7582
&corev1.ConfigMap{
@@ -79,16 +86,16 @@ func TestNewController(t *testing.T) {
7986
},
8087
Data: map[string]string{
8188
"predicates": `
82-
[
83-
{"Name": "PodFitsResources"},
84-
{"Name": "NoMaxResourceCount", "Args": "{\"NumPartitions\": 100}"},
85-
{"Name": "EvenPodSpread", "Args": "{\"MaxSkew\": 2}"}
86-
]`,
89+
[
90+
{"Name": "PodFitsResources"},
91+
{"Name": "NoMaxResourceCount", "Args": "{\"NumPartitions\": 100}"},
92+
{"Name": "EvenPodSpread", "Args": "{\"MaxSkew\": 2}"}
93+
]`,
8794
"priorities": `
88-
[
89-
{"Name": "AvailabilityZonePriority", "Weight": 10, "Args": "{\"MaxSkew\": 2}"},
90-
{"Name": "LowestOrdinalPriority", "Weight": 2}
91-
]`,
95+
[
96+
{"Name": "AvailabilityZonePriority", "Weight": 10, "Args": "{\"MaxSkew\": 2}"},
97+
{"Name": "LowestOrdinalPriority", "Weight": 2}
98+
]`,
9299
},
93100
},
94101
&corev1.ConfigMap{
@@ -99,23 +106,25 @@ func TestNewController(t *testing.T) {
99106
Data: map[string]string{
100107
"predicates": `[]`,
101108
"priorities": `
102-
[
103-
{"Name": "RemoveWithEvenPodSpreadPriority", "Weight": 10, "Args": "{\"MaxSkew\": 2}"},
104-
{"Name": "RemoveWithAvailabilityZonePriority", "Weight": 10, "Args": "{\"MaxSkew\": 2}"},
105-
{"Name": "RemoveWithHighestOrdinalPriority", "Weight": 2}
106-
]`,
109+
[
110+
{"Name": "RemoveWithEvenPodSpreadPriority", "Weight": 10, "Args": "{\"MaxSkew\": 2}"},
111+
{"Name": "RemoveWithAvailabilityZonePriority", "Weight": 10, "Args": "{\"MaxSkew\": 2}"},
112+
{"Name": "RemoveWithHighestOrdinalPriority", "Weight": 2}
113+
]`,
107114
},
108115
},
109116
)
110117

111118
ctx = clientpool.WithKafkaClientPool(ctx)
112119

120+
t.Setenv("SYSTEM_NAMESPACE", systemNamespace)
113121
t.Setenv("AUTOSCALER_REFRESH_PERIOD", RefreshPeriod)
114122
t.Setenv("POD_CAPACITY", PodCapacity)
115123
t.Setenv("DISPATCHERS_MIN_REPLICAS", MinReplicas)
116124
t.Setenv("SCHEDULER_CONFIG", ConfigKafkaSchedulerName)
117125
t.Setenv("DESCHEDULER_CONFIG", ConfigKafkaDeSchedulerName)
118126
t.Setenv("AUTOSCALER_CONFIG", ConfigKafkaAutoscalerName)
127+
119128
controller := NewController(ctx, configmap.NewStaticWatcher(&corev1.ConfigMap{
120129
ObjectMeta: metav1.ObjectMeta{
121130
Name: "config-kafka-features",
@@ -127,7 +136,6 @@ func TestNewController(t *testing.T) {
127136
}
128137

129138
func TestEnqueueConsumerFromConsumerGroup(t *testing.T) {
130-
131139
capture := types.NamespacedName{}
132140

133141
f := enqueueConsumerGroupFromConsumer(func(name types.NamespacedName) {
@@ -155,3 +163,36 @@ func TestEnqueueConsumerFromConsumerGroup(t *testing.T) {
155163
require.Equal(t, or.Name, capture.Name)
156164
require.Equal(t, ns, capture.Namespace)
157165
}
166+
167+
// Test covering the bug:
168+
// - https://github.com/knative-extensions/eventing-kafka-broker/issues/4542
169+
//
170+
// The config value "MinReplicas" was ignored in createKafkaScheduler when the config was copied.
171+
// To make sure this does not happen for the other values too, they were added to the test.
172+
func TestCreateKafkaSchedulerSetsCorrectConfig(t *testing.T) {
173+
ctx := fakeCtxWithDispatcher(t)
174+
ctx, cancel := context.WithCancel(ctx)
175+
t.Cleanup(cancel)
176+
177+
rp, _ := time.ParseDuration(RefreshPeriod)
178+
c, _ := strconv.ParseInt(PodCapacity, 10, 32)
179+
mr, _ := strconv.ParseInt(MinReplicas, 10, 32)
180+
expectedConfig := SchedulerConfig{
181+
RefreshPeriod: rp,
182+
Capacity: int32(c),
183+
MinReplicas: int32(mr),
184+
}
185+
informer := filteredFactory.Get(ctx, internalsapi.DispatcherLabelSelectorStr).Core().V1().Pods()
186+
187+
scheduler := createKafkaScheduler(
188+
ctx,
189+
expectedConfig,
190+
kafkainternals.SourceStatefulSetName,
191+
informer,
192+
)
193+
194+
require.Equal(t, kafkainternals.SourceStatefulSetName, scheduler.StatefulSetName)
195+
require.Equal(t, expectedConfig.RefreshPeriod, scheduler.RefreshPeriod)
196+
require.Equal(t, expectedConfig.Capacity, scheduler.Capacity)
197+
require.Equal(t, expectedConfig.MinReplicas, scheduler.MinReplicas)
198+
}

‎control-plane/pkg/reconciler/consumergroup/evictor_test.go‎

Lines changed: 8 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,6 @@
1717
package consumergroup
1818

1919
import (
20-
"context"
2120
"testing"
2221

2322
"github.com/stretchr/testify/require"
@@ -26,24 +25,21 @@ import (
2625
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
2726
eventingduckv1alpha1 "knative.dev/eventing/pkg/apis/duck/v1alpha1"
2827
kubeclient "knative.dev/pkg/client/injection/kube/client/fake"
29-
filteredFactory "knative.dev/pkg/client/injection/kube/informers/factory/filtered"
3028
"knative.dev/pkg/ptr"
31-
reconcilertesting "knative.dev/pkg/reconciler/testing"
3229

33-
internalsapi "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internalskafkaeventing"
3430
kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internalskafkaeventing/v1alpha1"
3531
_ "knative.dev/eventing-kafka-broker/control-plane/pkg/client/injection/client/fake"
3632
kafkainternalsclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/injection/client/fake"
3733
)
3834

3935
func TestNewEvictor(t *testing.T) {
40-
ctx, _ := reconcilertesting.SetupFakeContext(t, withFilteredSelectors)
36+
ctx := fakeCtxWithDispatcher(t)
4137

4238
require.NotPanics(t, func() { newEvictor(ctx, zap.String("k", "n")) })
4339
}
4440

4541
func TestEvictorNilPodNoPanic(t *testing.T) {
46-
ctx, _ := reconcilertesting.SetupFakeContext(t, withFilteredSelectors)
42+
ctx := fakeCtxWithDispatcher(t)
4743

4844
var pod *corev1.Pod
4945

@@ -77,7 +73,7 @@ func TestEvictorNilPodNoPanic(t *testing.T) {
7773
}
7874

7975
func TestEvictorEvictSuccess(t *testing.T) {
80-
ctx, _ := reconcilertesting.SetupFakeContext(t, withFilteredSelectors)
76+
ctx := fakeCtxWithDispatcher(t)
8177

8278
pod := &corev1.Pod{
8379
ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "name"},
@@ -114,7 +110,7 @@ func TestEvictorEvictSuccess(t *testing.T) {
114110
}
115111

116112
func TestEvictorNoEvictionEmptyPlacement(t *testing.T) {
117-
ctx, _ := reconcilertesting.SetupFakeContext(t, withFilteredSelectors)
113+
ctx := fakeCtxWithDispatcher(t)
118114

119115
pod := &corev1.Pod{
120116
ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "name"},
@@ -148,7 +144,7 @@ func TestEvictorNoEvictionEmptyPlacement(t *testing.T) {
148144
}
149145

150146
func TestEvictorNoEviction(t *testing.T) {
151-
ctx, _ := reconcilertesting.SetupFakeContext(t, withFilteredSelectors)
147+
ctx := fakeCtxWithDispatcher(t)
152148

153149
pod := &corev1.Pod{
154150
ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "name"},
@@ -184,7 +180,7 @@ func TestEvictorNoEviction(t *testing.T) {
184180
}
185181

186182
func TestEvictorEvictSuccessConsumerGroupSchedulingInProgress(t *testing.T) {
187-
ctx, _ := reconcilertesting.SetupFakeContext(t, withFilteredSelectors)
183+
ctx := fakeCtxWithDispatcher(t)
188184

189185
pod := &corev1.Pod{
190186
ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "name"},
@@ -215,7 +211,7 @@ func TestEvictorEvictSuccessConsumerGroupSchedulingInProgress(t *testing.T) {
215211
}
216212

217213
func TestEvictorEvictPodNotFound(t *testing.T) {
218-
ctx, _ := reconcilertesting.SetupFakeContext(t, withFilteredSelectors)
214+
ctx := fakeCtxWithDispatcher(t)
219215

220216
pod := &corev1.Pod{
221217
ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "name"},
@@ -241,7 +237,7 @@ func TestEvictorEvictPodNotFound(t *testing.T) {
241237
require.Nil(t, err)
242238
}
243239
func TestEvictorEvictConsumerGroupNotFound(t *testing.T) {
244-
ctx, _ := reconcilertesting.SetupFakeContext(t, withFilteredSelectors)
240+
ctx := fakeCtxWithDispatcher(t)
245241

246242
pod := &corev1.Pod{
247243
ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "name"},
@@ -266,9 +262,3 @@ func TestEvictorEvictConsumerGroupNotFound(t *testing.T) {
266262

267263
require.Nil(t, err)
268264
}
269-
270-
func withFilteredSelectors(ctx context.Context) context.Context {
271-
return filteredFactory.WithSelectors(ctx,
272-
internalsapi.DispatcherLabelSelectorStr,
273-
)
274-
}

0 commit comments

Comments
 (0)