From a2d4cbd0e066831d53c0a852e26c392d7a55c2d2 Mon Sep 17 00:00:00 2001 From: Lana Andreasyan Date: Mon, 8 Jun 2026 22:54:11 +0000 Subject: [PATCH] chore: add storage capacity for templated nodes and update labels --- .../latest/templates/manager-deployment.yaml | 4 +- charts/latest/templates/manager-rbac.yaml | 20 +- charts/latest/values.yaml | 21 + cmd/manager/main.go | 27 +- docs/design/capacity-template.md | 148 +++++++ docs/design/webhooks.md | 2 +- internal/csi/core/lvm/controller_test.go | 17 +- internal/csi/core/lvm/lvm.go | 8 +- internal/csi/node/node.go | 2 +- internal/csi/node/node_test.go | 2 +- .../manager/capacitytemplate/controller.go | 376 ++++++++++++++++++ .../capacitytemplate/controller_test.go | 268 +++++++++++++ internal/manager/pvcleanup/controller.go | 2 +- internal/manager/pvcleanup/controller_test.go | 2 +- test/aks/aks.go | 2 +- 15 files changed, 880 insertions(+), 21 deletions(-) create mode 100644 docs/design/capacity-template.md create mode 100644 internal/manager/capacitytemplate/controller.go create mode 100644 internal/manager/capacitytemplate/controller_test.go diff --git a/charts/latest/templates/manager-deployment.yaml b/charts/latest/templates/manager-deployment.yaml index 564903a6..06a5a660 100644 --- a/charts/latest/templates/manager-deployment.yaml +++ b/charts/latest/templates/manager-deployment.yaml @@ -1,4 +1,4 @@ -{{- if or .Values.webhook.enforceEphemeral.enabled .Values.webhook.hyperconverged.enabled .Values.cleanup.pvCleanup.enabled }} +{{- if or .Values.webhook.enforceEphemeral.enabled .Values.webhook.hyperconverged.enabled .Values.cleanup.pvCleanup.enabled .Values.cleanup.capacityTemplate.enabled }} apiVersion: apps/v1 kind: Deployment metadata: @@ -61,6 +61,8 @@ spec: - --kube-api-burst={{ .Values.scalability.manager.kubeApi.burst }} - --v={{ .Values.observability.manager.log.level }} - --enable-pv-cleanup={{ .Values.cleanup.pvCleanup.enabled }} + - --enable-capacity-template={{ .Values.cleanup.capacityTemplate.enabled }} + - --capacity-template-node-group-label={{ .Values.cleanup.capacityTemplate.nodeGroupLabelKey }} {{- if .Values.webhook.enforceEphemeral.enabled }} - --enforce-ephemeral-webhook-config={{ .Values.name }}-enforce-ephemeral {{- end }} diff --git a/charts/latest/templates/manager-rbac.yaml b/charts/latest/templates/manager-rbac.yaml index 0fafd6f9..b90da6b2 100644 --- a/charts/latest/templates/manager-rbac.yaml +++ b/charts/latest/templates/manager-rbac.yaml @@ -1,4 +1,4 @@ -{{- if or .Values.webhook.enforceEphemeral.enabled .Values.webhook.hyperconverged.enabled .Values.cleanup.pvCleanup.enabled }} +{{- if or .Values.webhook.enforceEphemeral.enabled .Values.webhook.hyperconverged.enabled .Values.cleanup.pvCleanup.enabled .Values.cleanup.capacityTemplate.enabled }} apiVersion: rbac.authorization.k8s.io/v1 kind: ClusterRole metadata: @@ -40,7 +40,7 @@ rules: - list - watch {{- end }} -{{- if or .Values.webhook.enforceEphemeral.enabled .Values.webhook.hyperconverged.enabled }} +{{- if or .Values.webhook.enforceEphemeral.enabled .Values.webhook.hyperconverged.enabled .Values.cleanup.capacityTemplate.enabled }} - apiGroups: - storage.k8s.io resources: @@ -70,7 +70,7 @@ rules: - list - watch {{- end }} -{{- if or .Values.webhook.hyperconverged.enabled .Values.cleanup.pvCleanup.enabled }} +{{- if or .Values.webhook.hyperconverged.enabled .Values.cleanup.pvCleanup.enabled .Values.cleanup.capacityTemplate.enabled }} - apiGroups: - "" resources: @@ -80,6 +80,20 @@ rules: - list - watch {{- end }} +{{- if .Values.cleanup.capacityTemplate.enabled }} +- apiGroups: + - storage.k8s.io + resources: + - csistoragecapacities + verbs: + - get + - list + - watch + - create + - update + - patch + - delete +{{- end }} - apiGroups: - "" resources: diff --git a/charts/latest/values.yaml b/charts/latest/values.yaml index 07c992f8..3d67ab40 100644 --- a/charts/latest/values.yaml +++ b/charts/latest/values.yaml @@ -102,6 +102,27 @@ cleanup: # Enable the PV cleanup controller in the manager deployment enabled: true + # Capacity-template controller configuration. + # When enabled, the manager publishes one CSIStorageCapacity per + # (StorageClass x node group) so that Cluster Autoscaler can observe + # non-zero capacity on its simulated template nodes and trigger scale-up. + # Each object's NodeTopology is constrained to template nodes only via the + # 'cluster-autoscaler.kubernetes.io/template-node=true' label that CA + # attaches to its simulated nodes (see kubernetes/autoscaler#9702), so + # these synthetic objects do not shadow real per-node capacity reported by + # the external-provisioner sidecar. + # StorageClasses opt in by setting the annotation + # 'localdisk.csi.acstor.io/template-capacity' to a resource quantity + # (e.g. '1800Gi'). + capacityTemplate: + enabled: false + # Node label whose values define a node group. Defaults to the upstream + # instance-type label (i.e. the VM SKU), since local NVMe capacity is + # determined by the SKU. Other common choices: + # kubernetes.azure.com/agentpool - AKS node pool + # topology.kubernetes.io/zone - failure domain + nodeGroupLabelKey: node.kubernetes.io/instance-type + # Webhook configuration. webhook: enforceEphemeral: diff --git a/cmd/manager/main.go b/cmd/manager/main.go index 4c7108aa..7359df87 100644 --- a/cmd/manager/main.go +++ b/cmd/manager/main.go @@ -24,6 +24,7 @@ import ( metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server" "sigs.k8s.io/controller-runtime/pkg/webhook" + "local-csi-driver/internal/manager/capacitytemplate" "local-csi-driver/internal/manager/pvcleanup" "local-csi-driver/internal/pkg/events" "local-csi-driver/internal/pkg/version" @@ -71,6 +72,8 @@ func main() { var printVersionAndExit bool var eventRecorderEnabled bool var enablePVCleanup bool + var enableCapacityTemplate bool + var capacityTemplateNodeGroupLabel string flag.StringVar(&namespace, "namespace", "default", "The namespace to use for creating objects.") @@ -109,6 +112,12 @@ func main() { "If enabled, the webhook will use the event recorder to record events. This is useful for debugging and monitoring purposes.") flag.BoolVar(&enablePVCleanup, "enable-pv-cleanup", true, "If enabled, the PV cleanup controller will be registered to clean up PVs when nodes are unavailable.") + flag.BoolVar(&enableCapacityTemplate, "enable-capacity-template", false, + "If enabled, publish one CSIStorageCapacity per (StorageClass x node group) so Cluster Autoscaler can observe non-zero capacity on simulated template nodes.") + flag.StringVar(&capacityTemplateNodeGroupLabel, "capacity-template-node-group-label", capacitytemplate.DefaultNodeGroupLabelKey, + "Node label key whose values define a node group for the capacity-template controller. "+ + "Defaults to node.kubernetes.io/instance-type (VM SKU). Other useful values: "+ + "kubernetes.azure.com/agentpool (AKS node pool), topology.kubernetes.io/zone.") // Initialize logger flags config. logConfig := textlogger.NewConfig(textlogger.VerbosityFlagName("v")) @@ -127,7 +136,7 @@ func main() { log.Info("starting webhook server") // Validate required flags - if enforceEphemeralWebhookConfig == "" && hyperconvergedWebhookConfig == "" && !enablePVCleanup { + if enforceEphemeralWebhookConfig == "" && hyperconvergedWebhookConfig == "" && !enablePVCleanup && !enableCapacityTemplate { logAndExit(fmt.Errorf("at least one feature must be enabled"), "no webhooks or PV cleanup controller configured") } @@ -299,6 +308,22 @@ func main() { log.Info("PV cleanup controller disabled") } + // Register capacity-template controller if enabled + if enableCapacityTemplate { + log.Info("registering capacity-template controller", + "nodeGroupLabelKey", capacityTemplateNodeGroupLabel) + if err := (&capacitytemplate.Reconciler{ + Client: mgr.GetClient(), + Namespace: namespace, + NodeGroupLabelKey: capacityTemplateNodeGroupLabel, + }).SetupWithManager(mgr); err != nil { + logAndExit(err, "unable to register capacity-template controller") + } + log.Info("capacity-template controller registered successfully") + } else { + log.Info("capacity-template controller disabled") + } + // Register webhooks once certificates are ready go func() { <-certSetupFinished diff --git a/docs/design/capacity-template.md b/docs/design/capacity-template.md new file mode 100644 index 00000000..7b7d834e --- /dev/null +++ b/docs/design/capacity-template.md @@ -0,0 +1,148 @@ +# local-csi-driver Capacity-Template Controller + +## Scenario + +The Kubernetes Cluster Autoscaler (CA) decides whether scaling up a node group +will help a pending pod by simulating scheduling onto a *template node* +synthesised from an existing node in that group. When a pod's PVC uses a CSI +driver whose `CSIStorageCapacity` objects are *node-specific* (e.g. selected by +`kubernetes.io/hostname`), no capacity object exists for the simulated template +node, the scheduler's storage-capacity check fails, and CA refuses to scale up +the group. This is upstream issue +[kubernetes/autoscaler#9700][autoscaler-9700]. + +The local-csi-driver runs the external-provisioner sidecar in node-local mode +(`--node-deployment`, with `Topology=true` and `--strict-topology`) so each +DaemonSet pod handles its own Provision/Delete requests. Capacity tracking is +a separate feature, enabled by `--enable-capacity` on the same sidecar; the +combination of strict per-node topology and capacity tracking causes each +DaemonSet pod to publish a `CSIStorageCapacity` keyed by hostname for +accurate per-node accounting. That accuracy is exactly what breaks CA's +simulation - the simulated template node has a synthetic hostname for which +no capacity object exists. + +## Goals + +- Allow CA to scale up node groups whose nodes use the local-csi-driver, + without giving up per-node capacity accuracy for live nodes. +- Work on any cluster (AKS, vanilla Kubernetes, Karpenter) and any grouping + dimension (VM SKU, AKS agent pool, zone) without code changes. +- Avoid touching `CSIStorageCapacity` objects published by the + external-provisioner sidecar. + +## Non-goals + +- Reporting accurate free capacity. The published value is an operator-supplied + *template* quantity, used only to satisfy CA's simulation. +- Replacing the external-provisioner's per-node capacity reporting for real + nodes. + +## Design + +### Mitigation in upstream Cluster Autoscaler + +[kubernetes/autoscaler#9702][autoscaler-9702] adds the label +`cluster-autoscaler.kubernetes.io/template-node=true` to every template node +CA generates during scale-up simulation. CSI vendors can use this label in a +dedicated `CSIStorageCapacity.NodeTopology` selector that matches only +template nodes, leaving real-node capacity reporting untouched. + +### Capacity-template controller + +A controller in the `local-csi-manager` Deployment publishes one +`CSIStorageCapacity` per `(StorageClass x node group)` pair. + +**Opt-in.** A `StorageClass` whose provisioner is `localdisk.csi.acstor.io` +opts in by setting the annotation: + +```yaml +metadata: + annotations: + localdisk.csi.acstor.io/template-capacity: "1800Gi" +``` + +The value is parsed as a `resource.Quantity` and used as the published +capacity for every node group. + +**Grouping.** Nodes are grouped by a configurable label (flag +`--capacity-template-node-group-label`). The default is +`node.kubernetes.io/instance-type` (VM SKU) because local NVMe capacity is a +property of the VM SKU, not of an arbitrary pool name, and the upstream +instance-type label is portable across cloud providers. Other useful values: + +| Label | Use case | +| ---------------------------------- | ----------------------------------------- | +| `node.kubernetes.io/instance-type` | VM SKU (default; portable, matches NVMe) | +| `kubernetes.azure.com/agentpool` | AKS node pool (per-pool overrides) | +| `topology.kubernetes.io/zone` | Failure domain (zonal capacity skew) | + +**Topology selector.** Each managed `CSIStorageCapacity.NodeTopology` +selects on **both** labels: + +```yaml +nodeTopology: + matchLabels: + node.kubernetes.io/instance-type: Standard_L8s_v3 + cluster-autoscaler.kubernetes.io/template-node: "true" +``` + +The `template-node=true` constraint means the object only matches CA's +simulated template nodes. Real nodes (which never carry that label) continue +to use the per-node objects published by the external-provisioner sidecar. + +**Reconciliation.** The controller does a full sync on every event and on a +periodic 5-minute resync. On each reconcile it: + +1. Lists `StorageClasses` with provisioner `localdisk.csi.acstor.io` and the + opt-in annotation. +2. Lists `Nodes` and collects the distinct values of the configured + group label. +3. For each `(class, group)` pair, creates or updates a `CSIStorageCapacity` + in the manager's namespace. +4. Garbage-collects managed objects whose `(class, group)` is no longer + desired. Managed objects are identified by the + `localdisk.csi.acstor.io/managed-by=capacitytemplate` label, so objects + published by the external-provisioner sidecar are never touched. + +Each managed object also carries `localdisk.csi.acstor.io/storageclass` and +`localdisk.csi.acstor.io/node-group` labels for human inspection. + +**Naming.** Objects are named `local-csi-template--`, +sanitised to DNS-1123 and truncated with a short hash if longer than 253 +characters. + +**Watches.** The reconciler watches: + +- `StorageClass` (filtered by provisioner) - opt-in changes. +- `Node` - filtered to events that add, remove, or change the configured + group label. +- Managed `CSIStorageCapacity` - filtered to objects with the managed-by + label, so external changes trigger reconvergence. + +### Disabled by default + +The controller is opt-in via `--enable-capacity-template` (Helm value +`cleanup.capacityTemplate.enabled`), since clusters that do not use Cluster +Autoscaler do not need it. + +## Limitations + +- The published capacity is a single per-StorageClass value applied to every + group. If two groups using the same grouping label value need different + template capacities, switch the grouping label to a finer-grained one + (e.g. agent pool instead of SKU). +- The mitigation requires CA at the version that adds the + `template-node=true` label (kubernetes/autoscaler#9702). Older CAs will + not match these capacity objects, but they will also not be harmed by them. + +## References + +- [kubernetes/autoscaler#9700][autoscaler-9700] - scale-up regression with + node-specific `CSIStorageCapacity`. +- [kubernetes/autoscaler#9702][autoscaler-9702] - template-node label + mitigation. +- [Kubernetes storage capacity tracking][k8s-csc]. + +[autoscaler-9700]: https://github.com/kubernetes/autoscaler/issues/9700 +[autoscaler-9702]: https://github.com/kubernetes/autoscaler/pull/9702 +[k8s-csc]: https://kubernetes.io/docs/concepts/storage/storage-capacity/ diff --git a/docs/design/webhooks.md b/docs/design/webhooks.md index 5519564b..a5f942b1 100644 --- a/docs/design/webhooks.md +++ b/docs/design/webhooks.md @@ -113,7 +113,7 @@ spec: required: nodeSelectorTerms: - matchExpressions: - - key: topology.localdisk.csi.acstor.io/node + - key: kubernetes.io/hostname operator: In values: - nvme-node-0 diff --git a/internal/csi/core/lvm/controller_test.go b/internal/csi/core/lvm/controller_test.go index e9fefe0b..b25562da 100644 --- a/internal/csi/core/lvm/controller_test.go +++ b/internal/csi/core/lvm/controller_test.go @@ -11,6 +11,7 @@ import ( "github.com/container-storage-interface/spec/lib/go/csi" "go.uber.org/mock/gomock" + corev1 "k8s.io/api/core/v1" "local-csi-driver/internal/csi/core/lvm" "local-csi-driver/internal/pkg/block" @@ -58,7 +59,7 @@ func TestLVM_Create(t *testing.T) { AccessibleTopology: []*csi.Topology{ { Segments: map[string]string{ - "topology.localdisk.csi.acstor.io/node": "nodename", + corev1.LabelHostname: "nodename", }, }, }, @@ -94,7 +95,7 @@ func TestLVM_Create(t *testing.T) { AccessibleTopology: []*csi.Topology{ { Segments: map[string]string{ - "topology.localdisk.csi.acstor.io/node": "nodename", + corev1.LabelHostname: "nodename", }, }, }, @@ -127,7 +128,7 @@ func TestLVM_Create(t *testing.T) { AccessibleTopology: []*csi.Topology{ { Segments: map[string]string{ - "topology.localdisk.csi.acstor.io/node": "nodename", + corev1.LabelHostname: "nodename", }, }, }, @@ -160,7 +161,7 @@ func TestLVM_Create(t *testing.T) { AccessibleTopology: []*csi.Topology{ { Segments: map[string]string{ - "topology.localdisk.csi.acstor.io/node": "nodename", + corev1.LabelHostname: "nodename", }, }, }, @@ -194,7 +195,7 @@ func TestLVM_Create(t *testing.T) { AccessibleTopology: []*csi.Topology{ { Segments: map[string]string{ - "topology.localdisk.csi.acstor.io/node": "nodename", + corev1.LabelHostname: "nodename", }, }, }, @@ -247,7 +248,7 @@ func TestLVM_Create(t *testing.T) { AccessibleTopology: []*csi.Topology{ { Segments: map[string]string{ - "topology.localdisk.csi.acstor.io/node": "nodename", + corev1.LabelHostname: "nodename", }, }, }, @@ -282,7 +283,7 @@ func TestLVM_Create(t *testing.T) { AccessibleTopology: []*csi.Topology{ { Segments: map[string]string{ - "topology.localdisk.csi.acstor.io/node": "nodename", + corev1.LabelHostname: "nodename", }, }, }, @@ -318,7 +319,7 @@ func TestLVM_Create(t *testing.T) { AccessibleTopology: []*csi.Topology{ { Segments: map[string]string{ - "topology.localdisk.csi.acstor.io/node": "nodename", + corev1.LabelHostname: "nodename", }, }, }, diff --git a/internal/csi/core/lvm/lvm.go b/internal/csi/core/lvm/lvm.go index ef6484ba..ed7d25aa 100644 --- a/internal/csi/core/lvm/lvm.go +++ b/internal/csi/core/lvm/lvm.go @@ -41,8 +41,12 @@ const ( DefaultVolumeGroupTag = "local-csi" // TopologyKey is the expected key used in the volume request to specify the - // node where the volume should be placed. - TopologyKey = "topology.localdisk.csi.acstor.io/node" + // node where the volume should be placed. We use the standard + // kubernetes.io/hostname Node label so that PV nodeAffinity and + // CSIStorageCapacity nodeTopology selectors target a label that already + // exists on every Node, rather than a driver-specific label managed by + // kubelet's CSI registration. + TopologyKey = corev1.LabelHostname // Raid0LvType is the logical volume type for raid0. raid0LvType = "raid0" diff --git a/internal/csi/node/node.go b/internal/csi/node/node.go index bca6f9d4..c6e9a034 100644 --- a/internal/csi/node/node.go +++ b/internal/csi/node/node.go @@ -594,7 +594,7 @@ func (ns *Server) NodeExpandVolume(ctx context.Context, req *csi.NodeExpandVolum // topologyKey returns the key for the topology map. func (ns *Server) topologyKey() string { - return fmt.Sprintf("topology.%s/node", ns.driver) + return lvm.TopologyKey } // topologyValue returns the value for the topology map. diff --git a/internal/csi/node/node_test.go b/internal/csi/node/node_test.go index 70af8660..2e83323f 100644 --- a/internal/csi/node/node_test.go +++ b/internal/csi/node/node_test.go @@ -48,7 +48,7 @@ const ( isNotMountPoint = validTargetPath + "notmount" invalidTargetPath = "invalidTargetPath" permissionError = "permissionError" - testTopologyKey = "topology." + driverName + "/node" + testTopologyKey = corev1.LabelHostname selectedNodeAnnotation = "testlocaldisk.csi.acstor.io/selected-node" selectedInitialNodeParam = "testlocaldisk.csi.acstor.io/selected-initial-node" ) diff --git a/internal/manager/capacitytemplate/controller.go b/internal/manager/capacitytemplate/controller.go new file mode 100644 index 00000000..1fcf0d43 --- /dev/null +++ b/internal/manager/capacitytemplate/controller.go @@ -0,0 +1,376 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +// Package capacitytemplate publishes one CSIStorageCapacity per +// (StorageClass x node group) so that the Kubernetes Cluster Autoscaler can +// observe non-zero capacity for the local-csi-driver on simulated template +// nodes and trigger scale-up. +// +// A StorageClass opts in by setting the annotation +// "localdisk.csi.acstor.io/template-capacity" to a parseable resource +// quantity (for example, "1800Gi"). The controller groups Nodes by a +// configurable label (default "node.kubernetes.io/instance-type", i.e. VM +// SKU) and publishes one CSIStorageCapacity object per (StorageClass x +// group) pair, scoped to the manager namespace. +// +// The grouping label is configurable: typical choices are +// - node.kubernetes.io/instance-type (default; VM SKU, portable across +// cloud providers and well-suited to local NVMe whose capacity is +// determined by the SKU) +// - kubernetes.azure.com/agentpool (AKS node pool) +// - topology.kubernetes.io/zone (failure domain) +// +// Each managed CSIStorageCapacity selects nodes by both the grouping label +// and "cluster-autoscaler.kubernetes.io/template-node=true", which Cluster +// Autoscaler attaches to its simulated template nodes (see +// kubernetes/autoscaler#9702). This restricts these synthetic capacity +// objects to template nodes only, so they do not shadow the accurate +// per-node capacity that the external-provisioner sidecar publishes for +// real nodes. +package capacitytemplate + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "strings" + "time" + + corev1 "k8s.io/api/core/v1" + storagev1 "k8s.io/api/storage/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/api/resource" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/builder" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/event" + "sigs.k8s.io/controller-runtime/pkg/handler" + "sigs.k8s.io/controller-runtime/pkg/log" + "sigs.k8s.io/controller-runtime/pkg/predicate" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + + "local-csi-driver/internal/csi/core/lvm" +) + +const ( + // CapacityAnnotation is the StorageClass annotation that opts a class in + // to template capacity publishing and provides the per-pool capacity + // value as a resource.Quantity (e.g. "1800Gi"). + CapacityAnnotation = "localdisk.csi.acstor.io/template-capacity" + + // ManagedByLabel marks CSIStorageCapacity objects owned by this + // controller so that stale objects can be reclaimed without touching + // capacity objects published by the external-provisioner sidecar. + ManagedByLabel = "localdisk.csi.acstor.io/managed-by" + ManagedByValue = "capacitytemplate" + + // StorageClassLabel and NodeGroupLabel are added to managed objects so + // they can be identified without parsing the name. + StorageClassLabel = "localdisk.csi.acstor.io/storageclass" + NodeGroupLabel = "localdisk.csi.acstor.io/node-group" + + // DefaultNodeGroupLabelKey is the well-known instance-type label and the + // controller's default grouping key. Local NVMe capacity is determined + // by the VM SKU, so SKU is the natural unit and works on any cloud + // provider that sets the upstream label. + DefaultNodeGroupLabelKey = "node.kubernetes.io/instance-type" + + // TemplateNodeLabelKey is the label Cluster Autoscaler attaches to its + // simulated template nodes. Including it in NodeTopology restricts our + // synthetic capacity objects to template nodes only, so they don't + // shadow accurate per-node capacity reported for real nodes by the + // external-provisioner sidecar. See kubernetes/autoscaler#9702. + TemplateNodeLabelKey = "cluster-autoscaler.kubernetes.io/template-node" + TemplateNodeLabelValue = "true" + + // namePrefix is the prefix used for all CSIStorageCapacity objects + // managed by this controller. + namePrefix = "local-csi-template-" + + // resyncRequeue forces a full reconciliation periodically as a safety + // net against missed events (e.g. StorageClass annotation edits that do + // not change the resource version we are watching). + resyncRequeue = 5 * time.Minute + + // dns1123MaxLen is the maximum length for an object name. + dns1123MaxLen = 253 +) + +// syntheticRequest is the single key used for full-sync reconciliation. The +// controller does not have a meaningful per-object key because each +// reconciliation evaluates the entire (StorageClass x pool) matrix. +var syntheticRequest = reconcile.Request{ + NamespacedName: types.NamespacedName{Name: "capacitytemplate-sync"}, +} + +// Reconciler publishes per-node-group CSIStorageCapacity objects. +type Reconciler struct { + client.Client + + // Namespace is the namespace where CSIStorageCapacity objects are + // created. Typically the manager pod's namespace. + Namespace string + + // NodeGroupLabelKey is the Node label whose values define a node group. + // Defaults to DefaultNodeGroupLabelKey when empty. + NodeGroupLabelKey string +} + +// Reconcile performs a full sync of the desired CSIStorageCapacity set. +func (r *Reconciler) Reconcile(ctx context.Context, _ ctrl.Request) (ctrl.Result, error) { + logger := log.FromContext(ctx) + groupKey := r.nodeGroupLabelKey() + + // 1. Gather StorageClasses that opt in. + var scList storagev1.StorageClassList + if err := r.List(ctx, &scList); err != nil { + return ctrl.Result{}, fmt.Errorf("list storageclasses: %w", err) + } + type scEntry struct { + name string + capacity resource.Quantity + } + var classes []scEntry + for i := range scList.Items { + sc := &scList.Items[i] + if sc.Provisioner != lvm.DriverName { + continue + } + raw, ok := sc.Annotations[CapacityAnnotation] + if !ok { + continue + } + qty, err := resource.ParseQuantity(raw) + if err != nil { + logger.Info("skipping StorageClass with invalid template-capacity annotation", + "storageclass", sc.Name, "value", raw, "error", err.Error()) + continue + } + classes = append(classes, scEntry{name: sc.Name, capacity: qty}) + } + + // 2. Gather distinct node-group values from Nodes. + var nodes corev1.NodeList + if err := r.List(ctx, &nodes); err != nil { + return ctrl.Result{}, fmt.Errorf("list nodes: %w", err) + } + groups := map[string]struct{}{} + for i := range nodes.Items { + v := nodes.Items[i].Labels[groupKey] + if v == "" { + continue + } + groups[v] = struct{}{} + } + + // 3. Build desired set and reconcile each. + type key struct{ sc, group string } + desired := map[key]struct{}{} + for _, sc := range classes { + for group := range groups { + desired[key{sc.name, group}] = struct{}{} + if err := r.upsert(ctx, sc.name, sc.capacity, group, groupKey); err != nil { + return ctrl.Result{}, err + } + } + } + + // 4. Garbage-collect managed objects that are no longer desired. + var existing storagev1.CSIStorageCapacityList + if err := r.List(ctx, &existing, + client.InNamespace(r.Namespace), + client.MatchingLabels{ManagedByLabel: ManagedByValue}, + ); err != nil { + return ctrl.Result{}, fmt.Errorf("list managed CSIStorageCapacity: %w", err) + } + for i := range existing.Items { + obj := &existing.Items[i] + k := key{ + sc: obj.Labels[StorageClassLabel], + group: obj.Labels[NodeGroupLabel], + } + if _, want := desired[k]; want { + continue + } + if err := r.Delete(ctx, obj); err != nil && !apierrors.IsNotFound(err) { + return ctrl.Result{}, fmt.Errorf("delete stale CSIStorageCapacity %s: %w", obj.Name, err) + } + logger.Info("deleted stale CSIStorageCapacity", "name", obj.Name, + "storageclass", k.sc, "nodeGroup", k.group) + } + + return ctrl.Result{RequeueAfter: resyncRequeue}, nil +} + +// upsert creates or updates the CSIStorageCapacity for (sc, group). +func (r *Reconciler) upsert(ctx context.Context, sc string, capacity resource.Quantity, group, groupKey string) error { + name := objectName(sc, group) + logger := log.FromContext(ctx).WithValues("name", name, "storageclass", sc, "nodeGroup", group) + + want := &storagev1.CSIStorageCapacity{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: r.Namespace, + Labels: map[string]string{ + ManagedByLabel: ManagedByValue, + StorageClassLabel: sc, + NodeGroupLabel: group, + }, + }, + StorageClassName: sc, + Capacity: &capacity, + NodeTopology: &metav1.LabelSelector{ + MatchLabels: map[string]string{ + groupKey: group, + TemplateNodeLabelKey: TemplateNodeLabelValue, + }, + }, + } + + existing := &storagev1.CSIStorageCapacity{} + err := r.Get(ctx, types.NamespacedName{Namespace: r.Namespace, Name: name}, existing) + if apierrors.IsNotFound(err) { + if err := r.Create(ctx, want); err != nil { + return fmt.Errorf("create CSIStorageCapacity %s: %w", name, err) + } + logger.Info("created CSIStorageCapacity", "capacity", capacity.String()) + return nil + } + if err != nil { + return fmt.Errorf("get CSIStorageCapacity %s: %w", name, err) + } + + if capacityEqual(existing, want) && labelsEqual(existing.Labels, want.Labels) && + topologyEqual(existing.NodeTopology, want.NodeTopology) && + existing.StorageClassName == want.StorageClassName { + return nil + } + existing.Labels = want.Labels + existing.StorageClassName = want.StorageClassName + existing.Capacity = want.Capacity + existing.NodeTopology = want.NodeTopology + if err := r.Update(ctx, existing); err != nil { + return fmt.Errorf("update CSIStorageCapacity %s: %w", name, err) + } + logger.Info("updated CSIStorageCapacity", "capacity", capacity.String()) + return nil +} + +func (r *Reconciler) nodeGroupLabelKey() string { + if r.NodeGroupLabelKey != "" { + return r.NodeGroupLabelKey + } + return DefaultNodeGroupLabelKey +} + +// SetupWithManager wires the reconciler to relevant watches. All events +// collapse into a single synthetic request so the reconciliation is a full +// sync rather than a per-object update. +func (r *Reconciler) SetupWithManager(mgr ctrl.Manager) error { + if r.Namespace == "" { + return fmt.Errorf("capacitytemplate: Namespace must be set") + } + + enqueueAll := handler.EnqueueRequestsFromMapFunc(func(context.Context, client.Object) []reconcile.Request { + return []reconcile.Request{syntheticRequest} + }) + + groupKey := r.nodeGroupLabelKey() + nodePred := predicate.Funcs{ + CreateFunc: func(e event.CreateEvent) bool { + return e.Object.GetLabels()[groupKey] != "" + }, + DeleteFunc: func(e event.DeleteEvent) bool { + return e.Object.GetLabels()[groupKey] != "" + }, + UpdateFunc: func(e event.UpdateEvent) bool { + return e.ObjectOld.GetLabels()[groupKey] != e.ObjectNew.GetLabels()[groupKey] + }, + GenericFunc: func(event.GenericEvent) bool { return false }, + } + + scPred := predicate.NewPredicateFuncs(func(obj client.Object) bool { + sc, ok := obj.(*storagev1.StorageClass) + if !ok { + return false + } + return sc.Provisioner == lvm.DriverName + }) + + managedPred := predicate.NewPredicateFuncs(func(obj client.Object) bool { + return obj.GetLabels()[ManagedByLabel] == ManagedByValue + }) + + return ctrl.NewControllerManagedBy(mgr). + Named("capacitytemplate"). + For(&storagev1.StorageClass{}, builder.WithPredicates(scPred)). + Watches(&corev1.Node{}, enqueueAll, builder.WithPredicates(nodePred)). + Watches(&storagev1.CSIStorageCapacity{}, enqueueAll, builder.WithPredicates(managedPred)). + Complete(r) +} + +// objectName produces a deterministic, DNS-1123 compliant name for the +// CSIStorageCapacity object of a (storageClass, group) pair. Long inputs are +// truncated and disambiguated with a short hash. +func objectName(sc, group string) string { + base := namePrefix + sanitize(sc) + "-" + sanitize(group) + if len(base) <= dns1123MaxLen { + return base + } + sum := sha256.Sum256([]byte(sc + "/" + group)) + suffix := "-" + hex.EncodeToString(sum[:4]) + return base[:dns1123MaxLen-len(suffix)] + suffix +} + +// sanitize replaces characters not allowed in DNS-1123 labels with '-'. +func sanitize(s string) string { + s = strings.ToLower(s) + var b strings.Builder + b.Grow(len(s)) + for _, r := range s { + switch { + case r >= 'a' && r <= 'z', r >= '0' && r <= '9', r == '-': + b.WriteRune(r) + default: + b.WriteByte('-') + } + } + return strings.Trim(b.String(), "-") +} + +func capacityEqual(a, b *storagev1.CSIStorageCapacity) bool { + if a.Capacity == nil || b.Capacity == nil { + return a.Capacity == b.Capacity + } + return a.Capacity.Cmp(*b.Capacity) == 0 +} + +func labelsEqual(a, b map[string]string) bool { + if len(a) != len(b) { + return false + } + for k, v := range a { + if b[k] != v { + return false + } + } + return true +} + +func topologyEqual(a, b *metav1.LabelSelector) bool { + if a == nil || b == nil { + return a == b + } + if len(a.MatchLabels) != len(b.MatchLabels) || len(a.MatchExpressions) != len(b.MatchExpressions) { + return false + } + for k, v := range a.MatchLabels { + if b.MatchLabels[k] != v { + return false + } + } + return true +} diff --git a/internal/manager/capacitytemplate/controller_test.go b/internal/manager/capacitytemplate/controller_test.go new file mode 100644 index 00000000..5974fdcb --- /dev/null +++ b/internal/manager/capacitytemplate/controller_test.go @@ -0,0 +1,268 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +package capacitytemplate + +import ( + "context" + "sort" + "strings" + "testing" + + corev1 "k8s.io/api/core/v1" + storagev1 "k8s.io/api/storage/v1" + "k8s.io/apimachinery/pkg/api/resource" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + clientgoscheme "k8s.io/client-go/kubernetes/scheme" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + "local-csi-driver/internal/csi/core/lvm" +) + +const testNamespace = "kube-system" + +func newScheme(t *testing.T) *runtime.Scheme { + t.Helper() + s := runtime.NewScheme() + if err := clientgoscheme.AddToScheme(s); err != nil { + t.Fatalf("add scheme: %v", err) + } + return s +} + +func newReconciler(t *testing.T, objs ...client.Object) (*Reconciler, client.Client) { + t.Helper() + c := fake.NewClientBuilder(). + WithScheme(newScheme(t)). + WithObjects(objs...). + Build() + return &Reconciler{ + Client: c, + Namespace: testNamespace, + }, c +} + +func node(name, group string) *corev1.Node { + return &corev1.Node{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Labels: map[string]string{DefaultNodeGroupLabelKey: group}, + }, + } +} + +func storageClass(name string, annotations map[string]string) *storagev1.StorageClass { + return &storagev1.StorageClass{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Annotations: annotations, + }, + Provisioner: lvm.DriverName, + } +} + +func listCapacities(t *testing.T, c client.Client) []storagev1.CSIStorageCapacity { + t.Helper() + var list storagev1.CSIStorageCapacityList + if err := c.List(context.Background(), &list, client.InNamespace(testNamespace)); err != nil { + t.Fatalf("list capacities: %v", err) + } + sort.Slice(list.Items, func(i, j int) bool { return list.Items[i].Name < list.Items[j].Name }) + return list.Items +} + +func TestReconcile_CreatesOnePerNodeGroupAndStorageClass(t *testing.T) { + sc := storageClass("local", map[string]string{CapacityAnnotation: "100Gi"}) + r, c := newReconciler(t, + sc, + node("n1", "groupA"), + node("n2", "groupA"), + node("n3", "groupB"), + ) + + if _, err := r.Reconcile(context.Background(), ctrl.Request{}); err != nil { + t.Fatalf("reconcile: %v", err) + } + + got := listCapacities(t, c) + if len(got) != 2 { + t.Fatalf("expected 2 CSIStorageCapacity, got %d", len(got)) + } + groups := map[string]string{} + for _, o := range got { + if o.Labels[ManagedByLabel] != ManagedByValue { + t.Errorf("%s missing managed-by label", o.Name) + } + if o.StorageClassName != "local" { + t.Errorf("%s storageClassName = %q, want local", o.Name, o.StorageClassName) + } + if o.Capacity == nil || o.Capacity.String() != "100Gi" { + t.Errorf("%s capacity = %v, want 100Gi", o.Name, o.Capacity) + } + if o.NodeTopology == nil || o.NodeTopology.MatchLabels[DefaultNodeGroupLabelKey] == "" { + t.Errorf("%s missing node-group topology selector", o.Name) + } + if o.NodeTopology != nil && o.NodeTopology.MatchLabels[TemplateNodeLabelKey] != TemplateNodeLabelValue { + t.Errorf("%s missing template-node selector label", o.Name) + } + groups[o.NodeTopology.MatchLabels[DefaultNodeGroupLabelKey]] = o.Name + } + if _, ok := groups["groupA"]; !ok { + t.Errorf("missing capacity for groupA: %v", groups) + } + if _, ok := groups["groupB"]; !ok { + t.Errorf("missing capacity for groupB: %v", groups) + } +} + +func TestReconcile_SkipsStorageClassesWithoutAnnotation(t *testing.T) { + r, c := newReconciler(t, + storageClass("local", nil), + storageClass("other", map[string]string{CapacityAnnotation: "1Gi"}), + node("n1", "groupA"), + ) + + if _, err := r.Reconcile(context.Background(), ctrl.Request{}); err != nil { + t.Fatalf("reconcile: %v", err) + } + + got := listCapacities(t, c) + if len(got) != 1 { + t.Fatalf("expected 1 CSIStorageCapacity, got %d (%v)", len(got), got) + } + if got[0].StorageClassName != "other" { + t.Errorf("got %q, want other", got[0].StorageClassName) + } +} + +func TestReconcile_SkipsForeignProvisioners(t *testing.T) { + foreign := &storagev1.StorageClass{ + ObjectMeta: metav1.ObjectMeta{ + Name: "foreign", + Annotations: map[string]string{CapacityAnnotation: "1Gi"}, + }, + Provisioner: "some.other/provisioner", + } + r, c := newReconciler(t, foreign, node("n1", "groupA")) + + if _, err := r.Reconcile(context.Background(), ctrl.Request{}); err != nil { + t.Fatalf("reconcile: %v", err) + } + + if got := listCapacities(t, c); len(got) != 0 { + t.Fatalf("expected 0 CSIStorageCapacity, got %d", len(got)) + } +} + +func TestReconcile_DeletesStaleCapacities(t *testing.T) { + // Pre-populate with a managed capacity for a (sc,group) that should no longer exist. + stale := &storagev1.CSIStorageCapacity{ + ObjectMeta: metav1.ObjectMeta{ + Name: objectName("local", "gone"), + Namespace: testNamespace, + Labels: map[string]string{ + ManagedByLabel: ManagedByValue, + StorageClassLabel: "local", + NodeGroupLabel: "gone", + }, + }, + StorageClassName: "local", + } + // Pre-populate with an UNMANAGED capacity that must be left alone. + foreign := &storagev1.CSIStorageCapacity{ + ObjectMeta: metav1.ObjectMeta{ + Name: "foreign-capacity", + Namespace: testNamespace, + }, + StorageClassName: "local", + } + r, c := newReconciler(t, + storageClass("local", map[string]string{CapacityAnnotation: "100Gi"}), + node("n1", "groupA"), + stale, + foreign, + ) + + if _, err := r.Reconcile(context.Background(), ctrl.Request{}); err != nil { + t.Fatalf("reconcile: %v", err) + } + + got := listCapacities(t, c) + names := map[string]bool{} + for _, o := range got { + names[o.Name] = true + } + if names[stale.Name] { + t.Errorf("stale managed capacity %q should have been deleted", stale.Name) + } + if !names["foreign-capacity"] { + t.Errorf("foreign-capacity should be preserved; got names: %v", names) + } +} + +func TestReconcile_UpdatesCapacityOnChange(t *testing.T) { + sc := storageClass("local", map[string]string{CapacityAnnotation: "100Gi"}) + r, c := newReconciler(t, sc, node("n1", "groupA")) + if _, err := r.Reconcile(context.Background(), ctrl.Request{}); err != nil { + t.Fatalf("first reconcile: %v", err) + } + + // Mutate annotation and reconcile again. + var fresh storagev1.StorageClass + if err := c.Get(context.Background(), client.ObjectKey{Name: "local"}, &fresh); err != nil { + t.Fatalf("get sc: %v", err) + } + fresh.Annotations[CapacityAnnotation] = "200Gi" + if err := c.Update(context.Background(), &fresh); err != nil { + t.Fatalf("update sc: %v", err) + } + if _, err := r.Reconcile(context.Background(), ctrl.Request{}); err != nil { + t.Fatalf("second reconcile: %v", err) + } + + got := listCapacities(t, c) + if len(got) != 1 { + t.Fatalf("expected 1, got %d", len(got)) + } + if got[0].Capacity.String() != "200Gi" { + t.Errorf("capacity = %s, want 200Gi", got[0].Capacity.String()) + } +} + +func TestObjectName_LongInputsHashed(t *testing.T) { + long := strings.Repeat("a", 300) + name := objectName(long, "group") + if len(name) > dns1123MaxLen { + t.Errorf("name length = %d, exceeds %d", len(name), dns1123MaxLen) + } + if !strings.HasPrefix(name, namePrefix) { + t.Errorf("name %q missing prefix %q", name, namePrefix) + } +} + +func TestObjectName_SanitisesUppercaseAndSymbols(t *testing.T) { + name := objectName("My_SC", "Group/01") + for _, r := range name { + if r >= 'A' && r <= 'Z' { + t.Errorf("name %q contains uppercase", name) + break + } + if r == '_' || r == '/' { + t.Errorf("name %q contains invalid char %q", name, r) + break + } + } +} + +func TestCapacityEqual(t *testing.T) { + q := resource.MustParse("1Gi") + q2 := resource.MustParse("1024Mi") + a := &storagev1.CSIStorageCapacity{Capacity: &q} + b := &storagev1.CSIStorageCapacity{Capacity: &q2} + if !capacityEqual(a, b) { + t.Errorf("1Gi and 1024Mi should be equal capacities") + } +} diff --git a/internal/manager/pvcleanup/controller.go b/internal/manager/pvcleanup/controller.go index d51ad33b..feee59e3 100644 --- a/internal/manager/pvcleanup/controller.go +++ b/internal/manager/pvcleanup/controller.go @@ -156,7 +156,7 @@ func (r *PVCleanupReconciler) Reconcile(ctx context.Context, req ctrl.Request) ( } // extractHostnamesFromPV extracts hostnames from PV's node affinity -// Specifically looks for topology.localdisk.csi.acstor.io/node topology constraints. +// Specifically looks for kubernetes.io/hostname topology constraints (lvm.TopologyKey). func extractHostnamesFromPV(pv *corev1.PersistentVolume) []string { if pv.Spec.NodeAffinity == nil || pv.Spec.NodeAffinity.Required == nil { return nil diff --git a/internal/manager/pvcleanup/controller_test.go b/internal/manager/pvcleanup/controller_test.go index 69515aa6..2f2eb6a0 100644 --- a/internal/manager/pvcleanup/controller_test.go +++ b/internal/manager/pvcleanup/controller_test.go @@ -95,7 +95,7 @@ func TestExtractHostnamesFromPV(t *testing.T) { { MatchExpressions: []corev1.NodeSelectorRequirement{ { - Key: "kubernetes.io/hostname", + Key: "topology.kubernetes.io/zone", Operator: corev1.NodeSelectorOpIn, Values: []string{"node-1"}, }, diff --git a/test/aks/aks.go b/test/aks/aks.go index 6e0bb648..5e0fbace 100644 --- a/test/aks/aks.go +++ b/test/aks/aks.go @@ -359,7 +359,7 @@ func pvAffinityForExistingNode(nodeAffinity *corev1.VolumeNodeAffinity, nodes ma for _, term := range nodeAffinity.Required.NodeSelectorTerms { for _, expr := range term.MatchExpressions { - if expr.Key == "topology.localdisk.csi.acstor.io/node" && expr.Operator == corev1.NodeSelectorOpIn { + if expr.Key == corev1.LabelHostname && expr.Operator == corev1.NodeSelectorOpIn { for _, value := range expr.Values { if _, exists := nodes[value]; exists { return true