| 295 | type Output = Result<http::Response<ResBody>, E>; |
| 296 | |
| 297 | fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { |
| 298 | let this = self.project(); |
| 299 | |
| 300 | match this.future.poll(cx) { |
| 301 | Poll::Ready(result) => { |
| 302 | if let Ok(response) = &result { |
| 303 | let status_code: i32 = match response.headers().get("grpc-status") { |
| 304 | None => 0, |
| 305 | Some(v) => match v.to_str() { |
| 306 | Ok(s) => s.parse().unwrap_or_default(), |
| 307 | Err(_) => 2, |
| 308 | }, |
| 309 | }; |
| 310 | let status_code = Code::from_i32(status_code); |
| 311 | |
| 312 | // Log to Prometheus |
| 313 | let labels = GrpcLabels { |
| 314 | service: this.service.clone(), |
| 315 | method: this.method.clone(), |
| 316 | status_code: format!("{:?}", status_code), |
| 317 | }; |
| 318 | GRPC_COUNTER.get_or_create(&labels).inc(); |
| 319 | GRPC_HISTOGRAM |
| 320 | .get_or_create(&labels) |
| 321 | .observe(this.start.elapsed().as_secs_f64()); |
| 322 | |
| 323 | // Log API request to Redis |
| 324 | let req_log = stream_pb::ApiRequestLog { |
| 325 | service: this.service.to_string(), |
| 326 | method: this.method.to_string(), |
| 327 | metadata: response |
| 328 | .headers() |
| 329 | .iter() |
| 330 | .filter(|(k, _)| k.as_str().starts_with("x-log-")) |
| 331 | .map(|(k, v)| { |
| 332 | ( |
| 333 | k.as_str() |
| 334 | .strip_prefix("x-log-") |
| 335 | .unwrap_or_default() |
| 336 | .to_string(), |
| 337 | v.to_str().unwrap().to_string(), |
| 338 | ) |
| 339 | }) |
| 340 | .collect(), |
| 341 | }; |
| 342 | |
| 343 | task::spawn(async move { |
| 344 | if let Err(e) = stream::api_request::log_request(&req_log).await { |
| 345 | error!(error = %e.full(), "Log request error"); |
| 346 | } |
| 347 | }); |
| 348 | } |
| 349 | Poll::Ready(result) |
| 350 | } |
| 351 | Poll::Pending => Poll::Pending, |
| 352 | } |
| 353 | } |
| 354 | } |