refactor

Author: nikhilsinhaparseable

src/catalog/mod.rs

@@ -261,9 +261,9 @@ async fn process_single_partition(
         handle_existing_partition(
             pos,
             partition_changes,
-            &mut meta.snapshot.manifest_list,
             storage,
             stream_name,
+            meta,
             events_ingested,
             ingestion_size,
             storage_size,
@@ -292,14 +292,15 @@ async fn process_single_partition(
 async fn handle_existing_partition(
     pos: usize,
     partition_changes: Vec<manifest::File>,
-    manifests: &mut [snapshot::ManifestItem],
     storage: Arc<dyn ObjectStorage>,
     stream_name: &str,
+    meta: &mut ObjectStoreFormat,
     events_ingested: u64,
     ingestion_size: u64,
     storage_size: u64,
     partition_lower: DateTime<Utc>,
 ) -> Result<Option<snapshot::ManifestItem>, ObjectStorageError> {
+    let manifests = &mut meta.snapshot.manifest_list;
     let path = partition_path(
         stream_name,
         manifests[pos].time_lower_bound,
@@ -329,7 +330,7 @@ async fn handle_existing_partition(
                 storage,
                 stream_name,
                 false,
-                ObjectStoreFormat::default(), // We don't have meta here, use default
+                meta.clone(),
                 events_ingested,
                 ingestion_size,
                 storage_size,

src/handlers/http/ingest.rs

@@ -29,6 +29,7 @@ use crate::event::error::EventError;
 use crate::event::format::known_schema::{self, KNOWN_SCHEMA_LIST};
 use crate::event::format::{self, EventFormat, LogSource, LogSourceEntry};
 use crate::event::{self, FORMAT_KEY, USER_AGENT_KEY};
+use crate::handlers::http::modal::utils::ingest_utils::validate_stream_for_ingestion;
 use crate::handlers::{
     CONTENT_TYPE_JSON, CONTENT_TYPE_PROTOBUF, EXTRACT_LOG_KEY, LOG_SOURCE_KEY,
     STREAM_NAME_HEADER_KEY, TELEMETRY_TYPE_KEY, TelemetryType,
@@ -117,20 +118,7 @@ pub async fn ingest(
 
     //if stream exists, fetch the stream log source
     //return error if the stream log source is otel traces or otel metrics
-    let stream = match PARSEABLE.get_stream(&stream_name) {
-        Ok(stream) => {
-            stream
-                .get_log_source()
-                .iter()
-                .find(|&stream_log_source_entry| {
-                    stream_log_source_entry.log_source_format != LogSource::OtelTraces
-                        && stream_log_source_entry.log_source_format != LogSource::OtelMetrics
-                })
-                .ok_or(PostError::IncorrectLogFormat(stream_name.clone()))?;
-            stream
-        }
-        Err(e) => return Err(PostError::from(e)),
-    };
+    let stream = validate_stream_for_ingestion(&stream_name)?;
 
     PARSEABLE
         .add_update_log_source(&stream_name, log_source_entry)
@@ -409,20 +397,7 @@ pub async fn post_event(
 
     //if stream exists, fetch the stream log source
     //return error if the stream log source is otel traces or otel metrics
-    let stream = match PARSEABLE.get_stream(&stream_name) {
-        Ok(stream) => {
-            stream
-                .get_log_source()
-                .iter()
-                .find(|&stream_log_source_entry| {
-                    stream_log_source_entry.log_source_format != LogSource::OtelTraces
-                        && stream_log_source_entry.log_source_format != LogSource::OtelMetrics
-                })
-                .ok_or(PostError::IncorrectLogFormat(stream_name.clone()))?;
-            stream
-        }
-        Err(e) => return Err(PostError::from(e)),
-    };
+    let stream = validate_stream_for_ingestion(&stream_name)?;
 
     if stream.get_time_partition().is_some() {
         return Err(PostError::CustomError(

src/handlers/http/modal/utils/ingest_utils.rs

@@ -23,7 +23,7 @@ use opentelemetry_proto::tonic::{
     logs::v1::LogsData, metrics::v1::MetricsData, trace::v1::TracesData,
 };
 use serde_json::Value;
-use std::collections::HashMap;
+use std::{collections::HashMap, sync::Arc};
 use tracing::warn;
 
 use crate::{
@@ -39,7 +39,7 @@ use crate::{
         },
     },
     otel::{logs::flatten_otel_logs, metrics::flatten_otel_metrics, traces::flatten_otel_traces},
-    parseable::PARSEABLE,
+    parseable::{PARSEABLE, Stream},
     storage::StreamType,
     utils::json::{convert_array_to_object, flatten::convert_to_array},
 };
@@ -268,6 +268,29 @@ fn verify_dataset_fields_count(stream_name: &str) -> Result<(), PostError> {
     Ok(())
 }
 
+pub fn validate_stream_for_ingestion(stream_name: &str) -> Result<Arc<Stream>, PostError> {
+    let stream = PARSEABLE.get_stream(stream_name)?;
+
+    // Validate that the stream's log source is compatible
+    stream
+        .get_log_source()
+        .iter()
+        .find(|&stream_log_source_entry| {
+            stream_log_source_entry.log_source_format != LogSource::OtelTraces
+                && stream_log_source_entry.log_source_format != LogSource::OtelMetrics
+        })
+        .ok_or(PostError::IncorrectLogFormat(stream_name.to_string()))?;
+
+    // Check for time partition
+    if stream.get_time_partition().is_some() {
+        return Err(PostError::CustomError(
+            "Ingestion is not allowed to stream with time partition".to_string(),
+        ));
+    }
+
+    Ok(stream)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

src/storage/object_storage.rs

@@ -114,17 +114,13 @@ async fn upload_single_parquet_file(
         .expect("filename is valid string");
 
     // Upload the file
-    if let Err(e) = store
+    store
         .upload_multipart(&RelativePathBuf::from(&stream_relative_path), &path)
         .await
-    {
-        error!("Failed to upload file {filename:?}: {e}");
-        return Ok(UploadResult {
-            stats_calculated: false,
-            file_path: path,
-            manifest_file: None,
-        });
-    }
+        .map_err(|e| {
+            error!("Failed to upload file {filename:?} to {stream_relative_path}: {e}");
+            ObjectStorageError::Custom(format!("Failed to upload {filename}: {e}"))
+        })?;
 
     // Update storage metrics
     update_storage_metrics(&path, &stream_name, filename)?;