rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-08-18 03:58:29 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: admin http api (#1026)

Browse Source 
* feat: catalog list

* feat: catalog list

* feat:api

* feat: leader info

* feat: use constant

* fix: ci

* feat: query heartbeat by ip

* ut: add test

* fix: cr

* fix: cr

* fix: cr
This commit is contained in:
Xieqijun
2023-02-22 14:18:37 +08:00
committed by GitHub
parent bcd44b90c1
commit 390e9095f6
6 changed files with 370 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/catalog/src/helper.rs
+4 -4
View File
@@ -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_]*";
src/meta-srv/src/error.rs
+13
View File
@@ -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<T> = std::result::Result<T, Error>;
@@ -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(),
src/meta-srv/src/service/admin.rs
+37
View File
@@ -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)
src/meta-srv/src/service/admin/heartbeat.rs
+77 -2
View File
@@ -29,14 +29,22 @@ pub struct HeartBeatHandler {
#[async_trait::async_trait]
impl HttpHandler for HeartBeatHandler {
async fn handle(&self, _: &str, _: &HashMap<String, String>) -> Result<http::Response<String>> {
async fn handle(
&self,
_: &str,
params: &HashMap<String, String>,
) -> Result<http::Response<String>> {
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<StatValue> = stat_kvs.into_values().collect();
let mut stat_vals: Vec<StatValue> = 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<StatValues> for String {
})
}
}
fn filter_by_addr(stat_vals: Vec<StatValue>, addr: &str) -> Vec<StatValue> {
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
);
}
}
src/meta-srv/src/service/admin/leader.rs
+42
View File
@@ -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<ElectionRef>,
}
#[async_trait::async_trait]
impl HttpHandler for LeaderHandler {
async fn handle(&self, _: &str, _: &HashMap<String, String>) -> Result<http::Response<String>> {
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)
}
}
src/meta-srv/src/service/admin/meta.rs
+197
View File
@@ -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<String, String>) -> Result<http::Response<String>> {
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<String, String>,
) -> Result<http::Response<String>> {
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<String, String>,
) -> Result<http::Response<String>> {
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<String, String>,
) -> Result<http::Response<String>> {
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<http::Response<String>> {
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<Vec<String>> {
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::<Vec<&str>>();
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]);
}
}