ananke/internal/cluster/orchestrator_longhorn_reconcile.go

package cluster

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"sort"
	"strings"
	"time"
)

const longhornSystemNamespace = "longhorn-system"

type longhornNodeList struct {
	Items []longhornNode `json:"items"`
}

type longhornNode struct {
	Metadata struct {
		Name string `json:"name"`
	} `json:"metadata"`
	Spec struct {
		AllowScheduling *bool `json:"allowScheduling"`
	} `json:"spec"`
	Status struct {
		Conditions []longhornCondition `json:"conditions"`
	} `json:"status"`
}

type longhornCondition struct {
	Type    string `json:"type"`
	Status  string `json:"status"`
	Reason  string `json:"reason"`
	Message string `json:"message"`
}

type daemonSetResource struct {
	Metadata struct {
		Namespace string `json:"namespace"`
		Name      string `json:"name"`
	} `json:"metadata"`
	Spec struct {
		Selector struct {
			MatchLabels map[string]string `json:"matchLabels"`
		} `json:"selector"`
	} `json:"spec"`
}

// reconcileLonghornKubernetesReadiness repairs safe Kubernetes/Longhorn drift.
// Signature: (o *Orchestrator) reconcileLonghornKubernetesReadiness(ctx context.Context) (int, error).
// Why: Kubernetes Ready=True is not enough for PVC scheduling when Longhorn's
// manager, labels, and node conditions disagree with the Kubernetes node model.
func (o *Orchestrator) reconcileLonghornKubernetesReadiness(ctx context.Context) (int, error) {
	longhornNodes, err := o.queryLonghornNodes(ctx)
	if err != nil {
		return 0, err
	}
	if len(longhornNodes.Items) == 0 {
		return 0, nil
	}
	managerDS, err := o.queryLonghornManagerDaemonSet(ctx)
	if err != nil {
		return 0, err
	}
	selector := normalizeSelectorLabels(managerDS.Spec.Selector.MatchLabels)
	if len(selector) == 0 {
		return 0, fmt.Errorf("longhorn manager daemonset has no matchLabels selector")
	}
	nodes, err := o.queryReadyNodes(ctx)
	if err != nil {
		return 0, err
	}
	kubeNodes := map[string]nodeReadyItem{}
	for _, node := range nodes.Items {
		name := strings.TrimSpace(node.Metadata.Name)
		if name != "" {
			kubeNodes[name] = node
		}
	}
	managerPods, err := o.longhornManagerPodsByNode(ctx, selector)
	if err != nil {
		return 0, err
	}

	repaired := 0
	errs := []string{}
	for _, lhNode := range longhornNodes.Items {
		name := strings.TrimSpace(lhNode.Metadata.Name)
		if name == "" {
			continue
		}
		kubeNode, ok := kubeNodes[name]
		if !ok {
			continue
		}
		if _, hasManager := managerPods[name]; hasManager {
			continue
		}
		readyCond := longhornConditionByType(lhNode, "Ready")
		if readyCond == nil || strings.EqualFold(strings.TrimSpace(readyCond.Status), "True") {
			continue
		}
		if !strings.EqualFold(strings.TrimSpace(readyCond.Reason), "ManagerPodMissing") {
			continue
		}
		missing := missingSelectorLabels(kubeNode.Metadata.Labels, selector)
		if len(missing) == 0 {
			o.log.Printf("warning: Longhorn manager pod missing on node=%s but node already matches selector; leaving for DaemonSet scheduler reconciliation", name)
			continue
		}

		labels := make([]string, 0, len(missing))
		for key, value := range missing {
			labels = append(labels, key+"="+value)
		}
		sort.Strings(labels)
		args := append([]string{"label", "node", name, "--overwrite"}, labels...)
		o.log.Printf("warning: restoring Longhorn manager selector labels on node=%s labels=%s reason=LonghornNodeExistsAndManagerPodMissing", name, strings.Join(labels, ","))
		if _, err := o.kubectl(ctx, 25*time.Second, args...); err != nil {
			errs = append(errs, fmt.Sprintf("%s restore Longhorn selector labels: %v", name, err))
			continue
		}
		repaired++
		o.noteStartupAutoHeal(fmt.Sprintf("restored Longhorn manager selector labels on %s: %s", name, strings.Join(labels, ",")))
	}
	if len(errs) > 0 {
		return repaired, errors.New(strings.Join(errs, "; "))
	}
	return repaired, nil
}

// queryLonghornNodes runs one orchestration or CLI step.
// Signature: (o *Orchestrator) queryLonghornNodes(ctx context.Context) (longhornNodeList, error).
// Why: Longhorn/Kubernetes drift reconciliation starts from Longhorn's own node
// model rather than assuming Kubernetes Ready is sufficient for storage.
func (o *Orchestrator) queryLonghornNodes(ctx context.Context) (longhornNodeList, error) {
	out, err := o.kubectl(ctx, 30*time.Second, "-n", "longhorn-system", "get", "nodes.longhorn.io", "-o", "json")
	if err != nil {
		if isNotFoundErr(err) {
			return longhornNodeList{}, nil
		}
		return longhornNodeList{}, fmt.Errorf("query longhorn nodes: %w", err)
	}
	if strings.TrimSpace(out) == "" {
		return longhornNodeList{}, nil
	}
	var nodes longhornNodeList
	if err := json.Unmarshal([]byte(out), &nodes); err != nil {
		return longhornNodeList{}, fmt.Errorf("decode longhorn nodes: %w", err)
	}
	return nodes, nil
}

// queryLonghornManagerDaemonSet runs one orchestration or CLI step.
// Signature: (o *Orchestrator) queryLonghornManagerDaemonSet(ctx context.Context) (daemonSetResource, error).
// Why: manager selector labels are installation policy, so repair must read
// the live DaemonSet selector instead of hard-coding a Titan label.
func (o *Orchestrator) queryLonghornManagerDaemonSet(ctx context.Context) (daemonSetResource, error) {
	out, err := o.kubectl(ctx, 20*time.Second, "-n", "longhorn-system", "get", "daemonset", "longhorn-manager", "-o", "json")
	if err != nil {
		return daemonSetResource{}, fmt.Errorf("query longhorn manager daemonset: %w", err)
	}
	var ds daemonSetResource
	if err := json.Unmarshal([]byte(out), &ds); err != nil {
		return daemonSetResource{}, fmt.Errorf("decode longhorn manager daemonset: %w", err)
	}
	return ds, nil
}

// longhornManagerPodsByNode runs one orchestration or CLI step.
// Signature: (o *Orchestrator) longhornManagerPodsByNode(ctx context.Context, selector map[string]string) (map[string]struct{}, error).
// Why: manager-pod presence is the postcondition for selector repair and
// distinguishes label drift from a healthy Longhorn node.
func (o *Orchestrator) longhornManagerPodsByNode(ctx context.Context, selector map[string]string) (map[string]struct{}, error) {
	args := []string{"-n", "longhorn-system", "get", "pods", "-o", "json"}
	if len(selector) > 0 {
		args = append(args, "-l", selectorLabelString(selector))
	}
	out, err := o.kubectl(ctx, 20*time.Second, args...)
	if err != nil {
		return nil, fmt.Errorf("query longhorn manager pods: %w", err)
	}
	var pods podList
	if err := json.Unmarshal([]byte(out), &pods); err != nil {
		return nil, fmt.Errorf("decode longhorn manager pods: %w", err)
	}
	byNode := map[string]struct{}{}
	for _, pod := range pods.Items {
		node := strings.TrimSpace(pod.Spec.NodeName)
		if node == "" {
			continue
		}
		if podMatchesLabels(pod.Metadata.Labels, selector) {
			byNode[node] = struct{}{}
		}
	}
	return byNode, nil
}

// normalizeSelectorLabels runs one orchestration or CLI step.
// Signature: normalizeSelectorLabels(labels map[string]string) map[string]string.
// Why: empty selector fragments should not produce accidental node label writes.
func normalizeSelectorLabels(labels map[string]string) map[string]string {
	out := map[string]string{}
	for key, value := range labels {
		key = strings.TrimSpace(key)
		value = strings.TrimSpace(value)
		if key != "" && value != "" {
			out[key] = value
		}
	}
	return out
}

// selectorLabelString runs one orchestration or CLI step.
// Signature: selectorLabelString(labels map[string]string) string.
// Why: kubectl label selectors need stable ordering for deterministic tests and
// readable logs.
func selectorLabelString(labels map[string]string) string {
	parts := make([]string, 0, len(labels))
	for key, value := range labels {
		parts = append(parts, key+"="+value)
	}
	sort.Strings(parts)
	return strings.Join(parts, ",")
}

// podMatchesLabels runs one orchestration or CLI step.
// Signature: podMatchesLabels(labels map[string]string, selector map[string]string) bool.
// Why: Longhorn manager pod checks should use the DaemonSet selector rather
// than assuming a specific app label.
func podMatchesLabels(labels map[string]string, selector map[string]string) bool {
	for key, value := range selector {
		if labels == nil || strings.TrimSpace(labels[key]) != value {
			return false
		}
	}
	return true
}

// missingSelectorLabels runs one orchestration or CLI step.
// Signature: missingSelectorLabels(labels map[string]string, selector map[string]string) map[string]string.
// Why: label-drift repair should write exactly the manager selector labels that
// are absent or wrong on the node.
func missingSelectorLabels(labels map[string]string, selector map[string]string) map[string]string {
	missing := map[string]string{}
	for key, value := range selector {
		if labels == nil || strings.TrimSpace(labels[key]) != value {
			missing[key] = value
		}
	}
	return missing
}

// longhornConditionByType runs one orchestration or CLI step.
// Signature: longhornConditionByType(node longhornNode, condType string) *longhornCondition.
// Why: Longhorn node condition parsing is shared by readiness and schedulability
// checks and should remain case-insensitive.
func longhornConditionByType(node longhornNode, condType string) *longhornCondition {
	for i := range node.Status.Conditions {
		if strings.EqualFold(strings.TrimSpace(node.Status.Conditions[i].Type), condType) {
			return &node.Status.Conditions[i]
		}
	}
	return nil
}