|
| 1 | +// Copyright The OpenTelemetry Authors |
| 2 | +// SPDX-License-Identifier: Apache-2.0 |
| 3 | + |
| 4 | +//! Shared helpers for gRPC receivers. |
| 5 | +//! |
| 6 | +//! Request lifecycle (see `otlp::server` for the per-signal services): |
| 7 | +//! - decode: gRPC body stays as OTLP bytes, wrapped into `OtapPdata` |
| 8 | +//! - subscribe: when `wait_for_result` is enabled, a slot is allocated and calldata recorded |
| 9 | +//! - send: payload is forwarded into the pipeline |
| 10 | +//! - wait (optional): task awaits an ACK or NACK routed by the subscription maps |
| 11 | +//! - respond: ACK/NACK is mapped back to the waiting gRPC request |
| 12 | +
|
| 13 | +use crate::otap_grpc::otlp::server_new::{AckSlot, RouteResponse}; |
| 14 | +use crate::otap_grpc::server_settings::GrpcServerSettings; |
| 15 | +use crate::pdata::OtapPdata; |
| 16 | +use otap_df_config::SignalType; |
| 17 | +use otap_df_engine::control::{AckMsg, NackMsg}; |
| 18 | +use tonic::transport::Server; |
| 19 | + |
| 20 | +/// Aggregates the per-signal ACK subscription maps that let us route responses back to callers. |
| 21 | +#[derive(Clone, Default)] |
| 22 | +pub struct AckRegistry { |
| 23 | + /// Subscription map for log acknowledgements. |
| 24 | + pub logs: Option<AckSlot>, |
| 25 | + /// Subscription map for metric acknowledgements. |
| 26 | + pub metrics: Option<AckSlot>, |
| 27 | + /// Subscription map for trace acknowledgements. |
| 28 | + pub traces: Option<AckSlot>, |
| 29 | +} |
| 30 | + |
| 31 | +impl AckRegistry { |
| 32 | + /// Creates a new bundle of optional subscription maps. |
| 33 | + #[must_use] |
| 34 | + pub fn new(logs: Option<AckSlot>, metrics: Option<AckSlot>, traces: Option<AckSlot>) -> Self { |
| 35 | + Self { |
| 36 | + logs, |
| 37 | + metrics, |
| 38 | + traces, |
| 39 | + } |
| 40 | + } |
| 41 | +} |
| 42 | + |
| 43 | +/// Routes an Ack message to the appropriate signal's subscription map. |
| 44 | +#[must_use] |
| 45 | +pub fn route_ack_response(states: &AckRegistry, ack: AckMsg<OtapPdata>) -> RouteResponse { |
| 46 | + let calldata = ack.calldata; |
| 47 | + let resp = Ok(()); |
| 48 | + let state = match ack.accepted.signal_type() { |
| 49 | + SignalType::Logs => states.logs.as_ref(), |
| 50 | + SignalType::Metrics => states.metrics.as_ref(), |
| 51 | + SignalType::Traces => states.traces.as_ref(), |
| 52 | + }; |
| 53 | + |
| 54 | + state |
| 55 | + .map(|s| s.route_response(calldata, resp)) |
| 56 | + .unwrap_or(RouteResponse::None) |
| 57 | +} |
| 58 | + |
| 59 | +/// Routes a Nack message to the appropriate shared state. |
| 60 | +#[must_use] |
| 61 | +pub fn route_nack_response(states: &AckRegistry, mut nack: NackMsg<OtapPdata>) -> RouteResponse { |
| 62 | + let calldata = std::mem::take(&mut nack.calldata); |
| 63 | + let signal_type = nack.refused.signal_type(); |
| 64 | + let resp = Err(nack); |
| 65 | + let state = match signal_type { |
| 66 | + SignalType::Logs => states.logs.as_ref(), |
| 67 | + SignalType::Metrics => states.metrics.as_ref(), |
| 68 | + SignalType::Traces => states.traces.as_ref(), |
| 69 | + }; |
| 70 | + |
| 71 | + state |
| 72 | + .map(|s| s.route_response(calldata, resp)) |
| 73 | + .unwrap_or(RouteResponse::None) |
| 74 | +} |
| 75 | + |
| 76 | +/// Handles the outcome from routing an Ack/Nack response. |
| 77 | +pub fn handle_route_response<T, F, G>( |
| 78 | + resp: RouteResponse, |
| 79 | + state: &mut T, |
| 80 | + mut on_sent: F, |
| 81 | + mut on_expired_or_invalid: G, |
| 82 | +) where |
| 83 | + F: FnMut(&mut T), |
| 84 | + G: FnMut(&mut T), |
| 85 | +{ |
| 86 | + match resp { |
| 87 | + RouteResponse::Sent => on_sent(state), |
| 88 | + RouteResponse::Expired | RouteResponse::Invalid => on_expired_or_invalid(state), |
| 89 | + RouteResponse::None => {} |
| 90 | + } |
| 91 | +} |
| 92 | + |
| 93 | +/// Tunes the maximum concurrent requests relative to the downstream capacity (channel connecting |
| 94 | +/// the receiver to the rest of the pipeline). |
| 95 | +pub fn tune_max_concurrent_requests(config: &mut GrpcServerSettings, downstream_capacity: usize) { |
| 96 | + // Fall back to the downstream channel capacity when it is tighter than the user setting. |
| 97 | + let safe_capacity = downstream_capacity.max(1); |
| 98 | + if config.max_concurrent_requests == 0 || config.max_concurrent_requests > safe_capacity { |
| 99 | + config.max_concurrent_requests = safe_capacity; |
| 100 | + } |
| 101 | +} |
| 102 | + |
| 103 | +/// Applies the shared server tuning options to a tonic server builder. |
| 104 | +pub fn apply_server_tuning<L>(builder: Server<L>, config: &GrpcServerSettings) -> Server<L> { |
| 105 | + let transport_limit = config |
| 106 | + .transport_concurrency_limit |
| 107 | + .and_then(|limit| if limit == 0 { None } else { Some(limit) }) |
| 108 | + .unwrap_or(config.max_concurrent_requests) |
| 109 | + .max(1); |
| 110 | + |
| 111 | + let fallback_streams = config.max_concurrent_requests.min(u32::MAX as usize) as u32; |
| 112 | + |
| 113 | + let mut builder = builder |
| 114 | + .concurrency_limit_per_connection(transport_limit) |
| 115 | + .load_shed(config.load_shed) |
| 116 | + .initial_stream_window_size(config.initial_stream_window_size) |
| 117 | + .initial_connection_window_size(config.initial_connection_window_size) |
| 118 | + .max_frame_size(config.max_frame_size) |
| 119 | + .http2_adaptive_window(Some(config.http2_adaptive_window)) |
| 120 | + .http2_keepalive_interval(config.http2_keepalive_interval) |
| 121 | + .http2_keepalive_timeout(config.http2_keepalive_timeout); |
| 122 | + |
| 123 | + let mut max_concurrent_streams = config |
| 124 | + .max_concurrent_streams |
| 125 | + .map(|value| if value == 0 { fallback_streams } else { value }) |
| 126 | + .unwrap_or(fallback_streams); |
| 127 | + if max_concurrent_streams == 0 { |
| 128 | + max_concurrent_streams = 1; |
| 129 | + } |
| 130 | + builder = builder.max_concurrent_streams(Some(max_concurrent_streams)); |
| 131 | + |
| 132 | + // Apply timeout if configured |
| 133 | + if let Some(timeout) = config.timeout { |
| 134 | + builder = builder.timeout(timeout); |
| 135 | + } |
| 136 | + |
| 137 | + builder |
| 138 | +} |
0 commit comments