mz_environmentd/http/
cluster.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10//! HTTP proxy for cluster replica endpoints.
11//!
12//! This module provides an HTTP proxy that forwards requests from environmentd
13//! to clusterd internal HTTP endpoints (profiling, metrics, tracing). This allows
14//! accessing clusterd endpoints through environmentd's canonical HTTP port without
15//! requiring direct network access to the clusterd pods.
16
17use std::sync::Arc;
18use std::time::Duration;
19
20use askama::Template;
21use axum::Extension;
22use axum::body::Body;
23use axum::extract::Path;
24use axum::http::{Request, StatusCode};
25use axum::response::{IntoResponse, Response};
26use http::HeaderValue;
27use hyper::Uri;
28use hyper_util::rt::TokioIo;
29use mz_controller::ReplicaHttpLocator;
30use mz_controller_types::{ClusterId, ReplicaId};
31use mz_ore::netio::{SocketAddrType, Stream};
32
33use crate::http::AuthedClient;
34
35/// Connection timeout for proxied requests.
36const CONNECT_TIMEOUT: Duration = Duration::from_secs(5);
37/// Overall request timeout for proxied requests.
38const REQUEST_TIMEOUT: Duration = Duration::from_secs(60);
39
40/// Configuration for the cluster HTTP proxy.
41pub struct ClusterProxyConfig {
42    /// Handle to look up replica HTTP addresses.
43    locator: Arc<ReplicaHttpLocator>,
44}
45
46impl ClusterProxyConfig {
47    /// Creates a new `ClusterProxyConfig`.
48    pub fn new(locator: Arc<ReplicaHttpLocator>) -> Self {
49        Self { locator }
50    }
51}
52
53/// Proxy handler for cluster replica HTTP endpoints (root path).
54///
55/// Route: `/api/cluster/:cluster_id/replica/:replica_id/process/:process/`
56///
57/// This handler handles requests to the root of a clusterd process's HTTP endpoint.
58pub(crate) async fn handle_cluster_proxy_root(
59    Path((cluster_id, replica_id, process)): Path<(String, String, usize)>,
60    config: Extension<Arc<ClusterProxyConfig>>,
61    req: Request<Body>,
62) -> Result<Response, (StatusCode, String)> {
63    handle_cluster_proxy_inner(cluster_id, replica_id, process, "", config, req).await
64}
65
66/// Proxy handler for cluster replica HTTP endpoints.
67///
68/// Route: `/api/cluster/:cluster_id/replica/:replica_id/process/:process/*path`
69///
70/// This handler proxies HTTP requests to the internal HTTP endpoint of a specific
71/// clusterd process. Each replica can have multiple processes (based on `scale`),
72/// and each process has its own HTTP endpoint.
73pub(crate) async fn handle_cluster_proxy(
74    Path((cluster_id, replica_id, process, path)): Path<(String, String, usize, String)>,
75    config: Extension<Arc<ClusterProxyConfig>>,
76    req: Request<Body>,
77) -> Result<Response, (StatusCode, String)> {
78    handle_cluster_proxy_inner(cluster_id, replica_id, process, &path, config, req).await
79}
80
81async fn handle_cluster_proxy_inner(
82    cluster_id: String,
83    replica_id: String,
84    process: usize,
85    path: &str,
86    config: Extension<Arc<ClusterProxyConfig>>,
87    mut req: Request<Body>,
88) -> Result<Response, (StatusCode, String)> {
89    // Parse cluster ID
90    let cluster_id: ClusterId = cluster_id.parse().map_err(|e| {
91        (
92            StatusCode::BAD_REQUEST,
93            format!("Invalid cluster_id '{cluster_id}': {e}"),
94        )
95    })?;
96
97    // Parse replica ID
98    let replica_id: ReplicaId = replica_id.parse().map_err(|e| {
99        (
100            StatusCode::BAD_REQUEST,
101            format!("Invalid replica_id '{replica_id}': {e}"),
102        )
103    })?;
104
105    // Look up HTTP address for this replica and process
106    let http_addr = config
107        .locator
108        .get_http_addr(cluster_id, replica_id, process)
109        .ok_or_else(|| {
110            (
111                StatusCode::NOT_FOUND,
112                format!(
113                    "No HTTP address found for cluster {cluster_id}, replica {replica_id}, process {process}"
114                ),
115            )
116        })?;
117
118    // Build target URI, preserving query string if present
119    let path_query = if let Some(query) = req.uri().query() {
120        format!("/{path}?{query}")
121    } else {
122        format!("/{path}")
123    };
124
125    let authority = match SocketAddrType::guess(&http_addr) {
126        // UDS addresses aren't valid URI authorities, use a placeholder.
127        SocketAddrType::Unix => format!("cluster_{cluster_id}_replica_{replica_id}_{process}"),
128        SocketAddrType::Turmoil => http_addr.trim_start_matches("turmoil:").to_owned(),
129        SocketAddrType::Inet => http_addr.clone(),
130    };
131    let uri = Uri::try_from(format!("http://{authority}{path_query}")).map_err(|e| {
132        (
133            StatusCode::INTERNAL_SERVER_ERROR,
134            format!("Invalid URI 'http://{authority}{path_query}': {e}"),
135        )
136    })?;
137
138    // Update request with new URI
139    *req.uri_mut() = uri.clone();
140
141    // Set Host header to target
142    if let Some(host) = uri.host() {
143        req.headers_mut().insert(
144            http::header::HOST,
145            HeaderValue::from_str(host).map_err(|e| {
146                (
147                    StatusCode::INTERNAL_SERVER_ERROR,
148                    format!("Invalid host header '{host}': {e}"),
149                )
150            })?,
151        );
152    }
153
154    // Signal that this is a one-shot connection so the server shuts down
155    // gracefully instead of waiting for more requests.
156    req.headers_mut()
157        .insert(http::header::CONNECTION, HeaderValue::from_static("close"));
158
159    // Connect to the target with timeout
160    let stream = tokio::time::timeout(CONNECT_TIMEOUT, Stream::connect(http_addr.as_str()))
161        .await
162        .map_err(|_| {
163            (
164                StatusCode::GATEWAY_TIMEOUT,
165                format!("Connection timeout to {http_addr} after {CONNECT_TIMEOUT:?}"),
166            )
167        })?
168        .map_err(|e| {
169            (
170                StatusCode::BAD_GATEWAY,
171                format!("Failed to connect to {http_addr}: {e}"),
172            )
173        })?;
174
175    // Perform HTTP/1.1 handshake
176    let io = TokioIo::new(stream);
177    let (mut sender, conn) = hyper::client::conn::http1::handshake(io)
178        .await
179        .map_err(|e| {
180            (
181                StatusCode::BAD_GATEWAY,
182                format!("HTTP handshake with {http_addr} failed: {e}"),
183            )
184        })?;
185
186    // Spawn task to drive the connection
187    mz_ore::task::spawn(|| format!("Proxy to {uri}"), async move {
188        if let Err(e) = conn.await {
189            tracing::debug!("Connection to clusterd {http_addr} closed: {e}");
190        }
191    });
192
193    // Send the request with timeout
194    tokio::time::timeout(REQUEST_TIMEOUT, sender.send_request(req))
195        .await
196        .map_err(|_| {
197            (
198                StatusCode::GATEWAY_TIMEOUT,
199                format!(
200                    "Request timeout to clusterd {cluster_id}/{replica_id}/process/{process} after {REQUEST_TIMEOUT:?}"
201                ),
202            )
203        })?
204        .map(|r| r.into_response())
205        .map_err(|e| {
206            (
207                StatusCode::BAD_GATEWAY,
208                format!(
209                    "Error proxying to clusterd {cluster_id}/{replica_id}/process/{process}: {e}"
210                ),
211            )
212        })
213}
214
215/// Information about a replica for the clusters page template.
216pub struct ReplicaInfo {
217    pub cluster_name: String,
218    pub replica_name: String,
219    pub cluster_id: String,
220    pub replica_id: String,
221    pub process_count: usize,
222    pub process_indices: Vec<usize>,
223}
224
225#[derive(Template)]
226#[template(path = "clusters.html")]
227struct ClustersTemplate<'a> {
228    version: &'a str,
229    replicas: Vec<ReplicaInfo>,
230}
231
232/// Handler for the clusters overview page.
233///
234/// Displays a table of all cluster replicas with links to their HTTP endpoints.
235pub(crate) async fn handle_clusters(client: AuthedClient) -> impl IntoResponse {
236    // Look up names from the catalog
237    let catalog = client.client.catalog_snapshot("clusters_page").await;
238
239    let mut replicas = Vec::new();
240    for cluster in catalog.clusters() {
241        for replica in cluster.replicas() {
242            let process_count = replica.config.location.num_processes();
243            replicas.push(ReplicaInfo {
244                cluster_name: cluster.name.clone(),
245                replica_name: replica.name.clone(),
246                cluster_id: cluster.id.to_string(),
247                replica_id: replica.replica_id.to_string(),
248                process_count,
249                process_indices: (0..process_count).collect(),
250            });
251        }
252    }
253
254    let _ = catalog;
255
256    // Sort by system clusters first (cluster_id starts with 's'), then cluster name, then replica name
257    replicas.sort_by(|a, b| {
258        let a_is_system = a.cluster_id.starts_with('s');
259        let b_is_system = b.cluster_id.starts_with('s');
260        b_is_system
261            .cmp(&a_is_system)
262            .then_with(|| a.cluster_name.cmp(&b.cluster_name))
263            .then_with(|| a.replica_name.cmp(&b.replica_name))
264    });
265
266    mz_http_util::template_response(ClustersTemplate {
267        version: crate::BUILD_INFO.version,
268        replicas,
269    })
270}
mz_environmentd/http/cluster.rs

mz_environmentd/http/
cluster.rs
