feat: implement forbid/replace logic for k3s pods spawned via cron:run

plugins/scheduler-k3s/k8s.go

@@ -337,6 +337,51 @@ func (k KubernetesClient) DeleteNode(ctx context.Context, input DeleteNodeInput)
 	return k.Client.CoreV1().Nodes().Delete(ctx, input.Name, metav1.DeleteOptions{})
 }
 
+// DeletePodInput contains all the information needed to delete a Kubernetes pod
+type DeletePodInput struct {
+	// Name is the Kubernetes pod name
+	Name string
+
+	// Namespace is the Kubernetes namespace
+	Namespace string
+
+	// LabelSelector is the Kubernetes label selector
+	LabelSelector string
+}
+
+// DeletePod deletes a Kubernetes pod
+func (k KubernetesClient) DeletePod(ctx context.Context, input DeletePodInput) error {
+	if input.Name == "" && input.LabelSelector == "" {
+		return fmt.Errorf("name or label selector is required")
+	}
+
+	if input.Name != "" && input.LabelSelector != "" {
+		return fmt.Errorf("name and label selector cannot be used together")
+	}
+
+	if input.Name != "" {
+		return k.Client.CoreV1().Pods(input.Namespace).Delete(ctx, input.Name, metav1.DeleteOptions{})
+	}
+
+	// get the pods with the label selector
+	pods, err := k.ListPods(ctx, ListPodsInput{
+		Namespace:     input.Namespace,
+		LabelSelector: input.LabelSelector,
+	})
+	if err != nil {
+		return fmt.Errorf("Error listing pods: %w", err)
+	}
+
+	for _, pod := range pods {
+		err := k.Client.CoreV1().Pods(input.Namespace).Delete(ctx, pod.Name, metav1.DeleteOptions{})
+		if err != nil {
+			return fmt.Errorf("Error deleting pod %s: %w", pod.Name, err)
+		}
+	}
+
+	return nil
+}
+
 // DeleteSecretInput contains all the information needed to delete a Kubernetes secret
 type DeleteSecretInput struct {
 	// Name is the Kubernetes secret name

plugins/scheduler-k3s/triggers.go

@@ -1212,10 +1212,44 @@ func TriggerSchedulerRun(scheduler string, appName string, envCount int, args []
 		extraEnv["TRACE"] = "true"
 	}
 
+	namespace := getComputedNamespace(appName)
+	clientset, err := NewKubernetesClient()
+	if err != nil {
+		return fmt.Errorf("Error creating kubernetes client: %w", err)
+	}
+
+	if err := clientset.Ping(); err != nil {
+		return fmt.Errorf("kubernetes api not available: %w", err)
+	}
+
 	processType := "run"
 	if os.Getenv("DOKKU_CRON_ID") != "" {
 		processType = "cron"
 		labels["dokku.com/cron-id"] = os.Getenv("DOKKU_CRON_ID")
+		concurrencyPolicy := strings.ToUpper(os.Getenv("DOKKU_CONCURRENCY_POLICY"))
+		switch concurrencyPolicy {
+		case "forbid":
+			// check if there is a running pod with the same dokku.com/cron-id label
+			pods, err := clientset.ListPods(context.Background(), ListPodsInput{
+				Namespace:     namespace,
+				LabelSelector: fmt.Sprintf("dokku.com/cron-id=%s", os.Getenv("DOKKU_CRON_ID")),
+			})
+			if err != nil {
+				return fmt.Errorf("Error listing pods: %w", err)
+			}
+			if len(pods) > 0 {
+				return fmt.Errorf("There is a running pod with the same dokku.com/cron-id label")
+			}
+		case "replace":
+			// delete any existing pod with the same dokku.com/cron-id label
+			err := clientset.DeletePod(context.Background(), DeletePodInput{
+				Namespace:     namespace,
+				LabelSelector: fmt.Sprintf("dokku.com/cron-id=%s", os.Getenv("DOKKU_CRON_ID")),
+			})
+			if err != nil {
+				return fmt.Errorf("Error deleting pod: %w", err)
+			}
+		}
 	}
 
 	imageSourceType, err := common.DockerInspect(image, "{{ index .Config.Labels \"com.dokku.builder-type\" }}")
@@ -1245,7 +1279,6 @@ func TriggerSchedulerRun(scheduler string, appName string, envCount int, args []
 		entrypoint = "/exec"
 	}
 
-	namespace := getComputedNamespace(appName)
 	helmAgent, err := NewHelmAgent(namespace, DevNullPrinter)
 	if err != nil {
 		return fmt.Errorf("Error creating helm agent: %w", err)
@@ -1275,15 +1308,6 @@ func TriggerSchedulerRun(scheduler string, appName string, envCount int, args []
 
 	attachToPod := os.Getenv("DOKKU_DETACH_CONTAINER") != "1"
 
-	clientset, err := NewKubernetesClient()
-	if err != nil {
-		return fmt.Errorf("Error creating kubernetes client: %w", err)
-	}
-
-	if err := clientset.Ping(); err != nil {
-		return fmt.Errorf("kubernetes api not available: %w", err)
-	}
-
 	imagePullSecrets := getComputedImagePullSecrets(appName)
 	if imagePullSecrets == "" {
 		imagePullSecrets = fmt.Sprintf("ims-%s.%d", appName, deploymentID)