Merge pull request #312 from uselagoon/fix-queue-cancellations

refactor: build qos and cache logic

Author: shreddedbacon

api/lagoon/v1beta2/lagoonbuild_helpers.go

@@ -115,7 +115,7 @@ func TestSortQueuedBuilds(t *testing.T) {
 	}
 }
 
-func TestSortQueuedNamespaceBuilds(t *testing.T) {
+func TestSortQueuedNamespaceBuildsByCreation(t *testing.T) {
 	type args struct {
 		namespace     string
 		pendingBuilds []string
@@ -127,7 +127,7 @@ func TestSortQueuedNamespaceBuilds(t *testing.T) {
 		wantErr bool
 	}{
 		{
-			name: "test1 - namespace1 builds only sorted by priority then creation",
+			name: "test1 - namespace1 builds only sorted by creation",
 			args: args{
 				namespace: "namespace1",
 				pendingBuilds: []string{
@@ -154,13 +154,13 @@ func TestSortQueuedNamespaceBuilds(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got, err := SortQueuedNamespaceBuilds(tt.args.namespace, tt.args.pendingBuilds)
+			got, err := SortQueuedNamespaceBuildsByCreation(tt.args.namespace, tt.args.pendingBuilds)
 			if (err != nil) != tt.wantErr {
-				t.Errorf("SortQueuedNamespaceBuilds() error = %v, wantErr %v", err, tt.wantErr)
+				t.Errorf("SortQueuedNamespaceBuildsByCreation() error = %v, wantErr %v", err, tt.wantErr)
 				return
 			}
 			if !reflect.DeepEqual(got, tt.want) {
-				t.Errorf("SortQueuedNamespaceBuilds() = %v, want %v", got, tt.want)
+				t.Errorf("SortQueuedNamespaceBuildsByCreation() = %v, want %v", got, tt.want)
 			}
 		})
 	}

api/lagoon/v1beta2/lagoonbuild_helpers_test.go

@@ -18,6 +18,7 @@ package v1beta2
 import (
 	"encoding/json"
 	"strings"
+	"time"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
@@ -182,6 +183,7 @@ type CachedBuildItem struct {
 	Step              string `json:"step"`
 	DockerBuild       bool   `json:"dockerBuild"`
 	CreationTimestamp int64  `json:"creationTimestamp"`
+	UpdatedTimestamp  int64  `json:"updatedTimestamp"`
 }
 
 func (q *CachedBuildItem) String() string {
@@ -203,6 +205,7 @@ func NewCachedBuildItem(lagoonBuild LagoonBuild, status string, dockerBuild bool
 		Step:              lagoonBuild.Labels["lagoon.sh/buildStep"],
 		DockerBuild:       dockerBuild,
 		CreationTimestamp: lagoonBuild.CreationTimestamp.Unix(),
+		UpdatedTimestamp:  time.Now().UTC().Unix(),
 	}
 }
 
@@ -213,6 +216,7 @@ type CachedBuildQueueItem struct {
 	Position          int    `json:"position"`
 	Length            int    `json:"length"`
 	CreationTimestamp int64  `json:"creationTimestamp"`
+	UpdatedTimestamp  int64  `json:"updatedTimestamp"`
 }
 
 func (q *CachedBuildQueueItem) String() string {
@@ -234,5 +238,21 @@ func NewCachedBuildQueueItem(lagoonBuild LagoonBuild, priority, position, length
 		Position:          position,
 		Length:            length,
 		CreationTimestamp: lagoonBuild.CreationTimestamp.Unix(),
+		UpdatedTimestamp:  time.Now().UTC().Unix(),
 	}
 }
+
+const (
+	BuildFinalizer = "finalizer.lagoonbuild.crd.lagoon.sh/v1beta2"
+
+	// NotOwnedByControllerMessage is used to describe an error where the controller was unable to start the build because
+	// the `lagoon.sh/controller` label does not match this controllers name
+	NotOwnedByControllerMessage = `Build was cancelled due to an issue with the build controller.
+This issue is related to the deployment system, not the repository or code base changes.
+Contact your Lagoon support team for help`
+	// MissingLabelsMessage is used to describe an error where the controller was unable to start the build because
+	// the `lagoon.sh/controller` label is missing
+	MissingLabelsMessage = `"Build was cancelled due to namespace configuration issue. A label or labels are missing on the namespace.
+This issue is related to the deployment system, not the repository or code base changes.
+Contact your Lagoon support team for help`
+)

api/lagoon/v1beta2/lagoonbuild_types.go

@@ -8,11 +8,13 @@ import (
 	"time"
 
 	"github.com/go-logr/logr"
+	lru "github.com/hashicorp/golang-lru/v2"
 	"github.com/uselagoon/machinery/api/schema"
 	"github.com/uselagoon/remote-controller/internal/helpers"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
+	runtime "k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/selection"
 	"k8s.io/apimachinery/pkg/types"
 	ctrl "sigs.k8s.io/controller-runtime"
@@ -240,10 +242,8 @@ func updateLagoonTask(namespace string, taskSpec LagoonTaskSpec) ([]byte, error)
 }
 
 // CancelTask handles cancelling tasks or handling if a tasks no longer exists.
-func CancelTask(ctx context.Context, cl client.Client, namespace string, body []byte) (bool, []byte, error) {
+func CancelTask(ctx context.Context, cl client.Client, namespace string, jobSpec *LagoonTaskSpec) (bool, []byte, error) {
 	opLog := ctrl.Log.WithName("handlers").WithName("LagoonTasks")
-	jobSpec := &LagoonTaskSpec{}
-	_ = json.Unmarshal(body, jobSpec)
 	var jobPod corev1.Pod
 	// @TODO: use `taskName` in the future only
 	taskName := fmt.Sprintf("lagoon-task-%s-%s", jobSpec.Task.ID, helpers.HashString(jobSpec.Task.ID)[0:6])
@@ -345,3 +345,46 @@ func SortQueuedNamespaceTasks(namespace string, pendingTasks []string) ([]Cached
 	})
 	return tasks, nil
 }
+
+func SeedTaskStartup(cl client.Client, scheme *runtime.Scheme, controllerNamespace string,
+	tasksCache *lru.Cache[string, string], tasksQueueCache *lru.Cache[string, string],
+) error {
+	runningTasks := &LagoonTaskList{}
+	listOption := (&client.ListOptions{}).ApplyOptions([]client.ListOption{
+		client.MatchingLabels(map[string]string{
+			"lagoon.sh/controller": controllerNamespace, // created by this controller
+			"lagoon.sh/taskStatus": BuildStatusRunning.String(),
+		}),
+	})
+	if err := cl.List(context.Background(), runningTasks, listOption); err != nil {
+		return fmt.Errorf("unable to list running LagoonTasks, there may be none or something went wrong: %v", err)
+	}
+	for _, build := range runningTasks.Items {
+		bc := NewCachedTaskItem(build, "Running")
+		tasksCache.Add(build.Name, bc.String())
+	}
+	pendingTasks := &LagoonTaskList{}
+	listOption = (&client.ListOptions{}).ApplyOptions([]client.ListOption{
+		client.MatchingLabels(map[string]string{
+			"lagoon.sh/controller": controllerNamespace, // created by this controller
+			"lagoon.sh/taskStatus": BuildStatusPending.String(),
+		}),
+	})
+	if err := cl.List(context.Background(), pendingTasks, listOption); err != nil {
+		return fmt.Errorf("unable to list pending LagoonTasks, there may be none or something went wrong: %v", err)
+	}
+	sortTasks(pendingTasks)
+	position := 1
+	for _, build := range pendingTasks.Items {
+		bc := NewCachedTaskQueueItem(build, position, len(pendingTasks.Items))
+		tasksQueueCache.Add(build.Name, bc.String())
+	}
+	return nil
+}
+
+func sortTasks(pendingTasks *LagoonTaskList) {
+	sort.Slice(pendingTasks.Items, func(i, j int) bool {
+		// sort by creation timestamp
+		return pendingTasks.Items[i].CreationTimestamp.Before(&pendingTasks.Items[j].CreationTimestamp)
+	})
+}

api/lagoon/v1beta2/lagoontask_helpers.go

@@ -253,3 +253,7 @@ func NewCachedTaskQueueItem(lagoonTask LagoonTask, position, length int) CachedT
 		CreationTimestamp: lagoonTask.CreationTimestamp.Unix(),
 	}
 }
+
+const (
+	TaskFinalizer = "finalizer.lagoontask.crd.lagoon.sh/v1beta2"
+)

api/lagoon/v1beta2/lagoontask_types.go

@@ -20,6 +20,7 @@ import (
 	"crypto/tls"
 	"flag"
 	"fmt"
+	_ "net/http/pprof"
 	"net/url"
 	"os"
 	"strings"
@@ -32,6 +33,7 @@ import (
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
 
 	"github.com/uselagoon/remote-controller/internal/dockerhost"
@@ -174,7 +176,7 @@ func main() {
 	var pruneLongRunningPodsCron string
 
 	var lffQoSEnabled bool
-	var qosMaxBuilds int
+	var qosTotalBuilds int
 	var qosMaxContainerBuilds int
 	var qosDefaultPriority int
 
@@ -394,11 +396,11 @@ func main() {
 	flag.IntVar(&timeoutForLongRunningTaskPods, "timeout-longrunning-task-pod-cleanup", 6, "How many hours a task pod should run before forcefully closed.")
 
 	// Build QoS configuration
-	flag.BoolVar(&lffQoSEnabled, "enable-qos", false, "Flag to enable this controller with QoS for builds.")
+	flag.BoolVar(&lffQoSEnabled, "enable-qos", true, "Deprecated: Flag to enable this controller with QoS for builds. No longer configurable")
 	// this flag remains the same, the number of max builds flag remains unchanged to be backwards compatible
 	flag.IntVar(&qosMaxContainerBuilds, "qos-max-builds", 20, "The total number of builds during the container build phase that can run at any one time.")
 	// this new flag is added but defaults to 0, if it is greater than `qos-max-builds` then it will be used, otherwise it will default to the value of `qos-max-builds`
-	flag.IntVar(&qosMaxBuilds, "qos-total-builds", 0, "The total number of builds that can run at any one time. Defaults to qos-max-builds if not provided or less than qos-max-builds.")
+	flag.IntVar(&qosTotalBuilds, "qos-total-builds", 0, "The total number of builds that can run at any one time. Defaults to qos-max-builds if not provided or less than qos-max-builds.")
 	flag.IntVar(&qosDefaultPriority, "qos-default", 5, "The default qos priority value to apply if one is not provided.")
 
 	// Task QoS configuration
@@ -565,7 +567,7 @@ func main() {
 		os.Exit(1)
 	}
 
-	cacheSize := helpers.GetEnvInt("CACHE_SIZE", 1000)
+	cacheSize := helpers.GetEnvInt("CACHE_SIZE", 2000)
 	// create the cancellation cache
 	cache := expirable.NewLRU[string, string](cacheSize, nil, time.Minute*60)
 	// create queue cache
@@ -744,8 +746,12 @@ func main() {
 		TLSSkipVerify:         tlsSkipVerify,
 	}
 
+	// setup harbor config
+	var lagoonHarbor *harbor.Harbor
+	lagoonHarbor, _ = harbor.New(harborConfig)
+
 	deletion := deletions.New(mgr.GetClient(),
-		harborConfig,
+		lagoonHarbor,
 		deletions.DeleteConfig{
 			PVCRetryAttempts: pvcRetryAttempts,
 			PVCRetryInterval: pvcRetryInterval,
@@ -756,6 +762,7 @@ func main() {
 
 	messaging := messenger.New(config,
 		mgr.GetClient(),
+		mgr.GetAPIReader(),
 		startupConnectionAttempts,
 		startupConnectionInterval,
 		controllerNamespace,
@@ -767,7 +774,11 @@ func main() {
 		enableDebug,
 		lffSupportK8UPv2,
 		cache,
-		harborConfig,
+		lagoonHarbor,
+		lagoonTargetName,
+		buildsCache,
+		buildsQueueCache,
+		qosDefaultPriority,
 	)
 
 	reuseCache, _ := lru.New[string, string](cacheSize)
@@ -790,19 +801,13 @@ func main() {
 		buildCache,
 	)
 	c := cron.New()
-	// if we are running with MQ support, then start the consumer handler
-
-	if enableMQ {
-		setupLog.Info("starting messaging handler")
-		go messaging.Consumer(lagoonTargetName)
-	}
 
 	// this ensures that the max number of builds is not less than the container builds support
-	if qosMaxBuilds < qosMaxContainerBuilds {
-		qosMaxBuilds = qosMaxContainerBuilds
+	if qosTotalBuilds < qosMaxContainerBuilds {
+		qosTotalBuilds = qosMaxContainerBuilds
 	}
 	buildQoSConfigv1beta2 := lagoonv1beta2ctrl.BuildQoS{
-		MaxBuilds:          qosMaxBuilds,
+		TotalBuilds:        qosTotalBuilds,
 		MaxContainerBuilds: qosMaxContainerBuilds,
 		DefaultPriority:    qosDefaultPriority,
 	}
@@ -828,6 +833,7 @@ func main() {
 	}
 
 	resourceCleanup := pruner.New(mgr.GetClient(),
+		mgr.GetAPIReader(),
 		buildsToKeep,
 		buildPodsToKeep,
 		tasksToKeep,
@@ -896,7 +902,6 @@ func main() {
 		// use cron to run a task pod cleanup task
 		// this will check any Lagoon task pods and attempt to delete them
 		_, err := c.AddFunc(harborCredentialCron, func() {
-			lagoonHarbor, _ := harbor.New(harborConfig)
 			lagoonHarbor.RotateRobotCredentials(context.Background(), mgr.GetClient())
 		})
 		if err != nil {
@@ -930,12 +935,24 @@ func main() {
 
 	c.Start()
 
-	// @TODO: maybe insert a pre-controller start state collector to try and seed the queue/build caches before the controllers start
+	// create a temporary client to use in seed functions
+	tmpClient, _ := client.New(ctrl.GetConfigOrDie(), client.Options{
+		Scheme: scheme,
+	})
+	// pre-seed the queues with the current state of builds
+	if err := lagoonv1beta2.SeedBuildStartup(tmpClient, scheme, controllerNamespace, qosDefaultPriority, buildsCache, buildsQueueCache); err != nil {
+		setupLog.Error(err, "unable to seed controller startup state")
+	}
+	// pre-seed the queues with the current state of tasks
+	if err := lagoonv1beta2.SeedTaskStartup(tmpClient, scheme, controllerNamespace, tasksCache, tasksQueueCache); err != nil {
+		setupLog.Error(err, "unable to seed controller startup state")
+	}
 
 	setupLog.Info("starting build controller")
 	// v1beta2 is the latest version
 	if err = (&lagoonv1beta2ctrl.LagoonBuildReconciler{
 		Client:                mgr.GetClient(),
+		APIReader:             mgr.GetAPIReader(),
 		Log:                   ctrl.Log.WithName("v1beta2").WithName("LagoonBuild"),
 		Scheme:                mgr.GetScheme(),
 		EnableMQ:              enableMQ,
@@ -973,8 +990,7 @@ func main() {
 		LFFBackupWeeklyRandom:            lffBackupWeeklyRandom,
 		LFFRouterURL:                     lffRouterURL,
 		LFFHarborEnabled:                 lffHarborEnabled,
-		Harbor:                           harborConfig,
-		LFFQoSEnabled:                    lffQoSEnabled,
+		Harbor:                           lagoonHarbor,
 		BuildQoS:                         buildQoSConfigv1beta2,
 		NativeCronPodMinFrequency:        nativeCronPodMinFrequency,
 		LagoonTargetName:                 lagoonTargetName,
@@ -1038,6 +1054,7 @@ func main() {
 	setupLog.Info("starting build pod monitor controller")
 	if err = (&lagoonv1beta2ctrl.BuildMonitorReconciler{
 		Client:                mgr.GetClient(),
+		APIReader:             mgr.GetAPIReader(),
 		Log:                   ctrl.Log.WithName("v1beta2").WithName("LagoonBuildPodMonitor"),
 		Scheme:                mgr.GetScheme(),
 		EnableMQ:              enableMQ,
@@ -1047,7 +1064,6 @@ func main() {
 		RandomNamespacePrefix: randomPrefix,
 		EnableDebug:           enableDebug,
 		LagoonTargetName:      lagoonTargetName,
-		LFFQoSEnabled:         lffQoSEnabled,
 		BuildQoS:              buildQoSConfigv1beta2,
 		Cache:                 cache,
 		DockerHost:            dockerhosts,
@@ -1081,18 +1097,25 @@ func main() {
 	if lffHarborEnabled {
 		if err = (&harborctrl.HarborCredentialReconciler{
 			Client:              mgr.GetClient(),
+			APIReader:           mgr.GetAPIReader(),
 			Log:                 ctrl.Log.WithName("harbor").WithName("HarborCredentialReconciler"),
 			Scheme:              mgr.GetScheme(),
 			LFFHarborEnabled:    lffHarborEnabled,
 			ControllerNamespace: controllerNamespace,
-			Harbor:              harborConfig,
+			Harbor:              lagoonHarbor,
 		}).SetupWithManager(mgr); err != nil {
 			setupLog.Error(err, "unable to create controller", "controller", "HarborCredentialReconciler")
 			os.Exit(1)
 		}
 	}
 	// +kubebuilder:scaffold:builder
 
+	// if we are running with MQ support, then start the consumer handler
+	if enableMQ {
+		setupLog.Info("starting messaging handler")
+		go messaging.Consumer(lagoonTargetName)
+	}
+
 	setupLog.Info("starting manager")
 	if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {
 		setupLog.Error(err, "problem running manager")

api/lagoon/v1beta2/zz_generated.deepcopy.go

@@ -15,15 +15,13 @@ spec:
         - "--leader-elect"
         - "--build-pod-cleanup-cron=*/1 * * * *"
         - "--task-pod-cleanup-cron=*/1 * * * *"
-        - "--harbor-credential-cron=*/1 * * * *"
+        - "--harbor-credential-cron=*/2 * * * *"
         - "--harbor-robot-account-expiry=1d"
         - "--enable-harbor"
         - "--harbor-enable-project-webhook"
         - "--enable-debug"
-        - "--enable-qos"
         - "--qos-total-builds=5"
         - "--qos-max-builds=3"
-        - "--enable-deprecated-apis"
         - "--lagoon-feature-flag-support-k8upv2"
         - "--skip-tls-verify"
         - "--enable-task-qos"