fix(opensearch): handle index not present errors in search api (#4965)

Co-authored-by: hyperswitch-bot[bot] <148525504+hyperswitch-bot[bot]@users.noreply.github.com>
Co-authored-by: ivor-juspay <[email protected]>

crates/analytics/docs/README.md

@@ -101,4 +101,18 @@ Here's an example of how to do this:
 [default.features]
 audit_trail=true
 system_metrics=true
-```
+global_search=true
+```
+
+## Viewing the data on OpenSearch Dashboard
+
+To view the data on the OpenSearch dashboard perform the following steps:
+
+- Go to the OpenSearch Dashboard home and click on `Stack Management` under the Management tab
+- Select `Index Patterns`
+- Click on `Create index pattern`
+- Define an index pattern with the same name that matches your indices and click on `Next Step`
+- Select a time field that will be used for time-based queries
+- Save the index pattern
+
+Now, head on to the `Discover` tab, to select the newly created index pattern and query the data

crates/analytics/src/opensearch.rs

@@ -76,6 +76,8 @@ pub enum OpenSearchError {
     ResponseError,
     #[error("Opensearch query building error")]
     QueryBuildingError,
+    #[error("Opensearch deserialisation error")]
+    DeserialisationError,
 }
 
 impl ErrorSwitch<OpenSearchError> for QueryBuildingError {
@@ -111,6 +113,12 @@ impl ErrorSwitch<ApiErrorResponse> for OpenSearchError {
                 "Query building error",
                 None,
             )),
+            Self::DeserialisationError => ApiErrorResponse::InternalServerError(ApiError::new(
+                "IR",
+                0,
+                "Deserialisation error",
+                None,
+            )),
         }
     }
 }

crates/analytics/src/search.rs

@@ -4,7 +4,7 @@ use api_models::analytics::search::{
 };
 use common_utils::errors::{CustomResult, ReportSwitchExt};
 use error_stack::ResultExt;
-use serde_json::Value;
+use router_env::tracing;
 use strum::IntoEnumIterator;
 
 use crate::opensearch::{
@@ -22,27 +22,59 @@ pub async fn msearch_results(
         .add_filter_clause("merchant_id".to_string(), merchant_id.to_string())
         .switch()?;
 
-    let response_body = client
+    let response_text: OpenMsearchOutput = client
         .execute(query_builder)
         .await
         .change_context(OpenSearchError::ConnectionError)?
-        .json::<OpenMsearchOutput<Value>>()
+        .text()
         .await
-        .change_context(OpenSearchError::ResponseError)?;
+        .change_context(OpenSearchError::ResponseError)
+        .and_then(|body: String| {
+            serde_json::from_str::<OpenMsearchOutput>(&body)
+                .change_context(OpenSearchError::DeserialisationError)
+                .attach_printable(body.clone())
+        })?;
+
+    let response_body: OpenMsearchOutput = response_text;
 
     Ok(response_body
         .responses
         .into_iter()
         .zip(SearchIndex::iter())
-        .map(|(index_hit, index)| GetSearchResponse {
-            count: index_hit.hits.total.value,
-            index,
-            hits: index_hit
-                .hits
-                .hits
-                .into_iter()
-                .map(|hit| hit._source)
-                .collect(),
+        .map(|(index_hit, index)| match index_hit {
+            OpensearchOutput::Success(success) => {
+                if success.status == 200 {
+                    GetSearchResponse {
+                        count: success.hits.total.value,
+                        index,
+                        hits: success
+                            .hits
+                            .hits
+                            .into_iter()
+                            .map(|hit| hit.source)
+                            .collect(),
+                    }
+                } else {
+                    tracing::error!("Unexpected status code: {}", success.status,);
+                    GetSearchResponse {
+                        count: 0,
+                        index,
+                        hits: Vec::new(),
+                    }
+                }
+            }
+            OpensearchOutput::Error(error) => {
+                tracing::error!(
+                    index = ?index,
+                    error_response = ?error,
+                    "Search error"
+                );
+                GetSearchResponse {
+                    count: 0,
+                    index,
+                    hits: Vec::new(),
+                }
+            }
         })
         .collect())
 }
@@ -65,22 +97,54 @@ pub async fn search_results(
         .set_offset_n_count(search_req.offset, search_req.count)
         .switch()?;
 
-    let response_body = client
+    let response_text: OpensearchOutput = client
         .execute(query_builder)
         .await
         .change_context(OpenSearchError::ConnectionError)?
-        .json::<OpensearchOutput<Value>>()
+        .text()
         .await
-        .change_context(OpenSearchError::ResponseError)?;
+        .change_context(OpenSearchError::ResponseError)
+        .and_then(|body: String| {
+            serde_json::from_str::<OpensearchOutput>(&body)
+                .change_context(OpenSearchError::DeserialisationError)
+                .attach_printable(body.clone())
+        })?;
+
+    let response_body: OpensearchOutput = response_text;
 
-    Ok(GetSearchResponse {
-        count: response_body.hits.total.value,
-        index: req.index,
-        hits: response_body
-            .hits
-            .hits
-            .into_iter()
-            .map(|hit| hit._source)
-            .collect(),
-    })
+    match response_body {
+        OpensearchOutput::Success(success) => {
+            if success.status == 200 {
+                Ok(GetSearchResponse {
+                    count: success.hits.total.value,
+                    index: req.index,
+                    hits: success
+                        .hits
+                        .hits
+                        .into_iter()
+                        .map(|hit| hit.source)
+                        .collect(),
+                })
+            } else {
+                tracing::error!("Unexpected status code: {}", success.status);
+                Ok(GetSearchResponse {
+                    count: 0,
+                    index: req.index,
+                    hits: Vec::new(),
+                })
+            }
+        }
+        OpensearchOutput::Error(error) => {
+            tracing::error!(
+                index = ?req.index,
+                error_response = ?error,
+                "Search error"
+            );
+            Ok(GetSearchResponse {
+                count: 0,
+                index: req.index,
+                hits: Vec::new(),
+            })
+        }
+    }
 }

crates/api_models/src/analytics/search.rs

@@ -48,19 +48,40 @@ pub struct GetSearchResponse {
 }
 
 #[derive(Debug, serde::Deserialize)]
-pub struct OpenMsearchOutput<T> {
-    pub responses: Vec<OpensearchOutput<T>>,
+pub struct OpenMsearchOutput {
+    pub responses: Vec<OpensearchOutput>,
 }
 
 #[derive(Debug, serde::Deserialize)]
-pub struct OpensearchOutput<T> {
-    pub hits: OpensearchResults<T>,
+#[serde(untagged)]
+pub enum OpensearchOutput {
+    Success(OpensearchSuccess),
+    Error(OpensearchError),
 }
 
 #[derive(Debug, serde::Deserialize)]
-pub struct OpensearchResults<T> {
+pub struct OpensearchError {
+    pub error: OpensearchErrorDetails,
+    pub status: u16,
+}
+
+#[derive(Debug, serde::Deserialize)]
+pub struct OpensearchErrorDetails {
+    #[serde(rename = "type")]
+    pub error_type: String,
+    pub reason: String,
+}
+
+#[derive(Debug, serde::Deserialize)]
+pub struct OpensearchSuccess {
+    pub status: u16,
+    pub hits: OpensearchHits,
+}
+
+#[derive(Debug, serde::Deserialize)]
+pub struct OpensearchHits {
     pub total: OpensearchResultsTotal,
-    pub hits: Vec<OpensearchHits<T>>,
+    pub hits: Vec<OpensearchHit>,
 }
 
 #[derive(Debug, serde::Deserialize)]
@@ -69,6 +90,7 @@ pub struct OpensearchResultsTotal {
 }
 
 #[derive(Debug, serde::Deserialize)]
-pub struct OpensearchHits<T> {
-    pub _source: T,
+pub struct OpensearchHit {
+    #[serde(rename = "_source")]
+    pub source: Value,
 }