praxis-proxy-protocol 0.6.0

HTTP, TCP, and protocol adapters for Praxis
Documentation 
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
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108

// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2024 Praxis Contributors

//! Raw TCP/L4 bidirectional forwarding protocol.

use std::{sync::Arc, time::Duration};

use arc_swap::ArcSwap;
use pingora_core::services::listening::Service;
use praxis_core::{ProxyError, config::Config};
use praxis_filter::{FilterPipeline, FilterRegistry};
use tokio::sync::{Semaphore, watch};

use crate::{ListenerPipelines, Protocol};

/// TCP connection metrics (Prometheus counters, gauges and histograms).
pub(crate) mod metrics;
/// Bidirectional TCP proxy application.
pub(crate) mod proxy;
/// TLS configuration and listener grouping utilities.
mod tls_setup;

// -----------------------------------------------------------------------------
// PingoraTcp
// -----------------------------------------------------------------------------

/// Pingora-backed raw TCP/L4 protocol implementation.
///
/// Groups TCP listeners by `(upstream address, idle timeout, max duration)`,
/// creating one bidirectional forwarder per unique combination. Implements [`Protocol`].
///
/// [`Protocol`]: crate::Protocol
pub struct PingoraTcp;

#[expect(clippy::too_many_lines, reason = "linear registration with shutdown collection")]
impl Protocol for PingoraTcp {
    fn register(
        self: Box<Self>,
        server: &mut praxis_core::PingoraServerRuntime,
        config: &Config,
        pipelines: &ListenerPipelines,
    ) -> Result<Vec<watch::Sender<bool>>, ProxyError> {
        let groups = tls_setup::group_tcp_listeners(config);
        tls_setup::validate_tcp_group_consistency(&groups)?;
        #[expect(clippy::expect_used, reason = "empty pipeline is infallible")]
        let fallback_pipeline = Arc::new(ArcSwap::from_pointee(
            FilterPipeline::build(&mut [], &FilterRegistry::with_builtins()).expect("empty pipeline is valid"),
        ));

        let mut cert_watcher_shutdowns = Vec::new();

        for ((upstream_opt, cluster_opt, timeout_ms, max_dur_secs), listeners) in groups {
            let pipeline = listeners
                .first()
                .and_then(|l| pipelines.get(&l.name))
                .map_or_else(|| Arc::clone(&fallback_pipeline), Arc::clone);

            let session_timeout = timeout_ms.map(Duration::from_millis);
            let max_duration = max_dur_secs.map(Duration::from_secs);
            let service_name = match (upstream_opt.as_deref(), cluster_opt.as_deref()) {
                (Some(addr), _) => format!("tcp-proxy:{addr}"),
                (_, Some(cluster)) => format!("tcp-proxy:cluster:{cluster}"),
                _ => "tcp-proxy:filter-routed".to_owned(),
            };
            let connection_semaphore = listeners
                .first()
                .and_then(|l| l.max_connections)
                .map(|max| Arc::new(Semaphore::new(max as usize)));
            // `from_shared` keeps labels as refcounted `Arc<str>`s: they
            // are cloned per connection, and owned `String` labels would
            // deep-copy on every clone.
            let listener_names: std::collections::HashMap<String, ::metrics::SharedString> = listeners
                .iter()
                .map(|l| {
                    (
                        l.address.clone(),
                        ::metrics::SharedString::from_shared(Arc::from(l.name.as_str())),
                    )
                })
                .collect();
            let default_listener_name = listeners.first().map_or_else(
                || ::metrics::SharedString::const_str("unknown"),
                |l| ::metrics::SharedString::from_shared(Arc::from(l.name.as_str())),
            );
            let app = proxy::PingoraTcpProxy::new(
                upstream_opt.clone(),
                cluster_opt.map(Arc::from),
                pipeline,
                session_timeout,
                max_duration,
                connection_semaphore,
                config.insecure_options.allow_private_upstreams,
                listener_names,
                default_listener_name,
            );
            let mut service = Service::new(service_name, app);

            cert_watcher_shutdowns.extend(tls_setup::register_tcp_listeners(
                &mut service,
                &listeners,
                upstream_opt.as_deref(),
            )?);
            server.server_mut().add_service(service);
        }

        Ok(cert_watcher_shutdowns)
    }
}