diff --git a/src/catalog/src/helper.rs b/src/catalog/src/helper.rs index 5d2d39ed71..182533b1d8 100644 --- a/src/catalog/src/helper.rs +++ b/src/catalog/src/helper.rs @@ -24,10 +24,10 @@ use serde::{Deserialize, Serialize, Serializer}; use snafu::{ensure, OptionExt, ResultExt}; use table::metadata::{RawTableInfo, TableId, TableVersion}; -const CATALOG_KEY_PREFIX: &str = "__c"; -const SCHEMA_KEY_PREFIX: &str = "__s"; -const TABLE_GLOBAL_KEY_PREFIX: &str = "__tg"; -const TABLE_REGIONAL_KEY_PREFIX: &str = "__tr"; +pub const CATALOG_KEY_PREFIX: &str = "__c"; +pub const SCHEMA_KEY_PREFIX: &str = "__s"; +pub const TABLE_GLOBAL_KEY_PREFIX: &str = "__tg"; +pub const TABLE_REGIONAL_KEY_PREFIX: &str = "__tr"; const ALPHANUMERICS_NAME_PATTERN: &str = "[a-zA-Z_][a-zA-Z0-9_]*"; diff --git a/src/meta-srv/src/error.rs b/src/meta-srv/src/error.rs index fb70ea004e..a5dd4cd621 100644 --- a/src/meta-srv/src/error.rs +++ b/src/meta-srv/src/error.rs @@ -12,6 +12,8 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::string::FromUtf8Error; + use common_error::prelude::*; use tonic::codegen::http; use tonic::{Code, Status}; @@ -259,6 +261,15 @@ pub enum Error { #[snafu(display("Distributed lock is not configured"))] LockNotConfig { backtrace: Backtrace }, + + #[snafu(display("Invalid utf-8 value, source: {:?}", source))] + InvalidUtf8Value { + source: FromUtf8Error, + backtrace: Backtrace, + }, + + #[snafu(display("Missing required parameter, param: {:?}", param))] + MissingRequiredParameter { param: String }, } pub type Result = std::result::Result; @@ -303,6 +314,7 @@ impl ErrorExt for Error { | Error::ExceededRetryLimit { .. } | Error::StartGrpc { .. } => StatusCode::Internal, Error::EmptyKey { .. } + | Error::MissingRequiredParameter { .. } | Error::EmptyTableName { .. } | Error::InvalidLeaseKey { .. } | Error::InvalidStatKey { .. } @@ -319,6 +331,7 @@ impl ErrorExt for Error { | Error::MoveValue { .. } | Error::InvalidKvsLength { .. } | Error::InvalidTxnResult { .. } + | Error::InvalidUtf8Value { .. } | Error::Unexpected { .. } => StatusCode::Unexpected, Error::TableNotFound { .. } => StatusCode::TableNotFound, Error::InvalidCatalogValue { source, .. } => source.status_code(), diff --git a/src/meta-srv/src/service/admin.rs b/src/meta-srv/src/service/admin.rs index 5d15c173ce..771f12aaf1 100644 --- a/src/meta-srv/src/service/admin.rs +++ b/src/meta-srv/src/service/admin.rs @@ -14,6 +14,8 @@ mod health; mod heartbeat; +mod leader; +mod meta; use std::collections::HashMap; use std::convert::Infallible; @@ -36,6 +38,41 @@ pub fn make_admin_service(meta_srv: MetaSrv) -> Admin { }, ); + let router = router.route( + "/catalogs", + meta::CatalogsHandler { + kv_store: meta_srv.kv_store(), + }, + ); + + let router = router.route( + "/schemas", + meta::SchemasHandler { + kv_store: meta_srv.kv_store(), + }, + ); + + let router = router.route( + "/tables", + meta::TablesHandler { + kv_store: meta_srv.kv_store(), + }, + ); + + let router = router.route( + "/table", + meta::TableHandler { + kv_store: meta_srv.kv_store(), + }, + ); + + let router = router.route( + "/leader", + leader::LeaderHandler { + election: meta_srv.election(), + }, + ); + let router = Router::nest("/admin", router); Admin::new(router) diff --git a/src/meta-srv/src/service/admin/heartbeat.rs b/src/meta-srv/src/service/admin/heartbeat.rs index 489c236b56..1f83efe8de 100644 --- a/src/meta-srv/src/service/admin/heartbeat.rs +++ b/src/meta-srv/src/service/admin/heartbeat.rs @@ -29,14 +29,22 @@ pub struct HeartBeatHandler { #[async_trait::async_trait] impl HttpHandler for HeartBeatHandler { - async fn handle(&self, _: &str, _: &HashMap) -> Result> { + async fn handle( + &self, + _: &str, + params: &HashMap, + ) -> Result> { let meta_peer_client = self .meta_peer_client .as_ref() .context(error::NoMetaPeerClientSnafu)?; let stat_kvs = meta_peer_client.get_all_dn_stat_kvs().await?; - let stat_vals: Vec = stat_kvs.into_values().collect(); + let mut stat_vals: Vec = stat_kvs.into_values().collect(); + + if let Some(addr) = params.get("addr") { + stat_vals = filter_by_addr(stat_vals, addr); + } let result = StatValues { stat_vals }.try_into()?; http::Response::builder() @@ -61,3 +69,70 @@ impl TryFrom for String { }) } } + +fn filter_by_addr(stat_vals: Vec, addr: &str) -> Vec { + stat_vals + .into_iter() + .filter(|stat_val| stat_val.stats.iter().any(|stat| stat.addr == addr)) + .collect() +} + +#[cfg(test)] +mod tests { + use crate::handler::node_stat::Stat; + use crate::keys::StatValue; + use crate::service::admin::heartbeat::filter_by_addr; + + #[tokio::test] + async fn test_filter_by_addr() { + let stat_value1 = StatValue { + stats: vec![ + Stat { + addr: "127.0.0.1:3001".to_string(), + timestamp_millis: 1, + ..Default::default() + }, + Stat { + addr: "127.0.0.1:3001".to_string(), + timestamp_millis: 2, + ..Default::default() + }, + ], + }; + + let stat_value2 = StatValue { + stats: vec![ + Stat { + addr: "127.0.0.1:3002".to_string(), + timestamp_millis: 3, + ..Default::default() + }, + Stat { + addr: "127.0.0.1:3002".to_string(), + timestamp_millis: 4, + ..Default::default() + }, + Stat { + addr: "127.0.0.1:3002".to_string(), + timestamp_millis: 5, + ..Default::default() + }, + ], + }; + + let mut stat_vals = vec![stat_value1, stat_value2]; + stat_vals = filter_by_addr(stat_vals, "127.0.0.1:3002"); + assert_eq!(stat_vals.len(), 1); + assert_eq!(stat_vals.get(0).unwrap().stats.len(), 3); + assert_eq!( + stat_vals + .get(0) + .unwrap() + .stats + .get(0) + .unwrap() + .timestamp_millis, + 3 + ); + } +} diff --git a/src/meta-srv/src/service/admin/leader.rs b/src/meta-srv/src/service/admin/leader.rs new file mode 100644 index 0000000000..2c650a1924 --- /dev/null +++ b/src/meta-srv/src/service/admin/leader.rs @@ -0,0 +1,42 @@ +// Copyright 2023 Greptime Team +// +// 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. +use std::collections::HashMap; + +use snafu::ResultExt; +use tonic::codegen::http; + +use crate::error::{self, Result}; +use crate::metasrv::ElectionRef; +use crate::service::admin::HttpHandler; + +pub struct LeaderHandler { + pub election: Option, +} + +#[async_trait::async_trait] +impl HttpHandler for LeaderHandler { + async fn handle(&self, _: &str, _: &HashMap) -> Result> { + if let Some(election) = &self.election { + let leader_addr = election.leader().await?.0; + return http::Response::builder() + .status(http::StatusCode::OK) + .body(leader_addr) + .context(error::InvalidHttpBodySnafu); + } + http::Response::builder() + .status(http::StatusCode::OK) + .body("election info is None".to_string()) + .context(error::InvalidHttpBodySnafu) + } +} diff --git a/src/meta-srv/src/service/admin/meta.rs b/src/meta-srv/src/service/admin/meta.rs new file mode 100644 index 0000000000..bafa741d91 --- /dev/null +++ b/src/meta-srv/src/service/admin/meta.rs @@ -0,0 +1,197 @@ +// Copyright 2023 Greptime Team +// +// 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. + +use std::collections::HashMap; + +use api::v1::meta::{RangeRequest, RangeResponse}; +use catalog::helper::{CATALOG_KEY_PREFIX, SCHEMA_KEY_PREFIX, TABLE_GLOBAL_KEY_PREFIX}; +use snafu::{OptionExt, ResultExt}; +use tonic::codegen::http; + +use crate::error::Result; +use crate::service::admin::HttpHandler; +use crate::service::store::ext::KvStoreExt; +use crate::service::store::kv::KvStoreRef; +use crate::{error, util}; + +pub struct CatalogsHandler { + pub kv_store: KvStoreRef, +} + +pub struct SchemasHandler { + pub kv_store: KvStoreRef, +} + +pub struct TablesHandler { + pub kv_store: KvStoreRef, +} + +pub struct TableHandler { + pub kv_store: KvStoreRef, +} + +#[async_trait::async_trait] +impl HttpHandler for CatalogsHandler { + async fn handle(&self, _: &str, _: &HashMap) -> Result> { + get_http_response_by_prefix(String::from(CATALOG_KEY_PREFIX), &self.kv_store).await + } +} + +#[async_trait::async_trait] +impl HttpHandler for SchemasHandler { + async fn handle( + &self, + _: &str, + params: &HashMap, + ) -> Result> { + let catalog = params + .get("catalog_name") + .context(error::MissingRequiredParameterSnafu { + param: "catalog_name", + })?; + let prefix = format!("{SCHEMA_KEY_PREFIX}-{catalog}",); + get_http_response_by_prefix(prefix, &self.kv_store).await + } +} + +#[async_trait::async_trait] +impl HttpHandler for TablesHandler { + async fn handle( + &self, + _: &str, + params: &HashMap, + ) -> Result> { + let catalog = params + .get("catalog_name") + .context(error::MissingRequiredParameterSnafu { + param: "catalog_name", + })?; + + let schema = params + .get("schema_name") + .context(error::MissingRequiredParameterSnafu { + param: "schema_name", + })?; + let prefix = format!("{TABLE_GLOBAL_KEY_PREFIX}-{catalog}-{schema}",); + get_http_response_by_prefix(prefix, &self.kv_store).await + } +} + +#[async_trait::async_trait] +impl HttpHandler for TableHandler { + async fn handle( + &self, + _: &str, + params: &HashMap, + ) -> Result> { + let table_name = params + .get("full_table_name") + .map(|full_table_name| full_table_name.replace('.', "-")) + .context(error::MissingRequiredParameterSnafu { + param: "full_table_name", + })?; + let table_key = format!("{TABLE_GLOBAL_KEY_PREFIX}-{table_name}"); + + let response = self.kv_store.get(table_key.into_bytes()).await?; + let mut value: String = "Not found result".to_string(); + if let Some(key_value) = response { + value = String::from_utf8(key_value.value).context(error::InvalidUtf8ValueSnafu)?; + } + http::Response::builder() + .status(http::StatusCode::OK) + .body(value) + .context(error::InvalidHttpBodySnafu) + } +} + +/// Get kv_store's key list with http response format by prefix key +async fn get_http_response_by_prefix( + key_prefix: String, + kv_store: &KvStoreRef, +) -> Result> { + let keys = get_keys_by_prefix(key_prefix, kv_store).await?; + let body = serde_json::to_string(&keys).context(error::SerializeToJsonSnafu { + input: format!("{keys:?}"), + })?; + + http::Response::builder() + .status(http::StatusCode::OK) + .body(body) + .context(error::InvalidHttpBodySnafu) +} + +/// Get kv_store's key list by prefix key +async fn get_keys_by_prefix(key_prefix: String, kv_store: &KvStoreRef) -> Result> { + let key_prefix_u8 = key_prefix.clone().into_bytes(); + let range_end = util::get_prefix_end_key(&key_prefix_u8); + let req = RangeRequest { + key: key_prefix_u8, + range_end, + ..Default::default() + }; + + let response: RangeResponse = kv_store.range(req).await?; + + let kvs = response.kvs; + let mut values = Vec::with_capacity(kvs.len()); + for kv in kvs { + let value = String::from_utf8(kv.key).context(error::InvalidUtf8ValueSnafu)?; + let split_list = value.split(&key_prefix).collect::>(); + if let Some(v) = split_list.get(1) { + values.push(v.to_string()); + } + } + Ok(values) +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use api::v1::meta::PutRequest; + + use crate::service::admin::meta::get_keys_by_prefix; + use crate::service::store::kv::KvStoreRef; + use crate::service::store::memory::MemStore; + + #[tokio::test] + async fn test_get_list_by_prefix() { + let in_mem = Arc::new(MemStore::new()) as KvStoreRef; + + in_mem + .put(PutRequest { + key: "test_key1".as_bytes().to_vec(), + value: "test_val1".as_bytes().to_vec(), + ..Default::default() + }) + .await + .unwrap(); + + in_mem + .put(PutRequest { + key: "test_key2".as_bytes().to_vec(), + value: "test_val2".as_bytes().to_vec(), + ..Default::default() + }) + .await + .unwrap(); + + let keys = get_keys_by_prefix(String::from("test_key"), &in_mem) + .await + .unwrap(); + + assert_eq!("1", keys[0]); + assert_eq!("2", keys[1]); + } +}