removed unwraps in favor of error propagation

Eshanatnight · Eshanatnight · commit 7bd8102067ff · 2024-04-18T13:53:45.000+05:30
diff --git a/server/src/analytics.rs b/server/src/analytics.rs
@@ -74,7 +74,7 @@ pub struct Report {
 }
 
 impl Report {
-    pub async fn new() -> Self {
+    pub async fn new() -> anyhow::Result<Self> {
         let mut upt: f64 = 0.0;
         if let Ok(uptime) = uptime_lib::get() {
             upt = uptime.as_secs_f64();
@@ -91,9 +91,9 @@ impl Report {
             cpu_count = info.cpus().len();
             mem_total = info.total_memory();
         }
-        let ingestor_metrics = fetch_ingestors_metrics().await;
+        let ingestor_metrics = fetch_ingestors_metrics().await?;
 
-        Self {
+        Ok(Self {
             deployment_id: storage::StorageMetadata::global().deployment_id,
             uptime: upt,
             report_created_at: Utc::now(),
@@ -113,7 +113,7 @@ impl Report {
             total_json_bytes: ingestor_metrics.4,
             total_parquet_bytes: ingestor_metrics.5,
             metrics: build_metrics().await,
-        }
+        })
     }
 
     pub async fn send(&self) {
@@ -148,7 +148,7 @@ fn total_event_stats() -> (u64, u64, u64) {
     (total_events, total_json_bytes, total_parquet_bytes)
 }
 
-async fn fetch_ingestors_metrics() -> (u64, u64, usize, u64, u64, u64) {
+async fn fetch_ingestors_metrics() -> anyhow::Result<(u64, u64, usize, u64, u64, u64)> {
     let event_stats = total_event_stats();
     let mut node_metrics =
         NodeMetrics::new(total_streams(), event_stats.0, event_stats.1, event_stats.2);
@@ -181,24 +181,24 @@ async fn fetch_ingestors_metrics() -> (u64, u64, usize, u64, u64, u64) {
                 .header(header::CONTENT_TYPE, "application/json")
                 .send()
                 .await
-                .unwrap(); // should respond
+                .expect("should respond");
 
-            let data = serde_json::from_slice::<NodeMetrics>(&resp.bytes().await.unwrap()).unwrap();
+            let data = serde_json::from_slice::<NodeMetrics>(&resp.bytes().await?)?;
             vec.push(data);
             active_ingestors += 1;
         }
 
         node_metrics.accumulate(&mut vec);
     }
 
-    (
+    Ok((
         active_ingestors,
         offline_ingestors,
         node_metrics.stream_count,
         node_metrics.total_events_count,
         node_metrics.total_json_bytes,
         node_metrics.total_parquet_bytes,
-    )
+    ))
 }
 
 async fn build_metrics() -> HashMap<String, Value> {
@@ -220,14 +220,23 @@ async fn build_metrics() -> HashMap<String, Value> {
     metrics
 }
 
-pub fn init_analytics_scheduler() {
+pub fn init_analytics_scheduler() -> anyhow::Result<()> {
     log::info!("Setting up schedular for anonymous user analytics");
 
     let mut scheduler = AsyncScheduler::new();
     scheduler
         .every(ANALYTICS_SEND_INTERVAL_SECONDS)
         .run(move || async {
-            Report::new().await.send().await;
+            Report::new()
+                .await
+                .unwrap_or_else(|err| {
+                    // panicing because seperate thread
+                    // TODO: a better way to handle this
+                    log::error!("Error while sending analytics: {}", err.to_string());
+                    panic!("{}", err.to_string());
+                })
+                .send()
+                .await;
         });
 
     tokio::spawn(async move {
@@ -236,6 +245,8 @@ pub fn init_analytics_scheduler() {
             tokio::time::sleep(Duration::from_secs(10)).await;
         }
     });
+
+    Ok(())
 }
 
 #[derive(Serialize, Deserialize, Default, Debug)]
diff --git a/server/src/handlers/http.rs b/server/src/handlers/http.rs
@@ -85,7 +85,7 @@ pub async fn fetch_schema(stream_name: &str) -> anyhow::Result<arrow_schema::Sch
         .await?
         .iter()
         // we should be able to unwrap as we know the data is valid schema
-        .map(|byte_obj| serde_json::from_slice(byte_obj).unwrap())
+        .map(|byte_obj| serde_json::from_slice(byte_obj).expect("data is valid json"))
         .collect_vec();
 
     let new_schema = Schema::try_merge(res)?;
@@ -97,7 +97,7 @@ pub async fn fetch_schema(stream_name: &str) -> anyhow::Result<arrow_schema::Sch
 pub async fn send_query_request_to_ingestor(query: &Query) -> anyhow::Result<Vec<Value>> {
     // send the query request to the ingestor
     let mut res = vec![];
-    let ima = get_ingestor_info().await.unwrap();
+    let ima = get_ingestor_info().await?;
 
     for im in ima.iter() {
         let uri = format!(
diff --git a/server/src/handlers/http/cluster/mod.rs b/server/src/handlers/http/cluster/mod.rs
@@ -37,6 +37,8 @@ use chrono::Utc;
 use http::StatusCode;
 use itertools::Itertools;
 use relative_path::RelativePathBuf;
+use serde::de::Error;
+use serde_json::error::Error as SerdeError;
 use serde_json::Value as JsonValue;
 use url::Url;
 
@@ -262,9 +264,13 @@ pub async fn get_cluster_info() -> Result<impl Responder, StreamError> {
                     StreamError::SerdeError(err)
                 })?
                 .get("staging")
-                .unwrap()
+                .ok_or(StreamError::SerdeError(SerdeError::missing_field(
+                    "staging",
+                )))?
                 .as_str()
-                .unwrap()
+                .ok_or(StreamError::SerdeError(SerdeError::custom(
+                    "staging path not a string/ not provided",
+                )))?
                 .to_string();
 
             (true, sp, None, status)
@@ -304,7 +310,9 @@ pub async fn get_cluster_metrics() -> Result<impl Responder, PostError> {
             &ingestor.domain_name,
             base_path_without_preceding_slash()
         ))
-        .unwrap();
+        .map_err(|err| {
+            PostError::Invalid(anyhow::anyhow!("Invalid URL in Ingestor Metadata: {}", err))
+        })?;
 
         let res = reqwest::Client::new()
             .get(uri)
diff --git a/server/src/handlers/http/logstream.rs b/server/src/handlers/http/logstream.rs
@@ -169,7 +169,7 @@ pub async fn put_stream(req: HttpRequest, body: Bytes) -> Result<impl Responder,
     }
 
     if !body.is_empty() && static_schema_flag == "true" {
-        let static_schema: StaticSchema = serde_json::from_slice(&body).unwrap();
+        let static_schema: StaticSchema = serde_json::from_slice(&body)?;
         let parsed_schema = convert_static_schema_to_arrow_schema(static_schema);
         if let Ok(parsed_schema) = parsed_schema {
             schema = parsed_schema;
@@ -357,7 +357,7 @@ pub async fn get_stats(req: HttpRequest) -> Result<impl Responder, StreamError>
         None
     };
 
-    let hash_map = STREAM_INFO.read().unwrap();
+    let hash_map = STREAM_INFO.read().expect("Readable");
     let stream_meta = &hash_map
         .get(&stream_name)
         .ok_or(StreamError::StreamNotFound(stream_name.clone()))?;
@@ -396,7 +396,7 @@ pub async fn get_stats(req: HttpRequest) -> Result<impl Responder, StreamError>
         stats
     };
 
-    let stats = serde_json::to_value(stats).unwrap();
+    let stats = serde_json::to_value(stats)?;
 
     Ok((web::Json(stats), StatusCode::OK))
 }
diff --git a/server/src/handlers/http/modal/ingest_server.rs b/server/src/handlers/http/modal/ingest_server.rs
@@ -42,6 +42,7 @@ use actix_web::body::MessageBody;
 use actix_web::Scope;
 use actix_web::{web, App, HttpServer};
 use actix_web_prometheus::PrometheusMetrics;
+use anyhow::anyhow;
 use async_trait::async_trait;
 use base64::Engine;
 use itertools::Itertools;
@@ -202,12 +203,11 @@ impl IngestServer {
             return Ok(());
         };
 
-        let resource = serde_json::to_string(&resource)
-            .unwrap()
+        let resource = serde_json::to_string(&resource)?
             .try_into_bytes()
-            .unwrap();
+            .map_err(|err| anyhow!(err))?;
 
-        store.put_object(&path, resource.clone()).await?;
+        store.put_object(&path, resource).await?;
 
         Ok(())
     }
diff --git a/server/src/handlers/http/modal/query_server.rs b/server/src/handlers/http/modal/query_server.rs
@@ -180,7 +180,7 @@ impl QueryServer {
         // all internal data structures populated now.
         // start the analytics scheduler if enabled
         if CONFIG.parseable.send_analytics {
-            analytics::init_analytics_scheduler();
+            analytics::init_analytics_scheduler()?;
         }
 
         self.start(prometheus, CONFIG.parseable.openid.clone())
diff --git a/server/src/handlers/http/modal/server.rs b/server/src/handlers/http/modal/server.rs
@@ -443,7 +443,7 @@ impl Server {
             sync::object_store_sync();
 
         if CONFIG.parseable.send_analytics {
-            analytics::init_analytics_scheduler();
+            analytics::init_analytics_scheduler()?;
         }
 
         tokio::spawn(handlers::livetail::server());
diff --git a/server/src/handlers/http/query.rs b/server/src/handlers/http/query.rs
@@ -70,7 +70,12 @@ pub async fn query(req: HttpRequest, query_request: Query) -> Result<impl Respon
     // create a visitor to extract the table name
     let mut visitor = TableScanVisitor::default();
     let _ = raw_logical_plan.visit(&mut visitor);
-    let table_name = visitor.into_inner().pop().unwrap();
+    let table_name = visitor
+        .into_inner()
+        .pop()
+        .ok_or(QueryError::MalformedQuery(
+            "No table found from sql".to_string(),
+        ))?;
 
     if CONFIG.parseable.mode == Mode::Query {
         if let Ok(new_schema) = fetch_schema(&table_name).await {
@@ -278,6 +283,8 @@ pub enum QueryError {
     ObjectStorage(#[from] ObjectStorageError),
     #[error("Evern Error: {0}")]
     EventError(#[from] EventError),
+    #[error("Error: {0}")]
+    MalformedQuery(String),
 }
 
 impl actix_web::ResponseError for QueryError {
diff --git a/server/src/handlers/livetail.rs b/server/src/handlers/livetail.rs
@@ -270,7 +270,7 @@ fn extract_cookie(header: &MetadataMap) -> Option<Cookie> {
         .iter()
         .filter_map(|value| value.to_str().ok())
         .flat_map(Cookie::split_parse)
-        .map(|value| value.unwrap())
+        .map(|value| value.expect("cookie is parseable"))
         .collect();
 
     cookies
diff --git a/server/src/metrics/prom_utils.rs b/server/src/metrics/prom_utils.rs
@@ -22,11 +22,12 @@ struct StorageMetrics {
 
 impl Default for Metrics {
     fn default() -> Self {
-        let socket = get_url();
+        let url = get_url();
         let address = format!(
             "http://{}:{}",
-            socket.domain().unwrap(),
-            socket.port().unwrap_or_default()
+            url.domain()
+                .unwrap_or(url.host_str().expect("should have a host")),
+            url.port().unwrap_or_default()
         );
         Metrics {
             address,
@@ -68,11 +69,11 @@ impl Metrics {
                     prom_dress.process_resident_memory_bytes += val;
                 }
             } else if sample.metric == "parseable_storage_size" {
-                if sample.labels.get("type").unwrap() == "data" {
+                if sample.labels.get("type").expect("type is present") == "data" {
                     if let PromValue::Gauge(val) = sample.value {
                         prom_dress.parseable_storage_size.data += val;
                     }
-                } else if sample.labels.get("type").unwrap() == "staging" {
+                } else if sample.labels.get("type").expect("type is present") == "staging" {
                     if let PromValue::Gauge(val) = sample.value {
                         prom_dress.parseable_storage_size.staging += val;
                     }
diff --git a/server/src/storage/localfs.rs b/server/src/storage/localfs.rs
@@ -162,15 +162,17 @@ impl ObjectStorage for LocalFS {
             let flag = entry
                 .path()
                 .file_name()
-                .unwrap_or_default()
+                .ok_or(ObjectStorageError::NoSuchKey(
+                    "Dir Entry Suggests no file present".to_string(),
+                ))?
                 .to_str()
-                .unwrap_or_default()
+                .expect("file name is parseable to str")
                 .contains("ingestor");
 
             if flag {
                 path_arr.push(RelativePathBuf::from_iter([
                     stream_name,
-                    entry.path().file_name().unwrap().to_str().unwrap(),
+                    entry.path().file_name().unwrap().to_str().unwrap(), // checking the error before hand
                 ]));
             }
         }
@@ -209,9 +211,11 @@ impl ObjectStorage for LocalFS {
             let path = entry
                 .path()
                 .file_name()
-                .unwrap()
+                .ok_or(ObjectStorageError::NoSuchKey(
+                    "Dir Entry suggests no file present".to_string(),
+                ))?
                 .to_str()
-                .unwrap()
+                .expect("file name is parseable to str")
                 .to_owned();
             let ingestor_file = filter_func(path);
 
@@ -390,9 +394,9 @@ impl ObjectStorage for LocalFS {
         self.root
             .iter()
             .last()
-            .unwrap()
+            .expect("can be unwrapped without checking as the path is absolute")
             .to_str()
-            .unwrap()
+            .expect("valid unicode")
             .to_string()
     }
 }
diff --git a/server/src/storage/object_storage.rs b/server/src/storage/object_storage.rs
@@ -220,7 +220,7 @@ pub trait ObjectStorage: Sync + 'static {
                 ]);
                 let data = self.get_object(&schema_path).await?;
                 // schema was not found in store, so it needs to be placed
-                self.put_schema(stream_name, &serde_json::from_slice(&data).unwrap())
+                self.put_schema(stream_name, &serde_json::from_slice(&data)?)
                     .await?;
 
                 data
@@ -331,7 +331,7 @@ pub trait ObjectStorage: Sync + 'static {
             .get("retention")
             .cloned();
         if let Some(retention) = retention {
-            Ok(serde_json::from_value(retention).unwrap())
+            Ok(serde_json::from_value(retention)?)
         } else {
             Ok(Retention::default())
         }
diff --git a/server/src/storage/store_metadata.rs b/server/src/storage/store_metadata.rs
@@ -128,7 +128,7 @@ pub async fn resolve_parseable_metadata() -> Result<StorageMetadata, ObjectStora
         EnvChange::NewStaging(mut metadata) => {
             // if server is started in ingest mode,we need to make sure that query mode has been started
             // i.e the metadata is updated to reflect the server mode = Query
-            if Mode::from_string(&metadata.server_mode).unwrap() == Mode::All && CONFIG.parseable.mode == Mode::Ingest {
+            if Mode::from_string(&metadata.server_mode).map_err(ObjectStorageError::Custom)? == Mode::All && CONFIG.parseable.mode == Mode::Ingest {
                 Err("Starting Ingest Mode is not allowed, Since Query Server has not been started yet")
             } else {
                 create_dir_all(CONFIG.staging_dir())?;
@@ -201,7 +201,8 @@ fn determine_environment(
             // if both staging and remote have same deployment id
             if staging.deployment_id == remote.deployment_id {
                 EnvChange::None(remote)
-            } else if Mode::from_string(&remote.server_mode).unwrap() == Mode::All
+            } else if Mode::from_string(&remote.server_mode).expect("server mode is valid here")
+                == Mode::All
                 && (CONFIG.parseable.mode == Mode::Query || CONFIG.parseable.mode == Mode::Ingest)
             {
                 // if you are switching to distributed mode from standalone mode

Original file line number	Diff line number	Diff line change
`@@ -180,7 +180,7 @@ impl QueryServer {`
`180`	`180`	`// all internal data structures populated now.`
`181`	`181`	`// start the analytics scheduler if enabled`
`182`	`182`	`if CONFIG.parseable.send_analytics {`
`183`		`- analytics::init_analytics_scheduler();`
	`183`	`+ analytics::init_analytics_scheduler()?;`
`184`	`184`	`}`
`185`	`185`
`186`	`186`	`self.start(prometheus, CONFIG.parseable.openid.clone())`
Original file line number	Diff line number	Diff line change
`@@ -443,7 +443,7 @@ impl Server {`
`443`	`443`	`sync::object_store_sync();`
`444`	`444`
`445`	`445`	`if CONFIG.parseable.send_analytics {`
`446`		`- analytics::init_analytics_scheduler();`
	`446`	`+ analytics::init_analytics_scheduler()?;`
`447`	`447`	`}`
`448`	`448`
`449`	`449`	`tokio::spawn(handlers::livetail::server());`