feat(rest): expose json rpc methods

Mayastor already exposes SPDK JSON RPC methods through its gRPC
interface. This gRPC interface is further exposed up through the REST
API. Now generic JSON gRPC methods can be invoked using a REST call.

Author: Paul Yoong

mbus-api/src/message_bus/v0.rs

@@ -240,6 +240,12 @@ pub trait MessageBusTrait: Sized {
         request.request().await?;
         Ok(())
     }
+
+    /// Generic JSON gRPC call
+    #[tracing::instrument(level = "debug", err)]
+    async fn json_grpc_call(request: JsonGrpcRequest) -> BusResult<String> {
+        Ok(request.request().await?)
+    }
 }
 
 /// Implementation of the bus interface trait

mbus-api/src/send.rs

@@ -2,7 +2,7 @@ use super::*;
 
 // todo: replace with proc-macros
 
-/// Main Message trait, which should tipically be used to send
+/// Main Message trait, which should typically be used to send
 /// MessageBus messages.
 /// Implements Message trait for the type `S` with the reply type
 /// `R`, the message id `I`, the default channel `C`.

mbus-api/src/v0.rs

@@ -26,6 +26,8 @@ pub enum ChannelVs {
     Nexus,
     /// Keep it In Sync Service
     Kiiss,
+    /// Json gRPC Service
+    JsonGrpc,
 }
 impl Default for ChannelVs {
     fn default() -> Self {
@@ -102,6 +104,8 @@ pub enum MessageIdVs {
     AddVolumeNexus,
     /// Remove nexus from volume
     RemoveVolumeNexus,
+    /// Generic JSON gRPC message
+    JsonGrpc,
 }
 
 // Only V0 should export this macro
@@ -367,6 +371,11 @@ bus_impl_string_id!(ReplicaId, "UUID of a mayastor pool replica");
 bus_impl_string_id!(NexusId, "UUID of a mayastor nexus");
 bus_impl_string_id_percent_decoding!(ChildUri, "URI of a mayastor nexus child");
 bus_impl_string_id!(VolumeId, "UUID of a mayastor volume");
+bus_impl_string_id!(JsonGrpcMethod, "JSON gRPC method");
+bus_impl_string_id!(
+    JsonGrpcParams,
+    "Parameters to be passed to a JSON gRPC method"
+);
 
 /// Pool Service
 /// Get all the pools from specific node or None for all nodes
@@ -953,3 +962,16 @@ pub struct RemoveVolumeNexus {
     pub node: Option<NodeId>,
 }
 bus_impl_message_all!(RemoveVolumeNexus, RemoveVolumeNexus, (), Volume);
+
+/// Generic JSON gRPC request
+#[derive(Serialize, Deserialize, Default, Debug, Clone)]
+#[serde(rename_all = "camelCase")]
+pub struct JsonGrpcRequest {
+    /// id of the mayastor instance
+    pub node: NodeId,
+    /// JSON gRPC method to call
+    pub method: JsonGrpcMethod,
+    /// parameters to be passed to the above method
+    pub params: JsonGrpcParams,
+}
+bus_impl_message_all!(JsonGrpcRequest, JsonGrpc, String, JsonGrpc);

rest/service/src/v0/jsongrpc.rs

@@ -0,0 +1,35 @@
+//! Provides a REST interface to interact with JSON gRPC methods.
+//! These methods are typically used to control SPDK directly.
+
+use super::*;
+use mbus_api::v0::JsonGrpcRequest;
+
+/// Configure the functions that this service supports.
+pub(crate) fn configure(cfg: &mut paperclip::actix::web::ServiceConfig) {
+    cfg.service(json_grpc_call);
+}
+
+// A PUT request is required so that method parameters can be passed in the
+// body.
+//
+// # Example
+// To create a malloc bdev:
+// ```
+//  curl -X PUT "https://localhost:8080/v0/nodes/mayastor/jsongrpc/bdev_malloc_create" \
+//  -H "accept: application/json" -H "Content-Type: application/json" \
+//  -d '{"block_size": 512, "num_blocks": 64, "name": "Malloc0"}'
+// ```
+#[put("/v0", "/nodes/{node}/jsongrpc/{method}", tags(JsonGrpc))]
+async fn json_grpc_call(
+    web::Path((node, method)): web::Path<(NodeId, JsonGrpcMethod)>,
+    body: web::Json<serde_json::Value>,
+) -> Result<Json<String>, RestError> {
+    RestRespond::result(
+        MessageBus::json_grpc_call(JsonGrpcRequest {
+            node,
+            method,
+            params: body.into_inner().to_string().into(),
+        })
+        .await,
+    )
+}

rest/service/src/v0/mod.rs

@@ -3,6 +3,7 @@
 //! Ex: /v0/nodes
 
 pub mod children;
+pub mod jsongrpc;
 pub mod nexuses;
 pub mod nodes;
 pub mod pools;
@@ -45,6 +46,7 @@ fn configure(cfg: &mut paperclip::actix::web::ServiceConfig) {
     nexuses::configure(cfg);
     children::configure(cfg);
     volumes::configure(cfg);
+    jsongrpc::configure(cfg);
 }
 
 pub(super) fn configure_api<T, B>(

rest/src/lib.rs

@@ -35,7 +35,7 @@ impl ActixRestClient {
 
         match url.scheme() {
             "https" => Self::new_https(&url, trace),
-            "http" => Self::new_http(&url, trace),
+            "http" => Ok(Self::new_http(&url, trace)),
             invalid => {
                 let msg = format!("Invalid url scheme: {}", invalid);
                 Err(anyhow::Error::msg(msg))
@@ -65,12 +65,12 @@ impl ActixRestClient {
         })
     }
     /// creates a new client
-    fn new_http(url: &url::Url, trace: bool) -> anyhow::Result<Self> {
-        Ok(Self {
+    fn new_http(url: &url::Url, trace: bool) -> Self {
+        Self {
             client: Client::new(),
             url: url.to_string().trim_end_matches('/').into(),
             trace,
-        })
+        }
     }
     async fn get_vec<R>(&self, urn: String) -> anyhow::Result<Vec<R>>
     where

services/Cargo.toml

@@ -24,6 +24,11 @@ path = "volume/src/server.rs"
 name = "deployer"
 path = "deployer/src/bin.rs"
 
+[[bin]]
+name = "jsongrpc"
+path = "jsongrpc/src/server.rs"
+
+
 [lib]
 name = "common"
 path = "common/src/lib.rs"

services/common/src/wrapper/v0/mod.rs

@@ -29,6 +29,7 @@ use tonic::transport::Channel;
 
 /// Common error type for send/receive
 #[derive(Debug, Snafu)]
+#[snafu(visibility = "pub")]
 #[allow(missing_docs)]
 pub enum SvcError {
     #[snafu(display("Failed to get nodes from the node service"))]
@@ -85,6 +86,17 @@ pub enum SvcError {
     InvalidArguments {},
     #[snafu(display("Not implemented"))]
     NotImplemented {},
+    #[snafu(display(
+        "Json RPC call failed for method '{}' with parameters '{}'. Error {}",
+        method,
+        params,
+        error,
+    ))]
+    JsonRpc {
+        method: String,
+        params: String,
+        error: String,
+    },
 }
 
 impl From<NotEnough> for SvcError {

services/jsongrpc/src/server.rs

@@ -0,0 +1,79 @@
+pub mod service;
+
+use async_trait::async_trait;
+use common::*;
+use mbus_api::{v0::*, *};
+use service::*;
+use std::{convert::TryInto, marker::PhantomData};
+use structopt::StructOpt;
+use tracing::info;
+
+#[derive(Debug, StructOpt)]
+struct CliArgs {
+    /// The Nats Server URL to connect to
+    /// (supports the nats schema)
+    /// Default: nats://127.0.0.1:4222
+    #[structopt(long, short, default_value = "nats://127.0.0.1:4222")]
+    nats: String,
+}
+
+/// Needed so we can implement the ServiceSubscriber trait for
+/// the message types external to the crate
+#[derive(Clone, Default)]
+struct ServiceHandler<T> {
+    data: PhantomData<T>,
+}
+
+macro_rules! impl_service_handler {
+    // RequestType is the message bus request type
+    // ServiceFnName is the name of the service function to route the request
+    // into
+    ($RequestType:ident, $ServiceFnName:ident) => {
+        #[async_trait]
+        impl ServiceSubscriber for ServiceHandler<$RequestType> {
+            async fn handler(&self, args: Arguments<'_>) -> Result<(), Error> {
+                let request: ReceivedMessage<$RequestType> =
+                    args.request.try_into()?;
+
+                let reply = JsonGrpcSvc::$ServiceFnName(&request.inner())
+                    .await
+                    .map_err(|error| Error::ServiceError {
+                        message: error.full_string(),
+                    })?;
+                request.reply(reply).await
+            }
+            fn filter(&self) -> Vec<MessageId> {
+                vec![$RequestType::default().id()]
+            }
+        }
+    };
+}
+
+impl_service_handler!(JsonGrpcRequest, json_grpc_call);
+
+fn init_tracing() {
+    if let Ok(filter) = tracing_subscriber::EnvFilter::try_from_default_env() {
+        tracing_subscriber::fmt().with_env_filter(filter).init();
+    } else {
+        tracing_subscriber::fmt().with_env_filter("info").init();
+    }
+}
+
+#[tokio::main]
+async fn main() {
+    init_tracing();
+
+    let cli_args = CliArgs::from_args();
+    info!("Using options: {:?}", &cli_args);
+
+    server(cli_args).await;
+}
+
+async fn server(cli_args: CliArgs) {
+    Service::builder(cli_args.nats, ChannelVs::JsonGrpc)
+        .connect()
+        .await
+        .with_subscription(ServiceHandler::<JsonGrpcRequest>::default())
+        .run()
+        .await;
+}

services/jsongrpc/src/service.rs

@@ -0,0 +1,44 @@
+// clippy warning caused by the instrument macro
+#![allow(clippy::unit_arg)]
+
+use ::rpc::mayastor::{JsonRpcReply, JsonRpcRequest};
+use common::wrapper::v0::{BusGetNode, SvcError};
+use mbus_api::message_bus::v0::{MessageBus, *};
+use rpc::mayastor::json_rpc_client::JsonRpcClient;
+use snafu::ResultExt;
+
+#[derive(Clone, Default)]
+pub(super) struct JsonGrpcSvc {}
+
+/// JSON gRPC service implementation
+impl JsonGrpcSvc {
+    /// Generic JSON gRPC call issued to Mayastor using the JsonRpcClient.
+    pub(super) async fn json_grpc_call(
+        request: &JsonGrpcRequest,
+    ) -> Result<String, SvcError> {
+        let node =
+            MessageBus::get_node(&request.node)
+                .await
+                .context(BusGetNode {
+                    node: request.node.clone(),
+                })?;
+        let mut client =
+            JsonRpcClient::connect(format!("http://{}", node.grpc_endpoint))
+                .await
+                .unwrap();
+        let response: JsonRpcReply = client
+            .json_rpc_call(JsonRpcRequest {
+                method: request.method.to_string(),
+                params: request.params.to_string(),
+            })
+            .await
+            .map_err(|error| SvcError::JsonRpc {
+                method: request.method.to_string(),
+                params: request.params.to_string(),
+                error: error.to_string(),
+            })?
+            .into_inner();
+
+        Ok(response.result)
+    }
+}