(incubator-devlake) branch kw-6983-optimize-bp-update created (now 36d7f5194)

[kl...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

klesh pushed a change to branch kw-6983-optimize-bp-update
in repository https://gitbox.apache.org/repos/asf/incubator-devlake.git


      at 36d7f5194 refactor: optimize bp cronjob scheduling

This branch includes the following new commits:

     new 36d7f5194 refactor: optimize bp cronjob scheduling

The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails.  The revisions
listed as "add" were already present in the repository and have only
been added to this reference.

(incubator-devlake) 01/01: refactor: optimize bp cronjob scheduling

[kl...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

klesh pushed a commit to branch kw-6983-optimize-bp-update
in repository https://gitbox.apache.org/repos/asf/incubator-devlake.git

commit 36d7f5194402a2b36988763b1d63b111b69084c8
Author: Klesh Wong <zh...@merico.dev>
AuthorDate: Thu Feb 22 16:19:23 2024 +0800

    refactor: optimize bp cronjob scheduling
---
 backend/server/services/blueprint.go | 73 ++++++++++++++++++------------------
 backend/server/services/pipeline.go  |  2 +-
 2 files changed, 38 insertions(+), 37 deletions(-)

diff --git a/backend/server/services/blueprint.go b/backend/server/services/blueprint.go
index fa2850895..447286a2d 100644
--- a/backend/server/services/blueprint.go
+++ b/backend/server/services/blueprint.go
@@ -68,22 +68,8 @@ func (bj BlueprintJob) Run() {
 
 // CreateBlueprint accepts a Blueprint instance and insert it to database
 func CreateBlueprint(blueprint *models.Blueprint) errors.Error {
-	err := validateBlueprintAndMakePlan(blueprint)
-	if err != nil {
-		return err
-	}
-	err = bpManager.SaveDbBlueprint(blueprint)
-	if err != nil {
-		return err
-	}
-	if err := SanitizeBlueprint(blueprint); err != nil {
-		return errors.Convert(err)
-	}
-	err = ReloadBlueprints(cronManager)
-	if err != nil {
-		return errors.Internal.Wrap(err, "error reloading blueprints")
-	}
-	return nil
+	_, err := saveBlueprint(blueprint)
+	return err
 }
 
 // GetBlueprints returns a paginated list of Blueprints based on `query`
@@ -204,9 +190,9 @@ func saveBlueprint(blueprint *models.Blueprint) (*models.Blueprint, errors.Error
 	}
 
 	// reload schedule
-	err = ReloadBlueprints(cronManager)
+	err = reloadBlueprint(blueprint)
 	if err != nil {
-		return nil, errors.Internal.Wrap(err, "error reloading blueprints")
+		return nil, err
 	}
 	// done
 	return blueprint, nil
@@ -260,13 +246,10 @@ func DeleteBlueprint(id uint64) errors.Error {
 }
 
 var blueprintReloadLock sync.Mutex
+var bpCronIdMap map[uint64]cron.EntryID
 
-// ReloadBlueprints FIXME ...
-func ReloadBlueprints(c *cron.Cron) (err errors.Error) {
-	// preventing concurrent reloads. It would be better to use Table Lock , however, it requires massive refactor
-	// like the `bpManager` must accept transaction. Use mutex as a temporary fix.
-	blueprintReloadLock.Lock()
-	defer blueprintReloadLock.Unlock()
+// ReloadBlueprints reloades cronjobs based on blueprints
+func ReloadBlueprints() (err errors.Error) {
 
 	enable := true
 	isManual := false
@@ -277,27 +260,45 @@ func ReloadBlueprints(c *cron.Cron) (err errors.Error) {
 	if err != nil {
 		return err
 	}
-	for _, e := range c.Entries() {
-		c.Remove(e.ID)
+	for _, e := range cronManager.Entries() {
+		cronManager.Remove(e.ID)
 	}
-	c.Stop()
+	cronManager.Stop()
+	bpCronIdMap = make(map[uint64]cron.EntryID, len(blueprints))
 	for _, blueprint := range blueprints {
-		blueprintLog.Info("Add blueprint id:[%d] cronConfg[%s] to cron job", blueprint.ID, blueprint.CronConfig)
-		blueprintJob := &BlueprintJob{
-			Blueprint: blueprint,
-		}
-		if _, err := c.AddJob(blueprint.CronConfig, blueprintJob); err != nil {
-			blueprintLog.Error(err, failToCreateCronJob)
-			return errors.Default.Wrap(err, "created cron job failed")
-		}
+		reloadBlueprint(blueprint)
 	}
 	if len(blueprints) > 0 {
-		c.Start()
+		cronManager.Start()
 	}
 	logger.Info("total %d blueprints were scheduled", len(blueprints))
 	return nil
 }
 
+func reloadBlueprint(blueprint *models.Blueprint) errors.Error {
+	// preventing concurrent reloads. It would be better to use Table Lock , however, it requires massive refactor
+	// like the `bpManager` must accept transaction. Use mutex as a temporary fix.
+	blueprintReloadLock.Lock()
+	defer blueprintReloadLock.Unlock()
+
+	cronId, scheduled := bpCronIdMap[blueprint.ID]
+	if scheduled {
+		cronManager.Remove(cronId)
+		delete(bpCronIdMap, blueprint.ID)
+		logger.Info("removed blueprint %d from cronjobs, cron id: %v", blueprint.ID, cronId)
+	}
+	if blueprint.Enable && blueprint.IsManual == false {
+		if cronId, err := cronManager.AddJob(blueprint.CronConfig, &BlueprintJob{blueprint}); err != nil {
+			blueprintLog.Error(err, failToCreateCronJob)
+			return errors.Default.Wrap(err, "created cron job failed")
+		} else {
+			bpCronIdMap[blueprint.ID] = cronId
+			logger.Info("added blueprint %d to cronjobs, cron id: %v, cron config: %s", blueprint.ID, cronId, blueprint.CronConfig)
+		}
+	}
+	return nil
+}
+
 func createPipelineByBlueprint(blueprint *models.Blueprint, syncPolicy *models.SyncPolicy) (*models.Pipeline, errors.Error) {
 	var plan models.PipelinePlan
 	var err errors.Error
diff --git a/backend/server/services/pipeline.go b/backend/server/services/pipeline.go
index 1783c9421..2a1ac8ccb 100644
--- a/backend/server/services/pipeline.go
+++ b/backend/server/services/pipeline.go
@@ -103,7 +103,7 @@ func pipelineServiceInit() {
 		panic(err)
 	}
 
-	err = ReloadBlueprints(cronManager)
+	err = ReloadBlueprints()
 	if err != nil {
 		panic(err)
 	}