diff --git a/pkg/kube/longhorn-utils.sh b/pkg/kube/longhorn-utils.sh index b25797c2773..01c3cc44bf6 100644 --- a/pkg/kube/longhorn-utils.sh +++ b/pkg/kube/longhorn-utils.sh @@ -247,6 +247,19 @@ Longhorn_is_ready() { return 1 fi + # Longhorn runs a volume's engine and replica processes inside the + # instance-manager pod, so the node cannot serve a volume until one is + # running. It is owned by an InstanceManager CR rather than a DaemonSet, + # so the daemonset sweep above cannot observe it. + imState=$(kubectl -n longhorn-system get instancemanagers.longhorn.io -o json | jq -r --arg n "$node" '[.items[] | select(.spec.nodeID==$n) | .status.currentState] | index("running")') + if [ "$imState" = "null" ]; then + if [ -n "${bootLhRdyComplete}" ]; then + # Allow the final ready log message when its reached. + bootLhRdyComplete="" + fi + return 1 + fi + if [ -z "${bootLhRdyComplete}" ]; then logmsg "longhorn ds ready, node:$node nodedeploymentmap:$(echo "$ndm" | tr -d '\n')" bootLhRdyComplete="1" diff --git a/pkg/pillar/kubeapi/kubeapi.go b/pkg/pillar/kubeapi/kubeapi.go index 5e2fbfcab9e..75ab5c5b3cc 100644 --- a/pkg/pillar/kubeapi/kubeapi.go +++ b/pkg/pillar/kubeapi/kubeapi.go @@ -234,7 +234,7 @@ func WaitForKubernetes(agentName string, ps *pubsub.PubSub, stillRunning *time.T var lastUnmet error doneCh := make(chan struct{}, 1) go func() { - nodeReadyErr = wait.PollImmediate(time.Second, time.Minute*20, func() (bool, error) { + nodeReadyErr = wait.PollImmediate(time.Second, componentsReadyTimeout(opts), func() (bool, error) { if err := nodeReadyByName(client, nodeName); err != nil { lastUnmet = fmt.Errorf("node not ready: %w", err) return false, nil @@ -329,7 +329,7 @@ func checkLonghornReady(client kubernetes.Interface, nodeName string) error { } } - return nil + return instanceManagerReady(ctx, nodeName) } // nodeReadyByName confirms this device's Kubernetes node object exists. nodeName is the diff --git a/pkg/pillar/kubeapi/longhorninstancemanager.go b/pkg/pillar/kubeapi/longhorninstancemanager.go new file mode 100644 index 00000000000..bab057ac18d --- /dev/null +++ b/pkg/pillar/kubeapi/longhorninstancemanager.go @@ -0,0 +1,106 @@ +// Copyright (c) 2026 Zededa, Inc. +// SPDX-License-Identifier: Apache-2.0 + +//go:build k + +package kubeapi + +import ( + "context" + "fmt" + "time" + + lhv1beta2 "github.com/longhorn/longhorn-manager/k8s/pkg/apis/longhorn/v1beta2" + "github.com/longhorn/longhorn-manager/k8s/pkg/client/clientset/versioned" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +const ( + // baseComponentsReadyTimeout bounds the wait for the node object plus any + // optional components when Longhorn is not among them. + baseComponentsReadyTimeout = 20 * time.Minute + + // longhornComponentsReadyTimeout applies once Longhorn is in the predicate. + // The node cannot serve a volume until the instance-manager pod runs, and + // that pod pulls a ~440 MB image: measured at 8m20s and 8m41s on single-disk + // topologies but over 20 minutes on two-disk ZFS, which alone exhausts the + // base budget and leaves nothing for the node and kubevirt checks ahead of + // it. Sized to clear the slowest observed pull with room to spare. + longhornComponentsReadyTimeout = 45 * time.Minute +) + +// componentsReadyTimeout returns the deadline for the readiness poll, widened +// when the caller waits on Longhorn. +func componentsReadyTimeout(opts WaitForKubernetesOptions) time.Duration { + if opts.WaitForLonghorn { + return longhornComponentsReadyTimeout + } + return baseComponentsReadyTimeout +} + +// instanceManagerLister is the subset of the generated Longhorn client this +// file needs, kept narrow so the state logic below can be exercised without a +// live API server. +type instanceManagerLister interface { + List(ctx context.Context, opts metav1.ListOptions) (*lhv1beta2.InstanceManagerList, error) +} + +// instanceManagerRunningOnNode reports whether nodeName has an InstanceManager +// in the running state. +// +// Longhorn runs a volume's engine and replica processes inside the +// instance-manager pod, so the node cannot serve any volume until one is +// running. The pod is owned by an InstanceManager CR rather than a DaemonSet, +// which is why the daemonset sweep in checkLonghornReady cannot observe it. +// +// InstanceManager.Spec.NodeID carries the Kubernetes node name, the same value +// checkLonghornReady uses to select per-node DaemonSet pods. +func instanceManagerRunningOnNode(ctx context.Context, lister instanceManagerLister, + nodeName string) (bool, error) { + ims, err := lister.List(ctx, metav1.ListOptions{}) + if err != nil { + return false, err + } + for _, im := range ims.Items { + if im.Spec.NodeID != nodeName { + continue + } + if im.Status.CurrentState == lhv1beta2.InstanceManagerStateRunning { + return true, nil + } + } + return false, nil +} + +// instanceManagerReady is the gate checkLonghornReady applies once the Longhorn +// DaemonSets look healthy. It is a variable so that tests driving +// checkLonghornReady with a fake clientset can substitute it: the real +// implementation builds a Longhorn client from the on-device kubeconfig, which +// a fake clientset cannot supply. +var instanceManagerReady = checkLonghornInstanceManagerReady + +// checkLonghornInstanceManagerReady fails while nodeName has no running +// InstanceManager. +// +// Longhorn creates the CR during node setup rather than on first volume +// request, so waiting on it cannot deadlock against a volume whose own creation +// is gated on storage readiness. +func checkLonghornInstanceManagerReady(ctx context.Context, nodeName string) error { + config, err := GetKubeConfig() + if err != nil { + return fmt.Errorf("longhorn instance-manager: kubeconfig: %v", err) + } + lhClient, err := versioned.NewForConfig(config) + if err != nil { + return fmt.Errorf("longhorn instance-manager: versioned client: %v", err) + } + running, err := instanceManagerRunningOnNode(ctx, + lhClient.LonghornV1beta2().InstanceManagers(longhornNamespace), nodeName) + if err != nil { + return fmt.Errorf("longhorn instance-manager: list: %v", err) + } + if !running { + return fmt.Errorf("longhorn instance-manager not running on node") + } + return nil +} diff --git a/pkg/pillar/kubeapi/longhorninstancemanager_test.go b/pkg/pillar/kubeapi/longhorninstancemanager_test.go new file mode 100644 index 00000000000..db0d67421db --- /dev/null +++ b/pkg/pillar/kubeapi/longhorninstancemanager_test.go @@ -0,0 +1,196 @@ +// Copyright (c) 2026 Zededa, Inc. +// SPDX-License-Identifier: Apache-2.0 + +//go:build k + +package kubeapi + +import ( + "context" + "errors" + "testing" + "time" + + lhv1beta2 "github.com/longhorn/longhorn-manager/k8s/pkg/apis/longhorn/v1beta2" + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/client-go/kubernetes/fake" +) + +type fakeInstanceManagerLister struct { + items []lhv1beta2.InstanceManager + err error +} + +func (f fakeInstanceManagerLister) List(context.Context, metav1.ListOptions) ( + *lhv1beta2.InstanceManagerList, error) { + if f.err != nil { + return nil, f.err + } + return &lhv1beta2.InstanceManagerList{Items: f.items}, nil +} + +func instanceManager(nodeID string, state lhv1beta2.InstanceManagerState) lhv1beta2.InstanceManager { + return lhv1beta2.InstanceManager{ + Spec: lhv1beta2.InstanceManagerSpec{NodeID: nodeID}, + Status: lhv1beta2.InstanceManagerStatus{CurrentState: state}, + } +} + +func TestInstanceManagerRunningOnNode(t *testing.T) { + const thisNode = "node-a" + + testMatrix := map[string]struct { + items []lhv1beta2.InstanceManager + expectReady bool + }{ + "running on this node": { + items: []lhv1beta2.InstanceManager{instanceManager(thisNode, lhv1beta2.InstanceManagerStateRunning)}, + expectReady: true, + }, + // The reported failure: the pod is still pulling its ~440 MB image, so + // the CR exists but cannot serve a volume yet. + "starting on this node": { + items: []lhv1beta2.InstanceManager{instanceManager(thisNode, lhv1beta2.InstanceManagerStateStarting)}, + expectReady: false, + }, + "error on this node": { + items: []lhv1beta2.InstanceManager{instanceManager(thisNode, lhv1beta2.InstanceManagerStateError)}, + expectReady: false, + }, + "running only on another node": { + items: []lhv1beta2.InstanceManager{instanceManager("node-b", lhv1beta2.InstanceManagerStateRunning)}, + expectReady: false, + }, + "another node running, this one starting": { + items: []lhv1beta2.InstanceManager{ + instanceManager("node-b", lhv1beta2.InstanceManagerStateRunning), + instanceManager(thisNode, lhv1beta2.InstanceManagerStateStarting), + }, + expectReady: false, + }, + "several on this node, one running": { + items: []lhv1beta2.InstanceManager{ + instanceManager(thisNode, lhv1beta2.InstanceManagerStateStopped), + instanceManager(thisNode, lhv1beta2.InstanceManagerStateRunning), + }, + expectReady: true, + }, + "none at all": { + items: nil, + expectReady: false, + }, + } + + for name, test := range testMatrix { + t.Run(name, func(t *testing.T) { + ready, err := instanceManagerRunningOnNode(context.Background(), + fakeInstanceManagerLister{items: test.items}, thisNode) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if ready != test.expectReady { + t.Errorf("ready = %v, want %v", ready, test.expectReady) + } + }) + } +} + +func TestInstanceManagerRunningOnNodeListError(t *testing.T) { + listErr := errors.New("api unreachable") + ready, err := instanceManagerRunningOnNode(context.Background(), + fakeInstanceManagerLister{err: listErr}, "node-a") + if !errors.Is(err, listErr) { + t.Errorf("err = %v, want %v", err, listErr) + } + if ready { + t.Error("ready = true on list error, want false") + } +} + +// The instance-manager pull is the slowest thing in the readiness predicate, so +// adding it to checkLonghornReady must come with a budget that can absorb it -- +// the two-disk ZFS leg regressed on the un-widened 20m deadline. +func TestComponentsReadyTimeout(t *testing.T) { + withLH := componentsReadyTimeout(WaitForKubernetesOptions{WaitForLonghorn: true}) + withoutLH := componentsReadyTimeout(WaitForKubernetesOptions{}) + if withLH <= withoutLH { + t.Errorf("longhorn timeout %v must exceed base %v", withLH, withoutLH) + } + if withLH < 30*time.Minute { + t.Errorf("longhorn timeout %v too small for a 20m+ instance-manager pull", withLH) + } +} + +const imTestNode = "im-test-node" + +func imDaemonset(name string) *appsv1.DaemonSet { + return &appsv1.DaemonSet{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: longhornNamespace}, + Spec: appsv1.DaemonSetSpec{ + Template: corev1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{"app": name}}, + }, + }, + } +} + +func imPod(dsName string) *corev1.Pod { + return &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: dsName + "-pod", + Namespace: longhornNamespace, + Labels: map[string]string{"app": dsName}, + }, + Spec: corev1.PodSpec{NodeName: imTestNode}, + Status: corev1.PodStatus{ + Phase: corev1.PodRunning, + ContainerStatuses: []corev1.ContainerStatus{{Ready: true}}, + }, + } +} + +func imHealthyDaemonsets() []runtime.Object { + names := []string{"longhorn-manager", "longhorn-csi-plugin", "engine-image-ei-abcdef12"} + objs := make([]runtime.Object, 0, len(names)*2) + for _, n := range names { + objs = append(objs, imDaemonset(n), imPod(n)) + } + return objs +} + +// Healthy DaemonSets alone must not make the node ready: the instance-manager +// gate runs afterwards and its verdict is what checkLonghornReady returns. +func TestCheckLonghornReadyAppliesInstanceManagerGate(t *testing.T) { + gateErr := errors.New("longhorn instance-manager not running on node") + + testMatrix := map[string]struct { + gate func(context.Context, string) error + expectErr error + }{ + "gate satisfied": { + gate: func(context.Context, string) error { return nil }, + expectErr: nil, + }, + "gate unsatisfied": { + gate: func(context.Context, string) error { return gateErr }, + expectErr: gateErr, + }, + } + + for name, test := range testMatrix { + t.Run(name, func(t *testing.T) { + saved := instanceManagerReady + t.Cleanup(func() { instanceManagerReady = saved }) + instanceManagerReady = test.gate + + client := fake.NewSimpleClientset(imHealthyDaemonsets()...) + err := checkLonghornReady(client, imTestNode) + if !errors.Is(err, test.expectErr) { + t.Errorf("err = %v, want %v", err, test.expectErr) + } + }) + } +}