GeezerVison is showing github.com — [connectors] Give transport output endpoints a health and metrics channel by bestouff · Pull Request #7639 · feldera/feldera
[connectors] Give transport output endpoints a health and metrics channel
#7639
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement . We’ll occasionally send you account related emails.
Already on GitHub?
Sign in
to your account
Uh oh!
There was an error while loading. Please reload this page.
Show file tree
Hide file tree
Changes from all commits
Commits
File filter
Filter by extension
.md
(2)
.rs
(23)
All 2 file types selected
Viewed files
Conversations
Failed to load comments.
Loading
Uh oh!
There was an error while loading. Please reload this page.
Jump to
Jump to file
Failed to load files.
Loading
Uh oh!
There was an error while loading. Please reload this page.
Diff view
Hide whitespace
Diff view
Hide whitespace
There are no files selected for viewing
Original file line number
Diff line number
Diff line change
Expand Up
@@ -6,6 +6,7 @@
mod canonical;
mod engine;
mod mock_input_consumer;
mod mock_output_callbacks;
mod mock_output_consumer;
mod updates;
mod wait;
Expand All
@@ -18,6 +19,7 @@ pub use engine::{
pub use mock_input_consumer::{
ErrorCallback, MockInputConsumer, MockInputConsumerState, MockInputParser, MockInputParserState,
};
pub use mock_output_callbacks::MockOutputCallbacks;
pub use mock_output_consumer::MockOutputConsumer;
pub use updates::{Updates, WEIGHT_COLUMN};
pub use wait::{async_wait, default_timeout_ms, wait};
Expand Down
Original file line number
Diff line number
Diff line change
@@ -0,0 +1,22 @@
use crate::metrics::ConnectorMetrics;
use crate::transport::OutputEndpointCallbacks;
use anyhow::Error as AnyError;
use feldera_types::adapter_stats::ConnectorHealth;
use std::sync::Arc;
use tracing::info;
/// Output endpoint callbacks for tests that drive an endpoint without a
/// controller: logs what the endpoint reports.
pub struct MockOutputCallbacks;
impl OutputEndpointCallbacks for MockOutputCallbacks {
fn error(&self, fatal: bool, error: AnyError, tag: Option<&'static str>) {
info!("output endpoint error: ({fatal:?}, {error:?}, {tag:?})");
}
fn update_connector_health(&self, health: ConnectorHealth) {
info!("output endpoint health: {health:?}");
}
fn set_custom_metrics(&self, _metrics: Arc<dyn ConnectorMetrics>) {}
}
Original file line number
Diff line number
Diff line change
Expand Up
@@ -1360,8 +1360,36 @@ where
.map_err(|e| anyhow::anyhow!("unable to parse checkpointed connector state (checkpointed state: {metadata}; parse error: {e})"))
}
#[doc(hidden)]
pub type AsyncErrorCallback = Box<dyn Fn(bool, AnyError, Option<&'static str>) + Send + Sync>;
/// The controller side of an output endpoint.
///
/// [`OutputEndpoint::connect`] hands the endpoint one of these. The endpoint
/// keeps it and calls it from any thread, for example from a delivery callback
/// that the transport library runs on its own thread, to report what happens
/// outside the [`OutputEndpoint`] methods. This is the output counterpart of
/// the reporting methods of [`InputConsumer`].
pub trait OutputEndpointCallbacks: Send + Sync {
/// Reports an asynchronous error, i.e., an error that happens outside the
/// context of the [`OutputEndpoint::push_buffer`] method. For instance, a
/// reliable message bus like Kafka may notify the endpoint about a failure
/// to deliver a previously sent message via an async callback. `fatal`
/// indicates an error that the endpoint cannot recover from. The optional
/// `tag` groups errors of one kind, e.g., for rate limiting.
fn error(&self, fatal: bool, error: AnyError, tag: Option<&'static str>);
/// Updates the health status of the connector.
///
/// An endpoint that retries an operation for a long time, such as a sink
/// that waits for a broker to come back, reports itself unhealthy while it
/// waits and healthy again once it succeeds. The status is visible in the
/// pipeline statistics.
fn update_connector_health(&self, health: ConnectorHealth);
/// Registers connector-specific metrics for Prometheus export.
///
/// An endpoint may call this once to provide an [`Arc<dyn ConnectorMetrics>`]
/// whose [`ConnectorMetrics::metrics`] will be polled on every scrape.
fn set_custom_metrics(&self, metrics: Arc<dyn ConnectorMetrics>);
}
/// Command handler API exposed by connectors.
///
Expand Down
Expand Up
@@ -1435,16 +1463,12 @@ pub trait OutputEndpoint: Send {
/// Finishes establishing the connection to the output endpoint.
///
/// If the endpoint encounters any errors during output, now or later, it
/// invokes `async_error_callback` to notify the client about asynchronous
/// errors, i.e., errors that happen outside the context of the
/// [`OutputEndpoint::push_buffer`] method. For instance, a reliable message
/// bus like Kafka may notify the endpoint about a failure to deliver a
/// previously sent message via an async callback. If the endpoint is unable
/// to handle this error, it must forward it to the client via the
/// `async_error_callback`. The first argument of the callback is a flag
/// that indicates a fatal error that the endpoint cannot recover from.
fn connect(&mut self, async_error_callback: AsyncErrorCallback) -> AnyResult<()>;
/// `callbacks` is the endpoint's channel back to the controller. If the
/// endpoint encounters an error that it cannot handle during output, now or
/// later, it must forward it through [`OutputEndpointCallbacks::error`].
/// The endpoint may also report its health and register its metrics
/// through the other methods, at connect time or later.
fn connect(&mut self, callbacks: Arc<dyn OutputEndpointCallbacks>) -> AnyResult<()>;
/// Maximum buffer size that this transport can transmit.
/// The encoder should not generate buffers exceeding this size.
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -65,7 +65,7 @@ use feldera_adapterlib::metrics::{ConnectorMetrics, ValueType};
use feldera_adapterlib::preprocess::PreprocessorWatchdog;
use feldera_adapterlib::soft_delete::SoftDeleteHandle;
use feldera_adapterlib::transport::{
CommandHandler, InputReader, OutputBatchType, Resume, Watermark,
CommandHandler, InputReader, OutputBatchType, OutputEndpointCallbacks, Resume, Watermark,
};
use feldera_adapterlib::utils::datafusion::{create_runtime_env, create_session_context};
use feldera_ir::LirCircuit;
Expand Down
Expand Up
@@ -8714,19 +8714,11 @@ impl ControllerInner {
let (encoder, command_handler) = if let Some(mut endpoint) = endpoint {
endpoint
.connect(Box::new(
move |fatal: bool, e: AnyError, error_tag: Option<&str>| {
if let Some(controller) = self_weak.upgrade() {
controller.output_transport_error(
endpoint_id,
&endpoint_name_str,
fatal,
e,
error_tag,
)
}
},
))
.connect(Arc::new(OutputCallbacks {
controller: self_weak,
endpoint_id,
endpoint_name: endpoint_name_str,
}))
.map_err(|e| ControllerError::output_transport_error(endpoint_name, true, e))?;
let command_handler = endpoint.command_handler();
Expand Down
Expand Up
@@ -10296,6 +10288,45 @@ impl InputConsumer for InputProbe {
}
}
/// The controller side of a transport output endpoint, handed to
/// [`OutputEndpoint::connect`].
///
/// Holds a weak reference, since the endpoint may call it from a transport
/// thread after the controller is gone.
struct OutputCallbacks {
controller: Weak<ControllerInner>,
endpoint_id: EndpointId,
endpoint_name: String,
}
impl OutputEndpointCallbacks for OutputCallbacks {
fn error(&self, fatal: bool, error: AnyError, tag: Option<&'static str>) {
if let Some(controller) = self.controller.upgrade() {
controller.output_transport_error(
self.endpoint_id,
&self.endpoint_name,
fatal,
error,
tag,
);
}
}
fn update_connector_health(&self, health: ConnectorHealth) {
if let Some(controller) = self.controller.upgrade() {
controller.update_output_connector_health(self.endpoint_id, health);
}
}
fn set_custom_metrics(&self, metrics: Arc<dyn ConnectorMetrics>) {
if let Some(controller) = self.controller.upgrade() {
controller
.status
.set_output_custom_metrics(self.endpoint_id, metrics);
}
}
}
/// An output probe inserted between the encoder and the output transport
/// endpoint to track stats.
struct OutputProbe {
Expand Down
Loading
Oops, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
Loading
Oops, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.