From 6c6243f2d0b5e28bb7fc95ae45b3bd4b05f365f8 Mon Sep 17 00:00:00 2001 From: Sandipsinh Rathod Date: Mon, 19 Feb 2024 14:02:35 +0530 Subject: [PATCH 1/2] add custom reflection server for easy debug --- Cargo.lock | 1 + Cargo.toml | 1 + build.rs | 7 +- renovate.json | 4 +- src/main.rs | 10 +- src/server.rs | 384 ++++++++++++++++++++++++++++++++++++++++++++++++++ 6 files changed, 399 insertions(+), 8 deletions(-) create mode 100644 src/server.rs diff --git a/Cargo.lock b/Cargo.lock index 55c05eb..f619742 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -821,6 +821,7 @@ dependencies = [ "prost", "prost-types", "tokio", + "tokio-stream", "tonic", "tonic-build", "tonic-reflection", diff --git a/Cargo.toml b/Cargo.toml index 5ec9096..493556b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -16,6 +16,7 @@ tower = "0.4.13" hyper-util = {version = "0.1.3",features = ["tokio"]} http-body-util = "0.1.0" anyhow = "1.0.79" +tokio-stream = "0.1.14" [build-dependencies] tonic-build = "0.10.2" diff --git a/build.rs b/build.rs index 4f4fd27..4eaccbc 100644 --- a/build.rs +++ b/build.rs @@ -3,12 +3,15 @@ use std::path::PathBuf; fn main() { let mut news = PathBuf::from(env!("CARGO_MANIFEST_DIR")); news.push("news.proto"); + tonic_build::compile_protos(news.clone()).expect("Failed to compile protos"); + news.pop(); + news.push("reflection.proto"); tonic_build::compile_protos(news).expect("Failed to compile protos"); - let out_dir = PathBuf::from(std::env::var("OUT_DIR").unwrap()); + /* let out_dir = PathBuf::from(std::env::var("OUT_DIR").unwrap()); tonic_build::configure() .file_descriptor_set_path(out_dir.join("news_descriptor.bin")) .compile(&["news.proto"], &["proto"]) - .unwrap(); + .unwrap();*/ } diff --git a/renovate.json b/renovate.json index 5db72dd..22a9943 100644 --- a/renovate.json +++ b/renovate.json @@ -1,6 +1,4 @@ { "$schema": "https://docs.renovatebot.com/renovate-schema.json", - "extends": [ - "config:recommended" - ] + "extends": ["config:recommended"] } diff --git a/src/main.rs b/src/main.rs index e2aa03f..665ed4b 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,6 +1,9 @@ +mod server; + use tonic::{transport::Server as TonicServer, Response, Status}; use crate::news::news_service_server::NewsServiceServer; +use crate::server::Builder; use anyhow::Result; use news::news_service_server::NewsService; use news::{MultipleNewsId, News, NewsId, NewsList}; @@ -9,6 +12,7 @@ use tower::make::Shared; pub mod news { tonic::include_proto!("news"); // The package name specified in your .proto + // tonic::include_proto!("grpc.reflection.v1alpha"); // The package name specified in your .proto pub(crate) const FILE_DESCRIPTOR_SET: &[u8] = tonic::include_file_descriptor_set!("news_descriptor"); } @@ -152,10 +156,10 @@ async fn main() -> Result<()> { let addr = ([127, 0, 0, 1], 50051).into(); let news_service = MyNewsService::new(); - let service = tonic_reflection::server::Builder::configure() + + let service = Builder::configure() .register_encoded_file_descriptor_set(news::FILE_DESCRIPTOR_SET) - .build() - .unwrap(); + .build()?; println!("NewsService server listening on {}", addr); diff --git a/src/server.rs b/src/server.rs new file mode 100644 index 0000000..be8d7ff --- /dev/null +++ b/src/server.rs @@ -0,0 +1,384 @@ +#![allow(unused)] + +use prost::{DecodeError, Message}; +use prost_types::{ + DescriptorProto, EnumDescriptorProto, FieldDescriptorProto, FileDescriptorProto, + FileDescriptorSet, +}; +use std::collections::HashMap; +use std::fmt::{Display, Formatter}; +use std::sync::Arc; +use tokio::sync::mpsc; +use tokio_stream::{wrappers::ReceiverStream, StreamExt}; +use tonic::{Request, Response, Status, Streaming}; +use tonic_reflection::pb::server_reflection_request::MessageRequest; +use tonic_reflection::pb::server_reflection_response::MessageResponse; +pub use tonic_reflection::pb::server_reflection_server::{ + ServerReflection, ServerReflectionServer, +}; +use tonic_reflection::pb::{ + ExtensionNumberResponse, FileDescriptorResponse, ListServiceResponse, ServerReflectionRequest, + ServerReflectionResponse, ServiceResponse, +}; + +/// Represents an error in the construction of a gRPC Reflection Service. +#[derive(Debug)] +pub enum Error { + /// An error was encountered decoding a `prost_types::FileDescriptorSet` from a buffer. + DecodeError(prost::DecodeError), + /// An invalid `prost_types::FileDescriptorProto` was encountered. + InvalidFileDescriptorSet(String), +} + +impl From for Error { + fn from(e: DecodeError) -> Self { + Error::DecodeError(e) + } +} + +impl std::error::Error for Error {} + +impl Display for Error { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + match self { + Error::DecodeError(_) => f.write_str("error decoding FileDescriptorSet from buffer"), + Error::InvalidFileDescriptorSet(s) => { + write!(f, "invalid FileDescriptorSet - {}", s) + } + } + } +} + +/// A builder used to construct a gRPC Reflection Service. +#[derive(Debug)] +pub struct Builder<'b> { + file_descriptor_sets: Vec, + encoded_file_descriptor_sets: Vec<&'b [u8]>, + include_reflection_service: bool, + + service_names: Vec, + use_all_service_names: bool, + symbols: HashMap>, +} + +impl<'b> Builder<'b> { + /// Create a new builder that can configure a gRPC Reflection Service. + pub fn configure() -> Self { + Builder { + file_descriptor_sets: Vec::new(), + encoded_file_descriptor_sets: Vec::new(), + include_reflection_service: true, + + service_names: Vec::new(), + use_all_service_names: true, + symbols: HashMap::new(), + } + } + + /// Registers an instance of `prost_types::FileDescriptorSet` with the gRPC Reflection + /// Service builder. + pub fn register_file_descriptor_set(mut self, file_descriptor_set: FileDescriptorSet) -> Self { + self.file_descriptor_sets.push(file_descriptor_set); + self + } + + /// Registers a byte slice containing an encoded `prost_types::FileDescriptorSet` with + /// the gRPC Reflection Service builder. + pub fn register_encoded_file_descriptor_set( + mut self, + encoded_file_descriptor_set: &'b [u8], + ) -> Self { + self.encoded_file_descriptor_sets + .push(encoded_file_descriptor_set); + self + } + + /// Serve the gRPC Reflection Service descriptor via the Reflection Service. This is enabled + /// by default - set `include` to false to disable. + pub fn include_reflection_service(mut self, include: bool) -> Self { + self.include_reflection_service = include; + self + } + + /// Advertise a fully-qualified gRPC service name. + /// + /// If not called, then all services present in the registered file descriptor sets + /// will be advertised. + pub fn with_service_name(mut self, name: impl Into) -> Self { + self.use_all_service_names = false; + self.service_names.push(name.into()); + self + } + + /// Build a gRPC Reflection Service to be served via Tonic. + pub fn build(mut self) -> Result, Error> { + if self.include_reflection_service { + self = self + .register_encoded_file_descriptor_set(tonic_reflection::pb::FILE_DESCRIPTOR_SET); + } + + for encoded in &self.encoded_file_descriptor_sets { + let decoded = FileDescriptorSet::decode(*encoded)?; + self.file_descriptor_sets.push(decoded); + } + + let all_fds = self.file_descriptor_sets.clone(); + let mut files: HashMap> = HashMap::new(); + + for fds in all_fds { + for fd in fds.file { + let name = match fd.name.clone() { + None => { + return Err(Error::InvalidFileDescriptorSet("missing name".to_string())); + } + Some(n) => n, + }; + + if files.contains_key(&name) { + continue; + } + + let fd = Arc::new(fd); + files.insert(name, fd.clone()); + + self.process_file(fd)?; + } + } + + let service_names = self + .service_names + .iter() + .map(|name| ServiceResponse { name: name.clone() }) + .collect(); + + Ok(ServerReflectionServer::new(ReflectionService { + state: Arc::new(ReflectionServiceState { + service_names, + files, + symbols: self.symbols, + }), + })) + } + + fn process_file(&mut self, fd: Arc) -> Result<(), Error> { + let prefix = &fd.package.clone().unwrap_or_default(); + + for msg in &fd.message_type { + self.process_message(fd.clone(), prefix, msg)?; + } + + for en in &fd.enum_type { + self.process_enum(fd.clone(), prefix, en)?; + } + + for service in &fd.service { + let service_name = extract_name(prefix, "service", service.name.as_ref())?; + if self.use_all_service_names { + self.service_names.push(service_name.clone()); + } + self.symbols.insert(service_name.clone(), fd.clone()); + + for method in &service.method { + let method_name = extract_name(&service_name, "method", method.name.as_ref())?; + self.symbols.insert(method_name, fd.clone()); + } + } + + Ok(()) + } + + fn process_message( + &mut self, + fd: Arc, + prefix: &str, + msg: &DescriptorProto, + ) -> Result<(), Error> { + let message_name = extract_name(prefix, "message", msg.name.as_ref())?; + self.symbols.insert(message_name.clone(), fd.clone()); + + for nested in &msg.nested_type { + self.process_message(fd.clone(), &message_name, nested)?; + } + + for en in &msg.enum_type { + self.process_enum(fd.clone(), &message_name, en)?; + } + + for field in &msg.field { + self.process_field(fd.clone(), &message_name, field)?; + } + + for oneof in &msg.oneof_decl { + let oneof_name = extract_name(&message_name, "oneof", oneof.name.as_ref())?; + self.symbols.insert(oneof_name, fd.clone()); + } + + Ok(()) + } + + fn process_enum( + &mut self, + fd: Arc, + prefix: &str, + en: &EnumDescriptorProto, + ) -> Result<(), Error> { + let enum_name = extract_name(prefix, "enum", en.name.as_ref())?; + self.symbols.insert(enum_name.clone(), fd.clone()); + + for value in &en.value { + let value_name = extract_name(&enum_name, "enum value", value.name.as_ref())?; + self.symbols.insert(value_name, fd.clone()); + } + + Ok(()) + } + + fn process_field( + &mut self, + fd: Arc, + prefix: &str, + field: &FieldDescriptorProto, + ) -> Result<(), Error> { + let field_name = extract_name(prefix, "field", field.name.as_ref())?; + self.symbols.insert(field_name, fd); + Ok(()) + } +} + +fn extract_name( + prefix: &str, + name_type: &str, + maybe_name: Option<&String>, +) -> Result { + match maybe_name { + None => Err(Error::InvalidFileDescriptorSet(format!( + "missing {} name", + name_type + ))), + Some(name) => { + if prefix.is_empty() { + Ok(name.to_string()) + } else { + Ok(format!("{}.{}", prefix, name)) + } + } + } +} + +#[derive(Debug)] +struct ReflectionServiceState { + service_names: Vec, + files: HashMap>, + symbols: HashMap>, +} + +impl ReflectionServiceState { + fn list_services(&self) -> MessageResponse { + MessageResponse::ListServicesResponse(ListServiceResponse { + service: self.service_names.clone(), + }) + } + + fn symbol_by_name(&self, symbol: &str) -> Result { + match self.symbols.get(symbol) { + None => Err(Status::not_found(format!("symbol '{}' not found", symbol))), + Some(fd) => { + let mut encoded_fd = Vec::new(); + if fd.clone().encode(&mut encoded_fd).is_err() { + return Err(Status::internal("encoding error")); + }; + + Ok(MessageResponse::FileDescriptorResponse( + FileDescriptorResponse { + file_descriptor_proto: vec![encoded_fd], + }, + )) + } + } + } + + fn file_by_filename(&self, filename: &str) -> Result { + match self.files.get(filename) { + None => Err(Status::not_found(format!("file '{}' not found", filename))), + Some(fd) => { + let mut encoded_fd = Vec::new(); + if fd.clone().encode(&mut encoded_fd).is_err() { + return Err(Status::internal("encoding error")); + } + + Ok(MessageResponse::FileDescriptorResponse( + FileDescriptorResponse { + file_descriptor_proto: vec![encoded_fd], + }, + )) + } + } + } +} + +#[derive(Debug)] +struct ReflectionService { + state: Arc, +} + +#[tonic::async_trait] +impl ServerReflection for ReflectionService { + type ServerReflectionInfoStream = ReceiverStream>; + + async fn server_reflection_info( + &self, + req: Request>, + ) -> Result, Status> { + let mut req_rx = req.into_inner(); + let (resp_tx, resp_rx) = mpsc::channel::>(1); + + let state = self.state.clone(); + + tokio::spawn(async move { + while let Some(req) = req_rx.next().await { + let req = match req { + Ok(req) => req, + Err(_) => { + return; + } + }; + + let resp_msg = match req.message_request.clone() { + None => Err(Status::invalid_argument("invalid MessageRequest")), + Some(msg) => match msg { + MessageRequest::FileByFilename(s) => state.file_by_filename(&s), + MessageRequest::FileContainingSymbol(s) => state.symbol_by_name(&s), + MessageRequest::FileContainingExtension(_) => { + Err(Status::not_found("extensions are not supported")) + } + MessageRequest::AllExtensionNumbersOfType(_) => { + // NOTE: Workaround. Some grpc clients (e.g. grpcurl) expect this method not to fail. + // https://github.com/hyperium/tonic/issues/1077 + Ok(MessageResponse::AllExtensionNumbersResponse( + ExtensionNumberResponse::default(), + )) + } + MessageRequest::ListServices(_) => Ok(state.list_services()), + }, + }; + + match resp_msg { + Ok(resp_msg) => { + let resp = ServerReflectionResponse { + valid_host: req.host.clone(), + original_request: Some(req.clone()), + message_response: Some(resp_msg), + }; + resp_tx.send(Ok(resp)).await.expect("send"); + } + Err(status) => { + resp_tx.send(Err(status)).await.expect("send"); + return; + } + } + } + }); + + Ok(Response::new(ReceiverStream::new(resp_rx))) + } +} From 7c04a02bb37ed2f04f15b4385bdb7e71ff551ce2 Mon Sep 17 00:00:00 2001 From: Sandipsinh Rathod Date: Mon, 19 Feb 2024 14:16:31 +0530 Subject: [PATCH 2/2] upgrade deps --- Cargo.lock | 8 ++++---- Cargo.toml | 4 ++-- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index f619742..777e682 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1010,9 +1010,9 @@ dependencies = [ [[package]] name = "tonic" -version = "0.10.2" +version = "0.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d560933a0de61cf715926b9cac824d4c883c2c43142f787595e48280c40a1d0e" +checksum = "76c4eb7a4e9ef9d4763600161f12f5070b92a578e1b634db88a6887844c91a13" dependencies = [ "async-stream", "async-trait", @@ -1050,9 +1050,9 @@ dependencies = [ [[package]] name = "tonic-reflection" -version = "0.10.2" +version = "0.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3fa37c513df1339d197f4ba21d28c918b9ef1ac1768265f11ecb6b7f1cba1b76" +checksum = "548c227bd5c0fae5925812c4ec6c66ffcfced23ea370cb823f4d18f0fc1cb6a7" dependencies = [ "prost", "prost-types", diff --git a/Cargo.toml b/Cargo.toml index 493556b..a1bdc99 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,8 +8,8 @@ edition = "2021" [dependencies] hyper = {version = "0.14.28",features = ["full"]} tokio = {version = "1.35.1",features = ["full"]} -tonic = "0.10.2" -tonic-reflection = "0.10.2" +tonic = "0.11.0" +tonic-reflection = "0.11.0" prost = "0.12.3" prost-types = "0.12.3" tower = "0.4.13"