perf: check for experiments and auto operarions before subscribing to pubsub #626

Ubisoft-potato · 2023-12-26T02:01:27Z

Part of #612

Summary

Implement serverless architecture for persisters:
- event-persister-evaluation-events-ops
- event-persister-goal-events-ops
- event-persister-evaluation-events-dwh
- event-persister-goal-events-dwh
reference: perf: reduce costs in the background services #612

cre8ivejp · 2023-12-26T05:13:04Z

pkg/eventpersisterdwh/storage/v2/sql/count_experiment.sql

@@ -0,0 +1,4 @@
+select count(*)
+from experiment
+where deleted = 0


The conditions must match the conditions when listing the experiments.
The status and the 2-day buffer after the experiment has stopped.

Use this as a reference.
https://github.com/bucketeer-io/bucketeer/blob/main/pkg/eventpersisterdwh/persister/evaluation_event.go#L227-L279

Can you try this?
The current condition will get too much data unnecessarily.

SELECT (SELECT count(*) FROM experiment WHERE status = 1) + (SELECT count(*) FROM experiment WHERE status = 2 and DATE_ADD(from_unixtime(stop_at), INTERVAL 2 DAY) > NOW() ) as count;

Looks good, let me try it!

But one thing I need to mention is that performing calculations in the WHERE clause when execute this query, it will not use the index that we created.

I think we can use this sub query to count stopped and still in 2-days buffer experiments:

SELECT count(*) FROM experiment WHERE status = 2 AND stop_at > UNIX_TIMESTAMP(DATE_SUB(NOW(), INTERVAL 2 DAY))

So we will not do calculation on stop_at column.

Sounds good!

cre8ivejp · 2023-12-26T11:44:51Z

pkg/eventpersisterdwh/storage/v2/sql/count_experiment.sql

@@ -0,0 +1,4 @@
+SELECT (SELECT count(*) FROM experiment WHERE status = 1) + (SELECT count(*)


Can you format the query?

SELECT ( SELECT count(*) FROM experiment WHERE status = 1 ) + ( SELECT count(*) FROM experiment WHERE status = 2 AND stop_at > UNIX_TIMESTAMP( DATE_SUB(NOW(), INTERVAL 2 DAY) ) ) as count;

cre8ivejp · 2024-01-09T10:13:10Z

pkg/eventpersisterops/persister/persister.go

@@ -173,23 +173,29 @@ func (p *persister) Run() error {
 				continue
 			}
 			if exist {
+				p.logger.Debug("There are not been triggered auto ops rules")


nit: There are untriggered auto ops rules

cre8ivejp · 2024-01-09T10:13:45Z

pkg/eventpersisterops/persister/persister.go

 				}
 			} else {
+				p.logger.Debug("There are no not been triggered auto ops rules")


nit: There are no untriggered auto ops rules

cre8ivejp · 2024-01-09T10:16:29Z

pkg/eventpersisterdwh/persister/persister.go

@@ -204,7 +208,9 @@ func (p *PersisterDWH) Run() error {
 			}
 			timer.Reset(p.opts.checkInterval)
 		case <-p.ctx.Done():
+			p.logger.Info("Context is done")


Please change it to debug level.

cre8ivejp · 2024-01-09T10:16:37Z

pkg/eventpersisterdwh/persister/persister.go

 			if p.IsRunning() {
+				p.logger.Info("Puller is running, stop pulling messages")


cre8ivejp · 2024-01-09T10:16:48Z

pkg/eventpersisterops/persister/persister.go

 					p.rateLimitedPuller = puller.NewRateLimitedPuller(p.puller, p.opts.maxMPS)
 					p.group = errgroup.Group{}
 				}
 			}
 			timer.Reset(p.opts.checkInterval)
 		case <-p.ctx.Done():
+			p.logger.Info("Context is done")


Please change it to debug level.

cre8ivejp · 2024-01-09T10:16:54Z

pkg/eventpersisterops/persister/persister.go

 			if p.IsRunning() {
+				p.logger.Info("Puller is running, stop pulling messages")


Ubisoft-potato · 2024-01-15T07:14:57Z

pkg/eventpersisterdwh/persister/persister.go

@@ -177,21 +225,65 @@ func (p *PersisterDWH) Check(ctx context.Context) health.Status {
 		p.logger.Error("Unhealthy due to context Done is closed", zap.Error(p.ctx.Err()))
 		return health.Unhealthy
 	default:
-		if p.group.FinishedCount() > 0 {
-			p.logger.Error("Unhealthy", zap.Int32("FinishedCount", p.group.FinishedCount()))
+		if p.group.FailedCount() > 0 {


Since the persister will be run periodically, it's essential to monitor the FailedCount instead of the FinishedCount to prevent it from stopping after a health check.

Ubisoft-potato · 2024-01-15T07:15:34Z

pkg/eventpersisterops/persister/persister.go

@@ -161,21 +210,65 @@ func (p *persister) Check(ctx context.Context) health.Status {
 		p.logger.Error("Unhealthy due to context Done is closed", zap.Error(p.ctx.Err()))
 		return health.Unhealthy
 	default:
-		if p.group.FinishedCount() > 0 {
-			p.logger.Error("Unhealthy", zap.Int32("FinishedCount", p.group.FinishedCount()))
+		if p.group.FailedCount() > 0 {


github-actions · 2024-01-19T02:58:04Z

CLA Assistant Lite bot All Contributors have signed the CLA.

Ubisoft-potato · 2024-01-19T03:13:20Z

/cla sign

cre8ivejp · 2024-01-22T02:59:34Z

pkg/eventpersisterdwh/persister/persister.go

+	for {
+		select {
+		case <-timer.C:
+			// check if there are running experiment


Suggested change

// check if there are running experiment

// check if there are running experiments

cre8ivejp · 2024-01-22T03:03:10Z

pkg/eventpersisterdwh/persister/persister.go

+			if exist {
+				p.logger.Debug("There are running experiments")
+				if !p.IsRunning() {
+					p.group = errgroup.Group{}


This should be below err = p.createNewPuller(). Otherwise, if it fails to create the puller, it will leave the p.group set.

cre8ivejp · 2024-01-22T03:04:43Z

pkg/eventpersisterdwh/persister/persister.go

+					p.logger.Debug("Puller is running, stop pulling messages")
+					p.unsubscribe()
+				}
+				// delete subscription if subscription exists


Suggested change

// delete subscription if subscription exists

// delete subscription if it exists

cre8ivejp · 2024-01-22T03:09:41Z

pkg/eventpersisterdwh/persister/persister.go

+		select {
+		case <-subscription:
+			p.isRunning = true
+			p.logger.Debug("Puller start subscribing")


Suggested change

p.logger.Debug("Puller start subscribing")

p.logger.Debug("Puller started subscribing")

cre8ivejp · 2024-01-22T03:12:59Z

pkg/eventpersisterdwh/persister/persister.go

+						p.unsubscribe()
+						return nil
+					}
+					p.logger.Error("Puller pulling messages error", zap.Error(err))


Suggested change

p.logger.Error("Puller pulling messages error", zap.Error(err))

p.logger.Error("Failed to pull messages", zap.Error(err))

cre8ivejp · 2024-01-22T03:13:32Z

pkg/eventpersisterdwh/persister/persister.go

+				err := p.rateLimitedPuller.Run(ctx)
+				if err != nil {
+					if strings.Contains(err.Error(), pubsubErrNotFound) {
+						p.logger.Debug("Subscription does not exist",


Suggested change

p.logger.Debug("Subscription does not exist",

p.logger.Debug("Failed to pull messages. Subscription does not exist",

cre8ivejp · 2024-01-22T03:15:42Z

pkg/eventpersisterdwh/persister/persister.go

+			}
+			err := p.group.Wait()
+			if err != nil {
+				p.logger.Error("Running puller error", zap.Error(err))


Suggested change

p.logger.Error("Running puller error", zap.Error(err))

p.logger.Error("Failed while running pull messages", zap.Error(err))

cre8ivejp · 2024-01-22T03:18:30Z

pkg/pubsub/pubsub.go

@@ -246,7 +251,45 @@ func (c *Client) subscription(id, topicID string) (*pubsub.Subscription, error)
 		if err == nil {
 			return sub, nil
 		}
+		if strings.Contains(err.Error(), rpcErrAlreadyExists) {
+			c.logger.Debug("Subscription already exists, use it directly",


Suggested change

c.logger.Debug("Subscription already exists, use it directly",

c.logger.Debug("Subscription already exists. Using the current one",

cre8ivejp · 2024-01-22T03:21:49Z

pkg/pubsub/pubsub.go

Can you add this line before the continue here?

lastErr = err

We should save the last error. Otherwise, when it returns the timeout error, it won't have any error log.

cre8ivejp · 2024-01-22T03:22:50Z

pkg/pubsub/pubsub.go

+	sub := c.Client.Subscription(id)
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	err := sub.Delete(ctx)


Suggested change

err := sub.Delete(ctx)

return sub.Delete(ctx)

cre8ivejp · 2024-01-29T07:06:39Z

pkg/eventpersisterdwh/storage/v2/sql/count_experiment.sql

+            experiment
+        WHERE
+            status = 2
+          AND stop_at > UNIX_TIMESTAMP(


@Ubisoft-potato, can you fix the indent in these two lines?

cre8ivejp · 2024-01-29T07:08:31Z

pkg/eventpersisterops/storage/v2/sql/count_auto_ops_rules.sql

@@ -0,0 +1,4 @@
+select count(*)


To keep the consistency with the other queries, please use uppercase only.

cre8ivejp

Thank you!

Ubisoft-potato force-pushed the serverless branch from 947e7dc to 9ae10e4 Compare December 26, 2023 03:01

Ubisoft-potato marked this pull request as ready for review December 26, 2023 03:07

Ubisoft-potato requested review from cre8ivejp and masaaania December 26, 2023 03:07

cre8ivejp reviewed Dec 26, 2023

View reviewed changes

cre8ivejp reviewed Jan 9, 2024

View reviewed changes

Ubisoft-potato requested a review from cre8ivejp January 9, 2024 10:46

Ubisoft-potato commented Jan 15, 2024

View reviewed changes

cre8ivejp changed the title ~~feat: serverless for background service~~ perf: check for experiments and auto operarions before subscribing to pubsub Jan 15, 2024

Ubisoft-potato force-pushed the serverless branch 2 times, most recently from 1d4c16b to 1142e19 Compare January 18, 2024 09:18

Ubisoft-potato requested a review from kentakozuka as a code owner January 19, 2024 02:57

Ubisoft-potato force-pushed the serverless branch from d3e2277 to 17f6ad0 Compare January 19, 2024 06:27

cre8ivejp reviewed Jan 22, 2024

View reviewed changes

Ubisoft-potato and others added 12 commits January 29, 2024 12:00

feat: serverless arch

cbd20d2

fix: gofmt

28df9e3

feat: wait for puller shutdown

0895960

feat: handle stopped experiments

f006dec

feat: use sql to count experiments

5e79fde

chore: format sql

8da07d6

chore: add mysql params

a35fbce

fix: create rateLimitedPuller

4b15b0a

chore: add debug logs

444ae0a

chore: update log level

6ef2bef

feat: wait group

48cedc1

feat: delete subscription

a39cd25

Ubisoft-potato and others added 17 commits January 29, 2024 12:00

feat: go format

f278fc4

feat: add health check

dc0ee36

feat: remove health check

89b9a2f

feat: add pubsub log

d50816f

feat: using detach api

6fe3b0e

fix: mock puller

4dd69bd

fix: mock puller

c3df884

feat: check detach status before subscribing

e0bac67

feat: add debug log

5170200

feat: add debug log

15b9ae7

feat: add debug log

02393cc

feat: skip FailedPrecondition and NotFound error

3c0dbbc

feat: delete subscription

6174212

feat: improve create puller

fd40e95

feat: change to internal error

fc3cda3

feat: update sub delete log

ff7efd4

feat: minor changes

cdea535

Ubisoft-potato force-pushed the serverless branch from b97e6ef to cdea535 Compare January 29, 2024 04:01

cre8ivejp reviewed Jan 29, 2024

View reviewed changes

Ubisoft-potato added 2 commits January 29, 2024 15:14

fix: indent

3707bb4

feat: use uppercase

acd6f52

Ubisoft-potato requested a review from cre8ivejp January 29, 2024 07:18

cre8ivejp approved these changes Feb 13, 2024

View reviewed changes

cre8ivejp merged commit b4f98d3 into bucketeer-io:main Feb 13, 2024
11 checks passed

bucketeer-bot mentioned this pull request Feb 13, 2024

chore(main): release 0.4.0 #802

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

perf: check for experiments and auto operarions before subscribing to pubsub #626

perf: check for experiments and auto operarions before subscribing to pubsub #626

Ubisoft-potato commented Dec 26, 2023 •

edited by cre8ivejp

Loading

cre8ivejp Dec 26, 2023 •

edited

Loading

cre8ivejp Dec 26, 2023

Ubisoft-potato Dec 26, 2023 •

edited

Loading

Ubisoft-potato Dec 26, 2023

cre8ivejp Dec 26, 2023

cre8ivejp Dec 26, 2023

Ubisoft-potato Dec 26, 2023

cre8ivejp Jan 9, 2024

cre8ivejp Jan 9, 2024

cre8ivejp Jan 9, 2024

cre8ivejp Jan 9, 2024

cre8ivejp Jan 9, 2024

cre8ivejp Jan 9, 2024

Ubisoft-potato Jan 15, 2024 •

edited

Loading

Ubisoft-potato Jan 15, 2024

github-actions bot commented Jan 19, 2024 •

edited

Loading

Ubisoft-potato commented Jan 19, 2024

cre8ivejp Jan 22, 2024

cre8ivejp Jan 22, 2024

cre8ivejp Jan 22, 2024

cre8ivejp Jan 22, 2024

cre8ivejp Jan 22, 2024

cre8ivejp Jan 22, 2024

cre8ivejp Jan 22, 2024

cre8ivejp Jan 22, 2024

cre8ivejp Jan 22, 2024

cre8ivejp Jan 22, 2024

cre8ivejp Jan 29, 2024

Ubisoft-potato Jan 29, 2024 •

edited

Loading

cre8ivejp Jan 29, 2024

cre8ivejp left a comment

		@@ -0,0 +1,4 @@
		SELECT (SELECT count() FROM experiment WHERE status = 1) + (SELECT count()

		if p.IsRunning() {
		p.logger.Info("Puller is running, stop pulling messages")

	// check if there are running experiment
	// check if there are running experiments

	// delete subscription if subscription exists
	// delete subscription if it exists

	p.logger.Debug("Puller start subscribing")
	p.logger.Debug("Puller started subscribing")

	p.logger.Error("Puller pulling messages error", zap.Error(err))
	p.logger.Error("Failed to pull messages", zap.Error(err))

	p.logger.Debug("Subscription does not exist",
	p.logger.Debug("Failed to pull messages. Subscription does not exist",

	p.logger.Error("Running puller error", zap.Error(err))
	p.logger.Error("Failed while running pull messages", zap.Error(err))

	c.logger.Debug("Subscription already exists, use it directly",
	c.logger.Debug("Subscription already exists. Using the current one",

perf: check for experiments and auto operarions before subscribing to pubsub #626

perf: check for experiments and auto operarions before subscribing to pubsub #626

Conversation

Ubisoft-potato commented Dec 26, 2023 • edited by cre8ivejp Loading

Summary

cre8ivejp Dec 26, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Ubisoft-potato Dec 26, 2023 • edited Loading

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Ubisoft-potato Jan 15, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

github-actions bot commented Jan 19, 2024 • edited Loading

Ubisoft-potato commented Jan 19, 2024

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Ubisoft-potato Jan 29, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cre8ivejp left a comment

Choose a reason for hiding this comment

Ubisoft-potato commented Dec 26, 2023 •

edited by cre8ivejp

Loading

cre8ivejp Dec 26, 2023 •

edited

Loading

Ubisoft-potato Dec 26, 2023 •

edited

Loading

Ubisoft-potato Jan 15, 2024 •

edited

Loading

github-actions bot commented Jan 19, 2024 •

edited

Loading

Ubisoft-potato Jan 29, 2024 •

edited

Loading