diff --git a/.github/workflows/prepare-longhorn-v2-node.yaml b/.github/workflows/prepare-longhorn-v2-node.yaml index 1883ee19b..69a8e9f0b 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, 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 ]] ;; 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 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/README.md b/scripts/migrate-longhorn-v2/README.md index 37b9a51b8..ddcae85dc 100644 --- a/scripts/migrate-longhorn-v2/README.md +++ b/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-`. It reads only the native huge-page allocation and selected `/proc/meminfo` counters, bracketed by unchanged 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/main.go b/scripts/migrate-longhorn-v2/main.go index f589ff859..be8828e29 100644 --- a/scripts/migrate-longhorn-v2/main.go +++ b/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() 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_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..849f2405b --- /dev/null +++ b/scripts/migrate-longhorn-v2/provider_volume_observe.go @@ -0,0 +1,217 @@ +package main + +import ( + "context" + "errors" + "maps" + "net/netip" + "regexp" + "slices" + "strings" +) + +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, Hardware, Address 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 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") + } + return inventoryExecuteCommand(ctx, program, args...) +} + +func providerFootprintReadAllowed(program string, args ...string) bool { + 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 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 + } + talosClient, err := openInventoryTalosClient(ctx) + if err != nil { + return providerVolumeUsageReceipt{}, errors.New("the protected Talos reader is unavailable") + } + receipt, err := observeProviderVolumeUsage(ctx, c, talosClient.MachineClient) + 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 + } + 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)) + 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 + } + disks, err := readProviderDevices(ctx, c, epoch.Address) + if err != nil { + return providerVolumeUsageReceipt{}, err + } + 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") + } + } + 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, "metadata", "uid") == "" || text(n, "status", "nodeInfo", "bootID") == "" { + return providerStatsEpoch{}, errors.New("provider filesystem statistics worker is unverified") + } + 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..816c544ca --- /dev/null +++ b/scripts/migrate-longhorn-v2/provider_volume_talos.go @@ -0,0 +1,115 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "io" + "regexp" + "strconv" + "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_" + +var providerDiskPath = regexp.MustCompile(`^/dev/[a-z][a-z0-9-]+$`) + +type providerTalosMountAPI interface { + Mounts(context.Context, *emptypb.Empty, ...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.WithNode(ctx, address), 15*time.Second) + defer stop() + response, err := api.Mounts(ctx, &emptypb.Empty{}, grpc.MaxCallRecvMsgSize(4<<20)) + messages := response.GetMessages() + 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) +} + +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 new file mode 100644 index 000000000..20780dac9 --- /dev/null +++ b/scripts/migrate-longhorn-v2/provider_volume_usage.go @@ -0,0 +1,67 @@ +package main + +import ( + "errors" + "sort" +) + +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 || + !kubeNamespace.MatchString(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 +} 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..98b78ff09 --- /dev/null +++ b/scripts/migrate-longhorn-v2/provider_volume_usage_test.go @@ -0,0 +1,129 @@ +package main + +import ( + "slices" + "testing" + + machine "github.com/siderolabs/talos/pkg/machinery/api/machine" +) + +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(*machine.MountStat) []*machine.MountStat + proven bool + }{ + {"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) { + 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[claim.ClaimUID] != 4<<30 { + t.Fatalf("matching native usage was rejected: %v", err) + } + } else if err == nil { + 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}, + {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"}, 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}, + {[]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") + } + 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") + } +}