sma: initial Storage Management Agent structures

Storage Management Agent is a gRPC server that provides an abstraction
layer above the SPDK RPC interface.  The interface aims to provide a set
of methods for managing various protocols (e.g. NVMe, virtio-blk) while
hiding the details of a particular transport.

The external API is defined by `lib/python/spdk/sma/proto/sma.proto`
protobuf file.  It defines the generic gRPC service methods and their
requests/responses.  Device-specific messages are defined in their own
files.  This patch also defines messages for creating NVMe and NVMe/TCP
devices.

This patch implements a gRPC service that delegates the work to a
specific device type.  A DeviceManager is a class that implements some
of the methods defined by the service for a given type of devices (e.g.
NVMe, virtio-blk, NVMe/TCP, etc.).  For now, only the RPC for creating a
device is implemented, others are added in subsequent patches.

The series implements the generic calls as well as their NVMe/TCP
implementation.  Support for other devce types could be easily added by
creating a new device manager and defining its protobuf parameter
definition.

Signed-off-by: Konrad Sztyber <konrad.sztyber@intel.com>
Change-Id: I17cde3b31d3514878f1027cfcd112b48848f6123
Reviewed-on: https://review.spdk.io/gerrit/c/spdk/spdk/+/10273
Community-CI: Broadcom CI <spdk-ci.pdl@broadcom.com>
Community-CI: Mellanox Build Bot
Tested-by: SPDK CI Jenkins <sys_sgci@intel.com>
Reviewed-by: Ben Walker <benjamin.walker@intel.com>
Reviewed-by: Jim Harris <james.r.harris@intel.com>

Makefile

@@ -47,6 +47,7 @@ DIRS-y += test
 DIRS-$(CONFIG_IPSEC_MB) += ipsecbuild
 DIRS-$(CONFIG_ISAL) += isalbuild
 DIRS-$(CONFIG_VFIO_USER) += vfiouserbuild
+DIRS-y += python
 
 .PHONY: all clean $(DIRS-y) include/spdk/config.h mk/config.mk \
 	cc_version cxx_version .libs_only_other .ldflags ldflags install \

python/Makefile

@@ -0,0 +1,62 @@
+#
+#  BSD LICENSE
+#
+#  Copyright (c) Intel Corporation.
+#  All rights reserved.
+#
+#  Redistribution and use in source and binary forms, with or without
+#  modification, are permitted provided that the following conditions
+#  are met:
+#
+#    * Redistributions of source code must retain the above copyright
+#      notice, this list of conditions and the following disclaimer.
+#    * Redistributions in binary form must reproduce the above copyright
+#      notice, this list of conditions and the following disclaimer in
+#      the documentation and/or other materials provided with the
+#      distribution.
+#    * Neither the name of Intel Corporation nor the names of its
+#      contributors may be used to endorse or promote products derived
+#      from this software without specific prior written permission.
+#
+#  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+#  "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+#  LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+#  A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+#  OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+#  SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+#  LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+#  DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+#  THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+#  (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+#  OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+#
+
+SPDK_ROOT_DIR := $(abspath $(CURDIR)/..)
+include $(SPDK_ROOT_DIR)/mk/spdk.common.mk
+
+define generate_protobuf
+$(1)/%_pb2.py $(1)/%_pb2_grpc.py: $(1)/%.proto
+	$(Q)python3 -m grpc_tools.protoc -I $(1) --python_out=$(1) \
+		--grpc_python_out=$(1) --proto_path=$(1) $$^
+endef
+
+# Check for the required modules here until there's a configure option for SMA
+ifeq ($(shell python3 -c 'import grpc; import grpc_tools' && echo 0),0)
+protodirs = spdk/sma/proto
+protodefs = $(foreach protodir,$(protodirs),$(wildcard $(protodir)/*.proto))
+protopy = $(foreach proto,$(basename $(protodefs)),$(addprefix $(proto),_pb2.py _pb2_grpc.py))
+endif
+
+all: $(protopy)
+
+clean:
+	$(Q)$(RM) $(protopy)
+
+# TODO: we should probably write a proper install rule here instead of just blindly copying all
+# python scripts when building the RPMs
+install:
+uninstall:
+
+$(foreach protodir,$(protodirs),$(eval $(call generate_protobuf,$(protodir))))
+
+.PHONY: all clean install uninstall

python/spdk/sma/__init__.py

@@ -0,0 +1,10 @@
+import os
+import sys
+
+# Fix up the import paths for the autogenerated files
+sys.path.append(os.path.dirname(__file__) + '/proto')
+
+from .sma import StorageManagementAgent     # noqa
+from .device import DeviceException         # noqa
+from .device import DeviceManager           # noqa
+from .device import NvmfTcpDeviceManager    # noqa

python/spdk/sma/common.py

@@ -0,0 +1,17 @@
+import uuid
+
+
+def format_volume_id(volume_id):
+    """Verifies volume_id and returns it as a str
+
+    Args:
+        volume_id: either a str (in which case it's only validated) or bytes object
+    """
+    try:
+        if type(volume_id) is bytes:
+            return str(uuid.UUID(bytes=volume_id))
+        elif type(volume_id) is str:
+            return str(uuid.UUID(hex=volume_id))
+    except ValueError:
+        pass
+    return None

python/spdk/sma/device/__init__.py

@@ -0,0 +1,3 @@
+from .device import DeviceException
+from .device import DeviceManager
+from .nvmf_tcp import NvmfTcpDeviceManager

python/spdk/sma/device/device.py

@@ -0,0 +1,31 @@
+from ..proto import sma_pb2
+
+
+class DeviceException(Exception):
+    def __init__(self, code, message):
+        self.code = code
+        self.message = message
+
+
+class DeviceManager:
+    def __init__(self, name, client):
+        self._client = client
+        self.name = name
+
+    def init(self, config):
+        pass
+
+    def create_device(self, request):
+        raise NotImplementedError()
+
+    def delete_device(self, request):
+        raise NotImplementedError()
+
+    def attach_volume(self, request):
+        raise NotImplementedError()
+
+    def detach_volume(self, request):
+        raise NotImplementedError()
+
+    def owns_device(self, id):
+        raise NotImplementedError()

python/spdk/sma/device/nvmf_tcp.py

@@ -0,0 +1,106 @@
+import grpc
+import logging
+from spdk.rpc.client import JSONRPCException
+from .device import DeviceManager, DeviceException
+from ..common import format_volume_id
+from ..proto import sma_pb2
+from ..proto import nvmf_tcp_pb2
+
+
+class NvmfTcpDeviceManager(DeviceManager):
+    def __init__(self, client):
+        super().__init__('nvmf_tcp', client)
+
+    def init(self, config):
+        self._has_transport = self._create_transport()
+
+    def _create_transport(self):
+        try:
+            with self._client() as client:
+                transports = client.call('nvmf_get_transports')
+                for transport in transports:
+                    if transport['trtype'].lower() == 'tcp':
+                        return True
+                # TODO: take the transport params from config
+                return client.call('nvmf_create_transport',
+                                   {'trtype': 'tcp'})
+        except JSONRPCException:
+            logging.error('Failed to query for NVMe/TCP transport')
+            return False
+
+    def _check_transport(f):
+        def wrapper(self, *args):
+            if not self._has_transport:
+                raise DeviceException(grpc.StatusCode.INTERNAL,
+                                      'NVMe/TCP transport is unavailable')
+            return f(self, *args)
+        return wrapper
+
+    def _get_params(self, request, params):
+        result = {}
+        for grpc_param, *rpc_param in params:
+            rpc_param = rpc_param[0] if rpc_param else grpc_param
+            result[rpc_param] = getattr(request, grpc_param)
+        return result
+
+    def _check_addr(self, addr, addrlist):
+        return next(filter(lambda a: (
+            a['trtype'].lower() == 'tcp' and
+            a['adrfam'].lower() == addr['adrfam'].lower() and
+            a['traddr'].lower() == addr['traddr'].lower() and
+            a['trsvcid'].lower() == addr['trsvcid'].lower()), addrlist), None) is not None
+
+    @_check_transport
+    def create_device(self, request):
+        params = request.nvmf_tcp
+        with self._client() as client:
+            try:
+                subsystems = client.call('nvmf_get_subsystems')
+                for subsystem in subsystems:
+                    if subsystem['nqn'] == params.subnqn:
+                        break
+                else:
+                    subsystem = None
+                    result = client.call('nvmf_create_subsystem',
+                                         {**self._get_params(params, [
+                                                ('subnqn', 'nqn'),
+                                                ('allow_any_host',)])})
+            except JSONRPCException:
+                raise DeviceException(grpc.StatusCode.INTERNAL,
+                                      'Failed to create NVMe/TCP device')
+            try:
+                for host in params.hosts:
+                    client.call('nvmf_subsystem_add_host',
+                                {'nqn': params.subnqn,
+                                 'host': host})
+                if subsystem is not None:
+                    for host in [h['nqn'] for h in subsystem['hosts']]:
+                        if host not in params.hosts:
+                            client.call('nvmf_subsystem_remove_host',
+                                        {'nqn': params.subnqn,
+                                         'host': host})
+
+                addr = self._get_params(params, [
+                                ('adrfam',),
+                                ('traddr',),
+                                ('trsvcid',)])
+                if subsystem is None or not self._check_addr(addr,
+                                                             subsystem['listen_addresses']):
+                    client.call('nvmf_subsystem_add_listener',
+                                {'nqn': params.subnqn,
+                                 'listen_address': {'trtype': 'tcp', **addr}})
+                volume_id = format_volume_id(request.volume.volume_id)
+                if volume_id is not None:
+                    result = client.call('nvmf_subsystem_add_ns',
+                                         {'nqn': params.subnqn,
+                                          'namespace': {
+                                              'bdev_name': volume_id}})
+            except JSONRPCException:
+                try:
+                    client.call('nvmf_delete_subsystem', {'nqn': params.subnqn})
+                except JSONRPCException:
+                    logging.warning(f'Failed to delete subsystem: {params.subnqn}')
+                raise DeviceException(grpc.StatusCode.INTERNAL,
+                                      'Failed to create NVMe/TCP device')
+
+        return sma_pb2.CreateDeviceResponse(handle=f'nvmf-tcp:{params.subnqn}')

python/spdk/sma/proto/.gitignore

@@ -0,0 +1,2 @@
+!__init__.py
+*.py

python/spdk/sma/proto/nvme.proto

@@ -0,0 +1,10 @@
+syntax = "proto3";
+
+package sma.nvme;
+
+message DeviceParameters {
+  // Physical function index
+  uint32 physical_id = 1;
+  // Virtual function index
+  uint32 virtual_id = 2;
+}

python/spdk/sma/proto/nvmf.proto

@@ -0,0 +1,35 @@
+syntax = "proto3";
+
+package sma.nvmf;
+
+option go_package = "spdk.io/sma/nvmf";
+
+// Defines an address of an NVMeoF endpoint
+message Address {
+  // Transport type ("rdma" or "tcp")
+  string trtype = 1;
+  // Transport address (IP)
+  string traddr = 2;
+  // Transport service identifier (port number)
+  string trsvcid = 3;
+}
+// NVMeoF connection using discovery service
+message VolumeDiscoveryParameters {
+  // One or more discovery endpoints
+  repeated Address discovery_endpoints = 1;
+}
+
+// Describes connection parameters for an NVMeoF volume (namespace)
+message VolumeConnectionParameters {
+  // Subsystem that the volume is exposed through.  A volume with a given
+  // GUID/UUID won't be created if it's attached to a different subsystem.  This
+  // field is optional and can be left empty.
+  string subnqn = 1;
+  // Host NQN to use when connecting to the subsystem exposing the volume (and,
+  // if using discovery, to the discovery subsystem too).
+  string hostnqn = 2;
+  oneof connection_params {
+    // Connection through discovery service
+    VolumeDiscoveryParameters discovery = 3;
+  }
+}

python/spdk/sma/proto/nvmf_tcp.proto

@@ -0,0 +1,20 @@
+syntax = "proto3";
+
+package sma.nvmf_tcp;
+
+// Create device NVMe/TCP-specific parameters
+message DeviceParameters {
+  // Subsystem NQN
+  string subnqn = 1;
+  // Address family ("ipv4", "ipv6")
+  string adrfam = 2;
+  // Transport address
+  string traddr = 3;
+  // Transport service ID (port number)
+  string trsvcid = 4;
+  // Allow any host to connect
+  bool allow_any_host = 5;
+  // List of host NQNs that are allowed to connect to the subsystem (if
+  // allow_any_host is false)
+  repeated string hosts = 6;
+}

python/spdk/sma/proto/sma.proto

@@ -0,0 +1,98 @@
+syntax = "proto3";
+
+import "nvme.proto";
+import "virtio_blk.proto";
+import "nvmf_tcp.proto";
+import "nvmf.proto";
+
+// This file provides the generic definitions for the Storage Management Agent
+// gRPC calls.  All of the methods are supposed to be idempotent.  Errors are
+// reported as standard gRPC status codes.
+
+package sma;
+
+// Parameters describing a volume
+message VolumeParameters {
+  // Volume GUID/UUID
+  bytes volume_id = 1;
+  oneof connection_params {
+    // NVMeoF volume
+    nvmf.VolumeConnectionParameters nvmf = 2;
+  }
+}
+
+// Create device request
+message CreateDeviceRequest {
+  // Volume to immediately attach to the created device.  This field may be
+  // optional for some device types (e.g. NVMe), while it may be required for
+  // others (e.g. virtio-blk).
+  VolumeParameters volume = 1;
+  // Device-specific parameters
+  oneof params {
+    // NVMe parameters
+    nvme.DeviceParameters nvme = 2;
+    // Virtio-blk parameters
+    virtio_blk.DeviceParameters virtio_blk = 3;
+    // NVMe/TCP parameters
+    nvmf_tcp.DeviceParameters nvmf_tcp = 4;
+  }
+}
+
+// Create device response
+message CreateDeviceResponse {
+  // Device handle that can uniquely identify a device within an instance of
+  // Storage Management Agent
+  string handle = 1;
+}
+
+// Delete device request
+message DeleteDeviceRequest {
+  // Device handle
+  string handle = 1;
+}
+
+// Delete device response
+message DeleteDeviceResponse {}
+
+// Attach volume request
+message AttachVolumeRequest {
+  // Volume parameters
+  VolumeParameters volume = 1;
+  // Device handle
+  string device_handle = 2;
+}
+
+// Attach volume response
+message AttachVolumeResponse {}
+
+// Detach volume request
+message DetachVolumeRequest {
+  // Volume GUID/UUID
+  bytes volume_id = 1;
+  // Device handle
+  string device_handle = 2;
+}
+
+// Detach volume response
+message DetachVolumeResponse {}
+
+// Storage Management Agent gRPC service definition
+service StorageManagementAgent {
+  // Creates a new device.  A device is an entity that can be used to expose
+  // volumes (e.g. an NVMeoF subsystem).
+  rpc CreateDevice (CreateDeviceRequest)
+    returns (CreateDeviceResponse) {}
+  // Deletes a device
+  rpc DeleteDevice (DeleteDeviceRequest)
+    returns (DeleteDeviceResponse) {}
+  // Attaches a volume to a specified device making it available through that
+  // device (e.g. for NVMeoF this results in adding a namespace to an NVMeoF
+  // subsystem).  The type of volume doesn't need to match the type of device
+  // (e.g. it's perfectly fine to attach an NVMe/TCP volume to a virtio-blk
+  // device).
+  rpc AttachVolume (AttachVolumeRequest)
+    returns (AttachVolumeResponse) {}
+  // Detaches a volume from a device
+  rpc DetachVolume (DetachVolumeRequest)
+    returns (DetachVolumeRequest) {}
+}

python/spdk/sma/proto/virtio_blk.proto

@@ -0,0 +1,10 @@
+syntax = "proto3";
+
+package sma.virtio_blk;
+
+message DeviceParameters {
+  // Physical function index
+  uint32 physical_id = 1;
+  // Virtual function index
+  uint32 virtual_id = 2;
+}

python/spdk/sma/sma.py

@@ -0,0 +1,49 @@
+from concurrent import futures
+from contextlib import contextmanager
+from multiprocessing import Lock
+import grpc
+import logging
+from .device import DeviceException
+from .proto import sma_pb2 as pb2
+from .proto import sma_pb2_grpc as pb2_grpc
+
+
+class StorageManagementAgent(pb2_grpc.StorageManagementAgentServicer):
+    def __init__(self, addr, port):
+        self._devices = {}
+        self._server = grpc.server(futures.ThreadPoolExecutor(max_workers=1))
+        self._server.add_insecure_port(f'{addr}:{port}')
+        pb2_grpc.add_StorageManagementAgentServicer_to_server(self, self._server)
+
+    def _grpc_method(f):
+        def wrapper(self, request, context):
+            logging.debug(f'{f.__name__}\n{request}')
+            return f(self, request, context)
+        return wrapper
+
+    def register_device(self, device_manager):
+        self._devices[device_manager.name] = device_manager
+
+    def run(self):
+        self._server.start()
+        self._server.wait_for_termination()
+
+    def _find_device(self, name):
+        return self._devices.get(name)
+
+    @_grpc_method
+    def CreateDevice(self, request, context):
+        response = pb2.CreateDeviceResponse()
+        try:
+            manager = self._find_device(request.WhichOneof('params'))
+            if manager is None:
+                raise DeviceException(grpc.StatusCode.INVALID_ARGUMENT,
+                                      'Unsupported device type')
+            response = manager.create_device(request)
+        except DeviceException as ex:
+            context.set_details(ex.message)
+            context.set_code(ex.code)
+        except NotImplementedError:
+            context.set_details('Method is not implemented by selected device type')
+            context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        return response

scripts/sma-client.py

@@ -0,0 +1,49 @@
+#!/usr/bin/env python3
+
+import grpc
+import google.protobuf.json_format as json_format
+import json
+import os
+import sys
+
+sys.path.append(os.path.dirname(__file__) + '/../python')
+
+import spdk.sma.proto.sma_pb2 as sma_pb2                        # noqa
+import spdk.sma.proto.sma_pb2_grpc as sma_pb2_grpc              # noqa
+import spdk.sma.proto.nvmf_tcp_pb2 as nvmf_tcp_pb2              # noqa
+import spdk.sma.proto.nvmf_tcp_pb2_grpc as nvmf_tcp_pb2_grpc    # noqa
+
+
+class Client:
+    def __init__(self, addr, port):
+        self._service = sma_pb2.DESCRIPTOR.services_by_name['StorageManagementAgent']
+        self.addr = addr
+        self.port = port
+
+    def _get_message_type(self, descriptor):
+        return getattr(sma_pb2, descriptor.name)
+
+    def _get_method_types(self, method_name):
+        method = self._service.methods_by_name.get(method_name)
+        return (self._get_message_type(method.input_type),
+                self._get_message_type(method.output_type))
+
+    def call(self, method, params):
+        with grpc.insecure_channel(f'{self.addr}:{self.port}') as channel:
+            stub = sma_pb2_grpc.StorageManagementAgentStub(channel)
+            func = getattr(stub, method)
+            input, output = self._get_method_types(method)
+            response = func(request=json_format.ParseDict(params, input()))
+            return json_format.MessageToDict(response,
+                                             preserving_proto_field_name=True)
+
+
+def main(args):
+    client = Client('localhost', 8080)
+    request = json.loads(sys.stdin.read())
+    result = client.call(request['method'], request.get('params', {}))
+    print(json.dumps(result, indent=2))
+
+
+if __name__ == '__main__':
+    main(sys.argv[1:])

scripts/sma.py

@@ -0,0 +1,27 @@
+#!/usr/bin/env python3
+
+from argparse import ArgumentParser
+import logging
+import os
+import sys
+
+sys.path.append(os.path.dirname(__file__) + '/../python')
+
+import spdk.sma as sma                      # noqa
+from spdk.rpc.client import JSONRPCClient   # noqa
+
+
+def build_client():
+    return JSONRPCClient('/var/tmp/spdk.sock')
+
+
+def register_device(agent, device):
+    device.init(None)
+    agent.register_device(device)
+
+
+if __name__ == '__main__':
+    logging.basicConfig(level=os.environ.get('SMA_LOGLEVEL', 'WARNING').upper())
+    agent = sma.StorageManagementAgent('localhost', 8080)
+    register_device(agent, sma.NvmfTcpDeviceManager(build_client))
+    agent.run()