Merge

.gitignore

@@ -7,7 +7,3 @@ web/dist
 web/tmp
 adapters
 plugins
-!qwdata/.gitkeep
-qwdata
-!grafana-storage/plugins
-grafana-storage

crates/web-api-plugin-tester/Cargo.toml

@@ -16,3 +16,5 @@ rand = { version = "0.8.5" }
 reqwest = { version = "0.11.18", features = ['json', 'serde_json', 'stream', 'rustls', 'tokio-util', 'socks', 'futures-channel'] }
 toml = { version = "0.7.4" }
 tracing = { version = "0.1.37" }
+anyhow = { version = "1.0.71" }
+rayon = { version = "1.7.0" }

crates/web-api-plugin-tester/src/main.rs

@@ -1,7 +1,6 @@
-use actix_web::http::StatusCode;
 use actix_web::{get, App, Error, HttpResponse, HttpServer};
+use futures::prelude::*;
 use plugin_api::{AppConfig, Plugin};
-use rand::prelude::SliceRandom;
 use reqwest::Client;
 use serde::*;
 use telemetry_plugin::TelemetryPlugin;
@@ -16,7 +15,7 @@ async fn main() -> std::io::Result<()> {
             let mut t = toml::Table::with_capacity(3);
             t.insert(
                 "endpoint".to_string(),
-                toml::Value::String("http://localhost:16686".to_string()),
+                toml::Value::String("http://0.0.0.0:16686".to_string()),
             );
             t.insert(
                 "log_level".to_string(),
@@ -49,19 +48,19 @@ const BASE_API_URL: &'static str = "https://jsonplaceholder.typicode.com";
 #[get("/posts")]
 async fn get_posts() -> Result<HttpResponse, Error> {
     // Randomly simulate errors in request handling
-    let choices = [200, 400, 401, 200, 500, 501, 200, 500];
-    let mut rng = rand::thread_rng();
-    let choice = choices.choose(&mut rng).unwrap().clone();
-    match choice {
-        400..=401 => Ok(HttpResponse::new(StatusCode::from_u16(choice).unwrap())),
-        500..=501 => Ok(HttpResponse::new(StatusCode::from_u16(choice).unwrap())),
-        _ => {
-            let posts = fetch_posts(20)
-                .await
-                .map_err(actix_web::error::ErrorInternalServerError)?;
-            Ok(HttpResponse::Ok().json(posts))
-        }
-    }
+    // let choices = [200, 400, 401, 200, 500, 501, 200, 500];
+    // let mut rng = rand::thread_rng();
+    // let choice = choices.choose(&mut rng).unwrap().clone();
+    // match choice {
+    // 400..=401 => Ok(HttpResponse::new(StatusCode::from_u16(choice).unwrap())),
+    // 500..=501 => Ok(HttpResponse::new(StatusCode::from_u16(choice).unwrap())),
+    // _ => {
+    let posts = fetch_posts(20)
+        .await
+        .map_err(actix_web::error::ErrorInternalServerError)?;
+    Ok(HttpResponse::Ok().json(posts))
+    // }
+    // }
 }
 
 // Fetching posts with a limit.
@@ -77,11 +76,23 @@ async fn fetch_posts(limit: usize) -> anyhow::Result<Vec<Post>> {
         .map(|(idx, post)| (idx, post.id))
         .collect();
 
-    // fetch post comments one after another.
-    for (index, post_id) in post_idx_to_ids {
-        let comments = fetch_comments(&client, post_id).await?;
-        posts[index].comments = comments
+    let mut res = futures::stream::iter(post_idx_to_ids.iter())
+        .map(|(idx, post_id)| {
+            let client = client.clone();
+            async move {
+                let comments = fetch_comments(&client, *post_id).await.unwrap();
+                (*idx, comments)
+            }
+        })
+        .buffer_unordered(3);
+
+    for (idx, comments) in res.next().await {
+        posts[idx].comments = comments;
     }
+    // fetch post comments one after another.
+    // for (index, post_id) in post_idx_to_ids {
+    //
+    // }
 
     Ok(posts)
 }
@@ -89,8 +100,7 @@ async fn fetch_posts(limit: usize) -> anyhow::Result<Vec<Post>> {
 #[instrument(level = "info", name = "fetch_comments")]
 async fn fetch_comments(client: &Client, id: i64) -> anyhow::Result<Vec<Comment>> {
     let url = format!("{BASE_API_URL}/posts/{id}/comments");
-    let mut comments: Vec<Comment> = client.get(url).send().await?.json().await?;
-    Ok(comments)
+    Ok(client.get(url).send().await?.json().await?)
 }
 
 #[derive(Debug, Serialize, Deserialize)]

docker-compose.yml

@@ -1,7 +1,7 @@
 version: '3'
 services:
   quickwit:
-    image: quickwit/quickwit:latest
+    image: quickwit/quickwit:v0.5.2
     command: run
     restart: always
     environment:
@@ -10,11 +10,11 @@ services:
     ports:
       - '7280:7280'
       - '7281:7281'
-    volumes:
-      - ./qwdata:/quickwit/qwdata
+#    volumes:
+#      - ./qwdata:/quickwit/qwdata
 
   jaeger:
-    image: jaegertracing/jaeger-query:latest
+    image: jaegertracing/jaeger-query:1.45
     restart: always
     depends_on:
       - quickwit
@@ -25,9 +25,8 @@ services:
       - '16686:16686'
 
   grafana:
-    image: grafana/grafana-enterprise:latest
+    image: grafana/grafana-enterprise:10.0.0
     restart: always
-    user: root
     depends_on:
       - quickwit
     environment:
@@ -35,4 +34,4 @@ services:
     ports:
       - '3000:3000'
     volumes:
-      - ./grafana-storage:/var/lib/grafana
+      - ./grafana/plugins:/var/lib/grafana/plugins