diff --git a/osac-csi-driver/AGENTS.md b/osac-csi-driver/AGENTS.md index 4a3592ab22..0165e07a4a 100644 --- a/osac-csi-driver/AGENTS.md +++ b/osac-csi-driver/AGENTS.md @@ -46,7 +46,6 @@ osac-csi-driver/ │ │ └── node.go # NodeServer (Stage/Unstage, Publish/Unpublish, vendor proxy) │ ├── fulfillment/ │ │ ├── volume.go # VolumeClient interface, VolumeInfo, CreateVolumeParams -│ │ ├── controlplane.go # ControlPlaneClient interface (Publish/Unpublish) │ │ └── stubs.go # In-memory stubs for development (no fulfillment-service needed) │ └── proxy/ │ └── proxy.go # gRPC connection manager for vendor CSI sockets (lazy, cached) @@ -73,41 +72,44 @@ Kubernetes PVC ControllerServer ──→ fulfillment-service Volume API (policy check, tier resolution, vendor dispatch) ↓ (CSI ControllerPublishVolume / ControllerUnpublishVolume) -ControllerServer ──→ fulfillment-service ControlPlane API +ControllerServer ──→ proxy Manager ──→ vendor CSI controller + (routed by "osac.backend"; OSAC-4187, 0.2 temporary) ↓ (CSI NodeStageVolume / NodePublishVolume) NodeServer ──→ proxy Manager ──→ vendor CSI node plugin (routed by "osac.backend" volume context key) ``` -The controller plugin **never talks to vendor CSI drivers directly** — all volume lifecycle and publish/unpublish operations go through the fulfillment-service, which handles policy checks, storage tier resolution, and vendor dispatch. Only the node plugin communicates directly with vendor CSI sockets for mount operations. +The controller plugin delegates **volume lifecycle** (CreateVolume, DeleteVolume) to the fulfillment-service, which handles policy checks, storage tier resolution, and vendor dispatch. For **attach/detach** (ControllerPublish/UnpublishVolume) the controller proxies directly to the vendor CSI controller selected by the `osac.backend` volume-context key — a temporary arrangement for milestone 0.2 (OSAC-4187), expected to be reworked in 0.3. The node plugin communicates directly with vendor CSI sockets for mount operations. + +`osac.backend` carries the **StorageBackend name** resolved from the volume's StorageTier, so the controller's `--vendor-controllers` map (and the node's `--vendor-sockets` map) are keyed by StorageBackend name. A backend whose vendor-controller endpoint is the sentinel `none` (e.g. node-local `lvms`/`topolvm`, which sets `attachRequired=false` and exposes no network CSI controller) makes attach/detach a no-op instead of dialing a vendor. ### Key Subsystems | Package | Purpose | |---------|---------| | `pkg/driver/` | CSI gRPC server (Identity, Controller, Node) implementing the meta-driver pattern | -| `pkg/fulfillment/` | Interfaces and stubs for fulfillment-service volume and control plane operations | +| `pkg/fulfillment/` | VolumeClient interface and in-memory stubs for fulfillment-service volume operations | | `pkg/proxy/` | Lazy gRPC connection manager for vendor CSI sockets (unix + TCP) | ### Volume Context Keys -The controller sets these keys in volume context at creation time, consumed by the node plugin: +The controller sets these keys in volume context at creation time, consumed by the node plugin (and, for attach/detach, by the controller itself): | Key | Purpose | |-----|---------| -| `osac.backend` | Routing key — identifies which vendor CSI socket to proxy to | +| `osac.backend` | Routing key (StorageBackend name) — selects the vendor CSI socket (node) and vendor CSI controller (controller attach/detach) to proxy to | | `osac.volume-id` | Vendor-side volume ID | | `osac.protocol` | Storage protocol (e.g., `nfs`) | ### Dual-Plugin Topology -- **Controller plugin** (Deployment): `CreateVolume`, `DeleteVolume`, `ControllerPublishVolume`, `ControllerUnpublishVolume` — delegates all operations to the fulfillment-service volume and control plane APIs +- **Controller plugin** (Deployment): `CreateVolume`, `DeleteVolume` via the fulfillment-service volume API; `ControllerPublishVolume`, `ControllerUnpublishVolume` proxied to the vendor CSI controller routed by `osac.backend` (OSAC-4187, 0.2 temporary) - **Node plugin** (DaemonSet): `NodeStageVolume`, `NodePublishVolume`, `NodeUnstageVolume`, `NodeUnpublishVolume` — routes to vendor node sockets via `osac.backend`; maintains in-memory `volumeBackends` map to track which vendor handled each volume's stage ### Stub Mode -The real gRPC fulfillment client is not yet implemented. If `--fulfillment-endpoint` is not set, the driver uses in-memory stubs (`VolumeStub`, `ControlPlaneStub`) for development. Setting `--fulfillment-endpoint` currently exits with an error. +The real gRPC fulfillment client is not yet implemented. If `--fulfillment-endpoint` is not set, the driver uses an in-memory `VolumeStub` for development, which reports the `local` backend so attach/detach no-op against the chart's default `local=none` mapping. Setting `--fulfillment-endpoint` currently exits with an error. ## Configuration diff --git a/osac-csi-driver/charts/csi-driver/templates/controller-deployment.yaml b/osac-csi-driver/charts/csi-driver/templates/controller-deployment.yaml index de91af5054..e2f483bc7f 100644 --- a/osac-csi-driver/charts/csi-driver/templates/controller-deployment.yaml +++ b/osac-csi-driver/charts/csi-driver/templates/controller-deployment.yaml @@ -43,6 +43,9 @@ spec: {{- if .Values.controller.vendorSockets }} - "--vendor-sockets={{ .Values.controller.vendorSockets }}" {{- end }} + {{- if .Values.controller.vendorControllers }} + - "--vendor-controllers={{ .Values.controller.vendorControllers }}" + {{- end }} - "--v=2" volumeMounts: - name: osac-socket-dir diff --git a/osac-csi-driver/charts/csi-driver/values.yaml b/osac-csi-driver/charts/csi-driver/values.yaml index 4fde41080f..03bb4b2bf7 100644 --- a/osac-csi-driver/charts/csi-driver/values.yaml +++ b/osac-csi-driver/charts/csi-driver/values.yaml @@ -11,6 +11,20 @@ controller: replicas: 1 clusterID: "" vendorSockets: "" + # Comma-separated backend=endpoint pairs for vendor CSI controllers, used to + # proxy attach/detach (ControllerPublish/UnpublishVolume). The routing key is + # the "osac.backend" volume-context value, which is the StorageBackend name + # resolved from the volume's StorageTier — so each key MUST match a + # StorageBackend's metadata.name that a Cloud Provider Admin created. The names + # below are the expected convention; align them with your actual backends. + # Endpoints for network-attached vendors are the Services created by the + # csi-backends chart (namespace osac-csi-backends, gRPC port 50051). Use the + # special value "none" for node-local backends (e.g. lvms/topolvm) that need no + # controller-side attach — publish/unpublish become a no-op for those. + # An entry for a backend that is not deployed is harmless — it is only dialed + # when a volume names it. + # OSAC-4187 (0.2, temporary): direct vendor-controller proxying; reworked in 0.3. + vendorControllers: "local=none,vast=vast-csi-controller.osac-csi-backends.svc:50051,pure=pure-csi-controller.osac-csi-backends.svc:50051,ontap=trident-csi-controller.osac-csi-backends.svc:50051" # Connection to the OSAC fulfillment-service (runs on the hub / control plane cluster). # The CSI driver runs on tenant clusters and needs credentials that the # fulfillment-service recognises. AAP provisions a Secret with a bearer token diff --git a/osac-csi-driver/cmd/osac-csi-driver/main.go b/osac-csi-driver/cmd/osac-csi-driver/main.go index 628fdb4d76..f88a420511 100644 --- a/osac-csi-driver/cmd/osac-csi-driver/main.go +++ b/osac-csi-driver/cmd/osac-csi-driver/main.go @@ -33,7 +33,11 @@ func main() { "Path to a file containing the bearer token for fulfillment-service authentication") grpcInsecure := flag.Bool("grpc-insecure", false, "Skip TLS server certificate verification") vendorSocketsFlag := flag.String("vendor-sockets", "", - "Comma-separated backend=socketpath pairs (e.g. ontap=/csi/trident/csi.sock)") + "Comma-separated backend=socketpath pairs for vendor node CSI sockets (e.g. ontap=/csi/trident/csi.sock)") + vendorControllersFlag := flag.String("vendor-controllers", "", + "Comma-separated backend=endpoint pairs for vendor CSI controllers, keyed "+ + "by StorageBackend name (e.g. ontap=trident-csi-controller.osac-csi-backends.svc:50051). "+ + "Use the value 'none' for node-local backends that need no attach (e.g. local=none)") driverName := flag.String("driver-name", "csi.osac.openshift.io", "CSI driver name") flag.Parse() @@ -43,19 +47,25 @@ func main() { os.Exit(1) } - vendorSockets, err := parseVendorSockets(*vendorSocketsFlag) + vendorSockets, err := parseBackendMap(*vendorSocketsFlag) if err != nil { fmt.Fprintf(os.Stderr, "Error parsing --vendor-sockets: %v\n", err) os.Exit(1) } + vendorControllers, err := parseBackendMap(*vendorControllersFlag) + if err != nil { + fmt.Fprintf(os.Stderr, "Error parsing --vendor-controllers: %v\n", err) + os.Exit(1) + } + klog.Infof("Starting OSAC CSI driver %s version %s (commit %s)", *driverName, version, gitCommit) klog.Infof("CSI endpoint: %s", *csiEndpoint) klog.Infof("Node ID: %s", *nodeID) klog.Infof("Vendor sockets: %v", vendorSockets) + klog.Infof("Vendor controllers: %v", vendorControllers) var volumeClient fulfillment.VolumeClient - var controlPlaneClient fulfillment.ControlPlaneClient if *fulfillmentEndpoint != "" { // Establish the gRPC connection to the fulfillment-service and back the @@ -77,13 +87,9 @@ func main() { volumeClient = fulfillment.NewVolumeStub("default-backend", "nfs") } - // Attach/publish still goes through the stub; the control-plane attach API - // is out of scope for OSAC-4109 (tracked separately in OSAC-3278/OSAC-4187). - controlPlaneClient = &fulfillment.ControlPlaneStub{} - d, err := driver.NewDriver( *driverName, version, *csiEndpoint, *nodeID, *clusterID, - volumeClient, controlPlaneClient, vendorSockets, + volumeClient, vendorSockets, vendorControllers, ) if err != nil { klog.Fatalf("Failed to create driver: %v", err) @@ -130,7 +136,9 @@ func (f *fileTokenSource) Token() (*oauth2.Token, error) { }, nil } -func parseVendorSockets(s string) (map[string]string, error) { +// parseBackendMap parses a comma-separated list of backend=value pairs into a +// map. It is used for both --vendor-sockets and --vendor-controllers. +func parseBackendMap(s string) (map[string]string, error) { result := make(map[string]string) if s == "" { return result, nil @@ -145,17 +153,17 @@ func parseVendorSockets(s string) (map[string]string, error) { parts := strings.SplitN(pair, "=", 2) if len(parts) != 2 { - return nil, fmt.Errorf("invalid vendor socket pair %q: expected format backend=socketpath", pair) + return nil, fmt.Errorf("invalid pair %q: expected format backend=value", pair) } backend := strings.TrimSpace(parts[0]) - socketPath := strings.TrimSpace(parts[1]) + value := strings.TrimSpace(parts[1]) - if backend == "" || socketPath == "" { - return nil, fmt.Errorf("invalid vendor socket pair %q: backend and socketpath must not be empty", pair) + if backend == "" || value == "" { + return nil, fmt.Errorf("invalid pair %q: backend and value must not be empty", pair) } - result[backend] = socketPath + result[backend] = value } return result, nil diff --git a/osac-csi-driver/pkg/driver/controller.go b/osac-csi-driver/pkg/driver/controller.go index 8b60a51bab..ab7e71a4aa 100644 --- a/osac-csi-driver/pkg/driver/controller.go +++ b/osac-csi-driver/pkg/driver/controller.go @@ -6,6 +6,7 @@ import ( csi "github.com/container-storage-interface/spec/lib/go/csi" "github.com/osac-project/osac/osac-csi-driver/pkg/fulfillment" + "github.com/osac-project/osac/osac-csi-driver/pkg/proxy" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" "k8s.io/klog/v2" @@ -14,26 +15,38 @@ import ( const ( defaultPollInitialInterval = 1 * time.Second defaultPollMaxInterval = 30 * time.Second + + // noAttachEndpoint is the sentinel vendor-controller endpoint for backends + // that need no controller-side attach/detach — node-local storage such as + // lvms/topolvm, whose CSIDriver sets attachRequired=false and exposes no + // network-reachable CSI controller. A backend mapped to this value makes + // ControllerPublish/UnpublishVolume a no-op instead of dialing a vendor. + noAttachEndpoint = "none" ) // ControllerServer implements the CSI Controller service. -// It delegates volume lifecycle to the OSAC fulfillment service and -// publish/unpublish operations to the OSAC control plane. +// It delegates volume lifecycle to the OSAC fulfillment service and proxies +// publish/unpublish operations to vendor CSI controllers. type ControllerServer struct { csi.UnimplementedControllerServer - volumes fulfillment.VolumeClient - controlPlane fulfillment.ControlPlaneClient - clusterID string + volumes fulfillment.VolumeClient + proxyMgr *proxy.Manager + // vendorControllers maps an "osac.backend" name to the gRPC endpoint of that + // vendor's CSI controller, used to proxy publish/unpublish operations. + vendorControllers map[string]string + clusterID string pollInitialInterval time.Duration pollMaxInterval time.Duration } -// NewControllerServer creates a new CSI controller server. -func NewControllerServer(vc fulfillment.VolumeClient, cpc fulfillment.ControlPlaneClient, clusterID string) *ControllerServer { +// NewControllerServer creates a new CSI controller server. vendorControllers maps +// an "osac.backend" name to the gRPC endpoint of that vendor's CSI controller. +func NewControllerServer(vc fulfillment.VolumeClient, proxyMgr *proxy.Manager, vendorControllers map[string]string, clusterID string) *ControllerServer { return &ControllerServer{ volumes: vc, - controlPlane: cpc, + proxyMgr: proxyMgr, + vendorControllers: vendorControllers, clusterID: clusterID, pollInitialInterval: defaultPollInitialInterval, pollMaxInterval: defaultPollMaxInterval, @@ -154,7 +167,13 @@ func (c *ControllerServer) DeleteVolume(ctx context.Context, req *csi.DeleteVolu return &csi.DeleteVolumeResponse{}, nil } -// ControllerPublishVolume publishes a volume to a node via the control plane. +// ControllerPublishVolume attaches a volume to a node by proxying to the vendor +// CSI controller for the volume's backend. +// +// OSAC-4187 (0.2, temporary): the controller talks to the vendor CSI controller +// directly, routed by "osac.backend", instead of going through a fulfillment +// control-plane attach API. This is a stop-gap for milestone 0.2 and is expected +// to be reworked in 0.3. func (c *ControllerServer) ControllerPublishVolume(ctx context.Context, req *csi.ControllerPublishVolumeRequest) (*csi.ControllerPublishVolumeResponse, error) { klog.Infof("ControllerPublishVolume called: volumeId=%s nodeId=%s", req.GetVolumeId(), req.GetNodeId()) @@ -168,20 +187,55 @@ func (c *ControllerServer) ControllerPublishVolume(ctx context.Context, req *csi return nil, status.Error(codes.InvalidArgument, "volume capability is required") } - if err := c.controlPlane.PublishVolume(ctx, req.GetVolumeId(), req.GetNodeId()); err != nil { + backend, vendorVolumeID, err := c.resolvePublishTarget(ctx, req) + if err != nil { + return nil, err + } + + vendorClient, err := c.vendorControllerClient(backend) + if err != nil { + return nil, err + } + if vendorClient == nil { + klog.Infof("Backend %q needs no controller attach (node-local), ControllerPublishVolume is a no-op: volumeId=%s", backend, req.GetVolumeId()) + return &csi.ControllerPublishVolumeResponse{}, nil + } + + // Translate the fulfillment volume id to the vendor-side volume id and + // forward the remaining fields (including secrets) unchanged. + vendorReq := &csi.ControllerPublishVolumeRequest{ + VolumeId: vendorVolumeID, + NodeId: req.GetNodeId(), + VolumeCapability: req.GetVolumeCapability(), + Readonly: req.GetReadonly(), + Secrets: req.GetSecrets(), + VolumeContext: req.GetVolumeContext(), + } + + resp, err := vendorClient.ControllerPublishVolume(ctx, vendorReq) + if err != nil { if st, ok := status.FromError(err); ok && st.Code() == codes.AlreadyExists { klog.Infof("Volume %s already published to node %s", req.GetVolumeId(), req.GetNodeId()) return &csi.ControllerPublishVolumeResponse{}, nil } - klog.Errorf("Failed to publish volume %s to node %s: %v", req.GetVolumeId(), req.GetNodeId(), err) + if isUnimplemented(err) { + klog.Infof("Vendor does not implement ControllerPublishVolume, treating as no-op: volumeId=%s", req.GetVolumeId()) + return &csi.ControllerPublishVolumeResponse{}, nil + } + klog.Errorf("Vendor ControllerPublishVolume failed for volume %s (vendor id %s): %v", req.GetVolumeId(), vendorVolumeID, err) return nil, err } klog.Infof("ControllerPublishVolume succeeded: volumeId=%s nodeId=%s", req.GetVolumeId(), req.GetNodeId()) - return &csi.ControllerPublishVolumeResponse{}, nil + return resp, nil } -// ControllerUnpublishVolume unpublishes a volume from a node via the control plane. +// ControllerUnpublishVolume detaches a volume from a node by proxying to the +// vendor CSI controller for the volume's backend. +// +// OSAC-4187 (0.2, temporary): see ControllerPublishVolume. The unpublish request +// carries no volume context, so the backend and vendor-side volume id are +// resolved from the fulfillment service. func (c *ControllerServer) ControllerUnpublishVolume(ctx context.Context, req *csi.ControllerUnpublishVolumeRequest) (*csi.ControllerUnpublishVolumeResponse, error) { klog.Infof("ControllerUnpublishVolume called: volumeId=%s nodeId=%s", req.GetVolumeId(), req.GetNodeId()) @@ -189,12 +243,40 @@ func (c *ControllerServer) ControllerUnpublishVolume(ctx context.Context, req *c return nil, status.Error(codes.InvalidArgument, "volume ID is required") } - if err := c.controlPlane.UnpublishVolume(ctx, req.GetVolumeId(), req.GetNodeId()); err != nil { + vol, err := c.volumes.GetVolume(ctx, req.GetVolumeId()) + if err != nil { + if st, ok := status.FromError(err); ok && st.Code() == codes.NotFound { + klog.Infof("Volume %s not found, treating unpublish as no-op", req.GetVolumeId()) + return &csi.ControllerUnpublishVolumeResponse{}, nil + } + return nil, err + } + + vendorClient, err := c.vendorControllerClient(vol.Backend) + if err != nil { + return nil, err + } + if vendorClient == nil { + klog.Infof("Backend %q needs no controller attach (node-local), ControllerUnpublishVolume is a no-op: volumeId=%s", vol.Backend, req.GetVolumeId()) + return &csi.ControllerUnpublishVolumeResponse{}, nil + } + + vendorReq := &csi.ControllerUnpublishVolumeRequest{ + VolumeId: vol.VendorVolumeID, + NodeId: req.GetNodeId(), + Secrets: req.GetSecrets(), + } + + if _, err := vendorClient.ControllerUnpublishVolume(ctx, vendorReq); err != nil { if st, ok := status.FromError(err); ok && st.Code() == codes.NotFound { klog.Infof("Volume %s already unpublished from node %s", req.GetVolumeId(), req.GetNodeId()) return &csi.ControllerUnpublishVolumeResponse{}, nil } - klog.Errorf("Failed to unpublish volume %s from node %s: %v", req.GetVolumeId(), req.GetNodeId(), err) + if isUnimplemented(err) { + klog.Infof("Vendor does not implement ControllerUnpublishVolume, treating as no-op: volumeId=%s", req.GetVolumeId()) + return &csi.ControllerUnpublishVolumeResponse{}, nil + } + klog.Errorf("Vendor ControllerUnpublishVolume failed for volume %s (vendor id %s): %v", req.GetVolumeId(), vol.VendorVolumeID, err) return nil, err } @@ -202,6 +284,64 @@ func (c *ControllerServer) ControllerUnpublishVolume(ctx context.Context, req *c return &csi.ControllerUnpublishVolumeResponse{}, nil } +// resolvePublishTarget determines the backend and vendor-side volume id for a +// publish request. It prefers the volume context (present on publish requests) +// and falls back to the fulfillment service for any missing value. +func (c *ControllerServer) resolvePublishTarget(ctx context.Context, req *csi.ControllerPublishVolumeRequest) (backend, vendorVolumeID string, err error) { + vctx := req.GetVolumeContext() + backend = vctx["osac.backend"] + vendorVolumeID = vctx["osac.volume-id"] + + if backend != "" && vendorVolumeID != "" { + return backend, vendorVolumeID, nil + } + + vol, err := c.volumes.GetVolume(ctx, req.GetVolumeId()) + if err != nil { + return "", "", err + } + if backend == "" { + backend = vol.Backend + } + if vendorVolumeID == "" { + vendorVolumeID = vol.VendorVolumeID + } + return backend, vendorVolumeID, nil +} + +// vendorControllerClient resolves the vendor CSI controller endpoint for a +// backend and returns a client connected to it. It returns a nil client (and nil +// error) when the backend is configured with the noAttachEndpoint sentinel, +// signalling that the caller should treat attach/detach as a no-op. +func (c *ControllerServer) vendorControllerClient(backend string) (csi.ControllerClient, error) { + endpoint, err := c.resolveVendorController(backend) + if err != nil { + return nil, err + } + if endpoint == noAttachEndpoint { + return nil, nil + } + conn, err := c.proxyMgr.GetConnection(endpoint) + if err != nil { + return nil, status.Errorf(codes.Unavailable, + "failed to connect to vendor CSI controller for backend %q: %v", backend, err) + } + return csi.NewControllerClient(conn), nil +} + +func (c *ControllerServer) resolveVendorController(backend string) (string, error) { + if backend == "" { + return "", status.Error(codes.InvalidArgument, + "cannot resolve vendor controller: backend is empty") + } + endpoint, ok := c.vendorControllers[backend] + if !ok { + return "", status.Errorf(codes.NotFound, + "no vendor controller configured for backend %q", backend) + } + return endpoint, nil +} + // ValidateVolumeCapabilities confirms volume existence and capabilities. func (c *ControllerServer) ValidateVolumeCapabilities(ctx context.Context, req *csi.ValidateVolumeCapabilitiesRequest) (*csi.ValidateVolumeCapabilitiesResponse, error) { klog.Infof("ValidateVolumeCapabilities called: volumeId=%s", req.GetVolumeId()) diff --git a/osac-csi-driver/pkg/driver/controller_test.go b/osac-csi-driver/pkg/driver/controller_test.go index ce2d62285e..05b1dea992 100644 --- a/osac-csi-driver/pkg/driver/controller_test.go +++ b/osac-csi-driver/pkg/driver/controller_test.go @@ -2,12 +2,17 @@ package driver import ( "context" + "net" + "path/filepath" + "sync" "sync/atomic" "testing" "time" csi "github.com/container-storage-interface/spec/lib/go/csi" "github.com/osac-project/osac/osac-csi-driver/pkg/fulfillment" + "github.com/osac-project/osac/osac-csi-driver/pkg/proxy" + "google.golang.org/grpc" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" ) @@ -37,28 +42,97 @@ func (m *mockVolumeClient) DeleteVolume(ctx context.Context, volumeID string) er return m.deleteVolumeFn(ctx, volumeID) } -type mockControlPlaneClient struct { - publishVolumeFn func(ctx context.Context, volumeID, nodeID string) error - unpublishVolumeFn func(ctx context.Context, volumeID, nodeID string) error +// fakeVendorController is an in-process CSI controller used to exercise the +// controller-side vendor proxy. It records the requests it receives and can be +// configured to return a specific response or error. +type fakeVendorController struct { + csi.UnimplementedControllerServer + + mu sync.Mutex + publishReqs []*csi.ControllerPublishVolumeRequest + unpublishReqs []*csi.ControllerUnpublishVolumeRequest + + publishResp *csi.ControllerPublishVolumeResponse + publishErr error + unpublishErr error +} + +func (f *fakeVendorController) ControllerPublishVolume(_ context.Context, req *csi.ControllerPublishVolumeRequest) (*csi.ControllerPublishVolumeResponse, error) { + f.mu.Lock() + f.publishReqs = append(f.publishReqs, req) + resp, err := f.publishResp, f.publishErr + f.mu.Unlock() + if err != nil { + return nil, err + } + if resp != nil { + return resp, nil + } + return &csi.ControllerPublishVolumeResponse{}, nil } -func (m *mockControlPlaneClient) PublishVolume(ctx context.Context, volumeID, nodeID string) error { - return m.publishVolumeFn(ctx, volumeID, nodeID) +func (f *fakeVendorController) ControllerUnpublishVolume(_ context.Context, req *csi.ControllerUnpublishVolumeRequest) (*csi.ControllerUnpublishVolumeResponse, error) { + f.mu.Lock() + f.unpublishReqs = append(f.unpublishReqs, req) + err := f.unpublishErr + f.mu.Unlock() + if err != nil { + return nil, err + } + return &csi.ControllerUnpublishVolumeResponse{}, nil } -func (m *mockControlPlaneClient) UnpublishVolume(ctx context.Context, volumeID, nodeID string) error { - return m.unpublishVolumeFn(ctx, volumeID, nodeID) +func (f *fakeVendorController) lastPublish() *csi.ControllerPublishVolumeRequest { + f.mu.Lock() + defer f.mu.Unlock() + if len(f.publishReqs) == 0 { + return nil + } + return f.publishReqs[len(f.publishReqs)-1] +} + +func (f *fakeVendorController) lastUnpublish() *csi.ControllerUnpublishVolumeRequest { + f.mu.Lock() + defer f.mu.Unlock() + if len(f.unpublishReqs) == 0 { + return nil + } + return f.unpublishReqs[len(f.unpublishReqs)-1] +} + +// startFakeVendorController serves the given controller on a unix socket and +// returns the socket path. The server is stopped when the test finishes. +func startFakeVendorController(t *testing.T, vendor csi.ControllerServer) string { + t.Helper() + socketPath := filepath.Join(t.TempDir(), "vendor.sock") + listener, err := net.Listen("unix", socketPath) + if err != nil { + t.Fatalf("listening on vendor socket: %v", err) + } + srv := grpc.NewServer() + csi.RegisterControllerServer(srv, vendor) + go func() { _ = srv.Serve(listener) }() + t.Cleanup(srv.Stop) + return socketPath } // --- helpers --- -func newTestController(vc fulfillment.VolumeClient, cpc fulfillment.ControlPlaneClient) *ControllerServer { - cs := NewControllerServer(vc, cpc, "test-cluster") +func newTestController(vc fulfillment.VolumeClient) *ControllerServer { + return newTestControllerWithVendor(vc, nil) +} + +func newTestControllerWithVendor(vc fulfillment.VolumeClient, vendorControllers map[string]string) *ControllerServer { + cs := NewControllerServer(vc, proxy.NewManager(nil), vendorControllers, "test-cluster") cs.pollInitialInterval = 1 * time.Millisecond cs.pollMaxInterval = 5 * time.Millisecond return cs } +func singleCap() *csi.VolumeCapability { + return defaultCaps()[0] +} + func defaultCaps() []*csi.VolumeCapability { return []*csi.VolumeCapability{ { @@ -122,7 +196,7 @@ func TestCreateVolume_Success(t *testing.T) { return vol, nil }, } - cs := newTestController(vc, &mockControlPlaneClient{}) + cs := newTestController(vc) resp, err := cs.CreateVolume(context.Background(), &csi.CreateVolumeRequest{ Name: "pvc-123", @@ -157,7 +231,7 @@ func TestCreateVolume_DefaultTenant(t *testing.T) { return availableVolume("vol-1", "pvc-123"), nil }, } - cs := newTestController(vc, &mockControlPlaneClient{}) + cs := newTestController(vc) _, err := cs.CreateVolume(context.Background(), &csi.CreateVolumeRequest{ Name: "pvc-123", @@ -190,7 +264,7 @@ func TestCreateVolume_PollsUntilAvailable(t *testing.T) { return availableVolume(volumeID, "pvc-123"), nil }, } - cs := newTestController(vc, &mockControlPlaneClient{}) + cs := newTestController(vc) resp, err := cs.CreateVolume(context.Background(), &csi.CreateVolumeRequest{ Name: "pvc-123", @@ -221,7 +295,7 @@ func TestCreateVolume_AlreadyExists(t *testing.T) { return []*fulfillment.VolumeInfo{vol}, nil }, } - cs := newTestController(vc, &mockControlPlaneClient{}) + cs := newTestController(vc) resp, err := cs.CreateVolume(context.Background(), &csi.CreateVolumeRequest{ Name: "pvc-123", @@ -247,7 +321,7 @@ func TestCreateVolume_AlreadyExistsDifferentCapacity(t *testing.T) { return []*fulfillment.VolumeInfo{vol}, nil }, } - cs := newTestController(vc, &mockControlPlaneClient{}) + cs := newTestController(vc) _, err := cs.CreateVolume(context.Background(), &csi.CreateVolumeRequest{ Name: "pvc-123", @@ -269,7 +343,7 @@ func TestCreateVolume_AlreadyExistsCompatibleCapacityRange(t *testing.T) { return []*fulfillment.VolumeInfo{vol}, nil }, } - cs := newTestController(vc, &mockControlPlaneClient{}) + cs := newTestController(vc) resp, err := cs.CreateVolume(context.Background(), &csi.CreateVolumeRequest{ Name: "pvc-123", @@ -294,7 +368,7 @@ func TestCreateVolume_AlreadyExistsNotFoundViaList(t *testing.T) { return nil, nil }, } - cs := newTestController(vc, &mockControlPlaneClient{}) + cs := newTestController(vc) _, err := cs.CreateVolume(context.Background(), &csi.CreateVolumeRequest{ Name: "pvc-123", @@ -319,7 +393,7 @@ func TestCreateVolume_ErrorState(t *testing.T) { }, nil }, } - cs := newTestController(vc, &mockControlPlaneClient{}) + cs := newTestController(vc) _, err := cs.CreateVolume(context.Background(), &csi.CreateVolumeRequest{ Name: "pvc-123", @@ -344,7 +418,7 @@ func TestCreateVolume_ContextCancelled(t *testing.T) { }, nil }, } - cs := newTestController(vc, &mockControlPlaneClient{}) + cs := newTestController(vc) ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) defer cancel() @@ -358,7 +432,7 @@ func TestCreateVolume_ContextCancelled(t *testing.T) { } func TestCreateVolume_MissingName(t *testing.T) { - cs := newTestController(&mockVolumeClient{}, &mockControlPlaneClient{}) + cs := newTestController(&mockVolumeClient{}) _, err := cs.CreateVolume(context.Background(), &csi.CreateVolumeRequest{ VolumeCapabilities: defaultCaps(), @@ -368,7 +442,7 @@ func TestCreateVolume_MissingName(t *testing.T) { } func TestCreateVolume_MissingCapabilities(t *testing.T) { - cs := newTestController(&mockVolumeClient{}, &mockControlPlaneClient{}) + cs := newTestController(&mockVolumeClient{}) _, err := cs.CreateVolume(context.Background(), &csi.CreateVolumeRequest{ Name: "pvc-123", @@ -378,7 +452,7 @@ func TestCreateVolume_MissingCapabilities(t *testing.T) { } func TestCreateVolume_MissingTier(t *testing.T) { - cs := newTestController(&mockVolumeClient{}, &mockControlPlaneClient{}) + cs := newTestController(&mockVolumeClient{}) _, err := cs.CreateVolume(context.Background(), &csi.CreateVolumeRequest{ Name: "pvc-123", @@ -394,7 +468,7 @@ func TestCreateVolume_CreateError(t *testing.T) { return nil, status.Error(codes.Unavailable, "connection refused") }, } - cs := newTestController(vc, &mockControlPlaneClient{}) + cs := newTestController(vc) _, err := cs.CreateVolume(context.Background(), &csi.CreateVolumeRequest{ Name: "pvc-123", @@ -414,7 +488,7 @@ func TestDeleteVolume_Success(t *testing.T) { return nil }, } - cs := newTestController(vc, &mockControlPlaneClient{}) + cs := newTestController(vc) _, err := cs.DeleteVolume(context.Background(), &csi.DeleteVolumeRequest{ VolumeId: "vol-1", @@ -428,7 +502,7 @@ func TestDeleteVolume_Success(t *testing.T) { } func TestDeleteVolume_MissingVolumeID(t *testing.T) { - cs := newTestController(&mockVolumeClient{}, &mockControlPlaneClient{}) + cs := newTestController(&mockVolumeClient{}) _, err := cs.DeleteVolume(context.Background(), &csi.DeleteVolumeRequest{}) assertCode(t, err, codes.InvalidArgument) @@ -440,7 +514,7 @@ func TestDeleteVolume_NotFoundIsSuccess(t *testing.T) { return status.Error(codes.NotFound, "volume not found") }, } - cs := newTestController(vc, &mockControlPlaneClient{}) + cs := newTestController(vc) _, err := cs.DeleteVolume(context.Background(), &csi.DeleteVolumeRequest{ VolumeId: "vol-1", @@ -456,7 +530,7 @@ func TestDeleteVolume_Error(t *testing.T) { return status.Error(codes.Unavailable, "service down") }, } - cs := newTestController(vc, &mockControlPlaneClient{}) + cs := newTestController(vc) _, err := cs.DeleteVolume(context.Background(), &csi.DeleteVolumeRequest{ VolumeId: "vol-1", @@ -467,69 +541,91 @@ func TestDeleteVolume_Error(t *testing.T) { // --- ControllerPublishVolume tests --- func TestControllerPublishVolume_Success(t *testing.T) { - var gotVolumeID, gotNodeID string - cpc := &mockControlPlaneClient{ - publishVolumeFn: func(_ context.Context, volumeID, nodeID string) error { - gotVolumeID = volumeID - gotNodeID = nodeID - return nil + vendor := &fakeVendorController{ + publishResp: &csi.ControllerPublishVolumeResponse{ + PublishContext: map[string]string{"vendor.device": "/dev/sdx"}, }, } - cs := newTestController(&mockVolumeClient{}, cpc) + socket := startFakeVendorController(t, vendor) + cs := newTestControllerWithVendor(&mockVolumeClient{}, map[string]string{"test-backend": socket}) - _, err := cs.ControllerPublishVolume(context.Background(), &csi.ControllerPublishVolumeRequest{ - VolumeId: "vol-1", - NodeId: "node-1", - VolumeCapability: &csi.VolumeCapability{ - AccessType: &csi.VolumeCapability_Mount{ - Mount: &csi.VolumeCapability_MountVolume{}, - }, - AccessMode: &csi.VolumeCapability_AccessMode{ - Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER, - }, + resp, err := cs.ControllerPublishVolume(context.Background(), &csi.ControllerPublishVolumeRequest{ + VolumeId: "vol-1", + NodeId: "node-1", + VolumeCapability: singleCap(), + VolumeContext: map[string]string{ + "osac.backend": "test-backend", + "osac.volume-id": "vendor-vol-1", }, }) if err != nil { t.Fatalf("unexpected error: %v", err) } - if gotVolumeID != "vol-1" { - t.Errorf("expected volume ID 'vol-1', got %q", gotVolumeID) + // The vendor's publish context must be forwarded back to the CO. + if resp.GetPublishContext()["vendor.device"] != "/dev/sdx" { + t.Errorf("expected vendor publish context to be forwarded, got %v", resp.GetPublishContext()) } - if gotNodeID != "node-1" { - t.Errorf("expected node ID 'node-1', got %q", gotNodeID) + last := vendor.lastPublish() + if last == nil { + t.Fatal("vendor did not receive a publish request") + } + // The fulfillment volume id must be translated to the vendor volume id. + if last.GetVolumeId() != "vendor-vol-1" { + t.Errorf("expected vendor volume ID 'vendor-vol-1', got %q", last.GetVolumeId()) + } + if last.GetNodeId() != "node-1" { + t.Errorf("expected node ID 'node-1', got %q", last.GetNodeId()) + } +} + +// When the request carries no volume context (or partial context), the backend +// and vendor volume id are resolved from the fulfillment service. +func TestControllerPublishVolume_ResolvesViaVolumeClient(t *testing.T) { + vendor := &fakeVendorController{} + socket := startFakeVendorController(t, vendor) + vc := &mockVolumeClient{ + getVolumeFn: func(_ context.Context, volumeID string) (*fulfillment.VolumeInfo, error) { + return availableVolume(volumeID, "pvc-123"), nil + }, + } + cs := newTestControllerWithVendor(vc, map[string]string{"test-backend": socket}) + + _, err := cs.ControllerPublishVolume(context.Background(), &csi.ControllerPublishVolumeRequest{ + VolumeId: "vol-1", + NodeId: "node-1", + VolumeCapability: singleCap(), + }) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + last := vendor.lastPublish() + if last == nil || last.GetVolumeId() != "vendor-vol-1" { + t.Fatalf("expected vendor volume ID 'vendor-vol-1' resolved via GetVolume, got %v", last) } } func TestControllerPublishVolume_MissingVolumeID(t *testing.T) { - cs := newTestController(&mockVolumeClient{}, &mockControlPlaneClient{}) + cs := newTestController(&mockVolumeClient{}) _, err := cs.ControllerPublishVolume(context.Background(), &csi.ControllerPublishVolumeRequest{ - NodeId: "node-1", - VolumeCapability: &csi.VolumeCapability{ - AccessType: &csi.VolumeCapability_Mount{ - Mount: &csi.VolumeCapability_MountVolume{}, - }, - }, + NodeId: "node-1", + VolumeCapability: singleCap(), }) assertCode(t, err, codes.InvalidArgument) } func TestControllerPublishVolume_MissingNodeID(t *testing.T) { - cs := newTestController(&mockVolumeClient{}, &mockControlPlaneClient{}) + cs := newTestController(&mockVolumeClient{}) _, err := cs.ControllerPublishVolume(context.Background(), &csi.ControllerPublishVolumeRequest{ - VolumeId: "vol-1", - VolumeCapability: &csi.VolumeCapability{ - AccessType: &csi.VolumeCapability_Mount{ - Mount: &csi.VolumeCapability_MountVolume{}, - }, - }, + VolumeId: "vol-1", + VolumeCapability: singleCap(), }) assertCode(t, err, codes.InvalidArgument) } func TestControllerPublishVolume_MissingCapability(t *testing.T) { - cs := newTestController(&mockVolumeClient{}, &mockControlPlaneClient{}) + cs := newTestController(&mockVolumeClient{}) _, err := cs.ControllerPublishVolume(context.Background(), &csi.ControllerPublishVolumeRequest{ VolumeId: "vol-1", @@ -538,24 +634,53 @@ func TestControllerPublishVolume_MissingCapability(t *testing.T) { assertCode(t, err, codes.InvalidArgument) } -func TestControllerPublishVolume_AlreadyExistsIsSuccess(t *testing.T) { - cpc := &mockControlPlaneClient{ - publishVolumeFn: func(_ context.Context, _, _ string) error { - return status.Error(codes.AlreadyExists, "already published") +// An unknown backend has no configured vendor controller to route to. +func TestControllerPublishVolume_UnknownBackend(t *testing.T) { + cs := newTestControllerWithVendor(&mockVolumeClient{}, map[string]string{}) + + _, err := cs.ControllerPublishVolume(context.Background(), &csi.ControllerPublishVolumeRequest{ + VolumeId: "vol-1", + NodeId: "node-1", + VolumeCapability: singleCap(), + VolumeContext: map[string]string{ + "osac.backend": "test-backend", + "osac.volume-id": "vendor-vol-1", + }, + }) + assertCode(t, err, codes.NotFound) +} + +// A nonexistent volume with no context cannot be resolved for publish. +func TestControllerPublishVolume_VolumeNotFound(t *testing.T) { + vc := &mockVolumeClient{ + getVolumeFn: func(_ context.Context, _ string) (*fulfillment.VolumeInfo, error) { + return nil, status.Error(codes.NotFound, "not found") }, } - cs := newTestController(&mockVolumeClient{}, cpc) + cs := newTestControllerWithVendor(vc, map[string]string{"test-backend": "unused"}) _, err := cs.ControllerPublishVolume(context.Background(), &csi.ControllerPublishVolumeRequest{ - VolumeId: "vol-1", - NodeId: "node-1", - VolumeCapability: &csi.VolumeCapability{ - AccessType: &csi.VolumeCapability_Mount{ - Mount: &csi.VolumeCapability_MountVolume{}, - }, - AccessMode: &csi.VolumeCapability_AccessMode{ - Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER, - }, + VolumeId: "vol-1", + NodeId: "node-1", + VolumeCapability: singleCap(), + }) + assertCode(t, err, codes.NotFound) +} + +func TestControllerPublishVolume_AlreadyExistsIsSuccess(t *testing.T) { + vendor := &fakeVendorController{ + publishErr: status.Error(codes.AlreadyExists, "already published"), + } + socket := startFakeVendorController(t, vendor) + cs := newTestControllerWithVendor(&mockVolumeClient{}, map[string]string{"test-backend": socket}) + + _, err := cs.ControllerPublishVolume(context.Background(), &csi.ControllerPublishVolumeRequest{ + VolumeId: "vol-1", + NodeId: "node-1", + VolumeCapability: singleCap(), + VolumeContext: map[string]string{ + "osac.backend": "test-backend", + "osac.volume-id": "vendor-vol-1", }, }) if err != nil { @@ -563,41 +688,81 @@ func TestControllerPublishVolume_AlreadyExistsIsSuccess(t *testing.T) { } } -func TestControllerPublishVolume_Error(t *testing.T) { - cpc := &mockControlPlaneClient{ - publishVolumeFn: func(_ context.Context, _, _ string) error { - return status.Error(codes.Unavailable, "service down") +// NFS-style vendors that do not implement controller attach are treated as a no-op. +func TestControllerPublishVolume_UnimplementedIsSuccess(t *testing.T) { + vendor := &fakeVendorController{ + publishErr: status.Error(codes.Unimplemented, "not implemented"), + } + socket := startFakeVendorController(t, vendor) + cs := newTestControllerWithVendor(&mockVolumeClient{}, map[string]string{"test-backend": socket}) + + _, err := cs.ControllerPublishVolume(context.Background(), &csi.ControllerPublishVolumeRequest{ + VolumeId: "vol-1", + NodeId: "node-1", + VolumeCapability: singleCap(), + VolumeContext: map[string]string{ + "osac.backend": "test-backend", + "osac.volume-id": "vendor-vol-1", }, + }) + if err != nil { + t.Fatalf("expected success for Unimplemented, got: %v", err) } - cs := newTestController(&mockVolumeClient{}, cpc) +} + +func TestControllerPublishVolume_Error(t *testing.T) { + vendor := &fakeVendorController{ + publishErr: status.Error(codes.Unavailable, "service down"), + } + socket := startFakeVendorController(t, vendor) + cs := newTestControllerWithVendor(&mockVolumeClient{}, map[string]string{"test-backend": socket}) _, err := cs.ControllerPublishVolume(context.Background(), &csi.ControllerPublishVolumeRequest{ - VolumeId: "vol-1", - NodeId: "node-1", - VolumeCapability: &csi.VolumeCapability{ - AccessType: &csi.VolumeCapability_Mount{ - Mount: &csi.VolumeCapability_MountVolume{}, - }, - AccessMode: &csi.VolumeCapability_AccessMode{ - Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER, - }, + VolumeId: "vol-1", + NodeId: "node-1", + VolumeCapability: singleCap(), + VolumeContext: map[string]string{ + "osac.backend": "test-backend", + "osac.volume-id": "vendor-vol-1", }, }) assertCode(t, err, codes.Unavailable) } +// A backend mapped to the "none" sentinel (node-local storage such as lvms) +// needs no controller-side attach, so publish is a no-op and no vendor +// controller is dialed. +func TestControllerPublishVolume_NoAttachBackendIsNoop(t *testing.T) { + cs := newTestControllerWithVendor(&mockVolumeClient{}, map[string]string{"local": noAttachEndpoint}) + + resp, err := cs.ControllerPublishVolume(context.Background(), &csi.ControllerPublishVolumeRequest{ + VolumeId: "vol-1", + NodeId: "node-1", + VolumeCapability: singleCap(), + VolumeContext: map[string]string{ + "osac.backend": "local", + "osac.volume-id": "vendor-vol-1", + }, + }) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(resp.GetPublishContext()) != 0 { + t.Errorf("expected empty publish context for no-attach backend, got %v", resp.GetPublishContext()) + } +} + // --- ControllerUnpublishVolume tests --- func TestControllerUnpublishVolume_Success(t *testing.T) { - var gotVolumeID, gotNodeID string - cpc := &mockControlPlaneClient{ - unpublishVolumeFn: func(_ context.Context, volumeID, nodeID string) error { - gotVolumeID = volumeID - gotNodeID = nodeID - return nil + vendor := &fakeVendorController{} + socket := startFakeVendorController(t, vendor) + vc := &mockVolumeClient{ + getVolumeFn: func(_ context.Context, volumeID string) (*fulfillment.VolumeInfo, error) { + return availableVolume(volumeID, "pvc-123"), nil }, } - cs := newTestController(&mockVolumeClient{}, cpc) + cs := newTestControllerWithVendor(vc, map[string]string{"test-backend": socket}) _, err := cs.ControllerUnpublishVolume(context.Background(), &csi.ControllerUnpublishVolumeRequest{ VolumeId: "vol-1", @@ -606,16 +771,22 @@ func TestControllerUnpublishVolume_Success(t *testing.T) { if err != nil { t.Fatalf("unexpected error: %v", err) } - if gotVolumeID != "vol-1" { - t.Errorf("expected volume ID 'vol-1', got %q", gotVolumeID) + last := vendor.lastUnpublish() + if last == nil { + t.Fatal("vendor did not receive an unpublish request") + } + // The backend and vendor volume id are resolved from the fulfillment service + // because the unpublish request carries no volume context. + if last.GetVolumeId() != "vendor-vol-1" { + t.Errorf("expected vendor volume ID 'vendor-vol-1', got %q", last.GetVolumeId()) } - if gotNodeID != "node-1" { - t.Errorf("expected node ID 'node-1', got %q", gotNodeID) + if last.GetNodeId() != "node-1" { + t.Errorf("expected node ID 'node-1', got %q", last.GetNodeId()) } } func TestControllerUnpublishVolume_MissingVolumeID(t *testing.T) { - cs := newTestController(&mockVolumeClient{}, &mockControlPlaneClient{}) + cs := newTestController(&mockVolumeClient{}) _, err := cs.ControllerUnpublishVolume(context.Background(), &csi.ControllerUnpublishVolumeRequest{ NodeId: "node-1", @@ -623,13 +794,35 @@ func TestControllerUnpublishVolume_MissingVolumeID(t *testing.T) { assertCode(t, err, codes.InvalidArgument) } -func TestControllerUnpublishVolume_NotFoundIsSuccess(t *testing.T) { - cpc := &mockControlPlaneClient{ - unpublishVolumeFn: func(_ context.Context, _, _ string) error { - return status.Error(codes.NotFound, "not published") +// A volume the fulfillment service no longer knows about is already detached. +func TestControllerUnpublishVolume_VolumeNotFoundIsSuccess(t *testing.T) { + vc := &mockVolumeClient{ + getVolumeFn: func(_ context.Context, _ string) (*fulfillment.VolumeInfo, error) { + return nil, status.Error(codes.NotFound, "volume not found") }, } - cs := newTestController(&mockVolumeClient{}, cpc) + cs := newTestControllerWithVendor(vc, nil) + + _, err := cs.ControllerUnpublishVolume(context.Background(), &csi.ControllerUnpublishVolumeRequest{ + VolumeId: "vol-1", + NodeId: "node-1", + }) + if err != nil { + t.Fatalf("expected success when volume not found, got: %v", err) + } +} + +func TestControllerUnpublishVolume_VendorNotFoundIsSuccess(t *testing.T) { + vendor := &fakeVendorController{ + unpublishErr: status.Error(codes.NotFound, "not published"), + } + socket := startFakeVendorController(t, vendor) + vc := &mockVolumeClient{ + getVolumeFn: func(_ context.Context, volumeID string) (*fulfillment.VolumeInfo, error) { + return availableVolume(volumeID, "pvc-123"), nil + }, + } + cs := newTestControllerWithVendor(vc, map[string]string{"test-backend": socket}) _, err := cs.ControllerUnpublishVolume(context.Background(), &csi.ControllerUnpublishVolumeRequest{ VolumeId: "vol-1", @@ -640,13 +833,53 @@ func TestControllerUnpublishVolume_NotFoundIsSuccess(t *testing.T) { } } +func TestControllerUnpublishVolume_UnimplementedIsSuccess(t *testing.T) { + vendor := &fakeVendorController{ + unpublishErr: status.Error(codes.Unimplemented, "not implemented"), + } + socket := startFakeVendorController(t, vendor) + vc := &mockVolumeClient{ + getVolumeFn: func(_ context.Context, volumeID string) (*fulfillment.VolumeInfo, error) { + return availableVolume(volumeID, "pvc-123"), nil + }, + } + cs := newTestControllerWithVendor(vc, map[string]string{"test-backend": socket}) + + _, err := cs.ControllerUnpublishVolume(context.Background(), &csi.ControllerUnpublishVolumeRequest{ + VolumeId: "vol-1", + NodeId: "node-1", + }) + if err != nil { + t.Fatalf("expected success for Unimplemented, got: %v", err) + } +} + +func TestControllerUnpublishVolume_UnknownBackend(t *testing.T) { + vc := &mockVolumeClient{ + getVolumeFn: func(_ context.Context, volumeID string) (*fulfillment.VolumeInfo, error) { + return availableVolume(volumeID, "pvc-123"), nil + }, + } + cs := newTestControllerWithVendor(vc, map[string]string{}) + + _, err := cs.ControllerUnpublishVolume(context.Background(), &csi.ControllerUnpublishVolumeRequest{ + VolumeId: "vol-1", + NodeId: "node-1", + }) + assertCode(t, err, codes.NotFound) +} + func TestControllerUnpublishVolume_Error(t *testing.T) { - cpc := &mockControlPlaneClient{ - unpublishVolumeFn: func(_ context.Context, _, _ string) error { - return status.Error(codes.Unavailable, "service down") + vendor := &fakeVendorController{ + unpublishErr: status.Error(codes.Unavailable, "service down"), + } + socket := startFakeVendorController(t, vendor) + vc := &mockVolumeClient{ + getVolumeFn: func(_ context.Context, volumeID string) (*fulfillment.VolumeInfo, error) { + return availableVolume(volumeID, "pvc-123"), nil }, } - cs := newTestController(&mockVolumeClient{}, cpc) + cs := newTestControllerWithVendor(vc, map[string]string{"test-backend": socket}) _, err := cs.ControllerUnpublishVolume(context.Background(), &csi.ControllerUnpublishVolumeRequest{ VolumeId: "vol-1", @@ -655,6 +888,27 @@ func TestControllerUnpublishVolume_Error(t *testing.T) { assertCode(t, err, codes.Unavailable) } +// A backend mapped to the "none" sentinel needs no controller-side detach, so +// unpublish is a no-op and no vendor controller is dialed. +func TestControllerUnpublishVolume_NoAttachBackendIsNoop(t *testing.T) { + vc := &mockVolumeClient{ + getVolumeFn: func(_ context.Context, volumeID string) (*fulfillment.VolumeInfo, error) { + vol := availableVolume(volumeID, "pvc-123") + vol.Backend = "local" + return vol, nil + }, + } + cs := newTestControllerWithVendor(vc, map[string]string{"local": noAttachEndpoint}) + + _, err := cs.ControllerUnpublishVolume(context.Background(), &csi.ControllerUnpublishVolumeRequest{ + VolumeId: "vol-1", + NodeId: "node-1", + }) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } +} + // --- ValidateVolumeCapabilities tests --- func TestValidateVolumeCapabilities_Success(t *testing.T) { @@ -663,7 +917,7 @@ func TestValidateVolumeCapabilities_Success(t *testing.T) { return availableVolume(volumeID, "pvc-123"), nil }, } - cs := newTestController(vc, &mockControlPlaneClient{}) + cs := newTestController(vc) resp, err := cs.ValidateVolumeCapabilities(context.Background(), &csi.ValidateVolumeCapabilitiesRequest{ VolumeId: "vol-1", @@ -683,7 +937,7 @@ func TestValidateVolumeCapabilities_VolumeNotFound(t *testing.T) { return nil, status.Errorf(codes.NotFound, "not found") }, } - cs := newTestController(vc, &mockControlPlaneClient{}) + cs := newTestController(vc) _, err := cs.ValidateVolumeCapabilities(context.Background(), &csi.ValidateVolumeCapabilitiesRequest{ VolumeId: "vol-1", @@ -695,7 +949,7 @@ func TestValidateVolumeCapabilities_VolumeNotFound(t *testing.T) { // --- ControllerGetCapabilities tests --- func TestControllerGetCapabilities(t *testing.T) { - cs := newTestController(&mockVolumeClient{}, &mockControlPlaneClient{}) + cs := newTestController(&mockVolumeClient{}) resp, err := cs.ControllerGetCapabilities(context.Background(), &csi.ControllerGetCapabilitiesRequest{}) if err != nil { diff --git a/osac-csi-driver/pkg/driver/driver.go b/osac-csi-driver/pkg/driver/driver.go index 95227cb326..0ab4f111e3 100644 --- a/osac-csi-driver/pkg/driver/driver.go +++ b/osac-csi-driver/pkg/driver/driver.go @@ -30,8 +30,11 @@ type Driver struct { node csi.NodeServer } -// NewDriver creates a new OSAC CSI driver instance. -func NewDriver(name, version, endpoint, nodeID, clusterID string, vc fulfillment.VolumeClient, cpc fulfillment.ControlPlaneClient, vendorSockets map[string]string) (*Driver, error) { +// NewDriver creates a new OSAC CSI driver instance. vendorSockets maps a backend +// name to the vendor node CSI socket (used by the node plugin), and +// vendorControllers maps a backend name to the vendor CSI controller endpoint +// (used by the controller plugin for publish/unpublish). +func NewDriver(name, version, endpoint, nodeID, clusterID string, vc fulfillment.VolumeClient, vendorSockets, vendorControllers map[string]string) (*Driver, error) { if name == "" { return nil, fmt.Errorf("driver name is required") } @@ -53,7 +56,7 @@ func NewDriver(name, version, endpoint, nodeID, clusterID string, vc fulfillment nodeID: nodeID, endpoint: endpoint, identity: NewIdentityServer(name, version), - controller: NewControllerServer(vc, cpc, clusterID), + controller: NewControllerServer(vc, proxyMgr, vendorControllers, clusterID), node: NewNodeServer(nodeID, proxyMgr, vendorSockets), }, nil } diff --git a/osac-csi-driver/pkg/fulfillment/client.go b/osac-csi-driver/pkg/fulfillment/client.go index d27474f9fa..eac152bcb1 100644 --- a/osac-csi-driver/pkg/fulfillment/client.go +++ b/osac-csi-driver/pkg/fulfillment/client.go @@ -1,4 +1,5 @@ // Package fulfillment provides the client interfaces for the OSAC fulfillment -// service. The CSI driver uses VolumeClient for volume lifecycle and -// ControlPlaneClient for publish/unpublish operations. +// service. The CSI driver uses VolumeClient for volume lifecycle; publish and +// unpublish are proxied to vendor CSI controllers by the driver's controller +// server (see pkg/driver), not routed through this package. package fulfillment diff --git a/osac-csi-driver/pkg/fulfillment/controlplane.go b/osac-csi-driver/pkg/fulfillment/controlplane.go deleted file mode 100644 index b3bdf0379c..0000000000 --- a/osac-csi-driver/pkg/fulfillment/controlplane.go +++ /dev/null @@ -1,10 +0,0 @@ -package fulfillment - -import "context" - -// ControlPlaneClient manages volume publish/unpublish operations through the -// OSAC control plane. Both methods block until the operation completes. -type ControlPlaneClient interface { - PublishVolume(ctx context.Context, volumeID, nodeID string) error - UnpublishVolume(ctx context.Context, volumeID, nodeID string) error -} diff --git a/osac-csi-driver/pkg/fulfillment/stubs.go b/osac-csi-driver/pkg/fulfillment/stubs.go index a53d2c0f17..8135caa4d2 100644 --- a/osac-csi-driver/pkg/fulfillment/stubs.go +++ b/osac-csi-driver/pkg/fulfillment/stubs.go @@ -94,16 +94,3 @@ func (s *VolumeStub) DeleteVolume(_ context.Context, volumeID string) error { delete(s.volumes, volumeID) return nil } - -// ControlPlaneStub is a no-op ControlPlaneClient used during development. -type ControlPlaneStub struct{} - -func (s *ControlPlaneStub) PublishVolume(_ context.Context, volumeID, nodeID string) error { - klog.Infof("control plane stub: PublishVolume(volumeID=%q, nodeID=%q)", volumeID, nodeID) - return nil -} - -func (s *ControlPlaneStub) UnpublishVolume(_ context.Context, volumeID, nodeID string) error { - klog.Infof("control plane stub: UnpublishVolume(volumeID=%q, nodeID=%q)", volumeID, nodeID) - return nil -} diff --git a/osac-csi-driver/test/sanity/fakevendor_test.go b/osac-csi-driver/test/sanity/fakevendor_test.go index 7c323636b8..d282b889bf 100644 --- a/osac-csi-driver/test/sanity/fakevendor_test.go +++ b/osac-csi-driver/test/sanity/fakevendor_test.go @@ -197,20 +197,10 @@ func (f *fakeVendor) ControllerPublishVolume( return nil, status.Error(codes.InvalidArgument, "node ID is required") } - f.mu.Lock() - _, exists := f.volumes[req.GetVolumeId()] - f.mu.Unlock() - - if !exists { - return nil, status.Errorf(codes.NotFound, - "volume %q not found", req.GetVolumeId()) - } - - if req.GetNodeId() != f.nodeID { - return nil, status.Errorf(codes.NotFound, - "node %q not found", req.GetNodeId()) - } - + // The OSAC meta-driver creates volumes through the fulfillment service, not + // through this vendor, so the vendor-side volume id it forwards on publish is + // one this double never saw via CreateVolume. Accept any volume/node id — the + // existence semantics are exercised against the real vendor, not here. return &csi.ControllerPublishVolumeResponse{ PublishContext: map[string]string{ "fake.device": "/dev/fake0", diff --git a/osac-csi-driver/test/sanity/sanity_test.go b/osac-csi-driver/test/sanity/sanity_test.go index 94b814c401..a495c9107a 100644 --- a/osac-csi-driver/test/sanity/sanity_test.go +++ b/osac-csi-driver/test/sanity/sanity_test.go @@ -14,8 +14,10 @@ import ( ) // metaDriverSkips lists CSI sanity tests that don't apply to the OSAC -// controller. The controller delegates publish/unpublish to the control -// plane stub which does not validate volume or node existence. +// controller. Publish/unpublish are proxied to the fake vendor controller, a +// permissive test double that accepts any volume and node id (volume creation +// is decoupled from attach in stub mode), so it does not reproduce the +// volume/node existence failures these tests expect. var metaDriverSkips = []string{ "ControllerPublishVolume.*should fail when the volume does not exist", "ControllerPublishVolume.*should fail when the node does not exist", @@ -45,11 +47,15 @@ func TestSanity(t *testing.T) { defer vendorSrv.GracefulStop() vc := fulfillment.NewVolumeStub(backendName, "nfs") - cpc := &fulfillment.ControlPlaneStub{} vendorSockets := map[string]string{ backendName: vendorSocket, } + // The controller proxies publish/unpublish to the same fake vendor, which + // serves the CSI controller service on this socket alongside the node service. + vendorControllers := map[string]string{ + backendName: vendorSocket, + } d, err := driver.NewDriver( "csi.osac.openshift.io", @@ -58,8 +64,8 @@ func TestSanity(t *testing.T) { nodeID, "test-cluster", vc, - cpc, vendorSockets, + vendorControllers, ) if err != nil { t.Fatalf("creating driver: %v", err)