GeezerVison is showing github.com — [connectors] Give transport output endpoints a health and metrics cha… · feldera/feldera@0baad9f
File treeExpand file treeCollapse file tree Open diff view settings Expand file treeCollapse file tree Open diff view settings Collapse file Copy file name to clipboardExpand all lines: crates/adapterlib/src/test.rs +2Lines changed: 2 additions & 0 deletions | Original file line number | Diff line number | Diff line change |
|---|
| 6 | 6 | mod canonical; | 7 | 7 | mod engine; | 8 | 8 | mod mock_input_consumer; | | 9 | +mod mock_output_callbacks; | 9 | 10 | mod mock_output_consumer; | 10 | 11 | mod updates; | 11 | 12 | mod wait; | @@ -18,6 +19,7 @@ pub use engine::{ | 18 | 19 | pub use mock_input_consumer::{ | 19 | 20 | ErrorCallback, MockInputConsumer, MockInputConsumerState, MockInputParser, MockInputParserState, | 20 | 21 | }; | | 22 | +pub use mock_output_callbacks::MockOutputCallbacks; | 21 | 23 | pub use mock_output_consumer::MockOutputConsumer; | 22 | 24 | pub use updates::{Updates, WEIGHT_COLUMN}; | 23 | 25 | pub use wait::{async_wait, default_timeout_ms, wait}; | |
Collapse file Copy file name to clipboard +22Lines changed: 22 additions & 0 deletions | Original file line number | Diff line number | Diff line change |
|---|
| | 1 | +use crate::metrics::ConnectorMetrics; | | 2 | +use crate::transport::OutputEndpointCallbacks; | | 3 | +use anyhow::Error as AnyError; | | 4 | +use feldera_types::adapter_stats::ConnectorHealth; | | 5 | +use std::sync::Arc; | | 6 | +use tracing::info; | | 7 | + | | 8 | +/// Output endpoint callbacks for tests that drive an endpoint without a | | 9 | +/// controller: logs what the endpoint reports. | | 10 | +pub struct MockOutputCallbacks; | | 11 | + | | 12 | +impl OutputEndpointCallbacks for MockOutputCallbacks { | | 13 | + fn error(&self, fatal: bool, error: AnyError, tag: Option<&'static str>) { | | 14 | + info!("output endpoint error: ({fatal:?}, {error:?}, {tag:?})"); | | 15 | + } | | 16 | + | | 17 | + fn update_connector_health(&self, health: ConnectorHealth) { | | 18 | + info!("output endpoint health: {health:?}"); | | 19 | + } | | 20 | + | | 21 | + fn set_custom_metrics(&self, _metrics: Arc<dyn ConnectorMetrics>) {} | | 22 | +} |
Collapse file Copy file name to clipboardExpand all lines: crates/adapterlib/src/transport.rs +36-12Lines changed: 36 additions & 12 deletions | Original file line number | Diff line number | Diff line change |
|---|
@@ -1352,8 +1352,36 @@ where | 1352 | 1352 | .map_err(|e| anyhow::anyhow!("unable to parse checkpointed connector state (checkpointed state: {metadata}; parse error: {e})")) | 1353 | 1353 | } | 1354 | 1354 |
| 1355 | | -#[doc(hidden)] | 1356 | | -pub type AsyncErrorCallback = Box<dyn Fn(bool, AnyError, Option<&'static str>) + Send + Sync>; | | 1355 | +/// The controller side of an output endpoint. | | 1356 | +/// | | 1357 | +/// [`OutputEndpoint::connect`] hands the endpoint one of these. The endpoint | | 1358 | +/// keeps it and calls it from any thread, for example from a delivery callback | | 1359 | +/// that the transport library runs on its own thread, to report what happens | | 1360 | +/// outside the [`OutputEndpoint`] methods. This is the output counterpart of | | 1361 | +/// the reporting methods of [`InputConsumer`]. | | 1362 | +pub trait OutputEndpointCallbacks: Send + Sync { | | 1363 | + /// Reports an asynchronous error, i.e., an error that happens outside the | | 1364 | + /// context of the [`OutputEndpoint::push_buffer`] method. For instance, a | | 1365 | + /// reliable message bus like Kafka may notify the endpoint about a failure | | 1366 | + /// to deliver a previously sent message via an async callback. `fatal` | | 1367 | + /// indicates an error that the endpoint cannot recover from. The optional | | 1368 | + /// `tag` groups errors of one kind, e.g., for rate limiting. | | 1369 | + fn error(&self, fatal: bool, error: AnyError, tag: Option<&'static str>); | | 1370 | + | | 1371 | + /// Updates the health status of the connector. | | 1372 | + /// | | 1373 | + /// An endpoint that retries an operation for a long time, such as a sink | | 1374 | + /// that waits for a broker to come back, reports itself unhealthy while it | | 1375 | + /// waits and healthy again once it succeeds. The status is visible in the | | 1376 | + /// pipeline statistics. | | 1377 | + fn update_connector_health(&self, health: ConnectorHealth); | | 1378 | + | | 1379 | + /// Registers connector-specific metrics for Prometheus export. | | 1380 | + /// | | 1381 | + /// An endpoint may call this once to provide an [`Arc<dyn ConnectorMetrics>`] | | 1382 | + /// whose [`ConnectorMetrics::metrics`] will be polled on every scrape. | | 1383 | + fn set_custom_metrics(&self, metrics: Arc<dyn ConnectorMetrics>); | | 1384 | +} | 1357 | 1385 |
| 1358 | 1386 | /// Command handler API exposed by connectors. | 1359 | 1387 | /// | @@ -1427,16 +1455,12 @@ pub trait OutputEndpoint: Send { | 1427 | 1455 |
| 1428 | 1456 | /// Finishes establishing the connection to the output endpoint. | 1429 | 1457 | /// | 1430 | | - /// If the endpoint encounters any errors during output, now or later, it | 1431 | | - /// invokes `async_error_callback` to notify the client about asynchronous | 1432 | | - /// errors, i.e., errors that happen outside the context of the | 1433 | | - /// [`OutputEndpoint::push_buffer`] method. For instance, a reliable message | 1434 | | - /// bus like Kafka may notify the endpoint about a failure to deliver a | 1435 | | - /// previously sent message via an async callback. If the endpoint is unable | 1436 | | - /// to handle this error, it must forward it to the client via the | 1437 | | - /// `async_error_callback`. The first argument of the callback is a flag | 1438 | | - /// that indicates a fatal error that the endpoint cannot recover from. | 1439 | | - fn connect(&mut self, async_error_callback: AsyncErrorCallback) -> AnyResult<()>; | | 1458 | + /// `callbacks` is the endpoint's channel back to the controller. If the | | 1459 | + /// endpoint encounters an error that it cannot handle during output, now or | | 1460 | + /// later, it must forward it through [`OutputEndpointCallbacks::error`]. | | 1461 | + /// The endpoint may also report its health and register its metrics | | 1462 | + /// through the other methods, at connect time or later. | | 1463 | + fn connect(&mut self, callbacks: Arc<dyn OutputEndpointCallbacks>) -> AnyResult<()>; | 1440 | 1464 |
| 1441 | 1465 | /// Maximum buffer size that this transport can transmit. | 1442 | 1466 | /// The encoder should not generate buffers exceeding this size. | |
Collapse file Copy file name to clipboardExpand all lines: crates/connectors/src/controller.rs +45-14Lines changed: 45 additions & 14 deletions | Original file line number | Diff line number | Diff line change |
|---|
@@ -65,7 +65,7 @@ use feldera_adapterlib::metrics::{ConnectorMetrics, ValueType}; | 65 | 65 | use feldera_adapterlib::preprocess::PreprocessorWatchdog; | 66 | 66 | use feldera_adapterlib::soft_delete::SoftDeleteHandle; | 67 | 67 | use feldera_adapterlib::transport::{ | 68 | | - CommandHandler, InputReader, OutputBatchType, Resume, Watermark, | | 68 | + CommandHandler, InputReader, OutputBatchType, OutputEndpointCallbacks, Resume, Watermark, | 69 | 69 | }; | 70 | 70 | use feldera_adapterlib::utils::datafusion::{create_runtime_env, create_session_context}; | 71 | 71 | use feldera_ir::LirCircuit; | @@ -8694,19 +8694,11 @@ impl ControllerInner { | 8694 | 8694 |
| 8695 | 8695 | let (encoder, command_handler) = if let Some(mut endpoint) = endpoint { | 8696 | 8696 | endpoint | 8697 | | - .connect(Box::new( | 8698 | | - move |fatal: bool, e: AnyError, error_tag: Option<&str>| { | 8699 | | - if let Some(controller) = self_weak.upgrade() { | 8700 | | - controller.output_transport_error( | 8701 | | - endpoint_id, | 8702 | | - &endpoint_name_str, | 8703 | | - fatal, | 8704 | | - e, | 8705 | | - error_tag, | 8706 | | - ) | 8707 | | - } | 8708 | | - }, | 8709 | | - )) | | 8697 | + .connect(Arc::new(OutputCallbacks { | | 8698 | + controller: self_weak, | | 8699 | + endpoint_id, | | 8700 | + endpoint_name: endpoint_name_str, | | 8701 | + })) | 8710 | 8702 | .map_err(|e| ControllerError::output_transport_error(endpoint_name, true, e))?; | 8711 | 8703 |
| 8712 | 8704 | let command_handler = endpoint.command_handler(); | @@ -10268,6 +10260,45 @@ impl InputConsumer for InputProbe { | 10268 | 10260 | } | 10269 | 10261 | } | 10270 | 10262 |
| | 10263 | +/// The controller side of a transport output endpoint, handed to | | 10264 | +/// [`OutputEndpoint::connect`]. | | 10265 | +/// | | 10266 | +/// Holds a weak reference, since the endpoint may call it from a transport | | 10267 | +/// thread after the controller is gone. | | 10268 | +struct OutputCallbacks { | | 10269 | + controller: Weak<ControllerInner>, | | 10270 | + endpoint_id: EndpointId, | | 10271 | + endpoint_name: String, | | 10272 | +} | | 10273 | + | | 10274 | +impl OutputEndpointCallbacks for OutputCallbacks { | | 10275 | + fn error(&self, fatal: bool, error: AnyError, tag: Option<&'static str>) { | | 10276 | + if let Some(controller) = self.controller.upgrade() { | | 10277 | + controller.output_transport_error( | | 10278 | + self.endpoint_id, | | 10279 | + &self.endpoint_name, | | 10280 | + fatal, | | 10281 | + error, | | 10282 | + tag, | | 10283 | + ); | | 10284 | + } | | 10285 | + } | | 10286 | + | | 10287 | + fn update_connector_health(&self, health: ConnectorHealth) { | | 10288 | + if let Some(controller) = self.controller.upgrade() { | | 10289 | + controller.update_output_connector_health(self.endpoint_id, health); | | 10290 | + } | | 10291 | + } | | 10292 | + | | 10293 | + fn set_custom_metrics(&self, metrics: Arc<dyn ConnectorMetrics>) { | | 10294 | + if let Some(controller) = self.controller.upgrade() { | | 10295 | + controller | | 10296 | + .status | | 10297 | + .set_output_custom_metrics(self.endpoint_id, metrics); | | 10298 | + } | | 10299 | + } | | 10300 | +} | | 10301 | + | 10271 | 10302 | /// An output probe inserted between the encoder and the output transport | 10272 | 10303 | /// endpoint to track stats. | 10273 | 10304 | struct OutputProbe { | |
|