refactor: single reqwest::Client for HTTP (#1071)

Co-authored-by: Nikhil Sinha <131262146+nikhilsinhaparseable@users.noreply.github.com>

Author: Devdutt Shenoi

Cargo.lock

@@ -73,6 +73,8 @@ relative-path = { version = "1.7", features = ["serde"] }
 reqwest = { version = "0.11.27", default-features = false, features = [
  "rustls-tls",
  "json",
+ "gzip",
+ "brotli",
 ] } # cannot update cause rustls is not latest `see rustls`
 rustls = "0.22.4" # cannot update to 0.23 actix has not caught up yet
 rustls-pemfile = "2.1.2"

Cargo.toml

@@ -22,8 +22,8 @@ use crate::handlers::http::cluster::utils::check_liveness;
 use crate::handlers::http::{base_path_without_preceding_slash, cluster};
 use crate::handlers::STREAM_NAME_HEADER_KEY;
 use crate::option::{Mode, CONFIG};
-use crate::storage;
 use crate::{metadata, stats};
+use crate::{storage, HTTP_CLIENT};
 
 use crate::stats::Stats;
 use actix_web::{web, HttpRequest, Responder};
@@ -132,9 +132,7 @@ impl Report {
  }
 
  pub async fn send(&self) {
- let client = reqwest::Client::new();
-
- let _ = client
+ let _ = HTTP_CLIENT
  .post(ANALYTICS_SERVER_URL)
  .header(STREAM_NAME_HEADER_KEY, "serverusageevent")
  .json(&self)
@@ -240,7 +238,7 @@ async fn fetch_ingestors_metrics(
  ))
  .expect("Should be a valid URL");
 
- let resp = reqwest::Client::new()
+ let resp = HTTP_CLIENT
  .get(uri)
  .header(header::AUTHORIZATION, im.token.clone())
  .header(header::CONTENT_TYPE, "application/json")

src/analytics.rs

@@ -21,12 +21,11 @@ use std::{
  fmt::{Debug, Display},
 };
 
-use crate::{about::current, storage::StorageMetadata};
+use crate::{about::current, storage::StorageMetadata, HTTP_CLIENT};
 
 use super::option::CONFIG;
 use chrono::{DateTime, Utc};
 use once_cell::sync::Lazy;
-use reqwest::Client;
 use serde::Serialize;
 use serde_json::{json, Value};
 use tracing::error;
@@ -38,7 +37,6 @@ static AUDIT_LOGGER: Lazy<Option<AuditLogger>> = Lazy::new(AuditLogger::new);
 
 // AuditLogger handles sending audit logs to a remote logging system
 pub struct AuditLogger {
- client: Client,
  log_endpoint: Url,
 }
 
@@ -62,16 +60,12 @@ impl AuditLogger {
  }
  };
 
- Some(AuditLogger {
- client: reqwest::Client::new(),
- log_endpoint,
- })
+ Some(AuditLogger { log_endpoint })
  }
 
  // Sends the audit log to the configured endpoint with proper authentication
  async fn send_log(&self, json: Value) {
- let mut req = self
- .client
+ let mut req = HTTP_CLIENT
  .post(self.log_endpoint.as_str())
  .json(&json)
  .header("x-p-stream", "audit_log");

src/audit.rs

@@ -33,6 +33,7 @@ use crate::stats::Stats;
 use crate::storage::object_storage::ingestor_metadata_path;
 use crate::storage::{ObjectStorageError, STREAM_ROOT_DIRECTORY};
 use crate::storage::{ObjectStoreFormat, PARSEABLE_ROOT_DIRECTORY};
+use crate::HTTP_CLIENT;
 use actix_web::http::header::{self, HeaderMap};
 use actix_web::{HttpRequest, Responder};
 use bytes::Bytes;
@@ -76,8 +77,6 @@ pub async fn sync_streams_with_ingestors(
  StreamError::Anyhow(err)
  })?;
 
- let client = reqwest::Client::new();
-
  for ingestor in ingestor_infos {
  if !utils::check_liveness(&ingestor.domain_name).await {
  warn!("Ingestor {} is not live", ingestor.domain_name);
@@ -89,7 +88,7 @@ pub async fn sync_streams_with_ingestors(
  base_path_without_preceding_slash(),
  stream_name
  );
- let res = client
+ let res = HTTP_CLIENT
  .put(url)
  .headers(reqwest_headers.clone())
  .header(header::AUTHORIZATION, &ingestor.token)
@@ -126,7 +125,6 @@ pub async fn sync_users_with_roles_with_ingestors(
  RBACError::Anyhow(err)
  })?;
 
- let client = reqwest::Client::new();
  let role = to_vec(&role.clone()).map_err(|err| {
  error!("Fatal: failed to serialize role: {:?}", err);
  RBACError::SerdeError(err)
@@ -143,7 +141,7 @@ pub async fn sync_users_with_roles_with_ingestors(
  username
  );
 
- let res = client
+ let res = HTTP_CLIENT
  .put(url)
  .header(header::AUTHORIZATION, &ingestor.token)
  .header(header::CONTENT_TYPE, "application/json")
@@ -177,7 +175,6 @@ pub async fn sync_user_deletion_with_ingestors(username: &String) -> Result<(),
  RBACError::Anyhow(err)
  })?;
 
- let client = reqwest::Client::new();
  for ingestor in ingestor_infos.iter() {
  if !utils::check_liveness(&ingestor.domain_name).await {
  warn!("Ingestor {} is not live", ingestor.domain_name);
@@ -190,7 +187,7 @@ pub async fn sync_user_deletion_with_ingestors(username: &String) -> Result<(),
  username
  );
 
- let res = client
+ let res = HTTP_CLIENT
  .delete(url)
  .header(header::AUTHORIZATION, &ingestor.token)
  .send()
@@ -231,7 +228,6 @@ pub async fn sync_user_creation_with_ingestors(
  user.roles.clone_from(role);
  }
  let username = user.username();
- let client = reqwest::Client::new();
 
  let user = to_vec(&user).map_err(|err| {
  error!("Fatal: failed to serialize user: {:?}", err);
@@ -250,7 +246,7 @@ pub async fn sync_user_creation_with_ingestors(
  username
  );
 
- let res = client
+ let res = HTTP_CLIENT
  .post(url)
  .header(header::AUTHORIZATION, &ingestor.token)
  .header(header::CONTENT_TYPE, "application/json")
@@ -283,7 +279,6 @@ pub async fn sync_password_reset_with_ingestors(username: &String) -> Result<(),
  error!("Fatal: failed to get ingestor info: {:?}", err);
  RBACError::Anyhow(err)
  })?;
- let client = reqwest::Client::new();
 
  for ingestor in ingestor_infos.iter() {
  if !utils::check_liveness(&ingestor.domain_name).await {
@@ -297,7 +292,7 @@ pub async fn sync_password_reset_with_ingestors(username: &String) -> Result<(),
  username
  );
 
- let res = client
+ let res = HTTP_CLIENT
  .post(url)
  .header(header::AUTHORIZATION, &ingestor.token)
  .header(header::CONTENT_TYPE, "application/json")
@@ -338,7 +333,6 @@ pub async fn sync_role_update_with_ingestors(
  RoleError::SerdeError(err)
  })?;
  let roles = Bytes::from(roles);
- let client = reqwest::Client::new();
 
  for ingestor in ingestor_infos.iter() {
  if !utils::check_liveness(&ingestor.domain_name).await {
@@ -352,7 +346,7 @@ pub async fn sync_role_update_with_ingestors(
  name
  );
 
- let res = client
+ let res = HTTP_CLIENT
  .put(url)
  .header(header::AUTHORIZATION, &ingestor.token)
  .header(header::CONTENT_TYPE, "application/json")
@@ -401,7 +395,7 @@ pub async fn fetch_daily_stats_from_ingestors(
  StreamError::Anyhow(anyhow::anyhow!("Invalid URL in Ingestor Metadata: {}", err))
  })?;
 
- let res = reqwest::Client::new()
+ let res = HTTP_CLIENT
  .get(uri)
  .header(header::AUTHORIZATION, &ingestor.token)
  .header(header::CONTENT_TYPE, "application/json")
@@ -512,8 +506,7 @@ pub async fn send_stream_delete_request(
  if !utils::check_liveness(&ingestor.domain_name).await {
  return Ok(());
  }
- let client = reqwest::Client::new();
- let resp = client
+ let resp = HTTP_CLIENT
  .delete(url)
  .header(header::CONTENT_TYPE, "application/json")
  .header(header::AUTHORIZATION, ingestor.token)
@@ -551,8 +544,7 @@ pub async fn send_retention_cleanup_request(
  if !utils::check_liveness(&ingestor.domain_name).await {
  return Ok(first_event_at);
  }
- let client = reqwest::Client::new();
- let resp = client
+ let resp = HTTP_CLIENT
  .post(url)
  .header(header::CONTENT_TYPE, "application/json")
  .header(header::AUTHORIZATION, ingestor.token)
@@ -603,7 +595,7 @@ pub async fn get_cluster_info() -> Result<impl Responder, StreamError> {
  ))
  .expect("should always be a valid url");
 
- let resp = reqwest::Client::new()
+ let resp = HTTP_CLIENT
  .get(uri)
  .header(header::AUTHORIZATION, ingestor.token.clone())
  .header(header::CONTENT_TYPE, "application/json")
@@ -752,7 +744,7 @@ async fn fetch_cluster_metrics() -> Result<Vec<Metrics>, PostError> {
  PostError::Invalid(anyhow::anyhow!("Invalid URL in Ingestor Metadata: {}", err))
  })?;
 
- let res = reqwest::Client::new()
+ let res = HTTP_CLIENT
  .get(uri)
  .header(header::AUTHORIZATION, &ingestor.token)
  .header(header::CONTENT_TYPE, "application/json")

src/handlers/http/cluster/mod.rs

@@ -16,8 +16,11 @@
  *
  */
 
-use crate::handlers::http::{
- base_path_without_preceding_slash, logstream::error::StreamError, modal::IngestorMetadata,
+use crate::{
+ handlers::http::{
+ base_path_without_preceding_slash, logstream::error::StreamError, modal::IngestorMetadata,
+ },
+ HTTP_CLIENT,
 };
 use actix_web::http::header;
 use chrono::{DateTime, Utc};
@@ -235,13 +238,13 @@ pub async fn check_liveness(domain_name: &str) -> bool {
  }
  };
 
- let reqw = reqwest::Client::new()
+ let req = HTTP_CLIENT
  .get(uri)
  .header(header::CONTENT_TYPE, "application/json")
  .send()
  .await;
 
- reqw.is_ok()
+ req.is_ok()
 }
 
 /// send a request to the ingestor to fetch its stats
@@ -255,8 +258,7 @@ pub async fn send_stats_request(
  return Ok(None);
  }
 
- let client = reqwest::Client::new();
- let res = client
+ let res = HTTP_CLIENT
  .get(url)
  .header(header::CONTENT_TYPE, "application/json")
  .header(header::AUTHORIZATION, ingestor.token)

src/handlers/http/cluster/utils.rs

@@ -23,7 +23,7 @@ use http::StatusCode;
 use itertools::Itertools;
 use serde_json::Value;
 
-use crate::{option::CONFIG, storage::STREAM_ROOT_DIRECTORY};
+use crate::{option::CONFIG, storage::STREAM_ROOT_DIRECTORY, HTTP_CLIENT};
 
 use self::{cluster::get_ingestor_info, query::Query};
 
@@ -109,7 +109,7 @@ pub async fn send_query_request_to_ingestor(query: &Query) -> anyhow::Result<Vec
  base_path_without_preceding_slash(),
  "query"
  );
- let reqw = reqwest::Client::new()
+ let reqw = HTTP_CLIENT
  .post(uri)
  .json(query)
  .header(http::header::AUTHORIZATION, im.token.clone())