Add Apache Pulsar Scaler #3021

liangyuanpeng · 2022-05-08T04:30:39Z

Provide a description of what has been changed

Checklist

Commits are signed with Developer Certificate of Origin (DCO - learn more)
Tests have been added
A PR is opened to update our Helm chart (repo) (if applicable, ie. when deployment manifests are modified)
A PR is opened to update the documentation on (repo) (if applicable)
Changelog has been updated and is aligned with our changelog requirements

Fixes #853

This PR is base on #1666

doc PR --> kedacore/keda-docs#761

Signed-off-by: Mark Mussett <mmussett@tibco.com>

Signed-off-by: Lan Liang <gcslyp@gmail.com>

…gger metadata Signed-off-by: Lan Liang <gcslyp@gmail.com>

liangyuanpeng · 2022-05-15T07:18:15Z

e2e CI failed and this is error:

Error: Label check failed: required any of ok-to-merge, but found 0.

Whant can i do for it and how can i check it myself, thanks!

JorTurFer · 2022-05-15T10:47:49Z

Whant can i do for it and how can i check it myself, thanks!

hey @liangyuanpeng ,
Thanks a ton for this new scaler! ❤️ ❤️
That specific CI only passes if e2e tests have passed. (it's a new mechanism that we have introduced to block PRs without e2e test). It's not mandatory because we can add the label manually if the PR doesn't need them.
In this case, talking about new scaler, it'd be nice if you can add some e2e tests.
BTW, there is other CI failing that requires changes in the code. Basically you should sort alphabetically the scalers in scale_handler.go and some code fixes

pkg/scaling/scale_handler.go

JorTurFer

Thanks a lot for this new scaler ❤️ ❤️ ❤️
I have left some comments inline

JorTurFer · 2022-05-15T11:00:54Z

pkg/scalers/pulsar_scaler.go

+func (s *pulsarScaler) GetStats() (*pulsarStats, error) {
+	stats := new(pulsarStats)
+
+	topic := strings.ReplaceAll(s.metadata.topic, "persistent://", "")


Do we need the full name topic in any case? I mean, inside the scaler the only usage I have seen is this and we replace the given value every time. Maybe if the scaler requests the topic with the prefix "persistent://" (because it's the common language in pulsar for instance) , maybe we can parse in during parsePulsarMetadata to do it only once.
WDYT?

Yes, you absolutely right, we just need to parse it once.

JorTurFer · 2022-05-15T11:05:37Z

pkg/scalers/pulsar_scaler.go

+		return nil, fmt.Errorf("error parsing pulsar metadata: %s", err)
+	}
+
+	client := &http.Client{}


Please, use the internal function to generate the httpClient, we use it to unify all the configurations. You can find an example here

JorTurFer · 2022-05-15T11:09:03Z

pkg/scalers/pulsar_scaler.go

+	topic := strings.ReplaceAll(s.metadata.topic, "persistent://", "")
+	statsURL := s.metadata.adminURL + "/admin/v2/persistent/" + topic + "/stats"
+
+	req, err := http.NewRequest("GET", statsURL, nil)


Please, propagate the context and use it here

JorTurFer · 2022-05-15T11:10:37Z

pkg/scalers/pulsar_scaler.go

+}
+
+// IsActive determines if we need to scale from zero
+func (s *pulsarScaler) IsActive(_ context.Context) (bool, error) {


Please, use this context and propagate it to http request using http.NewRequestWithContext

JorTurFer · 2022-05-15T11:10:44Z

pkg/scalers/pulsar_scaler.go

+}
+
+// GetMetrics returns value for a supported metric and an error if there is a problem getting the metric
+func (s *pulsarScaler) GetMetrics(_ context.Context, metricName string, _ labels.Selector) ([]external_metrics.ExternalMetricValue, error) {


Please, use this context and propagate it to http request using http.NewRequestWithContext

JorTurFer · 2022-05-15T11:15:01Z

pkg/scalers/pulsar_scaler.go

+	targetMetricValue := resource.NewQuantity(s.metadata.msgBacklogThreshold, resource.DecimalSI)
+	externalMetric := &v2beta2.ExternalMetricSource{
+		Metric: v2beta2.MetricIdentifier{
+			Name: kedautil.NormalizeString(fmt.Sprintf("%s-%s-%s", "pulsar", s.metadata.topic, s.metadata.subscription)),


For names, we use the internal function that ensures the metric name is unique inside the scaler, you can find an example here. The whole process is defined in the docs about GetMetricsSpecForScaling

Signed-off-by: Lan Liang <gcslyp@gmail.com>

…lete unuse log Signed-off-by: Lan Liang <gcslyp@gmail.com>

liangyuanpeng · 2022-05-18T15:21:02Z

it'd be nice if you can add some e2e tests

I have not experience for ts, but i'm really working for pulsar scaler e2e test. Will push it on this PR or next PR.

JorTurFer · 2022-05-21T17:57:37Z

it'd be nice if you can add some e2e tests

I have not experience for ts, but i'm really working for pulsar scaler e2e test. Will push it on this PR or next PR.

We prefer to merge together new scalers with the e2e tests but no worries, no rush with the PR, next release will be in 3 months 😄
If you need help with the e2e tests, you can write us directly in kubernetes slack and we'll be happy helping you.

JorTurFer · 2022-05-21T18:04:32Z

The changes look good, only some nits (apart from e2e tests):

Add a test to check that metricSpec is well generated (like this)
Add the changes to the Changelog.md
Solve Static Checks CI issues

Thanks a lot!

zroubalik · 2022-05-31T08:49:05Z

it'd be nice if you can add some e2e tests

I have not experience for ts, but i'm really working for pulsar scaler e2e test. Will push it on this PR or next PR.

@liangyuanpeng we are migrating e2e tests to Go, you can see this inital PR: #3079 Once that PR is merged you can start writing e2e tests based on that and not deal with TS.

liangyuanpeng · 2022-05-31T16:16:35Z

it'd be nice if you can add some e2e tests

I have not experience for ts, but i'm really working for pulsar scaler e2e test. Will push it on this PR or next PR.

@liangyuanpeng we are migrating e2e tests to Go, you can see this inital PR: #3079 Once that PR is merged you can start writing e2e tests based on that and not deal with TS.

@zroubalik Thanks, Actually i'm finished the e2e test, but i need more time to ready for this PR. I'm happy of i can write e2e test with go , so i will working when the #3079 is merged.

zroubalik · 2022-06-13T07:14:13Z

@liangyuanpeng the initial PR with Golang based e2e tests has been merged, feel free to use it as a base for Pulsar e2e tests.

liangyuanpeng · 2022-06-20T13:41:28Z

I will work for this on this week.

tomkerkhove · 2022-07-11T05:20:52Z

@liangyuanpeng Any thought if we can close this PR soon? We are planning to release in 2 weeks.

liangyuanpeng · 2022-07-12T05:53:44Z

@tomkerkhove Sorry for my late, I am busy with my job bu i still have interested int finish this PR and I will finish the e2e test asap. :)

Signed-off-by: Lan Liang <gcslyp@gmail.com>

zroubalik · 2022-07-13T15:52:31Z

@liangyuanpeng could you please add activation property as part of this PR, we have started to adding this existing scalers as you can see in these PRs:
#3358
#3360

liangyuanpeng · 2022-07-18T05:55:10Z

add activation property as part of this PR

Absolutely, push it today.

Signed-off-by: Lan Liang <gcslyp@gmail.com>

zroubalik · 2022-07-18T18:39:57Z

@liangyuanpeng could you please fixt he problems found in the Static Checks? https://github.com/kedacore/keda/runs/7393791519?check_suite_focus=true

Signed-off-by: Lan Liang <gcslyp@gmail.com>

zroubalik

Could you please rebase your branch and fix the outstanding issues?

tests/scalers_go/pulsar/pulsar_test.go:222:17: undeclared name: `WaitForDeploymentReplicaCount` (typecheck)
	assert.True(t, WaitForDeploymentReplicaCount(t, kc, consumerDeploymentName, testNamespace, 0, 60, 1),
^
tests/scalers_go/pulsar/pulsar_test.go:249:17: undeclared name: `WaitForDeploymentReplicaCount` (typecheck)
	assert.True(t, WaitForDeploymentReplicaCount(t, kc, consumerDeploymentName, testNamespace, 5, 300, 1),
^
tests/scalers_go/pulsar/pulsar_test.go:256:17: undeclared name: `WaitForDeploymentReplicaCount` (typecheck)
	assert.True(t, WaitForDeploymentReplicaCount(t, kc, consumerDeploymentName, testNamespace, 0, 300, 1),

-> replace WaitForDeploymentReplicaCount() with WaitForDeploymentReplicaReadyCount().

pkg/scalers/kafka_scaler.go:132:13: string `enable` has 3 occurrences, make it a constant (goconst)
		if val == "enable" {

-> make it constant

zroubalik · 2022-07-20T09:16:20Z

pkg/scalers/pulsar_scaler.go

+	if val, ok := config.TriggerMetadata["tls"]; ok {
+		val = strings.TrimSpace(val)
+
+		if val == "enable" {


make this constant pls, so it doesn't fail the static check

make this constant pls, so it doesn't fail the static check

Seems like the CI message is talking about kafka_scaler and this PR is working for pulsar_scaler, should i change it?

It's just a confusing warning message, it is caused by the "enabled" added in this new code in pulsar scaler.

…const Signed-off-by: Lan Liang <gcslyp@gmail.com>

JorTurFer

Hey,
Thanks for this awesome improvement! ❤️
I have left some comments inline and I'd like to request also to update the changelog adding an entry for this scaler. You can see the conventions for the changelog here

pkg/scalers/pulsar_scaler.go

JorTurFer · 2022-07-21T09:12:06Z

pkg/scalers/pulsar_scaler.go

+	ca        string
+
+	statsURL                   string
+	activationTargetQueryValue int64


I guess that this variable is for the activation feature, please use the same name as the target/threashold value with the prefix activation, I think that in this case it would be activationMsgBacklogThreshold

JorTurFer · 2022-07-21T09:15:05Z

pkg/scalers/pulsar_scaler.go

+	}
+
+	meta.activationTargetQueryValue = 0
+	if val, ok := config.TriggerMetadata["activationTargetQueryValue"]; ok {


if the target value is msgBacklog, based on the convention defined, the key should be activationMsgBacklog

JorTurFer · 2022-07-21T09:17:20Z

pkg/scalers/pulsar_scaler.go

+		}
+		return stats, nil
+	case 404:
+		return nil, nil


shouldn't we return error here?

JorTurFer · 2022-07-21T09:18:19Z

pkg/scalers/pulsar_scaler.go

+	}
+
+	if !found {
+		return nil, nil


we have to return an error here, otherwise we will have a panic in the code which calls here

JorTurFer · 2022-07-21T09:21:51Z

pkg/scalers/pulsar_scaler.go

+	metric := external_metrics.ExternalMetricValue{
+		MetricName: metricName,
+		Value:      *resource.NewQuantity(msgBacklog, resource.DecimalSI),
+		Timestamp:  metav1.Now(),
+	}


we have a helper to generate this, it's the func GenerateMetricMili, you can see a sample here

JorTurFer · 2022-07-21T09:22:10Z

pkg/scalers/pulsar_scaler.go

+func (s *pulsarScaler) GetMetricSpecForScaling(context.Context) []v2beta2.MetricSpec {
+	targetMetricValue := resource.NewQuantity(s.metadata.msgBacklogThreshold, resource.DecimalSI)
+
+	metricName := fmt.Sprintf("%s-%s-%s", "pulsar", s.metadata.topic, s.metadata.subscription)
+
+	externalMetric := &v2beta2.ExternalMetricSource{
+		Metric: v2beta2.MetricIdentifier{
+			Name: GenerateMetricNameWithIndex(s.metadata.scalerIndex, kedautil.NormalizeString(metricName)),
+		},
+		Target: v2beta2.MetricTarget{
+			Type:         v2beta2.AverageValueMetricType,
+			AverageValue: targetMetricValue,
+		},
+	}
+	metricSpec := v2beta2.MetricSpec{External: externalMetric, Type: pulsarMetricType}
+	return []v2beta2.MetricSpec{metricSpec}
+}


Some considerations here:

Use the function GetMetricTarget to generate the target section (a sample here)

I'd move the metricName generation to the parsing section to avoid parsing it every time

JorTurFer · 2022-07-21T09:27:15Z

tests/scalers_go/pulsar/pulsar_test.go

+		"replica count should be 0 after a minute")
+
+	testActivation(t, kc, data)
+	KubectlDeleteWithTemplate(t, data, "publishJobTemplate", publishJobTemplate)


I'd move this inside testActivation func because it's related with cleaning up the activation test

Simplify the logic. Co-authored-by: Jorge Turrado <jorge_turrado@hotmail.es> Co-authored-by: Jorge Turrado Ferrero <Jorge_turrado@hotmail.es>

…d add some error to return Signed-off-by: Lan Liang <gcslyp@gmail.com>

Signed-off-by: Lan Liang <gcslyp@gmail.com>

zroubalik · 2022-07-25T09:46:28Z

/run-e2e pulsar*
Update: You can check the progress here

JorTurFer

Looking good!
Only one more thing, could you add an entry in the changelog?

tests/scalers_go/pulsar/pulsar_test.go

zroubalik · 2022-08-04T13:55:28Z

@liangyuanpeng any update on this please? We would like to complete this by the end of the week. So we can ship a new release the next one.

zroubalik · 2022-08-08T07:09:39Z

@liangyuanpeng any updates please? we would need to merge it today in order to get it into the next release, if not that we will target 2.9.

tests/scalers_go/pulsar/pulsar_test.go

zroubalik · 2022-08-08T13:45:36Z

/run-e2e pulsar*
Update: You can check the progress here

mmussett and others added 4 commits March 12, 2021 10:29

Pulsar scaler initial commit

6944035

Signed-off-by: Mark Mussett <mmussett@tibco.com>

Updated changelog for Apache Pulsar scaler

9eeda25

Signed-off-by: Mark Mussett <mmussett@tibco.com>

Lint fixes applied

a9fc474

Signed-off-by: Mark Mussett <mmussett@tibco.com>

Merge main and change statsURL to adminURL

756b2de

Signed-off-by: Lan Liang <gcslyp@gmail.com>

liangyuanpeng requested a review from a team as a code owner May 8, 2022 04:30

Delete param of tenant and namespace, Change enable tls config to tri…

de04781

…gger metadata Signed-off-by: Lan Liang <gcslyp@gmail.com>

liangyuanpeng changed the title ~~[WIP] Support Scaler pulsar~~ Support Scaler pulsar May 15, 2022

JorTurFer reviewed May 15, 2022

View reviewed changes

pkg/scaling/scale_handler.go Show resolved Hide resolved

JorTurFer reviewed May 15, 2022

View reviewed changes

liangyuanpeng added 2 commits May 17, 2022 22:22

Add pulsar metadata param of statsURL and use httpclient of kedautil

c2bc85f

Signed-off-by: Lan Liang <gcslyp@gmail.com>

Use GenerateMetricNameWithIndex for pulsar scaler metrics name and de…

8e678d4

…lete unuse log Signed-off-by: Lan Liang <gcslyp@gmail.com>

JorTurFer requested a review from zroubalik May 21, 2022 18:02

zroubalik changed the title ~~Support Scaler pulsar~~ Add Apache Pulsar Scaler Jun 13, 2022

Merge branch 'main' of github.com:kedacore/keda into scaler_pulsar

1a42218

Merge branch 'main' of github.com:kedacore/keda into scaler_pulsar

dc36444

liangyuanpeng added 2 commits July 13, 2022 23:06

Add e2e test withh no tls for pulsar scaler

1c15cfa

Signed-off-by: Lan Liang <gcslyp@gmail.com>

Add messageCount for config

28ff48a

Signed-off-by: Lan Liang <gcslyp@gmail.com>

Pulsar scaler support activationTargetQueryValue and add test for it

9e16bcc

Signed-off-by: Lan Liang <gcslyp@gmail.com>

Fix code style

8c3bb3d

Signed-off-by: Lan Liang <gcslyp@gmail.com>

zroubalik reviewed Jul 20, 2022

View reviewed changes

liangyuanpeng added 2 commits July 21, 2022 13:30

Merge remote-tracking branch 'origin/main' into scaler_pulsar

83a8bec

Use WaitForDeploymentReplicaReadyCount for test and Change enable to …

eee95f8

…const Signed-off-by: Lan Liang <gcslyp@gmail.com>

JorTurFer reviewed Jul 21, 2022

View reviewed changes

liangyuanpeng and others added 4 commits July 22, 2022 13:48

Update pkg/scalers/pulsar_scaler.go

88143a5

Simplify the logic. Co-authored-by: Jorge Turrado <jorge_turrado@hotmail.es> Co-authored-by: Jorge Turrado Ferrero <Jorge_turrado@hotmail.es>

Change activationTargetQueryValue to activationMsgBacklogThreshold an…

0a158f5

…d add some error to return Signed-off-by: Lan Liang <gcslyp@gmail.com>

Fix conflict

0251a9b

Signed-off-by: Lan Liang <gcslyp@gmail.com>

Update pulsar scaler e2e test use activationMsgBacklogThreshold

7adfbc2

Signed-off-by: Lan Liang <gcslyp@gmail.com>

JorTurFer reviewed Jul 27, 2022

View reviewed changes

tests/scalers_go/pulsar/pulsar_test.go Outdated Show resolved Hide resolved

zroubalik reviewed Aug 2, 2022

View reviewed changes

tests/scalers_go/pulsar/pulsar_test.go Outdated Show resolved Hide resolved

tests/scalers_go/pulsar/pulsar_test.go Outdated Show resolved Hide resolved

tests/scalers_go/pulsar/pulsar_test.go Outdated Show resolved Hide resolved

zroubalik reviewed Aug 8, 2022

View reviewed changes

tests/scalers_go/pulsar/pulsar_test.go Outdated Show resolved Hide resolved

Update tests/scalers_go/pulsar/pulsar_test.go

e6c734b

zroubalik merged commit f986280 into kedacore:main Aug 8, 2022

Add Apache Pulsar Scaler #3021

Add Apache Pulsar Scaler #3021

Conversation

liangyuanpeng commented May 8, 2022 • edited Loading

Checklist

liangyuanpeng commented May 15, 2022

JorTurFer commented May 15, 2022

JorTurFer left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

liangyuanpeng commented May 18, 2022 • edited Loading

JorTurFer commented May 21, 2022

JorTurFer commented May 21, 2022 • edited Loading

zroubalik commented May 31, 2022

liangyuanpeng commented May 31, 2022

zroubalik commented Jun 13, 2022

liangyuanpeng commented Jun 20, 2022

tomkerkhove commented Jul 11, 2022

liangyuanpeng commented Jul 12, 2022

zroubalik commented Jul 13, 2022

liangyuanpeng commented Jul 18, 2022

zroubalik commented Jul 18, 2022

zroubalik left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

liangyuanpeng Jul 20, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

JorTurFer left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zroubalik commented Jul 25, 2022 • edited by github-actions bot Loading

JorTurFer left a comment

Choose a reason for hiding this comment

zroubalik commented Aug 4, 2022

zroubalik commented Aug 8, 2022

zroubalik commented Aug 8, 2022 • edited by github-actions bot Loading

liangyuanpeng commented May 8, 2022 •

edited

Loading

liangyuanpeng commented May 18, 2022 •

edited

Loading

JorTurFer commented May 21, 2022 •

edited

Loading

liangyuanpeng Jul 20, 2022 •

edited

Loading

zroubalik commented Jul 25, 2022 •

edited by github-actions bot

Loading

zroubalik commented Aug 8, 2022 •

edited by github-actions bot

Loading