dokku/plugins/scheduler-k3s/storage.go

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,
	}
}