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 tokio::net::TcpStream;
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 uri = Uri::try_from(format!("http://{http_addr}{path_query}")).map_err(|e| {
126        (
127            StatusCode::INTERNAL_SERVER_ERROR,
128            format!("Invalid URI 'http://{http_addr}{path_query}': {e}"),
129        )
130    })?;
131
132    // Update request with new URI
133    *req.uri_mut() = uri.clone();
134
135    // Set Host header to target
136    if let Some(host) = uri.host() {
137        req.headers_mut().insert(
138            http::header::HOST,
139            HeaderValue::from_str(host).map_err(|e| {
140                (
141                    StatusCode::INTERNAL_SERVER_ERROR,
142                    format!("Invalid host header '{host}': {e}"),
143                )
144            })?,
145        );
146    }
147
148    // Connect to the target with timeout
149    let stream = tokio::time::timeout(CONNECT_TIMEOUT, TcpStream::connect(&*http_addr))
150        .await
151        .map_err(|_| {
152            (
153                StatusCode::GATEWAY_TIMEOUT,
154                format!("Connection timeout to {http_addr} after {CONNECT_TIMEOUT:?}"),
155            )
156        })?
157        .map_err(|e| {
158            (
159                StatusCode::BAD_GATEWAY,
160                format!("Failed to connect to {http_addr}: {e}"),
161            )
162        })?;
163
164    // Perform HTTP/1.1 handshake
165    let io = TokioIo::new(stream);
166    let (mut sender, conn) = hyper::client::conn::http1::handshake(io)
167        .await
168        .map_err(|e| {
169            (
170                StatusCode::BAD_GATEWAY,
171                format!("HTTP handshake with {http_addr} failed: {e}"),
172            )
173        })?;
174
175    // Spawn task to drive the connection
176    mz_ore::task::spawn(|| format!("Proxy to {uri}"), async move {
177        if let Err(e) = conn.await {
178            tracing::debug!("Connection to clusterd {http_addr} closed: {e}");
179        }
180    });
181
182    // Send the request with timeout
183    tokio::time::timeout(REQUEST_TIMEOUT, sender.send_request(req))
184        .await
185        .map_err(|_| {
186            (
187                StatusCode::GATEWAY_TIMEOUT,
188                format!(
189                    "Request timeout to clusterd {cluster_id}/{replica_id}/process/{process} after {REQUEST_TIMEOUT:?}"
190                ),
191            )
192        })?
193        .map(|r| r.into_response())
194        .map_err(|e| {
195            (
196                StatusCode::BAD_GATEWAY,
197                format!(
198                    "Error proxying to clusterd {cluster_id}/{replica_id}/process/{process}: {e}"
199                ),
200            )
201        })
202}
203
204/// Information about a replica for the clusters page template.
205pub struct ReplicaInfo {
206    pub cluster_name: String,
207    pub replica_name: String,
208    pub cluster_id: String,
209    pub replica_id: String,
210    pub process_count: usize,
211    pub process_indices: Vec<usize>,
212}
213
214#[derive(Template)]
215#[template(path = "clusters.html")]
216struct ClustersTemplate<'a> {
217    version: &'a str,
218    replicas: Vec<ReplicaInfo>,
219}
220
221/// Handler for the clusters overview page.
222///
223/// Displays a table of all cluster replicas with links to their HTTP endpoints.
224pub(crate) async fn handle_clusters(client: AuthedClient) -> impl IntoResponse {
225    // Look up names from the catalog
226    let catalog = client.client.catalog_snapshot("clusters_page").await;
227
228    let mut replicas = Vec::new();
229    for cluster in catalog.clusters() {
230        for replica in cluster.replicas() {
231            let process_count = replica.config.location.num_processes();
232            replicas.push(ReplicaInfo {
233                cluster_name: cluster.name.clone(),
234                replica_name: replica.name.clone(),
235                cluster_id: cluster.id.to_string(),
236                replica_id: replica.replica_id.to_string(),
237                process_count,
238                process_indices: (0..process_count).collect(),
239            });
240        }
241    }
242
243    let _ = catalog;
244
245    // Sort by system clusters first (cluster_id starts with 's'), then cluster name, then replica name
246    replicas.sort_by(|a, b| {
247        let a_is_system = a.cluster_id.starts_with('s');
248        let b_is_system = b.cluster_id.starts_with('s');
249        b_is_system
250            .cmp(&a_is_system)
251            .then_with(|| a.cluster_name.cmp(&b.cluster_name))
252            .then_with(|| a.replica_name.cmp(&b.replica_name))
253    });
254
255    mz_http_util::template_response(ClustersTemplate {
256        version: crate::BUILD_INFO.version,
257        replicas,
258    })
259}
mz_environmentd/http/cluster.rs

mz_environmentd/http/
cluster.rs
