feat(ios-build): dispatch runner task by org id

.erda/migrations/pipeline/20240702-runner-scheduler-add-orgid.sql

@@ -0,0 +1,4 @@
+ALTER TABLE `dice_runner_tasks`
+    ADD COLUMN `org_id` bigint(20) NOT NULL DEFAULT '0' COMMENT 'org id' AFTER `job_id`;
+# add index
+CREATE INDEX `idx_org_id` ON `dice_runner_tasks` (`org_id`);

api/proto/core/pipeline/action_runner_scheduler/runner_task.proto

@@ -5,6 +5,7 @@ package erda.core.pipeline.action_runner_scheduler;
 import "google/api/annotations.proto";
 import "common/identity.proto";
 import "common/openapi.proto";
+import "github.com/envoyproxy/protoc-gen-validate/validate/validate.proto";
 
 option go_package = "github.com/erda-project/erda-proto-go/core/pipeline/action_runner_scheduler/pb";
 
@@ -87,6 +88,7 @@ message RunnerTask {
   repeated string commands = 7;
   repeated string targets = 8;
   string workdir = 9;
+  int64 orgID = 10 [json_name = "org_id"];
 }
 
 message RunnerTaskQueryRequest {
@@ -97,7 +99,9 @@ message RunnerTaskQueryResponse {
   RunnerTask data = 1;
 }
 
-message RunnerTaskFetchRequest {}
+message RunnerTaskFetchRequest {
+  repeated int64 org_id = 1 [(validate.rules).repeated.min_items = 1, (validate.rules).repeated.items.int64.gt = 0];
+}
 
 message RunnerTaskFetchResponse {
   repeated RunnerTask data = 1;
@@ -109,6 +113,7 @@ message RunnerTaskCreateRequest {
   repeated string commands = 3;
   repeated string targets = 4;
   string workdir = 5;
+  int64 orgID = 6 [json_name = "org_id", (validate.rules).int64.gt = 0];
 }
 
 message RunnerTaskCreateResponse {

apistructs/runner_task.go

@@ -25,6 +25,7 @@ const (
 type RunnerTask struct {
 	ID             uint64   `json:"id"`
 	JobID          string   `json:"job_id"`
+	OrgID          int64    `json:"org_id"`
 	Status         string   `json:"status"` // pending running success failed
 	ContextDataUrl string   `json:"context_data_url"`
 	OpenApiToken   string   `json:"openapi_token"`
@@ -45,6 +46,7 @@ type QueryRunnerTaskResponse struct {
 
 type CreateRunnerTaskRequest struct {
 	JobID          string   `json:"job_id"`
+	OrgID          int64    `json:"org_id"`
 	ContextDataUrl string   `json:"context_data_url"`
 	Commands       []string `json:"commands"`
 	Targets        []string `json:"targets"`

cmd/action-runner/build.sh

@@ -0,0 +1,14 @@
+#!/usr/bin/env bash
+
+set -o errexit -o pipefail
+
+# goto current bash script dir
+cd "$(dirname "$0")"
+
+rm -fr bin
+
+ARCHES="amd64 arm64"
+
+for arch in $ARCHES; do
+    GOOS=darwin GOARCH=$arch go build -o bin/action-runner-$arch
+done

cmd/action-runner/config.json

@@ -1,4 +1,5 @@
 {
+  "org_ids": [2, 3, 4],
   "build_path": "/tmp/xxx",
   "failed_task_keep_hours": 3,
   "open_api": "",
@@ -15,4 +16,4 @@
   ],
   "upload_timeout_sec": 500,
   "print_upload_interval_sec": 5
-}
+}

cmd/action-runner/main.go

@@ -53,11 +53,16 @@ func readConfig(path string) *actionrunner.Conf {
 			os.Exit(-1)
 		}
 	}
+	// check org ids
+	if len(conf.OrgIDs) == 0 {
+		logrus.Error("org_ids could not be empty in config.json, format: [1, 2]")
+		os.Exit(-1)
+	}
 	conf.BuildPath = getEnv("BUILD_ROOT_PATH", conf.BuildPath)
 	if len(conf.BuildPath) <= 0 {
 		conf.BuildPath = os.TempDir()
 	}
-	conf.OpenAPI = getEnv("OPENAPI_UEL", conf.OpenAPI)
+	conf.OpenAPI = getEnv("OPENAPI_URL", conf.OpenAPI)
 	conf.Token = getEnv("TOKEN", conf.Token)
 	if conf.MaxTask < 1 {
 		conf.MaxTask = 1

internal/tools/pipeline/action-runner/apis.go

@@ -25,6 +25,7 @@ import (
 	"github.com/sirupsen/logrus"
 
 	"github.com/erda-project/erda-proto-go/core/file/pb"
+	runnerpb "github.com/erda-project/erda-proto-go/core/pipeline/action_runner_scheduler/pb"
 	"github.com/erda-project/erda/apistructs"
 	"github.com/erda-project/erda/pkg/http/httpclient"
 )
@@ -42,7 +43,7 @@ func (r *Runner) fetchTasks() []*Task {
 		Header("Content-Type", "application/json").
 		Header("Authorization", r.Conf.Token)
 	var resp TaskListResponse
-	httpResp, err := request.Do().JSON(&resp)
+	httpResp, err := request.JSONBody(&runnerpb.RunnerTaskFetchRequest{OrgId: r.Conf.OrgIDs}).Do().JSON(&resp)
 	if err != nil {
 		logrus.Errorf("fail to fetch task: %s", err)
 		return nil

internal/tools/pipeline/action-runner/conf.go

@@ -16,6 +16,7 @@ package actionrunner
 
 // Conf .
 type Conf struct {
+	OrgIDs                 []int64           `json:"org_ids"`
 	BuildPath              string            `json:"build_path"`
 	OpenAPI                string            `json:"open_api"`
 	Token                  string            `json:"token"`

internal/tools/pipeline/providers/action_runner_scheduler/db/runner_task.go

@@ -25,6 +25,7 @@ import (
 type RunnerTask struct {
 	dbengine.BaseModel
 	JobID          string `json:"job_id"`
+	OrgID          int64  `json:"org_id"`
 	Status         string `json:"status"` // pending running success failed
 	OpenApiToken   string `json:"openapi_token"`
 	ContextDataUrl string `json:"context_data_url"`
@@ -43,6 +44,7 @@ func (task RunnerTask) ToApiData() *apistructs.RunnerTask {
 	result := &apistructs.RunnerTask{
 		ID:             task.ID,
 		JobID:          task.JobID,
+		OrgID:          task.OrgID,
 		Status:         task.Status,
 		ContextDataUrl: task.ContextDataUrl,
 		OpenApiToken:   task.OpenApiToken,
@@ -60,6 +62,7 @@ func (task RunnerTask) ToPbData() *pb.RunnerTask {
 	result := &pb.RunnerTask{
 		ID:             task.ID,
 		JobID:          task.JobID,
+		OrgID:          task.OrgID,
 		Status:         task.Status,
 		ContextDataUrl: task.ContextDataUrl,
 		OpenApiToken:   task.OpenApiToken,
@@ -78,6 +81,7 @@ func (db *DBClient) CreateRunnerTask(request *pb.RunnerTaskCreateRequest) (uint6
 	targets, _ := json.Marshal(request.Targets)
 	task := &RunnerTask{
 		JobID:          request.JobID,
+		OrgID:          request.OrgID,
 		Status:         apistructs.RunnerTaskStatusPending,
 		ContextDataUrl: request.ContextDataUrl,
 		ResultDataUrl:  "",
@@ -92,11 +96,17 @@ func (db *DBClient) CreateRunnerTask(request *pb.RunnerTaskCreateRequest) (uint6
 	return task.ID, nil
 }
 
-func (db *DBClient) GetFirstPendingTask() (*RunnerTask, error) {
+func (db *DBClient) GetFirstPendingTask(orgIDs []int64) (*RunnerTask, error) {
 	var list []RunnerTask
-	err := db.Model(&RunnerTask{}).
-		Where("status =?", apistructs.RunnerTaskStatusPending).
-		Limit(1).Find(&list).Error
+	sql := db.Model(&RunnerTask{}).
+		Where("status = ?", apistructs.RunnerTaskStatusPending)
+	// handle org id
+	if len(orgIDs) > 0 {
+		sql = sql.Where("org_id in (?)", orgIDs)
+	}
+	// order, oldest first
+	sql = sql.Order("id ASC")
+	err := sql.Limit(1).Find(&list).Error
 	if err != nil {
 		return nil, err
 	}

internal/tools/pipeline/providers/action_runner_scheduler/runner.task.service.go

@@ -66,7 +66,7 @@ func (s *runnerTaskService) GetRunnerTask(ctx context.Context, req *pb.RunnerTas
 }
 
 func (s *runnerTaskService) FetchRunnerTask(ctx context.Context, req *pb.RunnerTaskFetchRequest) (*pb.RunnerTaskFetchResponse, error) {
-	task, err := s.dbClient.GetFirstPendingTask()
+	task, err := s.dbClient.GetFirstPendingTask(req.OrgId)
 	if err != nil {
 		return nil, apierrors.ErrFetchRunnerTask.InternalError(err)
 	}