scalo 2.10.12

Self-regulating runtime for Rust data-plane services. Backpressure, load shedding and adaptive scaling are on by default.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90

// Project:   scalo
// File:      src/transport/vector_compat/sink.rs
// Purpose:   Vector gRPC sink compatibility wrapper
// Language:  Rust
//
// License:   Apache-2.0
// Copyright: (c) 2026 HYPERI PTY LIMITED

//! Vector-compatible gRPC sink.
//!
//! Sends events to a Vector source via the `vector.Vector/PushEvents` RPC.
//! Use this to push events to a downstream Vector pipeline.

use super::convert::json_to_event_wrapper;
use super::proto::vector;
use crate::transport::error::{TransportError, TransportResult};

/// Client that sends events to a Vector source via gRPC.
///
/// Converts JSON values to Vector's protobuf `EventWrapper` format
/// and sends them via `PushEvents`.
pub struct VectorCompatClient {
    client: vector::vector_client::VectorClient<tonic::transport::Channel>,
}

impl VectorCompatClient {
    /// Connect to a Vector source endpoint.
    ///
    /// Uses lazy connection -- the actual TCP connection is established
    /// on the first RPC call.
    ///
    /// # Errors
    ///
    /// Returns error if the endpoint URI is invalid.
    pub fn connect_lazy(endpoint: &str) -> TransportResult<Self> {
        // Bound the response decode size. `usize::MAX` contradicts the never-OOM
        // doctrine -- even a (trusted) Vector server response should not be able
        // to drive an unbounded allocation. 64 MiB is far above any real ack
        // response and matches the generous end of the transport size envelope.
        const MAX_DECODE_BYTES: usize = 64 * 1024 * 1024;

        let channel = tonic::transport::Channel::from_shared(endpoint.to_string())
            .map_err(|e| TransportError::Config(format!("invalid Vector endpoint: {e}")))?
            .connect_lazy();

        let client = vector::vector_client::VectorClient::new(channel)
            .max_decoding_message_size(MAX_DECODE_BYTES)
            .accept_compressed(tonic::codec::CompressionEncoding::Gzip)
            .send_compressed(tonic::codec::CompressionEncoding::Gzip);

        Ok(Self { client })
    }

    /// Send JSON values as Vector log events.
    ///
    /// Each JSON value is wrapped as a Vector `Log` event inside an `EventWrapper`.
    ///
    /// # Errors
    ///
    /// Returns error if the gRPC call fails.
    pub async fn send_events(&self, values: &[serde_json::Value]) -> TransportResult<()> {
        let events: Vec<_> = values.iter().map(json_to_event_wrapper).collect();

        let request = vector::PushEventsRequest { events };

        self.client
            .clone()
            .push_events(request)
            .await
            .map_err(|e| TransportError::Send(format!("Vector PushEvents failed: {e}")))?;

        Ok(())
    }

    /// Check if the remote Vector source is healthy.
    ///
    /// # Errors
    ///
    /// Returns error if the health check RPC fails.
    pub async fn health_check(&self) -> TransportResult<bool> {
        let response = self
            .client
            .clone()
            .health_check(vector::HealthCheckRequest {})
            .await
            .map_err(|e| TransportError::Connection(format!("Vector health check failed: {e}")))?;

        Ok(response.into_inner().status == vector::ServingStatus::Serving as i32)
    }
}