diff --git a/src/handler/http/request/alerts/destinations.rs b/src/handler/http/request/alerts/destinations.rs index fef039bce8..74d0217436 100644 --- a/src/handler/http/request/alerts/destinations.rs +++ b/src/handler/http/request/alerts/destinations.rs @@ -49,12 +49,12 @@ pub async fn save_destination( Some(_) => destinations::save_destination(org_id, name, dest).await, None => Ok(HttpResponse::BadRequest().json(MetaHttpResponse::error( http::StatusCode::BAD_REQUEST.into(), - Some("Please specify valid template".to_string()), + "Please specify valid template".to_string(), ))), }, Err(_) => Ok(HttpResponse::BadRequest().json(MetaHttpResponse::error( http::StatusCode::BAD_REQUEST.into(), - Some("Please specify valid template".to_string()), + "Please specify valid template".to_string(), ))), } } diff --git a/src/handler/http/request/ingest/mod.rs b/src/handler/http/request/logs/ingest.rs similarity index 95% rename from src/handler/http/request/ingest/mod.rs rename to src/handler/http/request/logs/ingest.rs index 97b79fa44f..b835cf2638 100644 --- a/src/handler/http/request/ingest/mod.rs +++ b/src/handler/http/request/logs/ingest.rs @@ -20,8 +20,8 @@ use crate::service::logs; /** Elasticsearch compatible _bulk ingestion API */ #[utoipa::path( context_path = "/api", - tag = "Ingestion", - operation_id = "IngestionBulk", + tag = "Logs", + operation_id = "LogsIngestionBulk", security( ("Authorization"= []) ), @@ -47,8 +47,8 @@ pub async fn bulk( /** ndjson (newline delimited json) multi ingestion API */ #[utoipa::path( context_path = "/api", - tag = "Ingestion", - operation_id = "IngestionMulti", + tag = "Logs", + operation_id = "LogsIngestionMulti", security( ("Authorization"= []) ), @@ -75,8 +75,8 @@ pub async fn multi( /** json ingestion API, accepts array of json records */ #[utoipa::path( context_path = "/api", - tag = "Ingestion", - operation_id = "IngestionJson", + tag = "Logs", + operation_id = "LogsIngestionJson", security( ("Authorization"= []) ), diff --git a/src/handler/http/request/logs/mod.rs b/src/handler/http/request/logs/mod.rs new file mode 100644 index 0000000000..c3cc2f049f --- /dev/null +++ b/src/handler/http/request/logs/mod.rs @@ -0,0 +1,15 @@ +// Copyright 2022 Zinc Labs Inc. and Contributors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +pub mod ingest; diff --git a/src/handler/http/request/mod.rs b/src/handler/http/request/mod.rs index 173980fda1..3a8f9590f6 100644 --- a/src/handler/http/request/mod.rs +++ b/src/handler/http/request/mod.rs @@ -15,7 +15,7 @@ pub mod alerts; pub mod dashboards; pub mod functions; -pub mod ingest; +pub mod logs; pub mod organization; pub mod prom; pub mod search; diff --git a/src/handler/http/request/prom/mod.rs b/src/handler/http/request/prom/mod.rs index dbf54e3f33..f83a6ad9d0 100644 --- a/src/handler/http/request/prom/mod.rs +++ b/src/handler/http/request/prom/mod.rs @@ -32,7 +32,7 @@ pub async fn prometheus_write( Ok( HttpResponse::BadRequest().json(meta::http::HttpResponse::error( http::StatusCode::BAD_REQUEST.into(), - Some("Bad Request".to_string()), + "Bad Request".to_string(), )), ) } diff --git a/src/handler/http/request/search/mod.rs b/src/handler/http/request/search/mod.rs index 3393c93fdc..3ac2dc7dfb 100644 --- a/src/handler/http/request/search/mod.rs +++ b/src/handler/http/request/search/mod.rs @@ -22,10 +22,9 @@ use std::time::Instant; use crate::common::http::get_stream_type_from_request; use crate::infra::config::{CONFIG, LOCKER}; -use crate::infra::metrics; -use crate::meta; +use crate::infra::{errors, metrics}; use crate::meta::http::HttpResponse as MetaHttpResponse; -use crate::meta::StreamType; +use crate::meta::{self, StreamType}; use crate::service::search as SearchService; /** Search using SQL against a stream data */ @@ -106,7 +105,7 @@ pub async fn search( Err(e) => { return Ok(HttpResponse::BadRequest().json(MetaHttpResponse::error( http::StatusCode::BAD_REQUEST.into(), - Some(e.to_string()), + e.to_string(), ))) } }; @@ -117,14 +116,14 @@ pub async fn search( Err(e) => { return Ok(HttpResponse::BadRequest().json(MetaHttpResponse::error( http::StatusCode::BAD_REQUEST.into(), - Some(e.to_string()), + e.to_string(), ))) } }; if let Err(e) = req.decode() { return Ok(HttpResponse::BadRequest().json(MetaHttpResponse::error( http::StatusCode::BAD_REQUEST.into(), - Some(e.to_string()), + e.to_string(), ))); } @@ -159,7 +158,6 @@ pub async fn search( Ok(HttpResponse::Ok().json(res)) } Err(err) => { - log::error!("search error: {:?}", err); let time = start.elapsed().as_secs_f64(); metrics::HTTP_RESPONSE_TIME .with_label_values(&[ @@ -179,12 +177,15 @@ pub async fn search( stream_type.to_string().as_str(), ]) .inc(); - Ok( - HttpResponse::InternalServerError().json(meta::http::HttpResponse::error( + log::error!("search error: {:?}", err); + Ok(match err { + errors::Error::ErrorCode(code) => HttpResponse::InternalServerError() + .json(meta::http::HttpResponse::error_code(code)), + _ => HttpResponse::InternalServerError().json(meta::http::HttpResponse::error( StatusCode::INTERNAL_SERVER_ERROR.into(), - Some(err.to_string()), + err.to_string(), )), - ) + }) } } } @@ -245,7 +246,7 @@ pub async fn around( Err(e) => { return Ok(HttpResponse::BadRequest().json(MetaHttpResponse::error( http::StatusCode::BAD_REQUEST.into(), - Some(e.to_string()), + e.to_string(), ))) } }; @@ -285,7 +286,6 @@ pub async fn around( let resp_forward = match SearchService::search(&org_id, stream_type, &req).await { Ok(res) => res, Err(err) => { - log::error!("search error: {:?}", err); let time = start.elapsed().as_secs_f64(); metrics::HTTP_RESPONSE_TIME .with_label_values(&[ @@ -305,12 +305,15 @@ pub async fn around( stream_type.to_string().as_str(), ]) .inc(); - return Ok( - HttpResponse::InternalServerError().json(meta::http::HttpResponse::error( + log::error!("search around error: {:?}", err); + return Ok(match err { + errors::Error::ErrorCode(code) => HttpResponse::InternalServerError() + .json(meta::http::HttpResponse::error_code(code)), + _ => HttpResponse::InternalServerError().json(meta::http::HttpResponse::error( StatusCode::INTERNAL_SERVER_ERROR.into(), - Some(err.to_string()), + err.to_string(), )), - ); + }); } }; @@ -334,7 +337,6 @@ pub async fn around( let resp_backward = match SearchService::search(&org_id, stream_type, &req).await { Ok(res) => res, Err(err) => { - log::error!("search error: {:?}", err); let time = start.elapsed().as_secs_f64(); metrics::HTTP_RESPONSE_TIME .with_label_values(&[ @@ -354,12 +356,15 @@ pub async fn around( stream_type.to_string().as_str(), ]) .inc(); - return Ok( - HttpResponse::InternalServerError().json(meta::http::HttpResponse::error( + log::error!("search around error: {:?}", err); + return Ok(match err { + errors::Error::ErrorCode(code) => HttpResponse::InternalServerError() + .json(meta::http::HttpResponse::error_code(code)), + _ => HttpResponse::InternalServerError().json(meta::http::HttpResponse::error( StatusCode::INTERNAL_SERVER_ERROR.into(), - Some(err.to_string()), + err.to_string(), )), - ); + }); } }; diff --git a/src/handler/http/request/stream/mod.rs b/src/handler/http/request/stream/mod.rs index 5f82738c7d..2e20552679 100644 --- a/src/handler/http/request/stream/mod.rs +++ b/src/handler/http/request/stream/mod.rs @@ -51,7 +51,7 @@ async fn schema( return Ok( HttpResponse::BadRequest().json(meta::http::HttpResponse::error( http::StatusCode::BAD_REQUEST.into(), - Some(e.to_string()), + e.to_string(), )), ) } @@ -92,7 +92,7 @@ async fn settings( return Ok( HttpResponse::BadRequest().json(meta::http::HttpResponse::error( http::StatusCode::BAD_REQUEST.into(), - Some(e.to_string()), + e.to_string(), )), ) } @@ -131,7 +131,7 @@ async fn delete( return Ok( HttpResponse::BadRequest().json(meta::http::HttpResponse::error( http::StatusCode::BAD_REQUEST.into(), - Some(e.to_string()), + e.to_string(), )), ) } @@ -165,7 +165,7 @@ async fn list(org_id: web::Path, req: HttpRequest) -> impl Responder { return Ok( HttpResponse::BadRequest().json(meta::http::HttpResponse::error( http::StatusCode::BAD_REQUEST.into(), - Some(e.to_string()), + e.to_string(), )), ) } diff --git a/src/handler/http/request/traces/mod.rs b/src/handler/http/request/traces/mod.rs index 6d37d7f281..45ff0d5ec1 100644 --- a/src/handler/http/request/traces/mod.rs +++ b/src/handler/http/request/traces/mod.rs @@ -37,7 +37,7 @@ pub async fn traces_write( Ok( HttpResponse::BadRequest().json(meta::http::HttpResponse::error( http::StatusCode::BAD_REQUEST.into(), - Some("Bad Request".to_string()), + "Bad Request".to_string(), )), ) } diff --git a/src/handler/http/request/users/mod.rs b/src/handler/http/request/users/mod.rs index 2476202f5e..bc99384c9a 100644 --- a/src/handler/http/request/users/mod.rs +++ b/src/handler/http/request/users/mod.rs @@ -99,7 +99,7 @@ pub async fn update( return Ok( HttpResponse::BadRequest().json(meta::http::HttpResponse::error( http::StatusCode::BAD_REQUEST.into(), - Some("Please specify appropriate fields to update user".to_string()), + "Please specify appropriate fields to update user".to_string(), )), ); } diff --git a/src/handler/http/router/mod.rs b/src/handler/http/router/mod.rs index 9d5bd31681..68435e446b 100644 --- a/src/handler/http/router/mod.rs +++ b/src/handler/http/router/mod.rs @@ -23,7 +23,7 @@ use super::auth::validator; use super::request::alerts::*; use super::request::dashboards::*; use super::request::functions; -use super::request::ingest; +use super::request::logs; use super::request::organization::*; use super::request::prom::*; use super::request::search; @@ -91,9 +91,9 @@ pub fn get_service_routes(cfg: &mut web::ServiceConfig) { .wrap(auth) .wrap(cors) .service(status::cache_status) - .service(ingest::bulk) - .service(ingest::multi) - .service(ingest::json) + .service(logs::ingest::bulk) + .service(logs::ingest::multi) + .service(logs::ingest::json) .service(search::search) .service(search::around) .service(stream::schema) diff --git a/src/handler/http/router/openapi.rs b/src/handler/http/router/openapi.rs index 20b35efa7d..362149cdd5 100644 --- a/src/handler/http/router/openapi.rs +++ b/src/handler/http/router/openapi.rs @@ -25,9 +25,9 @@ use crate::meta; request::stream::schema, request::stream::settings, request::stream::delete, - request::ingest::bulk, - request::ingest::multi, - request::ingest::json, + request::logs::ingest::bulk, + request::logs::ingest::multi, + request::logs::ingest::json, request::search::search, request::search::around, request::functions::list_functions, @@ -108,7 +108,7 @@ use crate::meta; modifiers(&SecurityAddon), tags( (name = "Auth", description = "User login authentication"), - (name = "Ingestion", description = "Data ingestion operations"), + (name = "Logs", description = "Logs data ingestion operations"), (name = "Search", description = "Search/Query operations"), (name = "Alerts", description = "Alerts retrieval & management operations"), (name = "Functions", description = "Functions retrieval & management operations"), diff --git a/src/infra/errors.rs b/src/infra/errors.rs index aefb492731..e9acb84860 100644 --- a/src/infra/errors.rs +++ b/src/infra/errors.rs @@ -28,10 +28,14 @@ pub enum Error { SerdeJsonError(#[from] serde_json::Error), #[error("SimdJsonError# {0}")] SimdJsonError(#[from] simd_json::Error), + #[error("DataFusionError# {0}")] + DataFusionError(#[from] datafusion::error::DataFusionError), #[error("Error# watcher is exists {0}")] WatcherExists(String), #[error("Error# {0}")] Message(String), + #[error("ErrorCode# {0}")] + ErrorCode(ErrorCodes), #[error("Not implemented")] NotImplemented, #[error("Unknown error")] @@ -44,6 +48,62 @@ pub enum DbError { KeyNotExists(String), } +#[derive(ThisError, Debug)] +pub enum ErrorCodes { + ServerInternalError(String), + SearchSQLNotValid(String), + SearchStreamNotFound(String), + FullTextSearchFieldNotFound, + SearchFieldNotFound, + SearchFunctionNotDefined, + SearchParquetFileNotFound, + SearchFieldHasNoCompatibleDataType, +} + +impl std::fmt::Display for ErrorCodes { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!( + f, + r#"{{"error_code": {}, "error_msg": "{}"}}"#, + self.get_code(), + self.get_message() + ) + } +} + +impl ErrorCodes { + pub fn get_code(&self) -> u16 { + match self { + ErrorCodes::ServerInternalError(_) => 10001, + ErrorCodes::SearchSQLNotValid(_) => 20001, + ErrorCodes::SearchStreamNotFound(_) => 20002, + ErrorCodes::FullTextSearchFieldNotFound => 20003, + ErrorCodes::SearchFieldNotFound => 20004, + ErrorCodes::SearchFunctionNotDefined => 20005, + ErrorCodes::SearchParquetFileNotFound => 20006, + ErrorCodes::SearchFieldHasNoCompatibleDataType => 20007, + } + } + pub fn get_message(&self) -> String { + match self { + ErrorCodes::ServerInternalError(msg) => msg.to_owned(), + ErrorCodes::SearchSQLNotValid(sql) => format!("Search SQL not valid, sql: [{}]", sql), + ErrorCodes::SearchStreamNotFound(stream) => { + format!("Search stream not found, stream: [{}]", stream) + } + ErrorCodes::FullTextSearchFieldNotFound => { + "Full text search field not found".to_string() + } + ErrorCodes::SearchFieldNotFound => "Search field not found".to_string(), + ErrorCodes::SearchFunctionNotDefined => "Search function not defined".to_string(), + ErrorCodes::SearchParquetFileNotFound => "Search parquet file not found".to_string(), + ErrorCodes::SearchFieldHasNoCompatibleDataType => { + "Search field has no compatible data type".to_string() + } + } + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/meta/http.rs b/src/meta/http.rs index 12efe1f48c..2064483684 100644 --- a/src/meta/http.rs +++ b/src/meta/http.rs @@ -15,13 +15,20 @@ use serde::{Deserialize, Serialize}; use utoipa::ToSchema; +use crate::infra::errors; + +/// HTTP response +/// code 200 is success +/// code 400 is error +/// code 404 is not found +/// code 500 is internal server error +/// code 503 is service unavailable +/// code >= 1000 is custom error code +/// message is the message or error message #[derive(Clone, Debug, Serialize, Deserialize, ToSchema)] pub struct HttpResponse { pub code: u16, - #[serde(skip_serializing_if = "String::is_empty")] pub message: String, - #[serde(skip_serializing_if = "Option::is_none")] - pub error: Option, } #[derive(Clone, Debug, Serialize, Deserialize)] @@ -32,41 +39,49 @@ pub struct ESResponse { } impl HttpResponse { - pub fn _new(code: u16, message: String, error: Option) -> Self { - HttpResponse { - code, - message, - error, - } - } - pub fn message(code: u16, message: String) -> Self { + HttpResponse { code, message } + } + + pub fn error(code: u16, error: String) -> Self { HttpResponse { code, - message, - error: None, + message: error, } } - pub fn error(code: u16, error: Option) -> Self { + pub fn error_code(err: errors::ErrorCodes) -> Self { HttpResponse { - code, - message: "".to_string(), - error, + code: err.get_code(), + message: err.get_message(), } } } + #[cfg(test)] mod test { - use super::*; use actix_web::http; + + use super::*; + #[test] - fn test_err_response() { + fn test_http_response() { let msg = "This is an error response"; + let err = HttpResponse::message(http::StatusCode::OK.into(), msg.to_string()); + assert_eq!(err.code, http::StatusCode::OK); + assert_eq!(err.message, msg); + let err = HttpResponse::error( http::StatusCode::INTERNAL_SERVER_ERROR.into(), - Some(msg.to_string()), + msg.to_string(), ); - assert_eq!(err.error.unwrap(), msg); + assert_eq!(err.code, http::StatusCode::INTERNAL_SERVER_ERROR); + assert_eq!(err.message, msg); + + let errcode = errors::ErrorCodes::ServerInternalError(msg.to_string()); + let err = + HttpResponse::error_code(errors::ErrorCodes::ServerInternalError(msg.to_string())); + assert_eq!(err.code, errcode.get_code()); + assert_eq!(err.message, errcode.get_message()); } } diff --git a/src/service/alert_manager.rs b/src/service/alert_manager.rs index 2846823a11..8a5cb9b6a7 100644 --- a/src/service/alert_manager.rs +++ b/src/service/alert_manager.rs @@ -137,7 +137,7 @@ pub async fn handle_trigger(alert_name: &str, alert: Alert) { } } Err(err) => { - log::error!("search error: {:?}", err); + log::error!("alert_manager search error: {:?}", err); } } //} diff --git a/src/service/alerts/destinations.rs b/src/service/alerts/destinations.rs index 0a3393d679..7af0db8e4a 100644 --- a/src/service/alerts/destinations.rs +++ b/src/service/alerts/destinations.rs @@ -56,9 +56,9 @@ pub async fn delete_destination(org_id: String, name: String) -> Result Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( + Err(e) => Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( http::StatusCode::NOT_FOUND.into(), - Some(err.to_string()), + e.to_string(), ))), } } @@ -71,7 +71,7 @@ pub async fn get_destination(org_id: String, name: String) -> Result Ok(HttpResponse::Ok().json(alert)), Err(_) => Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( http::StatusCode::NOT_FOUND.into(), - Some("Alert destination not found".to_string()), + "Alert destination not found".to_string(), ))), } } diff --git a/src/service/alerts/mod.rs b/src/service/alerts/mod.rs index e28f21d1d7..89bc5279fa 100644 --- a/src/service/alerts/mod.rs +++ b/src/service/alerts/mod.rs @@ -52,7 +52,7 @@ pub async fn save_alert( if dest.is_none() { return Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( http::StatusCode::NOT_FOUND.into(), - Some(format!("Destination with name {} not found", in_dest)), + format!("Destination with name {} not found", in_dest), ))); } @@ -64,7 +64,7 @@ pub async fn save_alert( if fields.is_empty() { return Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( http::StatusCode::NOT_FOUND.into(), - Some(format!("Stream with name {} not found", stream_name)), + format!("Stream with name {} not found", stream_name), ))); } @@ -104,10 +104,10 @@ pub async fn save_alert( } else { return Ok(HttpResponse::BadRequest().json(MetaHttpResponse::error( http::StatusCode::BAD_REQUEST.into(), - Some(format!( + format!( "Column named {} not found on stream {}", &alert.condition.column, stream_name - )), + ), ))); } //} @@ -123,7 +123,7 @@ pub async fn save_alert( if sql.is_err() { return Ok(HttpResponse::BadRequest().json(MetaHttpResponse::error( http::StatusCode::BAD_REQUEST.into(), - Some(format!("Invalid query : {:?} ", sql.err())), + format!("Invalid query : {:?} ", sql.err()), ))); } } @@ -179,9 +179,9 @@ pub async fn delete_alert( http::StatusCode::OK.into(), "Alert deleted ".to_string(), ))), - Err(err) => Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( + Err(e) => Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( http::StatusCode::NOT_FOUND.into(), - Some(err.to_string()), + e.to_string(), ))), } } @@ -198,7 +198,7 @@ pub async fn get_alert( Ok(alert) => Ok(HttpResponse::Ok().json(alert)), Err(_) => Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( http::StatusCode::NOT_FOUND.into(), - Some("alert not found".to_string()), + "alert not found".to_string(), ))), } } diff --git a/src/service/alerts/templates.rs b/src/service/alerts/templates.rs index b8150d4222..6831a194db 100644 --- a/src/service/alerts/templates.rs +++ b/src/service/alerts/templates.rs @@ -56,9 +56,9 @@ pub async fn delete_template(org_id: String, name: String) -> Result Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( + Err(e) => Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( http::StatusCode::NOT_FOUND.into(), - Some(err.to_string()), + e.to_string(), ))), } } @@ -71,7 +71,7 @@ pub async fn get_template(org_id: String, name: String) -> Result Ok(HttpResponse::Ok().json(alert)), Err(_) => Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( http::StatusCode::NOT_FOUND.into(), - Some("Alert template not found".to_string()), + "Alert template not found".to_string(), ))), } } diff --git a/src/service/dashboards.rs b/src/service/dashboards.rs index 3251070012..8e9c4a6d6e 100644 --- a/src/service/dashboards.rs +++ b/src/service/dashboards.rs @@ -31,11 +31,11 @@ pub async fn get_dashboard(org_id: &str, name: &str) -> Result Ok(HttpResponse::Ok().json(dashboard)), Ok(None) => Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( StatusCode::NOT_FOUND.into(), - Some("Dashboard not found".to_string()), + "Dashboard not found".to_string(), ))), Err(_) => Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( StatusCode::NOT_FOUND.into(), - Some("Dashboard not found".to_string()), + "Dashboard not found".to_string(), ))), } } @@ -53,10 +53,10 @@ pub async fn save_dashboard( http::StatusCode::OK.into(), "Dashboard saved".to_string(), ))), - Err(err) => Ok( + Err(e) => Ok( HttpResponse::InternalServerError().json(meta::http::HttpResponse::error( StatusCode::BAD_REQUEST.into(), - Some(err.to_string()), + e.to_string(), )), ), } @@ -78,9 +78,9 @@ pub async fn delete_dashboard(org_id: &str, name: &str) -> Result Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( + Err(e) => Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( StatusCode::NOT_FOUND.into(), - Some(err.to_string()), + e.to_string(), ))), } } diff --git a/src/service/functions.rs b/src/service/functions.rs index f39ee5afa1..f956d279ad 100644 --- a/src/service/functions.rs +++ b/src/service/functions.rs @@ -61,7 +61,7 @@ pub async fn register_function( Ok( HttpResponse::BadRequest().json(meta::http::HttpResponse::error( http::StatusCode::BAD_REQUEST.into(), - Some(msg.to_string()), + msg.to_string(), )), ) } else { @@ -102,7 +102,7 @@ pub async fn register_function( Ok( HttpResponse::BadRequest().json(meta::http::HttpResponse::error( http::StatusCode::BAD_REQUEST.into(), - Some("Not valid function".to_string()), + "Not valid function".to_string(), )), ) } @@ -135,9 +135,9 @@ pub async fn delete_function( http::StatusCode::OK.into(), DELETED.to_string(), ))), - Err(err) => Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( + Err(e) => Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( StatusCode::NOT_FOUND.into(), - Some(err.to_string()), + e.to_string(), ))), } } diff --git a/src/service/logs/bulk.rs b/src/service/logs/bulk.rs index 566a360566..9a9e046dd9 100644 --- a/src/service/logs/bulk.rs +++ b/src/service/logs/bulk.rs @@ -53,7 +53,7 @@ pub async fn ingest( return Ok( HttpResponse::InternalServerError().json(MetaHttpResponse::error( http::StatusCode::INTERNAL_SERVER_ERROR.into(), - Some("not an ingester".to_string()), + "not an ingester".to_string(), )), ); } @@ -102,7 +102,7 @@ pub async fn ingest( return Ok( HttpResponse::InternalServerError().json(MetaHttpResponse::error( http::StatusCode::INTERNAL_SERVER_ERROR.into(), - Some(format!("stream [{}] is being deleted", stream_name)), + format!("stream [{}] is being deleted", stream_name), )), ); } diff --git a/src/service/logs/json.rs b/src/service/logs/json.rs index ea951e2d70..da694ec622 100644 --- a/src/service/logs/json.rs +++ b/src/service/logs/json.rs @@ -53,7 +53,7 @@ pub async fn ingest( return Ok( HttpResponse::InternalServerError().json(MetaHttpResponse::error( http::StatusCode::INTERNAL_SERVER_ERROR.into(), - Some("not an ingester".to_string()), + "not an ingester".to_string(), )), ); } @@ -63,7 +63,7 @@ pub async fn ingest( return Ok( HttpResponse::InternalServerError().json(MetaHttpResponse::error( http::StatusCode::INTERNAL_SERVER_ERROR.into(), - Some(format!("stream [{}] is being deleted", stream_name)), + format!("stream [{}] is being deleted", stream_name), )), ); } diff --git a/src/service/logs/multi.rs b/src/service/logs/multi.rs index fd62dc2f4f..1b677f9f3e 100644 --- a/src/service/logs/multi.rs +++ b/src/service/logs/multi.rs @@ -54,7 +54,7 @@ pub async fn ingest( return Ok( HttpResponse::InternalServerError().json(MetaHttpResponse::error( http::StatusCode::INTERNAL_SERVER_ERROR.into(), - Some("not an ingester".to_string()), + "not an ingester".to_string(), )), ); } @@ -64,7 +64,7 @@ pub async fn ingest( return Ok( HttpResponse::InternalServerError().json(MetaHttpResponse::error( http::StatusCode::INTERNAL_SERVER_ERROR.into(), - Some(format!("stream [{}] is being deleted", stream_name)), + format!("stream [{}] is being deleted", stream_name), )), ); } diff --git a/src/service/metrics/mod.rs b/src/service/metrics/mod.rs index afcb2c21ba..149f33f31f 100644 --- a/src/service/metrics/mod.rs +++ b/src/service/metrics/mod.rs @@ -66,7 +66,7 @@ pub async fn prometheus_write_proto( return Ok( HttpResponse::InternalServerError().json(meta::http::HttpResponse::error( http::StatusCode::INTERNAL_SERVER_ERROR.into(), - Some("not an ingester".to_string()), + "not an ingester".to_string(), )), ); } @@ -216,7 +216,7 @@ pub async fn prometheus_write_proto( return Ok(HttpResponse::InternalServerError().json( meta::http::HttpResponse::error( http::StatusCode::INTERNAL_SERVER_ERROR.into(), - Some(format!("stream [{}] is being deleted", metric_name)), + format!("stream [{}] is being deleted", metric_name), ), )); } diff --git a/src/service/search.rs b/src/service/search.rs index aa2a9285a4..afe73b1d77 100644 --- a/src/service/search.rs +++ b/src/service/search.rs @@ -14,6 +14,7 @@ use ::datafusion::arrow::{datatypes::Schema, ipc, json as arrow_json, record_batch::RecordBatch}; use ahash::AHashMap as HashMap; +use datafusion_common::DataFusionError; use http_auth_basic::Credentials; use std::io::Cursor; use std::sync::Arc; @@ -27,11 +28,11 @@ use crate::handler::grpc::cluster_rpc; use crate::infra::cluster; use crate::infra::config::{CONFIG, ROOT_USER}; use crate::infra::db::etcd; +use crate::infra::errors::{Error, ErrorCodes}; use crate::meta; use crate::meta::search::Response; use crate::meta::StreamType; -use crate::service::db; -use crate::service::file_list; +use crate::service::{db, file_list}; mod cache; pub mod datafusion; @@ -39,13 +40,13 @@ pub mod exec; pub mod sql; mod storage; -pub type SearchResult = Result<(HashMap>, usize, usize), anyhow::Error>; +pub type SearchResult = Result<(HashMap>, usize, usize), Error>; pub async fn search( org_id: &str, stream_type: StreamType, req: &meta::search::Request, -) -> Result { +) -> Result { let root_span = info_span!("service:search:enter"); let mut req: cluster_rpc::SearchRequest = req.to_owned().into(); @@ -72,20 +73,19 @@ pub async fn search( if result.is_err() { let err = result.err(); - log::error!("search error: {:?}", err); - return Err(anyhow::anyhow!("error: {:?}", err)); + return Err(Error::ErrorCode(ErrorCodes::ServerInternalError(format!( + "{:?}", + err + )))); } match result.unwrap() { Ok(res) => Ok(res), - Err(err) => { - log::error!("search error: {:?}", err); - Err(anyhow::anyhow!("error: {:?}", err)) - } + Err(err) => Err(err), } } #[tracing::instrument(name = "service:search:local:enter", skip(req))] -async fn search_in_local(req: cluster_rpc::SearchRequest) -> Result { +async fn search_in_local(req: cluster_rpc::SearchRequest) -> Result { let resp = match exec::search(&req).await { Ok(res) => res, Err(err) => return Err(err), @@ -104,7 +104,14 @@ async fn search_in_local(req: cluster_rpc::SearchRequest) -> Result>(); - let json_rows = arrow_json::writer::record_batches_to_json_rows(&batches[..])?; + let json_rows = match arrow_json::writer::record_batches_to_json_rows(&batches[..]) { + Ok(res) => res, + Err(err) => { + return Err(Error::ErrorCode(ErrorCodes::ServerInternalError( + err.to_string(), + ))); + } + }; let sources: Vec = json_rows .into_iter() .map(serde_json::Value::Object) @@ -125,7 +132,14 @@ async fn search_in_local(req: cluster_rpc::SearchRequest) -> Result>(); - let json_rows = arrow_json::writer::record_batches_to_json_rows(&batches[..])?; + let json_rows = match arrow_json::writer::record_batches_to_json_rows(&batches[..]) { + Ok(res) => res, + Err(err) => { + return Err(Error::ErrorCode(ErrorCodes::ServerInternalError( + err.to_string(), + ))); + } + }; let sources: Vec = json_rows .into_iter() .map(serde_json::Value::Object) @@ -155,7 +169,7 @@ async fn search_in_local(req: cluster_rpc::SearchRequest) -> Result Result { +async fn search_in_cluster(req: cluster_rpc::SearchRequest) -> Result { // start time let start = std::time::Instant::now(); let stream_type: StreamType = StreamType::from(req.stream_type.as_str()); @@ -167,8 +181,9 @@ async fn search_in_cluster(req: cluster_rpc::SearchRequest) -> Result res, Err(err) => { - log::error!("search in cluster get lock error: {}", err); - return Err(anyhow::anyhow!(err)); + return Err(Error::ErrorCode(ErrorCodes::ServerInternalError( + err.to_string(), + ))); } }; @@ -388,7 +403,14 @@ async fn search_in_cluster(req: cluster_rpc::SearchRequest) -> Result res, + Err(err) => { + return Err(Error::ErrorCode(ErrorCodes::ServerInternalError( + err.to_string(), + ))); + } + }; match result { Ok(res) => { results.push(res); @@ -398,7 +420,9 @@ async fn search_in_cluster(req: cluster_rpc::SearchRequest) -> Result Result res, - Err(err) => return Err(anyhow::anyhow!(err)), + Err(err) => { + return Err(Error::ErrorCode(ErrorCodes::ServerInternalError( + err.to_string(), + ))); + } }; } } @@ -490,7 +518,15 @@ async fn search_in_cluster(req: cluster_rpc::SearchRequest) -> Result Vec::new(), }; if !batches_query.is_empty() { - let json_rows = arrow_json::writer::record_batches_to_json_rows(&batches_query[0][..])?; + let json_rows = match arrow_json::writer::record_batches_to_json_rows(&batches_query[0][..]) + { + Ok(res) => res, + Err(err) => { + return Err(Error::ErrorCode(ErrorCodes::ServerInternalError( + err.to_string(), + ))) + } + }; let sources: Vec = json_rows .into_iter() .map(serde_json::Value::Object) @@ -506,7 +542,14 @@ async fn search_in_cluster(req: cluster_rpc::SearchRequest) -> Result res, + Err(err) => { + return Err(Error::ErrorCode(ErrorCodes::ServerInternalError( + err.to_string(), + ))); + } + }; let sources: Vec = json_rows .into_iter() .map(serde_json::Value::Object) @@ -538,6 +581,25 @@ async fn search_in_cluster(req: cluster_rpc::SearchRequest) -> Result Error { + let err = err.to_string(); + if err.contains("Schema error: No field named") { + return Error::ErrorCode(ErrorCodes::SearchFieldNotFound); + } + if err.contains("parquet not found") { + return Error::ErrorCode(ErrorCodes::SearchParquetFileNotFound); + } + if err.contains("Invalid function ") { + return Error::ErrorCode(ErrorCodes::SearchFunctionNotDefined); + } + if err.contains("Incompatible data types") { + return Error::ErrorCode(ErrorCodes::SearchFieldHasNoCompatibleDataType); + } + Error::ErrorCode(ErrorCodes::ServerInternalError( + "sql execute error".to_string(), + )) +} + struct MetadataMap<'a>(&'a mut tonic::metadata::MetadataMap); impl<'a> opentelemetry::propagation::Injector for MetadataMap<'a> { diff --git a/src/service/search/cache.rs b/src/service/search/cache.rs index 8b1d83ceda..2e464530fe 100644 --- a/src/service/search/cache.rs +++ b/src/service/search/cache.rs @@ -23,6 +23,7 @@ use super::datafusion::storage::file_list::SessionType; use super::sql::Sql; use crate::common::file::scan_files; use crate::infra::config::{self, CONFIG}; +use crate::infra::errors::{Error, ErrorCodes}; use crate::meta; use crate::service::db; use crate::service::file_list::calculate_local_files_size; @@ -54,7 +55,15 @@ pub async fn search( let span2 = info_span!("service:search:cache:calculate_files_size"); let guard2 = span2.enter(); - let scan_size = calculate_local_files_size(&files).await?; + let scan_size = match calculate_local_files_size(&files).await { + Ok(size) => size, + Err(err) => { + log::error!("calculate files size error: {}", err); + return Err(Error::ErrorCode(ErrorCodes::ServerInternalError( + "calculate files size error".to_string(), + ))); + } + }; log::info!( "[TRACE] cache->search: load files {}, scan_size {}", file_count, @@ -66,7 +75,15 @@ pub async fn search( let _guard3 = span3.enter(); // fetch all schema versions, get latest schema - let schema = db::schema::get(&sql.org_id, &sql.stream_name, Some(stream_type)).await?; + let schema = match db::schema::get(&sql.org_id, &sql.stream_name, Some(stream_type)).await { + Ok(schema) => schema, + Err(err) => { + log::error!("get schema error: {}", err); + return Err(Error::ErrorCode(ErrorCodes::SearchStreamNotFound( + sql.stream_name.clone(), + ))); + } + }; let schema = Arc::new( schema .to_owned() @@ -77,7 +94,7 @@ pub async fn search( id: session_id.to_string(), data_type: SessionType::Cache, }; - let result = super::datafusion::exec::sql( + let result = match super::datafusion::exec::sql( &session, stream_type, Some(schema), @@ -86,7 +103,14 @@ pub async fn search( &files, FileType::JSON, ) - .await?; + .await + { + Ok(res) => res, + Err(err) => { + log::error!("datafusion execute error: {}", err); + return Err(super::handle_datafusion_error(err)); + } + }; // searching done. drop(searching); @@ -96,10 +120,7 @@ pub async fn search( /// get file list from local cache, no need match_source, each file will be searched #[inline] -async fn get_file_list( - sql: &Sql, - stream_type: meta::StreamType, -) -> Result, anyhow::Error> { +async fn get_file_list(sql: &Sql, stream_type: meta::StreamType) -> Result, Error> { let pattern = format!( "{}/files/{}/{}/{}/*.json", &CONFIG.common.data_wal_dir, &sql.org_id, stream_type, &sql.stream_name diff --git a/src/service/search/exec.rs b/src/service/search/exec.rs index 91d6bab373..d8c338e103 100644 --- a/src/service/search/exec.rs +++ b/src/service/search/exec.rs @@ -19,12 +19,13 @@ use tracing::{info_span, Instrument}; use crate::handler::grpc::cluster_rpc; use crate::infra::config::CONFIG; +use crate::infra::errors::{Error, ErrorCodes}; use crate::meta::StreamType; #[tracing::instrument(name = "service:search:exec", skip(req))] pub async fn search( req: &cluster_rpc::SearchRequest, -) -> Result { +) -> Result { let start = std::time::Instant::now(); let sql = super::sql::Sql::new(req).await; if sql.is_err() { @@ -84,7 +85,11 @@ pub async fn search( Ok((search_result, file_count, scan_size)) => (search_result, file_count, scan_size), Err(err) => return Err(err), }, - Err(err) => return Err(anyhow::anyhow!(err)), + Err(err) => { + return Err(Error::ErrorCode(ErrorCodes::ServerInternalError( + err.to_string(), + ))) + } }; if !batches1.is_empty() { @@ -104,7 +109,11 @@ pub async fn search( Ok((search_result, file_count, scan_size)) => (search_result, file_count, scan_size), Err(err) => return Err(err), }, - Err(err) => return Err(anyhow::anyhow!(err)), + Err(err) => { + return Err(Error::ErrorCode(ErrorCodes::ServerInternalError( + err.to_string(), + ))) + } }; if !batches2.is_empty() { @@ -142,7 +151,10 @@ pub async fn search( .await { Ok(res) => res, - Err(err) => return Err(anyhow::anyhow!(err)), + Err(err) => { + log::error!("datafusion merge error: {}", err); + return Err(super::handle_datafusion_error(err)); + } }; } diff --git a/src/service/search/sql.rs b/src/service/search/sql.rs index 49b0258aaf..4fc906942d 100644 --- a/src/service/search/sql.rs +++ b/src/service/search/sql.rs @@ -23,6 +23,7 @@ use std::fmt::{Display, Formatter}; use crate::common::str; use crate::handler::grpc::cluster_rpc; use crate::infra::config::CONFIG; +use crate::infra::errors::{Error, ErrorCodes}; use crate::meta::sql::Sql as MetaSql; use crate::meta::StreamType; use crate::service::stream::get_stream_setting_fts_fields; @@ -70,7 +71,7 @@ impl Display for SqlMode { impl Sql { #[tracing::instrument(name = "service:search:sql:new", skip(req))] - pub async fn new(req: &cluster_rpc::SearchRequest) -> Result { + pub async fn new(req: &cluster_rpc::SearchRequest) -> Result { let req_query = req.query.as_ref().unwrap(); let req_time_range = (req_query.start_time, req_query.end_time); let org_id = req.org_id.clone(); @@ -80,12 +81,12 @@ impl Sql { // quick check SQL if origin_sql.is_empty() { - return Err(anyhow::anyhow!("Query SQL is empty")); + return Err(Error::ErrorCode(ErrorCodes::SearchSQLNotValid( + "".to_string(), + ))); } if !origin_sql.to_lowercase().contains(" from ") { - return Err(anyhow::anyhow!( - "Query SQL should likes [select * from table]" - )); + return Err(Error::ErrorCode(ErrorCodes::SearchSQLNotValid(origin_sql))); } // log::info!("[TRACE] origin_sql: {:?}", origin_sql); @@ -101,35 +102,35 @@ impl Sql { let re1 = Regex::new(r"(?i) (limit|offset|group|having|join|union) ").unwrap(); let re2 = Regex::new(r"(?i) (join|union) ").unwrap(); if sql_mode.eq(&SqlMode::Context) && re1.is_match(&origin_sql) { - return Err(anyhow::anyhow!( - "sql_mode=context, Query SQL is not supported [limit|offset|group by|having|join|union]", - )); + return Err(Error::ErrorCode(ErrorCodes::SearchSQLNotValid( + "sql_mode=context, Query SQL does not supported [limit|offset|group by|having|join|union]".to_string() + ))); } if sql_mode.eq(&SqlMode::Full) && re2.is_match(&origin_sql) { - return Err(anyhow::anyhow!( - "sql_mode=full, Query SQL is not supported [join|union]", - )); + return Err(Error::ErrorCode(ErrorCodes::SearchSQLNotValid( + "sql_mode=full, Query SQL does not supported [join|union]".to_string(), + ))); } // check time_range if req_time_range.0 > 0 && req_time_range.0 < Duration::seconds(1).num_microseconds().unwrap() { - return Err(anyhow::anyhow!( - "Query SQL time_range start_time should be microseconds", - )); + return Err(Error::ErrorCode(ErrorCodes::SearchSQLNotValid( + "Query SQL time_range start_time should be microseconds".to_string(), + ))); } if req_time_range.1 > 0 && req_time_range.1 < Duration::seconds(1).num_microseconds().unwrap() { - return Err(anyhow::anyhow!( - "Query SQL time_range start_time should be microseconds", - )); + return Err(Error::ErrorCode(ErrorCodes::SearchSQLNotValid( + "Query SQL time_range start_time should be microseconds".to_string(), + ))); } if req_time_range.0 > 0 && req_time_range.1 > 0 && req_time_range.1 < req_time_range.0 { - return Err(anyhow::anyhow!( - "Query SQL time_range start_time should be less than end_time", - )); + return Err(Error::ErrorCode(ErrorCodes::SearchSQLNotValid( + "Query SQL time_range start_time should be less than end_time".to_string(), + ))); } // check Agg SQL @@ -145,7 +146,9 @@ impl Sql { req_aggs.insert(agg.name.to_string(), agg.sql.to_string()); } if sql_mode.eq(&SqlMode::Full) && !req_aggs.is_empty() { - return Err(anyhow::anyhow!("sql_mode=full, Query not supported aggs")); + return Err(Error::ErrorCode(ErrorCodes::SearchSQLNotValid( + "sql_mode=full, Query not supported aggs".to_string(), + ))); } // check aggs @@ -154,14 +157,15 @@ impl Sql { } for sql in req_aggs.values() { if !re1.is_match(sql.as_str()) { - return Err(anyhow::anyhow!( - "Aggregation SQL only support 'from query' as context", - )); + return Err(Error::ErrorCode(ErrorCodes::SearchSQLNotValid( + "Aggregation SQL only support 'from query' as context".to_string(), + ))); } if re2.is_match(sql.as_str()) { - return Err(anyhow::anyhow!( - "Aggregation SQL is not supported 'select *' please specify the fields", - )); + return Err(Error::ErrorCode(ErrorCodes::SearchSQLNotValid( + "Aggregation SQL is not supported 'select *' please specify the fields" + .to_string(), + ))); } if re3.is_match(sql.as_str()) { let caps = re3.captures(sql.as_str()).unwrap(); @@ -173,16 +177,16 @@ impl Sql { let select_caps = match re4.captures(sql.as_str()) { Some(caps) => caps, None => { - return Err(anyhow::anyhow!( - "Aggregation SQL should start with 'select'", - )) + return Err(Error::ErrorCode(ErrorCodes::SearchSQLNotValid( + sql.to_owned(), + ))); } }; if !select_caps.get(1).unwrap().as_str().contains(group_by) { - return Err(anyhow::anyhow!( + return Err(Error::ErrorCode(ErrorCodes::ServerInternalError(format!( "Aggregation SQL used [group by] you should select the field [{}]", group_by - )); + )))); } } } @@ -190,7 +194,12 @@ impl Sql { // parse sql let meta = MetaSql::new(&origin_sql); if meta.is_err() { - return Err(anyhow::anyhow!(meta.err().unwrap())); + log::error!( + "parse sql error: {}, sql: {}", + meta.err().unwrap(), + origin_sql + ); + return Err(Error::ErrorCode(ErrorCodes::SearchSQLNotValid(origin_sql))); } let mut meta = meta.unwrap(); @@ -207,7 +216,9 @@ impl Sql { let re = Regex::new(&format!(r#"(?i) from[ '"]+{}[ '"]?"#, stream_name)).unwrap(); let caps = match re.captures(origin_sql.as_str()) { Some(caps) => caps, - None => return Err(anyhow::anyhow!("SQL should likes [select * from table]")), + None => { + return Err(Error::ErrorCode(ErrorCodes::SearchSQLNotValid(origin_sql))); + } }; origin_sql = origin_sql.replace(caps.get(0).unwrap().as_str(), " FROM tbl "); @@ -281,7 +292,9 @@ impl Sql { && !origin_sql.contains(" count(") && !origin_sql.contains(&CONFIG.common.time_stamp_col) { - return Err(anyhow::anyhow!("sql_mode=full, Query SQL must limit rows")); + return Err(Error::ErrorCode(ErrorCodes::SearchSQLNotValid( + "sql_mode=full, Query SQL must limit rows".to_string(), + ))); } } @@ -345,7 +358,7 @@ impl Sql { fulltext_search.push(format!("\"{}\" {} '%{}%'", field.name(), func, item.1)); } if fulltext_search.is_empty() { - return Err(anyhow::anyhow!("No full text search field found")); + return Err(Error::ErrorCode(ErrorCodes::FullTextSearchFieldNotFound)); } let fulltext_search = format!("({})", fulltext_search.join(" OR ")); origin_sql = origin_sql.replace(item.0.as_str(), &fulltext_search); @@ -452,7 +465,8 @@ impl Sql { } let sql_meta = MetaSql::new(sql.clone().as_str()); if sql_meta.is_err() { - return Err(sql_meta.err().unwrap()); + log::error!("parse sql error: {}, sql: {}", sql_meta.err().unwrap(), sql); + return Err(Error::ErrorCode(ErrorCodes::SearchSQLNotValid(sql))); } for cap in re_histogram.captures_iter(sql.clone().as_str()) { let attrs = cap diff --git a/src/service/search/storage.rs b/src/service/search/storage.rs index 7dc4542370..b7d2f0a608 100644 --- a/src/service/search/storage.rs +++ b/src/service/search/storage.rs @@ -23,6 +23,7 @@ use super::datafusion::storage::file_list::SessionType; use super::sql::Sql; use crate::infra::cache::file_data; use crate::infra::config::CONFIG; +use crate::infra::errors::{Error, ErrorCodes}; use crate::meta; use crate::service::{db, file_list}; @@ -77,12 +78,16 @@ pub async fn search( for task in tasks { match task.await { Ok(ret) => { - if let Err(e) = ret { - return Err(anyhow::anyhow!(e)); + if let Err(err) = ret { + return Err(Error::ErrorCode(ErrorCodes::ServerInternalError( + err.to_string(), + ))); } } - Err(e) => { - return Err(anyhow::anyhow!(e)); + Err(err) => { + return Err(Error::ErrorCode(ErrorCodes::ServerInternalError( + err.to_string(), + ))); } }; } @@ -93,9 +98,19 @@ pub async fn search( // fetch all schema versions, group files by version let schema_versions = - db::schema::get_versions(&sql.org_id, &sql.stream_name, Some(stream_type)).await?; + match db::schema::get_versions(&sql.org_id, &sql.stream_name, Some(stream_type)).await { + Ok(versions) => versions, + Err(err) => { + log::error!("get schema error: {}", err); + return Err(Error::ErrorCode(ErrorCodes::SearchStreamNotFound( + sql.stream_name.clone(), + ))); + } + }; if schema_versions.is_empty() { - return Err(anyhow::anyhow!("stream not found")); + return Err(Error::ErrorCode(ErrorCodes::SearchStreamNotFound( + sql.stream_name.clone(), + ))); } let schema_latest = schema_versions.last().unwrap(); let schema_latest_id = schema_versions.len() - 1; @@ -103,7 +118,15 @@ pub async fn search( HashMap::with_capacity(schema_versions.len()); let mut scan_size = 0; if !CONFIG.common.widening_schema_evoluation || schema_versions.len() == 1 { - scan_size = file_list::calculate_files_size(&files.to_vec()).await?; + scan_size = match file_list::calculate_files_size(&files.to_vec()).await { + Ok(size) => size, + Err(err) => { + log::error!("calculate files size error: {}", err); + return Err(Error::ErrorCode(ErrorCodes::ServerInternalError( + "calculate files size error".to_string(), + ))); + } + }; files_group.insert(schema_latest_id, files); } else { for file in files { @@ -193,12 +216,15 @@ pub async fn search( group.extend(v); } } - Err(e) => { - return Err(anyhow::anyhow!(e)); + Err(err) => { + log::error!("datafusion execute error: {}", err); + return Err(super::handle_datafusion_error(err)); } }, - Err(e) => { - return Err(anyhow::anyhow!(e)); + Err(err) => { + return Err(Error::ErrorCode(ErrorCodes::ServerInternalError( + err.to_string(), + ))); } }; } @@ -207,19 +233,25 @@ pub async fn search( } #[inline] -async fn get_file_list( - sql: &Sql, - stream_type: meta::StreamType, -) -> Result, anyhow::Error> { +async fn get_file_list(sql: &Sql, stream_type: meta::StreamType) -> Result, Error> { let (time_min, time_max) = sql.meta.time_range.unwrap(); - let results = file_list::get_file_list( + let results = match file_list::get_file_list( &sql.org_id, &sql.stream_name, Some(stream_type), time_min, time_max, ) - .await?; + .await + { + Ok(results) => results, + Err(err) => { + log::error!("get file list error: {}", err); + return Err(Error::ErrorCode(ErrorCodes::ServerInternalError( + "get file list error".to_string(), + ))); + } + }; let mut files = Vec::new(); for file in results { diff --git a/src/service/stream.rs b/src/service/stream.rs index 5b4042c7cb..017fae2302 100644 --- a/src/service/stream.rs +++ b/src/service/stream.rs @@ -50,7 +50,7 @@ pub async fn get_stream( } else { Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( StatusCode::NOT_FOUND.into(), - Some("stream not found".to_owned()), + "stream not found".to_string(), ))) } } @@ -175,7 +175,7 @@ pub async fn save_stream_settings( return Ok( HttpResponse::InternalServerError().json(MetaHttpResponse::error( http::StatusCode::INTERNAL_SERVER_ERROR.into(), - Some(format!("stream [{}] is being deleted", stream_name)), + format!("stream [{}] is being deleted", stream_name), )), ); } @@ -217,7 +217,7 @@ pub async fn delete_stream( if schema.is_empty() { return Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( StatusCode::NOT_FOUND.into(), - Some("stream not found".to_owned()), + "stream not found".to_string(), ))); } @@ -227,7 +227,7 @@ pub async fn delete_stream( return Ok( HttpResponse::InternalServerError().json(MetaHttpResponse::error( StatusCode::INTERNAL_SERVER_ERROR.into(), - Some(format!("failed to delete stream: {}", e)), + format!("failed to delete stream: {}", e), )), ); } @@ -237,7 +237,7 @@ pub async fn delete_stream( return Ok( HttpResponse::InternalServerError().json(MetaHttpResponse::error( StatusCode::INTERNAL_SERVER_ERROR.into(), - Some(format!("failed to delete stream: {}", e)), + format!("failed to delete stream: {}", e), )), ); } @@ -254,14 +254,14 @@ pub async fn delete_stream( return Ok( HttpResponse::InternalServerError().json(MetaHttpResponse::error( StatusCode::INTERNAL_SERVER_ERROR.into(), - Some(format!("failed to delete stream: {}", e)), + format!("failed to delete stream: {}", e), )), ); }; Ok(HttpResponse::Ok().json(MetaHttpResponse::message( StatusCode::OK.into(), - "stream deleted".to_owned(), + "stream deleted".to_string(), ))) } diff --git a/src/service/traces/mod.rs b/src/service/traces/mod.rs index d033a30c57..359899d437 100644 --- a/src/service/traces/mod.rs +++ b/src/service/traces/mod.rs @@ -63,7 +63,7 @@ pub async fn handle_trace_request( return Ok( HttpResponse::InternalServerError().json(meta::http::HttpResponse::error( http::StatusCode::INTERNAL_SERVER_ERROR.into(), - Some("not an ingester".to_string()), + "not an ingester".to_string(), )), ); } diff --git a/src/service/traces/otlp_http.rs b/src/service/traces/otlp_http.rs index 88d0e07b11..14f36c2fb4 100644 --- a/src/service/traces/otlp_http.rs +++ b/src/service/traces/otlp_http.rs @@ -68,7 +68,7 @@ pub async fn traces_json( return Ok( HttpResponse::InternalServerError().json(meta::http::HttpResponse::error( http::StatusCode::INTERNAL_SERVER_ERROR.into(), - Some("not an ingester".to_string()), + "not an ingester".to_string(), )), ); } @@ -253,7 +253,7 @@ pub async fn traces_json( return Ok( HttpResponse::BadRequest().json(meta::http::HttpResponse::error( http::StatusCode::BAD_REQUEST.into(), - Some("Bad Request".to_string()), + "Bad Request".to_string(), )), ); } diff --git a/src/service/triggers.rs b/src/service/triggers.rs index 8204eaa79c..e8c0f48816 100644 --- a/src/service/triggers.rs +++ b/src/service/triggers.rs @@ -42,9 +42,9 @@ pub async fn delete_trigger(alert_name: String) -> Result { http::StatusCode::OK.into(), "Trigger deleted ".to_string(), ))), - Err(err) => Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( + Err(e) => Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( StatusCode::NOT_FOUND.into(), - Some(err.to_string()), + e.to_string(), ))), } } @@ -61,7 +61,7 @@ pub async fn get_alert( Ok(alert) => Ok(HttpResponse::Ok().json(alert)), Err(_) => Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( StatusCode::NOT_FOUND.into(), - Some("alert not found".to_string()), + "alert not found".to_string(), ))), } } diff --git a/src/service/users.rs b/src/service/users.rs index 462ae4a71e..ef92e57eed 100644 --- a/src/service/users.rs +++ b/src/service/users.rs @@ -178,7 +178,7 @@ pub async fn update_user( } Err(_) => Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( StatusCode::NOT_FOUND.into(), - Some("User not found".to_string()), + "User not found".to_string(), ))), } } else { @@ -193,13 +193,13 @@ pub async fn update_user( } None => Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( StatusCode::NOT_FOUND.into(), - Some("User not found".to_string()), + "User not found".to_string(), ))), } } else { Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( StatusCode::NOT_FOUND.into(), - Some("User not found".to_string()), + "User not found".to_string(), ))) } } @@ -251,13 +251,13 @@ pub async fn add_user_to_org( } else { Ok(HttpResponse::Unauthorized().json(MetaHttpResponse::error( StatusCode::UNAUTHORIZED.into(), - Some("Not Allowed".to_string()), + "Not Allowed".to_string(), ))) } } else { Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( StatusCode::NOT_FOUND.into(), - Some("User not found".to_string()), + "User not found".to_string(), ))) } } @@ -326,13 +326,13 @@ pub async fn remove_user_from_org(org_id: &str, email_id: &str) -> Result Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( StatusCode::NOT_FOUND.into(), - Some("User for the organization not found".to_owned()), + "User for the organization not found".to_string(), ))), } } @@ -344,9 +344,9 @@ pub async fn delete_user(email_id: &str) -> Result { http::StatusCode::OK.into(), "User deleted".to_string(), ))), - Err(err) => Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( + Err(e) => Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( StatusCode::NOT_FOUND.into(), - Some(err.to_string()), + e.to_string(), ))), } }