[adapters] Add metrics for additional memory used by connectors.

blp · blp · commit d8b6c5772504 · 2025-11-14T00:13:45.000Z
Signed-off-by: Ben Pfaff &lt;blp@feldera.com&gt;
diff --git a/crates/adapterlib/src/format.rs b/crates/adapterlib/src/format.rs
@@ -302,6 +302,14 @@ pub trait OutputConsumer: Send {
         num_records: usize,
     );
     fn batch_end(&mut self);
+
+    /// Returns the approximate amount of memory used by the connector's
+    /// underlying implementation.  For the Kafka connectors, for example, this
+    /// is the amount of memory used by librdkafka.  Not all connectors use a
+    /// substantial amount of memory, so the default implementation returns 0.
+    fn memory(&self) -> usize {
+        0
+    }
 }
 
 /// The largest weight of a record that can be output using
diff --git a/crates/adapterlib/src/transport.rs b/crates/adapterlib/src/transport.rs
@@ -620,6 +620,14 @@ pub trait InputReader: Send + Sync {
     fn disconnect(&self) {
         self.request(InputReaderCommand::Disconnect);
     }
+
+    /// Returns the approximate amount of memory used by the connector's
+    /// underlying implementation.  For the Kafka connectors, for example, this
+    /// is the amount of memory used by librdkafka.  Not all connectors use a
+    /// substantial amount of memory, so the default implementation returns 0.
+    fn memory(&self) -> usize {
+        0
+    }
 }
 
 /// Position in an input stream, including the timestamp when the data was ingested
@@ -991,6 +999,14 @@ pub trait OutputEndpoint: Send {
 
     /// Whether this endpoint is [fault tolerant](crate#fault-tolerance).
     fn is_fault_tolerant(&self) -> bool;
+
+    /// Returns the approximate amount of memory used by the connector's
+    /// underlying implementation.  For the Kafka connectors, for example, this
+    /// is the amount of memory used by librdkafka.  Not all connectors use a
+    /// substantial amount of memory, so the default implementation returns 0.
+    fn memory(&self) -> usize {
+        0
+    }
 }
 
 /// An [UnboundedReceiver] wrapper for [InputReaderCommand] for fault-tolerant connectors.
diff --git a/crates/adapters/src/controller.rs b/crates/adapters/src/controller.rs
@@ -32,7 +32,7 @@ use crate::controller::sync::{
 };
 use crate::samply::SamplySpan;
 use crate::server::metrics::{
-    HistogramDiv, LabelStack, MetricsFormatter, MetricsWriter, ValueType,
+    HistogramDiv, LabelStack, MetricsFormatter, MetricsWriter, Value, ValueType,
 };
 use crate::server::{InitializationState, ServerState};
 use crate::transport::clock::now_endpoint_config;
@@ -1172,7 +1172,7 @@ impl Controller {
             &CHECKPOINT_SYNC_PULL_FAILURES,
         );
 
-        fn write_input_metric<F, M>(
+        fn write_input_metric<F, M, T>(
             metrics: &mut MetricsWriter<F>,
             labels: &LabelStack,
             status: &ControllerStatus,
@@ -1182,14 +1182,12 @@ impl Controller {
             func: M,
         ) where
             F: MetricsFormatter,
-            M: Fn(&InputEndpointMetrics) -> &AtomicU64,
+            M: Fn(&InputEndpointStatus) -> T,
+            T: Value,
         {
             metrics.values(name, help, value_type, |w| {
                 for input in status.input_status().values() {
-                    w.write_value(
-                        &labels.with("endpoint", &input.endpoint_name),
-                        func(&input.metrics),
-                    );
+                    w.write_value(&labels.with("endpoint", &input.endpoint_name), func(input));
                 }
             });
         }
@@ -1200,7 +1198,7 @@ impl Controller {
             "input_connector_bytes_total",
             "Total number of bytes received by an input connector.",
             ValueType::Counter,
-            |m| &m.total_bytes,
+            |s| s.metrics.total_bytes.load(Ordering::Relaxed),
         );
         write_input_metric(
             metrics,
@@ -1209,7 +1207,7 @@ impl Controller {
             "input_connector_records_total",
             "Total number of records received by an input connector.",
             ValueType::Counter,
-            |m| &m.total_records,
+            |m| m.metrics.total_records.load(Ordering::Relaxed),
         );
         write_input_metric(
             metrics,
@@ -1218,7 +1216,7 @@ impl Controller {
             "input_connector_buffered_records",
             "Amount of data currently buffered by an input connector, in records.",
             ValueType::Gauge,
-            |m| &m.buffered_records,
+            |s| s.metrics.buffered_records.load(Ordering::Relaxed),
         );
         write_input_metric(
             metrics,
@@ -1227,7 +1225,7 @@ impl Controller {
             "input_connector_buffered_records_bytes",
             "Amount of data currently buffered by an input connector, in bytes.",
             ValueType::Gauge,
-            |m| &m.buffered_bytes,
+            |s| s.metrics.buffered_bytes.load(Ordering::Relaxed),
         );
         write_input_metric(
             metrics,
@@ -1236,7 +1234,7 @@ impl Controller {
             "input_connector_errors_transport_total",
             "Total number of errors encountered by the input connector at the transport layer.",
             ValueType::Counter,
-            |m| &m.num_transport_errors,
+            |s| s.metrics.num_transport_errors.load(Ordering::Relaxed),
         );
         write_input_metric(
             metrics,
@@ -1245,7 +1243,16 @@ impl Controller {
             "input_connector_errors_parse_total",
             "Total number of errors encountered parsing records received by the input connector.",
             ValueType::Counter,
-            |m| &m.num_parse_errors,
+            |s| s.metrics.num_parse_errors.load(Ordering::Relaxed),
+        );
+        write_input_metric(
+            metrics,
+            labels,
+            status,
+            "input_connector_extra_memory_bytes",
+            "Additional memory used by an input connector beyond that used for buffered records.",
+            ValueType::Gauge,
+            |s| s.reader.as_ref().map_or(0, |reader| reader.memory()),
         );
 
         fn write_input_histogram<F, M>(
@@ -1378,6 +1385,15 @@ impl Controller {
             ValueType::Counter,
             |m| &m.num_encode_errors,
         );
+        write_output_metric(
+            metrics,
+            labels,
+            status,
+            "output_connector_extra_memory_bytes",
+            "Additional memory used by an output connector beyond that used for buffered records.",
+            ValueType::Gauge,
+            |m| &m.memory,
+        );
     }
 
     /// Execute a SQL query over materialized tables and views;
@@ -4649,6 +4665,9 @@ impl ControllerInner {
                 return;
             }
 
+            controller
+                .status
+                .update_output_memory(endpoint_id, encoder.consumer().memory());
             if output_buffer.flush_needed(&output_buffer_config) {
                 // One of the triggering conditions for flushing the output buffer is satisfied:
                 // go ahead and flush the buffer; we will check for more messages at the next iteration
@@ -5535,6 +5554,10 @@ impl OutputConsumer for OutputProbe {
             );
         })
     }
+
+    fn memory(&self) -> usize {
+        self.endpoint.memory()
+    }
 }
 
 /// An in-progress checkpoint.
diff --git a/crates/adapters/src/controller/stats.rs b/crates/adapters/src/controller/stats.rs
@@ -1004,6 +1004,15 @@ impl ControllerStatus {
         self.update_total_completed_records();
     }
 
+    pub fn update_output_memory(&self, endpoint_id: EndpointId, memory: usize) {
+        if let Some(endpoint_stats) = self.output_status().get(&endpoint_id) {
+            endpoint_stats
+                .metrics
+                .memory
+                .store(memory as u64, Ordering::Release);
+        }
+    }
+
     pub fn output_buffered_batches(&self, endpoint_id: EndpointId, total_processed_records: u64) {
         if let Some(endpoint_stats) = self.output_status().get(&endpoint_id) {
             endpoint_stats.output_buffered_batches(total_processed_records);
@@ -1952,6 +1961,10 @@ pub struct OutputEndpointMetrics {
     /// of this endpoint is equal to the output of the circuit after
     /// processing `total_processed_input_records` records.
     pub total_processed_input_records: AtomicU64,
+
+    /// Extra memory in use beyond that used for queuing records.  Not all
+    /// output connectors report this.
+    pub memory: AtomicU64,
 }
 
 impl OutputEndpointMetrics {
@@ -1970,6 +1983,7 @@ impl OutputEndpointMetrics {
             num_encode_errors: AtomicU64::new(initial_statistics.num_encode_errors),
             num_transport_errors: AtomicU64::new(initial_statistics.num_transport_errors),
             total_processed_input_records: AtomicU64::new(total_processed_input_records),
+            memory: AtomicU64::new(0),
         }
     }
 
@@ -1988,6 +2002,7 @@ impl OutputEndpointMetrics {
                 total_processed_input_records: this
                     .total_processed_input_records
                     .load(Ordering::Relaxed),
+                memory: this.memory.load(Ordering::Relaxed),
             }
         }
 
@@ -2020,6 +2035,7 @@ pub struct OutputEndpointMetricsSnapshot {
     pub num_encode_errors: u64,
     pub num_transport_errors: u64,
     pub total_processed_input_records: u64,
+    pub memory: u64,
 }
 
 /// Output endpoint status information.
diff --git a/crates/adapters/src/transport/kafka.rs b/crates/adapters/src/transport/kafka.rs
@@ -370,38 +370,47 @@ struct MemoryUseReporter {
     /// will take some time to reach what we hope is a steady state.
     start: Instant,
 
-    /// The memory use we last reported and when we reported it.
+    /// The most recently measured memory use, in bytes.
+    current: u64,
+
+    /// The peak memory use we last reported, in bytes, and when we reported it.
     ///
     /// This is a peak value: we only ever report a new value when the usage
     /// increases substantially from the previously reported value.
-    last_report: Option<(Instant, u64)>,
+    peak: Option<(Instant, u64)>,
 }
 
 impl MemoryUseReporter {
     fn new() -> Self {
         Self {
             start: Instant::now(),
-            last_report: None,
+            current: 0,
+            peak: None,
         }
     }
+    /// The most recent measured memory use in bytes.
+    fn current(&self) -> usize {
+        self.current as usize
+    }
     fn update(&mut self, statistics: &Statistics) {
         /// Minimum time before first report.
         const REPORT_DELAY: Duration = Duration::from_secs(60);
 
         /// Minimum amount of memory to report on.
         const MIN_MEMORY: u64 = 1024 * 1024;
 
-        if self.start.elapsed() < REPORT_DELAY {
-            return;
-        }
-
         let mut memory = 0;
         for topic in statistics.topics.values() {
             for partition in topic.partitions.values() {
                 memory += partition.msgq_bytes + partition.xmit_msgq_bytes + partition.fetchq_size;
             }
         }
-        match &self.last_report {
+        self.current = memory;
+        if self.start.elapsed() < REPORT_DELAY {
+            return;
+        }
+
+        match &self.peak {
             None if memory > MIN_MEMORY => {
                 info!(
                     "Buffered {} after {} seconds",
@@ -420,6 +429,6 @@ impl MemoryUseReporter {
             }
             _ => return,
         }
-        self.last_report = Some((Instant::now(), memory));
+        self.peak = Some((Instant::now(), memory));
     }
 }
diff --git a/crates/adapters/src/transport/kafka/ft/input.rs b/crates/adapters/src/transport/kafka/ft/input.rs
@@ -74,7 +74,7 @@ impl KafkaFtInputEndpoint {
 }
 
 struct KafkaFtInputReader {
-    _inner: Arc<KafkaFtInputReaderInner>,
+    inner: Arc<KafkaFtInputReaderInner>,
     command_sender: UnboundedSender<InputReaderCommand>,
     poller_thread: Thread,
 }
@@ -757,7 +757,7 @@ impl KafkaFtInputReader {
             .expect("failed to spawn Kafka input poller thread");
         let poller_thread = poller_handle.thread().clone();
         Ok(KafkaFtInputReader {
-            _inner: inner,
+            inner,
             command_sender,
             poller_thread,
         })
@@ -795,6 +795,15 @@ impl InputReader for KafkaFtInputReader {
     fn is_closed(&self) -> bool {
         self.command_sender.is_closed()
     }
+    fn memory(&self) -> usize {
+        self.inner
+            .kafka_consumer
+            .context()
+            .memory_use_reporter
+            .lock()
+            .unwrap()
+            .current()
+    }
 }
 
 impl Drop for KafkaFtInputReader {
diff --git a/crates/adapters/src/transport/kafka/ft/output.rs b/crates/adapters/src/transport/kafka/ft/output.rs
@@ -321,6 +321,15 @@ impl OutputEndpoint for KafkaOutputEndpoint {
     fn is_fault_tolerant(&self) -> bool {
         true
     }
+
+    fn memory(&self) -> usize {
+        self.kafka_producer
+            .context()
+            .memory_use_reporter
+            .lock()
+            .unwrap()
+            .current()
+    }
 }
 
 struct DataProducerContext {
diff --git a/crates/adapters/src/transport/kafka/nonft/output.rs b/crates/adapters/src/transport/kafka/nonft/output.rs
@@ -250,6 +250,15 @@ impl OutputEndpoint for KafkaOutputEndpoint {
     fn is_fault_tolerant(&self) -> bool {
         false
     }
+
+    fn memory(&self) -> usize {
+        self.kafka_producer
+            .context()
+            .memory_use_reporter
+            .lock()
+            .unwrap()
+            .current()
+    }
 }
 
 #[cfg(test)]
diff --git a/docs.feldera.com/docs/operations/metrics.md b/docs.feldera.com/docs/operations/metrics.md
@@ -139,6 +139,7 @@ invisible to users unless a pause or checkpoint happens mid-batch.
 | <a name='input_connector_completion_latency_seconds'>`input_connector_completion_latency_seconds`</a> |histogram | Time between when the connector receives new data and when the pipeline processes this data, computes output updates, and sends these updates to all output connectors, over the last 600 seconds or 10,000 samples. |
 | <a name='input_connector_errors_parse_total'>`input_connector_errors_parse_total`</a> |counter | Total number of errors encountered parsing records received by the input connector. |
 | <a name='input_connector_errors_transport_total'>`input_connector_errors_transport_total`</a> |counter | Total number of errors encountered by the input connector at the transport layer. |
+| <a name='input_connector_extra_memory_bytes'>`input_connector_extra_memory_bytes`</a> |gauge | Additional memory used by an input connector beyond that used for buffered records. |
 | <a name='input_connector_processing_latency_seconds'>`input_connector_processing_latency_seconds`</a> |histogram | Time between when the connector receives new data and when the pipeline processes this data and computes output updates, over the last 600 seconds or 10,000 samples. |
 | <a name='input_connector_records_total'>`input_connector_records_total`</a> |counter | Total number of records received by an input connector. |
 
@@ -158,5 +159,6 @@ These metrics accumulate across checkpoint and resume.
 | <a name='output_connector_bytes_total'>`output_connector_bytes_total`</a> |counter | Total number of bytes of records sent by the output connector. |
 | <a name='output_connector_errors_encode_total'>`output_connector_errors_encode_total`</a> |counter | Total number of errors encountered encoding records to send. |
 | <a name='output_connector_errors_transport_total'>`output_connector_errors_transport_total`</a> |counter | Total number of errors encountered at the transport layer sending records. |
+| <a name='output_connector_extra_memory_bytes'>`output_connector_extra_memory_bytes`</a> |gauge | Additional memory used by an output connector beyond that used for buffered records. |
 | <a name='output_connector_records_total'>`output_connector_records_total`</a> |counter | Total number of records sent by the output connector. |
 
