rust/neon
1
0
Fork 0
mirror of https://github.com/neondatabase/neon.git synced 2026-01-15 01:12:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

remove one measuredstream layer of indirection

Browse Source
This commit is contained in:
Conrad Ludgate
2025-05-19 16:21:32 +01:00
parent 68d561664b
commit 008cd84e7b
1 changed files with 8 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
proxy/src/proxy/passthrough.rs
View File

@@ -16,7 +16,7 @@ use crate::usage_metrics::{Ids, MetricCounterRecorder, USAGE_METRICS};
#[tracing::instrument(level = "debug", skip_all)]
pub(crate) async fn proxy_pass(
client: impl AsyncRead + AsyncWrite + Unpin,
compute: impl AsyncRead + AsyncWrite + Unpin,
mut compute: impl AsyncRead + AsyncWrite + Unpin,
aux: MetricsAuxInfo,
private_link_id: Option<SmolStr>,
) -> Result<(), ErrorSource> {
@@ -29,24 +29,16 @@ pub(crate) async fn proxy_pass(
let metrics = &Metrics::get().proxy.io_bytes;
let m_sent = metrics.with_labels(Direction::Tx);
let m_recv = metrics.with_labels(Direction::Rx);
let mut client = MeasuredStream::new(
client,
|_| {},
|cnt| {
// Number of bytes we sent to the client (outbound).
metrics.get_metric(m_sent).inc_by(cnt as u64);
usage_tx.record_egress(cnt as u64);
|bytes_read| {
metrics.get_metric(m_recv).inc_by(bytes_read as u64);
usage_tx.record_ingress(bytes_read as u64);
},
);
let m_recv = metrics.with_labels(Direction::Rx);
let mut compute = MeasuredStream::new(
compute,
|_| {},
|cnt| {
// Number of bytes the client sent to the compute node (inbound).
metrics.get_metric(m_recv).inc_by(cnt as u64);
usage_tx.record_ingress(cnt as u64);
|bytes_flushed| {
metrics.get_metric(m_sent).inc_by(bytes_flushed as u64);
usage_tx.record_egress(bytes_flushed as u64);
},
);