diff --git a/pkg/dispatcher/dispatcher.go b/pkg/dispatcher/dispatcher.go index e732c7eb..64f7a235 100644 --- a/pkg/dispatcher/dispatcher.go +++ b/pkg/dispatcher/dispatcher.go @@ -955,6 +955,7 @@ func buildVolumes(rd *blockstoriov1alpha1.ResourceDefinition, target *blockstori SizeKib: vd.SizeKib, StoragePool: pool, SeedFromGI: seedFromGI(target, vd.VolumeNumber), + ObservedGI: observedGI(target, vd.VolumeNumber), SourceSnapshot: srcSnapshot, MetaPool: metaPool, Minor: minor, @@ -1014,6 +1015,18 @@ func seedFromGI(target *blockstoriov1alpha1.Resource, volumeNumber int32) string return "" } +// observedGI is the current GI the satellite last reported for the given +// volume of target, or "" when it never reported one. +func observedGI(target *blockstoriov1alpha1.Resource, volumeNumber int32) string { + for i := range target.Status.Volumes { + if target.Status.Volumes[i].VolumeNumber == volumeNumber { + return target.Status.Volumes[i].CurrentGI + } + } + + return "" +} + // kubeInternalNICName is the well-known NetInterface name carrying the // corev1.Node InternalIP (a host-routable address). Populated by the // register / label-sync path when blockstor knows the kube view of a diff --git a/pkg/dispatcher/observed_gi_test.go b/pkg/dispatcher/observed_gi_test.go new file mode 100644 index 00000000..95d03e5b --- /dev/null +++ b/pkg/dispatcher/observed_gi_test.go @@ -0,0 +1,53 @@ +// 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 dispatcher + +import ( + "testing" + + blockstoriov1alpha1 "github.com/cozystack/blockstor/api/v1alpha1" +) + +// The satellite tells its own crashed replica from a foreign superblock by +// the GI it last reported, so every desired volume carries it. +func TestBuildVolumesCarriesTheReportedGI(t *testing.T) { + t.Parallel() + + rd := &blockstoriov1alpha1.ResourceDefinition{} + rd.Spec.VolumeDefinitions = []blockstoriov1alpha1.ResourceDefinitionVolume{ + {VolumeNumber: 0, SizeKib: 4096}, + {VolumeNumber: 1, SizeKib: 4096}, + } + + target := &blockstoriov1alpha1.Resource{} + target.Status.Volumes = []blockstoriov1alpha1.ResourceVolumeStatus{{VolumeNumber: 0, CurrentGI: "ABCDEF0123456789"}} + + vols := buildVolumes(rd, target) + if len(vols) != 2 { + t.Fatalf("got %d volumes, want 2", len(vols)) + } + + if vols[0].GetObservedGI() != "ABCDEF0123456789" { + t.Errorf("volume 0 carries %q, want the reported GI", vols[0].GetObservedGI()) + } + + if vols[1].GetObservedGI() != "" { + t.Errorf("volume 1 never reported a GI and carries %q", vols[1].GetObservedGI()) + } +} diff --git a/pkg/drbd/dumpmd.go b/pkg/drbd/dumpmd.go new file mode 100644 index 00000000..3fecc179 --- /dev/null +++ b/pkg/drbd/dumpmd.go @@ -0,0 +1,248 @@ +// 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 drbd + +import ( + "bufio" + "bytes" + "context" + "strings" + + "github.com/cockroachdb/errors" +) + +// MetadataUUIDs runs `drbdadm dump-md ` and returns every +// generation UUID the metadata carries: the current UUID, the bitmap UUID of +// every peer slot and the history UUIDs. `get-gi` shows only one peer slot, +// and when a Primary loses a peer the old current UUID moves into that +// peer's bitmap slot, so a check against get-gi alone misses it. +func (a *Adm) MetadataUUIDs(ctx context.Context, target string) ([]string, error) { + out, err := a.exec.Run(ctx, "drbdadm", "dump-md", target) + if err != nil { + return nil, errors.Wrapf(err, "drbdadm dump-md %s", target) + } + + uuids, err := ParseMetadataUUIDs(out) + if err != nil { + return nil, errors.Wrapf(err, "drbdadm dump-md %s", target) + } + + return uuids, nil +} + +// The dump carries the whole bitmap, one line per run or per few words, so +// the lines are short; the buffer still grows to dumpLineMax for a dump whose +// format changes. The output as a whole is held in memory: for a large, +// heavily dirty volume that is tens of megabytes, once per check. +const ( + dumpLineBuffer = 64 << 10 + dumpLineMax = 16 << 20 +) + +// ParseMetadataUUIDs extracts the generation UUIDs from a dump-md listing, +// upper-cased and without the 0x prefix; zero UUIDs are left out. +// +// drbdmeta prints them as statements and blocks (print_dump_gi in +// drbd-utils' drbdmeta.c). In the v09 format the current UUID and each peer +// slot's bitmap UUID are single statements, and the history is a block of +// several UUIDs to a line: +// +// current-uuid 0x9F00AA11BB22CC33; +// peer[1] { +// bitmap-uuid 0x6E3A5D0B2C1F4A96; +// } +// history-uuids { +// 0x1234567890ABCDEF; 0x0000000000000000; ... +// } +// +// The v08 format has a single block of four UUIDs, current, bitmap and two +// of history, then the flags. Nothing else is read: the bitmap words and the +// device UUID are not generation identifiers. +// +// A dump that names no current UUID, a v09 `current-uuid` statement that +// gives one or a v08 `uuid` block that leads with one, or that yields no +// UUID at all, is not one +// this parser can read: empty output, a keyword +// drbdmeta renamed, or a line the scanner gave up on. That is an error, never +// an empty answer, since the caller would read an empty answer as "this +// superblock holds none of our generations" and re-initialise it. +func ParseMetadataUUIDs(dump []byte) ([]string, error) { + var parsed uuidDump + + scanner := bufio.NewScanner(bytes.NewReader(dump)) + scanner.Buffer(make([]byte, 0, dumpLineBuffer), dumpLineMax) + + for scanner.Scan() { + parsed.line(strings.TrimSpace(scanner.Text())) + } + + err := scanner.Err() + if err != nil { + return nil, errors.Wrap(err, "read the dump-md output") + } + + if !parsed.current || len(parsed.out) == 0 { + return nil, ErrUnreadableMetadataDump + } + + return parsed.out, nil +} + +// uuidDump is what ParseMetadataUUIDs has read so far. +type uuidDump struct { + out []string + block bool + current bool + // v08Lead is set while the v08 block has not yet given its first + // statement, which is the current UUID. + v08Lead bool +} + +func (d *uuidDump) line(line string) { + if d.block { + body, _, closed := strings.Cut(line, "}") + d.blockText(body, closed) + + return + } + + key, rest, _ := strings.Cut(line, " ") + + switch key { + case dumpKeyCurrent, "bitmap-uuid": + before := len(d.out) + d.out = appendUUIDStatements(d.out, rest) + // A current statement counts only when it gave a UUID: one + // without the 0x prefix, not hex, or zero is unreadable too. + d.current = d.current || (key == dumpKeyCurrent && len(d.out) > before) + case "history-uuids", dumpKeyV08: + d.v08Lead = key == dumpKeyV08 + + body, opens := strings.CutPrefix(strings.TrimSpace(rest), "{") + if !opens { + d.blockText(rest, true) + + return + } + + body, _, closed := strings.Cut(body, "}") + d.blockText(body, closed) + } +} + +// blockText reads text of an open block, which closed ends. In the v08 +// block the first statement must be a readable UUID for the dump to name a +// current one. +func (d *uuidDump) blockText(text string, closed bool) { + if first, ok := firstStatement(text); ok && d.v08Lead { + _, readable := uuidStatement(first) + d.current = d.current || readable + d.v08Lead = false + } + + d.out = appendUUIDStatements(d.out, text) + d.block = !closed + d.v08Lead = d.v08Lead && d.block +} + +// The statements that name a current UUID: v09's own one, and v08's block +// that leads with it. +const ( + dumpKeyCurrent = "current-uuid" + dumpKeyV08 = "uuid" +) + +// ErrUnreadableMetadataDump reports dump-md output that names no current +// UUID, which this parser therefore cannot vouch for. +var ErrUnreadableMetadataDump = errors.New("dump-md printed no current UUID this parser recognises") + +// appendUUIDStatements adds every statement of text that is a bare non-zero +// hex UUID, such as `0x6E3A5D0B2C1F4A96;`. A keyed statement, such as v08's +// `flags 0x00000011;` inside the uuid block, is not one. +func appendUUIDStatements(out []string, text string) []string { + for statement := range strings.SplitSeq(text, ";") { + if uuid, ok := uuidStatement(statement); ok { + out = append(out, uuid) + } + } + + return out +} + +// uuidStatement answers the UUID a statement is, when it is a bare non-zero +// hex UUID. +func uuidStatement(statement string) (string, bool) { + digits, ok := strings.CutPrefix(strings.TrimSpace(statement), "0x") + if !ok || !isHex(digits) || strings.Trim(digits, "0") == "" { + return "", false + } + + return strings.ToUpper(digits), true +} + +// firstStatement answers the first non-empty statement of text, if it has +// one. +func firstStatement(text string) (string, bool) { + for statement := range strings.SplitSeq(text, ";") { + if trimmed := strings.TrimSpace(statement); trimmed != "" { + return trimmed, true + } + } + + return "", false +} + +func isHex(digits string) bool { + if digits == "" { + return false + } + + for _, c := range strings.ToLower(digits) { + if !strings.ContainsRune("0123456789abcdef", c) { + return false + } + } + + return true +} + +// SameGeneration compares two generation UUIDs the way DRBD does: the +// lowest bit marks a UUID issued while Primary and is cleared when the UUID +// moves into history, so it takes no part in the identity. +func SameGeneration(a, b string) bool { + return strings.EqualFold(clearPrimaryBit(a), clearPrimaryBit(b)) +} + +func clearPrimaryBit(uuid string) string { + uuid = strings.ToUpper(strings.TrimPrefix(strings.TrimPrefix(uuid, "0x"), "0X")) + if uuid == "" { + return uuid + } + + last := uuid[len(uuid)-1] + + const hex = "0123456789ABCDEF" + + idx := strings.IndexByte(hex, last) + if idx < 0 { + return uuid + } + + return uuid[:len(uuid)-1] + string(hex[idx&^1]) +} diff --git a/pkg/drbd/dumpmd_test.go b/pkg/drbd/dumpmd_test.go new file mode 100644 index 00000000..f76fdc77 --- /dev/null +++ b/pkg/drbd/dumpmd_test.go @@ -0,0 +1,179 @@ +// 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 drbd + +import ( + "slices" + "strings" + "testing" +) + +// The shape drbdmeta prints for a v09 volume (print_dump_gi in drbd-utils' +// drbdmeta.c): one statement per line for the current and the bitmap UUIDs, +// the history as a block of four to a line, then the sizes and the bitmap. +const dumpV09 = `# DRBD meta data dump +# 2026-10-24 12:00:00 +0000 [1792800000] +# node-1> drbdmeta 1000 v09 /dev/vg/pvc-r_00000 internal dump-md +# + +version "v09"; + +max-peers 2; +# md_size_sect 2048 +# md_offset 10737414144 +# al_offset 10737385472 +# bm_offset 10737057792 + +node-id 0; +current-uuid 0x9F00AA11BB22CC33; +flags 0x00000011; +members 0x0000000000000003; +features 0x0000000000000000; +peer[0] { + bitmap-index -1; + bitmap-uuid 0x0000000000000000; + bitmap-dagtag 0x0000000000000000; + flags 0x00000000; +} +peer[1] { + bitmap-index 0; + bitmap-uuid 0x6e3a5d0b2c1f4a96; + bitmap-dagtag 0x0000000000000000; + flags 0x00000005; # PEER_CONNECTED | PEER_HAVE_DATA +} +history-uuids { + 0x1234567890ABCDEF; 0x0000000000000000; 0x0000000000000000; 0x0000000000000000; + 0x0000000000000000; 0x0000000000000000; 0x0000000000000000; 0x00000000000000A0; +} +# al-extents 1237; +la-size-sect 20971520; +bm-byte-per-bit 4096; +device-uuid 0x5A5A5A5A5A5A5A5A; +la-peer-max-bio-size 1048576; +al-stripes 1; +al-stripe-size-4k 8; +# bm-bytes 327680; +bitmap[0] { + # at 0kB + 5120 times 0x0000000000000000; + 0xDEADBEEFDEADBEEF; 0x0000000000000000; 0x0000000000000000; 0x0000000000000000; +} +# bits-set[0] 64; +` + +const dumpV08 = `version "v08"; + +# md_size_sect 262144 +uuid { + 0x9F00AA11BB22CC33; 0x6E3A5D0B2C1F4A96; 0x1234567890ABCDEF; 0x0000000000000000; + flags 0x00000011; +} +la-size-sect 20971520; +bm { + 0xDEADBEEFDEADBEEF; +} +` + +func TestParseMetadataUUIDsReadsEverySlot(t *testing.T) { + t.Parallel() + + for _, tc := range []struct { + name string + dump string + want []string + }{ + { + name: "v09", + dump: dumpV09, + want: []string{"9F00AA11BB22CC33", "6E3A5D0B2C1F4A96", "1234567890ABCDEF", "00000000000000A0"}, + }, + { + name: "v08", + dump: dumpV08, + want: []string{"9F00AA11BB22CC33", "6E3A5D0B2C1F4A96", "1234567890ABCDEF"}, + }, + { + name: "history on one line", + dump: "current-uuid 0x9F00AA11BB22CC33;\nhistory-uuids { 0x1234567890ABCDEF; }\n" + + "bm {\n 0xDEADBEEFDEADBEEF;\n}\n", + want: []string{"9F00AA11BB22CC33", "1234567890ABCDEF"}, + }, + { + name: "v08 block on one line", + dump: "version \"v08\";\nuuid { 0x9F00AA11BB22CC33; 0x6E3A5D0B2C1F4A96; flags 0x00000011; }\n", + want: []string{"9F00AA11BB22CC33", "6E3A5D0B2C1F4A96"}, + }, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + got, err := ParseMetadataUUIDs([]byte(tc.dump)) + if err != nil { + t.Fatalf("parse: %v", err) + } + + if !slices.Equal(got, tc.want) { + t.Fatalf("got %v, want %v", got, tc.want) + } + }) + } +} + +// Output that names no current UUID is not read as "no generations": empty +// output, a renamed keyword and a line the scanner gives up on are errors. +func TestParseMetadataUUIDsRefusesWhatItCannotRead(t *testing.T) { + t.Parallel() + + for name, dump := range map[string]string{ + "empty": "", + "current without 0x": "version \"v09\";\ncurrent-uuid 6E3A5D0B2C1F4A96;\n", + "zero current": "version \"v09\";\ncurrent-uuid 0x0000000000000000;\n", + "unreadable current next to a bitmap UUID": "version \"v09\";\ncurrent-uuid 6E3A5D0B2C1F4A96;\n" + + "peer[1] {\n bitmap-uuid 0x1234567890ABCDEF;\n}\n", + "v08 zero current next to a bitmap UUID": "version \"v08\";\nuuid {\n" + + " 0x0000000000000000; 0x6E3A5D0B2C1F4A96; 0x1234567890ABCDEF; 0x0000000000000000;\n flags 0x00000011;\n}\n", + "v08 current without 0x": "version \"v08\";\nuuid {\n" + + " 9F00AA11BB22CC33; 0x6E3A5D0B2C1F4A96; 0x1234567890ABCDEF; 0x0000000000000000;\n flags 0x00000011;\n}\n", + "v08 block on one line with a zero current": "version \"v08\";\nuuid { 0x0000000000000000; 0x6E3A5D0B2C1F4A96; }\n", + "renamed keyword": "version \"v09\";\ncurrent-gi 0x9F00AA11BB22CC33;\npeer[1] {\n bitmap-uuid 0x6E3A5D0B2C1F4A96;\n}\n", + "line too long": "current-uuid 0x9F00AA11BB22CC33;\n" + strings.Repeat("x", dumpLineMax+1) + "\n", + } { + t.Run(name, func(t *testing.T) { + t.Parallel() + + if got, err := ParseMetadataUUIDs([]byte(dump)); err == nil { + t.Fatalf("parse = %v, nil; want an error", got) + } + }) + } +} + +// The lowest bit marks a UUID issued while Primary and is cleared when the +// UUID moves into history. +func TestSameGenerationIgnoresThePrimaryBit(t *testing.T) { + t.Parallel() + + if !SameGeneration("6E3A5D0B2C1F4A97", "6e3a5d0b2c1f4a96") { + t.Error("a UUID and its history form are not the same generation") + } + + if SameGeneration("6E3A5D0B2C1F4A97", "6E3A5D0B2C1F4A95") { + t.Error("two generations that differ above the Primary bit compare equal") + } +} diff --git a/pkg/satellite/controllers/manager.go b/pkg/satellite/controllers/manager.go index 34b8ee41..4fdc35ca 100644 --- a/pkg/satellite/controllers/manager.go +++ b/pkg/satellite/controllers/manager.go @@ -215,16 +215,7 @@ func ensureWiredDefaults(cfg *Config) { // budget — Scenario 5.34 added the third runnable and the // inline chain tipped over the limit. func addBackgroundRunnables(mgr manager.Manager, cfg *Config) error { - err := mgr.Add(&ObserverRunnable{ - Client: mgr.GetClient(), - Exec: cfg.Exec, - NodeName: cfg.NodeName, - // Phase 11.7: producer end of the observer-trigger channel. - // Fires on kernel-state changes (resource lifecycle, role, - // disk, conn, repl) so the ResourceReconciler wakes even - // when no apiserver write bumps Generation. - ReconcileTrigger: cfg.ReconcileTrigger, - }) + err := mgr.Add(newObserverRunnableFromConfig(mgr.GetClient(), cfg)) if err != nil { return errors.Wrap(err, "add ObserverRunnable") } diff --git a/pkg/satellite/controllers/observer.go b/pkg/satellite/controllers/observer.go index d5853728..30fa9e1d 100644 --- a/pkg/satellite/controllers/observer.go +++ b/pkg/satellite/controllers/observer.go @@ -657,6 +657,24 @@ const ( drbdQuorumYes = "yes" ) +// newObserverRunnableFromConfig builds the observer the manager runs, so the +// wiring from cfg (the GI record above all, which nothing else would notice +// missing) lives where a unit test can drive it. Production caller: +// manager.go addBackgroundRunnables. +func newObserverRunnableFromConfig(cli client.Client, cfg *Config) *ObserverRunnable { + return &ObserverRunnable{ + Client: cli, + Exec: cfg.Exec, + NodeName: cfg.NodeName, + RecordGI: cfg.Apply.RecordObservedGI, + // Phase 11.7: producer end of the observer-trigger channel. + // Fires on kernel-state changes (resource lifecycle, role, + // disk, conn, repl) so the ResourceReconciler wakes even + // when no apiserver write bumps Generation. + ReconcileTrigger: cfg.ReconcileTrigger, + } +} + // ObserverRunnable tails `drbdsetup events2` and writes the parsed // observations onto matching Resource CRDs' Status subresource // via SSA. Phase 10.6: replaces the retired gRPC @@ -677,6 +695,11 @@ type ObserverRunnable struct { // uses to route observations to the right CRD. NodeName string + // RecordGI keeps the GI of every data-bearing volume it is called + // with in the satellite's own record, which outlives the status + // field (satellite.Reconciler.RecordObservedGI). Nil records nothing. + RecordGI func(rdName string, vol int32, gi string) + // connCache holds the latest observed per-peer connection state // keyed by `/`. We apply the full snapshot per // resource on every connection event because SSA with the same @@ -1146,20 +1169,20 @@ func (o *ObserverRunnable) backfillCurrentGI(ctx context.Context, adm *drbd.Adm, for i := range ev.Volumes { vol := &ev.Volumes[i] - if vol.CurrentUUID != "" || vol.BackingDev == "" { - continue - } - if !isDataBearingDiskState(drbd.DiskState(vol.DiskState)) { continue } - current, err := adm.CurrentGI(ctx, ev.ResourceName, vol.VolumeNumber, vol.BackingDev) - if err != nil || current == "" { - continue + if vol.CurrentUUID == "" && vol.BackingDev != "" { + current, err := adm.CurrentGI(ctx, ev.ResourceName, vol.VolumeNumber, vol.BackingDev) + if err == nil { + vol.CurrentUUID = current + } } - vol.CurrentUUID = current + if vol.CurrentUUID != "" && o.RecordGI != nil { + o.RecordGI(ev.ResourceName, vol.VolumeNumber, vol.CurrentUUID) + } } } diff --git a/pkg/satellite/controllers/observer_record_gi_test.go b/pkg/satellite/controllers/observer_record_gi_test.go new file mode 100644 index 00000000..9c24afaa --- /dev/null +++ b/pkg/satellite/controllers/observer_record_gi_test.go @@ -0,0 +1,83 @@ +// 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 controllers + +import ( + "context" + "os" + "path/filepath" + "testing" + + "github.com/cozystack/blockstor/pkg/drbd" + "github.com/cozystack/blockstor/pkg/satellite" + "github.com/cozystack/blockstor/pkg/storage" +) + +// The satellite keeps its own record of the GIs its volumes carried, since +// the status field is dropped when a volume leaves a data-bearing state. The +// observer feeds it every GI it reads, backfilled or not, and nothing for a +// volume without data. +func TestObserverRecordsTheGIOfDataBearingVolumes(t *testing.T) { + t.Parallel() + + fx := storage.NewFakeExec() + fx.Expect("drbdmeta --force pvc-r/0 v09 /dev/vg/pvc-r_00000 internal get-gi --node-id 0", + storage.FakeResponse{Stdout: []byte("6E3A5D0B2C1F4A96:0:0:0\n")}) + + recorded := map[int32]string{} + o := &ObserverRunnable{RecordGI: func(_ string, vol int32, gi string) { recorded[vol] = gi }} + + ev := &observation{ResourceName: "pvc-r", Volumes: []volumeObservation{ + {VolumeNumber: 0, DiskState: "UpToDate", BackingDev: "/dev/vg/pvc-r_00000"}, + {VolumeNumber: 1, DiskState: "UpToDate", CurrentUUID: "1111111111111110"}, + {VolumeNumber: 2, DiskState: "Inconsistent", CurrentUUID: "2222222222222220"}, + }} + + o.backfillCurrentGI(context.Background(), drbd.NewAdm(fx), ev) + + if recorded[0] != "6E3A5D0B2C1F4A96" || recorded[1] != "1111111111111110" { + t.Errorf("recorded %v, want the backfilled and the reported GI", recorded) + } + + if _, ok := recorded[2]; ok { + t.Error("a volume without data had its GI recorded") + } +} + +// The observer the manager runs feeds the reconciler's GI record. Without +// that wiring the record stays empty and every crashed replica is +// initialised again, with no test or log line to say so. +func TestManagerObserverRecordsIntoTheReconciler(t *testing.T) { + t.Parallel() + + dir := t.TempDir() + cfg := &Config{Apply: satellite.NewReconciler(satellite.ReconcilerConfig{StateDir: dir})} + + o := newObserverRunnableFromConfig(nil, cfg) + if o.RecordGI == nil { + t.Fatal("the observer records no GIs") + } + + o.RecordGI("pvc-w", 0, "6e3a5d0b2c1f4a96") + + data, err := os.ReadFile(filepath.Join(dir, "pvc-w.gi-history")) + if err != nil || string(data) != "0 6E3A5D0B2C1F4A96\n" { + t.Fatalf("record = %q, %v; want the GI in the reconciler's history", data, err) + } +} diff --git a/pkg/satellite/createmd_test.go b/pkg/satellite/createmd_test.go index 4780cba1..4578449c 100644 --- a/pkg/satellite/createmd_test.go +++ b/pkg/satellite/createmd_test.go @@ -21,6 +21,7 @@ package satellite import ( "context" "errors" + "fmt" "io" "os" "path/filepath" @@ -600,23 +601,17 @@ func uncleanFixture( return rec, dr, devices, filepath.Join(dir, "pvc-unclean.md-created"), fx } -// A lower disk carrying an unclean superblock is initialised, on both -// activation arms. +// A lower disk carrying an unclean superblock and no reported GI is +// initialised, on both activation arms. // -// It is worth being explicit about what these two pin, because the -// obvious reading of the second one is wrong. An unapplied activity log -// does NOT mean "this volume is established": a volume added to a live -// resource arrives with firstActivation=false on a lower disk that is -// brand new, and one carved from a recycled pool extent carries -// whatever superblock the previous tenant left. Gating re-initialisation -// on firstActivation stalled exactly that shape — the e2e suite caught -// vol-1 never leaving Diskless while the kernel had it UpToDate. -// -// The cost is that a replica which crashed while Primary and writing -// looks the same to drbdmeta and gets re-initialised too. Separating -// them needs a per-volume record of "we have initialised this before", -// which does not exist yet; these tests pin the behaviour as it stands -// rather than the behaviour we want. +// An unapplied activity log does NOT mean "this volume is established": a +// volume added to a live resource arrives with firstActivation=false on a +// lower disk that is brand new, and one carved from a recycled pool extent +// carries whatever superblock the previous tenant left. Gating +// re-initialisation on firstActivation stalled exactly that shape — the e2e +// suite caught vol-1 never leaving Diskless while the kernel had it UpToDate. +// What tells an established replica apart is a GI it reported, see +// TestEnsureMetadataRepairsTheCrashOfAReplicaWithAReportedGI. func TestEnsureMetadataInitialisesAnUncleanLowerDisk(t *testing.T) { for _, firstActivation := range []bool{true, false} { rec, dr, devices, mdMarkerPath, fx := uncleanFixture(t) @@ -652,8 +647,7 @@ func TestEnsureMetadataInitialisesAnUncleanLowerDisk(t *testing.T) { // reaches established resources. // // The behaviour is the same as its ensureMetadata twin and pinned for the -// same reason — see TestEnsureMetadataInitialisesAnUncleanLowerDisk for why -// re-initialising here is what the platform does today, and what it costs. +// same reason — see TestEnsureMetadataInitialisesAnUncleanLowerDisk. func TestEnsurePerVolumeMetadataInitialisesAnUncleanLowerDisk(t *testing.T) { rec, dr, devices, _, fx := uncleanFixture(t) @@ -670,3 +664,405 @@ func TestEnsurePerVolumeMetadataInitialisesAnUncleanLowerDisk(t *testing.T) { t.Error("the volume was left without metadata") } + +const ( + uncleanReportedGI = "6E3A5D0B2C1F4A96" + uncleanDumpMD = "drbdadm dump-md pvc-unclean/0" + uncleanApplyAL = "drbdadm apply-al pvc-unclean/0" +) + +var errApplyALFailed = errors.New("drbdadm apply-al: exit status 10") + +func ranCommand(fx *storage.FakeExec, fragment string) bool { + for _, line := range fx.CommandLines() { + if strings.Contains(line, fragment) { + return true + } + } + + return false +} + +// dumpWithPeerBitmap is dump-md output, in the shape drbdmeta prints for +// v09, for local node-id 0 whose current UUID moved on after it lost peer 1: +// the reported GI sits only in peer 1's bitmap slot, which +// `get-gi --node-id 0` never shows. +func dumpWithPeerBitmap(current, peerBitmap string) []byte { + return dumpV09(current, peerBitmap, "0000000000000000") +} + +// dumpWithHistory is the same after a resync moved the GI out of the bitmap +// slot and into the history block. +func dumpWithHistory(current, history string) []byte { + return dumpV09(current, "0000000000000000", history) +} + +func dumpV09(current, peerBitmap, history string) []byte { + return []byte("# DRBD meta data dump\nversion \"v09\";\n\nmax-peers 7;\n" + + "node-id 0;\ncurrent-uuid 0x" + current + ";\nflags 0x00000011;\n" + + "members 0x0000000000000003;\nfeatures 0x0000000000000000;\n" + + "peer[0] {\n bitmap-index -1;\n bitmap-uuid 0x0000000000000000;\n" + + " bitmap-dagtag 0x0000000000000000;\n flags 0x00000000;\n}\n" + + "peer[1] {\n bitmap-index 0;\n bitmap-uuid 0x" + peerBitmap + ";\n" + + " bitmap-dagtag 0x0000000000000000;\n flags 0x00000005;\n}\n" + + "history-uuids {\n 0x" + history + "; 0x0000000000000000; 0x0000000000000000; 0x0000000000000000;\n}\n" + + "la-size-sect 2097152;\nbm-byte-per-bit 4096;\ndevice-uuid 0x5A5A5A5A5A5A5A5A;\n" + + "bitmap[0] {\n # at 0kB\n 64 times 0x0000000000000000;\n}\n") +} + +func runMetadataPath(t *testing.T, path string, rec *Reconciler, dr *intent.DesiredResource, devices map[int32]string, marker string) error { + t.Helper() + + if path == "ensureMetadata" { + return rec.ensureMetadata(context.Background(), dr, devices, marker, false) + } + + return rec.ensurePerVolumeMetadata(context.Background(), dr, devices, false) +} + +// A replica that went down while Primary and writing leaves its activity log +// open. With create-md --force over it, its GI and dirty bitmap were gone, +// and with them the only up-to-date copy when it was the last one. A volume +// with a known GI is repaired with apply-al instead, on both paths, and the +// GI is looked for in every slot: a Primary that lost a peer keeps its old +// current UUID only in that peer's bitmap slot, and the Primary bit of a UUID +// is not part of its identity. +func TestEnsureMetadataRepairsTheCrashOfAReplicaWithAKnownGI(t *testing.T) { + for _, path := range []string{"ensureMetadata", "ensurePerVolumeMetadata"} { + for slot, dump := range map[string][]byte{ + "peer bitmap": dumpWithPeerBitmap("9F00AA11BB22CC33", uncleanReportedGI), + "history": dumpWithHistory("9F00AA11BB22CC33", uncleanReportedGI), + } { + testRepairsTheCrash(t, path, slot, dump) + } + } +} + +func testRepairsTheCrash(t *testing.T, path, slot string, dump []byte) { + t.Helper() + + { + for name, record := range map[string]func(*Reconciler, *intent.DesiredResource){ + "status": func(_ *Reconciler, dr *intent.DesiredResource) { dr.Volumes[0].ObservedGI = uncleanReportedGI }, + "history": func(rec *Reconciler, _ *intent.DesiredResource) { + rec.RecordObservedGI("pvc-unclean", 0, "6E3A5D0B2C1F4A97") + }, + } { + t.Run(path+"/"+slot+"/"+name, func(t *testing.T) { + rec, dr, devices, mdMarkerPath, fx := uncleanFixture(t) + record(rec, dr) + + fx.Expect(uncleanApplyAL, storage.FakeResponse{}) + fx.ExpectSequence(uncleanDumpMD, + storage.FakeResponse{Err: errDumpMdUnclean}, + storage.FakeResponse{Stdout: dump}) + + if err := runMetadataPath(t, path, rec, dr, devices, mdMarkerPath); err != nil { + t.Fatalf("%s: %v", path, err) + } + + if !ranCommand(fx, "apply-al") { + t.Error("the activity log was not applied") + } + + if ranCommand(fx, "create-md") { + t.Error("create-md ran over a replica with a known GI; its GI and bitmap are gone") + } + }) + } + } +} + +// The status field is not the record: it is dropped whenever the volume +// leaves a data-bearing disk state. The satellite's own history survives +// that, and survives a restart. +func TestRecordedGIsSurviveARestart(t *testing.T) { + rec := uncleanReconciler(t) + rec.RecordObservedGI("pvc-unclean", 0, uncleanReportedGI) + rec.RecordObservedGI("pvc-unclean", 0, uncleanReportedGI) + + again := NewReconciler(ReconcilerConfig{StateDir: rec.cfg.StateDir}) + if got := mustRecordedGIs(t, again, "pvc-unclean", 0); len(got) != 1 || got[0] != uncleanReportedGI { + t.Fatalf("recorded GIs after a restart = %v, want [%s]", got, uncleanReportedGI) + } +} + +// A known GI is not enough on its own: the superblock under the dirty log +// has to name it, or it is not this replica's and gets metadata of its own. +func TestEnsureMetadataInitialisesAForeignSuperblockDespiteAKnownGI(t *testing.T) { + for _, path := range []string{"ensureMetadata", "ensurePerVolumeMetadata"} { + t.Run(path, func(t *testing.T) { + rec, dr, devices, mdMarkerPath, fx := uncleanFixture(t) + dr.Volumes[0].ObservedGI = uncleanReportedGI + + fx.Expect(uncleanApplyAL, storage.FakeResponse{}) + fx.ExpectSequence(uncleanDumpMD, + storage.FakeResponse{Err: errDumpMdUnclean}, + storage.FakeResponse{Stdout: dumpWithPeerBitmap("1111111111111111", "2222222222222222")}) + + if err := runMetadataPath(t, path, rec, dr, devices, mdMarkerPath); err != nil { + t.Fatalf("%s: %v", path, err) + } + + if !ranCommand(fx, "create-md") { + t.Error("a superblock that does not name the known GI was kept") + } + }) + } +} + +// A repair or a check that fails is not a licence to re-initialise, and +// not a licence to keep either: the known GI proves the volume carried +// data-bearing metadata, and a superblock that could not be read is not +// proven to be it. A dump the parser cannot read, empty or with a renamed +// keyword, is such a failure, not a foreign superblock. Both metadata paths +// answer the same, the per-volume one being what a rebooted node takes. +func TestEnsureMetadataFailsClosedWhenTheRepairFails(t *testing.T) { + afterApply := func(second storage.FakeResponse) func(*storage.FakeExec) { + return func(fx *storage.FakeExec) { + fx.Expect(uncleanApplyAL, storage.FakeResponse{}) + fx.ExpectSequence(uncleanDumpMD, storage.FakeResponse{Err: errDumpMdUnclean}, second) + } + } + + for _, path := range []string{"ensureMetadata", "ensurePerVolumeMetadata"} { + for name, setup := range map[string]func(*storage.FakeExec){ + "apply-al": func(fx *storage.FakeExec) { + fx.Expect(uncleanApplyAL, storage.FakeResponse{Err: errApplyALFailed}) + }, + "dump-md": afterApply(storage.FakeResponse{Err: errApplyALFailed}), + "empty dump": afterApply(storage.FakeResponse{Stdout: []byte("")}), + "renamed keyword": afterApply(storage.FakeResponse{Stdout: []byte("version \"v09\";\ncurrent-gi 0x" + uncleanReportedGI + ";\n")}), + "current without 0x": afterApply(storage.FakeResponse{ + Stdout: []byte("version \"v09\";\ncurrent-uuid " + uncleanReportedGI + ";\n"), + }), + "zero current": afterApply(storage.FakeResponse{Stdout: []byte("version \"v09\";\ncurrent-uuid 0x0000000000000000;\n")}), + } { + t.Run(path+"/"+name, func(t *testing.T) { + rec, dr, devices, mdMarkerPath, fx := uncleanFixture(t) + dr.Volumes[0].ObservedGI = uncleanReportedGI + setup(fx) + + if err := runMetadataPath(t, path, rec, dr, devices, mdMarkerPath); err == nil { + t.Fatalf("a failed %s was answered with success", name) + } + + if ranCommand(fx, "create-md") { + t.Errorf("create-md ran after %s failed on a replica with a known GI", name) + } + }) + } + } +} + +// External metadata takes the same check as internal: dump-md reads the +// meta disk from the .res wherever it lives, so a repaired volume is kept +// when its superblock names a known GI and initialised when it does not. +func TestEnsureMetadataChecksRepairedExternalMetadata(t *testing.T) { + for name, tc := range map[string]struct { + dump []byte + keep bool + }{ + "its own": {dump: dumpWithPeerBitmap("9F00AA11BB22CC33", uncleanReportedGI), keep: true}, + "a foreign": {dump: dumpWithPeerBitmap("1111111111111111", "2222222222222222")}, + } { + t.Run(name, func(t *testing.T) { + rec, dr, devices, mdMarkerPath, fx := uncleanFixture(t) + dr.Volumes[0].ObservedGI = uncleanReportedGI + dr.Volumes[0].MetaPool = "meta" + + fx.Expect(uncleanApplyAL, storage.FakeResponse{}) + fx.ExpectSequence(uncleanDumpMD, + storage.FakeResponse{Err: errDumpMdUnclean}, + storage.FakeResponse{Stdout: tc.dump}) + + if err := rec.ensureMetadata(context.Background(), dr, devices, mdMarkerPath, false); err != nil { + t.Fatalf("ensureMetadata: %v", err) + } + + if ranCommand(fx, "create-md") == tc.keep { + t.Errorf("create-md ran = %v over %s superblock under external metadata", !tc.keep, name) + } + }) + } +} + +// Deleting a resource takes its GI record with it, file and cache, so a +// replica created again under the name starts with no record and records +// its first GI even when it is the one the old replica last had. +func TestDeleteResourceForgetsTheGIRecord(t *testing.T) { + rec := uncleanReconciler(t) + rec.RecordObservedGI("pvc-unclean", 0, uncleanReportedGI) + + resp, err := rec.DeleteResource(context.Background(), &intent.DeleteResourceRequest{Name: "pvc-unclean"}) + if err != nil || !resp.GetOk() { + t.Fatalf("DeleteResource: %v, %+v", err, resp) + } + + if got := mustRecordedGIs(t, rec, "pvc-unclean", 0); len(got) != 0 { + t.Fatalf("the deleted resource kept its GI record: %v", got) + } + + rec.RecordObservedGI("pvc-unclean", 0, uncleanReportedGI) + + if got := mustRecordedGIs(t, rec, "pvc-unclean", 0); len(got) != 1 { + t.Fatalf("the re-created replica's first GI was not recorded: %v", got) + } +} + +// The record keeps the newest generations of a volume: a crash leaves the +// last few in the metadata, and those are the ones a check needs. +func TestRecordedGIsKeepTheNewest(t *testing.T) { + rec := uncleanReconciler(t) + + for i := range giHistoryPerVolume + 8 { + rec.RecordObservedGI("pvc-unclean", 0, fmt.Sprintf("%016X", i+1)) + } + + got := mustRecordedGIs(t, rec, "pvc-unclean", 0) + if len(got) != giHistoryPerVolume { + t.Fatalf("kept %d GIs, want %d", len(got), giHistoryPerVolume) + } + + if want := fmt.Sprintf("%016X", 9); got[0] != want { + t.Errorf("oldest kept = %s, want %s", got[0], want) + } + + if want := fmt.Sprintf("%016X", giHistoryPerVolume+8); got[len(got)-1] != want { + t.Errorf("newest kept = %s, want %s", got[len(got)-1], want) + } +} + +// The observer reports the same GI on every event; a repeat of the last one +// is answered from memory and does not rewrite the file. +func TestRecordObservedGISkipsARepeat(t *testing.T) { + rec := uncleanReconciler(t) + rec.RecordObservedGI("pvc-unclean", 0, uncleanReportedGI) + + if err := os.Remove(rec.giHistoryPath("pvc-unclean")); err != nil { + t.Fatalf("remove: %v", err) + } + + rec.RecordObservedGI("pvc-unclean", 0, uncleanReportedGI) + + if _, err := os.Stat(rec.giHistoryPath("pvc-unclean")); !os.IsNotExist(err) { + t.Errorf("a repeated GI rewrote the record: %v", err) + } +} + +func uncleanReconciler(t *testing.T) *Reconciler { + t.Helper() + + rec, _, _, _, _ := uncleanFixture(t) //nolint:dogsled // only the reconciler is needed here + + return rec +} + +// A volume removed from the definition takes its recorded generations +// with it, so one added again under the same number starts with no record; +// the volumes that stay keep theirs. +func TestRemovedVolumeLosesItsGIRecord(t *testing.T) { + rec := uncleanReconciler(t) + rec.RecordObservedGI("pvc-unclean", 0, uncleanReportedGI) + rec.RecordObservedGI("pvc-unclean", 1, "1111111111111110") + + rec.keepGIHistoryOf("pvc-unclean", []int32{0}) + + if got := mustRecordedGIs(t, rec, "pvc-unclean", 1); len(got) != 0 { + t.Errorf("removed volume 1 kept its record: %v", got) + } + + if got := mustRecordedGIs(t, rec, "pvc-unclean", 0); len(got) != 1 { + t.Errorf("volume 0 lost its record: %v", got) + } + + rec.RecordObservedGI("pvc-unclean", 1, "1111111111111110") + + if got := mustRecordedGIs(t, rec, "pvc-unclean", 1); len(got) != 1 { + t.Errorf("the re-added volume's first GI was not recorded: %v", got) + } +} + +func mustRecordedGIs(t *testing.T, rec *Reconciler, rdName string, vol int32) []string { + t.Helper() + + got, err := rec.recordedGIs(rdName, vol) + if err != nil { + t.Fatalf("read the record: %v", err) + } + + return got +} + +// A record that cannot be read is not "no record": the repair fails on it +// with no create-md, on both paths, and an observation does not write over +// it with its one GI. A damaged line counts as unreadable: skipping it +// could leave the volume with no record. +func TestUnreadableGIRecordFailsClosed(t *testing.T) { + for _, path := range []string{"ensureMetadata", "ensurePerVolumeMetadata"} { + for damage, content := range map[string]string{ + "directory": "", + "line with no GI": "0\n", + "volume not a number": "zero " + uncleanReportedGI + "\n", + } { + t.Run(path+"/"+damage, func(t *testing.T) { + rec, dr, devices, mdMarkerPath, fx := uncleanFixture(t) + record := rec.giHistoryPath("pvc-unclean") + + var err error + if damage == "directory" { + // A directory where the file should be: every read of it fails. + err = os.Mkdir(record, 0o700) + } else { + err = os.WriteFile(record, []byte(content), 0o600) + } + + if err != nil { + t.Fatalf("damage the record: %v", err) + } + + if err := runMetadataPath(t, path, rec, dr, devices, mdMarkerPath); err == nil { + t.Fatal("a repair over an unreadable record was answered with success") + } + + if ranCommand(fx, "create-md") { + t.Error("create-md ran over a volume whose record could not be read") + } + + rec.RecordObservedGI("pvc-unclean", 0, uncleanReportedGI) + + if damage == "directory" { + if info, err := os.Stat(record); err != nil || !info.IsDir() { + t.Errorf("an observation wrote over the unreadable record: %v", err) + } + + return + } + + if got, _ := os.ReadFile(record); string(got) != content { + t.Errorf("an observation wrote over the damaged record: %q", got) + } + }) + } + } +} + +// The apply drops the record of a volume the definition no longer has: the +// drop runs where the desired volumes are known, in applyOne. +func TestApplyDropsTheGIRecordOfARemovedVolume(t *testing.T) { + rec, dr, _, _, _ := uncleanFixture(t) //nolint:dogsled // only the reconciler and its resource are needed + rec.RecordObservedGI("pvc-unclean", 0, uncleanReportedGI) + rec.RecordObservedGI("pvc-unclean", 1, "1111111111111110") + + // dr has volume 0 only; the outcome of the rest of the apply does not + // matter here. + _, _ = rec.Apply(context.Background(), []*intent.DesiredResource{dr}) + + if got := mustRecordedGIs(t, rec, "pvc-unclean", 1); len(got) != 0 { + t.Errorf("volume 1, gone from the definition, kept its record: %v", got) + } + + if got := mustRecordedGIs(t, rec, "pvc-unclean", 0); len(got) != 1 { + t.Errorf("volume 0 lost its record: %v", got) + } +} diff --git a/pkg/satellite/gi_history.go b/pkg/satellite/gi_history.go new file mode 100644 index 00000000..2200a9a2 --- /dev/null +++ b/pkg/satellite/gi_history.go @@ -0,0 +1,295 @@ +// 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 satellite + +import ( + "bufio" + "bytes" + "errors" + "fmt" + "log/slog" + "os" + "path/filepath" + "slices" + "strconv" + "strings" + + intent "github.com/cozystack/blockstor/pkg/satellite/intent" +) + +// giHistorySuffix names the per-resource file, in the state directory, +// that records the generation identifiers this satellite has seen its +// replica's volumes carry. It is the per-volume record "this replica held +// established metadata" that repairUncleanMetadata decides on. The Resource +// status is not that record: its current GI is dropped whenever the volume +// leaves a data-bearing disk state, which is exactly when a crash leaves it. +const giHistorySuffix = ".gi-history" + +// giHistoryPerVolume bounds how many generations are kept per volume. A +// replica that crashed has its last few in the metadata, and those are the +// ones a check needs. +const giHistoryPerVolume = 32 + +// RecordObservedGI notes that volume vol of rdName carried gi. The observer +// calls it for every data-bearing volume it reads a GI of; repeated values +// cost a map lookup, not a write. +func (r *Reconciler) RecordObservedGI(rdName string, vol int32, generation string) { + if r.cfg.StateDir == "" || generation == "" { + return + } + + generation = strings.ToUpper(generation) + key := rdName + "/" + strconv.Itoa(int(vol)) + + r.mu.Lock() + defer r.mu.Unlock() + + if r.lastRecordedGI == nil { + r.lastRecordedGI = map[string]string{} + } + + if r.lastRecordedGI[key] == generation { + return + } + + history, err := r.readGIHistory(rdName) + if err != nil { + // Writing over a file that could not be read would replace the + // record with this one GI. + slog.Default().Warn("could not read a volume's generation record; not recording", "resource", rdName, + "volume", vol, "err", err) + + return + } + + if !slices.Contains(history[vol], generation) { + history[vol] = append(history[vol], generation) + if extra := len(history[vol]) - giHistoryPerVolume; extra > 0 { + history[vol] = history[vol][extra:] + } + + err = r.writeGIHistory(rdName, history) + if err != nil { + // Not fatal to the observation, but the record is what lets a + // crashed replica keep its metadata, so a state directory that + // cannot be written has to show. + slog.Default().Warn("could not record a volume's generation; a crash of this replica "+ + "would re-initialise its metadata", "resource", rdName, "volume", vol, "err", err) + + return + } + } + + r.lastRecordedGI[key] = generation +} + +// keepGIHistoryOf drops the record of every volume of rdName that is not in +// vols, the volumes the definition has now. A volume removed and added again +// under the same number is a new volume: it must not inherit the old one's +// generations, or a lower disk still carrying the old superblock would be +// kept rather than initialised. +func (r *Reconciler) keepGIHistoryOf(rdName string, vols []int32) { + if r.cfg.StateDir == "" { + return + } + + r.mu.Lock() + defer r.mu.Unlock() + + history, err := r.readGIHistory(rdName) + if err != nil { + slog.Default().Warn("could not read a generation record; keeping it as it is", "resource", rdName, "err", err) + + return + } + + dropped := false + + for vol := range history { + if !slices.Contains(vols, vol) { + delete(history, vol) + delete(r.lastRecordedGI, rdName+"/"+strconv.Itoa(int(vol))) + + dropped = true + } + } + + if !dropped { + return + } + + err = r.writeGIHistory(rdName, history) + if err != nil { + slog.Default().Warn("could not drop the generations of removed volumes", "resource", rdName, "err", err) + } +} + +// recordedGIs is what RecordObservedGI kept for volume vol of rdName. +func (r *Reconciler) recordedGIs(rdName string, vol int32) ([]string, error) { + r.mu.Lock() + defer r.mu.Unlock() + + history, err := r.readGIHistory(rdName) + if err != nil { + return nil, err + } + + return history[vol], nil +} + +func (r *Reconciler) giHistoryPath(rdName string) string { + return filepath.Join(r.cfg.StateDir, rdName+giHistorySuffix) +} + +// readGIHistory reads the file. A missing one is an empty history, which +// leaves the volume to be initialised as before; one that cannot be read is +// an error, never an empty history, since an empty one is what lets the +// repair re-initialise a volume that has a record. +// errDamagedGIRecord reports a generation record line this code did not +// write. +var errDamagedGIRecord = errors.New("damaged line") + +func (r *Reconciler) readGIHistory(rdName string) (map[int32][]string, error) { + out := map[int32][]string{} + + data, err := os.ReadFile(r.giHistoryPath(rdName)) + if os.IsNotExist(err) { + return out, nil + } + + if err != nil { + return nil, fmt.Errorf("read the generation record of %s: %w", rdName, err) + } + + // A line this code did not write means the file is damaged, and + // skipping it could leave a volume with no record: that fails too. + scanner := bufio.NewScanner(bytes.NewReader(data)) + for scanner.Scan() { + line := strings.TrimSpace(scanner.Text()) + if line == "" { + continue + } + + volField, generation, ok := strings.Cut(line, " ") + vol, convErr := strconv.ParseInt(volField, 10, 32) + + if !ok || generation == "" || convErr != nil { + return nil, fmt.Errorf("the generation record of %s: %w: %q", rdName, errDamagedGIRecord, line) + } + + out[int32(vol)] = append(out[int32(vol)], generation) + } + + err = scanner.Err() + if err != nil { + return nil, fmt.Errorf("read the generation record of %s: %w", rdName, err) + } + + return out, nil +} + +func (r *Reconciler) writeGIHistory(rdName string, history map[int32][]string) error { + vols := make([]int32, 0, len(history)) + for vol := range history { + vols = append(vols, vol) + } + + slices.Sort(vols) + + var buf bytes.Buffer + + for _, vol := range vols { + for _, gi := range history[vol] { + buf.WriteString(strconv.Itoa(int(vol)) + " " + gi + "\n") + } + } + + return writeFileDurably(r.giHistoryPath(rdName), buf.Bytes()) +} + +// forgetGIHistory drops the record of rdName and its cached last GIs, so a +// replica created again under the name starts with no record and its first +// GI is written rather than taken for one already on file. It holds the +// lock RecordObservedGI writes under, so a write already under way finishes +// first. An observation of the old replica processed after this one would +// still write a fresh file; by then the device is down and the observer has +// no GI left to read, so the window is an event already in its queue. +func (r *Reconciler) forgetGIHistory(rdName string) error { + r.mu.Lock() + defer r.mu.Unlock() + + prefix := rdName + "/" + for key := range r.lastRecordedGI { + if strings.HasPrefix(key, prefix) { + delete(r.lastRecordedGI, key) + } + } + + err := os.Remove(r.giHistoryPath(rdName)) + if err != nil && !os.IsNotExist(err) { + return err //nolint:wrapcheck // the path is in the error + } + + return nil +} + +// writeFileDurably replaces path with data so that a power loss leaves +// either the old content or the new: the record exists for a node that went +// down hard, and the GI it most needs is the one written just before. +func writeFileDurably(path string, data []byte) error { + tmp := path + ".tmp" + + err := writeAndSync(tmp, data) + if err != nil { + return errors.Join(err, os.Remove(tmp)) + } + + err = os.Rename(tmp, path) + if err != nil { + return errors.Join(err, os.Remove(tmp)) + } + + dir, err := os.Open(filepath.Dir(path)) + if err != nil { + return err //nolint:wrapcheck // best-effort record; the caller drops the error + } + + return errors.Join(dir.Sync(), dir.Close()) +} + +func writeAndSync(path string, data []byte) error { + file, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, resFilePerm) + if err != nil { + return err //nolint:wrapcheck // best-effort record; the caller drops the error + } + + _, err = file.Write(data) + + return errors.Join(err, file.Sync(), file.Close()) +} + +// desiredVolumeNumbers lists the volume numbers dr asks for. +func desiredVolumeNumbers(dr *intent.DesiredResource) []int32 { + out := make([]int32, 0, len(dr.GetVolumes())) + for _, vol := range dr.GetVolumes() { + out = append(out, vol.GetVolumeNumber()) + } + + return out +} diff --git a/pkg/satellite/intent/types.go b/pkg/satellite/intent/types.go index cf7d20d9..53a48529 100644 --- a/pkg/satellite/intent/types.go +++ b/pkg/satellite/intent/types.go @@ -382,6 +382,23 @@ type DesiredVolume struct { // file points at an already-existing block device. The path is // then propagated via the .res renderer's Volume.MetaDisk field. MetaPool string + + // ObservedGI is the current generation identifier the satellite last + // reported for this volume on this replica (Resource status), empty + // when none was ever reported. It is the per-volume record that this + // replica's lower disk carried established metadata: a dirty activity + // log on a volume with one is a crash to repair with apply-al, not a + // foreign superblock to re-initialise over. + ObservedGI string +} + +// GetObservedGI returns the GI last reported for this volume, or "". +func (x *DesiredVolume) GetObservedGI() string { + if x == nil { + return "" + } + + return x.ObservedGI } // GetVolumeNumber returns the per-RD volume index. diff --git a/pkg/satellite/reconciler.go b/pkg/satellite/reconciler.go index 8a005f29..853e5064 100644 --- a/pkg/satellite/reconciler.go +++ b/pkg/satellite/reconciler.go @@ -257,6 +257,10 @@ type Reconciler struct { // recovery by `grace` (default 30s). seenStuckAt map[string]time.Time + // lastRecordedGI is the last GI RecordObservedGI wrote per + // "/", so a steady GI is not rewritten every tick. + lastRecordedGI map[string]string + // restoreBlankFallback records, per "/", that a // snapshot-restore-backed volume could NOT be materialised from the // snapshot on THIS node and fell back to a blank CreateVolume (no local @@ -894,6 +898,12 @@ func (r *Reconciler) DeleteResource(ctx context.Context, req *intent.DeleteResou }, nil } } + + err := r.forgetGIHistory(req.GetName()) + if err != nil { + //nolint:nilerr // surfaced as ok=false; gRPC error reserved for transport faults + return &intent.DeleteResourceResponse{Ok: false, Message: err.Error()}, nil + } } r.forgetPool(req.GetName()) @@ -924,6 +934,8 @@ func (r *Reconciler) applyOne(ctx context.Context, dr *intent.DesiredResource) * diskless := isDiskless(dr.GetFlags()) + r.keepGIHistoryOf(dr.GetName(), desiredVolumeNumbers(dr)) + if isInactive(dr.GetFlags()) { r.applyInactive(ctx, dr, res) @@ -3542,38 +3554,12 @@ func (r *Reconciler) ensureMetadata(ctx context.Context, dr *intent.DesiredResou hasMD, probeErr := r.cfg.Adm.HasMD(ctx, target) - // A dirty activity log means the metadata is THERE, unreadable - // until it is applied — and this is where blockstor cannot yet - // tell two very different situations apart. - // - // A lower disk just materialised carries someone else's - // superblock: a ZFS clone or snapshot receive brings the - // source's, and a volume carved from a pool that has been used - // before brings whatever the previous tenant left. Both need - // metadata of their own, and re-initialising is the point of - // the step. But an established replica that went down while - // Primary and writing leaves an activity log that looks - // IDENTICAL to drbdmeta, and create-md --force there destroys - // its GI and dirty bitmap. - // - // firstActivation does not separate them: it is per RESOURCE, - // so adding a volume to a live resource arrives with - // firstActivation=false on a lower disk that is nonetheless - // brand new. Gating on it stalled exactly that — the e2e suite - // caught vol-1 never leaving Diskless while the kernel had it - // UpToDate, because the reconcile aborted before stamping - // status. - // - // So the pre-existing behaviour stands for now: treat it as - // "no usable metadata" and re-initialise. That is fail-open on - // the crash-recovery case, which is a real exposure and is NOT - // made better by this comment — it is recorded here because - // closing it needs a per-VOLUME record of "we have initialised - // this before", which does not exist today. That record, and - // the apply-al repair that should replace re-initialisation - // once it does, belong in their own change with their own e2e. + // A dirty activity log means the metadata is there and cannot be + // read until it is applied; see repairUncleanMetadata for how + // this replica's own crash is told apart from a superblock that + // is not its own. if errors.Is(probeErr, drbd.ErrMetadataUnclean) { - hasMD, probeErr = false, nil + hasMD, probeErr = r.repairUncleanMetadata(ctx, dr.GetName(), vol) } if probeErr != nil { @@ -3706,6 +3692,90 @@ func isInitialUpToDateWinner(dr *intent.DesiredResource, firstActivation bool) b !dr.GetPeerHasData() } +// repairUncleanMetadata answers a metadata probe that found a dirty activity +// log. It returns whether the volume now carries usable metadata of its own +// (true: keep it), or false for "initialise it", which is what the caller +// did for every dirty log before. +// +// drbdmeta words two different situations the same way. A replica that went +// down while Primary and writing left its own activity log open, and +// create-md --force over it destroys its generation identifiers and dirty +// bitmap; if it is the only up-to-date copy, that is the copy. A lower disk +// just materialised can carry somebody else's superblock: a ZFS clone or a +// snapshot receive brings the source's, and an extent reused from a pool +// brings the previous tenant's. That one needs metadata of its own. +// +// The record that separates them is the set of GIs this replica's volume was +// seen to carry: the satellite's own history file (RecordObservedGI), plus the +// GI in the Resource status. A volume with neither never carried established +// metadata here, so it is initialised as before: a clone, a receive, a volume +// added to a live resource, a replica created again. One with a record is +// repaired with apply-al and then checked: some UUID now in its metadata, the +// current one, a peer slot's bitmap UUID or a history UUID, has to be one it +// was seen with, or the superblock is not this replica's and it is +// initialised after all. A repair or a check that fails is an error, not a +// re-initialisation: the record proves there was data-bearing metadata here. +// +// External metadata takes the same check: dump-md finds the meta disk from +// the rendered .res, wherever it lives. Its record holds only the GI in the +// status for now: the observer reads a GI with get-gi, which assumes internal +// metadata, so the history file does not fill for such a volume. +func (r *Reconciler) repairUncleanMetadata( + ctx context.Context, rdName string, vol *intent.DesiredVolume, +) (bool, error) { + target := fmt.Sprintf("%s/%d", rdName, vol.GetVolumeNumber()) + logger := log.FromContext(ctx).WithValues("volume", target) + + known, err := r.recordedGIs(rdName, vol.GetVolumeNumber()) + if err != nil { + return false, err + } + + if observed := vol.GetObservedGI(); observed != "" { + known = append(known, observed) + } + + if len(known) == 0 { + logger.Info("dirty activity log on a volume with no recorded GI; initialising its metadata") + + return false, nil + } + + err = r.cfg.Adm.ApplyAL(ctx, target) + if err != nil { + return false, errors.Wrapf(err, "apply the activity log of %s, which this replica held established metadata on", target) + } + + onDisk, err := r.cfg.Adm.MetadataUUIDs(ctx, target) + if err != nil { + return false, errors.Wrapf(err, "read the metadata of %s after apply-al", target) + } + + if !anySameGeneration(onDisk, known) { + logger.Info("the superblock under a dirty activity log is not this replica's; initialising its metadata", + "recordedGIs", known, "onDisk", onDisk) + + return false, nil + } + + logger.Info("repaired the activity log left by an unclean shutdown") + + return true, nil +} + +// anySameGeneration reports whether some UUID in onDisk is one of known. +func anySameGeneration(onDisk, known []string) bool { + for _, a := range onDisk { + for _, b := range known { + if drbd.SameGeneration(a, b) { + return true + } + } + } + + return false +} + // ensurePerVolumeMetadata stamps DRBD-9 metadata on every diskful // volume of `dr` that lacks it. Mirrors upstream LINSTOR's // DrbdLayer.adjustResource (satellite/.../DrbdLayer.java L702-723): @@ -3767,15 +3837,11 @@ func (r *Reconciler) ensurePerVolumeMetadata(ctx context.Context, dr *intent.Des hasMD, probeErr := r.cfg.Adm.HasMD(ctx, target) - // Same open problem as in ensureMetadata, and the same - // resolution for now. This helper runs past first activation, - // which reads like "established replica" — but the volume it is - // stamping can be one just added to a live resource, on a lower - // disk carved from a recycled pool extent whose previous tenant - // left an unclean superblock behind. Refusing here stalled that - // path outright. + // The same repair as in ensureMetadata: a volume just added to + // a live resource has no reported GI and is initialised, an + // established one that crashed is repaired. if errors.Is(probeErr, drbd.ErrMetadataUnclean) { - hasMD, probeErr = false, nil + hasMD, probeErr = r.repairUncleanMetadata(ctx, dr.GetName(), vol) } if probeErr != nil { diff --git a/pkg/storage/exec.go b/pkg/storage/exec.go index e5c13799..3309cc4b 100644 --- a/pkg/storage/exec.go +++ b/pkg/storage/exec.go @@ -134,6 +134,11 @@ type FakeExec struct { // Responses maps "name arg1 arg2 …" → canned output. // Falling back to empty stdout + nil error if absent. Responses map[string]FakeResponse + + // Sequences maps a command line to responses handed out one per call, + // in order, before Responses is consulted. For a command whose answer + // changes between calls, such as dump-md before and after apply-al. + Sequences map[string][]FakeResponse } // FakeCall records one invocation of FakeExec.Run. @@ -153,6 +158,18 @@ func NewFakeExec() *FakeExec { return &FakeExec{Responses: map[string]FakeResponse{}} } +// ExpectSequence queues responses for cmdLine, one per call. +func (f *FakeExec) ExpectSequence(cmdLine string, responses ...FakeResponse) { + f.mu.Lock() + defer f.mu.Unlock() + + if f.Sequences == nil { + f.Sequences = map[string][]FakeResponse{} + } + + f.Sequences[cmdLine] = append(f.Sequences[cmdLine], responses...) +} + // Run looks up a pre-registered response (full command line) and falls // back to an empty success if none was registered. func (f *FakeExec) Run(ctx context.Context, name string, args ...string) ([]byte, error) { @@ -192,6 +209,12 @@ func (f *FakeExec) RunWithStdin( key += " " + strings.Join(args, " ") } + if queue := f.Sequences[key]; len(queue) > 0 { + f.Sequences[key] = queue[1:] + + return queue[0].Stdout, queue[0].Err + } + if resp, ok := f.Responses[key]; ok { return resp.Stdout, resp.Err }