Experimental: volume bindings

cmd/main.go

@@ -15,6 +15,7 @@ import (
 	"github.com/opiproject/opi-spdk-bridge/pkg/frontend"
 	"github.com/opiproject/opi-spdk-bridge/pkg/kvm"
 	"github.com/opiproject/opi-spdk-bridge/pkg/middleend"
+	"github.com/opiproject/opi-spdk-bridge/pkg/volume"
 
 	pb "github.com/opiproject/opi-api/storage/v1alpha1/gen/go"
 	"google.golang.org/grpc"
@@ -47,9 +48,10 @@ func main() {
 	}
 	s := grpc.NewServer()
 
+	registry := volume.NewRegistry()
 	jsonRPC := spdk.NewSpdkJSONRPC(spdkAddress)
-	backendServer := backend.NewServer(jsonRPC)
-	middleendServer := middleend.NewServer(jsonRPC)
+	backendServer := backend.NewServer(jsonRPC, registry)
+	middleendServer := middleend.NewServer(jsonRPC, registry)
 
 	if useKvm {
 		log.Println("Creating KVM server.")

pkg/backend/backend.go

@@ -8,6 +8,7 @@ package backend
 import (
 	"github.com/opiproject/gospdk/spdk"
 	pb "github.com/opiproject/opi-api/storage/v1alpha1/gen/go"
+	"github.com/opiproject/opi-spdk-bridge/pkg/volume"
 )
 
 // TODO: can we combine all of volume types into a single list?
@@ -29,11 +30,12 @@ type Server struct {
 	rpc        spdk.JSONRPC
 	Volumes    VolumeParameters
 	Pagination map[string]int
+	registry   *volume.Registry
 }
 
 // NewServer creates initialized instance of BackEnd server communicating
 // with provided jsonRPC
-func NewServer(jsonRPC spdk.JSONRPC) *Server {
+func NewServer(jsonRPC spdk.JSONRPC, registry *volume.Registry) *Server {
 	return &Server{
 		rpc: jsonRPC,
 		Volumes: VolumeParameters{
@@ -42,5 +44,6 @@ func NewServer(jsonRPC spdk.JSONRPC) *Server {
 			NvmeVolumes: make(map[string]*pb.NVMfRemoteController),
 		},
 		Pagination: make(map[string]int),
+		registry:   registry,
 	}
 }

pkg/backend/null.go

@@ -14,6 +14,7 @@ import (
 	pc "github.com/opiproject/opi-api/common/v1/gen/go"
 	pb "github.com/opiproject/opi-api/storage/v1alpha1/gen/go"
 	"github.com/opiproject/opi-spdk-bridge/pkg/server"
+	"github.com/opiproject/opi-spdk-bridge/pkg/volume"
 
 	"github.com/google/uuid"
 	"github.com/ulule/deepcopier"
@@ -26,14 +27,20 @@ import (
 func (s *Server) CreateNullDebug(_ context.Context, in *pb.CreateNullDebugRequest) (*pb.NullDebug, error) {
 	log.Printf("CreateNullDebug: Received from client: %v", in)
 	// idempotent API when called with same key, should return same object
-	volume, ok := s.Volumes.NullVolumes[in.NullDebug.Handle.Value]
-	if ok {
-		log.Printf("Already existing NullDebug with id %v", in.NullDebug.Handle.Value)
-		return volume, nil
+	// vol, ok := s.Volumes.NullVolumes[in.NullDebug.Handle.Value]
+	vol := s.registry.Find(in.NullDebug.Handle.Value)
+	if vol != nil {
+		if null := vol.Descriptor().ToNullDebug(); null != nil {
+			log.Printf("Already existing NullDebug with id %v", in.NullDebug.Handle.Value)
+			return null, nil
+		} else {
+			// panic?
+		}
 	}
+	vol = volume.NewNullVolume(in.NullDebug)
 	// not found, so create a new one
 	params := spdk.BdevNullCreateParams{
-		Name:      in.NullDebug.Handle.Value,
+		Name:      vol.BdevName(),
 		BlockSize: 512,
 		NumBlocks: 64,
 	}
@@ -55,38 +62,43 @@ func (s *Server) CreateNullDebug(_ context.Context, in *pb.CreateNullDebugReques
 		log.Printf("error: %v", err)
 		return nil, err
 	}
+	s.registry.Add(in.NullDebug.Handle.Value, vol)
 	s.Volumes.NullVolumes[in.NullDebug.Handle.Value] = response
 	return response, nil
 }
 
 // DeleteNullDebug deletes a Null Debug instance
 func (s *Server) DeleteNullDebug(_ context.Context, in *pb.DeleteNullDebugRequest) (*emptypb.Empty, error) {
 	log.Printf("DeleteNullDebug: Received from client: %v", in)
-	volume, ok := s.Volumes.NullVolumes[in.Name]
-	if !ok {
+	vol := s.registry.Find(in.Name)
+	if vol == nil {
 		if in.AllowMissing {
 			return &emptypb.Empty{}, nil
 		}
 		err := status.Errorf(codes.NotFound, "unable to find key %s", in.Name)
 		log.Printf("error: %v", err)
 		return nil, err
 	}
+
+	err := s.registry.Delete(in.Name)
+	if err != nil {
+		return nil, status.Error(codes.Unknown, err.Error())
+	}
 	params := spdk.BdevNullDeleteParams{
 		Name: in.Name,
 	}
 	var result spdk.BdevNullDeleteResult
-	err := s.rpc.Call("bdev_null_delete", &params, &result)
+	err = s.rpc.Call("bdev_null_delete", &params, &result)
 	if err != nil {
 		log.Printf("error: %v", err)
 		return nil, err
 	}
 	log.Printf("Received from SPDK: %v", result)
 	if !result {
-		msg := fmt.Sprintf("Could not delete Null Dev: %s", volume.Handle.Value)
+		msg := fmt.Sprintf("Could not delete Null Dev: %s", in.Name)
 		log.Print(msg)
 		return nil, status.Errorf(codes.InvalidArgument, msg)
 	}
-	delete(s.Volumes.NullVolumes, volume.Handle.Value)
 	return &emptypb.Empty{}, nil
 }
 

pkg/middleend/encryption.go

@@ -30,6 +30,24 @@ func (s *Server) CreateEncryptedVolume(_ context.Context, in *pb.CreateEncrypted
 		return nil, status.Error(codes.InvalidArgument, err.Error())
 	}
 
+	encrVol := s.registry.Find(in.EncryptedVolume.EncryptedVolumeId.Value)
+	if encrVol != nil {
+		log.Printf("Already existing encrypted volume with id %v", in.EncryptedVolume.EncryptedVolumeId.Value)
+		if encr := encrVol.Descriptor().ToEncryptedVolume(); encr != nil {
+			return encr, nil
+		} else {
+			// panic?
+		}
+	}
+	vol := s.registry.Find(in.EncryptedVolume.VolumeId.Value)
+	if vol == nil {
+		return nil, status.Errorf(codes.NotFound, "Underlying volume %v not found", in.EncryptedVolume.VolumeId.Value)
+	}
+	encrVol, err := vol.CreateEncryptedVolume(in.EncryptedVolume)
+	if err != nil {
+		return nil, status.Errorf(codes.InvalidArgument, err.Error())
+	}
+
 	// first create a key
 	params1 := s.getAccelCryptoKeyCreateParams(in.EncryptedVolume)
 	var result1 spdk.AccelCryptoKeyCreateResult
@@ -46,12 +64,12 @@ func (s *Server) CreateEncryptedVolume(_ context.Context, in *pb.CreateEncrypted
 	}
 	// create bdev now
 	params := spdk.BdevCryptoCreateParams{
-		Name:         in.EncryptedVolume.EncryptedVolumeId.Value,
-		BaseBdevName: in.EncryptedVolume.VolumeId.Value,
+		Name:         encrVol.BdevName(),
+		BaseBdevName: vol.BdevName(),
 		KeyName:      in.EncryptedVolume.EncryptedVolumeId.Value,
 	}
 	var result spdk.BdevCryptoCreateResult
-	err := s.rpc.Call("bdev_crypto_create", &params, &result)
+	err = s.rpc.Call("bdev_crypto_create", &params, &result)
 	if err != nil {
 		log.Printf("error: %v", err)
 		return nil, err
@@ -68,17 +86,34 @@ func (s *Server) CreateEncryptedVolume(_ context.Context, in *pb.CreateEncrypted
 		log.Printf("error: %v", err)
 		return nil, err
 	}
+
+	s.registry.Add(in.EncryptedVolume.EncryptedVolumeId.Value, encrVol)
 	return response, nil
 }
 
 // DeleteEncryptedVolume deletes an encrypted volume
 func (s *Server) DeleteEncryptedVolume(_ context.Context, in *pb.DeleteEncryptedVolumeRequest) (*emptypb.Empty, error) {
 	log.Printf("DeleteEncryptedVolume: Received from client: %v", in)
+
+	vol := s.registry.Find(in.Name)
+	if vol == nil {
+		if in.AllowMissing {
+			return &emptypb.Empty{}, nil
+		}
+		err := status.Errorf(codes.NotFound, "unable to find key %s", in.Name)
+		log.Printf("error: %v", err)
+		return nil, err
+	}
+
+	err := s.registry.Delete(in.Name)
+	if err != nil {
+		return nil, status.Error(codes.Unknown, err.Error())
+	}
 	bdevCryptoDeleteParams := spdk.BdevCryptoDeleteParams{
-		Name: in.Name,
+		Name: vol.BdevName(),
 	}
 	var bdevCryptoDeleteResult spdk.BdevCryptoDeleteResult
-	err := s.rpc.Call("bdev_crypto_delete", &bdevCryptoDeleteParams, &bdevCryptoDeleteResult)
+	err = s.rpc.Call("bdev_crypto_delete", &bdevCryptoDeleteParams, &bdevCryptoDeleteResult)
 	if err != nil {
 		log.Printf("error: %v", err)
 		return nil, err
@@ -91,7 +126,7 @@ func (s *Server) DeleteEncryptedVolume(_ context.Context, in *pb.DeleteEncrypted
 	}
 
 	keyDestroyParams := spdk.AccelCryptoKeyDestroyParams{
-		KeyName: in.Name,
+		KeyName: vol.BdevName(),
 	}
 	var keyDestroyResult spdk.AccelCryptoKeyDestroyResult
 	err = s.rpc.Call("accel_crypto_key_destroy", &keyDestroyParams, &keyDestroyResult)

pkg/middleend/middleend.go

@@ -8,6 +8,7 @@ package middleend
 import (
 	"github.com/opiproject/gospdk/spdk"
 	pb "github.com/opiproject/opi-api/storage/v1alpha1/gen/go"
+	"github.com/opiproject/opi-spdk-bridge/pkg/volume"
 )
 
 // VolumeParameters contains MiddleEnd volume related structures
@@ -23,16 +24,18 @@ type Server struct {
 	rpc        spdk.JSONRPC
 	volumes    VolumeParameters
 	Pagination map[string]int
+	registry   *volume.Registry
 }
 
 // NewServer creates initialized instance of MiddleEnd server communicating
 // with provided jsonRPC
-func NewServer(jsonRPC spdk.JSONRPC) *Server {
+func NewServer(jsonRPC spdk.JSONRPC, registry *volume.Registry) *Server {
 	return &Server{
 		rpc: jsonRPC,
 		volumes: VolumeParameters{
 			qosVolumes: make(map[string]*pb.QosVolume),
 		},
 		Pagination: make(map[string]int),
+		registry:   registry,
 	}
 }

pkg/middleend/qos.go

@@ -13,7 +13,6 @@ import (
 	pb "github.com/opiproject/opi-api/storage/v1alpha1/gen/go"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
-	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/emptypb"
 )
 
@@ -24,20 +23,33 @@ func (s *Server) CreateQosVolume(_ context.Context, in *pb.CreateQosVolumeReques
 		log.Println("error:", err)
 		return nil, status.Error(codes.InvalidArgument, err.Error())
 	}
-	if volume, ok := s.volumes.qosVolumes[in.QosVolume.QosVolumeId.Value]; ok {
+	qosVol := s.registry.Find(in.QosVolume.QosVolumeId.Value)
+	if qosVol != nil {
 		log.Printf("Already existing QoS volume with id %v", in.QosVolume.QosVolumeId.Value)
-		return volume, nil
+		if qos := qosVol.Descriptor().ToQosVolume(); qos != nil {
+			return qos, nil
+		} else {
+			// panic?
+		}
+	}
+	vol := s.registry.Find(in.QosVolume.VolumeId.Value)
+	if vol == nil {
+		return nil, status.Errorf(codes.NotFound, "Underlying volume %v not found", in.QosVolume.VolumeId.Value)
+	}
+	qosVol, err := vol.CreateQosVolume(in.QosVolume)
+	if err != nil {
+		return nil, status.Errorf(codes.InvalidArgument, err.Error())
 	}
 
 	params := spdk.BdevQoSParams{
-		Name:           in.QosVolume.VolumeId.Value,
+		Name:           qosVol.BdevName(),
 		RwIosPerSec:    int(in.QosVolume.LimitMax.RwIopsKiops * 1000),
 		RwMbytesPerSec: int(in.QosVolume.LimitMax.RwBandwidthMbs),
 		RMbytesPerSec:  int(in.QosVolume.LimitMax.RdBandwidthMbs),
 		WMbytesPerSec:  int(in.QosVolume.LimitMax.RdBandwidthMbs),
 	}
 	var result spdk.BdevQoSResult
-	err := s.rpc.Call("bdev_set_qos_limit", &params, &result)
+	err = s.rpc.Call("bdev_set_qos_limit", &params, &result)
 	if err != nil {
 		log.Printf("error: %v", err)
 		return nil, spdk.ErrFailedSpdkCall
@@ -49,31 +61,36 @@ func (s *Server) CreateQosVolume(_ context.Context, in *pb.CreateQosVolumeReques
 		return nil, spdk.ErrUnexpectedSpdkCallResult
 	}
 
-	s.volumes.qosVolumes[in.QosVolume.QosVolumeId.Value] = proto.Clone(in.QosVolume).(*pb.QosVolume)
+	s.registry.Add(in.QosVolume.QosVolumeId.Value, qosVol)
+	// s.volumes.qosVolumes[in.QosVolume.QosVolumeId.Value] = proto.Clone(in.QosVolume).(*pb.QosVolume)
 	return in.QosVolume, nil
 }
 
 // DeleteQosVolume creates a QoS volume
 func (s *Server) DeleteQosVolume(_ context.Context, in *pb.DeleteQosVolumeRequest) (*emptypb.Empty, error) {
 	log.Printf("CreateQosVolume: Received from client: %v", in)
-	qosVolume, ok := s.volumes.qosVolumes[in.Name]
-	if !ok {
+	vol := s.registry.Find(in.Name)
+	if vol == nil {
 		if in.AllowMissing {
 			return &emptypb.Empty{}, nil
 		}
 		err := status.Errorf(codes.NotFound, "unable to find key %s", in.Name)
 		log.Printf("error: %v", err)
 		return nil, err
 	}
+	err := s.registry.Delete(in.Name)
+	if err != nil {
+		return nil, status.Error(codes.Unknown, err.Error())
+	}
 	params := spdk.BdevQoSParams{
-		Name:           qosVolume.VolumeId.Value,
+		Name:           vol.BdevName(),
 		RwIosPerSec:    0,
 		RwMbytesPerSec: 0,
 		RMbytesPerSec:  0,
 		WMbytesPerSec:  0,
 	}
 	var result spdk.BdevQoSResult
-	err := s.rpc.Call("bdev_set_qos_limit", &params, &result)
+	err = s.rpc.Call("bdev_set_qos_limit", &params, &result)
 	if err != nil {
 		log.Printf("error: %v", err)
 		return nil, spdk.ErrFailedSpdkCall
@@ -85,7 +102,7 @@ func (s *Server) DeleteQosVolume(_ context.Context, in *pb.DeleteQosVolumeReques
 		return nil, spdk.ErrUnexpectedSpdkCallResult
 	}
 
-	delete(s.volumes.qosVolumes, in.Name)
+	// delete(s.volumes.qosVolumes, in.Name)
 	return &emptypb.Empty{}, nil
 }
 

pkg/volume/descriptor.go

@@ -0,0 +1,37 @@
+// SPDX-License-Identifier: Apache-2.0
+// Copyright (C) 2023 Intel Corporation
+
+// Package volume contains volume layer abstractions
+package volume
+
+import pb "github.com/opiproject/opi-api/storage/v1alpha1/gen/go"
+
+type Descriptor struct {
+	value any
+}
+
+// return copies??
+
+func (d Descriptor) ToNullDebug() *pb.NullDebug {
+	val, ok := d.value.(*pb.NullDebug)
+	if ok {
+		return val
+	}
+	return nil
+}
+
+func (d Descriptor) ToQosVolume() *pb.QosVolume {
+	val, ok := d.value.(*pb.QosVolume)
+	if ok {
+		return val
+	}
+	return nil
+}
+
+func (d Descriptor) ToEncryptedVolume() *pb.EncryptedVolume {
+	val, ok := d.value.(*pb.EncryptedVolume)
+	if ok {
+		return val
+	}
+	return nil
+}