feat: adds runtime_metrics

src/catalog/src/information_schema.rs

@@ -16,6 +16,7 @@ mod columns;
 mod key_column_usage;
 mod memory_table;
 mod predicate;
+mod runtime_metrics;
 mod schemata;
 mod table_names;
 mod tables;
@@ -46,6 +47,7 @@ use self::columns::InformationSchemaColumns;
 use crate::error::Result;
 use crate::information_schema::key_column_usage::InformationSchemaKeyColumnUsage;
 use crate::information_schema::memory_table::{get_schema_columns, MemoryTable};
+use crate::information_schema::runtime_metrics::InformationSchemaMetrics;
 use crate::information_schema::schemata::InformationSchemaSchemata;
 use crate::information_schema::tables::InformationSchemaTables;
 use crate::CatalogManager;
@@ -145,6 +147,10 @@ impl InformationSchemaProvider {
  tables.insert(TABLES.to_string(), self.build_table(TABLES).unwrap());
  tables.insert(SCHEMATA.to_string(), self.build_table(SCHEMATA).unwrap());
  tables.insert(COLUMNS.to_string(), self.build_table(COLUMNS).unwrap());
+ tables.insert(
+ RUNTIME_METRICS.to_string(),
+ self.build_table(RUNTIME_METRICS).unwrap(),
+ );
  tables.insert(
  KEY_COLUMN_USAGE.to_string(),
  self.build_table(KEY_COLUMN_USAGE).unwrap(),
@@ -209,6 +215,7 @@ impl InformationSchemaProvider {
  self.catalog_name.clone(),
  self.catalog_manager.clone(),
  )) as _),
+ RUNTIME_METRICS => Some(Arc::new(InformationSchemaMetrics::new())),
  _ => None,
  }
  }

src/catalog/src/information_schema/runtime_metrics.rs

@@ -0,0 +1,304 @@
+// 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::sync::Arc;
+
+use arrow_schema::SchemaRef as ArrowSchemaRef;
+use common_catalog::consts::INFORMATION_SCHEMA_RUNTIME_METRICS_TABLE_ID;
+use common_error::ext::BoxedError;
+use common_query::physical_plan::TaskContext;
+use common_recordbatch::adapter::RecordBatchStreamAdapter;
+use common_recordbatch::{RecordBatch, SendableRecordBatchStream};
+use datafusion::physical_plan::stream::RecordBatchStreamAdapter as DfRecordBatchStreamAdapter;
+use datafusion::physical_plan::streaming::PartitionStream as DfPartitionStream;
+use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
+use datatypes::prelude::{ConcreteDataType, MutableVector};
+use datatypes::scalars::ScalarVectorBuilder;
+use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
+use datatypes::vectors::{
+ ConstantVector, Float64VectorBuilder, StringVector, StringVectorBuilder, VectorRef,
+};
+use prometheus::proto::{LabelPair, MetricType};
+use snafu::ResultExt;
+use store_api::storage::{ScanRequest, TableId};
+
+use super::{InformationTable, RUNTIME_METRICS};
+use crate::error::{CreateRecordBatchSnafu, InternalSnafu, Result};
+
+pub(super) struct InformationSchemaMetrics {
+ schema: SchemaRef,
+}
+
+const METRIC_NAME: &str = "metric_name";
+const METRIC_VALUE: &str = "value";
+const METRIC_LABELS: &str = "labels";
+const NODE: &str = "node";
+const NODE_TYPE: &str = "node_type";
+
+/// The `information_schema.runtime_metrics` virtual table.
+/// It provides the GreptimeDB runtime metrics for the users by SQL.
+impl InformationSchemaMetrics {
+ pub(super) fn new() -> Self {
+ Self {
+ schema: Self::schema(),
+ }
+ }
+
+ fn schema() -> SchemaRef {
+ Arc::new(Schema::new(vec![
+ ColumnSchema::new(METRIC_NAME, ConcreteDataType::string_datatype(), false),
+ ColumnSchema::new(METRIC_VALUE, ConcreteDataType::float64_datatype(), false),
+ ColumnSchema::new(METRIC_LABELS, ConcreteDataType::string_datatype(), true),
+ ColumnSchema::new(NODE, ConcreteDataType::string_datatype(), false),
+ ColumnSchema::new(NODE_TYPE, ConcreteDataType::string_datatype(), false),
+ ]))
+ }
+
+ fn builder(&self) -> InformationSchemaMetricsBuilder {
+ InformationSchemaMetricsBuilder::new(self.schema.clone())
+ }
+}
+
+impl InformationTable for InformationSchemaMetrics {
+ fn table_id(&self) -> TableId {
+ INFORMATION_SCHEMA_RUNTIME_METRICS_TABLE_ID
+ }
+
+ fn table_name(&self) -> &'static str {
+ RUNTIME_METRICS
+ }
+
+ fn schema(&self) -> SchemaRef {
+ self.schema.clone()
+ }
+
+ fn to_stream(&self, request: ScanRequest) -> Result<SendableRecordBatchStream> {
+ let schema = self.schema.arrow_schema().clone();
+ let mut builder = self.builder();
+ let stream = Box::pin(DfRecordBatchStreamAdapter::new(
+ schema,
+ futures::stream::once(async move {
+ builder
+ .make_metrics(Some(request))
+ .await
+ .map(|x| x.into_df_record_batch())
+ .map_err(Into::into)
+ }),
+ ));
+ Ok(Box::pin(
+ RecordBatchStreamAdapter::try_new(stream)
+ .map_err(BoxedError::new)
+ .context(InternalSnafu)?,
+ ))
+ }
+}
+
+struct InformationSchemaMetricsBuilder {
+ schema: SchemaRef,
+
+ metric_names: StringVectorBuilder,
+ metric_values: Float64VectorBuilder,
+ metric_labels: StringVectorBuilder,
+}
+
+impl InformationSchemaMetricsBuilder {
+ fn new(schema: SchemaRef) -> Self {
+ Self {
+ schema,
+ metric_names: StringVectorBuilder::with_capacity(42),
+ metric_values: Float64VectorBuilder::with_capacity(42),
+ metric_labels: StringVectorBuilder::with_capacity(42),
+ }
+ }
+
+ fn add_metric(&mut self, metric_name: &str, labels: String, metric_value: f64) {
+ self.metric_names.push(Some(metric_name));
+ self.metric_values.push(Some(metric_value));
+ self.metric_labels.push(Some(&labels));
+ }
+
+ async fn make_metrics(&mut self, _request: Option<ScanRequest>) -> Result<RecordBatch> {
+ let metric_families = prometheus::gather();
+
+ for mf in metric_families {
+ let mf_type = mf.get_field_type();
+ let mf_name = mf.get_name();
+
+ for m in mf.get_metric() {
+ match mf_type {
+ MetricType::COUNTER => self.add_metric(
+ mf_name,
+ join_labels(m.get_label(), None),
+ m.get_counter().get_value(),
+ ),
+ MetricType::GAUGE => self.add_metric(
+ mf_name,
+ join_labels(m.get_label(), None),
+ m.get_gauge().get_value(),
+ ),
+ MetricType::HISTOGRAM => {
+ let h = m.get_histogram();
+ let mut inf_seen = false;
+ let metric_name = format!("{}_bucket", mf_name);
+ for b in h.get_bucket() {
+ let upper_bound = b.get_upper_bound();
+ self.add_metric(
+ &metric_name,
+ join_labels(m.get_label(), Some(("le", upper_bound.to_string()))),
+ b.get_cumulative_count() as f64,
+ );
+ if upper_bound.is_sign_positive() && upper_bound.is_infinite() {
+ inf_seen = true;
+ }
+ }
+ if !inf_seen {
+ self.add_metric(
+ &metric_name,
+ join_labels(m.get_label(), Some(("le", "+Inf".to_string()))),
+ h.get_sample_count() as f64,
+ );
+ }
+ self.add_metric(
+ format!("{}_sum", mf_name).as_str(),
+ join_labels(m.get_label(), None),
+ h.get_sample_sum(),
+ );
+ self.add_metric(
+ format!("{}_count", mf_name).as_str(),
+ join_labels(m.get_label(), None),
+ h.get_sample_count() as f64,
+ );
+ }
+ MetricType::SUMMARY => {
+ let s = m.get_summary();
+ for q in s.get_quantile() {
+ self.add_metric(
+ mf_name,
+ join_labels(
+ m.get_label(),
+ Some(("quantile", q.get_quantile().to_string())),
+ ),
+ q.get_value(),
+ );
+ }
+ self.add_metric(
+ format!("{}_sum", mf_name).as_str(),
+ join_labels(m.get_label(), None),
+ s.get_sample_sum(),
+ );
+ self.add_metric(
+ format!("{}_count", mf_name).as_str(),
+ join_labels(m.get_label(), None),
+ s.get_sample_count() as f64,
+ );
+ }
+ MetricType::UNTYPED => {
+ // `TextEncoder` `MetricType::UNTYPED` unimplemented
+ // To keep the implementation consistent and not cause unexpected panics, we do nothing here.
+ }
+ };
+ }
+ }
+
+ self.finish()
+ }
+
+ fn finish(&mut self) -> Result<RecordBatch> {
+ let unknowns = Arc::new(StringVector::from(vec!["unknown"]));
+ let unknowns = Arc::new(ConstantVector::new(unknowns, self.metric_names.len()));
+
+ let columns: Vec<VectorRef> = vec![
+ Arc::new(self.metric_names.finish()),
+ Arc::new(self.metric_values.finish()),
+ Arc::new(self.metric_labels.finish()),
+ // TODO(dennis): supports node and node_type for cluster
+ unknowns.clone(),
+ unknowns,
+ ];
+
+ RecordBatch::new(self.schema.clone(), columns).context(CreateRecordBatchSnafu)
+ }
+}
+
+impl DfPartitionStream for InformationSchemaMetrics {
+ fn schema(&self) -> &ArrowSchemaRef {
+ self.schema.arrow_schema()
+ }
+
+ fn execute(&self, _: Arc<TaskContext>) -> DfSendableRecordBatchStream {
+ let schema = self.schema.arrow_schema().clone();
+ let mut builder = self.builder();
+ Box::pin(DfRecordBatchStreamAdapter::new(
+ schema,
+ futures::stream::once(async move {
+ builder
+ .make_metrics(None)
+ .await
+ .map(|x| x.into_df_record_batch())
+ .map_err(Into::into)
+ }),
+ ))
+ }
+}
+
+fn join_labels(pairs: &[LabelPair], addon: Option<(&'static str, String)>) -> String {
+ let mut labels = Vec::with_capacity(pairs.len() + 1 + if addon.is_some() { 1 } else { 0 });
+ for label in pairs {
+ labels.push(format!("{}={}", label.get_name(), label.get_value(),));
+ }
+ if let Some(addon) = addon {
+ labels.push(format!("{}={}", addon.0, addon.1));
+ }
+ labels.sort_unstable();
+ labels.join(", ")
+}
+
+#[cfg(test)]
+mod tests {
+ use common_recordbatch::RecordBatches;
+
+ use super::*;
+
+ fn new_pair(name: &str, value: &str) -> LabelPair {
+ let mut pair = LabelPair::new();
+ pair.set_name(name.to_string());
+ pair.set_value(value.to_string());
+ pair
+ }
+
+ #[test]
+ fn test_join_labels() {
+ let pairs = vec![new_pair("le", "0.999"), new_pair("host", "host1")];
+
+ assert_eq!("host=host1, le=0.999", join_labels(&pairs, None));
+ assert_eq!(
+ "a=a_value, host=host1, le=0.999",
+ join_labels(&pairs, Some(("a", "a_value".to_string())))
+ );
+ }
+
+ #[tokio::test]
+ async fn test_make_metrics() {
+ let metrics = InformationSchemaMetrics::new();
+
+ let stream = metrics.to_stream(ScanRequest::default()).unwrap();
+
+ let batches = RecordBatches::try_collect(stream).await.unwrap();
+
+ let result_literal = batches.pretty_print().unwrap();
+
+ assert!(result_literal.contains(METRIC_NAME));
+ assert!(result_literal.contains(METRIC_VALUE));
+ }
+}

src/catalog/src/information_schema/table_names.rs

@@ -38,3 +38,4 @@ pub const TABLE_PRIVILEGES: &str = "table_privileges";
 pub const TRIGGERS: &str = "triggers";
 pub const GLOBAL_STATUS: &str = "global_status";
 pub const SESSION_STATUS: &str = "session_status";
+pub const RUNTIME_METRICS: &str = "runtime_metrics";

src/common/catalog/src/consts.rs

@@ -80,6 +80,8 @@ pub const INFORMATION_SCHEMA_TRIGGERS_TABLE_ID: u32 = 24;
 pub const INFORMATION_SCHEMA_GLOBAL_STATUS_TABLE_ID: u32 = 25;
 /// id for information_schema.SESSION_STATUS
 pub const INFORMATION_SCHEMA_SESSION_STATUS_TABLE_ID: u32 = 26;
+/// id for information_schema.RUNTIME_METRICS
+pub const INFORMATION_SCHEMA_RUNTIME_METRICS_TABLE_ID: u32 = 27;
 /// ----- End of information_schema tables -----
 
 pub const MITO_ENGINE: &str = "mito";

tests/cases/standalone/common/show/show_databases_tables.result

@@ -38,6 +38,7 @@ show tables;
 | profiling |
 | referential_constraints |
 | routines |
+| runtime_metrics |
 | schema_privileges |
 | schemata |
 | session_status |