package driver
import (
"context"
"fmt"
csi "github.com/container-storage-interface/spec/lib/go/csi"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
"k8s.io/klog/v2"
"gitcode.com/openFuyao/ub-ssu-csi/pkg/backend"
"gitcode.com/openFuyao/ub-ssu-csi/pkg/kube"
)
const ubseNameMaxLen = 48
type ControllerServer struct {
csi.UnimplementedControllerServer
storage backend.StorageManager
kube.PVCReader
}
func NewControllerServer(storage backend.StorageManager, pvcReader kube.PVCReader) *ControllerServer {
return &ControllerServer{
storage: storage,
PVCReader: pvcReader,
}
}
func (cs *ControllerServer) CreateVolume(
ctx context.Context,
req *csi.CreateVolumeRequest,
) (*csi.CreateVolumeResponse, error) {
klog.InfoS("CreateVolume called", "name", req.GetName())
size, err := validateCreateVolumeRequest(req)
if err != nil {
return nil, err
}
pvcAnnotations, err := cs.readPVCAnnotations(ctx, req.GetParameters())
if err != nil {
stErr := status.Errorf(codes.Aborted, "read pvc annotations: %v", err)
klog.ErrorS(stErr, "CreateVolume failed", "name", req.GetName())
return nil, stErr
}
params, err := parseVolumeParams(req.GetParameters(), pvcAnnotations)
if err != nil {
stErr := status.Error(codes.InvalidArgument, err.Error())
klog.ErrorS(stErr, "CreateVolume invalid params", "name", req.GetName())
return nil, stErr
}
klog.InfoS("AllocVolume",
"name", req.GetName(),
"sizeBytes", size,
"nsnum", params.nsnum,
"strategy", params.allocStrategy,
"lbaFormat", params.lbaFormat)
vol, err := cs.storage.AllocVolume(req.GetName(), size, params.toLogicalConfig())
if err != nil {
stErr := status.Errorf(codes.Internal, "AllocVolume failed: %v", err)
klog.ErrorS(stErr, "CreateVolume failed", "name", req.GetName())
return nil, stErr
}
if vol == nil {
stErr := status.Error(codes.Internal, "AllocVolume returned nil volume")
klog.ErrorS(stErr, "CreateVolume failed", "name", req.GetName())
return nil, stErr
}
resp, err := buildCreateVolumeResponse(vol, params)
if err != nil {
stErr := status.Error(codes.Internal, err.Error())
klog.ErrorS(stErr, "CreateVolume failed", "name", req.GetName())
return nil, stErr
}
klog.InfoS("CreateVolume succeeded",
"name", req.GetName(),
"volumeID", vol.VolumeID,
"nsnum", len(vol.Namespaces),
"sizeBytes", vol.SizeBytes)
return resp, nil
}
func validateCreateVolumeRequest(req *csi.CreateVolumeRequest) (uint64, error) {
if len(req.GetName()) == 0 {
err := status.Error(codes.InvalidArgument, "volume name missing")
klog.ErrorS(err, "CreateVolume invalid request", "reason", "name missing")
return 0, err
}
if len(req.GetName()) > ubseNameMaxLen {
err := status.Errorf(codes.InvalidArgument,
"volume name length %d exceeds UBSE limit %d",
len(req.GetName()), ubseNameMaxLen)
klog.ErrorS(err, "CreateVolume invalid request", "name", req.GetName(), "len", len(req.GetName()))
return 0, err
}
if len(req.GetVolumeCapabilities()) == 0 {
err := status.Error(codes.InvalidArgument, "volume capabilities missing")
klog.ErrorS(err, "CreateVolume invalid request", "name", req.GetName(), "reason", "capabilities missing")
return 0, err
}
if err := validateVolumeCapabilities(req.GetVolumeCapabilities()); err != nil {
stErr := status.Error(codes.InvalidArgument, err.Error())
klog.ErrorS(stErr, "CreateVolume invalid request", "name", req.GetName())
return 0, stErr
}
size, err := parseRequestedSize(req.GetCapacityRange())
if err != nil {
stErr := status.Error(codes.InvalidArgument, err.Error())
klog.ErrorS(stErr, "CreateVolume invalid request", "name", req.GetName())
return 0, stErr
}
return size, nil
}
func buildCreateVolumeResponse(vol *backend.VolumeInfo, params volumeParams) (*csi.CreateVolumeResponse, error) {
vol.RaidLevel = params.raidLevel
vol.ChunkSize = params.chunksize
volCtx, err := buildVolumeContext(vol)
if err != nil {
return nil, fmt.Errorf("build volume context: %w", err)
}
return &csi.CreateVolumeResponse{
Volume: &csi.Volume{
VolumeId: vol.VolumeID,
CapacityBytes: int64(vol.SizeBytes),
VolumeContext: volCtx,
},
}, nil
}
func (cs *ControllerServer) readPVCAnnotations(
ctx context.Context,
parameters map[string]string,
) (map[string]string, error) {
pvcName := parameters[pvcNameKey]
pvcNamespace := parameters[pvcNamespaceKey]
if pvcName == "" || pvcNamespace == "" {
klog.InfoS("extra-create-metadata PVC keys absent, skipping PVC annotations")
return nil, nil
}
return cs.GetPVCAnnotations(ctx, pvcNamespace, pvcName)
}
func validateVolumeCapabilities(caps []*csi.VolumeCapability) error {
for _, c := range caps {
if c == nil {
return fmt.Errorf("volume capability is nil")
}
if c.GetAccessMode().GetMode() != csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER {
return fmt.Errorf("only ReadWriteOnce access mode is supported, got %v", c.GetAccessMode().GetMode())
}
switch c.GetAccessType().(type) {
case *csi.VolumeCapability_Block:
case *csi.VolumeCapability_Mount:
default:
return fmt.Errorf("unsupported access type: must be Block or Mount")
}
}
return nil
}
func (cs *ControllerServer) DeleteVolume(
ctx context.Context,
req *csi.DeleteVolumeRequest,
) (*csi.DeleteVolumeResponse, error) {
klog.InfoS("DeleteVolume called", "volumeID", req.GetVolumeId())
if len(req.GetVolumeId()) == 0 {
err := status.Error(codes.InvalidArgument, "volume ID missing")
klog.ErrorS(err, "DeleteVolume invalid request", "reason", "volume ID missing")
return nil, err
}
if err := cs.storage.DeleteVolume(req.GetVolumeId()); err != nil {
stErr := status.Errorf(codes.Internal, "DeleteVolume failed: %v", err)
klog.ErrorS(stErr, "DeleteVolume failed", "volumeID", req.GetVolumeId())
return nil, stErr
}
klog.InfoS("DeleteVolume succeeded", "volumeID", req.GetVolumeId())
return &csi.DeleteVolumeResponse{}, nil
}
func (cs *ControllerServer) ValidateVolumeCapabilities(
ctx context.Context,
req *csi.ValidateVolumeCapabilitiesRequest,
) (*csi.ValidateVolumeCapabilitiesResponse, error) {
klog.InfoS("ValidateVolumeCapabilities called", "volumeID", req.GetVolumeId())
if len(req.GetVolumeId()) == 0 {
return nil, status.Error(codes.InvalidArgument, "volume ID missing")
}
if req.GetVolumeCapabilities() == nil {
return nil, status.Error(codes.InvalidArgument, "volume capabilities missing")
}
return &csi.ValidateVolumeCapabilitiesResponse{
Confirmed: &csi.ValidateVolumeCapabilitiesResponse_Confirmed{
VolumeCapabilities: req.GetVolumeCapabilities(),
},
}, nil
}
func (cs *ControllerServer) ControllerGetCapabilities(
ctx context.Context,
req *csi.ControllerGetCapabilitiesRequest,
) (*csi.ControllerGetCapabilitiesResponse, error) {
klog.InfoS("ControllerGetCapabilities called")
return &csi.ControllerGetCapabilitiesResponse{
Capabilities: []*csi.ControllerServiceCapability{
{
Type: &csi.ControllerServiceCapability_Rpc{
Rpc: &csi.ControllerServiceCapability_RPC{
Type: csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME,
},
},
},
},
}, nil
}