Skip to content

Commit 56f8d4a

Browse files
omerap12omerap12
committed
refactor liklus scaler
Signed-off-by: Omer Aplatony <[email protected]>
1 parent 55bd5bf commit 56f8d4a

File tree

2 files changed

+136
-114
lines changed

2 files changed

+136
-114
lines changed

‎pkg/scalers/liiklus_scaler.go

+23-57
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,6 @@ package scalers
33
import (
44
"context"
55
"fmt"
6-
"strconv"
76
"time"
87

98
"github.com/go-logr/logr"
@@ -27,12 +26,12 @@ type liiklusScaler struct {
2726
}
2827

2928
type liiklusMetadata struct {
30-
lagThreshold int64
31-
activationLagThreshold int64
32-
address string
33-
topic string
34-
group string
35-
groupVersion uint32
29+
LagThreshold int64 `keda:"name=lagThreshold,order=triggerMetadata,default=10"`
30+
ActivationLagThreshold int64 `keda:"name=activationLagThreshold,order=triggerMetadata,default=0"`
31+
Address string `keda:"name=address,order=triggerMetadata"`
32+
Topic string `keda:"name=topic,order=triggerMetadata"`
33+
Group string `keda:"name=group,order=triggerMetadata"`
34+
GroupVersion uint32 `keda:"name=groupVersion,order=triggerMetadata,default=0"`
3635
triggerIndex int
3736
}
3837

@@ -70,7 +69,7 @@ func NewLiiklusScaler(config *scalersconfig.ScalerConfig) (Scaler, error) {
7069
return nil, err
7170
}
7271

73-
conn, err := grpc.NewClient(lm.address,
72+
conn, err := grpc.NewClient(lm.Address,
7473
grpc.WithDefaultServiceConfig(grpcConfig),
7574
grpc.WithTransportCredentials(insecure.NewCredentials()))
7675
if err != nil {
@@ -94,21 +93,21 @@ func (s *liiklusScaler) GetMetricsAndActivity(ctx context.Context, metricName st
9493
return nil, false, err
9594
}
9695

97-
if totalLag/uint64(s.metadata.lagThreshold) > uint64(len(lags)) {
98-
totalLag = uint64(s.metadata.lagThreshold) * uint64(len(lags))
96+
if totalLag/uint64(s.metadata.LagThreshold) > uint64(len(lags)) {
97+
totalLag = uint64(s.metadata.LagThreshold) * uint64(len(lags))
9998
}
10099

101100
metric := GenerateMetricInMili(metricName, float64(totalLag))
102101

103-
return []external_metrics.ExternalMetricValue{metric}, totalLag > uint64(s.metadata.activationLagThreshold), nil
102+
return []external_metrics.ExternalMetricValue{metric}, totalLag > uint64(s.metadata.ActivationLagThreshold), nil
104103
}
105104

106105
func (s *liiklusScaler) GetMetricSpecForScaling(context.Context) []v2.MetricSpec {
107106
externalMetric := &v2.ExternalMetricSource{
108107
Metric: v2.MetricIdentifier{
109-
Name: GenerateMetricNameWithIndex(s.metadata.triggerIndex, kedautil.NormalizeString(fmt.Sprintf("liiklus-%s", s.metadata.topic))),
108+
Name: GenerateMetricNameWithIndex(s.metadata.triggerIndex, kedautil.NormalizeString(fmt.Sprintf("liiklus-%s", s.metadata.Topic))),
110109
},
111-
Target: GetMetricTarget(s.metricType, s.metadata.lagThreshold),
110+
Target: GetMetricTarget(s.metricType, s.metadata.LagThreshold),
112111
}
113112
metricSpec := v2.MetricSpec{External: externalMetric, Type: liiklusMetricType}
114113
return []v2.MetricSpec{metricSpec}
@@ -131,9 +130,9 @@ func (s *liiklusScaler) getLag(ctx context.Context) (uint64, map[uint32]uint64,
131130
ctx1, cancel1 := context.WithTimeout(ctx, 10*time.Second)
132131
defer cancel1()
133132
gor, err := s.client.GetOffsets(ctx1, &liiklus_service.GetOffsetsRequest{
134-
Topic: s.metadata.topic,
135-
Group: s.metadata.group,
136-
GroupVersion: s.metadata.groupVersion,
133+
Topic: s.metadata.Topic,
134+
Group: s.metadata.Group,
135+
GroupVersion: s.metadata.GroupVersion,
137136
})
138137
if err != nil {
139138
return 0, nil, err
@@ -142,7 +141,7 @@ func (s *liiklusScaler) getLag(ctx context.Context) (uint64, map[uint32]uint64,
142141
ctx2, cancel2 := context.WithTimeout(ctx, 10*time.Second)
143142
defer cancel2()
144143
geor, err := s.client.GetEndOffsets(ctx2, &liiklus_service.GetEndOffsetsRequest{
145-
Topic: s.metadata.topic,
144+
Topic: s.metadata.Topic,
146145
})
147146
if err != nil {
148147
return 0, nil, err
@@ -159,50 +158,17 @@ func (s *liiklusScaler) getLag(ctx context.Context) (uint64, map[uint32]uint64,
159158
}
160159

161160
func parseLiiklusMetadata(config *scalersconfig.ScalerConfig) (*liiklusMetadata, error) {
162-
lagThreshold := defaultLiiklusLagThreshold
163-
activationLagThreshold := defaultLiiklusActivationLagThreshold
164-
165-
if val, ok := config.TriggerMetadata[liiklusLagThresholdMetricName]; ok {
166-
t, err := strconv.ParseInt(val, 10, 64)
167-
if err != nil {
168-
return nil, fmt.Errorf("error parsing %s: %w", liiklusLagThresholdMetricName, err)
169-
}
170-
lagThreshold = t
161+
meta := &liiklusMetadata{}
162+
if err := config.TypedConfig(meta); err != nil {
163+
return nil, fmt.Errorf("error parsing liiklus metadata: %w", err)
171164
}
172-
173-
if val, ok := config.TriggerMetadata[liiklusActivationLagThresholdMetricName]; ok {
174-
t, err := strconv.ParseInt(val, 10, 64)
175-
if err != nil {
176-
return nil, fmt.Errorf("error parsing %s: %w", liiklusActivationLagThresholdMetricName, err)
177-
}
178-
activationLagThreshold = t
179-
}
180-
181-
groupVersion := uint32(0)
182-
if val, ok := config.TriggerMetadata["groupVersion"]; ok {
183-
t, err := strconv.ParseUint(val, 10, 32)
184-
if err != nil {
185-
return nil, fmt.Errorf("error parsing groupVersion: %w", err)
186-
}
187-
groupVersion = uint32(t)
188-
}
189-
190165
switch {
191-
case config.TriggerMetadata["topic"] == "":
166+
case meta.Topic == "":
192167
return nil, ErrLiiklusNoTopic
193-
case config.TriggerMetadata["address"] == "":
168+
case meta.Address == "":
194169
return nil, ErrLiiklusNoAddress
195-
case config.TriggerMetadata["group"] == "":
170+
case meta.Group == "":
196171
return nil, ErrLiiklusNoGroup
197172
}
198-
199-
return &liiklusMetadata{
200-
topic: config.TriggerMetadata["topic"],
201-
address: config.TriggerMetadata["address"],
202-
group: config.TriggerMetadata["group"],
203-
groupVersion: groupVersion,
204-
lagThreshold: lagThreshold,
205-
activationLagThreshold: activationLagThreshold,
206-
triggerIndex: config.TriggerIndex,
207-
}, nil
173+
return meta, nil
208174
}

‎pkg/scalers/liiklus_scaler_test.go

+113-57
Original file line numberDiff line numberDiff line change
@@ -2,8 +2,7 @@ package scalers
22

33
import (
44
"context"
5-
"errors"
6-
"strconv"
5+
"fmt"
76
"testing"
87

98
"github.com/go-logr/logr"
@@ -15,12 +14,10 @@ import (
1514
)
1615

1716
type parseLiiklusMetadataTestData struct {
18-
metadata map[string]string
19-
err error
20-
liiklusAddress string
21-
group string
22-
topic string
23-
threshold int64
17+
name string
18+
metadata map[string]string
19+
ExpectedErr error
20+
ExpectedMetatada *liiklusMetadata
2421
}
2522

2623
type liiklusMetricIdentifier struct {
@@ -30,12 +27,64 @@ type liiklusMetricIdentifier struct {
3027
}
3128

3229
var parseLiiklusMetadataTestDataset = []parseLiiklusMetadataTestData{
33-
{map[string]string{}, ErrLiiklusNoTopic, "", "", "", 0},
34-
{map[string]string{"topic": "foo"}, ErrLiiklusNoAddress, "", "", "", 0},
35-
{map[string]string{"topic": "foo", "address": "bar:6565"}, ErrLiiklusNoGroup, "", "", "", 0},
36-
{map[string]string{"topic": "foo", "address": "bar:6565", "group": "mygroup"}, nil, "bar:6565", "mygroup", "foo", 10},
37-
{map[string]string{"topic": "foo", "address": "bar:6565", "group": "mygroup", "activationLagThreshold": "aa"}, strconv.ErrSyntax, "bar:6565", "mygroup", "foo", 10},
38-
{map[string]string{"topic": "foo", "address": "bar:6565", "group": "mygroup", "lagThreshold": "15"}, nil, "bar:6565", "mygroup", "foo", 15},
30+
{
31+
name: "Empty metadata",
32+
metadata: map[string]string{},
33+
ExpectedErr: fmt.Errorf("error parsing liiklus metadata: " +
34+
"missing required parameter \"address\" in [triggerMetadata]\n" +
35+
"missing required parameter \"topic\" in [triggerMetadata]\n" +
36+
"missing required parameter \"group\" in [triggerMetadata]"),
37+
ExpectedMetatada: nil,
38+
},
39+
{
40+
name: "Empty address",
41+
metadata: map[string]string{"topic": "foo"},
42+
ExpectedErr: fmt.Errorf("error parsing liiklus metadata: " +
43+
"missing required parameter \"address\" in [triggerMetadata]\n" +
44+
"missing required parameter \"group\" in [triggerMetadata]"),
45+
ExpectedMetatada: nil,
46+
},
47+
{
48+
name: "Empty group",
49+
metadata: map[string]string{"topic": "foo", "address": "using-mock"},
50+
ExpectedErr: fmt.Errorf("error parsing liiklus metadata: " +
51+
"missing required parameter \"group\" in [triggerMetadata]"),
52+
ExpectedMetatada: nil,
53+
},
54+
{
55+
name: "Valid",
56+
metadata: map[string]string{"topic": "foo", "address": "using-mock", "group": "mygroup"},
57+
ExpectedErr: nil,
58+
ExpectedMetatada: &liiklusMetadata{
59+
LagThreshold: defaultLiiklusLagThreshold,
60+
ActivationLagThreshold: defaultLiiklusActivationLagThreshold,
61+
Address: "using-mock",
62+
Topic: "foo",
63+
Group: "mygroup",
64+
GroupVersion: 0,
65+
triggerIndex: 0,
66+
},
67+
},
68+
{
69+
name: "Invalid activationLagThreshold",
70+
metadata: map[string]string{"topic": "foo", "address": "using-mock", "group": "mygroup", "activationLagThreshold": "invalid"},
71+
ExpectedErr: fmt.Errorf("error parsing liiklus metadata: unable to set param \"activationLagThreshold\" value \"invalid\": unable to unmarshal to field type int64: invalid character 'i' looking for beginning of value"),
72+
ExpectedMetatada: nil,
73+
},
74+
{
75+
name: "Custom lagThreshold",
76+
metadata: map[string]string{"topic": "foo", "address": "using-mock", "group": "mygroup", "lagThreshold": "20"},
77+
ExpectedErr: nil,
78+
ExpectedMetatada: &liiklusMetadata{
79+
LagThreshold: 20,
80+
ActivationLagThreshold: defaultLiiklusActivationLagThreshold,
81+
Address: "using-mock",
82+
Topic: "foo",
83+
Group: "mygroup",
84+
GroupVersion: 0,
85+
triggerIndex: 0,
86+
},
87+
},
3988
}
4089

4190
var liiklusMetricIdentifiers = []liiklusMetricIdentifier{
@@ -45,38 +94,44 @@ var liiklusMetricIdentifiers = []liiklusMetricIdentifier{
4594

4695
func TestLiiklusParseMetadata(t *testing.T) {
4796
for _, testData := range parseLiiklusMetadataTestDataset {
48-
meta, err := parseLiiklusMetadata(&scalersconfig.ScalerConfig{TriggerMetadata: testData.metadata})
49-
if err != nil && testData.err == nil {
50-
t.Error("Expected success but got error", err)
51-
continue
52-
}
53-
if testData.err != nil && err == nil {
54-
t.Error("Expected error but got success")
55-
continue
56-
}
57-
if testData.err != nil && err != nil && !errors.Is(err, testData.err) {
58-
t.Errorf("Expected error %v but got %v", testData.err, err)
59-
continue
60-
}
61-
if err != nil {
62-
continue
63-
}
64-
if testData.liiklusAddress != meta.address {
65-
t.Errorf("Expected address %q but got %q\n", testData.liiklusAddress, meta.address)
66-
continue
67-
}
68-
if meta.group != testData.group {
69-
t.Errorf("Expected group %q but got %q\n", testData.group, meta.group)
70-
continue
71-
}
72-
if meta.topic != testData.topic {
73-
t.Errorf("Expected topic %q but got %q\n", testData.topic, meta.topic)
74-
continue
75-
}
76-
if meta.lagThreshold != testData.threshold {
77-
t.Errorf("Expected threshold %d but got %d\n", testData.threshold, meta.lagThreshold)
78-
continue
79-
}
97+
t.Run(testData.name, func(t *testing.T) {
98+
meta, err := parseLiiklusMetadata(&scalersconfig.ScalerConfig{TriggerMetadata: testData.metadata})
99+
100+
// error cases
101+
if testData.ExpectedErr != nil {
102+
if err == nil {
103+
t.Errorf("Expected error %v but got success", testData.ExpectedErr)
104+
} else if err.Error() != testData.ExpectedErr.Error() {
105+
t.Errorf("Expected error %v but got %v", testData.ExpectedErr, err)
106+
}
107+
return // Skip the rest of the checks for error cases
108+
}
109+
110+
// success cases
111+
if err != nil {
112+
t.Errorf("Expected success but got error %v", err)
113+
}
114+
if testData.ExpectedMetatada != nil {
115+
if testData.ExpectedMetatada.Address != meta.Address {
116+
t.Errorf("Expected address %q but got %q", testData.ExpectedMetatada.Address, meta.Address)
117+
}
118+
if meta.Group != testData.ExpectedMetatada.Group {
119+
t.Errorf("Expected group %q but got %q", testData.ExpectedMetatada.Group, meta.Group)
120+
}
121+
if meta.Topic != testData.ExpectedMetatada.Topic {
122+
t.Errorf("Expected topic %q but got %q", testData.ExpectedMetatada.Topic, meta.Topic)
123+
}
124+
if meta.LagThreshold != testData.ExpectedMetatada.LagThreshold {
125+
t.Errorf("Expected threshold %d but got %d", testData.ExpectedMetatada.LagThreshold, meta.LagThreshold)
126+
}
127+
if meta.ActivationLagThreshold != testData.ExpectedMetatada.ActivationLagThreshold {
128+
t.Errorf("Expected activation threshold %d but got %d", testData.ExpectedMetatada.ActivationLagThreshold, meta.ActivationLagThreshold)
129+
}
130+
if meta.GroupVersion != testData.ExpectedMetatada.GroupVersion {
131+
t.Errorf("Expected group version %d but got %d", testData.ExpectedMetatada.GroupVersion, meta.GroupVersion)
132+
}
133+
}
134+
})
80135
}
81136
}
82137

@@ -172,16 +227,17 @@ func TestLiiklusScalerGetMetricsBehavior(t *testing.T) {
172227

173228
func TestLiiklusGetMetricSpecForScaling(t *testing.T) {
174229
for _, testData := range liiklusMetricIdentifiers {
175-
meta, err := parseLiiklusMetadata(&scalersconfig.ScalerConfig{TriggerMetadata: testData.metadataTestData.metadata, TriggerIndex: testData.triggerIndex})
176-
if err != nil {
177-
t.Fatal("Could not parse metadata:", err)
178-
}
179-
mockLiiklusScaler := liiklusScaler{"", meta, nil, nil, logr.Discard()}
180-
181-
metricSpec := mockLiiklusScaler.GetMetricSpecForScaling(context.Background())
182-
metricName := metricSpec[0].External.Metric.Name
183-
if metricName != testData.name {
184-
t.Error("Wrong External metric source name:", metricName)
185-
}
230+
t.Run(testData.name, func(t *testing.T) {
231+
meta, err := parseLiiklusMetadata(&scalersconfig.ScalerConfig{TriggerMetadata: testData.metadataTestData.metadata})
232+
if err != nil {
233+
t.Fatal("Could not parse metadata:", err)
234+
}
235+
meta.triggerIndex = testData.triggerIndex
236+
mockLiiklusScaler := liiklusScaler{"", meta, nil, nil, logr.Discard()}
237+
metricSpec := mockLiiklusScaler.GetMetricSpecForScaling(context.Background())
238+
if metricSpec[0].External.Metric.Name != testData.name {
239+
t.Errorf("Wrong External metric source name: %s", metricSpec[0].External.Metric.Name)
240+
}
241+
})
186242
}
187243
}

0 commit comments

Comments
 (0)