This is an automated email from the ASF dual-hosted git repository.
fresh-borzoni pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 87303ea7 [metrics] Add writer pipeline metrics (#609)
87303ea7 is described below
commit 87303ea7500efd21885a90ea8cb58562474175f4
Author: Kaiqi Dong <[email protected]>
AuthorDate: Mon Jun 8 23:34:56 2026 +0200
[metrics] Add writer pipeline metrics (#609)
* add writer pipeline metrics
* Address comments
* lint
* fmt
---
crates/fluss/src/client/write/accumulator.rs | 30 ++
crates/fluss/src/client/write/batch.rs | 127 ++++++
crates/fluss/src/client/write/sender.rs | 424 ++++++++++++++++++++-
crates/fluss/src/client/write/writer_client.rs | 6 +
crates/fluss/src/metrics.rs | 221 +++++++++++
crates/fluss/src/record/arrow.rs | 5 +
.../fluss/src/record/kv/kv_record_batch_builder.rs | 5 +
crates/fluss/src/test_utils.rs | 20 +-
website/docs/user-guide/rust/api-reference.md | 7 +
9 files changed, 834 insertions(+), 11 deletions(-)
diff --git a/crates/fluss/src/client/write/accumulator.rs
b/crates/fluss/src/client/write/accumulator.rs
index 244edf73..b41759f2 100644
--- a/crates/fluss/src/client/write/accumulator.rs
+++ b/crates/fluss/src/client/write/accumulator.rs
@@ -133,6 +133,21 @@ impl MemoryLimiter {
self.waiting_count.load(Ordering::Relaxed) > 0
}
+ /// Total buffer memory in bytes (constant)
+ pub(crate) fn total_bytes(&self) -> usize {
+ self.max_memory
+ }
+
+ /// Currently-available buffer memory in bytes
+ pub(crate) fn available_bytes(&self) -> usize {
+ self.max_memory.saturating_sub(*self.state.lock())
+ }
+
+ /// Number of producer threads currently blocked waiting for buffer memory.
+ pub(crate) fn waiting_threads(&self) -> usize {
+ self.waiting_count.load(Ordering::Relaxed)
+ }
+
/// Mark the limiter as closed and wake all blocked producers.
fn close(&self) {
self.closed.store(true, Ordering::Release);
@@ -209,6 +224,21 @@ impl RecordAccumulator {
}
}
+ /// Total writer buffer memory in bytes (constant).
+ pub(crate) fn buffer_total_bytes(&self) -> usize {
+ self.memory_limiter.total_bytes()
+ }
+
+ /// Currently-available writer buffer memory in bytes.
+ pub(crate) fn buffer_available_bytes(&self) -> usize {
+ self.memory_limiter.available_bytes()
+ }
+
+ /// Number of producer threads blocked waiting for buffer memory.
+ pub(crate) fn buffer_waiting_threads(&self) -> usize {
+ self.memory_limiter.waiting_threads()
+ }
+
fn try_append(
&self,
record: &WriteRecord,
diff --git a/crates/fluss/src/client/write/batch.rs
b/crates/fluss/src/client/write/batch.rs
index fd70cb97..50148835 100644
--- a/crates/fluss/src/client/write/batch.rs
+++ b/crates/fluss/src/client/write/batch.rs
@@ -71,6 +71,14 @@ impl InnerWriteBatch {
max(0i64, now - self.create_ms)
}
+ /// Time the batch spent queued in the accumulator, from creation until it
+ /// was drained for sending. Returns 0 if the batch has not been drained
+ /// yet (`drained_ms == -1`). Distinct from [`Self::waited_time_ms`], which
+ /// measures the batch's current age for timeout purposes.
+ fn queue_time_ms(&self) -> i64 {
+ max(0i64, self.drained_ms - self.create_ms)
+ }
+
fn complete(&self, write_result: BatchWriteResult) -> bool {
if self
.completed
@@ -135,6 +143,20 @@ impl WriteBatch {
self.inner_batch().waited_time_ms(now)
}
+ /// Time the batch spent queued in the accumulator before being drained
+ /// (`drained_ms - create_ms`). See [`InnerWriteBatch::queue_time_ms`].
+ pub(crate) fn queue_time_ms(&self) -> i64 {
+ self.inner_batch().queue_time_ms()
+ }
+
+ /// Number of records currently in this batch. Used for writer metrics.
+ pub(crate) fn record_count(&self) -> i32 {
+ match self {
+ WriteBatch::ArrowLog(batch) => batch.arrow_builder.records_count(),
+ WriteBatch::Kv(batch) => batch.record_count(),
+ }
+ }
+
pub fn close(&mut self) -> Result<()> {
match self {
WriteBatch::ArrowLog(batch) => {
@@ -408,6 +430,11 @@ impl KvWriteBatch {
pub fn estimated_size_in_bytes(&self) -> usize {
self.kv_batch_builder.get_size_in_bytes()
}
+
+ /// Number of records appended to this KV batch. Used for writer metrics.
+ pub(crate) fn record_count(&self) -> i32 {
+ self.kv_batch_builder.record_count()
+ }
}
#[cfg(test)]
@@ -436,6 +463,106 @@ mod tests {
assert_eq!(batch.attempts(), 1);
}
+ #[test]
+ fn queue_time_ms_is_drained_minus_create() {
+ let table_path = TablePath::new("db".to_string(), "tbl".to_string());
+ let physical_path = PhysicalTablePath::of(Arc::new(table_path));
+ let mut batch = InnerWriteBatch::new(1, Arc::new(physical_path),
1_000);
+ // Not drained yet -> 0 (drained_ms == -1).
+ assert_eq!(batch.queue_time_ms(), 0);
+ batch.drained(1_150);
+ assert_eq!(batch.queue_time_ms(), 150);
+ // drained() keeps the max; an earlier timestamp does not shrink it.
+ batch.drained(1_120);
+ assert_eq!(batch.queue_time_ms(), 150);
+ }
+
+ #[test]
+ fn record_count_reflects_appended_rows() {
+ use crate::client::WriteRecord;
+ use crate::compression::{
+ ArrowCompressionInfo, ArrowCompressionType,
DEFAULT_NON_ZSTD_COMPRESSION_LEVEL,
+ };
+ use crate::metadata::{DataField, DataTypes, RowType};
+ use crate::row::GenericRow;
+
+ let row_type = RowType::new(vec![DataField::new(
+ "id".to_string(),
+ DataTypes::int(),
+ None,
+ )]);
+ let table_path = TablePath::new("db".to_string(), "tbl".to_string());
+ let table_info = Arc::new(build_table_info(table_path.clone(), 1, 1));
+ let physical_table_path =
Arc::new(PhysicalTablePath::of(Arc::new(table_path)));
+
+ let arrow_batch = ArrowLogWriteBatch::new(
+ 1,
+ Arc::clone(&physical_table_path),
+ 1,
+ ArrowCompressionInfo {
+ compression_type: ArrowCompressionType::None,
+ compression_level: DEFAULT_NON_ZSTD_COMPRESSION_LEVEL,
+ },
+ &row_type,
+ 0,
+ false,
+ 2 * 1024 * 1024,
+ Arc::new(ArrowCompressionRatioEstimator::default()),
+ )
+ .unwrap();
+ let mut batch = WriteBatch::ArrowLog(arrow_batch);
+
+ assert_eq!(batch.record_count(), 0);
+ for _ in 0..5 {
+ let mut row = GenericRow::new(1);
+ row.set_field(0, 1_i32);
+ let record = WriteRecord::for_append(
+ Arc::clone(&table_info),
+ Arc::clone(&physical_table_path),
+ 1,
+ &row,
+ );
+ batch.try_append(&record).unwrap();
+ }
+ assert_eq!(batch.record_count(), 5);
+ }
+
+ #[test]
+ fn kv_record_count_reflects_appended_records() {
+ use crate::client::WriteRecord;
+ use crate::metadata::KvFormat;
+
+ let table_path = TablePath::new("db".to_string(), "tbl".to_string());
+ let table_info = Arc::new(build_table_info(table_path.clone(), 1, 1));
+ let physical_path =
Arc::new(PhysicalTablePath::of(Arc::new(table_path)));
+
+ let mut batch = WriteBatch::Kv(KvWriteBatch::new(
+ 1,
+ Arc::clone(&physical_path),
+ 1,
+ 64 * 1024,
+ KvFormat::COMPACTED,
+ None,
+ 0,
+ ));
+
+ assert_eq!(batch.record_count(), 0);
+ for _ in 0..3 {
+ let record = WriteRecord::for_upsert(
+ Arc::clone(&table_info),
+ Arc::clone(&physical_path),
+ 1,
+ Bytes::from(vec![1_u8, 2_u8, 3_u8]),
+ None,
+ WriteFormat::CompactedKv,
+ None,
+ Some(RowBytes::Owned(Bytes::from(vec![1_u8, 2_u8, 3_u8]))),
+ );
+ batch.try_append(&record).unwrap();
+ }
+ assert_eq!(batch.record_count(), 3);
+ }
+
#[test]
fn test_arrow_log_write_batch_estimated_size() {
use crate::client::WriteRecord;
diff --git a/crates/fluss/src/client/write/sender.rs
b/crates/fluss/src/client/write/sender.rs
index 8e738d0d..baf5a9b3 100644
--- a/crates/fluss/src/client/write/sender.rs
+++ b/crates/fluss/src/client/write/sender.rs
@@ -23,6 +23,7 @@ use crate::client::{ReadyWriteBatch, RecordAccumulator};
use crate::error::Error::UnexpectedError;
use crate::error::{FlussError, Result};
use crate::metadata::{PhysicalTablePath, TableBucket, TablePath};
+use crate::metrics::WriterMetrics;
use crate::proto::{
PbProduceLogRespForBucket, PbPutKvRespForBucket, PbTablePath,
ProduceLogResponse, PutKvResponse,
};
@@ -39,7 +40,7 @@ use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
-use std::time::Duration;
+use std::time::{Duration, Instant};
use tokio::sync::mpsc;
type SendFuture<'a> = Pin<Box<dyn Future<Output = Result<()>> + Send + 'a>>;
@@ -62,9 +63,11 @@ pub struct Sender {
max_request_timeout_ms: i32,
retries: i32,
idempotence_manager: Arc<IdempotenceManager>,
+ metrics: Arc<WriterMetrics>,
}
impl Sender {
+ #[allow(clippy::too_many_arguments)]
pub fn new(
metadata: Arc<Metadata>,
accumulator: Arc<RecordAccumulator>,
@@ -73,6 +76,7 @@ impl Sender {
ack: i16,
retries: i32,
idempotence_manager: Arc<IdempotenceManager>,
+ metrics: Arc<WriterMetrics>,
) -> Self {
Self {
running: AtomicBool::new(true),
@@ -84,6 +88,7 @@ impl Sender {
max_request_timeout_ms,
retries,
idempotence_manager,
+ metrics,
}
}
@@ -306,6 +311,15 @@ impl Sender {
if batches.is_empty() {
return Ok(());
}
+
+ // Record attempted-send per-batch metrics for the whole drained set
+ // up front, before any early return (unknown leader, connection
+ // failure) can drop batches. So a leader/connection failure
+ // still counts toward `records_send_total` / `bytes_send_total`.
+ // Retries re-drain the batch and therefore contribute one sample per
+ // send attempt.
+ self.record_request_batch_metrics(&batches);
+
let mut records_by_bucket = HashMap::new();
let mut write_batch_by_table: HashMap<TableId, Vec<TableBucket>> =
HashMap::new();
@@ -371,8 +385,8 @@ impl Sender {
}
};
- // let's put in back into records_by_bucket
- // since response handle will use it.
+ // Put batches back into records_by_bucket since response handling
+ // will use them.
for request_batch in request_batches {
records_by_bucket.insert(request_batch.table_bucket.clone(),
request_batch);
}
@@ -448,8 +462,14 @@ impl Sender {
records_by_bucket: &mut HashMap<TableBucket, ReadyWriteBatch>,
) -> Result<()> {
macro_rules! send {
- ($request:expr) => {
- match connection.request($request).await {
+ ($request:expr) => {{
+ // Record send latency for the request round trip regardless of
+ // outcome, so it is captured before the success/error branch.
+ let send_start = Instant::now();
+ let response_result = connection.request($request).await;
+ self.metrics
+ .record_send_latency_ms(send_start.elapsed().as_secs_f64()
* 1000.0);
+ match response_result {
Ok(response) => {
self.handle_write_response(
table_id,
@@ -471,7 +491,7 @@ impl Sender {
.await
}
}
- };
+ }};
}
match write_request {
@@ -723,9 +743,28 @@ impl Sender {
Ok(Self::is_invalid_metadata_error(error).then_some(physical_table_path))
}
+ /// Record per-batch writer throughput/queue metrics for a drained set of
+ /// batches. Invoked once at the start of `send_write_request`, before the
+ /// leader lookup / connection / serialization steps, so every drained
+ /// batch is counted exactly once per send attempt regardless of whether
+ /// the send later succeeds. Because this runs before serialization,
+ /// `estimated_size_in_bytes` is the pre-serialization estimate rather
than
+ /// the final encoded length.
+ fn record_request_batch_metrics(&self, request_batches:
&[ReadyWriteBatch]) {
+ for request_batch in request_batches {
+ let batch = &request_batch.write_batch;
+ self.metrics.record_sent_batch(
+ batch.record_count(),
+ batch.estimated_size_in_bytes(),
+ batch.queue_time_ms(),
+ );
+ }
+ }
+
fn re_enqueue_batch(&self, ready_write_batch: ReadyWriteBatch) {
self.remove_from_inflight_batches(&ready_write_batch);
- // TODO: add retry metrics (Java:
writerMetricGroup.recordsRetryTotal().inc(recordCount))
+ self.metrics
+
.record_records_retry(ready_write_batch.write_batch.record_count());
self.accumulator.re_enqueue(ready_write_batch);
}
@@ -842,6 +881,16 @@ impl Sender {
let mut next_delay_ms: u64 = 1;
loop {
+ // Sample buffer-pool gauges once per loop iteration. Cheap (three
+ // field reads) and naturally sampled. Java registers these as
+ // lazy gauge suppliers on the accumulator; the push model means
+ // we refresh them here on the sender's own cadence.
+ self.metrics.record_buffer_state(
+ self.accumulator.buffer_total_bytes(),
+ self.accumulator.buffer_available_bytes(),
+ self.accumulator.buffer_waiting_threads(),
+ );
+
// Spawn writer-ID init task if needed and not already running.
if init_futs.is_empty()
&& self.idempotence_manager.is_enabled()
@@ -1020,14 +1069,19 @@ impl WriteResponse for PutKvResponse {
mod tests {
use super::*;
use crate::client::WriteRecord;
- use crate::cluster::Cluster;
+ use crate::cluster::{Cluster, ServerType};
use crate::config::Config;
use crate::metadata::{PhysicalTablePath, TablePath};
- use crate::proto::{PbProduceLogRespForBucket, ProduceLogResponse};
+ use crate::proto::{
+ ApiVersionsResponse, PbApiVersion, PbProduceLogRespForBucket,
ProduceLogResponse,
+ };
use crate::row::{Datum, GenericRow};
use crate::rpc::FlussError;
- use crate::test_utils::{build_cluster_arc, build_table_info};
+ use crate::test_utils::{build_cluster_arc, build_cluster_arc_with_port,
build_table_info};
+ use prost::Message;
use std::collections::{HashMap, HashSet};
+ use tokio::io::{AsyncReadExt, AsyncWriteExt};
+ use tokio::net::TcpListener;
fn disabled_idempotence() -> Arc<IdempotenceManager> {
Arc::new(IdempotenceManager::new(false, 5))
@@ -1076,6 +1130,7 @@ mod tests {
1,
1,
idempotence,
+ Arc::new(crate::metrics::WriterMetrics::new()),
);
let (batch, _handle) =
@@ -1100,6 +1155,350 @@ mod tests {
Ok(())
}
+ #[test]
+ fn retriable_error_records_retry_metric() {
+ use metrics_util::debugging::{DebugValue, DebuggingRecorder};
+
+ let recorder = DebuggingRecorder::new();
+ let snapshotter = recorder.snapshotter();
+
+ let result: Result<()> = metrics::with_local_recorder(&recorder, || {
+ let table_path = Arc::new(TablePath::new("db".to_string(),
"tbl".to_string()));
+ let cluster = build_cluster_arc(table_path.as_ref(), 1, 1);
+ let metadata = Arc::new(Metadata::new_for_test(cluster.clone()));
+ let idempotence = disabled_idempotence();
+ let accumulator = Arc::new(RecordAccumulator::new(
+ Config::default(),
+ Arc::clone(&idempotence),
+ ));
+ // Construct the sender inside the recorder scope so its cached
+ // metric handles bind to the local recorder.
+ let sender = Sender::new(
+ metadata,
+ accumulator.clone(),
+ 1024 * 1024,
+ 1000,
+ 1,
+ 1,
+ idempotence,
+ Arc::new(crate::metrics::WriterMetrics::new()),
+ );
+
+ let (batch, _handle) =
+ build_ready_batch(accumulator.as_ref(), cluster.clone(),
table_path.clone())?;
+ let mut inflight = HashMap::new();
+ inflight.insert(1, vec![batch]);
+ sender.add_to_inflight_batches(&inflight);
+ let batch = inflight.remove(&1).unwrap().pop().unwrap();
+ let record_count = batch.write_batch.record_count();
+ assert_eq!(record_count, 1, "single-record batch expected");
+
+ sender.handle_write_batch_error(
+ batch,
+ FlussError::RequestTimeOut,
+ "timeout".to_string(),
+ )?;
+ Ok(())
+ });
+ result.expect("retry handling");
+
+ let entries = snapshotter.snapshot().into_vec();
+ let retry_total = entries.iter().find_map(|(key, _, _, val)| {
+ if key.key().name() == crate::metrics::WRITER_RECORDS_RETRY_TOTAL {
+ match val {
+ DebugValue::Counter(v) => Some(*v),
+ _ => None,
+ }
+ } else {
+ None
+ }
+ });
+ assert_eq!(retry_total, Some(1));
+ }
+
+ #[test]
+ fn record_request_batch_metrics_emits_per_batch_send_stats() {
+ use metrics_util::debugging::{DebugValue, DebuggingRecorder};
+
+ let recorder = DebuggingRecorder::new();
+ let snapshotter = recorder.snapshotter();
+
+ metrics::with_local_recorder(&recorder, || -> Result<()> {
+ let table_path = Arc::new(TablePath::new("db".to_string(),
"tbl".to_string()));
+ let cluster = build_cluster_arc(table_path.as_ref(), 1, 1);
+ let metadata = Arc::new(Metadata::new_for_test(cluster.clone()));
+ let idempotence = disabled_idempotence();
+ let accumulator = Arc::new(RecordAccumulator::new(
+ Config::default(),
+ Arc::clone(&idempotence),
+ ));
+ // Construct the sender inside the recorder scope so its cached
+ // metric handles bind to the local recorder.
+ let sender = Sender::new(
+ metadata,
+ accumulator.clone(),
+ 1024 * 1024,
+ 1000,
+ 1,
+ 1,
+ idempotence,
+ Arc::new(crate::metrics::WriterMetrics::new()),
+ );
+
+ // build_ready_batch drains the batch (sets drained_ms) and
appends a
+ // single record, mirroring the state batches are in when
+ // `send_write_request` records their metrics.
+ let (batch, _handle) =
+ build_ready_batch(accumulator.as_ref(), cluster.clone(),
table_path.clone())?;
+ assert_eq!(batch.write_batch.record_count(), 1);
+
+ sender.record_request_batch_metrics(std::slice::from_ref(&batch));
+ Ok(())
+ })
+ .expect("record batch metrics");
+
+ let entries = snapshotter.snapshot().into_vec();
+
+ let counter = |name: &str| {
+ entries.iter().find_map(|(key, _, _, val)| {
+ if key.key().name() == name {
+ match val {
+ DebugValue::Counter(v) => Some(*v),
+ _ => None,
+ }
+ } else {
+ None
+ }
+ })
+ };
+ let histogram = |name: &str| {
+ entries.iter().find_map(|(key, _, _, val)| {
+ if key.key().name() == name {
+ match val {
+ DebugValue::Histogram(v) => {
+ Some(v.iter().map(|f|
f.into_inner()).collect::<Vec<f64>>())
+ }
+ _ => None,
+ }
+ } else {
+ None
+ }
+ })
+ };
+
+ // One batch with a single record -> records counter is 1, bytes > 0.
+ assert_eq!(counter(crate::metrics::WRITER_RECORDS_SEND_TOTAL),
Some(1));
+ let bytes_send =
+ counter(crate::metrics::WRITER_BYTES_SEND_TOTAL).expect("bytes
send counter emitted");
+ assert!(
+ bytes_send > 0,
+ "expected non-zero bytes_send, got {bytes_send}"
+ );
+
+ // Each histogram observes exactly one sample for the single batch.
+ assert_eq!(
+ histogram(crate::metrics::WRITER_RECORDS_PER_BATCH),
+ Some(vec![1.0])
+ );
+ let bytes_per_batch =
+
histogram(crate::metrics::WRITER_BYTES_PER_BATCH).expect("bytes_per_batch
emitted");
+ assert_eq!(bytes_per_batch.len(), 1);
+ assert!(bytes_per_batch[0] > 0.0);
+ let queue_time =
+
histogram(crate::metrics::WRITER_BATCH_QUEUE_TIME_MS).expect("queue_time
emitted");
+ assert_eq!(queue_time.len(), 1);
+ assert!(queue_time[0] >= 0.0);
+ }
+
+ #[test]
+ fn send_write_request_error_still_records_attempted_send_metrics() {
+ use metrics_util::debugging::{DebugValue, DebuggingRecorder};
+
+ let recorder = DebuggingRecorder::new();
+ let snapshotter = recorder.snapshotter();
+
+ metrics::with_local_recorder(&recorder, || -> Result<()> {
+ let rt = tokio::runtime::Builder::new_current_thread()
+ .enable_all()
+ .build()
+ .expect("runtime");
+
+ rt.block_on(async {
+ let listener = TcpListener::bind("127.0.0.1:0")
+ .await
+ .expect("bind mock server");
+ let port = listener.local_addr().expect("listener
addr").port();
+
+ // Handshake-only mock server:
+ // respond to ApiVersions so connection setup succeeds, but do
+ // not advertise ProduceLog. That makes produce request fail
+ // during version resolution in `connection.request(...)`.
+ let server_task = tokio::spawn(async move {
+ let (mut stream, _) =
listener.accept().await.expect("accept");
+ let mut len_buf = [0u8; 4];
+ if stream.read_exact(&mut len_buf).await.is_err() {
+ return;
+ }
+ let len = i32::from_be_bytes(len_buf) as usize;
+ let mut payload = vec![0u8; len];
+ if stream.read_exact(&mut payload).await.is_err() {
+ return;
+ }
+
+ // Header layout: api_key(2) + api_version(2) +
request_id(4)
+ let request_id =
+ i32::from_be_bytes([payload[4], payload[5],
payload[6], payload[7]]);
+
+ let mut body = Vec::new();
+ ApiVersionsResponse {
+ api_versions: vec![PbApiVersion {
+ api_key: 1000, // ApiVersion
+ min_version: 0,
+ max_version: 0,
+ }],
+ server_type:
Some(ServerType::TabletServer.to_type_id()),
+ }
+ .encode(&mut body)
+ .expect("encode ApiVersionsResponse");
+
+ let mut resp = Vec::with_capacity(5 + body.len());
+ resp.push(0u8); // success response type
+ resp.extend_from_slice(&request_id.to_be_bytes());
+ resp.extend_from_slice(&body);
+ let resp_len = (resp.len() as i32).to_be_bytes();
+ let _ = stream.write_all(&resp_len).await;
+ let _ = stream.write_all(&resp).await;
+ let _ = stream.flush().await;
+ });
+
+ let table_path = Arc::new(TablePath::new("db".to_string(),
"tbl".to_string()));
+ let cluster = build_cluster_arc_with_port(table_path.as_ref(),
1, 1, port as u32);
+ let metadata =
Arc::new(Metadata::new_for_test(cluster.clone()));
+ let idempotence = disabled_idempotence();
+ let accumulator = Arc::new(RecordAccumulator::new(
+ Config::default(),
+ Arc::clone(&idempotence),
+ ));
+ let sender = Sender::new(
+ metadata,
+ accumulator.clone(),
+ 1024 * 1024,
+ 1000,
+ 1,
+ 1,
+ idempotence,
+ Arc::new(crate::metrics::WriterMetrics::new()),
+ );
+
+ let (batch, _handle) =
+ build_ready_batch(accumulator.as_ref(), cluster.clone(),
table_path)?;
+ sender.send_write_request(1, 1, vec![batch]).await?;
+ let _ = server_task.await;
+ Ok(())
+ })
+ })
+ .expect("sender attempted-send metrics");
+
+ let entries = snapshotter.snapshot().into_vec();
+
+ let send_latency_samples = entries.iter().find_map(|(key, _, _, val)| {
+ if key.key().name() == crate::metrics::WRITER_SEND_LATENCY_MS {
+ match val {
+ DebugValue::Histogram(v) => Some(v.len()),
+ _ => None,
+ }
+ } else {
+ None
+ }
+ });
+ assert_eq!(
+ send_latency_samples,
+ Some(1),
+ "send latency must be recorded even when request attempt fails"
+ );
+
+ let attempted_records = entries.iter().find_map(|(key, _, _, val)| {
+ if key.key().name() == crate::metrics::WRITER_RECORDS_SEND_TOTAL {
+ match val {
+ DebugValue::Counter(v) => Some(*v),
+ _ => None,
+ }
+ } else {
+ None
+ }
+ });
+ assert_eq!(
+ attempted_records,
+ Some(1),
+ "records_send_total should count attempted sends"
+ );
+ }
+
+ #[test]
+ fn
send_write_request_unknown_leader_still_records_attempted_send_metrics() {
+ use metrics_util::debugging::{DebugValue, DebuggingRecorder};
+
+ let recorder = DebuggingRecorder::new();
+ let snapshotter = recorder.snapshotter();
+
+ metrics::with_local_recorder(&recorder, || -> Result<()> {
+ let rt = tokio::runtime::Builder::new_current_thread()
+ .enable_all()
+ .build()
+ .expect("runtime");
+
+ rt.block_on(async {
+ let table_path = Arc::new(TablePath::new("db".to_string(),
"tbl".to_string()));
+ let cluster = build_cluster_arc(table_path.as_ref(), 1, 1);
+ let metadata =
Arc::new(Metadata::new_for_test(cluster.clone()));
+ let idempotence = disabled_idempotence();
+ let accumulator = Arc::new(RecordAccumulator::new(
+ Config::default(),
+ Arc::clone(&idempotence),
+ ));
+ let sender = Sender::new(
+ metadata,
+ accumulator.clone(),
+ 1024 * 1024,
+ 1000,
+ 1,
+ 1,
+ idempotence,
+ Arc::new(crate::metrics::WriterMetrics::new()),
+ );
+
+ let (batch, _handle) =
+ build_ready_batch(accumulator.as_ref(), cluster.clone(),
table_path)?;
+ // Destination 999 is absent from cluster metadata, so the send
+ // bails out with LeaderNotAvailableException before the batch
is
+ // serialized or dispatched. Metrics must still be recorded so
the
+ // count matches Java, which updates writer metrics over the
whole
+ // drained set regardless of send outcome.
+ sender.send_write_request(999, 1, vec![batch]).await?;
+ Ok(())
+ })
+ })
+ .expect("sender attempted-send metrics");
+
+ let entries = snapshotter.snapshot().into_vec();
+
+ let attempted_records = entries.iter().find_map(|(key, _, _, val)| {
+ if key.key().name() == crate::metrics::WRITER_RECORDS_SEND_TOTAL {
+ match val {
+ DebugValue::Counter(v) => Some(*v),
+ _ => None,
+ }
+ } else {
+ None
+ }
+ });
+ assert_eq!(
+ attempted_records,
+ Some(1),
+ "records_send_total must count batches dropped before send on
unknown leader"
+ );
+ }
+
#[tokio::test]
async fn handle_write_batch_error_fails() -> Result<()> {
let table_path = Arc::new(TablePath::new("db".to_string(),
"tbl".to_string()));
@@ -1118,6 +1517,7 @@ mod tests {
1,
0,
idempotence,
+ Arc::new(crate::metrics::WriterMetrics::new()),
);
let (batch, handle) = build_ready_batch(accumulator.as_ref(),
cluster.clone(), table_path)?;
@@ -1154,6 +1554,7 @@ mod tests {
1,
0,
idempotence,
+ Arc::new(crate::metrics::WriterMetrics::new()),
);
let (batch, handle) = build_ready_batch(accumulator.as_ref(), cluster,
table_path)?;
@@ -1198,6 +1599,7 @@ mod tests {
-1,
i32::MAX,
Arc::clone(&idempotence),
+ Arc::new(crate::metrics::WriterMetrics::new()),
);
// build_ready_batch drains the batch, which assigns seq=0 and adds
in-flight
@@ -1244,6 +1646,7 @@ mod tests {
-1,
0,
Arc::clone(&idempotence),
+ Arc::new(crate::metrics::WriterMetrics::new()),
);
// build_ready_batch drains the batch, which assigns seq=0 and adds
in-flight
@@ -1289,6 +1692,7 @@ mod tests {
-1,
i32::MAX,
Arc::clone(&idempotence),
+ Arc::new(crate::metrics::WriterMetrics::new()),
);
// build_ready_batch drains the batch, which assigns seq=0 and adds
in-flight
diff --git a/crates/fluss/src/client/write/writer_client.rs
b/crates/fluss/src/client/write/writer_client.rs
index ffdf96b1..73d91c7c 100644
--- a/crates/fluss/src/client/write/writer_client.rs
+++ b/crates/fluss/src/client/write/writer_client.rs
@@ -29,6 +29,7 @@ use crate::config::Config;
use crate::config::NoKeyAssigner;
use crate::error::{Error, Result};
use crate::metadata::{PhysicalTablePath, TableInfo};
+use crate::metrics::WriterMetrics;
use bytes::Bytes;
use dashmap::DashMap;
use log::warn;
@@ -66,6 +67,10 @@ impl WriterClient {
Arc::clone(&idempotence_manager),
));
+ // Writer metrics are emitted unlabeled (global per process). Resolve
the
+ // recorder once and share the cached handles with the sender.
+ let metrics = Arc::new(WriterMetrics::new());
+
let sender = Arc::new(Sender::new(
metadata.clone(),
accumulator.clone(),
@@ -74,6 +79,7 @@ impl WriterClient {
ack,
config.writer_retries,
Arc::clone(&idempotence_manager),
+ Arc::clone(&metrics),
));
let join_handle = tokio::spawn(async move {
diff --git a/crates/fluss/src/metrics.rs b/crates/fluss/src/metrics.rs
index 9f820ffe..e707f0e6 100644
--- a/crates/fluss/src/metrics.rs
+++ b/crates/fluss/src/metrics.rs
@@ -259,6 +259,138 @@ fn scanner_histogram(name: &'static str, database: &str,
table: &str) -> metrics
)
}
+// ---------------------------------------------------------------------------
+// Writer pipeline metrics
+//
+//
+// Java's `WriterMetricGroup` carries only the `client_id`
+// variable inherited from `ClientMetricGroup` -- no table or bucket label
+// (one series per client). The Rust `metrics` facade has no `client_id`
+// concept, so writer metrics are emitted UNLABELED (global per process).
+// TODO: A future `client.id` config option can attach a `client_id` label
without
+// breaking these series (it only splits the existing global series).
+//
+// Semantic deviations from Java:
+// * Java `sendLatencyMs` / `batchQueueTimeMs` are volatile-long gauges
+// (latest sample only); Rust uses histograms for full p50/p95/p99.
+// * Java `recordSendPerSecond` / `bytesSendPerSecond` /
`recordsRetryPerSecond`
+// are `MeterView` rates; Rust emits raw counters and lets the exporter
+// compute `rate()`.
+// ---------------------------------------------------------------------------
+
+/// Histogram: elapsed ms for each write request (ProduceLog / PutKv) round
+/// trip.
+pub const WRITER_SEND_LATENCY_MS: &str = "fluss.client.writer.send_latency_ms";
+
+/// Histogram: ms a batch spent queued in the accumulator (`drained_ms -
+/// create_ms`).
+pub const WRITER_BATCH_QUEUE_TIME_MS: &str =
"fluss.client.writer.batch_queue_time_ms";
+
+/// Counter: total records handed to the cluster across all sent batches.
+pub const WRITER_RECORDS_SEND_TOTAL: &str =
"fluss.client.writer.records_send.total";
+
+/// Counter: total serialized batch bytes sent.
+pub const WRITER_BYTES_SEND_TOTAL: &str =
"fluss.client.writer.bytes_send.total";
+
+/// Counter: total records re-enqueued for retry.
+pub const WRITER_RECORDS_RETRY_TOTAL: &str =
"fluss.client.writer.records_retry.total";
+
+/// Histogram: records per sent batch.
+pub const WRITER_RECORDS_PER_BATCH: &str =
"fluss.client.writer.records_per_batch";
+
+/// Histogram: serialized bytes per sent batch.
+pub const WRITER_BYTES_PER_BATCH: &str = "fluss.client.writer.bytes_per_batch";
+
+/// Gauge: total writer buffer memory in bytes (constant).
+pub const WRITER_BUFFER_TOTAL_BYTES: &str =
"fluss.client.writer.buffer_total_bytes";
+
+/// Gauge: currently-available writer buffer memory in bytes.
+pub const WRITER_BUFFER_AVAILABLE_BYTES: &str =
"fluss.client.writer.buffer_available_bytes";
+
+/// Gauge: number of producer threads blocked waiting for buffer memory --
+/// a high-signal backpressure indicator.
+pub const WRITER_BUFFER_WAITING_THREADS: &str =
"fluss.client.writer.buffer_waiting_threads";
+
+/// Cached, unlabeled writer-pipeline metric handles.
+///
+/// Constructed once per [`crate::client::write::WriterClient`] and shared
+/// (`Arc`) into the `Sender`.
+/// Like [`ScannerMetrics`], every cached handle is bound to whichever
+/// recorder is installed when [`Self::new`] runs. Construct it *after*
+/// installing the production recorder; in tests, construct it inside the
+/// `metrics::with_local_recorder(...)` closure. With no recorder installed,
+/// all `record_*` calls are zero-overhead no-ops.
+pub(crate) struct WriterMetrics {
+ send_latency_ms: metrics::Histogram,
+ batch_queue_time_ms: metrics::Histogram,
+ records_send_total: metrics::Counter,
+ bytes_send_total: metrics::Counter,
+ records_retry_total: metrics::Counter,
+ records_per_batch: metrics::Histogram,
+ bytes_per_batch: metrics::Histogram,
+ buffer_total_bytes: metrics::Gauge,
+ buffer_available_bytes: metrics::Gauge,
+ buffer_waiting_threads: metrics::Gauge,
+}
+
+impl WriterMetrics {
+ /// Build a fresh handle cache. Resolves the currently installed recorder
+ /// once per metric.
+ pub(crate) fn new() -> Self {
+ Self {
+ send_latency_ms: metrics::histogram!(WRITER_SEND_LATENCY_MS),
+ batch_queue_time_ms:
metrics::histogram!(WRITER_BATCH_QUEUE_TIME_MS),
+ records_send_total: metrics::counter!(WRITER_RECORDS_SEND_TOTAL),
+ bytes_send_total: metrics::counter!(WRITER_BYTES_SEND_TOTAL),
+ records_retry_total: metrics::counter!(WRITER_RECORDS_RETRY_TOTAL),
+ records_per_batch: metrics::histogram!(WRITER_RECORDS_PER_BATCH),
+ bytes_per_batch: metrics::histogram!(WRITER_BYTES_PER_BATCH),
+ buffer_total_bytes: metrics::gauge!(WRITER_BUFFER_TOTAL_BYTES),
+ buffer_available_bytes:
metrics::gauge!(WRITER_BUFFER_AVAILABLE_BYTES),
+ buffer_waiting_threads:
metrics::gauge!(WRITER_BUFFER_WAITING_THREADS),
+ }
+ }
+
+ pub(crate) fn record_send_latency_ms(&self, value: f64) {
+ self.send_latency_ms.record(value);
+ }
+
+ /// Record per-batch send statistics (records, bytes, queue time) for one
+ /// built+sent batch.
+ pub(crate) fn record_sent_batch(
+ &self,
+ record_count: i32,
+ batch_bytes: usize,
+ queue_time_ms: i64,
+ ) {
+ let records = record_count.max(0) as u64;
+ let bytes = batch_bytes as u64;
+ self.records_send_total.increment(records);
+ self.bytes_send_total.increment(bytes);
+ self.records_per_batch.record(record_count.max(0) as f64);
+ self.bytes_per_batch.record(batch_bytes as f64);
+ self.batch_queue_time_ms.record(queue_time_ms.max(0) as f64);
+ }
+
+ pub(crate) fn record_records_retry(&self, record_count: i32) {
+ self.records_retry_total
+ .increment(record_count.max(0) as u64);
+ }
+
+ /// Push the current buffer-pool gauges. Called once per sender poll-loop
+ /// iteration (Java registers these as lazy suppliers on the accumulator).
+ pub(crate) fn record_buffer_state(
+ &self,
+ total_bytes: usize,
+ available_bytes: usize,
+ waiting_threads: usize,
+ ) {
+ self.buffer_total_bytes.set(total_bytes as f64);
+ self.buffer_available_bytes.set(available_bytes as f64);
+ self.buffer_waiting_threads.set(waiting_threads as f64);
+ }
+}
+
/// Returns a label value for reportable API keys, matching Java's
/// `ConnectionMetrics.REPORT_API_KEYS` filter (`ProduceLog`, `FetchLog`,
/// `PutKv`, `Lookup`). Returns `None` for admin/metadata/auth calls to
@@ -651,4 +783,93 @@ mod tests {
Some(&3),
);
}
+
+ #[test]
+ fn writer_metrics_emit_all_writer_series() {
+ let recorder = DebuggingRecorder::new();
+ let snapshotter = recorder.snapshotter();
+
+ metrics::with_local_recorder(&recorder, || {
+ let m = WriterMetrics::new();
+ // Two sent batches: (3 records, 300 bytes, 12ms queue) and
+ // (2 records, 200 bytes, 8ms queue).
+ m.record_sent_batch(3, 300, 12);
+ m.record_sent_batch(2, 200, 8);
+ m.record_send_latency_ms(5.5);
+ m.record_records_retry(4);
+ m.record_buffer_state(64 * 1024 * 1024, 32 * 1024 * 1024, 2);
+ });
+
+ let snapshot = snapshotter.snapshot();
+ let entries: Vec<_> = snapshot.into_vec();
+
+ // Counters accumulate across both batches.
+ assert_eq!(find_counter!(entries, WRITER_RECORDS_SEND_TOTAL), Some(5));
+ assert_eq!(find_counter!(entries, WRITER_BYTES_SEND_TOTAL), Some(500));
+ assert_eq!(find_counter!(entries, WRITER_RECORDS_RETRY_TOTAL),
Some(4));
+
+ // Histograms capture one sample per batch.
+ assert_eq!(
+ find_histogram!(entries, WRITER_RECORDS_PER_BATCH),
+ Some(vec![3.0, 2.0])
+ );
+ assert_eq!(
+ find_histogram!(entries, WRITER_BYTES_PER_BATCH),
+ Some(vec![300.0, 200.0])
+ );
+ assert_eq!(
+ find_histogram!(entries, WRITER_BATCH_QUEUE_TIME_MS),
+ Some(vec![12.0, 8.0])
+ );
+ assert_eq!(
+ find_histogram!(entries, WRITER_SEND_LATENCY_MS),
+ Some(vec![5.5])
+ );
+
+ // Buffer gauges hold the latest pushed value.
+ assert_eq!(
+ find_gauge!(entries, WRITER_BUFFER_TOTAL_BYTES),
+ Some((64 * 1024 * 1024) as f64)
+ );
+ assert_eq!(
+ find_gauge!(entries, WRITER_BUFFER_AVAILABLE_BYTES),
+ Some((32 * 1024 * 1024) as f64)
+ );
+ assert_eq!(
+ find_gauge!(entries, WRITER_BUFFER_WAITING_THREADS),
+ Some(2.0)
+ );
+ }
+
+ /// Writer metrics carry no labels.
+ #[test]
+ fn writer_metrics_are_unlabeled() {
+ let recorder = DebuggingRecorder::new();
+ let snapshotter = recorder.snapshotter();
+
+ metrics::with_local_recorder(&recorder, || {
+ let m = WriterMetrics::new();
+ m.record_sent_batch(1, 10, 1);
+ });
+
+ let snapshot = snapshotter.snapshot();
+ let entries: Vec<_> = snapshot.into_vec();
+
+ let writer_entries: Vec<_> = entries
+ .iter()
+ .filter(|(key, _, _, _)|
key.key().name().starts_with("fluss.client.writer."))
+ .collect();
+ assert!(
+ !writer_entries.is_empty(),
+ "expected writer metrics to be emitted"
+ );
+ for (key, _, _, _) in writer_entries {
+ assert_eq!(
+ key.key().labels().count(),
+ 0,
+ "writer metric {} must be unlabeled",
+ key.key().name()
+ );
+ }
+ }
}
diff --git a/crates/fluss/src/record/arrow.rs b/crates/fluss/src/record/arrow.rs
index 5bd75924..33a0f6f4 100644
--- a/crates/fluss/src/record/arrow.rs
+++ b/crates/fluss/src/record/arrow.rs
@@ -561,6 +561,11 @@ impl MemoryLogRecordsArrowBuilder {
let estimated_body = self.estimated_compressed_size(body);
RECORD_BATCH_HEADER_SIZE + self.ipc_overhead + estimated_body
}
+
+ /// Number of records appended so far. Used for writer throughput metrics.
+ pub(crate) fn records_count(&self) -> i32 {
+ self.arrow_record_batch_builder.records_count()
+ }
}
/// Estimate the Arrow IPC overhead (metadata + body framing) for a given
schema.
diff --git a/crates/fluss/src/record/kv/kv_record_batch_builder.rs
b/crates/fluss/src/record/kv/kv_record_batch_builder.rs
index a2e472a3..f9500509 100644
--- a/crates/fluss/src/record/kv/kv_record_batch_builder.rs
+++ b/crates/fluss/src/record/kv/kv_record_batch_builder.rs
@@ -241,6 +241,11 @@ impl KvRecordBatchBuilder {
self.size_in_bytes
}
+ /// Number of records appended so far. Used for writer throughput metrics.
+ pub(crate) fn record_count(&self) -> i32 {
+ self.current_record_number
+ }
+
// ----------------------- Internal methods -------------------------------
/// Write the batch header.
diff --git a/crates/fluss/src/test_utils.rs b/crates/fluss/src/test_utils.rs
index f1e17e5f..ec192a50 100644
--- a/crates/fluss/src/test_utils.rs
+++ b/crates/fluss/src/test_utils.rs
@@ -37,7 +37,16 @@ pub(crate) fn build_table_info(table_path: TablePath,
table_id: i64, buckets: i3
}
pub(crate) fn build_cluster(table_path: &TablePath, table_id: i64, buckets:
i32) -> Cluster {
- let server = ServerNode::new(1, "127.0.0.1".to_string(), 9092,
ServerType::TabletServer);
+ build_cluster_with_port(table_path, table_id, buckets, 9092)
+}
+
+pub(crate) fn build_cluster_with_port(
+ table_path: &TablePath,
+ table_id: i64,
+ buckets: i32,
+ port: u32,
+) -> Cluster {
+ let server = ServerNode::new(1, "127.0.0.1".to_string(), port,
ServerType::TabletServer);
let mut servers = HashMap::new();
servers.insert(server.id(), server.clone());
@@ -89,6 +98,15 @@ pub(crate) fn build_cluster_arc(
Arc::new(build_cluster(table_path, table_id, buckets))
}
+pub(crate) fn build_cluster_arc_with_port(
+ table_path: &TablePath,
+ table_id: i64,
+ buckets: i32,
+ port: u32,
+) -> Arc<Cluster> {
+ Arc::new(build_cluster_with_port(table_path, table_id, buckets, port))
+}
+
/// Build an `Arc<ScannerMetrics>` for tests. Most callers don't install
/// a recorder, so the cached handles are no-ops; tests that *do* install
/// `metrics::with_local_recorder(...)` must call this *inside* the
diff --git a/website/docs/user-guide/rust/api-reference.md
b/website/docs/user-guide/rust/api-reference.md
index 285e84e7..7ef39577 100644
--- a/website/docs/user-guide/rust/api-reference.md
+++ b/website/docs/user-guide/rust/api-reference.md
@@ -128,6 +128,13 @@ Complete API reference for the Fluss Rust client.
| `fn append_arrow_batch(&self, batch: RecordBatch) ->
Result<WriteResultFuture>` | Append an Arrow RecordBatch |
| `async fn flush(&self) -> Result<()>`
| Flush all pending writes to the server |
+The writer pipeline emits `fluss.client.writer.*` metrics (send latency, batch
+queue time, records/bytes sent, retries, per-batch size, and buffer-pool
+gauges). Unlike scanner metrics, these are **unlabeled** (global per process),
+matching Java's single `WriterMetricGroup` per client (Java scopes only by
+`client_id`, which the Rust `metrics` facade has no concept of yet). The same
+series are shared by `AppendWriter` (log tables) and `UpsertWriter` (PK
tables).
+
## `TableScan<'a>`
| Method
| Description |
