From 5ee9a52322bf24710c0e2b888ef556563bdbb39d Mon Sep 17 00:00:00 2001 From: Nikolai Emil Damm Date: Tue, 10 Nov 2026 14:51:02 +0100 Subject: [PATCH 1/4] feat(storage): measure provider-backed volume usage --- .../workflows/prepare-longhorn-v2-node.yaml | 4 + scripts/migrate-longhorn-v2/README.md | 8 + scripts/migrate-longhorn-v2/main.go | 13 ++ .../provider_volume_inventory.go | 95 +++++++++ .../provider_volume_inventory_test.go | 49 +++++ .../provider_volume_observe.go | 196 ++++++++++++++++++ .../provider_volume_usage.go | 140 +++++++++++++ .../provider_volume_usage_test.go | 110 ++++++++++ 8 files changed, 615 insertions(+) create mode 100644 scripts/migrate-longhorn-v2/provider_volume_inventory.go create mode 100644 scripts/migrate-longhorn-v2/provider_volume_inventory_test.go create mode 100644 scripts/migrate-longhorn-v2/provider_volume_observe.go create mode 100644 scripts/migrate-longhorn-v2/provider_volume_usage.go create mode 100644 scripts/migrate-longhorn-v2/provider_volume_usage_test.go diff --git a/.github/workflows/prepare-longhorn-v2-node.yaml b/.github/workflows/prepare-longhorn-v2-node.yaml index d7fda39e2..125b47fe5 100644 --- a/.github/workflows/prepare-longhorn-v2-node.yaml +++ b/.github/workflows/prepare-longhorn-v2-node.yaml @@ -29,6 +29,7 @@ name: Prepare Longhorn V2 Migration inventory-replicas, inventory-storage-root, inventory-storage-layout, + inventory-provider-volume-footprint, inspect-host-memory, reboot-host-for-hugepages, repair-v1-reservation, @@ -131,6 +132,9 @@ jobs: inventory-storage-layout) [[ "${LONGHORN_V2_NODE}" =~ ^temporary-(v1-[12]|v2-1)$ && "${LONGHORN_V2_CONFIRM}" == "inventory-longhorn-storage-layout-${LONGHORN_V2_NODE}" ]] ;; + inventory-provider-volume-footprint) + [[ "${LONGHORN_V2_CONFIRM}" == inventory-provider-volume-footprint ]] + ;; refresh-clients) [[ "${LONGHORN_V2_CONFIRM}" == refresh-longhorn-v2-clients ]] ;; diff --git a/scripts/migrate-longhorn-v2/README.md b/scripts/migrate-longhorn-v2/README.md index 2570f6e20..2e39cf7d3 100644 --- a/scripts/migrate-longhorn-v2/README.md +++ b/scripts/migrate-longhorn-v2/README.md @@ -27,6 +27,14 @@ phases, encryption properties and counts, while withholding device paths and identities. It grants no reset clearance and never treats a failed read as absence. The original root-only mode keeps its existing scope and schema. +`inventory-provider-volume-footprint` reads the current provider-backed claims +and native filesystem usage from their running mounts. Confirm with +`inventory-provider-volume-footprint` on a first-attempt main dispatch. +It checks the original claim and volume bindings and the worker boot identities +on both sides of the read. The receipt reports aggregate GiB rounded up, with +unmounted claims accounted for at their full requested capacity. Use the result +as an input to capacity planning and validate each restore before cutover. + `inspect-host-memory` is a read-only diagnostic for a fixed permanent worker, confirmed with `inspect-longhorn-v2-host-memory-`. It reads only the native huge-page allocation and selected `/proc/meminfo` counters, bracketed by unchanged diff --git a/scripts/migrate-longhorn-v2/main.go b/scripts/migrate-longhorn-v2/main.go index fbefc35e1..bb70dc142 100644 --- a/scripts/migrate-longhorn-v2/main.go +++ b/scripts/migrate-longhorn-v2/main.go @@ -37,6 +37,8 @@ func main() { authorization = authorizeStorageRoot(os.Getenv, target) case "inventory-storage-layout": authorization = authorizeStorageLayout(os.Getenv, target) + case "inventory-provider-volume-footprint": + authorization = authorizeProviderVolumeFootprint(os.Getenv) default: authorization = authorizeRefresh(os.Getenv) } @@ -46,6 +48,17 @@ func main() { } ctx, cancel := context.WithTimeout(context.Background(), 110*time.Minute) defer cancel() + if operation == "inventory-provider-volume-footprint" { + receipt, err := observeProviderVolumeFootprint(ctx, client{run: providerFootprintReadCommand, context: "admin@prod"}, os.Getenv("GITHUB_SHA")) + if err != nil { + fmt.Fprintln(os.Stderr, "Provider volume footprint held:", err) + os.Exit(1) + } + fmt.Printf("LONGHORN_PROVIDER_VOLUME_FOOTPRINT=OBSERVED claims=%d measured=%d unmeasured=%d requestedGiBUpper=%d measuredUsedGiBUpper=%d unmeasuredRequestedGiBUpper=%d\n", + receipt.Claims, receipt.Measured, receipt.Unmeasured, providerUsageGiBUpper(receipt.RequestedBytes), + providerUsageGiBUpper(receipt.MeasuredUsedBytes), providerUsageGiBUpper(receipt.UnmeasuredRequestedBytes)) + return + } if operation == "repair-v1-reservation" { if err := repairV1Reservation(ctx, client{run: execute, context: "admin@prod"}, os.Getenv("GITHUB_SHA")); err != nil { fmt.Fprintln(os.Stderr, "V1 reservation repair held:", err) diff --git a/scripts/migrate-longhorn-v2/provider_volume_inventory.go b/scripts/migrate-longhorn-v2/provider_volume_inventory.go new file mode 100644 index 000000000..9e712e08b --- /dev/null +++ b/scripts/migrate-longhorn-v2/provider_volume_inventory.go @@ -0,0 +1,95 @@ +package main + +import ( + "errors" + "reflect" + "sort" + "strconv" + + kresource "k8s.io/apimachinery/pkg/api/resource" +) + +const hetznerCSIDriver = "csi.hetzner.cloud" + +type providerVolumeClaim struct { + Namespace, Claim, ClaimUID, ClaimVersion string + PV, PVUID, PVVersion, Handle string + RequestedBytes uint64 +} + +// The inventory binds PVC and PV identities on both sides of the live usage +// read. It never treats a missing or released original as zero capacity. +func selectProviderVolumeClaims(pvs, claims []object) ([]providerVolumeClaim, error) { + unverified := errors.New("provider volume inventory has incomplete or changed bindings") + if pvs == nil || claims == nil { + return nil, unverified + } + claimsByName := make(map[string]object, len(claims)) + for _, claim := range claims { + key := text(claim, "metadata", "namespace") + "/" + text(claim, "metadata", "name") + if claimsByName[key] != nil { + return nil, unverified + } + claimsByName[key] = claim + } + var selected []providerVolumeClaim + seenHandles := map[string]bool{} + seenClaims := map[string]bool{} + seenPVs := map[string]bool{} + for _, pv := range pvs { + if text(pv, "spec", "csi", "driver") != hetznerCSIDriver { + continue + } + bound := providerVolumeClaim{ + Namespace: text(pv, "spec", "claimRef", "namespace"), Claim: text(pv, "spec", "claimRef", "name"), + ClaimUID: text(pv, "spec", "claimRef", "uid"), PV: text(pv, "metadata", "name"), + PVUID: text(pv, "metadata", "uid"), PVVersion: text(pv, "metadata", "resourceVersion"), Handle: text(pv, "spec", "csi", "volumeHandle"), + } + handle, err := strconv.ParseInt(bound.Handle, 10, 64) + if err != nil || handle <= 0 || strconv.FormatInt(handle, 10) != bound.Handle || seenHandles[bound.Handle] || + bound.PVUID == "" || bound.PVVersion == "" || bound.ClaimUID == "" || seenClaims[bound.ClaimUID] || seenPVs[bound.PVUID] || + safeKubeRequest("persistentvolume", "", bound.PV, false) != nil || !kubeNamespace.MatchString(bound.Namespace) || + safeKubeRequest("persistentvolumeclaim", bound.Namespace, bound.Claim, false) != nil || + text(pv, "status", "phase") != "Bound" || value(pv, "metadata", "deletionTimestamp") != nil || !providerRWOFilesystem(pv) { + return nil, unverified + } + claim := claimsByName[bound.Namespace+"/"+bound.Claim] + bound.ClaimVersion = text(claim, "metadata", "resourceVersion") + class := text(pv, "spec", "storageClassName") + if bound.ClaimVersion == "" || text(claim, "metadata", "uid") != bound.ClaimUID || text(claim, "spec", "volumeName") != bound.PV || + text(claim, "status", "phase") != "Bound" || value(claim, "metadata", "deletionTimestamp") != nil || + class == "" || text(claim, "spec", "storageClassName") != class || !providerRWOFilesystem(claim) { + return nil, unverified + } + capacity := providerVolumeBytes(text(pv, "spec", "capacity", "storage")) + requested := providerVolumeBytes(text(claim, "spec", "resources", "requests", "storage")) + current := providerVolumeBytes(text(claim, "status", "capacity", "storage")) + if capacity == 0 || requested == 0 || requested > capacity || current != capacity { + return nil, unverified + } + bound.RequestedBytes = capacity + seenHandles[bound.Handle], seenClaims[bound.ClaimUID], seenPVs[bound.PVUID] = true, true, true + selected = append(selected, bound) + } + if len(selected) == 0 { + return nil, errors.New("no provider-backed source volume was found") + } + sort.Slice(selected, func(i, j int) bool { return selected[i].ClaimUID < selected[j].ClaimUID }) + return selected, nil +} + +func providerRWOFilesystem(o object) bool { + return text(o, "spec", "volumeMode") == "Filesystem" && reflect.DeepEqual(value(o, "spec", "accessModes"), []any{"ReadWriteOnce"}) +} + +func providerVolumeBytes(raw string) uint64 { + quantity, err := kresource.ParseQuantity(raw) + if err != nil { + return 0 + } + bytes, exact := quantity.AsInt64() + if !exact || bytes <= 0 { + return 0 + } + return uint64(bytes) +} diff --git a/scripts/migrate-longhorn-v2/provider_volume_inventory_test.go b/scripts/migrate-longhorn-v2/provider_volume_inventory_test.go new file mode 100644 index 000000000..18c933690 --- /dev/null +++ b/scripts/migrate-longhorn-v2/provider_volume_inventory_test.go @@ -0,0 +1,49 @@ +package main + +import "testing" + +func TestProviderInventoryBindsTheOriginalClaimAndVolume(t *testing.T) { + for _, test := range []struct { + name string + change func(object, object) + proven bool + }{ + {"exact bound provider claim", func(object, object) {}, true}, + {"replacement claim", func(_ object, pvc object) { valueObject(pvc, "metadata")["uid"] = "replacement" }, false}, + {"replacement binding", func(_ object, pvc object) { valueObject(pvc, "spec")["volumeName"] = "other" }, false}, + {"unsafe resize in flight", func(_ object, pvc object) { valueObject(pvc, "spec", "resources", "requests")["storage"] = "11Gi" }, false}, + {"provider handle mismatch", func(pv, _ object) { valueObject(pv, "spec", "csi")["volumeHandle"] = "unknown" }, false}, + {"released source", func(pv, _ object) { valueObject(pv, "status")["phase"] = "Released" }, false}, + {"block mode outside filesystem plan", func(pv, _ object) { valueObject(pv, "spec")["volumeMode"] = "Block" }, false}, + } { + t.Run(test.name, func(t *testing.T) { + pv := object{ + "metadata": object{"name": "pv", "uid": "pv-uid", "resourceVersion": "42"}, + "spec": object{ + "claimRef": object{"name": "data", "namespace": "provider-app", "uid": "pvc-uid"}, + "capacity": object{"storage": "10Gi"}, "storageClassName": "hcloud", + "csi": object{"driver": "csi.hetzner.cloud", "volumeHandle": "1234"}, + "accessModes": []any{"ReadWriteOnce"}, "volumeMode": "Filesystem", + }, + "status": object{"phase": "Bound"}, + } + pvc := object{ + "metadata": object{"name": "data", "namespace": "provider-app", "uid": "pvc-uid", "resourceVersion": "41"}, + "spec": object{ + "volumeName": "pv", "storageClassName": "hcloud", "accessModes": []any{"ReadWriteOnce"}, + "volumeMode": "Filesystem", "resources": object{"requests": object{"storage": "10Gi"}}, + }, + "status": object{"phase": "Bound", "capacity": object{"storage": "10Gi"}}, + } + test.change(pv, pvc) + claims, err := selectProviderVolumeClaims([]object{pv}, []object{pvc}) + if test.proven { + if err != nil || len(claims) != 1 || claims[0].RequestedBytes != 10<<30 || claims[0].ClaimUID != "pvc-uid" { + t.Fatalf("valid source inventory was rejected: %v", err) + } + } else if err == nil { + t.Fatal("an incomplete or changed source counted as a verified capacity input") + } + }) + } +} diff --git a/scripts/migrate-longhorn-v2/provider_volume_observe.go b/scripts/migrate-longhorn-v2/provider_volume_observe.go new file mode 100644 index 000000000..72c37a7b2 --- /dev/null +++ b/scripts/migrate-longhorn-v2/provider_volume_observe.go @@ -0,0 +1,196 @@ +package main + +import ( + "context" + "errors" + "maps" + "regexp" + "slices" + "strings" + "time" +) + +var providerStatsWorker = regexp.MustCompile(`^prod-worker-[123]$`) +var providerStatsHost = regexp.MustCompile(`^hcloud://[1-9][0-9]{0,18}$`) + +type providerVolumeUsageReceipt struct { + Claims, Measured, Unmeasured int + RequestedBytes, MeasuredUsedBytes, UnmeasuredRequestedBytes uint64 +} + +type providerStatsEpoch struct{ UID, ProviderID, BootID string } + +func providerUsageGiBUpper(bytes uint64) uint64 { + if bytes == 0 { + return 0 + } + return ((bytes - 1) >> 30) + 1 +} + +func authorizeProviderVolumeFootprint(getenv func(string) string) error { + if getenv("LONGHORN_V2_OPERATION") != "inventory-provider-volume-footprint" { + return errors.New("the explicit provider-volume footprint operation is required") + } + return authorizeWorkflow(getenv, "inventory-provider-volume-footprint") +} + +// Keep the production admin credential behind a read-only command allowlist. +// The kubelet proxy route is fixed, and only the reviewed storage workers can +// be addressed through it. +func providerFootprintReadCommand(ctx context.Context, input []byte, program string, args ...string) ([]byte, error) { + if input != nil || !providerFootprintReadAllowed(program, args...) { + return nil, errors.New("the provider footprint operation permits only fixed reads") + } + return inventoryExecuteCommand(ctx, program, args...) +} + +func providerFootprintReadAllowed(program string, args ...string) bool { + if program == "git" { + return inventoryReadAllowed(program, args...) + } + if program != "kubectl" || len(args) < 4 || + !slices.Equal(args[:3], []string{"--context=admin@prod", "--request-timeout=20s", "get"}) { + return false + } + switch tail := args[3:]; len(tail) { + case 1: + const prefix = "--raw=/api/v1/nodes/" + const suffix = "/proxy/stats/summary" + node, ok := strings.CutPrefix(tail[0], prefix) + if !ok { + return false + } + node, ok = strings.CutSuffix(node, suffix) + return ok && providerStatsWorker.MatchString(node) + case 2: + return slices.Equal(tail, []string{"persistentvolumes", "--output=json"}) + case 3: + if slices.Equal(tail, []string{"persistentvolumeclaims", "--output=json", "--all-namespaces"}) { + return true + } + if tail[0] == "node" { + return providerStatsWorker.MatchString(tail[1]) && tail[2] == "--output=json" + } + namespace, ok := strings.CutPrefix(tail[2], "--namespace=") + return ok && tail[0] == "pods" && tail[1] == "--output=json" && slices.Contains([]string{"kube-system", "observability", "openbao"}, namespace) + default: + return false + } +} + +func observeProviderVolumeFootprint(ctx context.Context, c client, source string) (providerVolumeUsageReceipt, error) { + if c.context != "admin@prod" { + return providerVolumeUsageReceipt{}, errors.New("protected production provider-volume observation is required") + } + if err := sourceCurrent(ctx, c.run, source); err != nil { + return providerVolumeUsageReceipt{}, err + } + claims, mounts, err := observeProviderVolumeProjection(ctx, c) + if err != nil { + return providerVolumeUsageReceipt{}, err + } + byNode := map[string][]providerPodMount{} + for _, mount := range mounts { + if !providerStatsWorker.MatchString(mount.Node) { + return providerVolumeUsageReceipt{}, errors.New("a provider claim is outside the reviewed worker roster") + } + byNode[mount.Node] = append(byNode[mount.Node], mount) + } + nodes := slices.Sorted(maps.Keys(byNode)) + epochs := make(map[string]providerStatsEpoch, len(nodes)) + used := make(map[string]uint64, len(mounts)) + for _, node := range nodes { + epoch, err := readProviderStatsEpoch(ctx, c, node) + if err != nil { + return providerVolumeUsageReceipt{}, err + } + raw, err := c.kubectl(ctx, nil, "get", "--raw=/api/v1/nodes/"+node+"/proxy/stats/summary") + if err != nil { + return providerVolumeUsageReceipt{}, errors.New("native kubelet filesystem statistics are unavailable") + } + stats, err := readProviderNodeUsage(raw, node, byNode[node], time.Now().UTC()) + if err != nil { + return providerVolumeUsageReceipt{}, err + } + maps.Copy(used, stats) + epochs[node] = epoch + } + afterClaims, afterMounts, err := observeProviderVolumeProjection(ctx, c) + if err != nil || !slices.Equal(claims, afterClaims) || !slices.Equal(mounts, afterMounts) { + return providerVolumeUsageReceipt{}, errors.New("provider claim bindings or consumer mounts changed during observation") + } + for _, node := range nodes { + after, err := readProviderStatsEpoch(ctx, c, node) + if err != nil || after != epochs[node] { + return providerVolumeUsageReceipt{}, errors.New("provider claim worker identity changed during observation") + } + } + if err := sourceCurrent(ctx, c.run, source); err != nil { + return providerVolumeUsageReceipt{}, err + } + return summarizeProviderVolumeUsage(claims, used) +} + +func summarizeProviderVolumeUsage(claims []providerVolumeClaim, used map[string]uint64) (providerVolumeUsageReceipt, error) { + receipt := providerVolumeUsageReceipt{Claims: len(claims)} + for _, claim := range claims { + amount, measured := used[claim.ClaimUID] + if ^uint64(0)-receipt.RequestedBytes < claim.RequestedBytes || amount > claim.RequestedBytes { + return providerVolumeUsageReceipt{}, errors.New("provider volume capacity exceeded the supported aggregate") + } + receipt.RequestedBytes += claim.RequestedBytes + if measured { + receipt.Measured++ + receipt.MeasuredUsedBytes += amount + } else { + receipt.Unmeasured++ + receipt.UnmeasuredRequestedBytes += claim.RequestedBytes + } + } + if receipt.Measured != len(used) { + return providerVolumeUsageReceipt{}, errors.New("filesystem statistics included an unverified provider claim") + } + return receipt, nil +} + +func observeProviderVolumeProjection(ctx context.Context, c client) ([]providerVolumeClaim, []providerPodMount, error) { + pvs, err := c.list(ctx, "persistentvolumes", "") + if err != nil { + return nil, nil, err + } + pvcs, err := c.list(ctx, "persistentvolumeclaims", "*") + if err != nil { + return nil, nil, err + } + claims, err := selectProviderVolumeClaims(pvs, pvcs) + if err != nil { + return nil, nil, err + } + namespaces := map[string]bool{} + for _, claim := range claims { + namespaces[claim.Namespace] = true + } + pods := []object{} + for _, namespace := range slices.Sorted(maps.Keys(namespaces)) { + current, err := c.list(ctx, "pods", namespace) + if err != nil { + return nil, nil, err + } + pods = append(pods, current...) + } + mounts, err := selectProviderPodMounts(claims, pods) + return claims, mounts, err +} + +func readProviderStatsEpoch(ctx context.Context, c client, name string) (providerStatsEpoch, error) { + if !providerStatsWorker.MatchString(name) { + return providerStatsEpoch{}, errors.New("provider filesystem statistics worker is unverified") + } + n, err := c.get(ctx, "node", "", name, false) + if err != nil || text(n, "metadata", "name") != name || value(n, "metadata", "deletionTimestamp") != nil || + !ready(valueObject(n, "status")) || !providerStatsHost.MatchString(text(n, "spec", "providerID")) || + text(n, "status", "nodeInfo", "bootID") == "" { + return providerStatsEpoch{}, errors.New("provider filesystem statistics worker is unverified") + } + return providerStatsEpoch{text(n, "metadata", "uid"), text(n, "spec", "providerID"), text(n, "status", "nodeInfo", "bootID")}, nil +} diff --git a/scripts/migrate-longhorn-v2/provider_volume_usage.go b/scripts/migrate-longhorn-v2/provider_volume_usage.go new file mode 100644 index 000000000..b94e3e98a --- /dev/null +++ b/scripts/migrate-longhorn-v2/provider_volume_usage.go @@ -0,0 +1,140 @@ +package main + +import ( + "encoding/json" + "errors" + "sort" + "time" + + "github.com/devantler-tech/platform/internal/strictjson" +) + +type providerPodMount struct { + Claim providerVolumeClaim + Node, Namespace, Pod, PodUID, Volume string +} + +// A finished job or dormant claim remains part of requested capacity. Only a +// unique current Pod mount can contribute observed filesystem usage. +func selectProviderPodMounts(claims []providerVolumeClaim, pods []object) ([]providerPodMount, error) { + unverified := errors.New("provider claim mounts changed or are ambiguous") + if len(claims) == 0 || pods == nil { + return nil, unverified + } + byClaim := make(map[string]providerVolumeClaim, len(claims)) + for _, claim := range claims { + byClaim[claim.Namespace+"/"+claim.Claim] = claim + } + seen := map[string]bool{} + mounts := make([]providerPodMount, 0, len(claims)) + for _, pod := range pods { + phase := text(pod, "status", "phase") + if phase == "Succeeded" || phase == "Failed" { + continue + } + ns, name, uid := text(pod, "metadata", "namespace"), text(pod, "metadata", "name"), text(pod, "metadata", "uid") + node := text(pod, "spec", "nodeName") + volumes, _ := value(pod, "spec", "volumes").([]any) + for _, entry := range volumes { + volume, _ := entry.(object) + claim, selected := byClaim[ns+"/"+text(volume, "persistentVolumeClaim", "claimName")] + if !selected { + continue + } + volumeName := text(volume, "name") + if seen[claim.ClaimUID] || phase != "Running" || value(pod, "metadata", "deletionTimestamp") != nil || + uid == "" || !kubeNamespace.MatchString(node) || safeKubeRequest("pods", ns, name, false) != nil || + !kubeNamespace.MatchString(volumeName) || !providerPodUsesMount(pod, volumeName) { + return nil, unverified + } + seen[claim.ClaimUID] = true + mounts = append(mounts, providerPodMount{Claim: claim, Node: node, Namespace: ns, Pod: name, PodUID: uid, Volume: volumeName}) + } + } + sort.Slice(mounts, func(i, j int) bool { return mounts[i].Claim.ClaimUID < mounts[j].Claim.ClaimUID }) + return mounts, nil +} + +func providerPodUsesMount(pod object, volume string) bool { + containers, _ := value(pod, "spec", "containers").([]any) + for _, entry := range containers { + container, _ := entry.(object) + mounts, _ := value(container, "volumeMounts").([]any) + for _, raw := range mounts { + mount, _ := raw.(object) + if text(mount, "name") == volume && text(mount, "mountPath") != "" { + return true + } + } + } + return false +} + +type providerFilesystemStats struct { + Name string `json:"name"` + Time time.Time `json:"time"` + PVCRef struct { + Name string `json:"name"` + Namespace string `json:"namespace"` + } `json:"pvcRef"` + Used *uint64 `json:"usedBytes"` + Available *uint64 `json:"availableBytes"` + Capacity *uint64 `json:"capacityBytes"` +} + +func readProviderNodeUsage(raw []byte, node string, mounts []providerPodMount, now time.Time) (map[string]uint64, error) { + unverified := errors.New("provider filesystem usage is incomplete or unrelated to the observed mount") + if len(raw) > 16<<20 || node == "" || len(mounts) == 0 || now.IsZero() || strictjson.Validate(raw, map[string][]string{ + "": {"node", "pods"}, + "node": {"nodeName"}, + "pods[]": {"podRef", "volume"}, + "pods[].podRef": {"name", "namespace", "uid"}, + "pods[].volume[]": {"name", "time", "pvcRef", "usedBytes", "availableBytes", "capacityBytes"}, + "pods[].volume[].pvcRef": {"name", "namespace"}, + }) != nil { + return nil, unverified + } + var summary struct { + Node struct { + Name string `json:"nodeName"` + } `json:"node"` + Pods []struct { + Ref struct { + UID string `json:"uid"` + Name string `json:"name"` + Namespace string `json:"namespace"` + } `json:"podRef"` + Volumes []providerFilesystemStats `json:"volume"` + } `json:"pods"` + } + if json.Unmarshal(raw, &summary) != nil || summary.Node.Name != node || summary.Pods == nil { + return nil, unverified + } + type podMountKey struct{ namespace, pod, uid, volume string } + lookup := make(map[podMountKey]providerFilesystemStats, len(mounts)) + for _, pod := range summary.Pods { + for _, volume := range pod.Volumes { + key := podMountKey{pod.Ref.Namespace, pod.Ref.Name, pod.Ref.UID, volume.Name} + if _, exists := lookup[key]; exists { + return nil, unverified + } + lookup[key] = volume + } + } + usage := make(map[string]uint64, len(mounts)) + for _, mount := range mounts { + stats, found := lookup[podMountKey{mount.Namespace, mount.Pod, mount.PodUID, mount.Volume}] + uid := mount.Claim.ClaimUID + _, repeated := usage[uid] + if !found || repeated || uid == "" || mount.PodUID == "" || mount.Node != node || + stats.PVCRef.Name != mount.Claim.Claim || stats.PVCRef.Namespace != mount.Claim.Namespace || + stats.Time.Before(now.Add(-3*time.Minute)) || stats.Time.After(now.Add(30*time.Second)) || + stats.Used == nil || stats.Available == nil || stats.Capacity == nil || *stats.Capacity == 0 || + *stats.Capacity > mount.Claim.RequestedBytes || *stats.Available > *stats.Capacity || + *stats.Used > *stats.Capacity-*stats.Available { + return nil, unverified + } + usage[uid] = *stats.Used + } + return usage, nil +} diff --git a/scripts/migrate-longhorn-v2/provider_volume_usage_test.go b/scripts/migrate-longhorn-v2/provider_volume_usage_test.go new file mode 100644 index 000000000..2086f3252 --- /dev/null +++ b/scripts/migrate-longhorn-v2/provider_volume_usage_test.go @@ -0,0 +1,110 @@ +package main + +import ( + "encoding/json" + "slices" + "testing" + "time" +) + +func TestProviderUsageBindsFreshStatsToTheObservedMount(t *testing.T) { + now := time.Date(2026, time.November, 10, 12, 0, 0, 0, time.UTC) + identity := providerVolumeClaim{Namespace: "provider-app", Claim: "data", ClaimUID: "claim-uid", RequestedBytes: 10 << 30} + mount := providerPodMount{Claim: identity, Node: "storage-worker", Namespace: "provider-app", Pod: "app-0", PodUID: "pod-uid", Volume: "data"} + for _, test := range []struct { + name string + edit func(object, object, object) + proven bool + }{ + {"matching mounted filesystem", func(object, object, object) {}, true}, + {"replacement Pod", func(_ object, pod, _ object) { valueObject(pod, "podRef")["uid"] = "other" }, false}, + {"different claim", func(_, _ object, volume object) { valueObject(volume, "pvcRef")["name"] = "other" }, false}, + {"different node", func(summary, _, _ object) { valueObject(summary, "node")["nodeName"] = "other" }, false}, + {"old filesystem measurement", func(_, _ object, volume object) { volume["time"] = now.Add(-10 * time.Minute) }, false}, + {"missing usage", func(_, _ object, volume object) { delete(volume, "usedBytes") }, false}, + {"larger than the provider volume", func(_, _ object, volume object) { volume["capacityBytes"] = 11 << 30 }, false}, + } { + t.Run(test.name, func(t *testing.T) { + volume := object{"name": "data", "pvcRef": object{"name": "data", "namespace": "provider-app"}, + "time": now.Add(-20 * time.Second), "usedBytes": 3 << 30, "availableBytes": 5 << 30, "capacityBytes": 9 << 30} + pod := object{"podRef": object{"uid": "pod-uid", "name": "app-0", "namespace": "provider-app"}, "volume": []object{volume}} + summary := object{"node": object{"nodeName": mount.Node}, "pods": []object{pod}} + test.edit(summary, pod, volume) + raw, err := json.Marshal(summary) + if err != nil { + t.Fatal(err) + } + got, err := readProviderNodeUsage(raw, mount.Node, []providerPodMount{mount}, now) + if test.proven { + if err != nil || got[identity.ClaimUID] != 3<<30 { + t.Fatalf("matching native usage was rejected: %v", err) + } + } else if err == nil { + t.Fatal("unrelated or incomplete filesystem stats qualified") + } + }) + } +} + +func TestProviderUsageKeepsUnmeasuredSourcesAtTheirFullRequest(t *testing.T) { + claims := []providerVolumeClaim{ + {Namespace: "provider-app", Claim: "data", ClaimUID: "live", RequestedBytes: 10 << 30}, + {Namespace: "provider-app", Claim: "archive", ClaimUID: "dormant", RequestedBytes: 20 << 30}, + } + pod := func(uid, claim, phase string) object { + return object{ + "metadata": object{"name": "app-" + uid, "namespace": "provider-app", "uid": uid}, + "spec": object{ + "nodeName": "prod-worker-1", + "volumes": []any{object{"name": "data", "persistentVolumeClaim": object{"claimName": claim}}}, + "containers": []any{object{"volumeMounts": []any{object{"name": "data", "mountPath": "/data"}}}}, + }, + "status": object{"phase": phase}, + } + } + mounts, err := selectProviderPodMounts(claims, []object{pod("live-reader", "data", "Running"), pod("old-job", "archive", "Succeeded")}) + if err != nil || len(mounts) != 1 || mounts[0].Claim.ClaimUID != "live" { + t.Fatalf("original running mount was not identified: %v", err) + } + if _, err := selectProviderPodMounts(claims, []object{pod("live-reader", "data", "Running"), pod("competing", "data", "Running")}); err == nil { + t.Fatal("multiple consumers of one provider volume qualified") + } + receipt, err := summarizeProviderVolumeUsage(claims, map[string]uint64{"live": 3 << 30}) + if err != nil || receipt.Measured != 1 || receipt.Unmeasured != 1 || receipt.MeasuredUsedBytes != 3<<30 || + receipt.UnmeasuredRequestedBytes != 20<<30 || receipt.RequestedBytes != 30<<30 { + t.Fatalf("unmeasured original was undercounted: %+v %v", receipt, err) + } + if _, err := summarizeProviderVolumeUsage(claims, map[string]uint64{"unknown-uid": 0}); err == nil { + t.Fatal("usage outside the verified source inventory was counted") + } +} + +func TestProviderFootprintNativeCommandHasOnlyFixedReadRoutes(t *testing.T) { + prefix := []string{"--context=admin@prod", "--request-timeout=20s", "get"} + for _, test := range []struct { + tail []string + allowed bool + }{ + {[]string{"persistentvolumes", "--output=json"}, true}, + {[]string{"persistentvolumeclaims", "--output=json", "--all-namespaces"}, true}, + {[]string{"pods", "--output=json", "--namespace=openbao"}, true}, + {[]string{"node", "prod-worker-2", "--output=json"}, true}, + {[]string{"--raw=/api/v1/nodes/prod-worker-1/proxy/stats/summary"}, true}, + {[]string{"--raw=/api/v1/nodes/prod-worker-1/proxy/stats/summary?foo=bar"}, false}, + {[]string{"--raw=/api/v1/nodes/prod-worker-1/proxy/exec"}, false}, + {[]string{"--raw=/api/v1/nodes/another-node/proxy/stats/summary"}, false}, + {[]string{"pods", "--output=json", "--namespace=other"}, false}, + {[]string{"secrets", "--output=json", "--all-namespaces"}, false}, + {[]string{"node", "prod-worker-1", "--output=json", "--watch"}, false}, + } { + if got := providerFootprintReadAllowed("kubectl", append(slices.Clone(prefix), test.tail...)...); got != test.allowed { + t.Errorf("fixed provider read authorization = %t, want %t for %v", got, test.allowed, test.tail) + } + } + if providerFootprintReadAllowed("kubectl", "--context=admin@prod", "--request-timeout=20s", "patch", "node", "prod-worker-1", "--output=json") { + t.Fatal("a write was accepted") + } + if providerFootprintReadAllowed("kubectl", "--context=other", "--request-timeout=20s", "get", "persistentvolumes", "--output=json") { + t.Fatal("a different context was accepted") + } +} From d0ecc207a61a52e19e9b61968bcc5b29ee18cd9c Mon Sep 17 00:00:00 2001 From: Nikolai Emil Damm Date: Tue, 10 Nov 2026 15:21:37 +0100 Subject: [PATCH 2/4] =?UTF-8?q?fix(ci):=20record=20the=20storage=20observe?= =?UTF-8?q?r=E2=80=99s=20Go=20dependency?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- go.mod | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/go.mod b/go.mod index bfb375c31..5c9b26275 100644 --- a/go.mod +++ b/go.mod @@ -46,6 +46,7 @@ require ( github.com/aws/aws-sdk-go-v2 v1.47.5 github.com/aws/aws-sdk-go-v2/service/s3 v1.114.6 github.com/google/go-github/v88 v88.0.0 + k8s.io/apimachinery v0.37.2 k8s.io/client-go v0.37.2 ) @@ -99,7 +100,6 @@ require ( golang.org/x/term v0.47.0 // indirect gopkg.in/inf.v0 v0.9.1 // indirect gopkg.in/ini.v1 v1.67.3 // indirect - k8s.io/apimachinery v0.37.2 // indirect k8s.io/klog/v2 v2.140.0 // indirect k8s.io/kube-openapi v0.0.0-20260721132016-d427ff9ee9ad // indirect k8s.io/utils v0.0.0-20260626114624-be93311217bd // indirect From a8081b3ba856fbf16ecf63b6a69bc2047caa0959 Mon Sep 17 00:00:00 2001 From: Nikolai Emil Damm Date: Tue, 10 Nov 2026 17:07:53 +0100 Subject: [PATCH 3/4] fix(storage): measure provider mounts with native Talos reads --- scripts/migrate-longhorn-v2/README.md | 6 +- .../migrate-longhorn-v2/inventory_runtime.go | 36 +++--- scripts/migrate-longhorn-v2/node.go | 4 + .../provider_volume_observe.go | 69 +++++++---- .../provider_volume_talos.go | 113 ++++++++++++++++++ .../provider_volume_usage.go | 75 +----------- .../provider_volume_usage_test.go | 73 ++++++----- 7 files changed, 234 insertions(+), 142 deletions(-) create mode 100644 scripts/migrate-longhorn-v2/provider_volume_talos.go diff --git a/scripts/migrate-longhorn-v2/README.md b/scripts/migrate-longhorn-v2/README.md index 2e39cf7d3..3f9edace0 100644 --- a/scripts/migrate-longhorn-v2/README.md +++ b/scripts/migrate-longhorn-v2/README.md @@ -28,10 +28,10 @@ identities. It grants no reset clearance and never treats a failed read as absen The original root-only mode keeps its existing scope and schema. `inventory-provider-volume-footprint` reads the current provider-backed claims -and native filesystem usage from their running mounts. Confirm with +and native Talos filesystem usage from their running mounts. Confirm with `inventory-provider-volume-footprint` on a first-attempt main dispatch. -It checks the original claim and volume bindings and the worker boot identities -on both sides of the read. The receipt reports aggregate GiB rounded up, with +It binds each running CSI mount to its provider disk and checks the claim, +volume, disk and worker identities on both sides of the read. The receipt reports aggregate GiB rounded up, with unmounted claims accounted for at their full requested capacity. Use the result as an input to capacity planning and validate each restore before cutover. diff --git a/scripts/migrate-longhorn-v2/inventory_runtime.go b/scripts/migrate-longhorn-v2/inventory_runtime.go index 91eaecb77..60beff93d 100644 --- a/scripts/migrate-longhorn-v2/inventory_runtime.go +++ b/scripts/migrate-longhorn-v2/inventory_runtime.go @@ -130,22 +130,35 @@ func newInventoryReader(ctx context.Context, address string) (nativeReplicaReade // workflow's memory-backed credentials and bounded block API. The unused-worker // profile additionally verifies its original partition before directory reads. func openInventoryReader(ctx context.Context, address, unusedEphemeralUUID string) (nativeReplicaReader, func() error, error) { + connection, err := openInventoryTalosClient(ctx) + if err != nil { + return nil, nil, err + } + blocks := longhorninventory.BoundedBlocks{API: v1alpha1.NewStateClient(connection.Conn()), Target: address} + var reader nativeReplicaReader = longhorninventory.Native{API: connection.MachineClient, Target: address, Blocks: blocks} + if unusedEphemeralUUID != "" { + reader = unusedTemporaryReader{nativeReplicaReader: reader, blocks: blocks, address: address, ephemeralUUID: unusedEphemeralUUID} + } + return reader, connection.Close, nil +} + +func openInventoryTalosClient(ctx context.Context) (*talos.Client, error) { directory := os.Getenv("LONGHORN_V2_CREDENTIAL_DIR") if !regexp.MustCompile(`^/dev/shm/longhorn-v2\.[A-Za-z0-9]+$`).MatchString(directory) || os.Getenv("TALOSCONFIG") != filepath.Join(directory, "talosconfig") { - return nil, nil, errInventory + return nil, errInventory } parent, err := os.Lstat(directory) if err != nil || !parent.IsDir() || parent.Mode().Perm() != 0700 { - return nil, nil, errInventory + return nil, errInventory } path := filepath.Join(directory, "talosconfig") before, err := os.Lstat(path) if err != nil || !before.Mode().IsRegular() || before.Mode().Perm() != 0600 || before.Size() <= 0 || before.Size() > 512<<10 { - return nil, nil, errInventory + return nil, errInventory } file, err := os.Open(path) if err != nil { - return nil, nil, errInventory + return nil, errInventory } opened, statErr := file.Stat() data, readErr := io.ReadAll(io.LimitReader(file, (512<<10)+1)) @@ -153,26 +166,21 @@ func openInventoryReader(ctx context.Context, address, unusedEphemeralUUID strin after, lastErr := os.Lstat(path) if statErr != nil || readErr != nil || closeErr != nil || lastErr != nil || len(data) != int(before.Size()) || !os.SameFile(before, opened) || !os.SameFile(before, after) || before.Size() != after.Size() || !before.ModTime().Equal(after.ModTime()) { - return nil, nil, errInventory + return nil, errInventory } config, err := talosconfig.FromBytes(data) if err != nil { - return nil, nil, errInventory + return nil, errInventory } endpoint, err := inventoryEndpoint(config) if err != nil { - return nil, nil, err + return nil, err } connection, err := talos.New(ctx, talos.WithConfig(config), talos.WithContextName(config.Context), talos.WithEndpoints(endpoint), talos.WithGRPCDialOptions(grpc.WithNoProxy())) if err != nil { - return nil, nil, errInventory - } - blocks := longhorninventory.BoundedBlocks{API: v1alpha1.NewStateClient(connection.Conn()), Target: address} - var reader nativeReplicaReader = longhorninventory.Native{API: connection.MachineClient, Target: address, Blocks: blocks} - if unusedEphemeralUUID != "" { - reader = unusedTemporaryReader{nativeReplicaReader: reader, blocks: blocks, address: address, ephemeralUUID: unusedEphemeralUUID} + return nil, errInventory } - return reader, connection.Close, nil + return connection, nil } func runReplicaInventory(ctx context.Context, target, source, runID string) error { diff --git a/scripts/migrate-longhorn-v2/node.go b/scripts/migrate-longhorn-v2/node.go index bdebb3ff1..4dc846cde 100644 --- a/scripts/migrate-longhorn-v2/node.go +++ b/scripts/migrate-longhorn-v2/node.go @@ -15,6 +15,10 @@ func verifyNode(n object, j journal, original bool) (string, error) { value(n, "metadata", "deletionTimestamp") != nil { return "", errors.New("worker hardware or Kubernetes identity changed") } + return workerPrivateAddress(n) +} + +func workerPrivateAddress(n object) (string, error) { addresses, _ := value(n, "status", "addresses").([]any) address := "" for _, raw := range addresses { diff --git a/scripts/migrate-longhorn-v2/provider_volume_observe.go b/scripts/migrate-longhorn-v2/provider_volume_observe.go index 72c37a7b2..837514f7a 100644 --- a/scripts/migrate-longhorn-v2/provider_volume_observe.go +++ b/scripts/migrate-longhorn-v2/provider_volume_observe.go @@ -4,10 +4,10 @@ import ( "context" "errors" "maps" + "net/netip" "regexp" "slices" "strings" - "time" ) var providerStatsWorker = regexp.MustCompile(`^prod-worker-[123]$`) @@ -18,7 +18,7 @@ type providerVolumeUsageReceipt struct { RequestedBytes, MeasuredUsedBytes, UnmeasuredRequestedBytes uint64 } -type providerStatsEpoch struct{ UID, ProviderID, BootID string } +type providerStatsEpoch struct{ UID, ProviderID, BootID, Hardware, Address string } func providerUsageGiBUpper(bytes uint64) uint64 { if bytes == 0 { @@ -34,9 +34,7 @@ func authorizeProviderVolumeFootprint(getenv func(string) string) error { return authorizeWorkflow(getenv, "inventory-provider-volume-footprint") } -// Keep the production admin credential behind a read-only command allowlist. -// The kubelet proxy route is fixed, and only the reviewed storage workers can -// be addressed through it. +// Keep the production credentials behind fixed Kubernetes and Talos reads. func providerFootprintReadCommand(ctx context.Context, input []byte, program string, args ...string) ([]byte, error) { if input != nil || !providerFootprintReadAllowed(program, args...) { return nil, errors.New("the provider footprint operation permits only fixed reads") @@ -45,23 +43,19 @@ func providerFootprintReadCommand(ctx context.Context, input []byte, program str } func providerFootprintReadAllowed(program string, args ...string) bool { - if program == "git" { - return inventoryReadAllowed(program, args...) + if (program == "git" || program == "talosctl") && inventoryReadAllowed(program, args...) { + return true + } + if program == "talosctl" && len(args) == 4 && slices.Equal(args[1:], []string{"get", "disks", "--output=json"}) { + target, ok := strings.CutPrefix(args[0], "--nodes=") + address, err := netip.ParseAddr(target) + return ok && err == nil && netip.MustParsePrefix("10.0.0.0/16").Contains(address) } if program != "kubectl" || len(args) < 4 || !slices.Equal(args[:3], []string{"--context=admin@prod", "--request-timeout=20s", "get"}) { return false } switch tail := args[3:]; len(tail) { - case 1: - const prefix = "--raw=/api/v1/nodes/" - const suffix = "/proxy/stats/summary" - node, ok := strings.CutPrefix(tail[0], prefix) - if !ok { - return false - } - node, ok = strings.CutSuffix(node, suffix) - return ok && providerStatsWorker.MatchString(node) case 2: return slices.Equal(tail, []string{"persistentvolumes", "--output=json"}) case 3: @@ -85,6 +79,25 @@ func observeProviderVolumeFootprint(ctx context.Context, c client, source string if err := sourceCurrent(ctx, c.run, source); err != nil { return providerVolumeUsageReceipt{}, err } + talosClient, err := openInventoryTalosClient(ctx) + if err != nil { + return providerVolumeUsageReceipt{}, errors.New("the protected Talos reader is unavailable") + } + receipt, err := observeProviderVolumeUsage(ctx, c, talosClient) + closeErr := talosClient.Close() + if err != nil { + return providerVolumeUsageReceipt{}, err + } + if closeErr != nil { + return providerVolumeUsageReceipt{}, errors.New("the protected Talos reader could not close") + } + if err := sourceCurrent(ctx, c.run, source); err != nil { + return providerVolumeUsageReceipt{}, err + } + return receipt, nil +} + +func observeProviderVolumeUsage(ctx context.Context, c client, api providerTalosMountAPI) (providerVolumeUsageReceipt, error) { claims, mounts, err := observeProviderVolumeProjection(ctx, c) if err != nil { return providerVolumeUsageReceipt{}, err @@ -98,36 +111,39 @@ func observeProviderVolumeFootprint(ctx context.Context, c client, source string } nodes := slices.Sorted(maps.Keys(byNode)) epochs := make(map[string]providerStatsEpoch, len(nodes)) + devices := make(map[string]map[string]string, len(nodes)) used := make(map[string]uint64, len(mounts)) for _, node := range nodes { epoch, err := readProviderStatsEpoch(ctx, c, node) if err != nil { return providerVolumeUsageReceipt{}, err } - raw, err := c.kubectl(ctx, nil, "get", "--raw=/api/v1/nodes/"+node+"/proxy/stats/summary") + disks, err := readProviderDevices(ctx, c, epoch.Address) if err != nil { - return providerVolumeUsageReceipt{}, errors.New("native kubelet filesystem statistics are unavailable") + return providerVolumeUsageReceipt{}, err } - stats, err := readProviderNodeUsage(raw, node, byNode[node], time.Now().UTC()) + stats, err := readProviderMounts(ctx, api, epoch.Address, byNode[node], disks) if err != nil { return providerVolumeUsageReceipt{}, err } maps.Copy(used, stats) epochs[node] = epoch + devices[node] = disks } afterClaims, afterMounts, err := observeProviderVolumeProjection(ctx, c) if err != nil || !slices.Equal(claims, afterClaims) || !slices.Equal(mounts, afterMounts) { return providerVolumeUsageReceipt{}, errors.New("provider claim bindings or consumer mounts changed during observation") } for _, node := range nodes { + disks, err := readProviderDevices(ctx, c, epochs[node].Address) + if err != nil || !maps.Equal(disks, devices[node]) { + return providerVolumeUsageReceipt{}, errors.New("provider device identity changed during observation") + } after, err := readProviderStatsEpoch(ctx, c, node) if err != nil || after != epochs[node] { return providerVolumeUsageReceipt{}, errors.New("provider claim worker identity changed during observation") } } - if err := sourceCurrent(ctx, c.run, source); err != nil { - return providerVolumeUsageReceipt{}, err - } return summarizeProviderVolumeUsage(claims, used) } @@ -189,8 +205,13 @@ func readProviderStatsEpoch(ctx context.Context, c client, name string) (provide n, err := c.get(ctx, "node", "", name, false) if err != nil || text(n, "metadata", "name") != name || value(n, "metadata", "deletionTimestamp") != nil || !ready(valueObject(n, "status")) || !providerStatsHost.MatchString(text(n, "spec", "providerID")) || - text(n, "status", "nodeInfo", "bootID") == "" { + text(n, "metadata", "uid") == "" || text(n, "status", "nodeInfo", "bootID") == "" { return providerStatsEpoch{}, errors.New("provider filesystem statistics worker is unverified") } - return providerStatsEpoch{text(n, "metadata", "uid"), text(n, "spec", "providerID"), text(n, "status", "nodeInfo", "bootID")}, nil + address, err := workerPrivateAddress(n) + if err != nil || c.verifyHost(ctx, address, n) != nil { + return providerStatsEpoch{}, errors.New("provider filesystem statistics worker hardware is unverified") + } + return providerStatsEpoch{text(n, "metadata", "uid"), text(n, "spec", "providerID"), + text(n, "status", "nodeInfo", "bootID"), text(n, "status", "nodeInfo", "systemUUID"), address}, nil } diff --git a/scripts/migrate-longhorn-v2/provider_volume_talos.go b/scripts/migrate-longhorn-v2/provider_volume_talos.go new file mode 100644 index 000000000..9d7c80b83 --- /dev/null +++ b/scripts/migrate-longhorn-v2/provider_volume_talos.go @@ -0,0 +1,113 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "io" + "regexp" + "strconv" + "strings" + "time" + + "github.com/devantler-tech/platform/internal/strictjson" + machine "github.com/siderolabs/talos/pkg/machinery/api/machine" + talos "github.com/siderolabs/talos/pkg/machinery/client" + "google.golang.org/grpc" +) + +const providerDeviceLink = "/dev/disk/by-id/scsi-0HC_Volume_" + +var providerDiskPath = regexp.MustCompile(`^/dev/[a-z][a-z0-9-]+$`) + +type providerTalosMountAPI interface { + Mounts(context.Context, ...grpc.CallOption) (*machine.MountsResponse, error) +} + +func readProviderDevices(ctx context.Context, c client, address string) (map[string]string, error) { + ctx, stop := context.WithTimeout(ctx, 15*time.Second) + defer stop() + raw, err := c.run(ctx, nil, "talosctl", "--nodes="+address, "get", "disks", "--output=json") + if err != nil { + return nil, errors.New("native Talos disk inventory is unavailable") + } + return parseProviderDevices(raw) +} + +func parseProviderDevices(raw []byte) (map[string]string, error) { + unverified := errors.New("native Talos disk identity is incomplete or ambiguous") + if len(raw) == 0 || len(raw) > 4<<20 { + return nil, unverified + } + decoder := json.NewDecoder(bytes.NewReader(raw)) + devices, paths, ids := map[string]string{}, map[string]bool{}, map[string]bool{} + for { + var entry json.RawMessage + err := decoder.Decode(&entry) + if err == io.EOF && len(ids) != 0 { + return devices, nil + } + var disk object + if err != nil || strictjson.Validate(entry, nil) != nil || json.Unmarshal(entry, &disk) != nil { + return nil, unverified + } + id, device := text(disk, "metadata", "id"), text(disk, "spec", "dev_path") + if text(disk, "metadata", "namespace") != "runtime" || text(disk, "metadata", "type") != "Disks.block.talos.dev" || + text(disk, "metadata", "phase") != "running" || id == "" || ids[id] || !providerDiskPath.MatchString(device) || paths[device] { + return nil, unverified + } + ids[id], paths[device] = true, true + links, _ := value(disk, "spec", "symlinks").([]any) + found := false + for _, rawLink := range links { + link, _ := rawLink.(string) + handle, provider := strings.CutPrefix(link, providerDeviceLink) + if !provider { + continue + } + parsed, err := strconv.ParseInt(handle, 10, 64) + if err != nil || parsed <= 0 || strconv.FormatInt(parsed, 10) != handle || devices[handle] != "" || found { + return nil, unverified + } + devices[handle] = device + found = true + } + } +} + +func readProviderMounts(ctx context.Context, api providerTalosMountAPI, address string, mounts []providerPodMount, devices map[string]string) (map[string]uint64, error) { + ctx, stop := context.WithTimeout(talos.WithNodes(ctx, address), 15*time.Second) + defer stop() + response, err := api.Mounts(ctx, grpc.MaxCallRecvMsgSize(4<<20)) + messages := response.GetMessages() + if err != nil || len(messages) != 1 || messages[0].GetMetadata().GetHostname() != address || + messages[0].GetMetadata().GetError() != "" || messages[0].GetMetadata().GetStatus().GetCode() != 0 { + return nil, errors.New("native Talos mount statistics came from an unverified host") + } + return matchProviderMounts(messages[0].GetStats(), mounts, devices) +} + +func matchProviderMounts(stats []*machine.MountStat, mounts []providerPodMount, devices map[string]string) (map[string]uint64, error) { + unverified := errors.New("native Talos filesystem usage does not match the original provider mount") + byPath := make(map[string]*machine.MountStat, len(stats)) + duplicates := map[string]bool{} + for _, entry := range stats { + path := entry.GetMountedOn() + duplicates[path] = byPath[path] != nil + byPath[path] = entry + } + used := make(map[string]uint64, len(mounts)) + for _, mount := range mounts { + path := "/var/lib/kubelet/pods/" + mount.PodUID + "/volumes/kubernetes.io~csi/" + mount.Claim.PV + "/mount" + device := devices[mount.Claim.Handle] + entry := byPath[path] + if duplicates[path] || device == "" || + (entry.GetFilesystem() != device && entry.GetFilesystem() != providerDeviceLink+mount.Claim.Handle) || + entry.GetSize() == 0 || entry.GetSize() > mount.Claim.RequestedBytes || entry.GetAvailable() > entry.GetSize() { + return nil, unverified + } + used[mount.Claim.ClaimUID] = entry.GetSize() - entry.GetAvailable() + } + return used, nil +} diff --git a/scripts/migrate-longhorn-v2/provider_volume_usage.go b/scripts/migrate-longhorn-v2/provider_volume_usage.go index b94e3e98a..20780dac9 100644 --- a/scripts/migrate-longhorn-v2/provider_volume_usage.go +++ b/scripts/migrate-longhorn-v2/provider_volume_usage.go @@ -1,12 +1,8 @@ package main import ( - "encoding/json" "errors" "sort" - "time" - - "github.com/devantler-tech/platform/internal/strictjson" ) type providerPodMount struct { @@ -43,7 +39,7 @@ func selectProviderPodMounts(claims []providerVolumeClaim, pods []object) ([]pro } volumeName := text(volume, "name") if seen[claim.ClaimUID] || phase != "Running" || value(pod, "metadata", "deletionTimestamp") != nil || - uid == "" || !kubeNamespace.MatchString(node) || safeKubeRequest("pods", ns, name, false) != nil || + !kubeNamespace.MatchString(uid) || !kubeNamespace.MatchString(node) || safeKubeRequest("pods", ns, name, false) != nil || !kubeNamespace.MatchString(volumeName) || !providerPodUsesMount(pod, volumeName) { return nil, unverified } @@ -69,72 +65,3 @@ func providerPodUsesMount(pod object, volume string) bool { } return false } - -type providerFilesystemStats struct { - Name string `json:"name"` - Time time.Time `json:"time"` - PVCRef struct { - Name string `json:"name"` - Namespace string `json:"namespace"` - } `json:"pvcRef"` - Used *uint64 `json:"usedBytes"` - Available *uint64 `json:"availableBytes"` - Capacity *uint64 `json:"capacityBytes"` -} - -func readProviderNodeUsage(raw []byte, node string, mounts []providerPodMount, now time.Time) (map[string]uint64, error) { - unverified := errors.New("provider filesystem usage is incomplete or unrelated to the observed mount") - if len(raw) > 16<<20 || node == "" || len(mounts) == 0 || now.IsZero() || strictjson.Validate(raw, map[string][]string{ - "": {"node", "pods"}, - "node": {"nodeName"}, - "pods[]": {"podRef", "volume"}, - "pods[].podRef": {"name", "namespace", "uid"}, - "pods[].volume[]": {"name", "time", "pvcRef", "usedBytes", "availableBytes", "capacityBytes"}, - "pods[].volume[].pvcRef": {"name", "namespace"}, - }) != nil { - return nil, unverified - } - var summary struct { - Node struct { - Name string `json:"nodeName"` - } `json:"node"` - Pods []struct { - Ref struct { - UID string `json:"uid"` - Name string `json:"name"` - Namespace string `json:"namespace"` - } `json:"podRef"` - Volumes []providerFilesystemStats `json:"volume"` - } `json:"pods"` - } - if json.Unmarshal(raw, &summary) != nil || summary.Node.Name != node || summary.Pods == nil { - return nil, unverified - } - type podMountKey struct{ namespace, pod, uid, volume string } - lookup := make(map[podMountKey]providerFilesystemStats, len(mounts)) - for _, pod := range summary.Pods { - for _, volume := range pod.Volumes { - key := podMountKey{pod.Ref.Namespace, pod.Ref.Name, pod.Ref.UID, volume.Name} - if _, exists := lookup[key]; exists { - return nil, unverified - } - lookup[key] = volume - } - } - usage := make(map[string]uint64, len(mounts)) - for _, mount := range mounts { - stats, found := lookup[podMountKey{mount.Namespace, mount.Pod, mount.PodUID, mount.Volume}] - uid := mount.Claim.ClaimUID - _, repeated := usage[uid] - if !found || repeated || uid == "" || mount.PodUID == "" || mount.Node != node || - stats.PVCRef.Name != mount.Claim.Claim || stats.PVCRef.Namespace != mount.Claim.Namespace || - stats.Time.Before(now.Add(-3*time.Minute)) || stats.Time.After(now.Add(30*time.Second)) || - stats.Used == nil || stats.Available == nil || stats.Capacity == nil || *stats.Capacity == 0 || - *stats.Capacity > mount.Claim.RequestedBytes || *stats.Available > *stats.Capacity || - *stats.Used > *stats.Capacity-*stats.Available { - return nil, unverified - } - usage[uid] = *stats.Used - } - return usage, nil -} diff --git a/scripts/migrate-longhorn-v2/provider_volume_usage_test.go b/scripts/migrate-longhorn-v2/provider_volume_usage_test.go index 2086f3252..98b78ff09 100644 --- a/scripts/migrate-longhorn-v2/provider_volume_usage_test.go +++ b/scripts/migrate-longhorn-v2/provider_volume_usage_test.go @@ -1,51 +1,65 @@ package main import ( - "encoding/json" "slices" "testing" - "time" + + machine "github.com/siderolabs/talos/pkg/machinery/api/machine" ) -func TestProviderUsageBindsFreshStatsToTheObservedMount(t *testing.T) { - now := time.Date(2026, time.November, 10, 12, 0, 0, 0, time.UTC) - identity := providerVolumeClaim{Namespace: "provider-app", Claim: "data", ClaimUID: "claim-uid", RequestedBytes: 10 << 30} - mount := providerPodMount{Claim: identity, Node: "storage-worker", Namespace: "provider-app", Pod: "app-0", PodUID: "pod-uid", Volume: "data"} +func TestProviderUsageBindsNativeStatsToTheObservedMount(t *testing.T) { + claim := providerVolumeClaim{ClaimUID: "claim-uid", PV: "pvc-original", Handle: "123", RequestedBytes: 10 << 30} + mount := providerPodMount{Claim: claim, PodUID: "pod-uid"} + devices := map[string]string{"123": "/dev/sdb"} for _, test := range []struct { name string - edit func(object, object, object) + edit func(*machine.MountStat) []*machine.MountStat proven bool }{ - {"matching mounted filesystem", func(object, object, object) {}, true}, - {"replacement Pod", func(_ object, pod, _ object) { valueObject(pod, "podRef")["uid"] = "other" }, false}, - {"different claim", func(_, _ object, volume object) { valueObject(volume, "pvcRef")["name"] = "other" }, false}, - {"different node", func(summary, _, _ object) { valueObject(summary, "node")["nodeName"] = "other" }, false}, - {"old filesystem measurement", func(_, _ object, volume object) { volume["time"] = now.Add(-10 * time.Minute) }, false}, - {"missing usage", func(_, _ object, volume object) { delete(volume, "usedBytes") }, false}, - {"larger than the provider volume", func(_, _ object, volume object) { volume["capacityBytes"] = 11 << 30 }, false}, + {"matching filesystem", func(s *machine.MountStat) []*machine.MountStat { return []*machine.MountStat{s} }, true}, + {"replacement Pod", func(s *machine.MountStat) []*machine.MountStat { + s.MountedOn = "/var/lib/kubelet/pods/other/volumes/kubernetes.io~csi/pvc-original/mount" + return []*machine.MountStat{s} + }, false}, + {"different device", func(s *machine.MountStat) []*machine.MountStat { + s.Filesystem = "/dev/sdc" + return []*machine.MountStat{s} + }, false}, + {"duplicate mount path", func(s *machine.MountStat) []*machine.MountStat { return []*machine.MountStat{s, s} }, false}, + {"oversized filesystem", func(s *machine.MountStat) []*machine.MountStat { + s.Size = 11 << 30 + return []*machine.MountStat{s} + }, false}, } { t.Run(test.name, func(t *testing.T) { - volume := object{"name": "data", "pvcRef": object{"name": "data", "namespace": "provider-app"}, - "time": now.Add(-20 * time.Second), "usedBytes": 3 << 30, "availableBytes": 5 << 30, "capacityBytes": 9 << 30} - pod := object{"podRef": object{"uid": "pod-uid", "name": "app-0", "namespace": "provider-app"}, "volume": []object{volume}} - summary := object{"node": object{"nodeName": mount.Node}, "pods": []object{pod}} - test.edit(summary, pod, volume) - raw, err := json.Marshal(summary) - if err != nil { - t.Fatal(err) - } - got, err := readProviderNodeUsage(raw, mount.Node, []providerPodMount{mount}, now) + stat := &machine.MountStat{MountedOn: "/var/lib/kubelet/pods/pod-uid/volumes/kubernetes.io~csi/pvc-original/mount", + Filesystem: "/dev/sdb", Size: 9 << 30, Available: 5 << 30} + got, err := matchProviderMounts(test.edit(stat), []providerPodMount{mount}, devices) if test.proven { - if err != nil || got[identity.ClaimUID] != 3<<30 { + if err != nil || got[claim.ClaimUID] != 4<<30 { t.Fatalf("matching native usage was rejected: %v", err) } } else if err == nil { - t.Fatal("unrelated or incomplete filesystem stats qualified") + t.Fatal("unrelated or ambiguous filesystem usage qualified") } }) } } +func TestProviderDevicesBindAUniqueNativeDisk(t *testing.T) { + good := `{"metadata":{"namespace":"runtime","type":"Disks.block.talos.dev","phase":"running","id":"sdb"},"spec":{"dev_path":"/dev/sdb","symlinks":["/dev/disk/by-id/scsi-0HC_Volume_123"]}}` + mapper := `{"metadata":{"namespace":"runtime","type":"Disks.block.talos.dev","phase":"running","id":"dm-0"},"spec":{"dev_path":"/dev/dm-0"}}` + devices, err := parseProviderDevices([]byte(good + mapper)) + if err != nil || devices["123"] != "/dev/sdb" { + t.Fatalf("matching disk was rejected: %v", err) + } + for _, raw := range []string{good + good, good + "{", `{"metadata":{},"metadata":{},"spec":{}}`} { + if _, err := parseProviderDevices([]byte(raw)); err == nil { + t.Fatal("an ambiguous disk inventory qualified") + } + } +} + func TestProviderUsageKeepsUnmeasuredSourcesAtTheirFullRequest(t *testing.T) { claims := []providerVolumeClaim{ {Namespace: "provider-app", Claim: "data", ClaimUID: "live", RequestedBytes: 10 << 30}, @@ -89,7 +103,7 @@ func TestProviderFootprintNativeCommandHasOnlyFixedReadRoutes(t *testing.T) { {[]string{"persistentvolumeclaims", "--output=json", "--all-namespaces"}, true}, {[]string{"pods", "--output=json", "--namespace=openbao"}, true}, {[]string{"node", "prod-worker-2", "--output=json"}, true}, - {[]string{"--raw=/api/v1/nodes/prod-worker-1/proxy/stats/summary"}, true}, + {[]string{"--raw=/api/v1/nodes/prod-worker-1/proxy/stats/summary"}, false}, {[]string{"--raw=/api/v1/nodes/prod-worker-1/proxy/stats/summary?foo=bar"}, false}, {[]string{"--raw=/api/v1/nodes/prod-worker-1/proxy/exec"}, false}, {[]string{"--raw=/api/v1/nodes/another-node/proxy/stats/summary"}, false}, @@ -107,4 +121,9 @@ func TestProviderFootprintNativeCommandHasOnlyFixedReadRoutes(t *testing.T) { if providerFootprintReadAllowed("kubectl", "--context=other", "--request-timeout=20s", "get", "persistentvolumes", "--output=json") { t.Fatal("a different context was accepted") } + if !providerFootprintReadAllowed("talosctl", "--nodes=10.0.1.2", "get", "disks", "--output=json") || + providerFootprintReadAllowed("talosctl", "--nodes=192.0.2.4", "get", "disks", "--output=json") || + providerFootprintReadAllowed("talosctl", "--nodes=10.0.1.2", "get", "machineconfig", "--output=json") { + t.Fatal("Talos read scope differs from the fixed disk inventory") + } } From a915b61dc6ae4cadf3dae76316598e01253f778a Mon Sep 17 00:00:00 2001 From: Nikolai Emil Damm Date: Tue, 10 Nov 2026 17:29:10 +0100 Subject: [PATCH 4/4] fix(storage): use the native one-to-one Talos mount API --- internal/longhorninventory/native.go | 12 +++++++----- .../migrate-longhorn-v2/provider_volume_observe.go | 2 +- scripts/migrate-longhorn-v2/provider_volume_talos.go | 12 +++++++----- 3 files changed, 15 insertions(+), 11 deletions(-) diff --git a/internal/longhorninventory/native.go b/internal/longhorninventory/native.go index a8e6f2388..10ecc9d06 100644 --- a/internal/longhorninventory/native.go +++ b/internal/longhorninventory/native.go @@ -100,7 +100,7 @@ func (n Native) listFixedRoot(ctx context.Context, root string) (Listing, error) if size > maxMessageBytes || total > MaxMetadataBytes || len(listing.Entries) >= MaxEntries { return Listing{}, nativeFailure{code: "bounds"} } - if !n.validMetadata(item.GetMetadata()) { + if !ValidNodeMetadata(n.Target, item.GetMetadata()) { return Listing{}, nativeFailure{code: "metadata"} } if item.GetError() != "" { @@ -144,7 +144,7 @@ func (n Native) verifyEmptyPartitionHolders(ctx context.Context, devicePath stri } entry, err := stream.Recv() if err != nil || !active(ctx) || entry == nil || proto.Size(entry) > maxMessageBytes || - !n.validMetadata(entry.GetMetadata()) || entry.GetError() != "" || entry.GetLink() != "" || + !ValidNodeMetadata(n.Target, entry.GetMetadata()) || entry.GetError() != "" || entry.GetLink() != "" || entry.GetRelativeName() != "." || entry.GetName() != root || !entry.GetIsDir() || !os.FileMode(entry.GetMode()).IsDir() { return refused } @@ -176,7 +176,7 @@ func (n Native) BootID(ctx context.Context) (string, error) { return strings.TrimSuffix(string(content), "\n"), nil } if recvErr != nil || !active(ctx) || item == nil || proto.Size(item) > maxMessageBytes || - !n.validMetadata(item.GetMetadata()) || len(content)+len(item.GetBytes()) > 37 { + !ValidNodeMetadata(n.Target, item.GetMetadata()) || len(content)+len(item.GetBytes()) > 37 { return "", errNative } content = append(content, item.GetBytes()...) @@ -184,7 +184,9 @@ func (n Native) BootID(ctx context.Context) (string, error) { return "", errNative } -func (n Native) validMetadata(metadata proto.Message) bool { +// ValidNodeMetadata checks the optional response metadata of a native Talos +// one-to-one request. The caller must independently bracket the host identity. +func ValidNodeMetadata(target string, metadata proto.Message) bool { if metadata == nil { // Native One2One routing does not inject a metadata envelope. Hardware and // boot identity must be bracketed independently by the caller. @@ -202,7 +204,7 @@ func (n Native) validMetadata(metadata proto.Message) bool { return false } hostname := message.Get(fields.ByName("hostname")).String() - if hostname != "" && hostname != n.Target { + if hostname != "" && hostname != target { return false } statusField := fields.ByName("status") diff --git a/scripts/migrate-longhorn-v2/provider_volume_observe.go b/scripts/migrate-longhorn-v2/provider_volume_observe.go index 837514f7a..849f2405b 100644 --- a/scripts/migrate-longhorn-v2/provider_volume_observe.go +++ b/scripts/migrate-longhorn-v2/provider_volume_observe.go @@ -83,7 +83,7 @@ func observeProviderVolumeFootprint(ctx context.Context, c client, source string if err != nil { return providerVolumeUsageReceipt{}, errors.New("the protected Talos reader is unavailable") } - receipt, err := observeProviderVolumeUsage(ctx, c, talosClient) + receipt, err := observeProviderVolumeUsage(ctx, c, talosClient.MachineClient) closeErr := talosClient.Close() if err != nil { return providerVolumeUsageReceipt{}, err diff --git a/scripts/migrate-longhorn-v2/provider_volume_talos.go b/scripts/migrate-longhorn-v2/provider_volume_talos.go index 9d7c80b83..816c544ca 100644 --- a/scripts/migrate-longhorn-v2/provider_volume_talos.go +++ b/scripts/migrate-longhorn-v2/provider_volume_talos.go @@ -11,10 +11,12 @@ import ( "strings" "time" + "github.com/devantler-tech/platform/internal/longhorninventory" "github.com/devantler-tech/platform/internal/strictjson" machine "github.com/siderolabs/talos/pkg/machinery/api/machine" talos "github.com/siderolabs/talos/pkg/machinery/client" "google.golang.org/grpc" + "google.golang.org/protobuf/types/known/emptypb" ) const providerDeviceLink = "/dev/disk/by-id/scsi-0HC_Volume_" @@ -22,7 +24,7 @@ const providerDeviceLink = "/dev/disk/by-id/scsi-0HC_Volume_" var providerDiskPath = regexp.MustCompile(`^/dev/[a-z][a-z0-9-]+$`) type providerTalosMountAPI interface { - Mounts(context.Context, ...grpc.CallOption) (*machine.MountsResponse, error) + Mounts(context.Context, *emptypb.Empty, ...grpc.CallOption) (*machine.MountsResponse, error) } func readProviderDevices(ctx context.Context, c client, address string) (map[string]string, error) { @@ -77,12 +79,12 @@ func parseProviderDevices(raw []byte) (map[string]string, error) { } func readProviderMounts(ctx context.Context, api providerTalosMountAPI, address string, mounts []providerPodMount, devices map[string]string) (map[string]uint64, error) { - ctx, stop := context.WithTimeout(talos.WithNodes(ctx, address), 15*time.Second) + ctx, stop := context.WithTimeout(talos.WithNode(ctx, address), 15*time.Second) defer stop() - response, err := api.Mounts(ctx, grpc.MaxCallRecvMsgSize(4<<20)) + response, err := api.Mounts(ctx, &emptypb.Empty{}, grpc.MaxCallRecvMsgSize(4<<20)) messages := response.GetMessages() - if err != nil || len(messages) != 1 || messages[0].GetMetadata().GetHostname() != address || - messages[0].GetMetadata().GetError() != "" || messages[0].GetMetadata().GetStatus().GetCode() != 0 { + if err != nil || ctx.Err() != nil || len(messages) != 1 || messages[0] == nil || + !longhorninventory.ValidNodeMetadata(address, messages[0].GetMetadata()) { return nil, errors.New("native Talos mount statistics came from an unverified host") } return matchProviderMounts(messages[0].GetStats(), mounts, devices)