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 = "973f49cde88a582fb65755cc572ebcf6fb93ccf7" }
+greptime-proto = { git = "https://github.com/realtaobo/greptime-proto.git", rev = "8671bd04ebca7a45880d10a3d4a1102f4fd5d8f1" }
 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,33 @@ pub enum Error {
         location: Location,
     },
 
+    #[snafu(display("Failed to get procedure client in {mode} mode"))]
+    GetProcedureClient {
+        mode: String,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
+    #[snafu(display("Failed to list procedures"))]
+    ListProcedures {
+        #[snafu(implicit)]
+        location: Location,
+        source: BoxedError,
+    },
+
+    #[snafu(display("Procedure id not found"))]
+    ProcedureIdNotFound {
+        #[snafu(implicit)]
+        location: Location,
+    },
+
+    #[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)]
@@ -266,7 +293,9 @@ impl ErrorExt for Error {
             | Error::FindRegionRoutes { .. }
             | Error::CacheNotFound { .. }
             | Error::CastManager { .. }
-            | Error::Json { .. } => StatusCode::Unexpected,
+            | Error::Json { .. }
+            | Error::GetProcedureClient { .. }
+            | Error::ProcedureIdNotFound { .. } => StatusCode::Unexpected,
 
             Error::ViewPlanColumnsChanged { .. } => StatusCode::InvalidArguments,
 
@@ -283,7 +312,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;
@@ -67,6 +68,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;
@@ -77,6 +79,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,
@@ -104,6 +107,7 @@ impl KvBackendCatalogManager {
                 backend,
             },
             cache_registry,
+            procedure_manager,
         })
     }
 
@@ -130,6 +134,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));