Unregister stateless agents from server on termination (#2606)

Closes #2027

---------

Co-authored-by: 6543 <6543@obermui.de>

agent/rpc/client_grpc.go

@@ -366,6 +366,11 @@ func (c *client) RegisterAgent(ctx context.Context, platform, backend, version s
 	return res.GetAgentId(), err
 }
 
+func (c *client) UnregisterAgent(ctx context.Context) error {
+	_, err := c.client.UnregisterAgent(ctx, &proto.Empty{})
+	return err
+}
+
 func (c *client) ReportHealth(ctx context.Context) (err error) {
 	retry := c.newBackOff()
 	req := new(proto.ReportHealthRequest)

cmd/agent/agent.go

@@ -118,9 +118,20 @@ func run(c *cli.Context) error {
 		context.Background(),
 		metadata.Pairs("hostname", hostname),
 	)
+
+	agentConfigPersisted := abool.New()
 	ctx = utils.WithContextSigtermCallback(ctx, func() {
-		println("ctrl+c received, terminating process")
+		log.Info().Msg("Termination signal is received, shutting down")
 		sigterm.Set()
+
+		// Remove stateless agents from server
+		if agentConfigPersisted.IsNotSet() {
+			log.Debug().Msg("Unregistering agent from server")
+			err := client.UnregisterAgent(ctx)
+			if err != nil {
+				log.Err(err).Msg("Failed to unregister agent from server")
+			}
+		}
 	})
 
 	// check if grpc server version is compatible with agent
@@ -165,7 +176,11 @@ func run(c *cli.Context) error {
 		return err
 	}
 
-	writeAgentConfig(agentConfig, agentConfigPath)
+	if agentConfigPath != "" {
+		if err := writeAgentConfig(agentConfig, agentConfigPath); err == nil {
+			agentConfigPersisted.Set()
+		}
+	}
 
 	labels := map[string]string{
 		"hostname": hostname,
@@ -187,6 +202,7 @@ func run(c *cli.Context) error {
 	go func() {
 		for {
 			if sigterm.IsSet() {
+				log.Debug().Msg("Terminating health reporting")
 				return
 			}
 
@@ -210,6 +226,7 @@ func run(c *cli.Context) error {
 
 			for {
 				if sigterm.IsSet() {
+					log.Debug().Msgf("terminating runner %d", i)
 					return
 				}
 

cmd/agent/config.go

@@ -35,6 +35,10 @@ func readAgentConfig(agentConfigPath string) AgentConfig {
 		AgentID: defaultAgentIDValue,
 	}
 
+	if agentConfigPath == "" {
+		return conf
+	}
+
 	rawAgentConf, err := os.ReadFile(agentConfigPath)
 	if err != nil {
 		if os.IsNotExist(err) {
@@ -54,11 +58,11 @@ func readAgentConfig(agentConfigPath string) AgentConfig {
 	return conf
 }
 
-func writeAgentConfig(conf AgentConfig, agentConfigPath string) {
+func writeAgentConfig(conf AgentConfig, agentConfigPath string) error {
 	rawAgentConf, err := json.Marshal(conf)
 	if err != nil {
 		log.Error().Err(err).Msg("could not marshal agent config")
-		return
+		return err
 	}
 
 	// get old config
@@ -68,6 +72,9 @@ func writeAgentConfig(conf AgentConfig, agentConfigPath string) {
 	if !bytes.Equal(rawAgentConf, oldRawAgentConf) {
 		if err := os.WriteFile(agentConfigPath, rawAgentConf, 0o644); err != nil {
 			log.Error().Err(err).Msgf("could not persist agent config at '%s'", agentConfigPath)
+			return err
 		}
 	}
+
+	return nil
 }

cmd/agent/config_test.go

@@ -44,7 +44,7 @@ func TestReadAgentIDFileExists(t *testing.T) {
 
 	// update existing config and check
 	actual.AgentID = 33
-	writeAgentConfig(actual, tmpF.Name())
+	_ = writeAgentConfig(actual, tmpF.Name())
 	actual = readAgentConfig(tmpF.Name())
 	assert.EqualValues(t, 33, actual.AgentID)
 
@@ -55,7 +55,7 @@ func TestReadAgentIDFileExists(t *testing.T) {
 	defer os.Remove(tmpF2.Name())
 
 	// write new config
-	writeAgentConfig(actual, tmpF2.Name())
+	_ = writeAgentConfig(actual, tmpF2.Name())
 	actual = readAgentConfig(tmpF2.Name())
 	assert.EqualValues(t, 33, actual.AgentID)
 }

cmd/agent/flags.go

@@ -54,7 +54,7 @@ var flags = []cli.Flag{
 	&cli.StringFlag{
 		EnvVars: []string{"WOODPECKER_AGENT_CONFIG_FILE"},
 		Name:    "agent-config",
-		Usage:   "agent config file path",
+		Usage:   "agent config file path, if set empty the agent will be stateless and unregister on termination",
 		Value:   "/etc/woodpecker/agent.conf",
 	},
 	&cli.StringSliceFlag{

pipeline/backend/kubernetes/kubernetes.go

@@ -41,6 +41,10 @@ import (
 	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
 )
 
+const (
+	EngineName = "kubernetes"
+)
+
 var noContext = context.Background()
 
 type kube struct {
@@ -98,7 +102,7 @@ func New(ctx context.Context) types.Engine {
 }
 
 func (e *kube) Name() string {
-	return "kubernetes"
+	return EngineName
 }
 
 func (e *kube) IsAvailable(context.Context) bool {

pipeline/rpc/peer.go

@@ -79,6 +79,9 @@ type Peer interface {
 	// RegisterAgent register our agent to the server
 	RegisterAgent(ctx context.Context, platform, backend, version string, capacity int) (int64, error)
 
+	// UnregisterAgent unregister our agent from the server
+	UnregisterAgent(ctx context.Context) error
+
 	// ReportHealth reports health status of the agent to the server
 	ReportHealth(c context.Context) error
 }

pipeline/rpc/proto/version.go

@@ -16,4 +16,4 @@ package proto
 
 // Version is the version of the woodpecker.proto file,
 // !IMPORTANT! increased by 1 each time it get changed !IMPORTANT!
-const Version int32 = 4
+const Version int32 = 5

pipeline/rpc/proto/woodpecker.proto

@@ -24,16 +24,17 @@ package proto;
 
 // Woodpecker Server Service
 service Woodpecker {
-  rpc Version       (Empty)                returns (VersionResponse) {}
-  rpc Next          (NextRequest)          returns (NextResponse) {}
-  rpc Init          (InitRequest)          returns (Empty) {}
-  rpc Wait          (WaitRequest)          returns (Empty) {}
-  rpc Done          (DoneRequest)          returns (Empty) {}
-  rpc Extend        (ExtendRequest)        returns (Empty) {}
-  rpc Update        (UpdateRequest)        returns (Empty) {}
-  rpc Log           (LogRequest)           returns (Empty) {}
-  rpc RegisterAgent (RegisterAgentRequest) returns (RegisterAgentResponse) {}
-  rpc ReportHealth  (ReportHealthRequest)  returns (Empty) {}
+  rpc Version         (Empty)                returns (VersionResponse) {}
+  rpc Next            (NextRequest)          returns (NextResponse) {}
+  rpc Init            (InitRequest)          returns (Empty) {}
+  rpc Wait            (WaitRequest)          returns (Empty) {}
+  rpc Done            (DoneRequest)          returns (Empty) {}
+  rpc Extend          (ExtendRequest)        returns (Empty) {}
+  rpc Update          (UpdateRequest)        returns (Empty) {}
+  rpc Log             (LogRequest)           returns (Empty) {}
+  rpc RegisterAgent   (RegisterAgentRequest) returns (RegisterAgentResponse) {}
+  rpc UnregisterAgent (Empty)                returns (Empty) {}
+  rpc ReportHealth    (ReportHealthRequest)  returns (Empty) {}
 }
 
 //