rust(feat): Add test server to CLI #450
Conversation
rust/crates/sift_cli/build.rs
Outdated
| @@ -0,0 +1,21 @@ | |||
| /// Build descriptor's so that the Black Hole gRPC server can | |||
| /// stand up the reflection service. | |||
| fn main() -> Result<(), Box<dyn std::error::Error>> { | |||
There was a problem hiding this comment.
There shouldn't be a need for you to compile these protos since they're already compiled and available in the sift_rs crate which you can add as a dependency like here:
sift/rust/crates/sift_cli/Cargo.toml
Line 28 in 737efa8
There was a problem hiding this comment.
This is just compiling the descriptors since we need that for the reflection service. We could either move this to sift_rs and create this when generating the protos, or keep it in the build step for sift_cli.
| use crate::cmd::test_server::metrics_streaming_client::Metrics; | ||
|
|
||
| pub async fn run(ctx: Context, args: TestServerArgs) -> Result<ExitCode> { | ||
| let local_address = args.local_address.unwrap_or("127.0.0.1:50051".into()); |
There was a problem hiding this comment.
I would prefer 0.0.0.0 since it's more user-friendly. Allows it to be reached inside of a docker network for example. Also as a note of good practice:
local_address = args.local_address.unwrap_or_else(|| "127.0.0.1:50051".to_string())for lazy evaluation of the fallback rather than immediate
|
|
||
| pub async fn run(ctx: Context, args: TestServerArgs) -> Result<ExitCode> { | ||
| let local_address = args.local_address.unwrap_or("127.0.0.1:50051".into()); | ||
| let addr = local_address.parse()?; |
There was a problem hiding this comment.
Generally helpful to include contextual information with errors for better traceability:
use anyhow::{Result, Context};
let addr = local_address.parse().context("foobar")?;There was a problem hiding this comment.
Same for all other places.
| // Initialize streaming client. | ||
| let mut streaming_client = | ||
| MetricsStreamingClient::build(ctx, &args.stream_metrics, &args.metrics_asset_name)?; | ||
| if streaming_client.is_some() { |
There was a problem hiding this comment.
Prefer:
if let Some(client) = streaming_client.as_mut() {
client.initialize.await.context("failed to initialize client").await?;
}|
|
||
| // Start task to ingest metrics to Sift. | ||
| let ingest_metrics_task = tokio::spawn(async move { | ||
| if streaming_client.is_none() { |
There was a problem hiding this comment.
Prefer:
let Some(mut client) = streaming_client else {
return;
};| return; | ||
| } | ||
|
|
||
| let mut client = streaming_client.unwrap(); |
There was a problem hiding this comment.
see:
let Some(mut client) = streaming_client else {
return;
};
solidiquis
left a comment
solidiquis
left a comment
There was a problem hiding this comment.
Definitely run: cargo fmt and cargo clippy. The latter will give you a lot of good feedback.
| last_total_num_messages = current_total_num_messages; | ||
|
|
||
| // Clear terminal and print metrics. | ||
| stdout() |
There was a problem hiding this comment.
Calling stdout() repeatedly is bad for performance because it will repeatedly get a handle and lock stdout. See this method to acquire the handle/lock once: https://doc.rust-lang.org/std/io/struct.Stdout.html#method.lock
| #[derive(Default)] | ||
| pub struct TestServer { | ||
| /// Total number of streams created. | ||
| total_num_streams: AtomicU32, |
There was a problem hiding this comment.
Do these need to be atomics?
There was a problem hiding this comment.
They're modified in the the grpc handlers, so they need to be some type of synchronized struct.
| #[tonic::async_trait] | ||
| impl PingService for TestServer { | ||
| async fn ping(&self, _request: Request<PingRequest>) -> Result<Response<PingResponse>, Status> { | ||
| let resp = PingResponse { |
There was a problem hiding this comment.
You might be able to just do PingResponse::default().
| return AnyhowOk(()); | ||
| } | ||
|
|
||
| _ = tokio::time::sleep(Duration::from_secs(1)) => { |
There was a problem hiding this comment.
Would it be better for this to be a tokio::time::interval? Just using sleep will skew over time, but it might not matter.
| last_total_num_messages = current_total_num_messages; | ||
|
|
||
| // Clear terminal and print metrics. | ||
| stdout |
There was a problem hiding this comment.
I think the server will often be ran separately in a background thread, or as a daemon; I think that that might make recording the logs to a file look weird? I've heard similar feedback for the CLI generally when running it from a script and recording logs to a file.
| async fn list_assets( | ||
| &self, | ||
| _request: Request<sift_rs::assets::v1::ListAssetsRequest>, | ||
| ) -> Result<Response<sift_rs::assets::v1::ListAssetsResponse>, Status> { |
There was a problem hiding this comment.
Instead of no-op successful calls, would it be better to return unimplemented? So in the future if we try to add API calls to the clients but forget to update the server, the response helps make it clear the server needs to be updated with support too?
|
|
||
| Ok(Response::new(GetIngestionConfigResponse { | ||
| ingestion_config: Some(IngestionConfig { | ||
| ingestion_config_id: ingestion_config.ingestion_config_id.clone(), |
There was a problem hiding this comment.
nit: I think you can clone the entire ingestion_config directly? I might be wrong though.
| &self, | ||
| _request: Request<CreateIngestionConfigFlowsRequest>, | ||
| ) -> Result<Response<CreateIngestionConfigFlowsResponse>, Status> { | ||
| Ok(Response::new(CreateIngestionConfigFlowsResponse::default())) |
There was a problem hiding this comment.
I think we might need to implement this API and list_ingestion_config_flows. SiftStream uses create_ingestion_config_flows to add new flows and initialization uses list_ingestion_config_flows to see what already exists and if the current flows are compatible.
If this server is only used for stress testing, maybe thats not necessary. But for reconnects/reinitialization, the added behavior might help catch some regressions with those workflows.
| self.total_num_bytes_read | ||
| .fetch_add(inner.encoded_len() as u64, Relaxed); | ||
|
|
||
| if self.done.load(Relaxed) { |
There was a problem hiding this comment.
Should this verify the request being received matches a known flow-config/ingestion-config-id? That would help catch issues where maybe the data arrives before the flow is added, or the data received doesn't quite match the definition.
Or would that be a future improvement?
No description provided.