Support changing labels (#201)

Implement proposal: https://github.com/go-gitea/gitea/issues/24540

Related:
- Protocol: https://gitea.com/gitea/actions-proto-def/pulls/9
- Gitea side: https://github.com/go-gitea/gitea/pull/24806

Co-authored-by: Jason Song <i@wolfogre.com>
Reviewed-on: https://gitea.com/gitea/act_runner/pulls/201
Reviewed-by: Jason Song <i@wolfogre.com>
Co-authored-by: sillyguodong <gedong_1994@163.com>
Co-committed-by: sillyguodong <gedong_1994@163.com>

Makefile

@@ -70,7 +70,7 @@ GO_PACKAGES_TO_VET ?= $(filter-out gitea.com/gitea/act_runner/internal/pkg/clien
 
 
 TAGS ?=
-LDFLAGS ?= -X "gitea.com/gitea/act_runner/internal/pkg/ver.version=$(RELASE_VERSION)"
+LDFLAGS ?= -X "gitea.com/gitea/act_runner/internal/pkg/ver.version=v$(RELASE_VERSION)"
 
 all: build
 

go.mod

@@ -3,7 +3,7 @@ module gitea.com/gitea/act_runner
 go 1.20
 
 require (
-	code.gitea.io/actions-proto-go v0.2.1
+	code.gitea.io/actions-proto-go v0.3.0
 	code.gitea.io/gitea-vet v0.2.3-0.20230113022436-2b1561217fa5
 	github.com/avast/retry-go/v4 v4.3.1
 	github.com/bufbuild/connect-go v1.3.1

go.sum

@@ -1,5 +1,5 @@
-code.gitea.io/actions-proto-go v0.2.1 h1:ToMN/8thz2q10TuCq8dL2d8mI+/pWpJcHCvG+TELwa0=
-code.gitea.io/actions-proto-go v0.2.1/go.mod h1:00ys5QDo1iHN1tHNvvddAcy2W/g+425hQya1cCSvq9A=
+code.gitea.io/actions-proto-go v0.3.0 h1:9Tvg8+TaaCXPKi6EnWl9vVgs2VZsj1Cs5afnsHa4AmM=
+code.gitea.io/actions-proto-go v0.3.0/go.mod h1:00ys5QDo1iHN1tHNvvddAcy2W/g+425hQya1cCSvq9A=
 code.gitea.io/gitea-vet v0.2.3-0.20230113022436-2b1561217fa5 h1:daBEK2GQeqGikJESctP5Cu1i33z5ztAD4kyQWiw185M=
 code.gitea.io/gitea-vet v0.2.3-0.20230113022436-2b1561217fa5/go.mod h1:zcNbT/aJEmivCAhfmkHOlT645KNOf9W2KnkLgFjGGfE=
 gitea.com/gitea/act v0.245.2-0.20230606002131-6ce5c93cc815 h1:u4rHwJLJnH6mej1BjEc4iubwknVeJmRVq9xQP9cAMeQ=

internal/app/cmd/daemon.go

@@ -12,6 +12,7 @@ import (
 	"strconv"
 	"strings"
 
+	"github.com/bufbuild/connect-go"
 	"github.com/mattn/go-isatty"
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
@@ -43,8 +44,13 @@ func runDaemon(ctx context.Context, configFile *string) func(cmd *cobra.Command,
 			return fmt.Errorf("failed to load registration file: %w", err)
 		}
 
+		lbls := reg.Labels
+		if len(cfg.Runner.Labels) > 0 {
+			lbls = cfg.Runner.Labels
+		}
+
 		ls := labels.Labels{}
-		for _, l := range reg.Labels {
+		for _, l := range lbls {
 			label, err := labels.Parse(l)
 			if err != nil {
 				log.WithError(err).Warnf("ignored invalid label %q", l)
@@ -71,6 +77,24 @@ func runDaemon(ctx context.Context, configFile *string) func(cmd *cobra.Command,
 		)
 
 		runner := run.NewRunner(cfg, reg, cli)
+		// declare the labels of the runner before fetching tasks
+		resp, err := runner.Declare(ctx, ls.Names())
+		if err != nil && connect.CodeOf(err) == connect.CodeUnimplemented {
+			// Gitea instance is older version. skip declare step.
+			log.Warn("Because the Gitea instance is an old version, skip declare labels and version.")
+		} else if err != nil {
+			log.WithError(err).Error("fail to invoke Declare")
+			return err
+		} else {
+			log.Infof("runner: %s, with version: %s, with labels: %v, declare successfully",
+				resp.Msg.Runner.Name, resp.Msg.Runner.Version, resp.Msg.Runner.Labels)
+			// if declare successfully, override the labels in the.runner file with valid labels in the config file (if specified)
+			reg.Labels = ls.ToStrings()
+			if err := config.SaveRegistration(cfg.Runner.File, reg); err != nil {
+				return fmt.Errorf("failed to save runner config: %w", err)
+			}
+		}
+
 		poller := poll.New(cfg, cli, runner)
 
 		poller.Poll(ctx)

internal/app/cmd/register.go

@@ -85,7 +85,7 @@ const (
 	StageInputInstance
 	StageInputToken
 	StageInputRunnerName
-	StageInputCustomLabels
+	StageInputLabels
 	StageWaitingForRegistration
 	StageExit
 )
@@ -101,7 +101,7 @@ type registerInputs struct {
 	InstanceAddr string
 	Token        string
 	RunnerName   string
-	CustomLabels []string
+	Labels       []string
 }
 
 func (r *registerInputs) validate() error {
@@ -111,8 +111,8 @@ func (r *registerInputs) validate() error {
 	if r.Token == "" {
 		return fmt.Errorf("token is empty")
 	}
-	if len(r.CustomLabels) > 0 {
-		return validateLabels(r.CustomLabels)
+	if len(r.Labels) > 0 {
+		return validateLabels(r.Labels)
 	}
 	return nil
 }
@@ -126,7 +126,7 @@ func validateLabels(ls []string) error {
 	return nil
 }
 
-func (r *registerInputs) assignToNext(stage registerStage, value string) registerStage {
+func (r *registerInputs) assignToNext(stage registerStage, value string, cfg *config.Config) registerStage {
 	// must set instance address and token.
 	// if empty, keep current stage.
 	if stage == StageInputInstance || stage == StageInputToken {
@@ -154,16 +154,33 @@ func (r *registerInputs) assignToNext(stage registerStage, value string) registe
 		return StageInputRunnerName
 	case StageInputRunnerName:
 		r.RunnerName = value
-		return StageInputCustomLabels
-	case StageInputCustomLabels:
-		r.CustomLabels = defaultLabels
+		// if there are some labels configured in config file, skip input labels stage
+		if len(cfg.Runner.Labels) > 0 {
+			ls := make([]string, 0, len(cfg.Runner.Labels))
+			for _, l := range cfg.Runner.Labels {
+				_, err := labels.Parse(l)
+				if err != nil {
+					log.WithError(err).Warnf("ignored invalid label %q", l)
+					continue
+				}
+				ls = append(ls, l)
+			}
+			if len(ls) == 0 {
+				log.Warn("no valid labels configured in config file, runner may not be able to pick up jobs")
+			}
+			r.Labels = ls
+			return StageWaitingForRegistration
+		}
+		return StageInputLabels
+	case StageInputLabels:
+		r.Labels = defaultLabels
 		if value != "" {
-			r.CustomLabels = strings.Split(value, ",")
+			r.Labels = strings.Split(value, ",")
 		}
 
-		if validateLabels(r.CustomLabels) != nil {
+		if validateLabels(r.Labels) != nil {
 			log.Infoln("Invalid labels, please input again, leave blank to use the default labels (for example, ubuntu-20.04:docker://node:16-bullseye,ubuntu-18.04:docker://node:16-buster,linux_arm:host)")
-			return StageInputCustomLabels
+			return StageInputLabels
 		}
 		return StageWaitingForRegistration
 	}
@@ -192,10 +209,10 @@ func registerInteractive(configFile string) error {
 		if err != nil {
 			return err
 		}
-		stage = inputs.assignToNext(stage, strings.TrimSpace(cmdString))
+		stage = inputs.assignToNext(stage, strings.TrimSpace(cmdString), cfg)
 
 		if stage == StageWaitingForRegistration {
-			log.Infof("Registering runner, name=%s, instance=%s, labels=%v.", inputs.RunnerName, inputs.InstanceAddr, inputs.CustomLabels)
+			log.Infof("Registering runner, name=%s, instance=%s, labels=%v.", inputs.RunnerName, inputs.InstanceAddr, inputs.Labels)
 			if err := doRegister(cfg, inputs); err != nil {
 				return fmt.Errorf("Failed to register runner: %w", err)
 			} else {
@@ -226,7 +243,7 @@ func printStageHelp(stage registerStage) {
 	case StageInputRunnerName:
 		hostname, _ := os.Hostname()
 		log.Infof("Enter the runner name (if set empty, use hostname: %s):\n", hostname)
-	case StageInputCustomLabels:
+	case StageInputLabels:
 		log.Infoln("Enter the runner labels, leave blank to use the default labels (comma-separated, for example, ubuntu-20.04:docker://node:16-bullseye,ubuntu-18.04:docker://node:16-buster,linux_arm:host):")
 	case StageWaitingForRegistration:
 		log.Infoln("Waiting for registration...")
@@ -242,12 +259,21 @@ func registerNoInteractive(configFile string, regArgs *registerArgs) error {
 		InstanceAddr: regArgs.InstanceAddr,
 		Token:        regArgs.Token,
 		RunnerName:   regArgs.RunnerName,
-		CustomLabels: defaultLabels,
+		Labels:       defaultLabels,
 	}
 	regArgs.Labels = strings.TrimSpace(regArgs.Labels)
+	// command line flag.
 	if regArgs.Labels != "" {
-		inputs.CustomLabels = strings.Split(regArgs.Labels, ",")
+		inputs.Labels = strings.Split(regArgs.Labels, ",")
 	}
+	// specify labels in config file.
+	if len(cfg.Runner.Labels) > 0 {
+		if regArgs.Labels != "" {
+			log.Warn("Labels from command will be ignored, use labels defined in config file.")
+		}
+		inputs.Labels = cfg.Runner.Labels
+	}
+
 	if inputs.RunnerName == "" {
 		inputs.RunnerName, _ = os.Hostname()
 		log.Infof("Runner name is empty, use hostname '%s'.", inputs.RunnerName)
@@ -302,7 +328,7 @@ func doRegister(cfg *config.Config, inputs *registerInputs) error {
 		Name:    inputs.RunnerName,
 		Token:   inputs.Token,
 		Address: inputs.InstanceAddr,
-		Labels:  inputs.CustomLabels,
+		Labels:  inputs.Labels,
 	}
 
 	ls := make([]string, len(reg.Labels))
@@ -314,7 +340,9 @@ func doRegister(cfg *config.Config, inputs *registerInputs) error {
 	resp, err := cli.Register(ctx, connect.NewRequest(&runnerv1.RegisterRequest{
 		Name:        reg.Name,
 		Token:       reg.Token,
-		AgentLabels: ls,
+		Version:     ver.Version(),
+		AgentLabels: ls, // Could be removed after Gitea 1.20
+		Labels:      ls,
 	}))
 	if err != nil {
 		log.WithError(err).Error("poller: cannot register new runner")

internal/app/run/runner.go

@@ -13,6 +13,7 @@ import (
 	"time"
 
 	runnerv1 "code.gitea.io/actions-proto-go/runner/v1"
+	"github.com/bufbuild/connect-go"
 	"github.com/docker/docker/api/types/container"
 	"github.com/nektos/act/pkg/artifactcache"
 	"github.com/nektos/act/pkg/common"
@@ -224,3 +225,10 @@ func parseDefaultActionsURLs(s string) []string {
 	}
 	return trimmed
 }
+
+func (r *Runner) Declare(ctx context.Context, labels []string) (*connect.Response[runnerv1.DeclareResponse], error) {
+	return r.client.Declare(ctx, connect.NewRequest(&runnerv1.DeclareRequest{
+		Version: ver.Version(),
+		Labels:  labels,
+	}))
+}

internal/pkg/client/header.go

@@ -6,5 +6,6 @@ package client
 const (
 	UUIDHeader  = "x-runner-uuid"
 	TokenHeader = "x-runner-token"
+	// Deprecated: could be removed after Gitea 1.20 released
 	VersionHeader = "x-runner-version"
 )

internal/pkg/client/http.go

@@ -39,6 +39,7 @@ func New(endpoint string, insecure bool, uuid, token, version string, opts ...co
 			if token != "" {
 				req.Header().Set(TokenHeader, token)
 			}
+			// TODO: version will be removed from request header after Gitea 1.20 released.
 			if version != "" {
 				req.Header().Set(VersionHeader, version)
 			}

internal/pkg/client/mocks/Client.go

@@ -33,6 +33,32 @@ func (_m *Client) Address() string {
 	return r0
 }
 
+// Declare provides a mock function with given fields: _a0, _a1
+func (_m *Client) Declare(_a0 context.Context, _a1 *connect.Request[runnerv1.DeclareRequest]) (*connect.Response[runnerv1.DeclareResponse], error) {
+	ret := _m.Called(_a0, _a1)
+
+	var r0 *connect.Response[runnerv1.DeclareResponse]
+	var r1 error
+	if rf, ok := ret.Get(0).(func(context.Context, *connect.Request[runnerv1.DeclareRequest]) (*connect.Response[runnerv1.DeclareResponse], error)); ok {
+		return rf(_a0, _a1)
+	}
+	if rf, ok := ret.Get(0).(func(context.Context, *connect.Request[runnerv1.DeclareRequest]) *connect.Response[runnerv1.DeclareResponse]); ok {
+		r0 = rf(_a0, _a1)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*connect.Response[runnerv1.DeclareResponse])
+		}
+	}
+
+	if rf, ok := ret.Get(1).(func(context.Context, *connect.Request[runnerv1.DeclareRequest]) error); ok {
+		r1 = rf(_a0, _a1)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // FetchTask provides a mock function with given fields: _a0, _a1
 func (_m *Client) FetchTask(_a0 context.Context, _a1 *connect.Request[runnerv1.FetchTaskRequest]) (*connect.Response[runnerv1.FetchTaskResponse], error) {
 	ret := _m.Called(_a0, _a1)

internal/pkg/config/config.example.yaml

@@ -26,6 +26,11 @@ runner:
   fetch_timeout: 5s
   # The interval for fetching the job from the Gitea instance.
   fetch_interval: 2s
+  # The labels of a runner are used to determine which jobs the runner can run, and how to run them.
+  # Like: ["macos-arm64:host", "ubuntu-latest:docker://node:16-bullseye", "ubuntu-22.04:docker://node:16-bullseye"]
+  # If it's empty when registering, it will ask for inputting labels.
+  # If it's empty when execute `deamon`, will use labels in `.runner` file.
+  labels: []
 
 cache:
   # Enable cache server to use actions/cache.

internal/pkg/config/config.go

@@ -29,6 +29,7 @@ type Runner struct {
 	Insecure      bool              `yaml:"insecure"`       // Insecure indicates whether the runner operates in an insecure mode.
 	FetchTimeout  time.Duration     `yaml:"fetch_timeout"`  // FetchTimeout specifies the timeout duration for fetching resources.
 	FetchInterval time.Duration     `yaml:"fetch_interval"` // FetchInterval specifies the interval duration for fetching resources.
+	Labels        []string          `yaml:"labels"`         // Labels specifies the labels of the runner. Labels are declared on each startup
 }
 
 // Cache represents the configuration for caching.

internal/pkg/labels/labels.go

@@ -82,3 +82,26 @@ func (l Labels) PickPlatform(runsOn []string) string {
 	// TODO: it may be not correct, what if the runner is used as host mode only?
 	return "node:16-bullseye"
 }
+
+func (l Labels) Names() []string {
+	names := make([]string, 0, len(l))
+	for _, label := range l {
+		names = append(names, label.Name)
+	}
+	return names
+}
+
+func (l Labels) ToStrings() []string {
+	ls := make([]string, 0, len(l))
+	for _, label := range l {
+		lbl := label.Name
+		if label.Schema != "" {
+			lbl += ":" + label.Schema
+			if label.Arg != "" {
+				lbl += ":" + label.Arg
+			}
+		}
+		ls = append(ls, lbl)
+	}
+	return ls
+}