diff --git a/pkg/satellite/reconciler.go b/pkg/satellite/reconciler.go index c41de5cb..7be70c05 100644 --- a/pkg/satellite/reconciler.go +++ b/pkg/satellite/reconciler.go @@ -60,6 +60,13 @@ type ReconcilerConfig struct { // StateDir is where `.res` files land. Required when Adm is set. StateDir string + // DeviceWaitTimeout/DeviceWaitPoll bound waitForBlockDevice, which + // closes the zfs-create/udev race before a freshly created zvol is + // consumed. Zero values fall back to deviceWaitTimeoutDefault / + // deviceWaitPollDefault; tests shrink them to keep the wait loop fast. + DeviceWaitTimeout time.Duration + DeviceWaitPoll time.Duration + // NodeName is this satellite's identifier; the reconciler uses it // to know which Peer entries describe local vs. remote. NodeName string @@ -1510,6 +1517,24 @@ func (r *Reconciler) applyStorage(ctx context.Context, dr *intent.DesiredResourc } devices[vol.GetVolumeNumber()] = status.DevicePath + + // zfs-create/udev race: `zfs create -V` returns before udev has + // finished creating the /dev/zvol// symlink, so the + // device path reported by VolumeStatus may not exist on disk yet. + // The storage-only (local SC) path is the most exposed: it runs + // runStorageOnlyMkfs -> runAutoMkfs immediately, with no DRBD + // setup in between to give udev time to settle, so it raced the + // symlink and failed with "mkfs.ext4 ...: The file ... does not + // exist and no size was specified." The DRBD path can lose the + // same race when drbdadm attaches the backing device. Block here + // until the device is usable so every downstream consumer (mkfs, + // DRBD attach, LUKS) sees a ready node; re-queue the resource if + // it never appears rather than acting on a missing device. + err = r.waitForBlockDevice(ctx, status.DevicePath) + if err != nil { + return nil, false, false, errors.Wrapf(err, + "wait for backing device %s/%d", dr.GetName(), vol.GetVolumeNumber()) + } } if len(dr.GetVolumes()) > 0 { @@ -1519,6 +1544,65 @@ func (r *Reconciler) applyStorage(ctx context.Context, dr *intent.DesiredResourc return devices, resized, cloned, nil } +// deviceWaitTimeoutDefault/deviceWaitPollDefault back the same-named +// ReconcilerConfig knobs when those are left zero (the production path). +const ( + deviceWaitTimeoutDefault = 60 * time.Second + deviceWaitPollDefault = 500 * time.Millisecond +) + +// waitForBlockDevice blocks until `device` is a usable block device or the +// bounded deadline elapses. It exists to close the zfs-create/udev race +// described at the applyStorage call site: after `zfs create -V` the +// /dev/zvol// symlink is created asynchronously by udev, and +// callers that consume the path immediately (notably the storage-only mkfs +// path) would otherwise act on a node that does not exist yet. +// +// `blockdev --getsize64` is the authoritative readiness probe — it opens +// the device and so errors on a missing or dangling node. Best-effort +// `udevadm settle` between attempts nudges udev to drain its event queue; +// it is intentionally ignored on error (it may be absent in unit tests or +// already idle). An empty device path (diskless replica) or a nil Exec is +// a no-op. On timeout it returns an error so the caller re-queues the +// resource instead of mkfs-ing / attaching a path that is not there. +func (r *Reconciler) waitForBlockDevice(ctx context.Context, device string) error { + if device == "" || r.cfg.Exec == nil { + return nil + } + + timeout := r.cfg.DeviceWaitTimeout + if timeout <= 0 { + timeout = deviceWaitTimeoutDefault + } + + poll := r.cfg.DeviceWaitPoll + if poll <= 0 { + poll = deviceWaitPollDefault + } + + deadline := time.Now().Add(timeout) + + for { + _, err := r.cfg.Exec.Run(ctx, "blockdev", "--getsize64", device) + if err == nil { + return nil + } + + if time.Now().After(deadline) { + return errors.Errorf("block device %s did not appear within %s "+ + "(zfs create -> udev /dev/zvol symlink race)", device, timeout) + } + + _, _ = r.cfg.Exec.Run(ctx, "udevadm", "settle", "--timeout=5") + + select { + case <-ctx.Done(): + return errors.Wrap(ctx.Err(), "wait for block device") + case <-time.After(poll): + } + } +} + // materializeVolume picks the right provider call: clone from a // snapshot when SourceSnapshot is set on the desired volume, // otherwise create blank. Parses `:` for the diff --git a/pkg/satellite/reconciler_zvol_device_wait_test.go b/pkg/satellite/reconciler_zvol_device_wait_test.go new file mode 100644 index 00000000..2e8a1623 --- /dev/null +++ b/pkg/satellite/reconciler_zvol_device_wait_test.go @@ -0,0 +1,157 @@ +// SPDX-License-Identifier: Apache-2.0 + +package satellite + +import ( + "context" + "errors" + "strings" + "testing" + "time" + + intent "github.com/cozystack/blockstor/pkg/satellite/intent" + "github.com/cozystack/blockstor/pkg/storage" +) + +// errFakeNoDevice is the canned "backing device not present yet" error the +// fake exec returns for blockdev probes before the (simulated) udev +// /dev/zvol symlink appears. +var errFakeNoDevice = errors.New("blockdev: cannot open device: No such file or directory") + +// zvolWaitFakeProvider is a no-op storage.Provider whose VolumeStatus +// reports a fixed device path. It lets the applyStorage device-wait tests +// drive the post-`zfs create` udev race without a real ZFS/LVM backend. +// UsableKib is large so applyStorage never takes the resize branch. +type zvolWaitFakeProvider struct{ dev string } + +func (zvolWaitFakeProvider) Kind() string { return ProviderKindZFSThin } + +func (zvolWaitFakeProvider) PoolStatus(context.Context) (storage.PoolStatus, error) { + return storage.PoolStatus{}, nil +} + +func (zvolWaitFakeProvider) CreateVolume(context.Context, storage.Volume) error { return nil } +func (zvolWaitFakeProvider) DeleteVolume(context.Context, storage.Volume) error { return nil } +func (zvolWaitFakeProvider) ResizeVolume(context.Context, storage.Volume) error { return nil } + +func (p zvolWaitFakeProvider) VolumeStatus(context.Context, storage.Volume) (storage.VolumeStatus, error) { + return storage.VolumeStatus{DevicePath: p.dev, UsableKib: 1 << 30, State: "PROVISIONED"}, nil +} + +func (zvolWaitFakeProvider) CreateSnapshot(context.Context, storage.Snapshot) error { return nil } +func (zvolWaitFakeProvider) DeleteSnapshot(context.Context, storage.Snapshot) error { return nil } + +func (zvolWaitFakeProvider) RestoreVolumeFromSnapshot( + context.Context, storage.Volume, storage.Snapshot, +) error { + return nil +} + +// countCalls counts how many recorded FakeExec calls match name + args +// exactly. Safe to read fx.Calls directly here: applyStorage runs +// synchronously in the test goroutine, no product goroutine is live. +func countCalls(fx *storage.FakeExec, name string, args ...string) int { + want := strings.Join(args, " ") + + n := 0 + for _, c := range fx.Calls { + if c.Name == name && strings.Join(c.Args, " ") == want { + n++ + } + } + + return n +} + +// TestApplyStorageWaitsForZvolDevice is the regression guard for the +// zfs-create/udev race: `zfs create -V` returns before udev has created +// the /dev/zvol// symlink, so the storage-only (local SC) path +// used to run mkfs immediately on a device that did not exist yet +// ("mkfs.ext4 ...: The file ... does not exist"). applyStorage must now +// poll `blockdev --getsize64` until the backing device is usable before +// returning it to the mkfs / DRBD-attach consumers. +// +// On the pre-fix code applyStorage never probes the device, so the +// blockdev call count is 0 and this test fails — exactly the fail-on-bug +// property the fix must carry. +func TestApplyStorageWaitsForZvolDevice(t *testing.T) { + t.Parallel() + + const dev = "/dev/zvol/data/pvc-zvolwait_00000" + + fx := storage.NewFakeExec() + // blockdev fails twice (udev has not made the symlink yet), then the + // device appears and it succeeds. Later calls fall through to fx + // (empty success). The sequencedFakeExec helper still records every + // forwarded call in fx.Calls for the count assertion below. + seq := &sequencedFakeExec{ + fx: fx, + seqKey: "blockdev --getsize64 " + dev, + seqResps: []storage.FakeResponse{ + {Err: errFakeNoDevice}, + {Err: errFakeNoDevice}, + {Stdout: []byte("1073741824\n")}, + }, + } + + rec := NewReconciler(ReconcilerConfig{ + Providers: map[string]storage.Provider{"p1": zvolWaitFakeProvider{dev: dev}}, + Exec: seq, + DeviceWaitPoll: time.Millisecond, + DeviceWaitTimeout: 5 * time.Second, + }) + + dr := &intent.DesiredResource{ + Name: "pvc-zvolwait", + Volumes: []*intent.DesiredVolume{{VolumeNumber: 0, SizeKib: 1024, StoragePool: "p1"}}, + } + + devices, _, _, err := rec.applyStorage(context.Background(), dr) + if err != nil { + t.Fatalf("applyStorage: unexpected error: %v", err) + } + + if devices[0] != dev { + t.Fatalf("device path: got %q, want %q", devices[0], dev) + } + + got := countCalls(fx, "blockdev", "--getsize64", dev) + if got < 3 { + t.Fatalf("blockdev --getsize64 calls: got %d, want >=3 "+ + "(applyStorage must poll for the backing device before returning)", got) + } +} + +// TestApplyStorageFailsWhenZvolDeviceNeverAppears pins the timeout side of +// the wait: if the backing device never materialises, applyStorage must +// surface an error so the resource is re-queued rather than mkfs-ing / +// attaching a path that is not there. +func TestApplyStorageFailsWhenZvolDeviceNeverAppears(t *testing.T) { + t.Parallel() + + const dev = "/dev/zvol/data/pvc-never_00000" + + fx := storage.NewFakeExec() + fx.Expect("blockdev --getsize64 "+dev, storage.FakeResponse{Err: errFakeNoDevice}) + + rec := NewReconciler(ReconcilerConfig{ + Providers: map[string]storage.Provider{"p1": zvolWaitFakeProvider{dev: dev}}, + Exec: fx, + DeviceWaitPoll: time.Millisecond, + DeviceWaitTimeout: 50 * time.Millisecond, + }) + + dr := &intent.DesiredResource{ + Name: "pvc-never", + Volumes: []*intent.DesiredVolume{{VolumeNumber: 0, SizeKib: 1024, StoragePool: "p1"}}, + } + + _, _, _, err := rec.applyStorage(context.Background(), dr) //nolint:dogsled // only the returned error is relevant here + if err == nil { + t.Fatal("applyStorage: want error when the backing device never appears, got nil") + } + + if !strings.Contains(err.Error(), "did not appear") { + t.Fatalf("error: got %v, want it to report the device never appeared", err) + } +} diff --git a/pkg/store/k8s/k8s.go b/pkg/store/k8s/k8s.go index 42ef7688..46083c57 100644 --- a/pkg/store/k8s/k8s.go +++ b/pkg/store/k8s/k8s.go @@ -79,7 +79,7 @@ func NewWithAPIReader(c ctrlclient.Client, apiReader ctrlclient.Reader) *Store { s.nodes = &nodes{c: c} s.storagePools = &storagePools{c: c} s.resourceGroups = &resourceGroups{c: c} - s.resourceDefinitions = &resourceDefinitions{c: c} + s.resourceDefinitions = &resourceDefinitions{c: c, apiReader: apiReader} s.resources = &resources{c: c} s.volumeDefinitions = &volumeDefinitions{c: c, apiReader: apiReader} s.snapshots = &snapshots{c: c} diff --git a/pkg/store/k8s/resource_definitions.go b/pkg/store/k8s/resource_definitions.go index d5191cb0..9e19e57a 100644 --- a/pkg/store/k8s/resource_definitions.go +++ b/pkg/store/k8s/resource_definitions.go @@ -37,6 +37,17 @@ import ( type resourceDefinitions struct { c ctrlclient.Client + + // apiReader is a direct, UNCACHED reader (mgr.GetAPIReader()). The + // cached client's Get trails a just-committed RD create by the + // informer watch round-trip, so a GET /v1/resource-definitions/{rd} + // that load-balances to a replica whose cache has not yet observed + // the create returns 404 while the RD already exists in the API + // server. linstor-csi then fails ControllerPublishVolume with a + // transient 404 and backs off, slowing bulk attach. On a cache + // NotFound we re-read live before concluding the RD is absent. nil in + // non-production stores (in-memory / unit) → keep the cached result. + apiReader ctrlclient.Reader } func (s *resourceDefinitions) List(ctx context.Context) ([]apiv1.ResourceDefinition, error) { @@ -61,15 +72,18 @@ func (s *resourceDefinitions) Get(ctx context.Context, name string) (apiv1.Resou var crd crdv1alpha1.ResourceDefinition err := s.c.Get(ctx, types.NamespacedName{Name: Name(name)}, &crd) - if err != nil { - if apierrors.IsNotFound(err) { - return apiv1.ResourceDefinition{}, errors.Wrapf(store.ErrNotFound, "resource definition %q", name) - } + if err == nil { + return crdToWireRD(&crd), nil + } + if !apierrors.IsNotFound(err) { return apiv1.ResourceDefinition{}, errors.Wrapf(err, "get ResourceDefinition %q", name) } - return crdToWireRD(&crd), nil + // Cache NotFound: re-read live before concluding the RD is absent, so a + // just-created RD not yet in this replica's informer cache is not + // reported as a spurious 404 to linstor-csi (see the apiReader doc). + return s.getUncached(ctx, name) } func (s *resourceDefinitions) Create(ctx context.Context, in *apiv1.ResourceDefinition) error { @@ -289,6 +303,27 @@ func (s *resourceDefinitions) Delete(ctx context.Context, name string) error { return nil } +// getUncached resolves a cache-miss RD Get against the direct API reader. +// nil apiReader (unit/in-memory stores) keeps the cached NotFound. +func (s *resourceDefinitions) getUncached(ctx context.Context, name string) (apiv1.ResourceDefinition, error) { + if s.apiReader == nil { + return apiv1.ResourceDefinition{}, errors.Wrapf(store.ErrNotFound, "resource definition %q", name) + } + + var crd crdv1alpha1.ResourceDefinition + + err := s.apiReader.Get(ctx, types.NamespacedName{Name: Name(name)}, &crd) + if err == nil { + return crdToWireRD(&crd), nil + } + + if apierrors.IsNotFound(err) { + return apiv1.ResourceDefinition{}, errors.Wrapf(store.ErrNotFound, "resource definition %q", name) + } + + return apiv1.ResourceDefinition{}, errors.Wrapf(err, "get ResourceDefinition %q (uncached)", name) +} + func crdToWireRD(crd *crdv1alpha1.ResourceDefinition) apiv1.ResourceDefinition { // Re-emit typed DRBDOptions + ExtraProps back as a flat Props // bag so golinstor (which only knows the upstream diff --git a/pkg/store/k8s/resource_definitions_rd404_internal_test.go b/pkg/store/k8s/resource_definitions_rd404_internal_test.go new file mode 100644 index 00000000..2fd7e8f6 --- /dev/null +++ b/pkg/store/k8s/resource_definitions_rd404_internal_test.go @@ -0,0 +1,91 @@ +// SPDX-License-Identifier: Apache-2.0 + +/* +Copyright 2026 Cozystack contributors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package k8s + +import ( + "context" + "errors" + "testing" + + "k8s.io/apimachinery/pkg/runtime" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + crdv1alpha1 "github.com/cozystack/blockstor/api/v1alpha1" + apiv1 "github.com/cozystack/blockstor/pkg/api/v1" + "github.com/cozystack/blockstor/pkg/store" +) + +// TestResourceDefinitionGetCacheMissFallsBackToAPIReader pins the RD-404 +// attach race. The apiserver runs multi-replica with no leader election, so a +// GET /v1/resource-definitions/{rd} can load-balance to a replica whose +// informer cache has not yet observed a just-committed RD create. Before the +// fix resourceDefinitions.Get read only the cached client and returned a +// spurious store.ErrNotFound, which the REST layer surfaces as a 404 to +// linstor-csi — ControllerPublishVolume then fails ("404 Not Found") and backs +// off, slowing bulk attach of freshly-provisioned PVCs. Get must fall back to +// the direct (uncached) API reader and resolve the RD that already exists. +func TestResourceDefinitionGetCacheMissFallsBackToAPIReader(t *testing.T) { + t.Parallel() + + scheme := runtime.NewScheme() + if err := crdv1alpha1.AddToScheme(scheme); err != nil { + t.Fatalf("AddToScheme: %v", err) + } + + const name = "pvc-rd404" + + crd := wireToCRDRD(&apiv1.ResourceDefinition{Name: name}) + + // cached: this replica's informer cache trails the create — RD absent. + cached := fake.NewClientBuilder().WithScheme(scheme).Build() + // apiReader: the direct, uncached read sees the committed RD. + apiReader := fake.NewClientBuilder().WithScheme(scheme).WithObjects(crd).Build() + + rd := &resourceDefinitions{c: cached, apiReader: apiReader} + + got, err := rd.Get(context.Background(), name) + if err != nil { + t.Fatalf("Get with apiReader fallback: unexpected error: %v", err) + } + + if got.Name != name { + t.Fatalf("got RD name %q, want %q", got.Name, name) + } +} + +// TestResourceDefinitionGetNilAPIReaderKeepsCachedNotFound pins that the +// fallback is gated on a non-nil apiReader: in-memory / unit stores (which +// have no informer and pass a nil reader) must preserve the cached NotFound +// rather than dereference a nil reader. +func TestResourceDefinitionGetNilAPIReaderKeepsCachedNotFound(t *testing.T) { + t.Parallel() + + scheme := runtime.NewScheme() + if err := crdv1alpha1.AddToScheme(scheme); err != nil { + t.Fatalf("AddToScheme: %v", err) + } + + cached := fake.NewClientBuilder().WithScheme(scheme).Build() + + rd := &resourceDefinitions{c: cached} + + if _, err := rd.Get(context.Background(), "pvc-absent"); !errors.Is(err, store.ErrNotFound) { + t.Fatalf("Get with nil apiReader: got %v, want store.ErrNotFound", err) + } +}