Influxdata
/
influxdb
mirror of https://github.com/influxdata/influxdb.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

refactor: align HTTP error handling w/ cloud 2 

Streamline HTTP error by using the same representation as cloud 2. Also
ensure that only internal server errors result in an error-level log
entry.

Fixes #3075.
pull/24376/head
Marco Neumann 2021-11-11 14:47:36 +01:00
parent 6f268f8260
commit 3667cb36fa
8 changed files with 276 additions and 172 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

163
influxdb_iox/src/influxdb_ioxd/http/error.rs Normal file
View File

@ -0,0 +1,163 @@
use hyper::{Body, Response, StatusCode};
/// Constants used in API error codes.
///
/// See <https://docs.influxdata.com/influxdb/v2.1/api/#operation/PostWrite>.
#[derive(Debug, PartialEq, Eq, Clone, Copy)]
#[allow(dead_code)]
pub enum HttpApiErrorCode {
InternalError,
NotFound,
Conflict,
Invalid,
UnprocessableEntity,
EmptyValue,
Unavailable,
Forbidden,
TooManyRequests,
Unauthorized,
MethodNotAllowed,
RequestTooLarge,
UnsupportedMediaType,
}
impl HttpApiErrorCode {
/// Get machine-readable text representation.
fn as_text(&self) -> &'static str {
match self {
Self::InternalError => "internal error",
Self::NotFound => "not found",
Self::Conflict => "conflict",
Self::Invalid => "invalid",
Self::UnprocessableEntity => "unprocessable entity",
Self::EmptyValue => "empty value",
Self::Unavailable => "unavailable",
Self::Forbidden => "forbidden",
Self::TooManyRequests => "too many requests",
Self::Unauthorized => "unauthorized",
Self::MethodNotAllowed => "method not allowed",
Self::RequestTooLarge => "request too large",
Self::UnsupportedMediaType => "unsupported media type",
}
}
/// Get tonic HTTP status code.
fn status_code(&self) -> StatusCode {
match self {
Self::InternalError => StatusCode::INTERNAL_SERVER_ERROR,
Self::NotFound => StatusCode::NOT_FOUND,
Self::Conflict => StatusCode::CONFLICT,
Self::Invalid => StatusCode::BAD_REQUEST,
Self::UnprocessableEntity => StatusCode::UNPROCESSABLE_ENTITY,
Self::EmptyValue => StatusCode::NO_CONTENT,
Self::Unavailable => StatusCode::SERVICE_UNAVAILABLE,
Self::Forbidden => StatusCode::FORBIDDEN,
Self::TooManyRequests => StatusCode::TOO_MANY_REQUESTS,
Self::Unauthorized => StatusCode::UNAUTHORIZED,
Self::MethodNotAllowed => StatusCode::METHOD_NOT_ALLOWED,
Self::RequestTooLarge => StatusCode::PAYLOAD_TOO_LARGE,
Self::UnsupportedMediaType => StatusCode::UNSUPPORTED_MEDIA_TYPE,
}
}
/// Check if the code is an internal server error.
fn is_internal(&self) -> bool {
matches!(self, Self::InternalError)
}
}
/// Error that is compatible with the Influxdata Cloud 2 HTTP API.
///
/// See <https://docs.influxdata.com/influxdb/v2.1/api/#operation/PostWrite>.
#[derive(Debug)]
pub struct HttpApiError {
/// Machine-readable error code.
code: HttpApiErrorCode,
/// Human-readable message.
msg: String,
}
impl HttpApiError {
/// Create new error from code and message.
pub fn new(code: HttpApiErrorCode, msg: impl Into<String>) -> Self {
Self {
code,
msg: msg.into(),
}
}
/// Generate response body for this error.
fn body(&self) -> Body {
let json = serde_json::json!({
"code": self.code.as_text().to_string(),
"message": self.msg.clone(),
})
.to_string();
Body::from(json)
}
/// Generate response for this error.
pub fn response(&self) -> Response<Body> {
Response::builder()
.status(self.code.status_code())
.body(self.body())
.unwrap()
}
/// Check if the error is an internal server error.
pub fn is_internal(&self) -> bool {
self.code.is_internal()
}
}
impl std::fmt::Display for HttpApiError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{}: {}", self.code.as_text(), self.msg)
}
}
impl std::error::Error for HttpApiError {}
/// Mixin-trait to simplify creation of [`HttpApiError`].
pub trait HttpApiErrorExt {
/// No data can be returned, but the server was asked to do so.
fn empty_value(&self) -> HttpApiError;
/// Internal server error. This is a bug / misconfiguration.
fn internal_error(&self) -> HttpApiError;
/// Invalid/bad request.
fn invalid(&self) -> HttpApiError;
/// Resource was not found.
fn not_found(&self) -> HttpApiError;
}
impl<E> HttpApiErrorExt for E
where
E: std::error::Error,
{
fn empty_value(&self) -> HttpApiError {
HttpApiError::new(HttpApiErrorCode::EmptyValue, self.to_string())
}
fn internal_error(&self) -> HttpApiError {
HttpApiError::new(HttpApiErrorCode::InternalError, self.to_string())
}
fn invalid(&self) -> HttpApiError {
HttpApiError::new(HttpApiErrorCode::Invalid, self.to_string())
}
fn not_found(&self) -> HttpApiError {
HttpApiError::new(HttpApiErrorCode::NotFound, self.to_string())
}
}
/// An error that can be transformed into a [`HttpApiError`].
pub trait HttpApiErrorSource: std::error::Error {
/// Create [`HttpApiError`].
fn to_http_api_error(&self) -> HttpApiError;
}

48
influxdb_iox/src/influxdb_ioxd/http/mod.rs
View File

@ -7,12 +7,15 @@ use hyper::{
};
use observability_deps::tracing::{debug, error};
use serde::Deserialize;
use snafu::{ResultExt, Snafu};
use snafu::Snafu;
use tokio_util::sync::CancellationToken;
use tower::Layer;
use trace_http::{ctx::TraceHeaderParser, tower::TraceLayer};
use crate::influxdb_ioxd::server_type::{RouteError, ServerType};
use crate::influxdb_ioxd::{
http::error::{HttpApiError, HttpApiErrorExt, HttpApiErrorSource},
server_type::ServerType,
};
#[cfg(feature = "heappy")]
mod heappy;
@ -20,6 +23,7 @@ mod heappy;
#[cfg(feature = "pprof")]
mod pprof;
pub mod error;
pub mod metrics;
pub mod utils;
pub mod write;
@ -62,23 +66,23 @@ pub enum ApplicationError {
#[snafu(display("pprof support is not compiled"))]
PProfIsNotCompiled,
#[snafu(display("Route error from run mode: {}", source))]
RunModeRouteError { source: Box<dyn RouteError> },
#[snafu(display("Route error from run mode: {}", e))]
RunModeRouteError { e: Box<dyn HttpApiErrorSource> },
}
impl RouteError for ApplicationError {
fn response(&self) -> Response<Body> {
impl HttpApiErrorSource for ApplicationError {
fn to_http_api_error(&self) -> HttpApiError {
match self {
Self::InvalidQueryString { .. } => self.bad_request(),
Self::PProf { .. } => self.internal_error(),
Self::Prost { .. } => self.internal_error(),
Self::ProstIO { .. } => self.internal_error(),
Self::EmptyFlamegraph => self.no_content(),
Self::HeappyIsNotCompiled => self.internal_error(),
Self::PProfIsNotCompiled => self.internal_error(),
e @ Self::InvalidQueryString { .. } => e.invalid(),
e @ Self::PProf { .. } => e.internal_error(),
e @ Self::Prost { .. } => e.internal_error(),
e @ Self::ProstIO { .. } => e.internal_error(),
e @ Self::EmptyFlamegraph => e.empty_value(),
e @ Self::HeappyIsNotCompiled => e.internal_error(),
e @ Self::PProfIsNotCompiled => e.internal_error(),
#[cfg(feature = "heappy")]
Self::HeappyError { .. } => self.internal_error(),
Self::RunModeRouteError { source } => source.response(),
e @ Self::HeappyError { .. } => e.internal_error(),
Self::RunModeRouteError { e } => e.to_http_api_error(),
}
}
}
@ -135,8 +139,7 @@ where
_ => server_type
.route_http_request(req)
.await
.map_err(|e| Box::new(e) as _)
.context(RunModeRouteError),
.map_err(|e| ApplicationError::RunModeRouteError { e: Box::new(e) }),
};
// TODO: Move logging to TraceLayer
@ -146,7 +149,12 @@ where
Ok(response)
}
Err(error) => {
error!(%error, %method, %uri, ?content_length, "Error while handling request");
let error: HttpApiError = error.to_http_api_error();
if error.is_internal() {
error!(%error, %method, %uri, ?content_length, "Error while handling request");
} else {
debug!(%error, %method, %uri, ?content_length, "Error while handling request");
}
Ok(error.response())
}
}
@ -236,6 +244,8 @@ impl PProfAllocsArgs {
#[cfg(feature = "pprof")]
async fn pprof_profile(req: Request<Body>) -> Result<Response<Body>, ApplicationError> {
use ::pprof::protos::Message;
use snafu::ResultExt;
let query_string = req.uri().query().unwrap_or_default();
let query: PProfArgs =
serde_urlencoded::from_str(query_string).context(InvalidQueryString { query_string })?;
@ -282,6 +292,8 @@ async fn pprof_profile(_req: Request<Body>) -> Result<Response<Body>, Applicatio
// If heappy support is enabled, call it
#[cfg(feature = "heappy")]
async fn pprof_heappy_profile(req: Request<Body>) -> Result<Response<Body>, ApplicationError> {
use snafu::ResultExt;
let query_string = req.uri().query().unwrap_or_default();
let query: PProfAllocsArgs =
serde_urlencoded::from_str(query_string).context(InvalidQueryString { query_string })?;

18
influxdb_iox/src/influxdb_ioxd/http/utils.rs
View File

@ -1,10 +1,10 @@
use bytes::{Bytes, BytesMut};
use futures::StreamExt;
use http::header::CONTENT_ENCODING;
use hyper::{Body, Response};
use hyper::Body;
use snafu::{ResultExt, Snafu};
use crate::influxdb_ioxd::server_type::RouteError;
use super::error::{HttpApiError, HttpApiErrorExt, HttpApiErrorSource};
#[allow(clippy::large_enum_variant)]
#[derive(Debug, Snafu)]
@ -28,14 +28,14 @@ pub enum ParseBodyError {
ClientHangup { source: hyper::Error },
}
impl RouteError for ParseBodyError {
fn response(&self) -> Response<Body> {
impl HttpApiErrorSource for ParseBodyError {
fn to_http_api_error(&self) -> HttpApiError {
match self {
Self::RequestSizeExceeded { .. } => self.bad_request(),
Self::InvalidContentEncoding { .. } => self.bad_request(),
Self::ReadingHeaderAsUtf8 { .. } => self.bad_request(),
Self::ReadingBodyAsGzip { .. } => self.bad_request(),
Self::ClientHangup { .. } => self.bad_request(),
e @ Self::RequestSizeExceeded { .. } => e.invalid(),
e @ Self::InvalidContentEncoding { .. } => e.invalid(),
e @ Self::ReadingHeaderAsUtf8 { .. } => e.invalid(),
e @ Self::ReadingBodyAsGzip { .. } => e.invalid(),
e @ Self::ClientHangup { .. } => e.invalid(),
}
}
}

40
influxdb_iox/src/influxdb_ioxd/http/write.rs
View File

@ -12,12 +12,12 @@ use observability_deps::tracing::debug;
use serde::Deserialize;
use snafu::{OptionExt, ResultExt, Snafu};
use crate::influxdb_ioxd::{
http::utils::parse_body,
server_type::{ApiErrorCode, RouteError, ServerType},
};
use crate::influxdb_ioxd::{http::utils::parse_body, server_type::ServerType};
use super::metrics::LineProtocolMetrics;
use super::{
error::{HttpApiError, HttpApiErrorExt, HttpApiErrorSource},
metrics::LineProtocolMetrics,
};
#[allow(clippy::large_enum_variant)]
#[derive(Debug, Snafu)]
@ -63,27 +63,17 @@ pub enum HttpWriteError {
},
}
impl RouteError for HttpWriteError {
fn response(&self) -> Response<Body> {
impl HttpApiErrorSource for HttpWriteError {
fn to_http_api_error(&self) -> HttpApiError {
match self {
Self::BucketMappingError { .. } => self.internal_error(),
Self::WritingPoints { .. } => self.internal_error(),
Self::ExpectedQueryString { .. } => self.bad_request(),
Self::InvalidQueryString { .. } => self.bad_request(),
Self::ReadingBodyAsUtf8 { .. } => self.bad_request(),
Self::ParsingLineProtocol { .. } => self.bad_request(),
Self::DatabaseNotFound { .. } => self.not_found(),
Self::ParseBody { source } => source.response(),
}
}
/// Map the error type into an API error code.
fn api_error_code(&self) -> u32 {
match self {
Self::DatabaseNotFound { .. } => ApiErrorCode::DB_NOT_FOUND.into(),
Self::ParseBody { source } => source.api_error_code(),
// A "catch all" error code
_ => ApiErrorCode::UNKNOWN.into(),
e @ Self::BucketMappingError { .. } => e.internal_error(),
e @ Self::WritingPoints { .. } => e.internal_error(),
e @ Self::ExpectedQueryString { .. } => e.invalid(),
e @ Self::InvalidQueryString { .. } => e.invalid(),
e @ Self::ReadingBodyAsUtf8 { .. } => e.invalid(),
e @ Self::ParsingLineProtocol { .. } => e.invalid(),
e @ Self::DatabaseNotFound { .. } => e.not_found(),
Self::ParseBody { source } => source.to_http_api_error(),
}
}
}

83
influxdb_iox/src/influxdb_ioxd/server_type/database/http.rs
View File

@ -30,12 +30,12 @@ use snafu::{OptionExt, ResultExt, Snafu};
use crate::influxdb_ioxd::{
http::{
error::{HttpApiError, HttpApiErrorExt, HttpApiErrorSource},
metrics::LineProtocolMetrics,
utils::parse_body,
write::{HttpDrivenWrite, InnerWriteError, RequestOrResponse, WriteInfo},
},
planner::Planner,
server_type::{ApiErrorCode, RouteError},
};
use dml::DmlWrite;
use std::{
@ -152,42 +152,30 @@ pub enum ApplicationError {
type Result<T, E = ApplicationError> = std::result::Result<T, E>;
impl RouteError for ApplicationError {
fn response(&self) -> Response<Body> {
impl HttpApiErrorSource for ApplicationError {
fn to_http_api_error(&self) -> HttpApiError {
match self {
Self::BucketMappingError { .. } => self.internal_error(),
Self::Query { .. } => self.internal_error(),
Self::ExpectedQueryString { .. } => self.bad_request(),
Self::InvalidQueryString { .. } => self.bad_request(),
Self::ReadingBodyAsUtf8 { .. } => self.bad_request(),
Self::ParsingDelete { .. } => self.bad_request(),
Self::BuildingDeletePredicate { .. } => self.bad_request(),
Self::ExecutingDelete { .. } => self.bad_request(),
Self::RouteNotFound { .. } => self.not_found(),
Self::DatabaseNameError { .. } => self.bad_request(),
Self::DatabaseNotFound { .. } => self.not_found(),
Self::CreatingResponse { .. } => self.internal_error(),
Self::FormattingResult { .. } => self.internal_error(),
Self::ParsingFormat { .. } => self.bad_request(),
Self::Planning { .. } => self.bad_request(),
Self::ServerIdNotSet => self.bad_request(),
Self::ServerNotInitialized => self.bad_request(),
Self::DatabaseNotInitialized { .. } => self.bad_request(),
Self::InternalServerError => self.internal_error(),
Self::ParseBody { source } => source.response(),
Self::WriteError { source } => source.response(),
}
}
/// Map the error type into an API error code.
fn api_error_code(&self) -> u32 {
match self {
Self::DatabaseNameError { .. } => ApiErrorCode::DB_INVALID_NAME.into(),
Self::DatabaseNotFound { .. } => ApiErrorCode::DB_NOT_FOUND.into(),
Self::ParseBody { source } => source.api_error_code(),
Self::WriteError { source } => source.api_error_code(),
// A "catch all" error code
_ => ApiErrorCode::UNKNOWN.into(),
e @ Self::BucketMappingError { .. } => e.internal_error(),
e @ Self::Query { .. } => e.internal_error(),
e @ Self::ExpectedQueryString { .. } => e.invalid(),
e @ Self::InvalidQueryString { .. } => e.invalid(),
e @ Self::ReadingBodyAsUtf8 { .. } => e.invalid(),
e @ Self::ParsingDelete { .. } => e.invalid(),
e @ Self::BuildingDeletePredicate { .. } => e.invalid(),
e @ Self::ExecutingDelete { .. } => e.invalid(),
e @ Self::RouteNotFound { .. } => e.not_found(),
e @ Self::DatabaseNameError { .. } => e.invalid(),
e @ Self::DatabaseNotFound { .. } => e.not_found(),
e @ Self::CreatingResponse { .. } => e.internal_error(),
e @ Self::FormattingResult { .. } => e.internal_error(),
e @ Self::ParsingFormat { .. } => e.invalid(),
e @ Self::Planning { .. } => e.invalid(),
e @ Self::ServerIdNotSet => e.invalid(),
e @ Self::ServerNotInitialized => e.invalid(),
e @ Self::DatabaseNotInitialized { .. } => e.invalid(),
e @ Self::InternalServerError => e.internal_error(),
Self::ParseBody { source } => source.to_http_api_error(),
Self::WriteError { source } => source.to_http_api_error(),
}
}
}
@ -769,6 +757,29 @@ mod tests {
check_response("query", response, StatusCode::OK, Some(res)).await;
}
#[tokio::test]
async fn test_query_invalid_name() {
let (client, test_server) = setup_test_data().await;
// send query data
let response = client
.get(&format!(
"{}/api/v3/query?d=&q={}",
test_server.url(),
"select%20*%20from%20h2o_temperature%20order%20by%20surface_degrees"
))
.send()
.await;
check_response(
"query",
response,
StatusCode::BAD_REQUEST,
Some(r#"{"code":"invalid","message":"Invalid database name: Database name length must be between 1 and 64 characters"}"#),
)
.await;
}
/// Run the specified SQL query and return formatted results as a string
async fn run_query(db: Arc<Db>, query: &str) -> Vec<RecordBatch> {
let ctx = db.new_query_context(None);

76
influxdb_iox/src/influxdb_ioxd/server_type/mod.rs
View File

@ -1,87 +1,17 @@
use std::sync::Arc;
use async_trait::async_trait;
use hyper::{Body, Request, Response, StatusCode};
use hyper::{Body, Request, Response};
use metric::Registry;
use snafu::Snafu;
use trace::TraceCollector;
use super::rpc::RpcBuilderInput;
use crate::influxdb_ioxd::{http::error::HttpApiErrorSource, rpc::RpcBuilderInput};
pub mod common_state;
pub mod database;
pub mod router;
/// Constants used in API error codes.
///
/// Expressing this as a enum prevents reuse of discriminants, and as they're
/// effectively consts this uses UPPER_SNAKE_CASE.
#[allow(non_camel_case_types, clippy::upper_case_acronyms)]
#[derive(Debug, PartialEq)]
pub enum ApiErrorCode {
/// An unknown/unhandled error
UNKNOWN = 100,
/// The database name in the request is invalid.
DB_INVALID_NAME = 101,
/// The database referenced already exists.
DB_ALREADY_EXISTS = 102,
/// The database referenced does not exist.
DB_NOT_FOUND = 103,
}
impl From<ApiErrorCode> for u32 {
fn from(v: ApiErrorCode) -> Self {
v as Self
}
}
pub trait RouteError: std::error::Error + snafu::AsErrorSource {
fn response(&self) -> Response<Body>;
fn bad_request(&self) -> Response<Body> {
Response::builder()
.status(StatusCode::BAD_REQUEST)
.body(self.body())
.unwrap()
}
fn internal_error(&self) -> Response<Body> {
Response::builder()
.status(StatusCode::INTERNAL_SERVER_ERROR)
.body(self.body())
.unwrap()
}
fn not_found(&self) -> Response<Body> {
Response::builder()
.status(StatusCode::NOT_FOUND)
.body(Body::empty())
.unwrap()
}
fn no_content(&self) -> Response<Body> {
Response::builder()
.status(StatusCode::NO_CONTENT)
.body(self.body())
.unwrap()
}
fn body(&self) -> Body {
let json =
serde_json::json!({"error": self.to_string(), "error_code": self.api_error_code()})
.to_string();
Body::from(json)
}
/// Map the error type into an API error code.
fn api_error_code(&self) -> u32 {
ApiErrorCode::UNKNOWN.into()
}
}
#[derive(Debug, Snafu)]
pub enum RpcError {
#[snafu(display("gRPC transport error: {}{}", source, details))]
@ -107,7 +37,7 @@ impl From<tonic::transport::Error> for RpcError {
#[async_trait]
pub trait ServerType: std::fmt::Debug + Send + Sync + 'static {
type RouteError: RouteError;
type RouteError: HttpApiErrorSource;
fn metric_registry(&self) -> Arc<Registry>;

18
influxdb_iox/src/influxdb_ioxd/server_type/router/http.rs
View File

@ -6,12 +6,10 @@ use dml::DmlWrite;
use hyper::{Body, Method, Request, Response};
use snafu::{ResultExt, Snafu};
use crate::influxdb_ioxd::{
http::{
metrics::LineProtocolMetrics,
write::{HttpDrivenWrite, InnerWriteError, RequestOrResponse},
},
server_type::RouteError,
use crate::influxdb_ioxd::http::{
error::{HttpApiError, HttpApiErrorExt, HttpApiErrorSource},
metrics::LineProtocolMetrics,
write::{HttpDrivenWrite, InnerWriteError, RequestOrResponse},
};
use super::RouterServerType;
@ -27,11 +25,11 @@ pub enum ApplicationError {
},
}
impl RouteError for ApplicationError {
fn response(&self) -> http::Response<hyper::Body> {
impl HttpApiErrorSource for ApplicationError {
fn to_http_api_error(&self) -> HttpApiError {
match self {
Self::RouteNotFound { .. } => self.not_found(),
Self::WriteError { source } => source.response(),
e @ Self::RouteNotFound { .. } => e.not_found(),
Self::WriteError { source } => source.to_http_api_error(),
}
}
}

2
influxdb_iox/tests/end_to_end_cases/http.rs
View File

@ -11,6 +11,6 @@ async fn test_http_error_messages() {
.await
.expect_err("Should have errored");
let expected_error = "HTTP request returned an error: 400 Bad Request, `{\"error\":\"Error parsing line protocol: error parsing line 1: A generic parsing error occurred: TakeWhile1\",\"error_code\":100}`";
let expected_error = r#"HTTP request returned an error: 400 Bad Request, `{"code":"invalid","message":"Error parsing line protocol: error parsing line 1: A generic parsing error occurred: TakeWhile1"}`"#;
assert_eq!(result.to_string(), expected_error);
}