feat(controller): scale runner pools from webhook demand annotation

Author: syscod3

internal/controller/impvmrunnerpool_controller.go

@@ -19,6 +19,7 @@ package controller
 import (
 	"context"
 	"fmt"
+	"strconv"
 	"time"
 
 	corev1 "k8s.io/api/core/v1"
@@ -50,6 +51,12 @@ type RunnerDriverFactory func(
 	pool *impv1alpha1.ImpVMRunnerPool,
 ) (runnerQueueDepthReader, error)
 
+const (
+	// AnnotationRunnerDemand is an optional immediate demand signal set by a webhook
+	// handler or external controller. Value is desired queued jobs as an int.
+	AnnotationRunnerDemand = "imp.dev/runner-demand"
+)
+
 // +kubebuilder:rbac:groups=imp.dev,resources=impvmrunnerpools,verbs=get;list;watch;create;update;patch;delete
 // +kubebuilder:rbac:groups=imp.dev,resources=impvmrunnerpools/status,verbs=get;update;patch
 // +kubebuilder:rbac:groups=imp.dev,resources=impvmtemplates,verbs=get;list;watch
@@ -135,6 +142,9 @@ func (r *ImpVMRunnerPoolReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 	} else if int32(queueDepth) > desiredCount {
 		desiredCount = int32(queueDepth)
 	}
+	if webhookDemand := runnerDemandFromAnnotation(pool); webhookDemand > desiredCount {
+		desiredCount = webhookDemand
+	}
 
 	toCreate := desiredCount - activeCount
 	if toCreate < 0 {
@@ -313,3 +323,20 @@ func pickSecretValue(m map[string][]byte, preferredKey string) string {
 	}
 	return ""
 }
+
+func runnerDemandFromAnnotation(pool *impv1alpha1.ImpVMRunnerPool) int32 {
+	if pool.Spec.JobDetection == nil ||
+		pool.Spec.JobDetection.Webhook == nil ||
+		!pool.Spec.JobDetection.Webhook.Enabled {
+		return 0
+	}
+	raw := pool.Annotations[AnnotationRunnerDemand]
+	if raw == "" {
+		return 0
+	}
+	n, err := strconv.Atoi(raw)
+	if err != nil || n < 0 {
+		return 0
+	}
+	return int32(n)
+}

internal/controller/impvmrunnerpool_controller_test.go

@@ -248,6 +248,53 @@ func TestRunnerPoolReconciler_scalesFromQueueDepth(t *testing.T) {
 	}
 }
 
+func TestRunnerPoolReconciler_scalesFromWebhookDemandAnnotation(t *testing.T) {
+	pool := &impv1alpha1.ImpVMRunnerPool{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "ci-pool",
+			Namespace: "ci",
+			Annotations: map[string]string{
+				AnnotationRunnerDemand: "2",
+			},
+		},
+		Spec: impv1alpha1.ImpVMRunnerPoolSpec{
+			TemplateName: "ubuntu-runner",
+			Platform: impv1alpha1.RunnerPlatformSpec{
+				Type:              "github-actions",
+				CredentialsSecret: "gh-creds",
+			},
+			Scaling: &impv1alpha1.RunnerScalingSpec{MinIdle: 0, MaxConcurrent: 5},
+			JobDetection: &impv1alpha1.RunnerJobDetectionSpec{
+				Webhook: &impv1alpha1.RunnerWebhookSpec{Enabled: true},
+			},
+		},
+	}
+	tpl := &impv1alpha1.ImpVMTemplate{
+		ObjectMeta: metav1.ObjectMeta{Name: "ubuntu-runner", Namespace: "ci"},
+		Spec:       impv1alpha1.ImpVMTemplateSpec{ClassRef: impv1alpha1.ClusterObjectRef{Name: "standard"}, Image: "ubuntu:22.04"},
+	}
+
+	scheme := newRunnerPoolTestScheme(t)
+	c := fake.NewClientBuilder().WithScheme(scheme).
+		WithObjects(pool, tpl).WithStatusSubresource(pool).Build()
+	r := &ImpVMRunnerPoolReconciler{Client: c, Scheme: scheme}
+
+	_, err := r.Reconcile(context.Background(), reconcile.Request{
+		NamespacedName: types.NamespacedName{Name: "ci-pool", Namespace: "ci"},
+	})
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	vmList := &impv1alpha1.ImpVMList{}
+	_ = c.List(context.Background(), vmList,
+		client.InNamespace("ci"),
+		client.MatchingLabels{impv1alpha1.LabelRunnerPool: "ci-pool"})
+	if len(vmList.Items) != 2 {
+		t.Fatalf("expected 2 webhook-demand VMs, got %d", len(vmList.Items))
+	}
+}
+
 type stubRunnerQueueDepthReader struct {
 	queueDepth int
 	err        error