Struct WebIngestionService 

pub struct WebIngestionService { /* private fields */ }

Implementations

impl WebIngestionService

pub fn new(lake: DataLakeConnection) -> WebIngestionService

pub async fn from_env() -> Result<Arc<WebIngestionService>, Error>

Reads MICROMEGAS_SQL_CONNECTION_STRING and MICROMEGAS_OBJECT_STORE_URI, connects to the data lake, runs ingestion migrations, and returns a ready-to-use service.

pub fn insert_block( &self, body: Bytes, ) -> impl Future<Output = Result<(), IngestionServiceError>>

pub fn insert_block_typed( &self, block: Block, ) -> impl Future<Output = Result<(), IngestionServiceError>>

Inserts a block whose payload is already typed (no envelope round-trip on the caller side).

The caller hands us a fully-built Block; we CBOR-encode the payload envelope once, write it to object storage, and INSERT the row. Used by the OTLP adapter where constructing the CBOR Block envelope just so insert_block could decode it would be wasted work.

pub fn insert_stream( &self, body: Bytes, ) -> impl Future<Output = Result<(), IngestionServiceError>>

Registers a stream whose blocks will be ingested in the transit format.

pub fn register_otel_stream( &self, stream_id: Uuid, process_id: Uuid, tags: Vec<String>, format: &str, ) -> impl Future<Output = Result<(), IngestionServiceError>>

Registers a stream produced by an OTLP ingestion path.

dependencies_metadata and objects_metadata are filled with the CBOR sentinel for an empty Vec<UserDefinedType> so legacy decode sites continue to work. format distinguishes per-block dispatch downstream (e.g. "otlp/v1/logs"). Stream properties are always empty for OTel — scope and per-event attrs live on individual rows during materialization, not on the stream.

Hack: piggybacking OTLP onto the transit-shaped streams row (with empty metadata sentinels) is expedient for two formats but won’t scale. To support more formats cleanly, dependencies_metadata, objects_metadata, and format should be merged into a single per-format payload column.

pub fn insert_process( &self, body: Bytes, ) -> impl Future<Output = Result<(), IngestionServiceError>>

pub fn register_otel_process( &self, process_id: Uuid, exe: String, username: String, computer: String, distro: String, cpu_brand: String, tsc_frequency: i64, start_time: DateTime<Utc>, start_ticks: i64, properties: Vec<Property>, ) -> impl Future<Output = Result<(), IngestionServiceError>>

Registers a process originating from OTLP. Idempotent via ON CONFLICT DO NOTHING.

realname is set equal to username (OTel has no separate “real name” concept). parent_process_id is always NULL — OTel has no parent-process model. insert_time is the server wall clock, matching the existing insert_process path.

Trait Implementations

impl Clone for WebIngestionService

fn clone(&self) -> WebIngestionService

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.