feat: scheduler-k3s renders pvc-backed volumes from storage attachments

Adds a per-entry storage helm chart that owns the PVC and (when host_path is set) the PV, and three triggers for the storage plugin to drive: storage-create installs/upgrades the chart with storage-class existence validation, storage-destroy uninstalls it, storage-status reports the PVC phase. App deployment and cron-job charts now consume storage-app-mounts and reference each PVC by name; the // todo: implement volumes blocks are gone. ProcessVolume gained sub_path, read_only, and persistent_claim fields and the deployment/cron-job templates render them.

plugins/scheduler-k3s/Makefile

@@ -1,5 +1,5 @@
 SUBCOMMANDS = subcommands/annotations:set subcommands/autoscaling-auth:set subcommands/autoscaling-auth:report subcommands/cluster:add subcommands/cluster:list subcommands/cluster:remove subcommands/ensure-charts subcommands/initialize subcommands/labels:set subcommands/profiles:add subcommands/profiles:list subcommands/profiles:remove subcommands/report subcommands/set subcommands/show-kubeconfig subcommands/uninstall
-TRIGGERS = triggers/core-post-deploy triggers/core-post-extract triggers/install triggers/post-app-clone-setup triggers/post-app-rename-setup triggers/post-certs-update triggers/post-certs-remove triggers/post-create triggers/post-delete triggers/report triggers/scheduler-app-status triggers/scheduler-deploy triggers/scheduler-enter triggers/scheduler-is-deployed triggers/scheduler-logs triggers/scheduler-proxy-config triggers/scheduler-proxy-logs triggers/scheduler-post-delete triggers/scheduler-run triggers/scheduler-run-list triggers/scheduler-stop triggers/scheduler-cron-write
+TRIGGERS = triggers/core-post-deploy triggers/core-post-extract triggers/install triggers/post-app-clone-setup triggers/post-app-rename-setup triggers/post-certs-update triggers/post-certs-remove triggers/post-create triggers/post-delete triggers/report triggers/scheduler-app-status triggers/scheduler-deploy triggers/scheduler-enter triggers/scheduler-is-deployed triggers/scheduler-logs triggers/scheduler-proxy-config triggers/scheduler-proxy-logs triggers/scheduler-post-delete triggers/scheduler-run triggers/scheduler-run-list triggers/scheduler-stop triggers/scheduler-cron-write triggers/storage-create triggers/storage-destroy triggers/storage-status
 BUILD = commands subcommands triggers
 PLUGIN_NAME = scheduler-k3s
 

plugins/scheduler-k3s/go.mod

@@ -13,6 +13,7 @@ require (
 	github.com/dokku/dokku/plugins/logs v0.0.0-00010101000000-000000000000
 	github.com/dokku/dokku/plugins/nginx-vhosts v0.0.0-00010101000000-000000000000
 	github.com/dokku/dokku/plugins/registry v0.0.0-00010101000000-000000000000
+	github.com/dokku/dokku/plugins/storage v0.0.0-00010101000000-000000000000
 	github.com/fatih/color v1.19.0
 	github.com/fluxcd/pkg/kustomize v1.31.0
 	github.com/go-openapi/jsonpointer v0.23.1
@@ -193,6 +194,8 @@ replace github.com/dokku/dokku/plugins/nginx-vhosts => ../nginx-vhosts
 
 replace github.com/dokku/dokku/plugins/registry => ../registry
 
+replace github.com/dokku/dokku/plugins/storage => ../storage
+
 replace github.com/joho/godotenv => github.com/joho/godotenv v1.2.0
 
 replace sigs.k8s.io/controller-runtime => sigs.k8s.io/controller-runtime v0.22.4

plugins/scheduler-k3s/src/triggers/triggers.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	"fmt"
 	"os"
 	"strconv"
@@ -158,6 +159,15 @@ func main() {
 		appName := flag.Arg(1)
 		format := flag.Arg(2)
 		err = scheduler_k3s.TriggerSchedulerRunList(scheduler, appName, format)
+	case "storage-create":
+		entryName := flag.Arg(0)
+		err = scheduler_k3s.TriggerStorageCreate(context.Background(), entryName)
+	case "storage-destroy":
+		entryName := flag.Arg(0)
+		err = scheduler_k3s.TriggerStorageDestroy(context.Background(), entryName)
+	case "storage-status":
+		entryName := flag.Arg(0)
+		err = scheduler_k3s.TriggerStorageStatus(context.Background(), entryName)
 	default:
 		err = fmt.Errorf("Invalid plugin trigger call: %s", trigger)
 	}

plugins/scheduler-k3s/storage.go

@@ -0,0 +1,313 @@
+package scheduler_k3s
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"strings"
+
+	"github.com/dokku/dokku/plugins/common"
+	storage "github.com/dokku/dokku/plugins/storage"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+// StorageChartValues are the helm chart inputs for the per-entry storage chart.
+type StorageChartValues struct {
+	Global StorageChartGlobal `yaml:"global"`
+}
+
+// StorageChartGlobal contains global values for the storage chart.
+type StorageChartGlobal struct {
+	EntryName     string            `yaml:"entry_name"`
+	Namespace     string            `yaml:"namespace"`
+	Size          string            `yaml:"size"`
+	AccessMode    string            `yaml:"access_mode,omitempty"`
+	StorageClass  string            `yaml:"storage_class,omitempty"`
+	HostPath      string            `yaml:"host_path,omitempty"`
+	ReclaimPolicy string            `yaml:"reclaim_policy,omitempty"`
+	Annotations   map[string]string `yaml:"annotations,omitempty"`
+	Labels        map[string]string `yaml:"labels,omitempty"`
+}
+
+// GetStorageReleaseName returns the helm release name for a storage entry.
+func GetStorageReleaseName(entryName string) string {
+	return fmt.Sprintf("storage-%s", entryName)
+}
+
+// readEntryFromStdin decodes a storage.Entry from stdin. The storage
+// plugin pipes the JSON-encoded entry to scheduler triggers.
+func readEntryFromStdin() (*storage.Entry, error) {
+	raw, err := io.ReadAll(os.Stdin)
+	if err != nil {
+		return nil, fmt.Errorf("unable to read storage entry from stdin: %w", err)
+	}
+	if len(raw) == 0 {
+		return nil, fmt.Errorf("no storage entry payload received on stdin")
+	}
+	entry := &storage.Entry{}
+	if err := json.Unmarshal(raw, entry); err != nil {
+		return nil, fmt.Errorf("unable to parse storage entry payload: %w", err)
+	}
+	return entry, nil
+}
+
+// TriggerStorageCreate provisions or updates the PVC/PV for a k3s storage
+// entry. Idempotent: re-runs against the same entry are a helm upgrade.
+func TriggerStorageCreate(ctx context.Context, entryName string) error {
+	if err := isKubernetesAvailable(); err != nil {
+		return fmt.Errorf("kubernetes not available: %w", err)
+	}
+
+	entry, err := readEntryFromStdin()
+	if err != nil {
+		return err
+	}
+	if entry.Name != entryName {
+		return fmt.Errorf("entry name mismatch: stdin=%q, arg=%q", entry.Name, entryName)
+	}
+	if entry.Scheduler != storage.SchedulerK3s {
+		return nil
+	}
+
+	if err := validateStorageClass(ctx, entry.StorageClass); err != nil {
+		return err
+	}
+
+	chartDir, err := os.MkdirTemp("", "dokku-storage-chart-")
+	if err != nil {
+		return fmt.Errorf("error creating chart directory: %w", err)
+	}
+	defer os.RemoveAll(chartDir)
+
+	releaseName := GetStorageReleaseName(entry.Name)
+	namespace := entry.Namespace
+	if namespace == "" {
+		namespace = "default"
+	}
+
+	if err := os.MkdirAll(filepath.Join(chartDir, "templates"), 0755); err != nil {
+		return fmt.Errorf("error creating chart templates directory: %w", err)
+	}
+
+	chart := &Chart{
+		ApiVersion: "v2",
+		AppVersion: "1.0.0",
+		Name:       releaseName,
+		Icon:       "https://dokku.com/assets/dokku-logo.svg",
+		Version:    "0.0.1",
+	}
+	if err := writeYaml(WriteYamlInput{Object: chart, Path: filepath.Join(chartDir, "Chart.yaml")}); err != nil {
+		return fmt.Errorf("error writing chart: %w", err)
+	}
+
+	for _, name := range []string{"persistent-volume-claim.yaml", "persistent-volume.yaml"} {
+		b, err := templates.ReadFile(filepath.Join("templates", "storage-chart", "templates", name))
+		if err != nil {
+			return fmt.Errorf("error reading %s: %w", name, err)
+		}
+		if err := os.WriteFile(filepath.Join(chartDir, "templates", name), b, 0644); err != nil {
+			return fmt.Errorf("error writing %s: %w", name, err)
+		}
+	}
+
+	values := &StorageChartValues{
+		Global: StorageChartGlobal{
+			EntryName:     entry.Name,
+			Namespace:     namespace,
+			Size:          entry.Size,
+			AccessMode:    entry.AccessMode,
+			StorageClass:  entry.StorageClass,
+			HostPath:      entry.HostPath,
+			ReclaimPolicy: entry.ReclaimPolicy,
+			Annotations:   entry.Annotations,
+			Labels:        entry.Labels,
+		},
+	}
+	if err := writeYaml(WriteYamlInput{Object: values, Path: filepath.Join(chartDir, "values.yaml")}); err != nil {
+		return fmt.Errorf("error writing values: %w", err)
+	}
+
+	if err := createKubernetesNamespace(ctx, namespace); err != nil {
+		return fmt.Errorf("error creating namespace: %w", err)
+	}
+
+	helmAgent, err := NewHelmAgent(namespace, DeployLogPrinter)
+	if err != nil {
+		return fmt.Errorf("error creating helm agent: %w", err)
+	}
+
+	chartPath, err := filepath.Abs(chartDir)
+	if err != nil {
+		return fmt.Errorf("error getting chart path: %w", err)
+	}
+
+	common.LogVerboseQuiet(fmt.Sprintf("Installing storage chart for %s", entry.Name))
+	if err := helmAgent.InstallOrUpgradeChart(ctx, ChartInput{
+		ChartPath:   chartPath,
+		Namespace:   namespace,
+		ReleaseName: releaseName,
+		Wait:        false,
+	}); err != nil {
+		return fmt.Errorf("error installing storage chart: %w", err)
+	}
+	return nil
+}
+
+// TriggerStorageDestroy removes the helm release backing a storage entry.
+func TriggerStorageDestroy(ctx context.Context, entryName string) error {
+	if err := isKubernetesAvailable(); err != nil {
+		return fmt.Errorf("kubernetes not available: %w", err)
+	}
+
+	entry, err := readEntryFromStdin()
+	if err != nil {
+		return err
+	}
+	if entry.Scheduler != storage.SchedulerK3s {
+		return nil
+	}
+
+	namespace := entry.Namespace
+	if namespace == "" {
+		namespace = "default"
+	}
+	releaseName := GetStorageReleaseName(entryName)
+
+	helmAgent, err := NewHelmAgent(namespace, DeployLogPrinter)
+	if err != nil {
+		return fmt.Errorf("error creating helm agent: %w", err)
+	}
+	exists, err := helmAgent.ChartExists(releaseName)
+	if err != nil {
+		return fmt.Errorf("error checking if storage chart exists: %w", err)
+	}
+	if !exists {
+		return nil
+	}
+
+	common.LogVerboseQuiet(fmt.Sprintf("Removing storage chart for %s", entryName))
+	if err := helmAgent.UninstallChart(releaseName); err != nil {
+		return fmt.Errorf("error uninstalling storage chart: %w", err)
+	}
+	return nil
+}
+
+// TriggerStorageStatus prints "Bound" if the entry's PVC is bound,
+// "Pending" or "Lost" otherwise. Used by the storage:wait command.
+func TriggerStorageStatus(ctx context.Context, entryName string) error {
+	if err := isKubernetesAvailable(); err != nil {
+		return fmt.Errorf("kubernetes not available: %w", err)
+	}
+	entry, err := readEntryFromStdin()
+	if err != nil {
+		return err
+	}
+	namespace := entry.Namespace
+	if namespace == "" {
+		namespace = "default"
+	}
+
+	clientset, err := NewKubernetesClient()
+	if err != nil {
+		return err
+	}
+	pvc, err := clientset.Client.CoreV1().PersistentVolumeClaims(namespace).Get(ctx, entryName, metav1.GetOptions{})
+	if err != nil {
+		return fmt.Errorf("error fetching PVC %s/%s: %w", namespace, entryName, err)
+	}
+	fmt.Println(string(pvc.Status.Phase))
+	return nil
+}
+
+// validateStorageClass surfaces a clear error when the requested storage
+// class doesn't exist in the cluster.
+func validateStorageClass(ctx context.Context, name string) error {
+	if name == "" {
+		return nil
+	}
+	clientset, err := NewKubernetesClient()
+	if err != nil {
+		return err
+	}
+	classes, err := clientset.Client.StorageV1().StorageClasses().List(ctx, metav1.ListOptions{})
+	if err != nil {
+		return fmt.Errorf("error listing storage classes: %w", err)
+	}
+	available := []string{}
+	for _, cls := range classes.Items {
+		if cls.Name == name {
+			return nil
+		}
+		available = append(available, cls.Name)
+	}
+	return fmt.Errorf("storage class %q does not exist; available: %s", name, strings.Join(available, ", "))
+}
+
+// AppMountPair mirrors the storage plugin's wire format so we can decode
+// the storage-app-mounts trigger output here.
+type AppMountPair struct {
+	Entry      *storage.Entry      `json:"entry"`
+	Attachment *storage.Attachment `json:"attachment"`
+}
+
+// LoadAppMounts asks the storage plugin for an app's mount pairs in the
+// given phase. Returns an empty slice when there are none.
+func LoadAppMounts(appName string, phase string) ([]AppMountPair, error) {
+	results, err := common.CallPlugnTrigger(common.PlugnTriggerInput{
+		Trigger: "storage-app-mounts",
+		Args:    []string{appName, phase},
+	})
+	if err != nil {
+		return nil, err
+	}
+	output := strings.TrimSpace(results.StdoutContents())
+	if output == "" {
+		return nil, nil
+	}
+	pairs := []AppMountPair{}
+	if err := json.Unmarshal([]byte(output), &pairs); err != nil {
+		return nil, fmt.Errorf("unable to parse storage-app-mounts output: %w", err)
+	}
+	return pairs, nil
+}
+
+// ToProcessVolumes converts each AppMountPair into a ProcessVolume. K3s
+// app deployments reference the PVC by name; the PVC itself is owned by
+// the storage entry's separate helm release. Any docker-local entries
+// found here are an error - they cannot be mounted on a k3s app.
+func ToProcessVolumes(pairs []AppMountPair) ([]ProcessVolume, error) {
+	volumes := []ProcessVolume{}
+	for _, pair := range pairs {
+		if pair.Entry == nil || pair.Attachment == nil {
+			continue
+		}
+		if pair.Entry.Scheduler == storage.SchedulerDockerLocal {
+			return nil, fmt.Errorf("storage entry %q is scheduler=docker-local but is mounted on a k3s app; recreate it with --scheduler k3s", pair.Entry.Name)
+		}
+		volumes = append(volumes, ProcessVolume{
+			Name:      pair.Entry.Name,
+			MountPath: pair.Attachment.ContainerPath,
+			SubPath:   pair.Attachment.Subpath,
+			ReadOnly:  pair.Attachment.Readonly,
+			PersistentClaim: &ProcessVolumePersistentClaim{
+				ClaimName: pair.Entry.Name,
+			},
+		})
+	}
+	return volumes, nil
+}
+
+// asK8sVolumeMount is a small helper used by tests / other callers that
+// want a corev1.VolumeMount from a ProcessVolume.
+func asK8sVolumeMount(v ProcessVolume) corev1.VolumeMount {
+	return corev1.VolumeMount{
+		Name:      v.Name,
+		MountPath: v.MountPath,
+		SubPath:   v.SubPath,
+		ReadOnly:  v.ReadOnly,
+	}
+}

plugins/scheduler-k3s/template.go

@@ -117,9 +117,12 @@ type ProcessValues struct {
 }
 
 type ProcessVolume struct {
-	Name      string                 `yaml:"name"`
-	MountPath string                 `yaml:"mount_path"`
-	EmptyDir  *ProcessVolumeEmptyDir `yaml:"empty_dir,omitempty"`
+	Name            string                        `yaml:"name"`
+	MountPath       string                        `yaml:"mount_path"`
+	SubPath         string                        `yaml:"sub_path,omitempty"`
+	ReadOnly        bool                          `yaml:"read_only,omitempty"`
+	EmptyDir        *ProcessVolumeEmptyDir        `yaml:"empty_dir,omitempty"`
+	PersistentClaim *ProcessVolumePersistentClaim `yaml:"persistent_claim,omitempty"`
 }
 
 type ProcessVolumeEmptyDir struct {
@@ -127,6 +130,10 @@ type ProcessVolumeEmptyDir struct {
 	SizeLimit string `yaml:"size_limit"`
 }
 
+type ProcessVolumePersistentClaim struct {
+	ClaimName string `yaml:"claim_name"`
+}
+
 type ProcessAnnotations struct {
 	CertificateAnnotations               map[string]string `yaml:"certificate,omitempty"`
 	CronJobAnnotations                   map[string]string `yaml:"cronjob,omitempty"`

plugins/scheduler-k3s/templates/chart/cron-job.yaml

@@ -127,12 +127,44 @@ spec:
             {{- if $.Values.global.image.working_dir }}
             workingDir: {{ $.Values.global.image.working_dir }}
             {{- end }}
+            {{- if $config.volumes }}
+            volumeMounts:
+            {{- range $volume := $config.volumes }}
+            - name: {{ $volume.name }}
+              mountPath: {{ $volume.mount_path }}
+              {{- if $volume.sub_path }}
+              subPath: {{ $volume.sub_path }}
+              {{- end }}
+              {{- if $volume.read_only }}
+              readOnly: true
+              {{- end }}
+            {{- end }}
+            {{- end }}
           {{- if $.Values.global.image.image_pull_secrets }}
           imagePullSecrets:
           - name: {{ $.Values.global.image.image_pull_secrets }}
           {{- end }}
           restartPolicy: Never
           serviceAccountName: {{ $.Values.global.app_name }}
+          {{- if $config.volumes }}
+          volumes:
+            {{- range $volume := $config.volumes }}
+            - name: {{ $volume.name }}
+              {{- if $volume.empty_dir }}
+              emptyDir:
+                {{- if $volume.empty_dir.medium }}
+                medium: {{ $volume.empty_dir.medium }}
+                {{- end }}
+                {{- if $volume.empty_dir.size_limit }}
+                sizeLimit: {{ $volume.empty_dir.size_limit }}
+                {{- end }}
+              {{- end }}
+              {{- if $volume.persistent_claim }}
+              persistentVolumeClaim:
+                claimName: {{ $volume.persistent_claim.claim_name }}
+              {{- end }}
+            {{- end }}
+          {{- end }}
   schedule: {{ $config.cron.schedule }}
   startingDeadlineSeconds: 60
   successfulJobsHistoryLimit: 10

plugins/scheduler-k3s/templates/chart/deployment.yaml

@@ -152,6 +152,12 @@ spec:
         {{- range $volume := $config.volumes }}
         - name: {{ $volume.name }}
           mountPath: {{ $volume.mount_path }}
+          {{- if $volume.sub_path }}
+          subPath: {{ $volume.sub_path }}
+          {{- end }}
+          {{- if $volume.read_only }}
+          readOnly: true
+          {{- end }}
         {{- end }}
         {{- end }}
       {{- if $.Values.global.image.image_pull_secrets }}
@@ -172,6 +178,10 @@ spec:
             sizeLimit: {{ $volume.empty_dir.size_limit }}
             {{- end }}
           {{- end }}
+          {{- if $volume.persistent_claim }}
+          persistentVolumeClaim:
+            claimName: {{ $volume.persistent_claim.claim_name }}
+          {{- end }}
         {{- end }}
       {{- end }}
 {{- end }}

plugins/scheduler-k3s/templates/storage-chart/templates/persistent-volume-claim.yaml

@@ -0,0 +1,29 @@
+apiVersion: v1
+kind: PersistentVolumeClaim
+metadata:
+  name: {{ .Values.global.entry_name }}
+  namespace: {{ .Values.global.namespace }}
+  labels:
+    app.kubernetes.io/name: {{ .Values.global.entry_name }}
+    app.kubernetes.io/managed-by: dokku
+    {{- range $k, $v := .Values.global.labels }}
+    {{ $k }}: {{ $v | quote }}
+    {{- end }}
+  annotations:
+    dokku.com/managed: "true"
+    {{- range $k, $v := .Values.global.annotations }}
+    {{ $k }}: {{ $v | quote }}
+    {{- end }}
+spec:
+  accessModes:
+  - {{ .Values.global.access_mode | default "ReadWriteOnce" }}
+  resources:
+    requests:
+      storage: {{ .Values.global.size }}
+  {{- if .Values.global.storage_class }}
+  storageClassName: {{ .Values.global.storage_class }}
+  {{- end }}
+  {{- if .Values.global.host_path }}
+  volumeName: {{ .Values.global.entry_name }}
+  storageClassName: ""
+  {{- end }}

plugins/scheduler-k3s/templates/storage-chart/templates/persistent-volume.yaml

@@ -0,0 +1,29 @@
+{{- if .Values.global.host_path }}
+apiVersion: v1
+kind: PersistentVolume
+metadata:
+  name: {{ .Values.global.entry_name }}
+  labels:
+    app.kubernetes.io/name: {{ .Values.global.entry_name }}
+    app.kubernetes.io/managed-by: dokku
+    {{- range $k, $v := .Values.global.labels }}
+    {{ $k }}: {{ $v | quote }}
+    {{- end }}
+  annotations:
+    dokku.com/managed: "true"
+    {{- range $k, $v := .Values.global.annotations }}
+    {{ $k }}: {{ $v | quote }}
+    {{- end }}
+spec:
+  capacity:
+    storage: {{ .Values.global.size }}
+  accessModes:
+  - {{ .Values.global.access_mode | default "ReadWriteOnce" }}
+  persistentVolumeReclaimPolicy: {{ .Values.global.reclaim_policy | default "Retain" }}
+  hostPath:
+    path: {{ .Values.global.host_path }}
+  storageClassName: ""
+  claimRef:
+    namespace: {{ .Values.global.namespace }}
+    name: {{ .Values.global.entry_name }}
+{{- end }}

plugins/scheduler-k3s/triggers.go

@@ -605,10 +605,19 @@ func TriggerSchedulerDeploy(scheduler string, appName string, imageTag string) e
 		})
 	}
 
+	deployMountPairs, err := LoadAppMounts(appName, "deploy")
+	if err != nil {
+		return fmt.Errorf("error loading storage app mounts: %w", err)
+	}
+	deployVolumes, err := ToProcessVolumes(deployMountPairs)
+	if err != nil {
+		return err
+	}
+	processVolumes = append(processVolumes, deployVolumes...)
+
 	for processType, processCount := range processes {
 		// todo: implement deployment annotations
 		// todo: implement pod annotations
-		// todo: implement volumes
 
 		healthchecks, ok := appJSON.Healthchecks[processType]
 		if !ok {
@@ -778,7 +787,6 @@ func TriggerSchedulerDeploy(scheduler string, appName string, imageTag string) e
 	for _, cronTask := range cronTasks {
 		// todo: implement deployment annotations
 		// todo: implement pod annotations
-		// todo: implement volumes
 		suffix := ""
 		for _, cronJob := range cronJobs {
 			if cronJob.Labels["dokku.com/cron-id"] == cronTask.ID {