feat: improve observability for procedure

Cargo.toml

@@ -120,7 +120,7 @@ etcd-client = { version = "0.13" }
 fst = "0.4.7"
 futures = "0.3"
 futures-util = "0.3"
-greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "c437b55725b7f5224fe9d46db21072b4a682ee4b" }
+greptime-proto = { git = "https://github.com/realtaobo/greptime-proto.git", rev = "b83e8318895e3f17334a3e1d5da9565455513816" }
 humantime = "2.1"
 humantime-serde = "1.1"
 itertools = "0.10"

src/catalog/Cargo.toml

@@ -22,6 +22,7 @@ common-config.workspace = true
 common-error.workspace = true
 common-macro.workspace = true
 common-meta.workspace = true
+common-procedure.workspace = true
 common-query.workspace = true
 common-recordbatch.workspace = true
 common-telemetry.workspace = true

src/catalog/src/error.rs

@@ -82,6 +82,20 @@ pub enum Error {
         location: Location,
     },
 
+    #[snafu(display("Failed to list procedures"))]
+    ListProcedures {
+        #[snafu(implicit)]
+        location: Location,
+        source: BoxedError,
+    },
+
+    #[snafu(display("convert proto data error"))]
+    ConvertProtoData {
+        #[snafu(implicit)]
+        location: Location,
+        source: BoxedError,
+    },
+
     #[snafu(display("Failed to re-compile script due to internal error"))]
     CompileScriptInternal {
         #[snafu(implicit)]
@@ -283,7 +297,9 @@ impl ErrorExt for Error {
             | Error::ListNodes { source, .. }
             | Error::ListSchemas { source, .. }
             | Error::ListTables { source, .. }
-            | Error::ListFlows { source, .. } => source.status_code(),
+            | Error::ListFlows { source, .. }
+            | Error::ListProcedures { source, .. }
+            | Error::ConvertProtoData { source, .. } => source.status_code(),
 
             Error::CreateTable { source, .. } => source.status_code(),
 

src/catalog/src/kvbackend/manager.rs

@@ -31,6 +31,7 @@ use common_meta::key::table_info::TableInfoValue;
 use common_meta::key::table_name::TableNameKey;
 use common_meta::key::{TableMetadataManager, TableMetadataManagerRef};
 use common_meta::kv_backend::KvBackendRef;
+use common_procedure::ProcedureManagerRef;
 use futures_util::stream::BoxStream;
 use futures_util::{StreamExt, TryStreamExt};
 use meta_client::client::MetaClient;
@@ -66,6 +67,7 @@ pub struct KvBackendCatalogManager {
     /// A sub-CatalogManager that handles system tables
     system_catalog: SystemCatalog,
     cache_registry: LayeredCacheRegistryRef,
+    procedure_manager: Option<ProcedureManagerRef>,
 }
 
 const CATALOG_CACHE_MAX_CAPACITY: u64 = 128;
@@ -76,6 +78,7 @@ impl KvBackendCatalogManager {
         meta_client: Option<Arc<MetaClient>>,
         backend: KvBackendRef,
         cache_registry: LayeredCacheRegistryRef,
+        procedure_manager: Option<ProcedureManagerRef>,
     ) -> Arc<Self> {
         Arc::new_cyclic(|me| Self {
             mode,
@@ -103,6 +106,7 @@ impl KvBackendCatalogManager {
                 backend,
             },
             cache_registry,
+            procedure_manager,
         })
     }
 
@@ -129,6 +133,10 @@ impl KvBackendCatalogManager {
     pub fn table_metadata_manager_ref(&self) -> &TableMetadataManagerRef {
         &self.table_metadata_manager
     }
+
+    pub fn procedure_manager(&self) -> Option<ProcedureManagerRef> {
+        self.procedure_manager.clone()
+    }
 }
 
 #[async_trait::async_trait]

src/catalog/src/system_schema/information_schema.rs

@@ -18,6 +18,7 @@ pub mod flows;
 mod information_memory_table;
 pub mod key_column_usage;
 mod partitions;
+mod procedure_info;
 mod region_peers;
 mod runtime_metrics;
 pub mod schemata;
@@ -188,6 +189,11 @@ impl SystemSchemaProviderInner for InformationSchemaProvider {
                 self.catalog_name.clone(),
                 self.flow_metadata_manager.clone(),
             )) as _),
+            PROCEDURE_INFO => Some(
+                Arc::new(procedure_info::InformationSchemaProcedureInfo::new(
+                    self.catalog_manager.clone(),
+                )) as _,
+            ),
             _ => None,
         }
     }
@@ -250,7 +256,10 @@ impl InformationSchemaProvider {
             self.build_table(TABLE_CONSTRAINTS).unwrap(),
         );
         tables.insert(FLOWS.to_string(), self.build_table(FLOWS).unwrap());
-
+        tables.insert(
+            PROCEDURE_INFO.to_string(),
+            self.build_table(PROCEDURE_INFO).unwrap(),
+        );
         // Add memory tables
         for name in MEMORY_TABLES.iter() {
             tables.insert((*name).to_string(), self.build_table(name).expect(name));