runner-controller: do not delete runner if it is busy (#103)

Currently, after refreshing the token, the controller re-creates the runner with the new token. This results in jobs being interrupted. This PR makes sure the pod is not restarted if it is busy.

Closes #74

controllers/integration_test.go

@@ -2,9 +2,10 @@ package controllers
 
 import (
 	"context"
-	"github.com/summerwind/actions-runner-controller/github/fake"
 	"time"
 
+	"github.com/summerwind/actions-runner-controller/github/fake"
+
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes/scheme"
@@ -44,7 +45,7 @@ func SetupIntegrationTest(ctx context.Context) *testEnvironment {
 		Status: 200,
 		Body:   workflowRunsFor3Replicas,
 	}
-	server := fake.NewServer(fake.WithFixedResponses(responses))
+	fakeGithubServer := fake.NewServer(fake.WithFixedResponses(responses))
 
 	BeforeEach(func() {
 		stopCh = make(chan struct{})
@@ -58,11 +59,16 @@ func SetupIntegrationTest(ctx context.Context) *testEnvironment {
 		mgr, err := ctrl.NewManager(cfg, ctrl.Options{})
 		Expect(err).NotTo(HaveOccurred(), "failed to create manager")
 
+		runnersList = fake.NewRunnersList()
+		server = runnersList.GetServer()
+		ghClient := newGithubClient(server)
+
 		replicasetController := &RunnerReplicaSetReconciler{
-			Client:   mgr.GetClient(),
-			Scheme:   scheme.Scheme,
-			Log:      logf.Log,
-			Recorder: mgr.GetEventRecorderFor("runnerreplicaset-controller"),
+			Client:       mgr.GetClient(),
+			Scheme:       scheme.Scheme,
+			Log:          logf.Log,
+			Recorder:     mgr.GetEventRecorderFor("runnerreplicaset-controller"),
+			GitHubClient: ghClient,
 		}
 		err = replicasetController.SetupWithManager(mgr)
 		Expect(err).NotTo(HaveOccurred(), "failed to setup controller")
@@ -76,7 +82,7 @@ func SetupIntegrationTest(ctx context.Context) *testEnvironment {
 		err = deploymentsController.SetupWithManager(mgr)
 		Expect(err).NotTo(HaveOccurred(), "failed to setup controller")
 
-		client := newGithubClient(server)
+		client := newGithubClient(fakeGithubServer)
 
 		autoscalerController := &HorizontalRunnerAutoscalerReconciler{
 			Client:       mgr.GetClient(),
@@ -99,7 +105,7 @@ func SetupIntegrationTest(ctx context.Context) *testEnvironment {
 	AfterEach(func() {
 		close(stopCh)
 
-		server.Close()
+		fakeGithubServer.Close()
 
 		err := k8sClient.Delete(ctx, ns)
 		Expect(err).NotTo(HaveOccurred(), "failed to delete test namespace")

controllers/runner_controller.go

@@ -203,12 +203,16 @@ func (r *RunnerReconciler) Reconcile(req ctrl.Request) (ctrl.Result, error) {
 			return ctrl.Result{}, err
 		}
 
-		if pod.Spec.Containers[0].Image != newPod.Spec.Containers[0].Image {
-			restart = true
-		}
-		if !reflect.DeepEqual(pod.Spec.Containers[0].Env, newPod.Spec.Containers[0].Env) {
+		runnerBusy, err := r.isRunnerBusy(ctx, runner.Spec.Organization, runner.Spec.Repository, runner.Name)
+		if err != nil {
+			log.Error(err, "Failed to check if runner is busy")
+			return ctrl.Result{}, nil
+		}
+
+		if !runnerBusy && (!reflect.DeepEqual(pod.Spec.Containers[0].Env, newPod.Spec.Containers[0].Env) || pod.Spec.Containers[0].Image != newPod.Spec.Containers[0].Image) {
 			restart = true
 		}
+
 		if !restart {
 			return ctrl.Result{}, err
 		}
@@ -225,6 +229,21 @@ func (r *RunnerReconciler) Reconcile(req ctrl.Request) (ctrl.Result, error) {
 	return ctrl.Result{}, nil
 }
 
+func (r *RunnerReconciler) isRunnerBusy(ctx context.Context, org, repo, name string) (bool, error) {
+	runners, err := r.GitHubClient.ListRunners(ctx, org, repo)
+	if err != nil {
+		return false, err
+	}
+
+	for _, runner := range runners {
+		if runner.GetName() == name {
+			return runner.GetBusy(), nil
+		}
+	}
+
+	return false, fmt.Errorf("runner not found")
+}
+
 func (r *RunnerReconciler) unregisterRunner(ctx context.Context, org, repo, name string) (bool, error) {
 	runners, err := r.GitHubClient.ListRunners(ctx, org, repo)
 	if err != nil {
@@ -234,6 +253,9 @@ func (r *RunnerReconciler) unregisterRunner(ctx context.Context, org, repo, name
 	id := int64(0)
 	for _, runner := range runners {
 		if runner.GetName() == name {
+			if runner.GetBusy() {
+				return false, fmt.Errorf("runner is busy")
+			}
 			id = runner.GetID()
 			break
 		}

controllers/runnerreplicaset_controller.go

@@ -31,14 +31,16 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/summerwind/actions-runner-controller/api/v1alpha1"
+	"github.com/summerwind/actions-runner-controller/github"
 )
 
 // RunnerReplicaSetReconciler reconciles a Runner object
 type RunnerReplicaSetReconciler struct {
 	client.Client
-	Log      logr.Logger
-	Recorder record.EventRecorder
-	Scheme   *runtime.Scheme
+	Log          logr.Logger
+	Recorder     record.EventRecorder
+	Scheme       *runtime.Scheme
+	GitHubClient *github.Client
 }
 
 // +kubebuilder:rbac:groups=actions.summerwind.dev,resources=runnerreplicasets,verbs=get;list;watch;create;update;patch;delete
@@ -96,8 +98,25 @@ func (r *RunnerReplicaSetReconciler) Reconcile(req ctrl.Request) (ctrl.Result, e
 	if available > desired {
 		n := available - desired
 
+		// get runners that are currently not busy
+		var notBusy []v1alpha1.Runner
+		for _, runner := range myRunners {
+			busy, err := r.isRunnerBusy(ctx, runner.Spec.Organization, runner.Spec.Repository, runner.Name)
+			if err != nil {
+				log.Error(err, "Failed to check if runner is busy")
+				return ctrl.Result{}, err
+			}
+			if !busy {
+				notBusy = append(notBusy, runner)
+			}
+		}
+
+		if len(notBusy) < n {
+			n = len(notBusy)
+		}
+
 		for i := 0; i < n; i++ {
-			if err := r.Client.Delete(ctx, &myRunners[i]); err != nil {
+			if err := r.Client.Delete(ctx, &notBusy[i]); err != nil {
 				log.Error(err, "Failed to delete runner resource")
 
 				return ctrl.Result{}, err
@@ -166,3 +185,19 @@ func (r *RunnerReplicaSetReconciler) SetupWithManager(mgr ctrl.Manager) error {
 		Owns(&v1alpha1.Runner{}).
 		Complete(r)
 }
+
+func (r *RunnerReplicaSetReconciler) isRunnerBusy(ctx context.Context, org, repo, name string) (bool, error) {
+	runners, err := r.GitHubClient.ListRunners(ctx, org, repo)
+	r.Log.Info("runners", "github", runners)
+	if err != nil {
+		return false, err
+	}
+
+	for _, runner := range runners {
+		if runner.GetName() == name {
+			return runner.GetBusy(), nil
+		}
+	}
+
+	return false, fmt.Errorf("runner not found")
+}

controllers/runnerreplicaset_controller_test.go

@@ -3,11 +3,14 @@ package controllers
 import (
 	"context"
 	"math/rand"
+	"net/http/httptest"
 	"time"
 
+	"github.com/google/go-github/v32/github"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/utils/pointer"
 	ctrl "sigs.k8s.io/controller-runtime"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 
@@ -17,6 +20,12 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	actionsv1alpha1 "github.com/summerwind/actions-runner-controller/api/v1alpha1"
+	"github.com/summerwind/actions-runner-controller/github/fake"
+)
+
+var (
+	runnersList *fake.RunnersList
+	server      *httptest.Server
 )
 
 // SetupTest will set up a testing environment.
@@ -41,11 +50,16 @@ func SetupTest(ctx context.Context) *corev1.Namespace {
 		mgr, err := ctrl.NewManager(cfg, ctrl.Options{})
 		Expect(err).NotTo(HaveOccurred(), "failed to create manager")
 
+		runnersList = fake.NewRunnersList()
+		server = runnersList.GetServer()
+		ghClient := newGithubClient(server)
+
 		controller := &RunnerReplicaSetReconciler{
-			Client:   mgr.GetClient(),
-			Scheme:   scheme.Scheme,
-			Log:      logf.Log,
-			Recorder: mgr.GetEventRecorderFor("runnerreplicaset-controller"),
+			Client:       mgr.GetClient(),
+			Scheme:       scheme.Scheme,
+			Log:          logf.Log,
+			Recorder:     mgr.GetEventRecorderFor("runnerreplicaset-controller"),
+			GitHubClient: ghClient,
 		}
 		err = controller.SetupWithManager(mgr)
 		Expect(err).NotTo(HaveOccurred(), "failed to setup controller")
@@ -61,6 +75,7 @@ func SetupTest(ctx context.Context) *corev1.Namespace {
 	AfterEach(func() {
 		close(stopCh)
 
+		server.Close()
 		err := k8sClient.Delete(ctx, ns)
 		Expect(err).NotTo(HaveOccurred(), "failed to delete test namespace")
 	})
@@ -124,6 +139,16 @@ var _ = Context("Inside of a new namespace", func() {
 							logf.Log.Error(err, "list runners")
 						}
 
+						for i, runner := range runners.Items {
+							runnersList.Add(&github.Runner{
+								ID:     pointer.Int64Ptr(int64(i) + 1),
+								Name:   pointer.StringPtr(runner.Name),
+								OS:     pointer.StringPtr("linux"),
+								Status: pointer.StringPtr("online"),
+								Busy:   pointer.BoolPtr(false),
+							})
+						}
+
 						return len(runners.Items)
 					},
 					time.Second*5, time.Millisecond*500).Should(BeEquivalentTo(1))
@@ -155,6 +180,16 @@ var _ = Context("Inside of a new namespace", func() {
 							logf.Log.Error(err, "list runners")
 						}
 
+						for i, runner := range runners.Items {
+							runnersList.Add(&github.Runner{
+								ID:     pointer.Int64Ptr(int64(i) + 1),
+								Name:   pointer.StringPtr(runner.Name),
+								OS:     pointer.StringPtr("linux"),
+								Status: pointer.StringPtr("online"),
+								Busy:   pointer.BoolPtr(false),
+							})
+						}
+
 						return len(runners.Items)
 					},
 					time.Second*5, time.Millisecond*500).Should(BeEquivalentTo(2))
@@ -186,6 +221,16 @@ var _ = Context("Inside of a new namespace", func() {
 							logf.Log.Error(err, "list runners")
 						}
 
+						for i, runner := range runners.Items {
+							runnersList.Add(&github.Runner{
+								ID:     pointer.Int64Ptr(int64(i) + 1),
+								Name:   pointer.StringPtr(runner.Name),
+								OS:     pointer.StringPtr("linux"),
+								Status: pointer.StringPtr("online"),
+								Busy:   pointer.BoolPtr(false),
+							})
+						}
+
 						return len(runners.Items)
 					},
 					time.Second*5, time.Millisecond*500).Should(BeEquivalentTo(0))