feat: support per-PVC adoption annotation at volume creation time

Add support for the `nasty-csi.io/adoptable: "true"` PVC annotation
to mark individual volumes for adoption at creation time, without
requiring the StorageClass-wide `markAdoptable` parameter.

The controller now initializes a k8s client and looks up the PVC
during CreateVolume to check for the annotation. This gives users
selective adoption — mark only important stateful volumes (databases,
etc.) without bulk-marking everything via the StorageClass.

Usage:
  apiVersion: v1
  kind: PersistentVolumeClaim
  metadata:
    name: my-database
    annotations:
      nasty-csi.io/adoptable: "true"
  spec:
    storageClassName: nasty-nfs
    ...

RBAC: no changes needed — the Helm chart already grants PVC get/list.

Author: fenio

pkg/driver/controller.go

@@ -13,6 +13,9 @@ import (
 	nastyapi "github.com/nasty-project/nasty-go"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
 	"k8s.io/klog/v2"
 )
 
@@ -174,6 +177,7 @@ type ControllerService struct {
 	csi.UnimplementedControllerServer
 	apiClient    nastyapi.ClientInterface
 	nodeRegistry *NodeRegistry
+	kubeClient   kubernetes.Interface
 	// publishedVolumes tracks volumes published to nodes with their readonly state.
 	// Key format: "volumeID:nodeID", value: readonly state.
 	// Used to detect incompatible re-publish attempts per CSI spec.
@@ -184,14 +188,45 @@ type ControllerService struct {
 
 // NewControllerService creates a new controller service.
 func NewControllerService(apiClient nastyapi.ClientInterface, nodeRegistry *NodeRegistry, clusterID string) *ControllerService {
+	var kubeClient kubernetes.Interface
+	if config, err := rest.InClusterConfig(); err == nil {
+		if cs, err := kubernetes.NewForConfig(config); err == nil {
+			kubeClient = cs
+		} else {
+			klog.Warningf("Failed to create k8s client: %v", err)
+		}
+	}
 	return &ControllerService{
 		apiClient:        apiClient,
 		nodeRegistry:     nodeRegistry,
+		kubeClient:       kubeClient,
 		clusterID:        clusterID,
 		publishedVolumes: make(map[string]bool),
 	}
 }
 
+// AnnotationAdoptable is the PVC annotation that marks a volume for adoption at creation time.
+const AnnotationAdoptable = "nasty-csi.io/adoptable"
+
+// pvcHasAdoptableAnnotation checks if the PVC that triggered this CreateVolume
+// has the nasty-csi.io/adoptable annotation set to "true".
+func (s *ControllerService) pvcHasAdoptableAnnotation(ctx context.Context, params map[string]string) bool {
+	if s.kubeClient == nil {
+		return false
+	}
+	pvcName := params["csi.storage.k8s.io/pvc/name"]
+	pvcNamespace := params["csi.storage.k8s.io/pvc/namespace"]
+	if pvcName == "" || pvcNamespace == "" {
+		return false
+	}
+	pvc, err := s.kubeClient.CoreV1().PersistentVolumeClaims(pvcNamespace).Get(ctx, pvcName, metav1.GetOptions{})
+	if err != nil {
+		klog.V(4).Infof("Failed to read PVC %s/%s for adoption annotation: %v", pvcNamespace, pvcName, err)
+		return false
+	}
+	return pvc.Annotations[AnnotationAdoptable] == VolumeContextValueTrue
+}
+
 // isDatasetPathVolumeID returns true if the volume ID is a full dataset path (new format).
 // New-format IDs contain "/" (e.g., "filesystem/parent/pvc-xxx"), while legacy IDs are plain names ("pvc-xxx").
 func isDatasetPathVolumeID(volumeID string) bool {

pkg/driver/controller_iscsi.go

@@ -164,6 +164,11 @@ func (s *ControllerService) createISCSIVolume(ctx context.Context, req *csi.Crea
 		return nil, err
 	}
 
+	// Per-PVC adoption annotation overrides StorageClass default
+	if !params.markAdoptable && s.pvcHasAdoptableAnnotation(ctx, req.GetParameters()) {
+		params.markAdoptable = true
+	}
+
 	klog.V(4).Infof("Creating iSCSI volume: %s with size: %d bytes", params.volumeName, params.requestedCapacity)
 
 	// Check if subvolume already exists (idempotency)

pkg/driver/controller_nfs.go

@@ -331,6 +331,11 @@ func (s *ControllerService) createNFSVolume(ctx context.Context, req *csi.Create
 		return nil, err
 	}
 
+	// Per-PVC adoption annotation overrides StorageClass default
+	if !params.markAdoptable && s.pvcHasAdoptableAnnotation(ctx, req.GetParameters()) {
+		params.markAdoptable = true
+	}
+
 	klog.V(4).Infof("Creating subvolume: %s/%s with capacity: %d bytes", params.filesystem, params.subvolumeName, params.requestedCapacity)
 
 	// Check if subvolume already exists (idempotency)

pkg/driver/controller_nvmeof.go

@@ -189,6 +189,11 @@ func (s *ControllerService) createNVMeOFVolume(ctx context.Context, req *csi.Cre
 		return nil, err
 	}
 
+	// Per-PVC adoption annotation overrides StorageClass default
+	if !params.markAdoptable && s.pvcHasAdoptableAnnotation(ctx, req.GetParameters()) {
+		params.markAdoptable = true
+	}
+
 	klog.V(4).Infof("Creating NVMe-oF volume: %s with size: %d bytes, NQN: %s",
 		params.volumeName, params.requestedCapacity, params.subsystemNQN)
 

pkg/driver/controller_smb.go

@@ -214,6 +214,11 @@ func (s *ControllerService) createSMBVolume(ctx context.Context, req *csi.Create
 		return nil, err
 	}
 
+	// Per-PVC adoption annotation overrides StorageClass default
+	if !params.markAdoptable && s.pvcHasAdoptableAnnotation(ctx, req.GetParameters()) {
+		params.markAdoptable = true
+	}
+
 	klog.V(4).Infof("Creating subvolume: %s/%s with capacity: %d bytes", params.filesystem, params.subvolumeName, params.requestedCapacity)
 
 	// Check if subvolume already exists (idempotency)