fix: Prevent panic in GetMetrics gRPC handler on invalid input (#1291)

Co-authored-by: Copilot <198982749+Copilot@users.noreply.github.com>
Co-authored-by: houseme <4829346+houseme@users.noreply.github.com>

rustfs/Cargo.toml

@@ -44,6 +44,7 @@ rustfs-appauth = { workspace = true }
 rustfs-audit = { workspace = true }
 rustfs-common = { workspace = true }
 rustfs-config = { workspace = true, features = ["constants", "notify"] }
+rustfs-credentials = { workspace = true }
 rustfs-ecstore = { workspace = true }
 rustfs-filemeta.workspace = true
 rustfs-iam = { workspace = true }

rustfs/src/admin/auth.rs

@@ -14,9 +14,9 @@
 
 use crate::auth::get_condition_values;
 use http::HeaderMap;
+use rustfs_credentials::Credentials;
 use rustfs_iam::store::object::ObjectStore;
 use rustfs_iam::sys::IamSys;
-use rustfs_policy::auth;
 use rustfs_policy::policy::Args;
 use rustfs_policy::policy::action::Action;
 use s3s::S3Result;
@@ -26,7 +26,7 @@ use std::sync::Arc;
 
 pub async fn validate_admin_request(
     headers: &HeaderMap,
-    cred: &auth::Credentials,
+    cred: &Credentials,
     is_owner: bool,
     deny_only: bool,
     actions: Vec<Action>,
@@ -49,7 +49,7 @@ pub async fn validate_admin_request(
 async fn check_admin_request_auth(
     iam_store: Arc<IamSys<ObjectStore>>,
     headers: &HeaderMap,
-    cred: &auth::Credentials,
+    cred: &Credentials,
     is_owner: bool,
     deny_only: bool,
     action: Action,

rustfs/src/admin/console.rs

@@ -23,7 +23,6 @@ use axum::{
     response::{IntoResponse, Response},
     routing::get,
 };
-use axum_extra::extract::Host;
 use axum_server::tls_rustls::RustlsConfig;
 use http::{HeaderMap, HeaderName, HeaderValue, Method, StatusCode, Uri};
 use mime_guess::from_path;
@@ -264,21 +263,27 @@ async fn version_handler() -> impl IntoResponse {
 ///
 /// # Arguments:
 /// - `uri`: The request URI.
-/// - `Host(host)`: The host extracted from the request.
 /// - `headers`: The request headers.
 ///
 /// # Returns:
 /// - 200 OK with JSON body containing the console configuration if initialized.
 /// - 500 Internal Server Error if configuration is not initialized.
-#[instrument(fields(host))]
-async fn config_handler(uri: Uri, Host(host): Host, headers: HeaderMap) -> impl IntoResponse {
+#[instrument(fields(uri))]
+async fn config_handler(uri: Uri, headers: HeaderMap) -> impl IntoResponse {
     // Get the scheme from the headers or use the URI scheme
     let scheme = headers
         .get(HeaderName::from_static("x-forwarded-proto"))
         .and_then(|value| value.to_str().ok())
         .unwrap_or_else(|| uri.scheme().map(|s| s.as_str()).unwrap_or("http"));
 
-    let raw_host = uri.host().unwrap_or(host.as_str());
+    // Prefer URI host, fallback to `Host` header
+    let header_host = headers
+        .get(http::header::HOST)
+        .and_then(|v| v.to_str().ok())
+        .unwrap_or_default();
+
+    let raw_host = uri.host().unwrap_or(header_host);
+
     let host_for_url = if let Ok(socket_addr) = raw_host.parse::<SocketAddr>() {
         // Successfully parsed, it's in IP:Port format.
         // For IPv6, we need to enclose it in brackets to form a valid URL.

rustfs/src/admin/handlers.rs

@@ -25,6 +25,7 @@ use hyper::StatusCode;
 use matchit::Params;
 use rustfs_common::heal_channel::HealOpts;
 use rustfs_config::{MAX_ADMIN_REQUEST_BODY_SIZE, MAX_HEAL_REQUEST_SIZE};
+use rustfs_credentials::get_global_action_cred;
 use rustfs_ecstore::admin_server_info::get_server_info;
 use rustfs_ecstore::bucket::bucket_target_sys::BucketTargetSys;
 use rustfs_ecstore::bucket::metadata::BUCKET_TARGETS_FILE;
@@ -35,7 +36,6 @@ use rustfs_ecstore::data_usage::{
     aggregate_local_snapshots, compute_bucket_usage, load_data_usage_from_backend, store_data_usage_in_backend,
 };
 use rustfs_ecstore::error::StorageError;
-use rustfs_ecstore::global::get_global_action_cred;
 use rustfs_ecstore::global::global_rustfs_port;
 use rustfs_ecstore::metrics_realtime::{CollectMetricsOpts, MetricType, collect_local_metrics};
 use rustfs_ecstore::new_object_layer_fn;

rustfs/src/admin/handlers/group.rs

@@ -19,7 +19,7 @@ use crate::{
 use http::{HeaderMap, StatusCode};
 use matchit::Params;
 use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE;
-use rustfs_ecstore::global::get_global_action_cred;
+use rustfs_credentials::get_global_action_cred;
 use rustfs_iam::error::{is_err_no_such_group, is_err_no_such_user};
 use rustfs_madmin::GroupAddRemove;
 use rustfs_policy::policy::action::{Action, AdminAction};

rustfs/src/admin/handlers/policies.rs

@@ -19,7 +19,7 @@ use crate::{
 use http::{HeaderMap, StatusCode};
 use matchit::Params;
 use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE;
-use rustfs_ecstore::global::get_global_action_cred;
+use rustfs_credentials::get_global_action_cred;
 use rustfs_iam::error::is_err_no_such_user;
 use rustfs_iam::store::MappedPolicy;
 use rustfs_policy::policy::{

rustfs/src/admin/handlers/service_account.rs

@@ -19,7 +19,7 @@ use http::HeaderMap;
 use hyper::StatusCode;
 use matchit::Params;
 use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE;
-use rustfs_ecstore::global::get_global_action_cred;
+use rustfs_credentials::get_global_action_cred;
 use rustfs_iam::error::is_err_no_such_service_account;
 use rustfs_iam::sys::{NewServiceAccountOpts, UpdateServiceAccountOpts};
 use rustfs_madmin::{

rustfs/src/admin/handlers/user.rs

@@ -19,7 +19,7 @@ use crate::{
 use http::{HeaderMap, StatusCode};
 use matchit::Params;
 use rustfs_config::{MAX_ADMIN_REQUEST_BODY_SIZE, MAX_IAM_IMPORT_SIZE};
-use rustfs_ecstore::global::get_global_action_cred;
+use rustfs_credentials::get_global_action_cred;
 use rustfs_iam::{
     store::{GroupInfo, MappedPolicy, UserType},
     sys::NewServiceAccountOpts,

rustfs/src/auth.rs

@@ -14,11 +14,10 @@
 
 use http::HeaderMap;
 use http::Uri;
-use rustfs_ecstore::global::get_global_action_cred;
+use rustfs_credentials::{Credentials, get_global_action_cred};
 use rustfs_iam::error::Error as IamError;
 use rustfs_iam::sys::SESSION_POLICY_NAME;
 use rustfs_iam::sys::get_claims_from_token_with_secret;
-use rustfs_policy::auth;
 use rustfs_utils::http::ip::get_source_ip_raw;
 use s3s::S3Error;
 use s3s::S3ErrorCode;
@@ -129,7 +128,7 @@ impl S3Auth for IAMAuth {
 }
 
 // check_key_valid checks the key is valid or not. return the user's credentials and if the user is the owner.
-pub async fn check_key_valid(session_token: &str, access_key: &str) -> S3Result<(auth::Credentials, bool)> {
+pub async fn check_key_valid(session_token: &str, access_key: &str) -> S3Result<(Credentials, bool)> {
     let Some(mut cred) = get_global_action_cred() else {
         return Err(S3Error::with_message(
             S3ErrorCode::InternalError,
@@ -187,7 +186,7 @@ pub async fn check_key_valid(session_token: &str, access_key: &str) -> S3Result<
     Ok((cred, owner))
 }
 
-pub fn check_claims_from_token(token: &str, cred: &auth::Credentials) -> S3Result<HashMap<String, Value>> {
+pub fn check_claims_from_token(token: &str, cred: &Credentials) -> S3Result<HashMap<String, Value>> {
     if !token.is_empty() && cred.access_key.is_empty() {
         return Err(s3_error!(InvalidRequest, "no access key"));
     }
@@ -235,9 +234,20 @@ pub fn get_session_token<'a>(uri: &'a Uri, hds: &'a HeaderMap) -> Option<&'a str
         .or_else(|| get_query_param(uri.query().unwrap_or_default(), "x-amz-security-token"))
 }
 
+/// Get condition values for policy evaluation
+///
+/// # Arguments
+/// * `header` - HTTP headers of the request
+/// * `cred` - User credentials
+/// * `version_id` - Optional version ID of the object
+/// * `region` - Optional region/location constraint
+///
+/// # Returns
+/// * `HashMap<String, Vec<String>>` - Condition values for policy evaluation
+///
 pub fn get_condition_values(
     header: &HeaderMap,
-    cred: &auth::Credentials,
+    cred: &Credentials,
     version_id: Option<&str>,
     region: Option<&str>,
 ) -> HashMap<String, Vec<String>> {
@@ -403,7 +413,14 @@ pub fn get_condition_values(
     args
 }
 
-// Get request authentication type
+/// Get request authentication type
+///
+/// # Arguments
+/// * `header` - HTTP headers of the request
+///
+/// # Returns
+/// * `AuthType` - The determined authentication type
+///
 pub fn get_request_auth_type(header: &HeaderMap) -> AuthType {
     if is_request_signature_v2(header) {
         AuthType::SignedV2
@@ -432,7 +449,14 @@ pub fn get_request_auth_type(header: &HeaderMap) -> AuthType {
     }
 }
 
-// Helper function to determine auth type and signature version
+/// Helper function to determine auth type and signature version
+///
+/// # Arguments
+/// * `header` - HTTP headers of the request
+///
+/// # Returns
+/// * `(String, String)` - Tuple of auth type and signature version
+///
 fn determine_auth_type_and_version(header: &HeaderMap) -> (String, String) {
     match get_request_auth_type(header) {
         AuthType::JWT => ("JWT".to_string(), String::new()),
@@ -450,7 +474,13 @@ fn determine_auth_type_and_version(header: &HeaderMap) -> (String, String) {
     }
 }
 
-// Verify if request has JWT
+/// Verify if request has JWT
+///
+/// # Arguments
+/// * `header` - HTTP headers of the request
+///
+/// # Returns
+/// * `bool` - True if request has JWT, false otherwise
 fn is_request_jwt(header: &HeaderMap) -> bool {
     if let Some(auth) = header.get("authorization") {
         if let Ok(auth_str) = auth.to_str() {
@@ -460,7 +490,13 @@ fn is_request_jwt(header: &HeaderMap) -> bool {
     false
 }
 
-// Verify if request has AWS Signature Version '4'
+/// Verify if request has AWS Signature Version '4'
+///
+/// # Arguments
+/// * `header` - HTTP headers of the request
+///
+/// # Returns
+/// * `bool` - True if request has AWS Signature Version '4', false otherwise
 fn is_request_signature_v4(header: &HeaderMap) -> bool {
     if let Some(auth) = header.get("authorization") {
         if let Ok(auth_str) = auth.to_str() {
@@ -470,7 +506,13 @@ fn is_request_signature_v4(header: &HeaderMap) -> bool {
     false
 }
 
-// Verify if request has AWS Signature Version '2'
+/// Verify if request has AWS Signature Version '2'
+///
+/// # Arguments
+/// * `header` - HTTP headers of the request
+///
+/// # Returns
+/// * `bool` - True if request has AWS Signature Version '2', false otherwise
 fn is_request_signature_v2(header: &HeaderMap) -> bool {
     if let Some(auth) = header.get("authorization") {
         if let Ok(auth_str) = auth.to_str() {
@@ -480,7 +522,13 @@ fn is_request_signature_v2(header: &HeaderMap) -> bool {
     false
 }
 
-// Verify if request has AWS PreSign Version '4'
+/// Verify if request has AWS PreSign Version '4'
+///
+/// # Arguments
+/// * `header` - HTTP headers of the request
+///
+/// # Returns
+/// * `bool` - True if request has AWS PreSign Version '4', false otherwise
 pub(crate) fn is_request_presigned_signature_v4(header: &HeaderMap) -> bool {
     if let Some(credential) = header.get(AMZ_CREDENTIAL) {
         return !credential.to_str().unwrap_or("").is_empty();
@@ -488,7 +536,13 @@ pub(crate) fn is_request_presigned_signature_v4(header: &HeaderMap) -> bool {
     false
 }
 
-// Verify request has AWS PreSign Version '2'
+/// Verify request has AWS PreSign Version '2'
+///
+/// # Arguments
+/// * `header` - HTTP headers of the request
+///
+/// # Returns
+/// * `bool` - True if request has AWS PreSign Version '2', false otherwise
 fn is_request_presigned_signature_v2(header: &HeaderMap) -> bool {
     if let Some(access_key) = header.get(AMZ_ACCESS_KEY_ID) {
         return !access_key.to_str().unwrap_or("").is_empty();
@@ -496,7 +550,13 @@ fn is_request_presigned_signature_v2(header: &HeaderMap) -> bool {
     false
 }
 
-// Verify if request has AWS Post policy Signature Version '4'
+/// Verify if request has AWS Post policy Signature Version '4'
+///
+/// # Arguments
+/// * `header` - HTTP headers of the request
+///
+/// # Returns
+/// * `bool` - True if request has AWS Post policy Signature Version '4', false otherwise
 fn is_request_post_policy_signature_v4(header: &HeaderMap) -> bool {
     if let Some(content_type) = header.get("content-type") {
         if let Ok(ct) = content_type.to_str() {
@@ -506,7 +566,7 @@ fn is_request_post_policy_signature_v4(header: &HeaderMap) -> bool {
     false
 }
 
-// Verify if the request has AWS Streaming Signature Version '4'
+/// Verify if the request has AWS Streaming Signature Version '4'
 fn is_request_sign_streaming_v4(header: &HeaderMap) -> bool {
     if let Some(content_sha256) = header.get("x-amz-content-sha256") {
         if let Ok(sha256_str) = content_sha256.to_str() {
@@ -567,7 +627,7 @@ pub fn get_query_param<'a>(query: &'a str, param_name: &str) -> Option<&'a str>
 mod tests {
     use super::*;
     use http::{HeaderMap, HeaderValue, Uri};
-    use rustfs_policy::auth::Credentials;
+    use rustfs_credentials::Credentials;
     use s3s::auth::SecretKey;
     use serde_json::json;
     use std::collections::HashMap;
@@ -605,7 +665,7 @@ mod tests {
 
     fn create_service_account_credentials() -> Credentials {
         let mut claims = HashMap::new();
-        claims.insert("sa-policy".to_string(), json!("test-policy"));
+        claims.insert(rustfs_credentials::IAM_POLICY_CLAIM_NAME_SA.to_string(), json!("test-policy"));
 
         Credentials {
             access_key: "service-access-key".to_string(),

rustfs/src/config/mod.rs

@@ -73,11 +73,11 @@ pub struct Opt {
     pub server_domains: Vec<String>,
 
     /// Access key used for authentication.
-    #[arg(long, default_value_t = rustfs_config::DEFAULT_ACCESS_KEY.to_string(), env = "RUSTFS_ACCESS_KEY")]
+    #[arg(long, default_value_t = rustfs_credentials::DEFAULT_ACCESS_KEY.to_string(), env = "RUSTFS_ACCESS_KEY")]
     pub access_key: String,
 
     /// Secret key used for authentication.
-    #[arg(long, default_value_t = rustfs_config::DEFAULT_SECRET_KEY.to_string(), env = "RUSTFS_SECRET_KEY")]
+    #[arg(long, default_value_t = rustfs_credentials::DEFAULT_SECRET_KEY.to_string(), env = "RUSTFS_SECRET_KEY")]
     pub secret_key: String,
 
     /// Enable console server

rustfs/src/main.rs

@@ -39,6 +39,7 @@ use rustfs_ahm::{
     scanner::data_scanner::ScannerConfig, shutdown_ahm_services,
 };
 use rustfs_common::{GlobalReadiness, SystemStage, set_global_addr};
+use rustfs_credentials::init_global_action_credentials;
 use rustfs_ecstore::{
     StorageAPI,
     bucket::metadata_sys::init_bucket_metadata_sys,
@@ -147,7 +148,16 @@ async fn run(opt: config::Opt) -> Result<()> {
     );
 
     // Set up AK and SK
-    rustfs_ecstore::global::init_global_action_credentials(Some(opt.access_key.clone()), Some(opt.secret_key.clone()));
+    match init_global_action_credentials(Some(opt.access_key.clone()), Some(opt.secret_key.clone())) {
+        Ok(_) => {
+            info!(target: "rustfs::main::run", "Global action credentials initialized successfully.");
+        }
+        Err(e) => {
+            let msg = format!("init_global_action_credentials failed: {e:?}");
+            error!("{msg}");
+            return Err(Error::other(msg));
+        }
+    };
 
     set_global_rustfs_port(server_port);
 

rustfs/src/server/http.rs

@@ -30,7 +30,7 @@ use hyper_util::{
 };
 use metrics::{counter, histogram};
 use rustfs_common::GlobalReadiness;
-use rustfs_config::{DEFAULT_ACCESS_KEY, DEFAULT_SECRET_KEY, MI_B, RUSTFS_TLS_CERT, RUSTFS_TLS_KEY};
+use rustfs_config::{MI_B, RUSTFS_TLS_CERT, RUSTFS_TLS_KEY};
 use rustfs_protos::proto_gen::node_service::node_service_server::NodeServiceServer;
 use rustfs_utils::net::parse_and_resolve_address;
 use rustls::ServerConfig;
@@ -212,10 +212,13 @@ pub async fn start_http_server(
         info!(target: "rustfs::main::startup","RustFS API: {api_endpoints}  {localhost_endpoint}");
         println!("RustFS Http API: {api_endpoints}  {localhost_endpoint}");
         println!("RustFS Start Time: {now_time}");
-        if DEFAULT_ACCESS_KEY.eq(&opt.access_key) && DEFAULT_SECRET_KEY.eq(&opt.secret_key) {
+        if rustfs_credentials::DEFAULT_ACCESS_KEY.eq(&opt.access_key)
+            && rustfs_credentials::DEFAULT_SECRET_KEY.eq(&opt.secret_key)
+        {
             warn!(
                 "Detected default credentials '{}:{}', we recommend that you change these values with 'RUSTFS_ACCESS_KEY' and 'RUSTFS_SECRET_KEY' environment variables",
-                DEFAULT_ACCESS_KEY, DEFAULT_SECRET_KEY
+                rustfs_credentials::DEFAULT_ACCESS_KEY,
+                rustfs_credentials::DEFAULT_SECRET_KEY
             );
         }
         info!(target: "rustfs::main::startup","For more information, visit https://rustfs.com/docs/");
@@ -685,7 +688,12 @@ fn handle_connection_error(err: &(dyn std::error::Error + 'static)) {
 
 #[allow(clippy::result_large_err)]
 fn check_auth(req: Request<()>) -> std::result::Result<Request<()>, Status> {
-    let token: MetadataValue<_> = "rustfs rpc".parse().unwrap();
+    let token_str = rustfs_credentials::get_grpc_token();
+
+    let token: MetadataValue<_> = token_str.parse().map_err(|e| {
+        error!("Failed to parse RUSTFS_GRPC_AUTH_TOKEN into gRPC metadata value: {}", e);
+        Status::internal("Invalid auth token configuration")
+    })?;
 
     match req.metadata().get("authorization") {
         Some(t) if token == t => Ok(req),

rustfs/src/storage/access.rs

@@ -17,7 +17,6 @@ use crate::auth::{check_key_valid, get_condition_values, get_session_token};
 use crate::license::license_check;
 use rustfs_ecstore::bucket::policy_sys::PolicySys;
 use rustfs_iam::error::Error as IamError;
-use rustfs_policy::auth;
 use rustfs_policy::policy::action::{Action, S3Action};
 use rustfs_policy::policy::{Args, BucketPolicyArgs};
 use s3s::access::{S3Access, S3AccessContext};
@@ -27,7 +26,7 @@ use std::collections::HashMap;
 #[allow(dead_code)]
 #[derive(Default, Clone)]
 pub(crate) struct ReqInfo {
-    pub cred: Option<auth::Credentials>,
+    pub cred: Option<rustfs_credentials::Credentials>,
     pub is_owner: bool,
     pub bucket: Option<String>,
     pub object: Option<String>,
@@ -107,7 +106,7 @@ pub async fn authorize_request<T>(req: &mut S3Request<T>, action: Action) -> S3R
     } else {
         let conditions = get_condition_values(
             &req.headers,
-            &auth::Credentials::default(),
+            &rustfs_credentials::Credentials::default(),
             req_info.version_id.as_deref(),
             req.region.as_deref(),
         );

rustfs/src/storage/ecfs.rs

@@ -93,12 +93,9 @@ use rustfs_kms::{
     types::{EncryptionMetadata, ObjectEncryptionContext},
 };
 use rustfs_notify::{EventArgsBuilder, notifier_global};
-use rustfs_policy::{
-    auth,
-    policy::{
-        action::{Action, S3Action},
-        {BucketPolicy, BucketPolicyArgs, Validator},
-    },
+use rustfs_policy::policy::{
+    action::{Action, S3Action},
+    {BucketPolicy, BucketPolicyArgs, Validator},
 };
 use rustfs_rio::{CompressReader, DecryptReader, EncryptReader, EtagReader, HardLimitReader, HashReader, Reader, WarpReader};
 use rustfs_s3select_api::{
@@ -4692,7 +4689,7 @@ impl S3 for FS {
             .await
             .map_err(ApiError::from)?;
 
-        let conditions = get_condition_values(&req.headers, &auth::Credentials::default(), None, None);
+        let conditions = get_condition_values(&req.headers, &rustfs_credentials::Credentials::default(), None, None);
 
         let read_only = PolicySys::is_allowed(&BucketPolicyArgs {
             bucket: &bucket,

rustfs/src/storage/tonic_service.rs

@@ -1774,11 +1774,34 @@ impl Node for NodeService {
 
     async fn get_metrics(&self, request: Request<GetMetricsRequest>) -> Result<Response<GetMetricsResponse>, Status> {
         let request = request.into_inner();
-        let mut buf_t = Deserializer::new(Cursor::new(request.metric_type));
-        let t: MetricType = Deserialize::deserialize(&mut buf_t).unwrap();
 
+        // Deserialize metric_type with error handling
+        let mut buf_t = Deserializer::new(Cursor::new(request.metric_type));
+        let t: MetricType = match Deserialize::deserialize(&mut buf_t) {
+            Ok(t) => t,
+            Err(err) => {
+                error!("Failed to deserialize metric_type: {}", err);
+                return Ok(Response::new(GetMetricsResponse {
+                    success: false,
+                    realtime_metrics: Bytes::new(),
+                    error_info: Some(format!("Invalid metric_type: {}", err)),
+                }));
+            }
+        };
+
+        // Deserialize opts with error handling
         let mut buf_o = Deserializer::new(Cursor::new(request.opts));
-        let opts: CollectMetricsOpts = Deserialize::deserialize(&mut buf_o).unwrap();
+        let opts: CollectMetricsOpts = match Deserialize::deserialize(&mut buf_o) {
+            Ok(opts) => opts,
+            Err(err) => {
+                error!("Failed to deserialize opts: {}", err);
+                return Ok(Response::new(GetMetricsResponse {
+                    success: false,
+                    realtime_metrics: Bytes::new(),
+                    error_info: Some(format!("Invalid opts: {}", err)),
+                }));
+            }
+        };
 
         let info = collect_local_metrics(t, &opts).await;
 
@@ -3648,4 +3671,32 @@ mod tests {
         // Should return None for non-existent disk
         assert!(disk.is_none());
     }
+
+    #[tokio::test]
+    async fn test_get_metrics_invalid_metric_type() {
+        let service = create_test_node_service();
+        let request = Request::new(GetMetricsRequest {
+            metric_type: Bytes::from(vec![0x00u8, 0x01u8]), // Invalid rmp data
+            opts: Bytes::new(),                             // Valid or invalid
+        });
+        let response = service.get_metrics(request).await.unwrap().into_inner();
+        assert!(!response.success);
+        assert!(response.error_info.is_some());
+    }
+
+    #[tokio::test]
+    async fn test_get_metrics_invalid_opts() {
+        let service = create_test_node_service();
+        // Serialize a valid MetricType
+        let metric_type = MetricType::DISK;
+        let metric_type_bytes = rmp_serde::to_vec(&metric_type).unwrap();
+
+        let request = Request::new(GetMetricsRequest {
+            metric_type: Bytes::from(metric_type_bytes),
+            opts: Bytes::from(vec![0x00u8, 0x01u8]), // Invalid rmp data
+        });
+        let response = service.get_metrics(request).await.unwrap().into_inner();
+        assert!(!response.success);
+        assert!(response.error_info.is_some());
+    }
 }