cozystack/internal/migrationcontroller/forklift_objects.go

// SPDX-License-Identifier: Apache-2.0

package migrationcontroller

import (
	"context"
	"fmt"
	"maps"
	"strings"

	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/types"

	migrationv1alpha1 "github.com/cozystack/cozystack/api/migration/v1alpha1"
)

// taskOwnership returns the labels and owner reference every piece of a task's
// Forklift scaffolding carries. The labels let the controller watch Migrations
// and map them back without trusting a label an unrelated object might set; the
// owner reference is what makes deleting the task take the scaffolding with it.
func taskOwnership(task *migrationv1alpha1.VMImportTask) (map[string]string, metav1.OwnerReference) {
	return map[string]string{
		migrationv1alpha1.ManagedByLabel:           migrationv1alpha1.ManagedByValue,
		migrationv1alpha1.OwningTaskNameLabel:      task.Name,
		migrationv1alpha1.OwningTaskNamespaceLabel: task.Namespace,
	}, ownerRef(
		migrationv1alpha1.GroupVersion.WithKind("VMImportTask"), task.Name, task.UID)
}

// outputMarkers are the labels every object a task produces carries: which
// task made it, and for which source VM. Labels rather than owner references,
// because outputs are meant to outlive the task that created them.
func outputMarkers(task *migrationv1alpha1.VMImportTask, vmID string) map[string]string {
	return map[string]string{
		migrationv1alpha1.OutputTaskUIDLabel: string(task.UID),
		migrationv1alpha1.OutputVMIDLabel:    sanitizeName(vmID),
	}
}

// stampOutput writes the markers onto an object being created, preserving any
// labels already set on it.
func stampOutput(obj metav1.Object, task *migrationv1alpha1.VMImportTask, vmID string) {
	labels := obj.GetLabels()
	if labels == nil {
		labels = map[string]string{}
	}
	maps.Copy(labels, outputMarkers(task, vmID))
	obj.SetLabels(labels)
}

// isOwnOutput reports whether an existing object is this task's own earlier
// output for this source VM, rather than something that merely shares its name.
// Absence of the marker is the answer that matters: it means a tenant object is
// standing where an output would go, which is a collision and never a resume.
func isOwnOutput(obj metav1.Object, task *migrationv1alpha1.VMImportTask, vmID string) bool {
	labels := obj.GetLabels()
	if labels == nil {
		return false
	}
	return labels[migrationv1alpha1.OutputTaskUIDLabel] == string(task.UID) &&
		labels[migrationv1alpha1.OutputVMIDLabel] == sanitizeName(vmID)
}

// providerRefs is the source/destination pair every Forklift object in a task
// points at. Both live in the task's namespace, because a Source does.
func providerRefs(task *migrationv1alpha1.VMImportTask, src *migrationv1alpha1.VMImportSource) map[string]any {
	return map[string]any{
		"source": map[string]any{
			"name":      sourceProviderName(src.Name),
			"namespace": task.Namespace,
		},
		"destination": map[string]any{
			"name":      destinationProviderName(src.Name),
			"namespace": task.Namespace,
		},
	}
}

// ensureMaps creates the NetworkMap and StorageMap a Plan requires. Both start
// empty — Forklift will not validate a Plan without them — and populateMaps
// fills them in from what the inventory says each VM actually uses, one VM at a
// time as its reconcile reaches this point.
func (r *VMImportTaskReconciler) ensureMaps(
	ctx context.Context,
	task *migrationv1alpha1.VMImportTask,
	src *migrationv1alpha1.VMImportSource,
	storageClass string,
) error {
	labels, owner := taskOwnership(task)
	name := mapName(task.Name)

	for _, gvk := range []struct {
		gvk schema.GroupVersionKind
	}{{networkMapGVK}, {storageMapGVK}} {
		existing := newObject(gvk.gvk)
		err := r.Get(ctx, types.NamespacedName{Namespace: task.Namespace, Name: name}, existing)
		if err == nil {
			continue
		}
		if !apierrors.IsNotFound(err) {
			return err
		}
		obj := newObject(gvk.gvk)
		obj.SetName(name)
		obj.SetNamespace(task.Namespace)
		obj.SetLabels(labels)
		obj.SetOwnerReferences([]metav1.OwnerReference{owner})
		spec := map[string]any{
			"provider": providerRefs(task, src),
			"map":      []any{},
		}
		if err := unstructured.SetNestedMap(obj.Object, spec, "spec"); err != nil {
			return err
		}
		if err := r.Create(ctx, obj); err != nil && !apierrors.IsAlreadyExists(err) {
			return err
		}
	}

	// The storage class is not part of the map until a datastore is known, but
	// it must be recorded somewhere the next pass can read it: the map's own
	// annotation is the natural place, and it keeps the class immutable for the
	// task's lifetime even if the cluster default changes mid-import.
	sm := newObject(storageMapGVK)
	if err := r.Get(ctx, types.NamespacedName{Namespace: task.Namespace, Name: name}, sm); err != nil {
		return err
	}
	annotations := sm.GetAnnotations()
	if annotations == nil {
		annotations = map[string]string{}
	}
	if annotations[storageClassAnnotation] == storageClass {
		return nil
	}
	annotations[storageClassAnnotation] = storageClass
	sm.SetAnnotations(annotations)
	return r.Update(ctx, sm)
}

// storageClassAnnotation records the resolved StorageClass on the task's
// StorageMap, so mappings learned later all land on the same class.
const storageClassAnnotation = "forklift.cozystack.io/storage-class"

// forkliftPowerStateOff is Forklift's `off` target power state
// (plan.TargetPowerStateOff upstream). Spelled out rather than imported: the
// controller deliberately builds Forklift objects as unstructured data and
// takes no compile-time dependency on the Forklift API module.
const forkliftPowerStateOff = "off"

// createPlan renders the Forklift Plan that migrates one VM.
//
// Raw copy (skipGuestConversion) is the only mode v1 offers, and it is the one
// that needs no privileged pod anywhere: Forklift emits a CDI DataVolume with a
// VDDK source and CDI's own importer — which satisfies the restricted Pod
// Security Standard — moves the bytes. Guest conversion needs a node-level
// seccomp profile and arrives as a separate, additive option.
func (r *VMImportTaskReconciler) createPlan(
	ctx context.Context,
	task *migrationv1alpha1.VMImportTask,
	src *migrationv1alpha1.VMImportSource,
	req *migrationv1alpha1.VMImportRequest,
) error {
	labels, owner := taskOwnership(task)
	obj := newObject(planGVK)
	obj.SetName(planName(task.Name, req.ID))
	obj.SetNamespace(task.Namespace)
	obj.SetLabels(labels)
	obj.SetOwnerReferences([]metav1.OwnerReference{owner})

	spec := map[string]any{
		"skipGuestConversion": true,
		"provider":            providerRefs(task, src),
		"map": map[string]any{
			"network": map[string]any{
				"name":      mapName(task.Name),
				"namespace": task.Namespace,
			},
			"storage": map[string]any{
				"name":      mapName(task.Name),
				"namespace": task.Namespace,
			},
		},
		// The destination is always the task's own namespace. There is no
		// field for anything else: an import lands where the object that
		// asked for it lives, which removes the whole class of
		// cross-namespace coercion bugs.
		"targetNamespace": task.Namespace,
		"vms": []any{
			map[string]any{
				"id": req.ID,
				// Keep the machine Forklift builds powered off. Left unset,
				// Forklift matches the source's pre-migration power state, so
				// migrating a running VM — every VM in a production cutover —
				// yields a target that boots the moment the transfer finishes.
				// That is a duplicate of a live machine on the network, and it
				// starts a guest on the very volume the handoff is about to
				// re-point. The VMInstance this controller builds is what
				// starts the imported guest, on the tenant's terms.
				"targetPowerState": forkliftPowerStateOff,
			},
		},
	}
	if err := unstructured.SetNestedMap(obj.Object, spec, "spec"); err != nil {
		return err
	}
	if err := r.Create(ctx, obj); err != nil && !apierrors.IsAlreadyExists(err) {
		return err
	}
	return nil
}

// ensureMigration creates the Migration that runs a validated Plan. A Plan
// describes the work; a Migration is the instruction to do it.
func (r *VMImportTaskReconciler) ensureMigration(
	ctx context.Context,
	task *migrationv1alpha1.VMImportTask,
	plan string,
) (*unstructured.Unstructured, error) {
	existing := newObject(migrationGVK)
	err := r.Get(ctx, types.NamespacedName{Namespace: task.Namespace, Name: plan}, existing)
	if err == nil {
		return existing, nil
	}
	if !apierrors.IsNotFound(err) {
		return nil, err
	}

	labels, owner := taskOwnership(task)
	obj := newObject(migrationGVK)
	obj.SetName(plan)
	obj.SetNamespace(task.Namespace)
	obj.SetLabels(labels)
	obj.SetOwnerReferences([]metav1.OwnerReference{owner})
	spec := map[string]any{
		"plan": map[string]any{
			"name":      plan,
			"namespace": task.Namespace,
		},
	}
	if err := unstructured.SetNestedMap(obj.Object, spec, "spec"); err != nil {
		return nil, err
	}
	if err := r.Create(ctx, obj); err != nil && !apierrors.IsAlreadyExists(err) {
		return nil, err
	}
	// Return the object as created; its status appears on a later pass.
	return obj, nil
}

// populateMaps adds one VM's networks and datastores to the task's maps.
//
// The topology comes from Forklift's inventory, because it exists nowhere else:
// a Plan's validation conditions name the VMs that are unmapped, never what
// they are unmapped to. The maps accumulate across the task's VMs — each pass
// adds only what is missing — so the task ends up with one entry per distinct
// network and datastore however many machines share them.
//
// Networks all map to the pod network. That is not a placeholder: the map
// shapes the interfaces of the KubeVirt VM Forklift creates, and this
// controller discards that VM unstarted — the imported machine's networking
// comes from the VMInstance it builds instead, which attaches to the pod
// network. When a richer network placement design lands, this is where a
// tenant-specified mapping would be honoured.
//
// Returns true when it changed something, so the caller waits for Forklift to
// re-validate rather than racing ahead.
func (r *VMImportTaskReconciler) populateMaps(
	ctx context.Context,
	task *migrationv1alpha1.VMImportTask,
	src *migrationv1alpha1.VMImportSource,
	req *migrationv1alpha1.VMImportRequest,
) (bool, error) {
	networks, datastores, err := r.sourceTopology(ctx, task, src, req)
	if err != nil {
		return false, err
	}
	if len(networks) == 0 && len(datastores) == 0 {
		return false, nil
	}

	name := mapName(task.Name)
	changed := false

	if len(networks) > 0 {
		nm := newObject(networkMapGVK)
		if err := r.Get(ctx, types.NamespacedName{Namespace: task.Namespace, Name: name}, nm); err != nil {
			return false, err
		}
		entries, _, _ := unstructured.NestedSlice(nm.Object, "spec", "map")
		added := false
		for _, id := range networks {
			if hasSourceID(entries, id) {
				continue
			}
			entries = append(entries, map[string]any{
				"source":      map[string]any{"id": id},
				"destination": map[string]any{"type": "pod"},
			})
			added = true
		}
		if added {
			if err := unstructured.SetNestedSlice(nm.Object, entries, "spec", "map"); err != nil {
				return false, err
			}
			if err := r.Update(ctx, nm); err != nil {
				return false, err
			}
			changed = true
		}
	}

	if len(datastores) > 0 {
		sm := newObject(storageMapGVK)
		if err := r.Get(ctx, types.NamespacedName{Namespace: task.Namespace, Name: name}, sm); err != nil {
			return false, err
		}
		storageClass := sm.GetAnnotations()[storageClassAnnotation]
		if storageClass == "" {
			return false, fmt.Errorf("StorageMap %s/%s carries no resolved storage class", task.Namespace, name)
		}
		entries, _, _ := unstructured.NestedSlice(sm.Object, "spec", "map")
		added := false
		for _, id := range datastores {
			if hasSourceID(entries, id) {
				continue
			}
			entries = append(entries, map[string]any{
				"source":      map[string]any{"id": id},
				"destination": map[string]any{"storageClass": storageClass},
			})
			added = true
		}
		if added {
			if err := unstructured.SetNestedSlice(sm.Object, entries, "spec", "map"); err != nil {
				return false, err
			}
			if err := r.Update(ctx, sm); err != nil {
				return false, err
			}
			changed = true
		}
	}

	return changed, nil
}

// sourceTopology asks Forklift's inventory which networks and datastores one
// VM actually uses.
//
// This is the only place the answer exists. Forklift's Plan conditions name the
// VMs that are unmapped, not what they are unmapped to, and no Forklift custom
// resource carries a VM's network or datastore IDs — see the comment on
// inventoryClient for the source references.
//
// Scoped to a single VM on purpose. Reading the whole task here would tie every
// VM's fate to every other one: a single mistyped managed-object reference — the
// likeliest error on this API — would fail the VM being reconciled with a
// message naming a different machine, where a missing VM is meant to fail alone
// while its siblings proceed. It is also what keeps the cost linear: one GET per
// VM per pass rather than one per VM *for* every VM, repeated on every requeue
// while anything is still transferring.
func (r *VMImportTaskReconciler) sourceTopology(
	ctx context.Context,
	task *migrationv1alpha1.VMImportTask,
	src *migrationv1alpha1.VMImportSource,
	req *migrationv1alpha1.VMImportRequest,
) (networks, datastores []string, err error) {
	if r.Inventory == nil {
		return nil, nil, fmt.Errorf("no inventory client configured; the network and storage maps cannot be built")
	}

	// The inventory addresses providers by UID, so the Provider object this
	// task's Source produced has to be read for it.
	provider := newObject(providerGVK)
	if err := r.Get(ctx, types.NamespacedName{
		Namespace: task.Namespace, Name: sourceProviderName(src.Name),
	}, provider); err != nil {
		return nil, nil, err
	}
	uid := string(provider.GetUID())
	if uid == "" {
		return nil, nil, fmt.Errorf("source Provider %s/%s has no UID yet", task.Namespace, sourceProviderName(src.Name))
	}

	vm, err := r.Inventory.VM(ctx, uid, req.ID)
	if err != nil {
		return nil, nil, err
	}
	return vm.networkIDs(), vm.datastoreIDs(), nil
}

func hasSourceID(entries []any, id string) bool {
	for _, raw := range entries {
		entry, ok := raw.(map[string]any)
		if !ok {
			continue
		}
		existing, _, _ := unstructured.NestedString(entry, "source", "id")
		if existing == id {
			return true
		}
		name, _, _ := unstructured.NestedString(entry, "source", "name")
		if name == id {
			return true
		}
	}
	return false
}

// planCriticalCondition returns Forklift's own words for why a Plan cannot run,
// or empty if nothing is critically wrong. Unmapped references are excluded:
// those are not failures, they are what populateMaps resolves.
func planCriticalCondition(plan *unstructured.Unstructured) string {
	conditions, _, _ := unstructured.NestedSlice(plan.Object, "status", "conditions")
	var critical []string
	for _, raw := range conditions {
		cond, ok := raw.(map[string]any)
		if !ok {
			continue
		}
		status, _, _ := unstructured.NestedString(cond, "status")
		category, _, _ := unstructured.NestedString(cond, "category")
		ctype, _, _ := unstructured.NestedString(cond, "type")
		message, _, _ := unstructured.NestedString(cond, "message")
		if status != "True" || category != "Critical" {
			continue
		}
		if strings.Contains(ctype, "NotMapped") {
			continue
		}
		critical = append(critical, fmt.Sprintf("%s: %s", ctype, message))
	}
	if len(critical) == 0 {
		return ""
	}
	return strings.Join(critical, "; ")
}

// planReady reports whether Forklift considers the Plan runnable.
func planReady(plan *unstructured.Unstructured) bool {
	conditions, _, _ := unstructured.NestedSlice(plan.Object, "status", "conditions")
	for _, raw := range conditions {
		cond, ok := raw.(map[string]any)
		if !ok {
			continue
		}
		ctype, _, _ := unstructured.NestedString(cond, "type")
		status, _, _ := unstructured.NestedString(cond, "status")
		if ctype == "Ready" && status == "True" {
			return true
		}
	}
	return false
}

// migrationProgress reads a Migration's status: whether the transfer finished,
// how far along it is, and the failure text if it failed.
func migrationProgress(migration *unstructured.Unstructured) (done bool, progress int32, failure string) {
	vms, _, _ := unstructured.NestedSlice(migration.Object, "status", "vms")
	if len(vms) == 0 {
		return false, 0, ""
	}
	vm, ok := vms[0].(map[string]any)
	if !ok {
		return false, 0, ""
	}

	if reasons, found, _ := unstructured.NestedStringSlice(vm, "error", "reasons"); found && len(reasons) > 0 {
		return false, 0, strings.Join(reasons, "; ")
	}
	phase, _, _ := unstructured.NestedString(vm, "phase")
	if phase == "Failed" {
		return false, 0, "the migration engine reported the transfer as failed"
	}

	// Progress is a completed/total pair per pipeline step, counted in MiB.
	// Prefer the disk-transfer step, because averaging a four-step pipeline
	// reads 25% before a byte has moved — but fall back to the sum over every
	// step that reports a total, since the step names are not part of any
	// contract and a release that renames them would otherwise peg progress at
	// zero for the whole transfer.
	var completed, total int64
	var namedCompleted, namedTotal int64
	pipeline, _, _ := unstructured.NestedSlice(vm, "pipeline")
	for _, raw := range pipeline {
		step, ok := raw.(map[string]any)
		if !ok {
			continue
		}
		c, _, _ := unstructured.NestedInt64(step, "progress", "completed")
		t, _, _ := unstructured.NestedInt64(step, "progress", "total")
		if t <= 0 {
			continue
		}
		completed += c
		total += t
		if name, _, _ := unstructured.NestedString(step, "name"); strings.Contains(strings.ToLower(name), "disktransfer") {
			namedCompleted += c
			namedTotal += t
		}
	}
	switch {
	case namedTotal > 0:
		progress = int32(namedCompleted * 100 / namedTotal)
	case total > 0:
		progress = int32(completed * 100 / total)
	}

	// Either signal means the transfer is over: the phase the engine settles
	// on, or the completion timestamp it stamps.
	if phase == "Completed" {
		return true, 100, ""
	}
	if ts, found, _ := unstructured.NestedString(vm, "completed"); found && ts != "" {
		return true, 100, ""
	}
	return false, progress, ""
}