Merge branch 'main' into schema-detect

Author: nikhilsinhaparseable

src/event/format/mod.rs

@@ -60,19 +60,25 @@ type EventSchema = Vec<Arc<Field>>;
 #[derive(Default, Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Hash)]
 pub enum LogSource {
     // AWS Kinesis sends logs in the format of a json array
+    #[serde(rename = "kinesis")]
     Kinesis,
     // OpenTelemetry sends logs according to the specification as explained here
     // https://github.com/open-telemetry/opentelemetry-proto/tree/v1.0.0/opentelemetry/proto/logs/v1
+    #[serde(rename = "otel-logs")]
     OtelLogs,
     // OpenTelemetry sends traces according to the specification as explained here
     // https://github.com/open-telemetry/opentelemetry-proto/blob/v1.0.0/opentelemetry/proto/trace/v1/trace.proto
+    #[serde(rename = "otel-metrics")]
     OtelMetrics,
     // OpenTelemetry sends traces according to the specification as explained here
     // https://github.com/open-telemetry/opentelemetry-proto/tree/v1.0.0/opentelemetry/proto/metrics/v1
+    #[serde(rename = "otel-traces")]
     OtelTraces,
     // Internal Stream format
+    #[serde(rename = "pmeta")]
     Pmeta,
     #[default]
+    #[serde(rename = "json")]
     // Json object or array
     Json,
     // Custom Log Sources e.g. "syslog"

src/handlers/http/query.rs

@@ -44,7 +44,7 @@ use crate::query::error::ExecuteError;
 use crate::query::{execute, CountsRequest, CountsResponse, Query as LogicalQuery};
 use crate::query::{TableScanVisitor, QUERY_SESSION};
 use crate::rbac::Users;
-use crate::response::QueryResponse;
+use crate::response::{QueryResponse, TIME_ELAPSED_HEADER};
 use crate::storage::object_storage::commit_schema_to_storage;
 use crate::storage::ObjectStorageError;
 use crate::utils::actix::extract_session_key_from_req;
@@ -122,22 +122,26 @@ pub async fn query(req: HttpRequest, query_request: Query) -> Result<HttpRespons
             Value::Array(vec![json!({column_name: count})])
         };
 
+        let total_time = format!("{:?}", time.elapsed());
         let time = time.elapsed().as_secs_f64();
 
         QUERY_EXECUTE_TIME
             .with_label_values(&[&table_name])
             .observe(time);
 
-        return Ok(HttpResponse::Ok().json(response));
+        return Ok(HttpResponse::Ok()
+            .insert_header((TIME_ELAPSED_HEADER, total_time.as_str()))
+            .json(response));
     }
 
     let (records, fields) = execute(query, &table_name).await?;
-
+    let total_time = format!("{:?}", time.elapsed());
     let response = QueryResponse {
         records,
         fields,
         fill_null: query_request.send_null,
         with_fields: query_request.fields,
+        total_time,
     }
     .to_http()?;
 

src/migration/mod.rs

@@ -246,6 +246,7 @@ async fn migrate_stream_metadata(
             stream_metadata_value = stream_metadata_migration::v1_v4(stream_metadata_value);
             stream_metadata_value = stream_metadata_migration::v4_v5(stream_metadata_value, stream);
             stream_metadata_value = stream_metadata_migration::v5_v6(stream_metadata_value);
+
             storage
                 .put_object(&path, to_bytes(&stream_metadata_value))
                 .await?;
@@ -259,6 +260,7 @@ async fn migrate_stream_metadata(
             stream_metadata_value = stream_metadata_migration::v2_v4(stream_metadata_value);
             stream_metadata_value = stream_metadata_migration::v4_v5(stream_metadata_value, stream);
             stream_metadata_value = stream_metadata_migration::v5_v6(stream_metadata_value);
+
             storage
                 .put_object(&path, to_bytes(&stream_metadata_value))
                 .await?;
@@ -272,13 +274,15 @@ async fn migrate_stream_metadata(
             stream_metadata_value = stream_metadata_migration::v3_v4(stream_metadata_value);
             stream_metadata_value = stream_metadata_migration::v4_v5(stream_metadata_value, stream);
             stream_metadata_value = stream_metadata_migration::v5_v6(stream_metadata_value);
+
             storage
                 .put_object(&path, to_bytes(&stream_metadata_value))
                 .await?;
         }
         Some("v4") => {
             stream_metadata_value = stream_metadata_migration::v4_v5(stream_metadata_value, stream);
             stream_metadata_value = stream_metadata_migration::v5_v6(stream_metadata_value);
+
             storage
                 .put_object(&path, to_bytes(&stream_metadata_value))
                 .await?;
@@ -289,7 +293,13 @@ async fn migrate_stream_metadata(
                 .put_object(&path, to_bytes(&stream_metadata_value))
                 .await?;
         }
-        _ => (),
+        _ => {
+            stream_metadata_value =
+                stream_metadata_migration::rename_log_source_v6(stream_metadata_value);
+            storage
+                .put_object(&path, to_bytes(&stream_metadata_value))
+                .await?;
+        }
     }
 
     Ok(stream_metadata_value)