建立 Go Task Scheduler 协议骨架

2026-09-20 17:52:38 +00:00
parent 5e5f2cc48d
commit 3c1fca1832
10 changed files with 394 additions and 0 deletions
@@ -6,6 +6,16 @@ on:
  pull_request:

 jobs:
+  go:
+    runs-on: [self-hosted, pod]
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-go@v6
+        with:
+          go-version-file: go.mod
+      - run: go test ./...
+      - run: go vet ./...
+
  python:
    runs-on: self-hosted
    steps:
@@ -29,6 +29,8 @@ dynamic-runner scheduler
 - 每个 executor 只执行一个 task，完成后销毁。
 - SPIFFE 身份从实际领取的 task 的 repository 和 job name 派生，不需要 queued 与
  in-progress webhook 的二阶段关联。
+- scheduler 只做确定性的身份派生与 executor 绑定，不维护业务授权 policy；Zot、
+  OpenBao 等资源服务继续是唯一授权决策点。
 - scheduler 的 runner registration credential 不进入 executor；executor 只得到执行
  当前 task 所需的短期 lease/capability。
 - task ACK、心跳和结果必须能够跨 scheduler 重启恢复；NATS 可以继续作为内部 handoff，
@@ -0,0 +1,9 @@
+module git.ddupan.top/panxiao81/gitea-dynamic-runner
+
+go 1.27
+
+require (
+	connectrpc.com/connect v1.20.0
+	gitea.dev/actionslib v1.0.0
+	google.golang.org/protobuf v1.36.12
+)
@@ -0,0 +1,8 @@
+connectrpc.com/connect v1.20.0 h1:6TNDAB+WeNd2uolWNlYczB5E0KNNaVMNUEx8JEUsPmQ=
+connectrpc.com/connect v1.20.0/go.mod h1:A2ygJrukXwWy32vkCAAHNVguZrqZ+jeZ9rGRnGR4dN4=
+gitea.dev/actionslib v1.0.0 h1:l0oFJP+P4Ds1rlCI5zk618dYkuBc2mU7Gz5wPeG0lZY=
+gitea.dev/actionslib v1.0.0/go.mod h1:6O8YHkqVTKSR0LL2e5VhIDePYzGTZCbfmSVqJWEhk9g=
+github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
+github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
+google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc=
+google.golang.org/protobuf v1.36.12/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
@@ -0,0 +1,61 @@
+// Package giteaactions provides the authenticated Gitea RunnerService client.
+package giteaactions
+
+import (
+	"context"
+	"net/http"
+	"strings"
+
+	"connectrpc.com/connect"
+	"gitea.dev/actionslib/pkg/protocol"
+	runnerv1 "gitea.dev/actionslib/runner/v1"
+	"gitea.dev/actionslib/runner/v1/runnerv1connect"
+)
+
+// Client is the subset of RunnerService owned by the scheduler.
+type Client struct {
+	runner runnerv1connect.RunnerServiceClient
+}
+
+// NewClient authenticates every RPC with the persistent scheduler runner.
+func NewClient(httpClient connect.HTTPClient, instanceURL, uuid, token string) *Client {
+	auth := connect.UnaryInterceptorFunc(func(next connect.UnaryFunc) connect.UnaryFunc {
+		return func(ctx context.Context, request connect.AnyRequest) (connect.AnyResponse, error) {
+			request.Header().Set("User-Agent", "gitea-dynamic-runner-go/0")
+			request.Header().Set(protocol.UUIDHeader, uuid)
+			request.Header().Set(protocol.TokenHeader, token)
+			return next(ctx, request)
+		}
+	})
+	baseURL := strings.TrimRight(instanceURL, "/") + "/api/actions"
+	return &Client{runner: runnerv1connect.NewRunnerServiceClient(
+		httpClient,
+		baseURL,
+		connect.WithInterceptors(auth),
+	)}
+}
+
+// Declare advertises the scheduler labels before tasks are fetched.
+func (c *Client) Declare(ctx context.Context, version string, labels []string) error {
+	_, err := c.runner.Declare(ctx, connect.NewRequest(&runnerv1.DeclareRequest{
+		Version: version,
+		Labels:  labels,
+	}))
+	return err
+}
+
+// FetchTask asks Gitea to atomically assign the next matching task.
+func (c *Client) FetchTask(ctx context.Context, tasksVersion int64) (*runnerv1.FetchTaskResponse, error) {
+	response, err := c.runner.FetchTask(ctx, connect.NewRequest(&runnerv1.FetchTaskRequest{
+		TasksVersion: tasksVersion,
+	}))
+	if err != nil {
+		return nil, err
+	}
+	return response.Msg, nil
+}
+
+// DefaultHTTPClient is suitable for the scheduler's long-lived connection.
+func DefaultHTTPClient() *http.Client {
+	return &http.Client{Transport: http.DefaultTransport}
+}
@@ -0,0 +1,78 @@
+package giteaactions
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"reflect"
+	"testing"
+
+	"connectrpc.com/connect"
+	"gitea.dev/actionslib/pkg/protocol"
+	runnerv1 "gitea.dev/actionslib/runner/v1"
+	"gitea.dev/actionslib/runner/v1/runnerv1connect"
+)
+
+type runnerService struct {
+	runnerv1connect.UnimplementedRunnerServiceHandler
+	t              *testing.T
+	declaredLabels []string
+	fetchedVersion int64
+	expectedUUID   string
+	expectedToken  string
+}
+
+func (s *runnerService) checkAuth(request connect.AnyRequest) {
+	s.t.Helper()
+	if got := request.Header().Get(protocol.UUIDHeader); got != s.expectedUUID {
+		s.t.Errorf("runner UUID header = %q", got)
+	}
+	if got := request.Header().Get(protocol.TokenHeader); got != s.expectedToken {
+		s.t.Errorf("runner token header = %q", got)
+	}
+}
+
+func (s *runnerService) Declare(_ context.Context, request *connect.Request[runnerv1.DeclareRequest]) (*connect.Response[runnerv1.DeclareResponse], error) {
+	s.checkAuth(request)
+	s.declaredLabels = request.Msg.Labels
+	return connect.NewResponse(&runnerv1.DeclareResponse{}), nil
+}
+
+func (s *runnerService) FetchTask(_ context.Context, request *connect.Request[runnerv1.FetchTaskRequest]) (*connect.Response[runnerv1.FetchTaskResponse], error) {
+	s.checkAuth(request)
+	s.fetchedVersion = request.Msg.TasksVersion
+	return connect.NewResponse(&runnerv1.FetchTaskResponse{
+		Task:         &runnerv1.Task{Id: 42},
+		TasksVersion: 8,
+	}), nil
+}
+
+func TestClientUsesOfficialRunnerProtocol(t *testing.T) {
+	service := &runnerService{
+		t:             t,
+		expectedUUID:  "runner-uuid",
+		expectedToken: "runner-token",
+	}
+	path, handler := runnerv1connect.NewRunnerServiceHandler(service)
+	mux := http.NewServeMux()
+	mux.Handle("/api/actions"+path, http.StripPrefix("/api/actions", handler))
+	server := httptest.NewServer(mux)
+	defer server.Close()
+
+	client := NewClient(server.Client(), server.URL, service.expectedUUID, service.expectedToken)
+	labels := []string{"self-hosted", "pod", "vm"}
+	if err := client.Declare(context.Background(), "0.1.0", labels); err != nil {
+		t.Fatal(err)
+	}
+	response, err := client.FetchTask(context.Background(), 7)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if !reflect.DeepEqual(service.declaredLabels, labels) {
+		t.Fatalf("declared labels = %#v", service.declaredLabels)
+	}
+	if service.fetchedVersion != 7 || response.GetTasksVersion() != 8 || response.GetTask().GetId() != 42 {
+		t.Fatalf("unexpected FetchTask exchange: request=%d response=%v", service.fetchedVersion, response)
+	}
+}
@@ -0,0 +1,71 @@
+// Package taskidentity derives workload identities from tasks assigned by Gitea.
+package taskidentity
+
+import (
+	"crypto/sha256"
+	"errors"
+	"fmt"
+	"regexp"
+	"strings"
+
+	runnerv1 "gitea.dev/actionslib/runner/v1"
+)
+
+var safeSegment = regexp.MustCompile(`^[A-Za-z0-9._-]+$`)
+
+// Identity is the trusted identity context extracted from a fetched task.
+type Identity struct {
+	Repository string
+	Task       string
+	SPIFFEID   string
+}
+
+// FromTask derives the repository/task SPIFFE ID from Gitea's trusted task
+// context. Workflow input never supplies or overrides the resulting ID.
+func FromTask(task *runnerv1.Task, trustDomain string) (Identity, error) {
+	if task == nil || task.Context == nil {
+		return Identity{}, errors.New("task context is required")
+	}
+
+	repository := strings.TrimSpace(task.Context.GetFields()["repository"].GetStringValue())
+	taskName := strings.TrimSpace(task.Context.GetFields()["job"].GetStringValue())
+	parts := strings.Split(repository, "/")
+	if len(parts) != 2 || parts[0] == "" || parts[1] == "" {
+		return Identity{}, fmt.Errorf("invalid task repository %q", repository)
+	}
+	if taskName == "" {
+		return Identity{}, errors.New("task job is required")
+	}
+	trustDomain = strings.TrimSpace(trustDomain)
+	if trustDomain == "" || strings.ContainsAny(trustDomain, "/ ") {
+		return Identity{}, fmt.Errorf("invalid trust domain %q", trustDomain)
+	}
+
+	path := strings.Join([]string{
+		sanitize(parts[0]),
+		sanitize(parts[1]),
+		sanitize(taskName),
+	}, "/")
+	return Identity{
+		Repository: repository,
+		Task:       taskName,
+		SPIFFEID:   "spiffe://" + trustDomain + "/ci/" + path,
+	}, nil
+}
+
+// sanitize intentionally matches the bootstrap Python implementation so
+// existing Zot and OpenBao policies keep their current identity names.
+func sanitize(value string) string {
+	if safeSegment.MatchString(value) {
+		return value
+	}
+	slug := strings.Trim(regexp.MustCompile(`[^A-Za-z0-9._-]+`).ReplaceAllString(value, "-"), "-._")
+	if len(slug) > 48 {
+		slug = strings.TrimRight(slug[:48], "-._")
+	}
+	if slug == "" {
+		slug = "segment"
+	}
+	digest := fmt.Sprintf("%x", sha256.Sum256([]byte(value)))[:12]
+	return slug + "-" + digest
+}
@@ -0,0 +1,64 @@
+package taskidentity
+
+import (
+	"testing"
+
+	runnerv1 "gitea.dev/actionslib/runner/v1"
+	"google.golang.org/protobuf/types/known/structpb"
+)
+
+func TestFromTaskUsesFetchedContext(t *testing.T) {
+	ctx, err := structpb.NewStruct(map[string]any{
+		"repository": "panxiao81/gitea-dynamic-runner",
+		"job":        "publish-images",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	got, err := FromTask(&runnerv1.Task{Id: 900, Context: ctx}, "ddupan.top")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got.Repository != "panxiao81/gitea-dynamic-runner" || got.Task != "publish-images" {
+		t.Fatalf("unexpected task identity context: %#v", got)
+	}
+	want := "spiffe://ddupan.top/ci/panxiao81/gitea-dynamic-runner/publish-images"
+	if got.SPIFFEID != want {
+		t.Fatalf("SPIFFE ID = %q, want %q", got.SPIFFEID, want)
+	}
+}
+
+func TestFromTaskMatchesBootstrapNormalization(t *testing.T) {
+	ctx, err := structpb.NewStruct(map[string]any{
+		"repository": "panxiao81/postgresql-tenant-operator",
+		"job":        "Run on Ubuntu",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	got, err := FromTask(&runnerv1.Task{Context: ctx}, "ddupan.top")
+	if err != nil {
+		t.Fatal(err)
+	}
+	want := "spiffe://ddupan.top/ci/panxiao81/postgresql-tenant-operator/Run-on-Ubuntu-8b7cd4c244fb"
+	if got.SPIFFEID != want {
+		t.Fatalf("SPIFFE ID = %q, want %q", got.SPIFFEID, want)
+	}
+}
+
+func TestFromTaskRejectsIncompleteServerContext(t *testing.T) {
+	for _, fields := range []map[string]any{
+		{"repository": "invalid", "job": "test"},
+		{"repository": "owner/repo", "job": ""},
+	} {
+		ctx, err := structpb.NewStruct(fields)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if _, err := FromTask(&runnerv1.Task{Context: ctx}, "ddupan.top"); err == nil {
+			t.Fatalf("expected invalid task context to fail: %#v", fields)
+		}
+	}
+}
@@ -0,0 +1,42 @@
+// Package taskscheduler owns tasks fetched through Gitea's RunnerService and
+// dispatches them to an executor only after their identity is known.
+package taskscheduler
+
+import (
+	"context"
+	"errors"
+
+	runnerv1 "gitea.dev/actionslib/runner/v1"
+
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskidentity"
+)
+
+// Assignment is the immutable input handed to a Pod or VM provisioner.
+type Assignment struct {
+	Task     *runnerv1.Task
+	Identity taskidentity.Identity
+}
+
+// Dispatcher creates exactly one executor for an already assigned Gitea task.
+// It must not register another runner or ask Gitea for a task.
+type Dispatcher interface {
+	Dispatch(context.Context, Assignment) error
+}
+
+// Scheduler implements the TaskRunner boundary used by Gitea Runner's poller.
+type Scheduler struct {
+	TrustDomain string
+	Dispatcher  Dispatcher
+}
+
+// Run derives identity from the fetched task before provisioning its executor.
+func (s *Scheduler) Run(ctx context.Context, task *runnerv1.Task) error {
+	if s.Dispatcher == nil {
+		return errors.New("executor dispatcher is required")
+	}
+	identity, err := taskidentity.FromTask(task, s.TrustDomain)
+	if err != nil {
+		return err
+	}
+	return s.Dispatcher.Dispatch(ctx, Assignment{Task: task, Identity: identity})
+}
@@ -0,0 +1,49 @@
+package taskscheduler
+
+import (
+	"context"
+	"testing"
+
+	runnerv1 "gitea.dev/actionslib/runner/v1"
+	"google.golang.org/protobuf/types/known/structpb"
+)
+
+type recordingDispatcher struct {
+	assignment Assignment
+}
+
+func (d *recordingDispatcher) Dispatch(_ context.Context, assignment Assignment) error {
+	d.assignment = assignment
+	return nil
+}
+
+func TestRunDerivesIdentityBeforeDispatch(t *testing.T) {
+	taskContext, err := structpb.NewStruct(map[string]any{
+		"repository": "panxiao81/gitea-dynamic-runner",
+		"job":        "test",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	task := &runnerv1.Task{Id: 42, Context: taskContext}
+	dispatcher := &recordingDispatcher{}
+	scheduler := Scheduler{TrustDomain: "ddupan.top", Dispatcher: dispatcher}
+
+	if err := scheduler.Run(context.Background(), task); err != nil {
+		t.Fatal(err)
+	}
+	if dispatcher.assignment.Task != task {
+		t.Fatal("dispatcher did not receive the fetched task")
+	}
+	want := "spiffe://ddupan.top/ci/panxiao81/gitea-dynamic-runner/test"
+	if dispatcher.assignment.Identity.SPIFFEID != want {
+		t.Fatalf("SPIFFE ID = %q, want %q", dispatcher.assignment.Identity.SPIFFEID, want)
+	}
+}
+
+func TestRunRequiresDispatcher(t *testing.T) {
+	scheduler := Scheduler{TrustDomain: "ddupan.top"}
+	if err := scheduler.Run(context.Background(), &runnerv1.Task{}); err == nil {
+		t.Fatal("expected missing dispatcher to fail")
+	}
+}