leekeiabstraction commented on code in PR #617: URL: https://github.com/apache/fluss-rust/pull/617#discussion_r3423844400
########## website/docs/user-guide/rust/metrics.md: ########## @@ -0,0 +1,179 @@ +--- +sidebar_position: 5 +--- +# Metrics + +The Fluss Rust client is instrumented with client-side metrics for the connection +layer, the write pipeline, and the read (scanner) pipeline. Metrics are emitted +through the [`metrics`](https://docs.rs/metrics) crate facade, so collecting them is +opt-in and costs nothing until you install a recorder. + +## How it works + +The client never decides *where* metrics go. It only emits them via the `metrics` +facade. Your application installs a global **recorder** (for example a Prometheus +exporter), and that recorder decides how to store and expose the values. + +- **No recorder installed** — every metric call is a zero-cost no-op. This is the + default, so the client adds no overhead unless you opt in. +- **Recorder installed** — values flow to whatever backend the recorder represents + (Prometheus, StatsD, OpenTelemetry, a test recorder, etc.). + +This differs from the Fluss Java client, where metric reporters are configured +server-side in `conf/server.yaml` (`metrics.reporters: jmx,prometheus`) and +discovered through plugins. The Rust client instead follows the idiomatic Rust +`metrics` ecosystem: the application owns recorder installation, and rate +computation is left to the backend (e.g. PromQL `rate()`) instead of a built-in +background rate thread. + +## Installing a recorder + +Use any [`metrics`-compatible exporter](https://docs.rs/metrics/latest/metrics/#related-crates). +The example below uses `metrics-exporter-prometheus` to expose a scrape endpoint: + +```rust +use metrics_exporter_prometheus::PrometheusBuilder; + +PrometheusBuilder::new() + .with_http_listener(([0, 0, 0, 0], 9000)) + .install() + .expect("failed to install Prometheus recorder"); +``` + +A full, runnable program is available as the `example-prometheus-metrics` example +in the `fluss-examples` crate. + +:::warning Install the recorder before writing or scanning +The client caches metric handles the first time a writer or scanner is created, +binding them to whichever recorder is installed at that moment. Install your global +recorder **before** calling `FlussConnection::new` (ideally as the very first thing +in `main`). If you install it after creating a writer or scanner, those metrics will Review Comment: This behaviour might catch user off guard. Is it worth making it so that an Error is returned instead? ########## crates/examples/src/example_prometheus_metrics.rs: ########## @@ -0,0 +1,136 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Exposes Fluss client metrics on a Prometheus scrape endpoint. +//! +//! Run a local cluster, then: +//! ```shell +//! cargo run -p fluss-examples --example example-prometheus-metrics +//! curl http://localhost:9000/metrics +//! ``` +//! The endpoint exposes `fluss_client_writer_*`, `fluss_client_scanner_*`, and +//! `fluss_client_requests_*` series produced by the workload below. + +#[cfg(not(target_env = "msvc"))] +#[global_allocator] +static GLOBAL: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc; + +use clap::Parser; +use fluss::client::FlussConnection; +use fluss::config::Config; +use fluss::error::Result; +use fluss::metadata::{DataTypes, Schema, TableDescriptor, TablePath}; +use fluss::row::{DataGetters, GenericRow}; +use metrics_exporter_prometheus::PrometheusBuilder; +use std::time::Duration; + +#[tokio::main] +pub async fn main() -> Result<()> { + // Install the global Prometheus recorder BEFORE creating any connection, + // writer, or scanner: the client caches metric handles on first use and + // binds them to whichever recorder is installed at that moment. + // + // `build()` (rather than `install()`) hands back a `PrometheusHandle` so the + // example can read its own metrics back and self-verify; the returned + // exporter future runs the HTTP scrape endpoint. + let (recorder, exporter) = PrometheusBuilder::new() + .with_http_listener(([0, 0, 0, 0], 9000)) + .build() + .expect("failed to build Prometheus recorder"); + let metrics_handle = recorder.handle(); + metrics::set_global_recorder(recorder).expect("failed to install global recorder"); + tokio::spawn(exporter); + println!("Metrics exposed on http://localhost:9000/metrics";); + + let mut config = Config::parse(); + config.bootstrap_servers = "127.0.0.1:9123".to_string(); + + let conn = FlussConnection::new(config).await?; + let admin = conn.get_admin()?; + + let table_path = TablePath::new("fluss", "rust_prometheus_metrics"); + let table_descriptor = TableDescriptor::builder() + .schema( + Schema::builder() + .column("id", DataTypes::int()) + .column("message", DataTypes::string()) + .build()?, + ) + .build()?; + admin + .create_table(&table_path, &table_descriptor, true) + .await?; + + let table = conn.get_table(&table_path).await?; + let append_writer = table.new_append()?.create_writer()?; + let log_scanner = table.new_scan().create_log_scanner()?; + log_scanner.subscribe(0, 0).await?; + + // Continuously write and read so the metrics keep updating and can be + // observed over multiple Prometheus scrapes. + let rows_per_iter = 100; + let mut id = 0i32; + let mut verified = false; + loop { + for _ in 0..rows_per_iter { + let mut row = GenericRow::new(2); + row.set_field(0, id); + row.set_field(1, "metrics demo"); + append_writer.append(&row)?; + id += 1; + } + append_writer.flush().await?; + + let scan_records = log_scanner.poll(Duration::from_secs(1)).await?; + let mut count = 0; + for record in scan_records { Review Comment: We are only verifying metrics for sent records, do we need the loop for polling records? If not, it might be worth removing this loop for a more concise example. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
