refactor: check error type before logging (#3697)

* refactor: check error type before logging

* chore: update log level for broken pipe

* refactor: leave a debugging output for non critial error

src/frontend/src/instance.rs

@@ -31,7 +31,7 @@ use catalog::CatalogManagerRef;
 use client::OutputData;
 use common_base::Plugins;
 use common_config::KvBackendConfig;
-use common_error::ext::BoxedError;
+use common_error::ext::{BoxedError, ErrorExt};
 use common_grpc::channel_manager::{ChannelConfig, ChannelManager};
 use common_meta::key::TableMetadataManagerRef;
 use common_meta::kv_backend::KvBackendRef;
@@ -40,8 +40,7 @@ use common_procedure::local::{LocalManager, ManagerConfig};
 use common_procedure::options::ProcedureConfig;
 use common_procedure::ProcedureManagerRef;
 use common_query::Output;
-use common_telemetry::logging::info;
-use common_telemetry::{error, tracing};
+use common_telemetry::{debug, error, info, tracing};
 use log_store::raft_engine::RaftEngineBackend;
 use meta_client::client::{MetaClient, MetaClientBuilder};
 use meta_client::MetaClientOptions;
@@ -324,7 +323,11 @@ impl SqlQueryHandler for Instance {
                         }
                         Err(e) => {
                             let redacted = sql::util::redact_sql_secrets(query.as_ref());
-                            error!(e; "Failed to execute query: {redacted}");
+                            if e.status_code().should_log_error() {
+                                error!(e; "Failed to execute query: {redacted}");
+                            } else {
+                                debug!("Failed to execute query: {redacted}, {e}");
+                            }
 
                             results.push(Err(e));
                             break;

src/frontend/src/script.rs

@@ -16,6 +16,7 @@ use std::collections::HashMap;
 use std::sync::Arc;
 
 use catalog::CatalogManagerRef;
+use common_error::ext::ErrorExt;
 use common_query::Output;
 use query::QueryEngineRef;
 use servers::query_handler::grpc::GrpcQueryHandler;
@@ -215,7 +216,10 @@ mod python {
             self.create_scripts_table_if_need(query_ctx.current_catalog())
                 .await
                 .map_err(|e| {
-                    error!(e; "Failed to create scripts table");
+                    if e.status_code().should_log_error() {
+                        error!(e; "Failed to create scripts table");
+                    }
+
                     servers::error::InternalSnafu {
                         err_msg: e.to_string(),
                     }
@@ -232,7 +236,10 @@ mod python {
                 )
                 .await
                 .map_err(|e| {
-                    error!(e; "Failed to insert script");
+                    if e.status_code().should_log_error() {
+                        error!(e; "Failed to insert script");
+                    }
+
                     BoxedError::new(e)
                 })
                 .context(servers::error::InsertScriptSnafu { name })?;
@@ -265,7 +272,10 @@ mod python {
                 )
                 .await
                 .map_err(|e| {
-                    error!(e; "Failed to execute script");
+                    if e.status_code().should_log_error() {
+                        error!(e; "Failed to execute script");
+                    }
+
                     BoxedError::new(e)
                 })
                 .context(servers::error::ExecuteScriptSnafu { name })

src/servers/src/mysql/server.rs

@@ -19,7 +19,7 @@ use std::sync::Arc;
 use async_trait::async_trait;
 use auth::UserProviderRef;
 use common_runtime::Runtime;
-use common_telemetry::{debug, error, warn};
+use common_telemetry::{debug, warn};
 use futures::StreamExt;
 use opensrv_mysql::{
     plain_run_with_options, secure_run_with_options, AsyncMysqlIntermediary, IntermediaryOptions,
@@ -138,7 +138,7 @@ impl MysqlServer {
                     Err(error) => warn!("Broken pipe: {}", error), // IoError doesn't impl ErrorExt.
                     Ok(io_stream) => {
                         if let Err(e) = io_stream.set_nodelay(true) {
-                            error!(e; "Failed to set TCP nodelay");
+                            warn!(e; "Failed to set TCP nodelay");
                         }
                         if let Err(error) =
                             Self::handle(io_stream, io_runtime, spawn_ref, spawn_config).await

src/servers/src/opentsdb.rs

@@ -22,9 +22,10 @@ use std::sync::Arc;
 
 use api::v1::RowInsertRequests;
 use async_trait::async_trait;
+use common_error::ext::ErrorExt;
 use common_query::prelude::{GREPTIME_TIMESTAMP, GREPTIME_VALUE};
 use common_runtime::Runtime;
-use common_telemetry::logging::error;
+use common_telemetry::logging::{debug, error, warn};
 use futures::StreamExt;
 use tokio::sync::broadcast;
 
@@ -86,18 +87,20 @@ impl OpentsdbServer {
                 match stream {
                     Ok(stream) => {
                         if let Err(e) = stream.set_nodelay(true) {
-                            error!(e; "Failed to set TCP nodelay");
+                            warn!(e; "Failed to set TCP nodelay");
                         }
                         let connection = Connection::new(stream);
                         let mut handler = Handler::new(query_handler, connection, shutdown);
 
                         let _handle = io_runtime.spawn(async move {
                             if let Err(e) = handler.run().await {
-                                error!(e; "Unexpected error when handling OpenTSDB connection");
+                                if e.status_code().should_log_error() {
+                                    error!(e; "Unexpected error when handling OpenTSDB connection");
+                                }
                             }
                         });
                     }
-                    Err(error) => error!("Broken pipe: {}", error), // IoError doesn't impl ErrorExt.
+                    Err(error) => debug!("Broken pipe: {}", error), // IoError doesn't impl ErrorExt.
                 };
             }
         })

src/servers/src/postgres/server.rs

@@ -19,7 +19,6 @@ use std::sync::Arc;
 use ::auth::UserProviderRef;
 use async_trait::async_trait;
 use common_runtime::Runtime;
-use common_telemetry::logging::error;
 use common_telemetry::{debug, warn};
 use futures::StreamExt;
 use pgwire::tokio::process_socket;
@@ -79,7 +78,7 @@ impl PostgresServer {
 
             async move {
                 match tcp_stream {
-                    Err(error) => error!("Broken pipe: {}", error), // IoError doesn't impl ErrorExt.
+                    Err(error) => debug!("Broken pipe: {}", error), // IoError doesn't impl ErrorExt.
                     Ok(io_stream) => {
                         let addr = match io_stream.peer_addr() {
                             Ok(addr) => {