chore(analytics): opensearch client creation based on config (#7810)

Co-authored-by: hyperswitch-bot[bot] <148525504+hyperswitch-bot[bot]@users.noreply.github.com>
This commit is contained in:
Shivansh Mathur
2025-04-17 13:32:22 +05:30
committed by GitHub
parent 5690864fd2
commit 2067bc3520
8 changed files with 61 additions and 23 deletions

View File

@ -887,6 +887,7 @@ region = "kms_region" # The AWS region used by the KMS SDK for decrypting data.
[opensearch]
host = "https://localhost:9200"
enabled = false
[opensearch.auth]
auth = "basic"

View File

@ -242,6 +242,7 @@ region = "report_download_config_region" # Region of the buc
[opensearch]
host = "https://localhost:9200"
enabled = false
[opensearch.auth]
auth = "basic"

View File

@ -940,6 +940,7 @@ keys = "accept-language,user-agent,x-profile-id"
[opensearch]
host = "https://localhost:9200"
enabled = false
[opensearch.auth]
auth = "basic"

View File

@ -763,6 +763,7 @@ keys = "accept-language,user-agent,x-profile-id"
[opensearch]
host = "https://opensearch:9200"
enabled = false
[opensearch.auth]
auth = "basic"

View File

@ -71,6 +71,8 @@ pub struct OpenSearchConfig {
host: String,
auth: OpenSearchAuth,
indexes: OpenSearchIndexes,
#[serde(default)]
enabled: bool,
}
impl Default for OpenSearchConfig {
@ -91,12 +93,15 @@ impl Default for OpenSearchConfig {
sessionizer_refunds: "sessionizer-refund-events".to_string(),
sessionizer_disputes: "sessionizer-dispute-events".to_string(),
},
enabled: false,
}
}
}
#[derive(Debug, thiserror::Error)]
pub enum OpenSearchError {
#[error("Opensearch is not enabled")]
NotEnabled,
#[error("Opensearch connection error")]
ConnectionError,
#[error("Opensearch NON-200 response content: '{0}'")]
@ -176,6 +181,12 @@ impl ErrorSwitch<ApiErrorResponse> for OpenSearchError {
"Access Forbidden error",
None,
)),
Self::NotEnabled => ApiErrorResponse::InternalServerError(ApiError::new(
"IR",
8,
"Opensearch is not enabled",
None,
)),
}
}
}
@ -408,15 +419,24 @@ impl OpenSearchAuth {
}
impl OpenSearchConfig {
pub async fn get_opensearch_client(&self) -> StorageResult<OpenSearchClient> {
Ok(OpenSearchClient::create(self)
pub async fn get_opensearch_client(&self) -> StorageResult<Option<OpenSearchClient>> {
if !self.enabled {
return Ok(None);
}
Ok(Some(
OpenSearchClient::create(self)
.await
.map_err(|_| StorageError::InitializationError)?)
.change_context(StorageError::InitializationError)?,
))
}
pub fn validate(&self) -> Result<(), ApplicationError> {
use common_utils::{ext_traits::ConfigExt, fp_utils::when};
if !self.enabled {
return Ok(());
}
when(self.host.is_default_or_empty(), || {
Err(ApplicationError::InvalidConfigurationValueError(
"Opensearch host must not be empty".into(),
@ -430,6 +450,7 @@ impl OpenSearchConfig {
Ok(())
}
}
#[derive(Debug, serde::Deserialize, PartialEq)]
#[serde(rename_all = "lowercase")]
pub enum OpenSearchHealthStatus {

View File

@ -2245,7 +2245,10 @@ pub mod routes {
.collect();
analytics::search::msearch_results(
&state.opensearch_client,
state
.opensearch_client
.as_ref()
.ok_or_else(|| error_stack::report!(OpenSearchError::NotEnabled))?,
req,
search_params,
SEARCH_INDEXES.to_vec(),
@ -2392,7 +2395,14 @@ pub mod routes {
})
})
.collect();
analytics::search::search_results(&state.opensearch_client, req, search_params)
analytics::search::search_results(
state
.opensearch_client
.as_ref()
.ok_or_else(|| error_stack::report!(OpenSearchError::NotEnabled))?,
req,
search_params,
)
.await
.map(ApplicationResponse::Json)
},

View File

@ -138,12 +138,15 @@ impl HealthCheckInterface for app::SessionState {
async fn health_check_opensearch(
&self,
) -> CustomResult<HealthState, errors::HealthCheckDBError> {
self.opensearch_client
if let Some(client) = self.opensearch_client.as_ref() {
client
.deep_health_check()
.await
.change_context(errors::HealthCheckDBError::OpensearchError)?;
Ok(HealthState::Running)
} else {
Ok(HealthState::NotApplicable)
}
}
async fn health_check_outgoing(

View File

@ -118,7 +118,7 @@ pub struct SessionState {
pub base_url: String,
pub tenant: Tenant,
#[cfg(feature = "olap")]
pub opensearch_client: Arc<OpenSearchClient>,
pub opensearch_client: Option<Arc<OpenSearchClient>>,
pub grpc_client: Arc<GrpcClients>,
pub theme_storage_client: Arc<dyn FileStorageInterface>,
pub locale: String,
@ -223,7 +223,7 @@ pub struct AppState {
#[cfg(feature = "olap")]
pub pools: HashMap<id_type::TenantId, AnalyticsProvider>,
#[cfg(feature = "olap")]
pub opensearch_client: Arc<OpenSearchClient>,
pub opensearch_client: Option<Arc<OpenSearchClient>>,
pub request_id: Option<RequestId>,
pub file_storage_client: Arc<dyn FileStorageInterface>,
pub encryption_client: Arc<dyn EncryptionManagementInterface>,
@ -342,12 +342,12 @@ impl AppState {
#[allow(clippy::expect_used)]
#[cfg(feature = "olap")]
let opensearch_client = Arc::new(
conf.opensearch
let opensearch_client = conf
.opensearch
.get_opensearch_client()
.await
.expect("Failed to create opensearch client"),
);
.expect("Failed to initialize OpenSearch client.")
.map(Arc::new);
#[allow(clippy::expect_used)]
let cache_store = get_cache_store(&conf.clone(), shut_down_signal, testable)
@ -501,7 +501,7 @@ impl AppState {
#[cfg(feature = "email")]
email_client: Arc::clone(&self.email_client),
#[cfg(feature = "olap")]
opensearch_client: Arc::clone(&self.opensearch_client),
opensearch_client: self.opensearch_client.clone(),
grpc_client: Arc::clone(&self.grpc_client),
theme_storage_client: self.theme_storage_client.clone(),
locale: locale.unwrap_or(common_utils::consts::DEFAULT_LOCALE.to_string()),