praxis_protocol/tcp/
mod.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2024 Praxis Contributors
3
4//! Raw TCP/L4 bidirectional forwarding protocol.
5
6use std::{sync::Arc, time::Duration};
7
8use arc_swap::ArcSwap;
9use pingora_core::services::listening::Service;
10use praxis_core::{ProxyError, config::Config};
11use praxis_filter::{FilterPipeline, FilterRegistry};
12use tokio::sync::{Semaphore, watch};
13
14use crate::{ListenerPipelines, Protocol};
15
16/// TCP connection metrics (Prometheus counters and histograms).
17pub(crate) mod metrics;
18/// Bidirectional TCP proxy application.
19pub(crate) mod proxy;
20/// TLS configuration and listener grouping utilities.
21mod tls_setup;
22
23// -----------------------------------------------------------------------------
24// PingoraTcp
25// -----------------------------------------------------------------------------
26
27/// Pingora-backed raw TCP/L4 protocol implementation.
28///
29/// Groups TCP listeners by `(upstream address, idle timeout, max duration)`,
30/// creating one bidirectional forwarder per unique combination. Implements [`Protocol`].
31///
32/// [`Protocol`]: crate::Protocol
33pub struct PingoraTcp;
34
35#[expect(clippy::too_many_lines, reason = "linear registration with shutdown collection")]
36impl Protocol for PingoraTcp {
37    fn register(
38        self: Box<Self>,
39        server: &mut praxis_core::PingoraServerRuntime,
40        config: &Config,
41        pipelines: &ListenerPipelines,
42    ) -> Result<Vec<watch::Sender<bool>>, ProxyError> {
43        let groups = tls_setup::group_tcp_listeners(config);
44        tls_setup::validate_tcp_group_consistency(&groups)?;
45        #[expect(clippy::expect_used, reason = "empty pipeline is infallible")]
46        let fallback_pipeline = Arc::new(ArcSwap::from_pointee(
47            FilterPipeline::build(&mut [], &FilterRegistry::with_builtins()).expect("empty pipeline is valid"),
48        ));
49
50        let mut cert_watcher_shutdowns = Vec::new();
51
52        for ((upstream_opt, cluster_opt, timeout_ms, max_dur_secs), listeners) in groups {
53            let pipeline = listeners
54                .first()
55                .and_then(|l| pipelines.get(&l.name))
56                .map_or_else(|| Arc::clone(&fallback_pipeline), Arc::clone);
57
58            let session_timeout = timeout_ms.map(Duration::from_millis);
59            let max_duration = max_dur_secs.map(Duration::from_secs);
60            let service_name = match (upstream_opt.as_deref(), cluster_opt.as_deref()) {
61                (Some(addr), _) => format!("tcp-proxy:{addr}"),
62                (_, Some(cluster)) => format!("tcp-proxy:cluster:{cluster}"),
63                _ => "tcp-proxy:filter-routed".to_owned(),
64            };
65            let connection_semaphore = listeners
66                .first()
67                .and_then(|l| l.max_connections)
68                .map(|max| Arc::new(Semaphore::new(max as usize)));
69            // `from_shared` keeps labels as refcounted `Arc<str>`s: they
70            // are cloned per connection, and owned `String` labels would
71            // deep-copy on every clone.
72            let listener_names: std::collections::HashMap<String, ::metrics::SharedString> = listeners
73                .iter()
74                .map(|l| {
75                    (
76                        l.address.clone(),
77                        ::metrics::SharedString::from_shared(Arc::from(l.name.as_str())),
78                    )
79                })
80                .collect();
81            let default_listener_name = listeners.first().map_or_else(
82                || ::metrics::SharedString::const_str("unknown"),
83                |l| ::metrics::SharedString::from_shared(Arc::from(l.name.as_str())),
84            );
85            let app = proxy::PingoraTcpProxy::new(
86                upstream_opt.clone(),
87                cluster_opt.map(Arc::from),
88                pipeline,
89                session_timeout,
90                max_duration,
91                connection_semaphore,
92                config.insecure_options.allow_private_upstreams,
93                listener_names,
94                default_listener_name,
95            );
96            let mut service = Service::new(service_name, app);
97
98            cert_watcher_shutdowns.extend(tls_setup::register_tcp_listeners(
99                &mut service,
100                &listeners,
101                upstream_opt.as_deref(),
102            )?);
103            server.server_mut().add_service(service);
104        }
105
106        Ok(cert_watcher_shutdowns)
107    }
108}
praxis_protocol/tcp/mod.rs

praxis_protocol/tcp/
mod.rs
