feat: init

pkg/csi/api/client.go

@@ -0,0 +1,166 @@
+package api
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+	"time"
+
+	"github.com/KubelanCloud/kks-csi-plugin/pkg/csi/provisioner"
+)
+
+type ClientConfig struct {
+	BaseURL string
+	Token   string
+	Timeout time.Duration
+}
+
+type Client struct {
+	baseURL string
+	token   string
+	client  *http.Client
+}
+
+func NewClient(cfg ClientConfig) *Client {
+	timeout := cfg.Timeout
+	if timeout <= 0 {
+		timeout = 30 * time.Second
+	}
+	return &Client{
+		baseURL: strings.TrimRight(strings.TrimSpace(cfg.BaseURL), "/"),
+		token:   strings.TrimSpace(cfg.Token),
+		client:  &http.Client{Timeout: timeout},
+	}
+}
+
+func (c *Client) Close() error {
+	return nil
+}
+
+func (c *Client) ClusterInfo(ctx context.Context) (provisioner.ClusterInfo, error) {
+	var out provisioner.ClusterInfo
+	if err := c.doJSON(ctx, http.MethodGet, "/v1/cluster", nil, &out); err != nil {
+		return provisioner.ClusterInfo{}, err
+	}
+	return out, nil
+}
+
+func (c *Client) CreateVolume(ctx context.Context, req provisioner.CreateVolumeRequest) (provisioner.Volume, error) {
+	var out provisioner.Volume
+	if err := c.doJSON(ctx, http.MethodPost, "/v1/volumes", req, &out); err != nil {
+		return provisioner.Volume{}, err
+	}
+	return out, nil
+}
+
+func (c *Client) DeleteVolume(ctx context.Context, volumeID string) error {
+	return c.doJSON(ctx, http.MethodDelete, "/v1/volumes/"+escapePath(volumeID), nil, nil)
+}
+
+func (c *Client) VolumeExists(ctx context.Context, volumeID string) (bool, error) {
+	var out provisioner.VolumeExistsResponse
+	err := c.doJSON(ctx, http.MethodGet, "/v1/volumes/"+escapePath(volumeID), nil, &out)
+	if err != nil {
+		if isNotFound(err) {
+			return false, nil
+		}
+		return false, err
+	}
+	return out.Exists, nil
+}
+
+func (c *Client) PublishVolume(ctx context.Context, volumeID, nodeID string) (provisioner.PublishVolumeResponse, error) {
+	var out provisioner.PublishVolumeResponse
+	req := provisioner.PublishVolumeRequest{NodeID: nodeID}
+	path := "/v1/volumes/" + escapePath(volumeID) + "/publish"
+	if err := c.doJSON(ctx, http.MethodPost, path, req, &out); err != nil {
+		return provisioner.PublishVolumeResponse{}, err
+	}
+	return out, nil
+}
+
+func (c *Client) UnpublishVolume(ctx context.Context, volumeID, nodeID string) error {
+	req := provisioner.UnpublishVolumeRequest{
+		NodeID:   nodeID,
+		VolumeID: volumeID,
+	}
+	return c.doJSON(ctx, http.MethodPost, "/v1/volumes/unpublish", req, nil)
+}
+
+func (c *Client) doJSON(ctx context.Context, method, path string, reqBody any, respBody any) error {
+	var body io.Reader
+	if reqBody != nil {
+		raw, err := json.Marshal(reqBody)
+		if err != nil {
+			return fmt.Errorf("marshal request: %w", err)
+		}
+		body = bytes.NewReader(raw)
+	}
+
+	req, err := http.NewRequestWithContext(ctx, method, c.baseURL+path, body)
+	if err != nil {
+		return fmt.Errorf("create request: %w", err)
+	}
+	req.Header.Set("Accept", "application/json")
+	if reqBody != nil {
+		req.Header.Set("Content-Type", "application/json")
+	}
+	if c.token != "" {
+		req.Header.Set("Authorization", "Bearer "+c.token)
+	}
+
+	resp, err := c.client.Do(req)
+	if err != nil {
+		return fmt.Errorf("request %s %s: %w", method, path, err)
+	}
+	defer resp.Body.Close()
+
+	raw, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return fmt.Errorf("read response: %w", err)
+	}
+
+	if resp.StatusCode >= 400 {
+		return &HTTPError{
+			StatusCode: resp.StatusCode,
+			Method:     method,
+			Path:       path,
+			Body:       strings.TrimSpace(string(raw)),
+		}
+	}
+
+	if respBody == nil || len(raw) == 0 {
+		return nil
+	}
+	if err := json.Unmarshal(raw, respBody); err != nil {
+		return fmt.Errorf("decode response: %w", err)
+	}
+	return nil
+}
+
+type HTTPError struct {
+	StatusCode int
+	Method     string
+	Path       string
+	Body       string
+}
+
+func (e *HTTPError) Error() string {
+	if e.Body == "" {
+		return fmt.Sprintf("%s %s: status %d", e.Method, e.Path, e.StatusCode)
+	}
+	return fmt.Sprintf("%s %s: status %d: %s", e.Method, e.Path, e.StatusCode, e.Body)
+}
+
+func isNotFound(err error) bool {
+	httpErr, ok := err.(*HTTPError)
+	return ok && httpErr.StatusCode == http.StatusNotFound
+}
+
+func escapePath(value string) string {
+	return strings.ReplaceAll(value, "/", "%2F")
+}

pkg/csi/api/client_test.go

@@ -0,0 +1,62 @@
+package api
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/KubelanCloud/kks-csi-plugin/pkg/csi/provisioner"
+)
+
+func TestClientClusterInfo(t *testing.T) {
+	t.Parallel()
+
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/v1/cluster" {
+			t.Fatalf("unexpected path: %s", r.URL.Path)
+		}
+		_ = json.NewEncoder(w).Encode(provisioner.ClusterInfo{
+			StorageID: "abc123",
+		})
+	}))
+	t.Cleanup(srv.Close)
+
+	client := NewClient(ClientConfig{BaseURL: srv.URL})
+	info, err := client.ClusterInfo(context.Background())
+	if err != nil {
+		t.Fatalf("ClusterInfo failed: %v", err)
+	}
+	if info.StorageID != "abc123" {
+		t.Fatalf("unexpected cluster info: %#v", info)
+	}
+}
+
+func TestClientCreateVolume(t *testing.T) {
+	t.Parallel()
+
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != http.MethodPost || r.URL.Path != "/v1/volumes" {
+			t.Fatalf("unexpected request: %s %s", r.Method, r.URL.Path)
+		}
+		w.WriteHeader(http.StatusCreated)
+		_ = json.NewEncoder(w).Encode(provisioner.Volume{
+			VolumeID:  "abc123/k8s-volumes/pvc-1",
+			SizeBytes: 1024,
+		})
+	}))
+	t.Cleanup(srv.Close)
+
+	client := NewClient(ClientConfig{BaseURL: srv.URL})
+	vol, err := client.CreateVolume(context.Background(), provisioner.CreateVolumeRequest{
+		Name:      "pvc-1",
+		SizeBytes: 1024,
+	})
+	if err != nil {
+		t.Fatalf("CreateVolume failed: %v", err)
+	}
+	if vol.VolumeID != "abc123/k8s-volumes/pvc-1" {
+		t.Fatalf("unexpected volume: %#v", vol)
+	}
+}

pkg/csi/client/backend.go

@@ -0,0 +1,15 @@
+package client
+
+import (
+	"github.com/KubelanCloud/kks-csi-plugin/config"
+	"github.com/KubelanCloud/kks-csi-plugin/pkg/csi/api"
+	"github.com/KubelanCloud/kks-csi-plugin/pkg/csi/provisioner"
+)
+
+func NewBackend(cfg *config.ClientConf) provisioner.Backend {
+	return api.NewClient(api.ClientConfig{
+		BaseURL: cfg.ServerURL,
+		Token:   cfg.BearerToken(),
+		Timeout: cfg.Timeout(),
+	})
+}

pkg/csi/driver/controller.go

@@ -0,0 +1,168 @@
+package driver
+
+import (
+	"context"
+
+	"github.com/container-storage-interface/spec/lib/go/csi"
+	"github.com/KubelanCloud/kks-csi-plugin/pkg/csi/provisioner"
+)
+
+type ControllerServer struct {
+	d *Driver
+}
+
+func newControllerServer(d *Driver) *ControllerServer {
+	return &ControllerServer{d: d}
+}
+
+func (s *ControllerServer) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest) (*csi.CreateVolumeResponse, error) {
+	if req.GetName() == "" {
+		return nil, invalidArgument("volume name is required")
+	}
+	if req.GetVolumeCapabilities() == nil {
+		return nil, invalidArgument("volume capabilities are required")
+	}
+
+	capacity := int64(1 * 1024 * 1024 * 1024)
+	if req.GetCapacityRange() != nil {
+		if req.GetCapacityRange().GetRequiredBytes() > 0 {
+			capacity = req.GetCapacityRange().GetRequiredBytes()
+		} else if req.GetCapacityRange().GetLimitBytes() > 0 {
+			capacity = req.GetCapacityRange().GetLimitBytes()
+		}
+	}
+
+	vol, err := s.d.backend.CreateVolume(ctx, provisioner.CreateVolumeRequest{
+		Name:      sanitizeVolumeName(req.GetName()),
+		SizeBytes: capacity,
+	})
+	if err != nil {
+		return nil, internalError(err)
+	}
+
+	return &csi.CreateVolumeResponse{
+		Volume: &csi.Volume{
+			VolumeId:      vol.VolumeID,
+			CapacityBytes: vol.SizeBytes,
+			VolumeContext: vol.VolumeContext,
+		},
+	}, nil
+}
+
+func (s *ControllerServer) DeleteVolume(ctx context.Context, req *csi.DeleteVolumeRequest) (*csi.DeleteVolumeResponse, error) {
+	if req.GetVolumeId() == "" {
+		return nil, invalidArgument("volume id is required")
+	}
+
+	if err := s.d.backend.DeleteVolume(ctx, req.GetVolumeId()); err != nil {
+		return nil, internalError(err)
+	}
+	return &csi.DeleteVolumeResponse{}, nil
+}
+
+func (s *ControllerServer) ControllerPublishVolume(ctx context.Context, req *csi.ControllerPublishVolumeRequest) (*csi.ControllerPublishVolumeResponse, error) {
+	if req.GetVolumeId() == "" {
+		return nil, invalidArgument("volume id is required")
+	}
+	if req.GetNodeId() == "" {
+		return nil, invalidArgument("node id is required")
+	}
+
+	pub, err := s.d.backend.PublishVolume(ctx, req.GetVolumeId(), req.GetNodeId())
+	if err != nil {
+		return nil, internalError(err)
+	}
+
+	return &csi.ControllerPublishVolumeResponse{
+		PublishContext: pub.PublishContext,
+	}, nil
+}
+
+func (s *ControllerServer) ControllerUnpublishVolume(ctx context.Context, req *csi.ControllerUnpublishVolumeRequest) (*csi.ControllerUnpublishVolumeResponse, error) {
+	if req.GetVolumeId() == "" {
+		return nil, invalidArgument("volume id is required")
+	}
+	nodeID := req.GetNodeId()
+	if nodeID == "" {
+		return &csi.ControllerUnpublishVolumeResponse{}, nil
+	}
+
+	if err := s.d.backend.UnpublishVolume(ctx, req.GetVolumeId(), nodeID); err != nil {
+		return nil, internalError(err)
+	}
+	return &csi.ControllerUnpublishVolumeResponse{}, nil
+}
+
+func (s *ControllerServer) ValidateVolumeCapabilities(ctx context.Context, req *csi.ValidateVolumeCapabilitiesRequest) (*csi.ValidateVolumeCapabilitiesResponse, error) {
+	_ = ctx
+	if req.GetVolumeId() == "" {
+		return nil, invalidArgument("volume id is required")
+	}
+	if req.GetVolumeCapabilities() == nil {
+		return nil, invalidArgument("volume capabilities are required")
+	}
+	return &csi.ValidateVolumeCapabilitiesResponse{
+		Confirmed: &csi.ValidateVolumeCapabilitiesResponse_Confirmed{
+			VolumeCapabilities: req.GetVolumeCapabilities(),
+		},
+	}, nil
+}
+
+func (s *ControllerServer) ListVolumes(ctx context.Context, req *csi.ListVolumesRequest) (*csi.ListVolumesResponse, error) {
+	_ = ctx
+	_ = req
+	return nil, unimplemented("ListVolumes is not supported")
+}
+
+func (s *ControllerServer) GetCapacity(ctx context.Context, req *csi.GetCapacityRequest) (*csi.GetCapacityResponse, error) {
+	_ = ctx
+	_ = req
+	return nil, unimplemented("GetCapacity is not supported")
+}
+
+func (s *ControllerServer) CreateSnapshot(ctx context.Context, req *csi.CreateSnapshotRequest) (*csi.CreateSnapshotResponse, error) {
+	_ = ctx
+	_ = req
+	return nil, unimplemented("CreateSnapshot is not supported")
+}
+
+func (s *ControllerServer) DeleteSnapshot(ctx context.Context, req *csi.DeleteSnapshotRequest) (*csi.DeleteSnapshotResponse, error) {
+	_ = ctx
+	_ = req
+	return nil, unimplemented("DeleteSnapshot is not supported")
+}
+
+func (s *ControllerServer) ListSnapshots(ctx context.Context, req *csi.ListSnapshotsRequest) (*csi.ListSnapshotsResponse, error) {
+	_ = ctx
+	_ = req
+	return nil, unimplemented("ListSnapshots is not supported")
+}
+
+func (s *ControllerServer) ControllerExpandVolume(ctx context.Context, req *csi.ControllerExpandVolumeRequest) (*csi.ControllerExpandVolumeResponse, error) {
+	_ = ctx
+	_ = req
+	return nil, unimplemented("ControllerExpandVolume is not supported")
+}
+
+func (s *ControllerServer) ControllerGetVolume(ctx context.Context, req *csi.ControllerGetVolumeRequest) (*csi.ControllerGetVolumeResponse, error) {
+	_ = ctx
+	_ = req
+	return nil, unimplemented("ControllerGetVolume is not supported")
+}
+
+func (s *ControllerServer) ControllerGetCapabilities(ctx context.Context, req *csi.ControllerGetCapabilitiesRequest) (*csi.ControllerGetCapabilitiesResponse, error) {
+	_ = ctx
+	_ = req
+	return &csi.ControllerGetCapabilitiesResponse{
+		Capabilities: []*csi.ControllerServiceCapability{
+			controllerCapability(csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME),
+			controllerCapability(csi.ControllerServiceCapability_RPC_PUBLISH_UNPUBLISH_VOLUME),
+		},
+	}, nil
+}
+
+func (s *ControllerServer) ControllerModifyVolume(ctx context.Context, req *csi.ControllerModifyVolumeRequest) (*csi.ControllerModifyVolumeResponse, error) {
+	_ = ctx
+	_ = req
+	return nil, unimplemented("ControllerModifyVolume is not supported")
+}

pkg/csi/driver/driver.go

@@ -0,0 +1,149 @@
+package driver
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"os"
+	"path/filepath"
+	"strings"
+
+	"github.com/container-storage-interface/spec/lib/go/csi"
+	"github.com/KubelanCloud/kks-csi-plugin/config"
+	"github.com/KubelanCloud/kks-csi-plugin/pkg/csi/client"
+	"github.com/KubelanCloud/kks-csi-plugin/pkg/csi/provisioner"
+	"go.uber.org/zap"
+	"google.golang.org/grpc"
+)
+
+type Driver struct {
+	cfg        *config.Config
+	log        *zap.SugaredLogger
+	backend    provisioner.Backend
+	storageID  string
+	identity   *IdentityServer
+	controller *ControllerServer
+	node       *NodeServer
+}
+
+func Run(ctx context.Context, cfg *config.Config, logger *zap.Logger) error {
+	if err := cfg.Validate(); err != nil {
+		return err
+	}
+	if logger == nil {
+		logger = zap.NewNop()
+	}
+	log := logger.Sugar()
+
+	backend := client.NewBackend(cfg.Client)
+	defer backend.Close()
+
+	cluster, err := backend.ClusterInfo(ctx)
+	if err != nil {
+		return fmt.Errorf("resolve storage info from csi server: %w", err)
+	}
+
+	d := &Driver{
+		cfg:       cfg,
+		log:       log,
+		backend:   backend,
+		storageID: cluster.StorageID,
+		identity:  newIdentityServer(cfg),
+	}
+
+	switch cfg.Driver.Mode {
+	case config.CSIModeController, config.CSIModeAll:
+		d.controller = newControllerServer(d)
+	case config.CSIModeNode:
+	}
+
+	switch cfg.Driver.Mode {
+	case config.CSIModeNode, config.CSIModeAll:
+		d.node = newNodeServer(d)
+	case config.CSIModeController:
+	}
+
+	endpoint, err := parseEndpoint(cfg.Driver.Endpoint)
+	if err != nil {
+		return err
+	}
+	if err := ensureSocketDir(endpoint); err != nil {
+		return err
+	}
+
+	listener, err := net.Listen(endpoint.network, endpoint.address)
+	if err != nil {
+		return fmt.Errorf("listen on %s: %w", cfg.Driver.Endpoint, err)
+	}
+	defer listener.Close()
+
+	server := grpc.NewServer()
+	csi.RegisterIdentityServer(server, d.identity)
+	if d.controller != nil {
+		csi.RegisterControllerServer(server, d.controller)
+	}
+	if d.node != nil {
+		csi.RegisterNodeServer(server, d.node)
+	}
+
+	log.Infof(
+		"starting csi client name=%s endpoint=%s mode=%s node_id=%s server=%s storage_id=%s",
+		cfg.Driver.Name,
+		cfg.Driver.Endpoint,
+		cfg.Driver.Mode,
+		cfg.Driver.NodeID,
+		cfg.Client.ServerURL,
+		cluster.StorageID,
+	)
+
+	errCh := make(chan error, 1)
+	go func() {
+		errCh <- server.Serve(listener)
+	}()
+
+	select {
+	case <-ctx.Done():
+		log.Info("shutting down csi client")
+		server.GracefulStop()
+		return nil
+	case err := <-errCh:
+		if err != nil {
+			return fmt.Errorf("csi grpc server stopped: %w", err)
+		}
+		return nil
+	}
+}
+
+type endpoint struct {
+	network string
+	address string
+}
+
+func parseEndpoint(raw string) (endpoint, error) {
+	if strings.HasPrefix(raw, "unix://") {
+		return endpoint{
+			network: "unix",
+			address: strings.TrimPrefix(raw, "unix://"),
+		}, nil
+	}
+	if strings.HasPrefix(raw, "tcp://") {
+		return endpoint{
+			network: "tcp",
+			address: strings.TrimPrefix(raw, "tcp://"),
+		}, nil
+	}
+	return endpoint{}, fmt.Errorf("unsupported endpoint %q (use unix:// or tcp://)", raw)
+}
+
+func ensureSocketDir(ep endpoint) error {
+	if ep.network != "unix" {
+		return nil
+	}
+	if err := os.MkdirAll(filepath.Dir(ep.address), 0o755); err != nil {
+		return fmt.Errorf("create socket directory: %w", err)
+	}
+	if err := os.Remove(ep.address); err != nil && !os.IsNotExist(err) {
+		return fmt.Errorf("remove stale socket: %w", err)
+	}
+	return nil
+}

pkg/csi/driver/identity.go

@@ -0,0 +1,82 @@
+package driver
+
+import (
+	"context"
+
+	"github.com/container-storage-interface/spec/lib/go/csi"
+	"github.com/KubelanCloud/kks-csi-plugin/config"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/types/known/wrapperspb"
+)
+
+type IdentityServer struct {
+	cfg *config.Config
+}
+
+func newIdentityServer(cfg *config.Config) *IdentityServer {
+	return &IdentityServer{cfg: cfg}
+}
+
+func (s *IdentityServer) GetPluginInfo(ctx context.Context, req *csi.GetPluginInfoRequest) (*csi.GetPluginInfoResponse, error) {
+	_ = ctx
+	_ = req
+	return &csi.GetPluginInfoResponse{
+		Name:          s.cfg.Driver.Name,
+		VendorVersion: "v0.1.0",
+	}, nil
+}
+
+func (s *IdentityServer) GetPluginCapabilities(ctx context.Context, req *csi.GetPluginCapabilitiesRequest) (*csi.GetPluginCapabilitiesResponse, error) {
+	_ = ctx
+	_ = req
+	return &csi.GetPluginCapabilitiesResponse{
+		Capabilities: []*csi.PluginCapability{
+			{
+				Type: &csi.PluginCapability_Service_{
+					Service: &csi.PluginCapability_Service{
+						Type: csi.PluginCapability_Service_CONTROLLER_SERVICE,
+					},
+				},
+			},
+		},
+	}, nil
+}
+
+func (s *IdentityServer) Probe(ctx context.Context, req *csi.ProbeRequest) (*csi.ProbeResponse, error) {
+	_ = ctx
+	_ = req
+	return &csi.ProbeResponse{Ready: wrapperspb.Bool(true)}, nil
+}
+
+func controllerCapability(t csi.ControllerServiceCapability_RPC_Type) *csi.ControllerServiceCapability {
+	return &csi.ControllerServiceCapability{
+		Type: &csi.ControllerServiceCapability_Rpc{
+			Rpc: &csi.ControllerServiceCapability_RPC{Type: t},
+		},
+	}
+}
+
+func nodeCapability(t csi.NodeServiceCapability_RPC_Type) *csi.NodeServiceCapability {
+	return &csi.NodeServiceCapability{
+		Type: &csi.NodeServiceCapability_Rpc{
+			Rpc: &csi.NodeServiceCapability_RPC{Type: t},
+		},
+	}
+}
+
+func invalidArgument(msg string) error {
+	return status.Error(codes.InvalidArgument, msg)
+}
+
+func notFound(msg string) error {
+	return status.Error(codes.NotFound, msg)
+}
+
+func internalError(err error) error {
+	return status.Errorf(codes.Internal, "%v", err)
+}
+
+func unimplemented(msg string) error {
+	return status.Error(codes.Unimplemented, msg)
+}

pkg/csi/driver/mount_linux.go

@@ -0,0 +1,120 @@
+//go:build linux
+
+package driver
+
+import (
+	"fmt"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strconv"
+	"strings"
+
+	mount "k8s.io/mount-utils"
+	utilexec "k8s.io/utils/exec"
+)
+
+const defaultLinuxFsType = "ext4"
+
+func newMounter() *mount.SafeFormatAndMount {
+	return mount.NewSafeFormatAndMount(mount.New(""), utilexec.New())
+}
+
+func scsiHostRescan() {
+	scsiPath := "/sys/class/scsi_host/"
+	entries, err := os.ReadDir(scsiPath)
+	if err != nil {
+		return
+	}
+	for _, entry := range entries {
+		name := filepath.Join(scsiPath, entry.Name(), "scan")
+		_ = os.WriteFile(name, []byte("- - -"), 0o666)
+	}
+}
+
+func findDiskByLUN(lun int) (string, error) {
+	scsiHostRescan()
+
+	sysPath := "/sys/bus/scsi/devices"
+	entries, err := os.ReadDir(sysPath)
+	if err != nil {
+		return "", fmt.Errorf("read %s: %w", sysPath, err)
+	}
+
+	for _, entry := range entries {
+		name := entry.Name()
+		parts := strings.Split(name, ":")
+		if len(parts) < 4 {
+			continue
+		}
+		deviceLUN, err := strconv.Atoi(parts[3])
+		if err != nil || deviceLUN != lun {
+			continue
+		}
+
+		vendorPath := filepath.Join(sysPath, name, "vendor")
+		vendorBytes, err := os.ReadFile(vendorPath)
+		if err != nil {
+			continue
+		}
+		vendor := strings.TrimSpace(string(vendorBytes))
+		if vendor != "QEMU" && strings.ToUpper(vendor) != "MSFT" {
+			continue
+		}
+
+		blockDir := filepath.Join(sysPath, name, "block")
+		devices, err := os.ReadDir(blockDir)
+		if err != nil || len(devices) == 0 {
+			continue
+		}
+		devName := devices[0].Name()
+
+		for _, devLinkPath := range []string{"/dev/disk/by-id/", "/dev/disk/by-path/"} {
+			if link, err := findDiskLink(devLinkPath, devName); err == nil {
+				return link, nil
+			}
+		}
+		return "/dev/" + devName, nil
+	}
+
+	return "", fmt.Errorf("failed to find disk by lun %d", lun)
+}
+
+func findDiskLink(devLinkPath, devName string) (string, error) {
+	entries, err := os.ReadDir(devLinkPath)
+	if err != nil {
+		return "", err
+	}
+	for _, entry := range entries {
+		linkPath := filepath.Join(devLinkPath, entry.Name())
+		target, err := os.Readlink(linkPath)
+		if err != nil {
+			continue
+		}
+		if strings.HasSuffix(target, devName) {
+			return linkPath, nil
+		}
+	}
+	return "", fmt.Errorf("device %s not found under %s", devName, devLinkPath)
+}
+
+func formatAndMount(source, target, fsType string, options []string, mounter *mount.SafeFormatAndMount) error {
+	return mounter.FormatAndMount(source, target, fsType, options)
+}
+
+func cleanupMountPoint(target string, mounter *mount.SafeFormatAndMount) error {
+	return mount.CleanupMountPoint(target, mounter, true)
+}
+
+func bindMount(source, target string) error {
+	out, err := exec.Command("mount", "--bind", source, target).CombinedOutput()
+	if err != nil {
+		return fmt.Errorf("bind mount %s -> %s: %w: %s", source, target, err, strings.TrimSpace(string(out)))
+	}
+	return nil
+}
+
+func isMounted(path string) bool {
+	out, err := exec.Command("findmnt", "-n", path).CombinedOutput()
+	return err == nil && strings.TrimSpace(string(out)) != ""
+}

pkg/csi/driver/mount_other.go

@@ -0,0 +1,39 @@
+//go:build !linux
+
+package driver
+
+import (
+	"fmt"
+
+	mount "k8s.io/mount-utils"
+	utilexec "k8s.io/utils/exec"
+)
+
+const defaultLinuxFsType = "ext4"
+
+func newMounter() *mount.SafeFormatAndMount {
+	return mount.NewSafeFormatAndMount(mount.New(""), utilexec.New())
+}
+
+func findDiskByLUN(lun int) (string, error) {
+	return "", fmt.Errorf("scsi volume mount is only supported on linux (lun %d)", lun)
+}
+
+func formatAndMount(source, target, fsType string, options []string, mounter *mount.SafeFormatAndMount) error {
+	return mounter.FormatAndMount(source, target, fsType, options)
+}
+
+func cleanupMountPoint(target string, mounter *mount.SafeFormatAndMount) error {
+	return mount.CleanupMountPoint(target, mounter, true)
+}
+
+func bindMount(source, target string) error {
+	m := newMounter()
+	return m.Mount(source, target, "", []string{"bind"})
+}
+
+func isMounted(path string) bool {
+	m := newMounter()
+	notMnt, err := m.IsLikelyNotMountPoint(path)
+	return err == nil && !notMnt
+}

pkg/csi/driver/node.go

@@ -0,0 +1,165 @@
+package driver
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+
+	"github.com/container-storage-interface/spec/lib/go/csi"
+	"github.com/KubelanCloud/kks-csi-plugin/pkg/csi/provisioner"
+)
+
+type NodeServer struct {
+	d *Driver
+}
+
+func newNodeServer(d *Driver) *NodeServer {
+	return &NodeServer{d: d}
+}
+
+func (s *NodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStageVolumeRequest) (*csi.NodeStageVolumeResponse, error) {
+	_ = ctx
+	if req.GetVolumeId() == "" {
+		return nil, invalidArgument("volume id is required")
+	}
+	if req.GetStagingTargetPath() == "" {
+		return nil, invalidArgument("staging target path is required")
+	}
+	if isBlockVolume(req.GetVolumeCapability()) {
+		return nil, unimplemented("block volumes are not supported")
+	}
+
+	lun, err := lunFromPublishContext(req.GetPublishContext())
+	if err != nil {
+		return nil, invalidArgument(err.Error())
+	}
+
+	device, err := findDiskByLUN(lun)
+	if err != nil {
+		return nil, internalError(err)
+	}
+
+	if err := os.MkdirAll(req.GetStagingTargetPath(), 0o755); err != nil {
+		return nil, internalError(fmt.Errorf("create staging path: %w", err))
+	}
+
+	fsType := defaultLinuxFsType
+	options := []string{}
+	if mnt := req.GetVolumeCapability().GetMount(); mnt != nil {
+		if mnt.FsType != "" {
+			fsType = mnt.FsType
+		}
+		options = append(options, mnt.MountFlags...)
+	}
+
+	mounter := newMounter()
+	if err := formatAndMount(device, req.GetStagingTargetPath(), fsType, options, mounter); err != nil {
+		return nil, internalError(fmt.Errorf("format and mount %s: %w", device, err))
+	}
+
+	return &csi.NodeStageVolumeResponse{}, nil
+}
+
+func (s *NodeServer) NodeUnstageVolume(ctx context.Context, req *csi.NodeUnstageVolumeRequest) (*csi.NodeUnstageVolumeResponse, error) {
+	_ = ctx
+	if req.GetVolumeId() == "" {
+		return nil, invalidArgument("volume id is required")
+	}
+	if req.GetStagingTargetPath() == "" {
+		return nil, invalidArgument("staging target path is required")
+	}
+
+	mounter := newMounter()
+	if err := cleanupMountPoint(req.GetStagingTargetPath(), mounter); err != nil {
+		return nil, internalError(err)
+	}
+	return &csi.NodeUnstageVolumeResponse{}, nil
+}
+
+func (s *NodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublishVolumeRequest) (*csi.NodePublishVolumeResponse, error) {
+	_ = ctx
+	if req.GetVolumeId() == "" {
+		return nil, invalidArgument("volume id is required")
+	}
+	if req.GetTargetPath() == "" {
+		return nil, invalidArgument("target path is required")
+	}
+	if req.GetStagingTargetPath() == "" {
+		return nil, invalidArgument("staging target path is required")
+	}
+
+	if err := os.MkdirAll(filepath.Dir(req.GetTargetPath()), 0o755); err != nil {
+		return nil, internalError(err)
+	}
+
+	if err := bindMount(req.GetStagingTargetPath(), req.GetTargetPath()); err != nil {
+		return nil, internalError(err)
+	}
+
+	return &csi.NodePublishVolumeResponse{}, nil
+}
+
+func (s *NodeServer) NodeUnpublishVolume(ctx context.Context, req *csi.NodeUnpublishVolumeRequest) (*csi.NodeUnpublishVolumeResponse, error) {
+	_ = ctx
+	if req.GetVolumeId() == "" {
+		return nil, invalidArgument("volume id is required")
+	}
+	if req.GetTargetPath() == "" {
+		return nil, invalidArgument("target path is required")
+	}
+
+	mounter := newMounter()
+	if err := cleanupMountPoint(req.GetTargetPath(), mounter); err != nil {
+		return nil, internalError(err)
+	}
+	return &csi.NodeUnpublishVolumeResponse{}, nil
+}
+
+func (s *NodeServer) NodeGetVolumeStats(ctx context.Context, req *csi.NodeGetVolumeStatsRequest) (*csi.NodeGetVolumeStatsResponse, error) {
+	_ = ctx
+	_ = req
+	return nil, unimplemented("NodeGetVolumeStats is not supported")
+}
+
+func (s *NodeServer) NodeExpandVolume(ctx context.Context, req *csi.NodeExpandVolumeRequest) (*csi.NodeExpandVolumeResponse, error) {
+	_ = ctx
+	_ = req
+	return nil, unimplemented("NodeExpandVolume is not supported")
+}
+
+func (s *NodeServer) NodeGetCapabilities(ctx context.Context, req *csi.NodeGetCapabilitiesRequest) (*csi.NodeGetCapabilitiesResponse, error) {
+	_ = ctx
+	_ = req
+	return &csi.NodeGetCapabilitiesResponse{
+		Capabilities: []*csi.NodeServiceCapability{
+			nodeCapability(csi.NodeServiceCapability_RPC_STAGE_UNSTAGE_VOLUME),
+		},
+	}, nil
+}
+
+func (s *NodeServer) NodeGetInfo(ctx context.Context, req *csi.NodeGetInfoRequest) (*csi.NodeGetInfoResponse, error) {
+	_ = ctx
+	_ = req
+	return &csi.NodeGetInfoResponse{
+		NodeId: s.d.cfg.Driver.NodeID,
+	}, nil
+}
+
+func lunFromPublishContext(publishContext map[string]string) (int, error) {
+	lunStr := strings.TrimSpace(publishContext[provisioner.PublishContextLUN])
+	if lunStr == "" {
+		return 0, fmt.Errorf("publish context %q is required", provisioner.PublishContextLUN)
+	}
+	lun, err := strconv.Atoi(lunStr)
+	if err != nil {
+		return 0, fmt.Errorf("invalid publish context %q: %w", provisioner.PublishContextLUN, err)
+	}
+	return lun, nil
+}
+
+func isBlockVolume(capability *csi.VolumeCapability) bool {
+	return capability != nil && capability.GetBlock() != nil
+}

pkg/csi/driver/volume.go

@@ -0,0 +1,7 @@
+package driver
+
+import "github.com/KubelanCloud/kks-csi-plugin/pkg/csi/volume"
+
+func sanitizeVolumeName(name string) string {
+	return volume.SanitizeName(name)
+}

pkg/csi/provisioner/types.go

@@ -0,0 +1,51 @@
+package provisioner
+
+import "context"
+
+const (
+	PublishContextLUN   = "scsi.lun"
+	PublishContextSlot  = "proxmox.scsi_slot"
+	PublishContextVolid = "proxmox.volid"
+)
+
+type ClusterInfo struct {
+	StorageID string `json:"storage_id"`
+}
+
+type Volume struct {
+	VolumeID      string            `json:"volume_id"`
+	SizeBytes     int64             `json:"size_bytes"`
+	VolumeContext map[string]string `json:"volume_context,omitempty"`
+}
+
+type CreateVolumeRequest struct {
+	Name      string `json:"name"`
+	SizeBytes int64  `json:"size_bytes"`
+}
+
+type PublishVolumeRequest struct {
+	NodeID string `json:"node_id"`
+}
+
+type PublishVolumeResponse struct {
+	PublishContext map[string]string `json:"publish_context"`
+}
+
+type UnpublishVolumeRequest struct {
+	NodeID   string `json:"node_id"`
+	VolumeID string `json:"volume_id"`
+}
+
+type VolumeExistsResponse struct {
+	Exists bool `json:"exists"`
+}
+
+type Backend interface {
+	ClusterInfo(ctx context.Context) (ClusterInfo, error)
+	CreateVolume(ctx context.Context, req CreateVolumeRequest) (Volume, error)
+	DeleteVolume(ctx context.Context, volumeID string) error
+	VolumeExists(ctx context.Context, volumeID string) (bool, error)
+	PublishVolume(ctx context.Context, volumeID, nodeID string) (PublishVolumeResponse, error)
+	UnpublishVolume(ctx context.Context, volumeID, nodeID string) error
+	Close() error
+}

pkg/csi/volume/id.go

@@ -0,0 +1,29 @@
+package volume
+
+import (
+	"fmt"
+	"path"
+	"strings"
+)
+
+func ID(storageID, name string) string {
+	return path.Join(storageID, name)
+}
+
+func Parse(volumeID string) (storageID, name string, err error) {
+	parts := strings.Split(volumeID, "/")
+	if len(parts) != 2 {
+		return "", "", fmt.Errorf("invalid volume id %q", volumeID)
+	}
+	return parts[0], parts[1], nil
+}
+
+func SanitizeName(name string) string {
+	name = strings.TrimSpace(name)
+	name = strings.ReplaceAll(name, " ", "-")
+	return name
+}
+
+func ExportKey(volumeID string) string {
+	return strings.NewReplacer("/", "_", " ", "_").Replace(volumeID)
+}

pkg/csi/volume/id_test.go

@@ -0,0 +1,25 @@
+package volume
+
+import "testing"
+
+func TestParseVolumeID(t *testing.T) {
+	storageID, name, err := Parse("nfs-prod/pvc-1")
+	if err != nil {
+		t.Fatalf("Parse failed: %v", err)
+	}
+	if storageID != "nfs-prod" || name != "pvc-1" {
+		t.Fatalf("unexpected parse result: %q %q", storageID, name)
+	}
+}
+
+func TestParseVolumeIDRejectsInvalid(t *testing.T) {
+	if _, _, err := Parse("only-one-part"); err == nil {
+		t.Fatal("expected error for invalid volume id")
+	}
+}
+
+func TestID(t *testing.T) {
+	if got := ID("nfs-prod", "pvc-1"); got != "nfs-prod/pvc-1" {
+		t.Fatalf("unexpected id: %q", got)
+	}
+}