openlineage_client/
transport.rs

1//! Pluggable sink for OpenLineage events.
2//!
3//! Mirrors OpenLineage's own `Transport` SPI naming. The default
4//! [`NoopTransport`] is used when no endpoint is configured; [`ConsoleTransport`]
5//! is handy for development and tests.
6
7use async_trait::async_trait;
8
9use crate::event::RunEvent;
10
11/// Error returned when a [`Transport`] fails to send an event.
12#[derive(Debug, thiserror::Error)]
13pub enum TransportError {
14    /// The event could not be serialized to JSON.
15    #[error("failed to serialize lineage event: {0}")]
16    Serialize(#[from] serde_json::Error),
17    /// A transport-specific delivery failure (e.g. network or backend error).
18    #[error("transport error: {0}")]
19    Other(String),
20}
21
22/// A sink that delivers OpenLineage events to a backend.
23///
24/// The only required method is [`emit`](Transport::emit). [`emit_batch`](Transport::emit_batch)
25/// and [`flush`](Transport::flush) have default implementations; override them
26/// when the backend can deliver events in bulk or buffers internally.
27///
28/// Implementations must never apply back-pressure that could stall the host
29/// workload — emission happens on a background drain task, but a transport that
30/// blocks indefinitely (e.g. on a hung upstream) can still starve every queued
31/// event. Bound your own IO with a timeout.
32#[async_trait]
33pub trait Transport: std::fmt::Debug + Send + Sync {
34    /// Delivers a single OpenLineage event to the backend.
35    ///
36    /// # Errors
37    /// Returns a [`TransportError`] if the event cannot be serialized or
38    /// delivered.
39    async fn emit(&self, event: &RunEvent) -> Result<(), TransportError>;
40
41    /// Delivers a batch of events. The default implementation calls
42    /// [`emit`](Transport::emit) for each event in order, stopping at the first
43    /// error. Override when the backend has a bulk endpoint (e.g. a batch POST).
44    ///
45    /// # Errors
46    /// Returns a [`TransportError`] if any event cannot be serialized or
47    /// delivered.
48    async fn emit_batch(&self, events: &[RunEvent]) -> Result<(), TransportError> {
49        for event in events {
50            self.emit(event).await?;
51        }
52        Ok(())
53    }
54
55    /// Flushes any internally-buffered events, blocking until they are delivered
56    /// (or fail). Called by [`OpenLineageClient::shutdown`](crate::OpenLineageClient::shutdown)
57    /// after the drain queue empties. The default is a no-op — override it only
58    /// for transports that buffer beyond a single [`emit`](Transport::emit) call.
59    ///
60    /// # Errors
61    /// Returns a [`TransportError`] if buffered events cannot be delivered.
62    async fn flush(&self) -> Result<(), TransportError> {
63        Ok(())
64    }
65}
66
67/// Drops events. The safe default when lineage is not configured.
68#[derive(Debug, Default)]
69pub struct NoopTransport;
70
71#[async_trait]
72impl Transport for NoopTransport {
73    async fn emit(&self, _event: &RunEvent) -> Result<(), TransportError> {
74        Ok(())
75    }
76}
77
78/// Logs each event as pretty JSON via `tracing`. For development/tests.
79#[derive(Debug, Default)]
80pub struct ConsoleTransport;
81
82#[async_trait]
83impl Transport for ConsoleTransport {
84    async fn emit(&self, event: &RunEvent) -> Result<(), TransportError> {
85        let json = serde_json::to_string_pretty(event)?;
86        tracing::info!(target: "openlineage", "{json}");
87        Ok(())
88    }
89}
openlineage_client/transport.rs

openlineage_client/
transport.rs
