feat: pipeline scheduler supports priority

backend/core/models/blueprint.go

@@ -41,6 +41,7 @@ type Blueprint struct {
 	AfterPlan    PipelinePlan           `json:"afterPlan" gorm:"serializer:encdec"`
 	Labels       []string               `json:"labels" gorm:"-"`
 	Connections  []*BlueprintConnection `json:"connections" gorm:"-"`
+	Priority     int                    `json:"priority"` // greater is higher
 	SyncPolicy   `gorm:"embedded"`
 	common.Model `swaggerignore:"true"`
 }

backend/core/models/migrationscripts/20250813_add_pipeline_priority.go

@@ -0,0 +1,57 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements.  See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package migrationscripts
+
+import (
+	"github.com/apache/incubator-devlake/core/context"
+	"github.com/apache/incubator-devlake/core/errors"
+	"github.com/apache/incubator-devlake/core/plugin"
+	"github.com/apache/incubator-devlake/helpers/migrationhelper"
+)
+
+var _ plugin.MigrationScript = (*addPipelinePriority)(nil)
+
+type addPipelinePriority struct{}
+
+type blueprint20250813 struct {
+	Priority int `json:"priority"`
+}
+
+func (blueprint20250813) TableName() string {
+	return "_devlake_blueprints"
+}
+
+type pipeline20250813 struct {
+	Priority int `json:"priority"`
+}
+
+func (pipeline20250813) TableName() string {
+	return "_devlake_pipelines"
+}
+
+func (script *addPipelinePriority) Up(basicRes context.BasicRes) errors.Error {
+	return migrationhelper.AutoMigrateTables(basicRes, new(blueprint20250813), new(pipeline20250813))
+}
+
+func (*addPipelinePriority) Version() uint64 {
+	return 20250813151534
+}
+
+func (*addPipelinePriority) Name() string {
+	return "add priority to blueprints and pipelines"
+}

backend/core/models/migrationscripts/register.go

@@ -139,5 +139,6 @@ func All() []plugin.MigrationScript {
 		new(increaseCqIssueComponentLength),
 		new(extendFieldSizeForCq),
 		new(addIssueFixVerion),
+		new(addPipelinePriority),
 	}
 }

backend/core/models/pipeline.go

@@ -66,6 +66,7 @@ type Pipeline struct {
 	SpentSeconds  int          `json:"spentSeconds"`
 	Stage         int          `json:"stage"`
 	Labels        []string     `json:"labels" gorm:"-"`
+	Priority      int          `json:"priority"` // greater is higher
 	SyncPolicy    `gorm:"embedded"`
 }
 
@@ -75,6 +76,7 @@ type NewPipeline struct {
 	Name        string       `json:"name"`
 	Plan        PipelinePlan `json:"plan" swaggertype:"array,string" example:"please check api /pipelines/<PLUGIN_NAME>/pipeline-plan"`
 	Labels      []string     `json:"labels"`
+	Priority    int          `json:"priority"` // greater is higher
 	BlueprintId uint64
 	SyncPolicy  `gorm:"embedded"`
 }

backend/plugins/org/impl/impl.go

@@ -61,6 +61,7 @@ func (p Org) SubTaskMetas() []plugin.SubTaskMeta {
 	return []plugin.SubTaskMeta{
 		tasks.ConnectUserAccountsExactMeta,
 		tasks.SetProjectMappingMeta,
+		tasks.SleepMeta,
 	}
 }
 

backend/plugins/org/tasks/sleep.go

@@ -0,0 +1,40 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements.  See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package tasks
+
+import (
+	"time"
+
+	"github.com/apache/incubator-devlake/core/errors"
+	"github.com/apache/incubator-devlake/core/plugin"
+)
+
+var SleepMeta = plugin.SubTaskMeta{
+	Name:             "sleep",
+	EntryPoint:       Sleep,
+	EnabledByDefault: false,
+	Description:      "for debugging only",
+	DomainTypes:      []string{plugin.DOMAIN_TYPE_CROSS},
+}
+
+// SetProjectMapping binds projects and scopes
+func Sleep(taskCtx plugin.SubTaskContext) errors.Error {
+	data := taskCtx.GetData().(*TaskData)
+	time.Sleep(time.Duration(data.Options.SleepSeconds) * time.Second)
+	return nil
+}

backend/plugins/org/tasks/task_data.go

@@ -22,6 +22,7 @@ import "github.com/apache/incubator-devlake/core/plugin"
 type Options struct {
 	ConnectionId    uint64           `json:"connectionId"`
 	ProjectMappings []ProjectMapping `json:"projectMappings"`
+	SleepSeconds    uint64           `json:"sleepSeconds"`
 }
 
 // ProjectMapping represents the relations between project and scopes

backend/server/services/blueprint.go

@@ -327,6 +327,7 @@ func createPipelineByBlueprint(blueprint *models.Blueprint, syncPolicy *models.S
 	newPipeline.Name = blueprint.Name
 	newPipeline.BlueprintId = blueprint.ID
 	newPipeline.Labels = blueprint.Labels
+	newPipeline.Priority = blueprint.Priority
 	newPipeline.SyncPolicy = blueprint.SyncPolicy
 
 	// if the plan is empty, we should not create the pipeline

backend/server/services/pipeline.go

@@ -273,7 +273,7 @@ func dequeuePipeline(runningParallelLabels []string) (pipeline *models.Pipeline,
 		dal.Groupby("id"),
 		dal.Having("count(_devlake_pipeline_labels.name)=0"),
 		dal.Select("id"),
-		dal.Orderby("id ASC"),
+		dal.Orderby("priority DESC, id ASC"),
 		dal.Limit(1),
 	)
 	if err == nil {

backend/server/services/pipeline_helper.go

@@ -67,6 +67,7 @@ func CreateDbPipeline(newPipeline *models.NewPipeline) (pipeline *models.Pipelin
 		Message:       "",
 		SpentSeconds:  0,
 		Plan:          newPipeline.Plan,
+		Priority:      newPipeline.Priority,
 		SyncPolicy:    newPipeline.SyncPolicy,
 	}
 	if newPipeline.BlueprintId != 0 {

config-ui/.yarnrc.yml

@@ -16,4 +16,6 @@
 #
 nodeLinker: node-modules
 
+npmRegistryServer: "https://registry.npmmirror.com"
+
 yarnPath: .yarn/releases/yarn-3.4.1.cjs

config-ui/package.json

@@ -69,5 +69,9 @@
     "typescript": "^5.1.6",
     "vite": "^5.1.4",
     "vite-plugin-svgr": "^4.2.0"
+  },
+  "volta": {
+    "node": "18.20.8",
+    "yarn": "3.4.1"
   }
 }