nvmeof: add nvmeof ability

create prototype of nvmeof csi-driver. create\delete volume were implemented

Signed-off-by: gadi-didi <gadi.didi@ibm.com>

Author: gadididi

internal/nvmeof/controller/controllerserver.go

@@ -0,0 +1,268 @@
+package controller
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+
+	"github.com/container-storage-interface/spec/lib/go/csi"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+
+	csicommon "github.com/ceph/ceph-csi/internal/csi-common"
+	"github.com/ceph/ceph-csi/internal/nvmeof"
+	"github.com/ceph/ceph-csi/internal/rbd"
+	rbdutil "github.com/ceph/ceph-csi/internal/rbd"
+	rbddriver "github.com/ceph/ceph-csi/internal/rbd/driver"
+	"github.com/ceph/ceph-csi/internal/util/log"
+)
+
+type Server struct {
+	csi.UnimplementedControllerServer
+
+	// backendServer handles the RBD requests
+	backendServer *rbd.ControllerServer
+
+	// gateway handles NVMe-oF operations
+	gateway *nvmeof.GatewayRpcClient
+}
+
+// NewControllerServer initialize a controller server for nvmeof CSI driver.
+func NewControllerServer(d *csicommon.CSIDriver, gatewayConfig *nvmeof.GatewayConfig) (*Server, error) {
+	// Initialize gateway client
+	gw, err := nvmeof.NewGatewayRpcClient(gatewayConfig)
+	if err != nil {
+		return nil, status.Errorf(codes.Unavailable, err.Error(), "failed to create NVMe-oF Gateway client: %v", err)
+	}
+
+	return &Server{
+		backendServer: rbddriver.NewControllerServer(d),
+		gateway:       gw,
+	}, nil
+}
+
+// ControllerGetCapabilities uses the RBD backendServer to return the
+// capabilities that were set in the Driver.Run() function.
+func (cs *Server) ControllerGetCapabilities(
+	ctx context.Context,
+	req *csi.ControllerGetCapabilitiesRequest,
+) (*csi.ControllerGetCapabilitiesResponse, error) {
+	return cs.backendServer.ControllerGetCapabilities(ctx, req)
+}
+
+// ValidateVolumeCapabilities checks whether the volume capabilities requested
+// are supported.
+func (cs *Server) ValidateVolumeCapabilities(
+	ctx context.Context,
+	req *csi.ValidateVolumeCapabilitiesRequest,
+) (*csi.ValidateVolumeCapabilitiesResponse, error) {
+	return cs.backendServer.ValidateVolumeCapabilities(ctx, req)
+}
+
+// CreateVolume creates a new RBD volume and exposes it through NVMe-oF Gateway
+func (cs *Server) CreateVolume(
+	ctx context.Context,
+	req *csi.CreateVolumeRequest,
+) (*csi.CreateVolumeResponse, error) {
+
+	// TODO - maybe need to check and modify the request to ensure it has the required fields NvmeOF supports (like nfs does)
+	// TODO - in case of failure, we should clean up the created resources (RBD volume, subsystem, etc.)
+	// TODO - move all hardcoded strings to constants
+
+	// Step 1: Create RBD volume through backend. if exists, it is ok.
+	res, err := cs.backendServer.CreateVolume(ctx, req)
+	if err != nil {
+		log.ErrorLog(ctx, "failed to create RBD volume: %v", err)
+		return nil, err
+	}
+
+	backend := res.GetVolume()
+	volumeID := backend.GetVolumeId()
+	rbdImageName := res.GetVolume().GetVolumeContext()["imageName"]
+	rbdPoolName := res.GetVolume().GetVolumeContext()["pool"]
+	log.DebugLog(ctx, "RBD volume created: rbdPoolName=%s  rbdImageName=%s  ", rbdPoolName, rbdImageName)
+
+	// Step 2: check if subsystem exists, if not create it
+	subsystemNQN := req.GetParameters()["subsystemNQN"]
+	if subsystemNQN == "" {
+		return nil, status.Errorf(codes.NotFound, "subsystemNQN parameter is required")
+	}
+	if err := cs.ensureSubsystem(ctx, subsystemNQN); err != nil {
+		log.ErrorLog(ctx, "failed to ensure (OR create) subsystem %s exists: %v", subsystemNQN, err)
+		return nil, err
+	}
+	// Step 3: Create namespace in the NVMe-oF subsystem
+	nsid, err := cs.gateway.CreateNamespace(ctx, subsystemNQN, rbdPoolName, rbdImageName)
+	if err != nil {
+		log.ErrorLog(ctx, "failed to create namespace for RBD %s/%s in subsystem %s: %v",
+			rbdPoolName, rbdImageName, subsystemNQN, err)
+		return nil, err
+	}
+
+	log.DebugLog(ctx, "NVMe-oF namespace created successfully: %s/%s with NSID: %d",
+		rbdPoolName, rbdImageName, nsid)
+
+	// Step 4: store the namespace ID and subsystem nqn in the volume context for ControllerPublishVolume
+	backend.VolumeContext["nvmeof.SubsystemNQN"] = subsystemNQN
+	backend.VolumeContext["nvmeof.NamespaceID"] = fmt.Sprintf("%d", nsid)
+
+	// step 5: Store the subsystem NQN and namespace ID in the RBD volume metadata
+	mgr := rbdutil.NewManager(cs.backendServer.Driver.GetInstanceID(), nil, req.GetSecrets())
+	defer mgr.Destroy(ctx)
+
+	rbdVol, err := mgr.GetVolumeByID(ctx, volumeID)
+	if err != nil {
+		return nil, status.Errorf(codes.Aborted, "failed to find volume with ID %q: %s", volumeID, err.Error())
+	}
+	defer rbdVol.Destroy(ctx)
+
+	err = rbdVol.SetMetadata("nvmeof.SubsystemNQN", subsystemNQN)
+	if err != nil {
+		log.ErrorLog(ctx, "failed to store subsystem metadata: %v", err)
+		return nil, status.Errorf(codes.Internal, "failed to store subsystem metadata: %v", err)
+	}
+
+	err = rbdVol.SetMetadata("nvmeof.NamespaceID", fmt.Sprintf("%d", nsid))
+	if err != nil {
+		log.ErrorLog(ctx, "failed to store NSID metadata: %v", err)
+		return nil, status.Errorf(codes.Internal, "failed to store NSID metadata: %v", err)
+	}
+	return &csi.CreateVolumeResponse{Volume: backend}, nil
+}
+
+func (cs *Server) DeleteVolume(
+	ctx context.Context,
+	req *csi.DeleteVolumeRequest,
+) (*csi.DeleteVolumeResponse, error) {
+
+	log.DebugLog(ctx, "NVMe-oF DeleteVolume: %s", req.GetVolumeId())
+
+	volumeID := req.GetVolumeId()
+	if volumeID == "" {
+		return nil, fmt.Errorf("volume ID is required")
+	}
+	// Step 1: Get the RBD volume by ID
+	mgr := rbdutil.NewManager(cs.backendServer.Driver.GetInstanceID(), nil, req.GetSecrets())
+	defer mgr.Destroy(ctx)
+	rbdVol, err := mgr.GetVolumeByID(ctx, volumeID)
+	if err != nil {
+		return nil, status.Errorf(codes.Aborted, "failed to find volume with ID %q: %s", volumeID, err.Error())
+	}
+	defer rbdVol.Destroy(ctx)
+
+	// Retrieve subsystem NQN and namespace ID from volume context
+	subsystemNQN, err := rbdVol.GetMetadata("nvmeof.SubsystemNQN")
+
+	if err != nil {
+		log.DebugLog(ctx, "No subsystem metadata found: %v", err)
+		subsystemNQN = ""
+	}
+	nsidStr, err := rbdVol.GetMetadata("nvmeof.NamespaceID")
+	if err != nil {
+		log.DebugLog(ctx, "No NSID metadata found: %v", err)
+		nsidStr = ""
+	}
+
+	// step 1:  Delete the namespace from the NVMe-oF subsystem first
+	if subsystemNQN != "" && nsidStr != "" {
+		nsid, err := strconv.ParseUint(nsidStr, 10, 32)
+		if err != nil {
+			log.ErrorLog(ctx, "failed to parse namespace ID %s: %v", nsidStr, err)
+			return nil, status.Error(codes.InvalidArgument, err.Error())
+		}
+		err = cs.gateway.DeleteNamespace(ctx, subsystemNQN, uint32(nsid))
+		if err != nil {
+			log.ErrorLog(ctx, "failed to delete namespace %d from subsystem %s: %v", nsid, subsystemNQN, err)
+			return nil, status.Errorf(codes.Internal, "failed to delete namespace %d from subsystem %s: %v", nsid, subsystemNQN, err)
+		}
+		log.DebugLog(ctx, "Namespace %d deleted from subsystem %s successfully", nsid, subsystemNQN)
+	} else {
+		log.WarningLogMsg("No subsystem NQN or namespace ID found in volume context, skipping namespace deletion")
+	}
+
+	// step 2: Delete the subsystem if it is empty (for now is one-to-one mapping?)
+	err = cs.cleanupEmptySubsystem(ctx, subsystemNQN)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "failed to delete subsystem %s: %v", subsystemNQN, err)
+	}
+
+	// step 3: Delete the RBD volume through backend
+	res, err := cs.backendServer.DeleteVolume(ctx, req)
+	if err != nil {
+		log.ErrorLog(ctx, "failed to delete RBD volume %s: %v", volumeID, err)
+		return nil, status.Error(codes.NotFound, err.Error())
+	}
+	log.DebugLog(ctx, "RBD volume %s deleted successfully", volumeID)
+
+	return res, nil
+}
+
+func (cs *Server) ControllerPublishVolume(
+	ctx context.Context,
+	req *csi.ControllerPublishVolumeRequest,
+) (*csi.ControllerPublishVolumeResponse, error) {
+	log.DebugLog(ctx, "NVMe-oF ControllerPublishVolume: %s", req.GetVolumeId())
+
+	// step 1: Validate the request parameters
+
+	// step 2: add host to the NVMe-oF subsystem OR maybe add ns host
+	// (depends on the design how many subsystems we have one-to-one or one-to-many)
+
+	// step 3: add listener\s to the NVMe-oF subsystem
+
+	return &csi.ControllerPublishVolumeResponse{}, nil
+}
+
+func (cs *Server) ControllerUnpublishVolume(
+	ctx context.Context,
+	req *csi.ControllerUnpublishVolumeRequest,
+) (*csi.ControllerUnpublishVolumeResponse, error) {
+	log.DebugLog(ctx, "NVMe-oF ControllerUnpublishVolume: %s", req.GetVolumeId())
+
+	// step 1: remove host from the NVMe-oF subsystem\ns (depends on the design)
+
+	// step 2: remove listener\s from the NVMe-oF subsystem
+
+	return &csi.ControllerUnpublishVolumeResponse{}, nil
+}
+
+// ensureSubsystem checks if the subsystem exists, and creates it if not.
+func (cs *Server) ensureSubsystem(ctx context.Context, subsystemNQN string) error {
+	exists, err := cs.gateway.SubsystemExists(ctx, subsystemNQN)
+	if err != nil {
+		return err
+	}
+	if exists {
+		return nil
+	}
+	// Create if doesn't exist (controller decision)
+	return cs.gateway.CreateSubsystem(ctx, subsystemNQN)
+}
+
+// cleanupEmptySubsystem checks if the subsystem is empty (no namespaces), if so, deletes it.
+func (cs *Server) cleanupEmptySubsystem(ctx context.Context, subsystemNQN string) error {
+	if subsystemNQN == "" {
+		return nil
+	}
+
+	// Check if subsystem has any remaining namespaces
+	// (You'd need to implement ListNamespaces in your gateway client)
+	namespaces, err := cs.gateway.ListNamespaces(ctx, subsystemNQN)
+	if err != nil {
+		return fmt.Errorf("failed to list namespaces: %w", err)
+	}
+
+	if len(namespaces.Namespaces) == 0 {
+		log.DebugLog(ctx, "Subsystem %s is empty, deleting", subsystemNQN)
+		err = cs.gateway.DeleteSubsystem(ctx, subsystemNQN)
+		if err != nil {
+			return fmt.Errorf("failed to delete empty subsystem: %w", err)
+		}
+		log.DebugLog(ctx, "Empty subsystem %s deleted", subsystemNQN)
+	} else {
+		log.DebugLog(ctx, "Subsystem %s still has %d namespaces, keeping",
+			subsystemNQN, len(namespaces.Namespaces))
+	}
+
+	return nil
+}