helix_driver_host/network/
http.rs

1use bytes::Bytes;
2use helix_core::effect::{HttpRequest, HttpResponse};
3use helix_core::ports::HttpRequester;
4use helix_core::PortError;
5use reqwest::header::USER_AGENT;
6use reqwest::{Client, Method};
7use std::collections::BTreeMap;
8use std::sync::atomic::{AtomicUsize, Ordering};
9use std::sync::Arc;
10use std::time::Instant;
11
12use crate::metrics::{
13    AsyncMetricSink, LabelKey, MetricEvent, MetricId, MetricLabels, NoopMetricSink,
14};
15
16use super::network_util::{
17    header_name, header_value, headers_to_strings, resolve_http_url, validate_url,
18};
19use super::{HostHeaderRegistry, HostNetworkConfig};
20
21/// Request shape after shared config/header resolution.
22#[derive(Debug, Clone, PartialEq, Eq)]
23pub struct PreparedHttpRequest {
24    pub method: String,
25    pub url: String,
26    pub headers: BTreeMap<String, String>,
27    pub body: Option<Bytes>,
28}
29
30/// Shared reqwest-backed HTTP port implementation.
31#[derive(Clone)]
32pub struct SharedHttpClient {
33    client: Client,
34    config: HostNetworkConfig,
35    headers: HostHeaderRegistry,
36    metrics: Arc<dyn AsyncMetricSink>,
37    inflight: Arc<AtomicUsize>,
38}
39
40impl SharedHttpClient {
41    pub fn new(config: HostNetworkConfig) -> Result<Self, PortError> {
42        Self::with_registry(config, HostHeaderRegistry::default())
43    }
44
45    pub fn with_registry(
46        config: HostNetworkConfig,
47        headers: HostHeaderRegistry,
48    ) -> Result<Self, PortError> {
49        validate_url(&config.api_base_url, "api_base_url")?;
50        if !config.default_api_base_url.is_empty() {
51            validate_url(&config.default_api_base_url, "default_api_base_url")?;
52        }
53        let mut builder = Client::builder().timeout(config.timeout);
54        if let Some(user_agent) = &config.user_agent {
55            builder = builder.user_agent(user_agent);
56        }
57        let client = builder
58            .build()
59            .map_err(|e| PortError::Http(format!("reqwest client build error: {e}")))?;
60        Ok(Self {
61            client,
62            config,
63            headers,
64            metrics: Arc::new(NoopMetricSink),
65            inflight: Arc::new(AtomicUsize::new(0)),
66        })
67    }
68
69    pub fn headers(&self) -> HostHeaderRegistry {
70        self.headers.clone()
71    }
72
73    pub fn with_metric_sink(mut self, metrics: Arc<dyn AsyncMetricSink>) -> Self {
74        self.metrics = metrics;
75        self
76    }
77
78    pub async fn prepare_for_test(
79        &self,
80        req: HttpRequest,
81    ) -> Result<PreparedHttpRequest, PortError> {
82        self.prepare(req).await
83    }
84
85    async fn prepare(&self, req: HttpRequest) -> Result<PreparedHttpRequest, PortError> {
86        let method = Method::from_bytes(req.method.as_bytes())
87            .map_err(|e| PortError::Http(format!("invalid method '{}': {e}", req.method)))?;
88        let url = resolve_http_url(&self.config.api_base_url, &req.url)?;
89        let mut headers = self.headers.snapshot().await;
90        if let Some(user_agent) = &self.config.user_agent {
91            headers.insert(USER_AGENT, header_value("user-agent", user_agent)?);
92        }
93        for (name, value) in &req.headers {
94            let name = header_name(name)?;
95            let value = header_value(name.as_str(), value)?;
96            headers.insert(name, value);
97        }
98        Ok(PreparedHttpRequest {
99            method: method.as_str().to_string(),
100            url,
101            headers: headers_to_strings(&headers),
102            body: req.body,
103        })
104    }
105
106    fn classify_error(e: &reqwest::Error) -> PortError {
107        if e.is_timeout() {
108            PortError::Transport(format!("timeout: {e}"))
109        } else if e.is_connect() || e.is_request() {
110            PortError::Transport(format!("network: {e}"))
111        } else {
112            PortError::Http(e.to_string())
113        }
114    }
115
116    async fn execute(&self, req: HttpRequest) -> Result<HttpResponse, PortError> {
117        let prepared = self.prepare(req).await?;
118        crate::network_debug::dump_http_request(&prepared);
119        let method = Method::from_bytes(prepared.method.as_bytes())
120            .map_err(|e| PortError::Http(format!("invalid method '{}': {e}", prepared.method)))?;
121        let mut builder = self.client.request(method, &prepared.url);
122        for (name, value) in &prepared.headers {
123            builder = builder.header(name, value);
124        }
125        if let Some(body) = prepared.body {
126            builder = builder.body(body);
127        }
128        let resp = builder.send().await.map_err(|e| Self::classify_error(&e))?;
129        let status = resp.status().as_u16();
130        let headers = resp
131            .headers()
132            .iter()
133            .map(|(k, v)| {
134                (
135                    k.as_str().to_string(),
136                    v.to_str().unwrap_or("<non-utf8>").to_string(),
137                )
138            })
139            .collect::<Vec<_>>();
140        let body = resp.bytes().await.map_err(|e| Self::classify_error(&e))?;
141        crate::network_debug::dump_http_response(status, &headers, &body);
142        Ok(HttpResponse {
143            status,
144            headers,
145            body,
146        })
147    }
148}
149
150#[async_trait::async_trait]
151impl HttpRequester for SharedHttpClient {
152    async fn request(&self, req: HttpRequest) -> Result<HttpResponse, PortError> {
153        if !self.metrics.is_enabled() {
154            return self.execute(req).await;
155        }
156        let request_bytes = req.body.as_ref().map_or(0, Bytes::len);
157        let started = Instant::now();
158        let inflight = self.inflight.fetch_add(1, Ordering::Relaxed) + 1;
159        record_http_gauge(self.metrics.as_ref(), inflight);
160        let result = self.execute(req).await;
161        let remaining = self
162            .inflight
163            .fetch_sub(1, Ordering::Relaxed)
164            .saturating_sub(1);
165        record_http_gauge(self.metrics.as_ref(), remaining);
166        let failed = match &result {
167            Ok(response) => response.status >= 400,
168            Err(_) => true,
169        };
170        let status = if failed { "error" } else { "ok" };
171        let labels = MetricLabels::one(LabelKey::Stage, "http")
172            .with(LabelKey::Protocol, "http")
173            .with(LabelKey::Operation, "request")
174            .with(LabelKey::Status, status);
175        let _ = self.metrics.try_record(MetricEvent::histogram(
176            MetricId::HttpRequestDurationSeconds,
177            started.elapsed().as_secs_f64(),
178            labels,
179        ));
180        let _ = self.metrics.try_record(MetricEvent::histogram(
181            MetricId::HttpRequestBytes,
182            request_bytes as f64,
183            labels,
184        ));
185        if let Ok(response) = &result {
186            let status_labels = labels.with(LabelKey::StatusClass, status_class(response.status));
187            let _ = self.metrics.try_record(MetricEvent::histogram(
188                MetricId::HttpResponseBytes,
189                response.body.len() as f64,
190                status_labels,
191            ));
192            let _ = self.metrics.try_record(MetricEvent::counter(
193                MetricId::HttpStatusTotal,
194                1.0,
195                status_labels,
196            ));
197        }
198        let _ =
199            self.metrics
200                .try_record(MetricEvent::counter(MetricId::OperationsTotal, 1.0, labels));
201        if failed {
202            if result
203                .as_ref()
204                .err()
205                .is_some_and(|error| error.to_string().to_ascii_lowercase().contains("timeout"))
206            {
207                let _ = self.metrics.try_record(MetricEvent::counter(
208                    MetricId::HttpTimeoutTotal,
209                    1.0,
210                    labels,
211                ));
212            }
213            let error_kind = if result.is_err() {
214                "http_request_failed"
215            } else {
216                "http_status_error"
217            };
218            let _ = self.metrics.try_record(MetricEvent::counter(
219                MetricId::ErrorsTotal,
220                1.0,
221                labels.with(LabelKey::ErrorKind, error_kind),
222            ));
223        }
224        result
225    }
226}
227
228/// 把 HTTP status 映射为固定低基数 class 标签。
229fn status_class(status: u16) -> &'static str {
230    match status {
231        100..=199 => "1xx",
232        200..=299 => "2xx",
233        300..=399 => "3xx",
234        400..=499 => "4xx",
235        500..=599 => "5xx",
236        _ => "other",
237    }
238}
239
240/// 发布当前 HTTP 实际在途请求数。
241fn record_http_gauge(metrics: &dyn AsyncMetricSink, inflight: usize) {
242    let _ = metrics.try_record(MetricEvent::gauge(
243        MetricId::HttpInflight,
244        inflight as f64,
245        MetricLabels::one(LabelKey::Stage, "http"),
246    ));
247}
helix_driver_host/network/http.rs

helix_driver_host/network/
http.rs
