feat(storage): measure provider-backed volume usage

.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,
             observe-database,
@@ -141,6 +142,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 ]]
               ;;

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

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,15 +176,17 @@ 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()...)
 	}
 	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")

scripts/migrate-longhorn-v2/README.md

@@ -42,6 +42,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 Talos filesystem usage from their running mounts. Confirm with
+`inventory-provider-volume-footprint` on a first-attempt main dispatch.
+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.
+
 `inspect-host-memory` is a read-only diagnostic for a fixed permanent worker,
 confirmed with `inspect-longhorn-v2-host-memory-<worker>`. It reads only the native
 huge-page allocation and selected `/proc/meminfo` counters, bracketed by unchanged

scripts/migrate-longhorn-v2/inventory_runtime.go

@@ -130,49 +130,57 @@ 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))
 	closeErr := file.Close()
 	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 {

scripts/migrate-longhorn-v2/main.go

@@ -39,6 +39,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)
 	}
@@ -48,6 +50,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 == "observe-database" {
 		observationContext, stop := context.WithTimeout(ctx, 5*time.Minute)
 		defer stop()

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 {

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)
+}

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")
+			}
+		})
+	}
+}