如何監控你的 Rust 應用程序？
在這篇文章中，我們將從 DevOps 的角度展示如何監控 Rust 應用程序並生成跟蹤數據。
我們使用廣泛認可的 Jaeger UI 進行分析跟蹤，以深入瞭解應用程序的行爲。並從這些痕跡中得出 RED(率、錯誤和持續時間) 指標，並在 Grafana 中監控它們。
我們的演練需要做到以下 4 點：
-
用 Actix 構建簡單的 web API。
-
推送痕跡數據和指標數據到 Quickwit。
-
使用 Jaeger UI 檢測、診斷和解決問題。
-
在 Grafana 中監控應用程序的 RED 指標 (率，錯誤，持續時間)。
在我們深入討論細節之前，請確保你的系統上安裝了以下軟件並正常運行：
-
Rust 1.68+
-
Docker
-
docker-compose
構建和測試 Rust 應用程序
我們將使用 Actix web 框架創建一個基本的 Rust 應用程序，它將從 JSONPlaceholder 網站的公共 web API 獲取帖子及其評論，並將其顯示爲 JSON。
項目初始化和核心依賴項
新建一個名爲 rust-app-tracing 的目錄並在目錄下創建一個 Rust 項目：
cargo new web-api在 Cargo.toml 文件中加入以下依賴項：
[dependencies]
actix-web = "4.1.0"
actix-web-opentelemetry = "0.13.0"
anyhow = "1.0.71"
futures = "0.3.28"
opentelemetry = { version = "0.19.0", default-features = false, features = ["trace", "rt-tokio"] }
opentelemetry-otlp = { version = "0.12.0", features = ["reqwest-client", "reqwest-rustls", "http-proto"] }
opentelemetry-semantic-conventions = "0.11.0"
reqwest = {version = "0.11.18", features = ["json"] }
serde = { version = "1.0.143", features = ["derive"] }
serde_json = "1.0.83"
tokio = { version = "1.24", features = ["full"] }
tracing = "0.1.36"
tracing-bunyan-formatter = "0.3.3"
tracing-opentelemetry = "0.19.0"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
rand = "0.8.5"
log = "0.4.17"
tracing-actix-web = "0.7"-
actix-web-opentelemetry：actix-web 框架的 open-telemetry 擴展。
-
openelemetry：Rust 的核心 open-telemetry SDK，包括跟蹤和度量。
-
opentelemetry-otlp：提供各種 open-telemetry 的輸出。
web API 應用程序代碼
首先，創建一個名爲 src/telemetry.rs 的文件來配置應用程序的跟蹤。我們將在其中設置所有的跟蹤配置。
use opentelemetry::{global, KeyValue};
use opentelemetry::sdk::propagation::TraceContextPropagator;
use opentelemetry::sdk::{trace, Resource};
use opentelemetry_otlp::WithExportConfig;
use tracing_bunyan_formatter::{BunyanFormattingLayer, JsonStorageLayer};
use tracing_subscriber::Registry;
use tracing_subscriber::{prelude::*, EnvFilter};
const SERVICE_NAME: &'static str = "quickwit-jaeger-demo";
pub fn init_telemetry(exporter_endpoint: &str) {
// 創建 gRPC 輸出
let exporter = opentelemetry_otlp::new_exporter()
.tonic()
.with_endpoint(exporter_endpoint);
// 定義跟蹤器
let tracer = opentelemetry_otlp::new_pipeline()
.tracing()
.with_exporter(exporter)
.with_trace_config(
trace::config().with_resource(Resource::new(vec![KeyValue::new(
opentelemetry_semantic_conventions::resource::SERVICE_NAME,
SERVICE_NAME.to_string(),
)])),
)
.install_batch(opentelemetry::runtime::Tokio)
.expect("Error: Failed to initialize the tracer.");
// 定義訂閱者
let subscriber = Registry::default();
// 定義過濾器，基於級別過濾跟蹤(trace, debug, info, warn, error)
let level_filter_layer = EnvFilter::try_from_default_env().unwrap_or(EnvFilter::new("INFO"));
// 添加我們自定義配置的跟蹤器
let tracing_layer = tracing_opentelemetry::layer().with_tracer(tracer);
// 用於打印到標準輸出
let formatting_layer = BunyanFormattingLayer::new(
SERVICE_NAME.to_string(),
std::io::stdout,
);
global::set_text_map_propagator(TraceContextPropagator::new());
subscriber
.with(level_filter_layer)
.with(tracing_layer)
.with(JsonStorageLayer)
.with(formatting_layer)
.init()
}接下來，讓我們實現 API 端點，同時向處理程序函數添加一些檢測。重要的是要注意，我們的重點不是這個應用程序做什麼，而是從應用程序生成有意義的和可利用的跟蹤數據。
首先，我們創建一個 src/models.rs 文件，用於序列化和反序列化帖子和評論：
use serde::{Serialize, Deserialize};
#[derive(Debug, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct Post {
pub user_id: i64,
pub id: i64,
pub title: String,
pub body: String,
#[serde(default)]
pub comments: Vec<Comment>,
}
#[derive(Debug, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct Comment {
pub post_id: i64,
pub id: i64,
pub name: String,
pub email: String,
pub body: String,
}接下來，讓我們實現 API 端點的處理程序。注意 instrument 屬性裝飾了一些函數。這就是我們如何啓用函數的跟蹤及對它執行後續任務的跟蹤。
創建一個 src/lib.rs 文件：
mod telemetry;
mod models;
use futures::StreamExt;
use rand::seq::SliceRandom;
pub use telemetry::init_telemetry;
pub use models::{Post, Comment};
use anyhow::anyhow;
use reqwest::{Client, StatusCode};
use serde::{de::DeserializeOwned};
use actix_web::{get, web, Error, HttpResponse};
use tracing::instrument;
const BASE_API_URL: &'static str = "https://jsonplaceholder.typicode.com";
// web api: /posts
pub fn fetch_posts_service() -> actix_web::Scope {
web::scope("/posts")
.service(get_posts)
}
#[instrument(level = "info", name = "get_posts", skip_all)]
#[get("")]
async fn get_posts() -> Result<HttpResponse, Error> {
// 隨機模擬請求處理中的錯誤
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))
}
}
}
// 獲取帖子
#[instrument(level = "info", name = "fetch_posts")]
async fn fetch_posts(limit: usize) -> anyhow::Result<Vec<Post>> {
let client = Client::new();
let url = format!("{}/posts", BASE_API_URL);
let mut posts: Vec<Post> = request_url(&client, &url).await?;
posts.truncate(limit);
let post_idx_to_ids: Vec<(usize, i64)> = posts.iter().enumerate().map(|(idx, post)| (idx, post.id)).collect();
// 獲取帖子評論
for (index, post_id) in post_idx_to_ids {
let comments = fetch_comments(&client, post_id).await?;
posts[index].comments = comments
}
Ok(posts)
}
// 獲取特定帖子的評論
#[instrument(level = "info", name = "fetch_comments", skip(client))]
async fn fetch_comments(client: &Client, post_id: i64) -> anyhow::Result<Vec<Comment>> {
let url = format!("{}/posts/{}/comments", BASE_API_URL, post_id);
let comments: Vec<Comment> = request_url(&client, &url).await?;
Ok(comments)
}
// 發送get請求和反序列化json響應
async fn request_url<T: DeserializeOwned>(client: &Client, url: &str) -> anyhow::Result<T> {
let response = client.get(url)
.send()
.await?;
match response.status() {
reqwest::StatusCode::OK =>
response.json::<T>()
.await
.map_err(|err| anyhow!(err.to_string()))
,
_ => Err(anyhow!(format!("Request error with statusCode `{}`", response.status()))),
}
}在上面的代碼中，我們只發送跟蹤數據。還可以使用功能強大的日誌收集器收集日誌並向接收後端中發送。
最後，我們編寫 src/main.rs 文件，用於啓動服務：
use actix_web::{App, HttpServer};
use tracing_actix_web::TracingLogger;
use web_api::{init_telemetry, fetch_posts_service};
const EXPORTER_ENDPOINT: &'static str = "http://localhost:7281";
#[actix_web::main]
async fn main() -> std::io::Result<()> {
init_telemetry(EXPORTER_ENDPOINT);
HttpServer::new(move || {
App::new()
.wrap(TracingLogger::default())
.service(fetch_posts_service())
})
.bind(("127.0.0.1", 9000))?
.run()
.await
}執行 cargo run，然後在瀏覽器中輸入：http://127.0.0.1:9000/posts，如圖：
監控 Rust 應用程序
使用 Quickwit 來提取跟蹤數據
我們將創建一個 rust-app-tracing/docker-compose.yaml 文件來簡化 Quickwit、Jaeger 和 Grafana 之間的設置。下面的 docker-compose 文件包含所有必要的配置。
version: '3'
services:
quickwit:
image: quickwit/quickwit:latest
command: run
restart: always
environment:
QW_ENABLE_OTLP_ENDPOINT: true
QW_ENABLE_JAEGER_ENDPOINT: true
ports:
- '7280:7280'
- '7281:7281'
volumes:
- ./qwdata:/quickwit/qwdata
jaeger:
image: jaegertracing/jaeger-query:latest
restart: always
depends_on:
- quickwit
environment:
SPAN_STORAGE_TYPE: 'grpc-plugin'
GRPC_STORAGE_SERVER: 'quickwit:7281'
ports:
- '16686:16686'
grafana:
image: grafana/grafana-enterprise:latest
restart: always
user: root
depends_on:
- quickwit
environment:
GF_PLUGINS_ALLOW_LOADING_UNSIGNED_PLUGINS: 'quickwit-quickwit-datasource'
ports:
- '3000:3000'
volumes:
- ./grafana-storage:/var/lib/grafana-
QW_ENABLE_OTLP_ENDPOINT：允許 Quickwit 接受和提取跟蹤數據和日誌數據。
-
SPAN_STORAGE_TYPE,GRPC_STORAGE_SERVER,QW_ENABLE_JAEGER_ENDPOINT：允許 Jaeger 從 Quickwit 提取跟蹤數據和日誌數據用於分析目的。
-
GF_PLUGINS_ALLOW_LOADING_UNSIGNED_PLUGINS：允許我們在 Grafana 中加載特定的插件。
在 rust-app-tracing 目錄下創建用於存儲 Quickwit 數據的 qwdata 目錄。
然後，在 rust-app-tracing 目錄下創建 grafana-storage/plugins 目錄，下載並放置 Quickwit Grafana 數據源插件在這個位置。
wget https://github.com/quickwit-oss/quickwit-datasource/releases/download/v0.2.0/quickwit-quickwit-datasource-0.2.0.zip \
&& mkdir -p grafana-storage/plugins \
&& unzip quickwit-quickwit-datasource-0.2.0.zip -d grafana-storage/plugins現在讓我們運行以下命令啓動所有服務 (Quickwit, Jaeger, Grafana)：
docker compose up -d現在進入 web-api 目錄，運行我們的 web 應用程序：
cargo run執行如下命令：
curl -X GET http://localhost:9000/posts等待大約 10 秒，新跟蹤將被編入索引並可用於搜索。
現在可以通過 CURL 搜索 otel-traces-v0_6 索引來檢查 Quickwit 是否已經索引了跟蹤數據。
curl -X POST http://localhost:7280/api/v1/otel-traces-v0_6/search -H 'Content-Type: application/json' -d '{ "query": "service_name:quickwit-jaeger-demo" }'你也可以使用 Quickwit UI 查看數據：http://localhost:7280/ui/search，
如圖：
使用 Jaeger UI 分析痕跡
Jaeger 容器已經運行，通過 http://localhost:16686 地址就可以可視化的查看應用程序的跟蹤數據。
從上面的截圖中可以看到，我們依次爲每個帖子獲取評論。也就是說，我們一個接一個地提出二十個請求，這使得整個請求處理時間更長。
我們能做得更好嗎？讓我們利用 Tokio 和 Rust futures crate 的異步流特性來併發地獲取評論。
修改 fetch_posts 函數，以便以批處理的方式併發地運行請求，每次處理 10 個請求。這應該能加快一點速度。
// 獲取帖子
#[instrument(level = "info", name = "fetch_posts")]
async fn fetch_posts(limit: usize) -> anyhow::Result<Vec<Post>> {
let client = Client::new();
let url = format!("{}/posts", BASE_API_URL);
let mut posts: Vec<Post> = request_url(&client, &url).await?;
posts.truncate(limit);
let post_idx_to_ids: Vec<(usize, i64)> = posts.iter().enumerate().map(|(idx, post)| (idx, post.id)).collect();
// 獲取帖子評論
// for (index, post_id) in post_idx_to_ids {
// let comments = fetch_comments(&client, post_id).await?;
// posts[index].comments = comments
// }
// 併發獲取帖子評論
let tasks: Vec<_> = post_idx_to_ids
.into_iter()
.map(|(index, post_id)| {
let moved_client = client.clone();
async move {
let comments_fetch_result = fetch_comments(&moved_client, post_id).await;
(index, comments_fetch_result)
}
})
.collect();
let mut stream = futures::stream::iter(tasks)
.buffer_unordered(10);
while let Some((index, comments_fetch_result)) = stream.next().await {
let comments = comments_fetch_result?;
posts[index].comments = comments;
}
Ok(posts)
}重新運行程序，再次查看，發現時間縮短了。
使用 Grafana 監控應用程序指標
Jaeger 適合集中檢查單個痕跡。但是，如果我們想要監視服務的延遲，該怎麼辦呢？如果我們想用給定的跟蹤元數據計算錯誤或請求的數量，該怎麼辦？
這就是 Grafana 儀表板發揮作用的地方，我們希望從追蹤的數據中構建 RED 指標，並在 Grafana 中可視化它們。
在瀏覽器中輸入 http://localhost:3000/login，使用 admin 作爲用戶名和密碼登錄。登錄後，我們可以連接到 Quickwit，並使用新發布的 Quickwit 數據源插件查詢我們的應用程序跟蹤數據。
總結
在這篇文章中，我們深入探討了如何實現 Rust 應用程序的分佈式跟蹤以及如何使用它來監視應用程序的性能。
本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/95MhAp341u3IdiPErFisng