fix: bugs causing server to not start in ingest mode

Author: Eshanatnight

server/src/handlers/airplane.rs

@@ -156,7 +156,7 @@ impl FlightService for AirServiceImpl {
 
         let time_delta = query.end - Utc::now();
 
-        let events = if CONFIG.parseable.mode == Mode::Query && time_delta.num_seconds() < 2 {
+        let events = if CONFIG.parseable.mode == Mode::Query && time_delta.num_seconds() < 1 {
             let sql = format!(
                 "{}\"query\": \"select * from {}\"{}",
                 L_CURLY, &stream_name, R_CURLY

server/src/handlers/http/cluster/utils.rs

@@ -16,7 +16,9 @@
  *
  */
 
-use crate::handlers::http::{logstream::error::StreamError, modal::IngestorMetadata};
+use crate::handlers::http::{
+    base_path_without_preceding_slash, logstream::error::StreamError, modal::IngestorMetadata,
+};
 use actix_web::http::header;
 use chrono::{DateTime, Utc};
 use http::StatusCode;
@@ -161,7 +163,11 @@ pub fn merge_quried_stats(stats: Vec<QueriedStats>) -> QueriedStats {
 }
 
 pub async fn check_liveness(domain_name: &str) -> bool {
-    let uri = match Url::parse(&format!("{}liveness", domain_name)) {
+    let uri = match Url::parse(&format!(
+        "{}{}/liveness",
+        domain_name,
+        base_path_without_preceding_slash()
+    )) {
         Ok(uri) => uri,
         Err(err) => {
             log::error!("Node Indentifier Failed To Parse: {}", err);

server/src/handlers/http/modal/ingest_server.rs

@@ -106,8 +106,6 @@ impl ParseableServer for IngestServer {
 
     /// implement the init method will just invoke the initialize method
     async fn init(&self) -> anyhow::Result<()> {
-        migrate_ingester_metadata().await?;
-
         self.validate()?;
 
         // check for querier state. Is it there, or was it there in the past
@@ -116,6 +114,7 @@ impl ParseableServer for IngestServer {
         self.validate_credentials().await?;
 
         let metadata = storage::resolve_parseable_metadata().await?;
+
         banner::print(&CONFIG, &metadata).await;
         rbac::map::init(&metadata);
         // set the info in the global metadata
@@ -217,6 +216,7 @@ impl IngestServer {
 
     // create the ingestor metadata and put the .ingestor.json file in the object store
     async fn set_ingestor_metadata(&self) -> anyhow::Result<()> {
+        migrate_ingester_metadata().await?;
         let store = CONFIG.storage().get_object_store();
 
         // find the meta file in staging if not generate new metadata

server/src/migration/metadata_migration.rs

@@ -126,7 +126,13 @@ pub fn update_v3(mut storage_metadata: JsonValue) -> JsonValue {
 
 pub async fn migrate_ingester_metadata() -> anyhow::Result<()> {
     let imp = ingestor_metadata_path(None);
-    let bytes = CONFIG.storage().get_object_store().get_object(&imp).await?;
+    let bytes = match CONFIG.storage().get_object_store().get_object(&imp).await {
+        Ok(bytes) => bytes,
+        Err(_) => {
+            log::debug!("No metadata found for ingester. So migration is not required");
+            return Ok(());
+        }
+    };
     let mut json = serde_json::from_slice::<JsonValue>(&bytes)?;
     let meta = json
         .as_object_mut()

server/src/utils/arrow/flight.rs

@@ -15,7 +15,9 @@ use tonic::transport::Channel;
 
 pub fn get_query_from_ticket(req: Request<Ticket>) -> Result<QueryJson, Status> {
     if CONFIG.parseable.mode == Mode::Ingest {
-        let query = serde_json::from_slice::<JsonValue>(&req.into_inner().ticket)
+        let inner = req.into_inner().ticket;
+        dbg!(&inner);
+        let query = serde_json::from_slice::<JsonValue>(&inner)
             .map_err(|_| Status::failed_precondition("Ticket is not valid json"))?["query"]
             .as_str()
             .ok_or_else(|| Status::failed_precondition("query is not valid string"))?
@@ -31,8 +33,10 @@ pub fn get_query_from_ticket(req: Request<Ticket>) -> Result<QueryJson, Status>
         })
     } else {
         Ok(
-            serde_json::from_slice::<QueryJson>(&req.into_inner().ticket)
-                .map_err(|err| Status::internal(err.to_string()))?,
+            serde_json::from_slice::<QueryJson>(&req.into_inner().ticket).map_err(|err| {
+                dbg!(&err);
+                Status::internal(err.to_string())
+            })?,
         )
     }
 }