feat: adds start_time and uptime to cluster_info

Cargo.toml

@@ -116,7 +116,7 @@ etcd-client = { git = "https://github.com/MichaelScofield/etcd-client.git", rev
 fst = "0.4.7"
 futures = "0.3"
 futures-util = "0.3"
-greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "aba235025ac5643c12bfdcefd656af11ad58ea8e" }
+greptime-proto = { git = "https://github.com/killme2008/greptime-proto.git", rev = "83991744468a010512758c2137ff3a4c9fd5eb91" }
 humantime = "2.1"
 humantime-serde = "1.1"
 itertools = "0.10"

src/catalog/Cargo.toml

@@ -31,6 +31,7 @@ datafusion.workspace = true
 datatypes.workspace = true
 futures = "0.3"
 futures-util.workspace = true
+humantime.workspace = true
 itertools.workspace = true
 lazy_static.workspace = true
 meta-client.workspace = true

src/catalog/src/information_schema/cluster_info.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use std::sync::{Arc, Weak};
+use std::time::Duration;
 
 use arrow_schema::SchemaRef as ArrowSchemaRef;
 use common_catalog::consts::INFORMATION_SCHEMA_CLUSTER_INFO_TABLE_ID;
@@ -24,13 +25,17 @@ use common_query::physical_plan::TaskContext;
 use common_recordbatch::adapter::RecordBatchStreamAdapter;
 use common_recordbatch::{RecordBatch, SendableRecordBatchStream};
 use common_telemetry::logging::warn;
+use common_time::timestamp::Timestamp;
 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, ScalarVectorBuilder, VectorRef};
 use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
+use datatypes::timestamp::TimestampMillisecond;
 use datatypes::value::Value;
-use datatypes::vectors::{StringVectorBuilder, UInt64VectorBuilder};
+use datatypes::vectors::{
+    StringVectorBuilder, TimestampMillisecondVectorBuilder, UInt64VectorBuilder,
+};
 use snafu::ResultExt;
 use store_api::storage::{ScanRequest, TableId};
 
@@ -44,7 +49,8 @@ const PEER_TYPE: &str = "peer_type";
 const PEER_ADDR: &str = "peer_addr";
 const VERSION: &str = "version";
 const GIT_COMMIT: &str = "git_commit";
-// TODO(dennis): adds `uptime`, `start_time` columns etc.
+const START_TIME: &str = "start_time";
+const UPTIME: &str = "uptime";
 
 const INIT_CAPACITY: usize = 42;
 
@@ -55,17 +61,21 @@ const INIT_CAPACITY: usize = 42;
 /// - `peer_addr`: the peer gRPC address.
 /// - `version`: the build package version of the peer.
 /// - `git_commit`: the build git commit hash of the peer.
+/// - `start_time`: the starting time of the peer.
+/// - `uptime`: the uptime of the peer.
 ///
 pub(super) struct InformationSchemaClusterInfo {
     schema: SchemaRef,
     catalog_manager: Weak<dyn CatalogManager>,
+    start_time_ms: u64,
 }
 
 impl InformationSchemaClusterInfo {
     pub(super) fn new(catalog_manager: Weak<dyn CatalogManager>) -> Self {
         Self {
             schema: Self::schema(),
             catalog_manager,
+            start_time_ms: common_time::util::current_time_millis() as u64,
         }
     }
 
@@ -76,11 +86,21 @@ impl InformationSchemaClusterInfo {
             ColumnSchema::new(PEER_ADDR, ConcreteDataType::string_datatype(), true),
             ColumnSchema::new(VERSION, ConcreteDataType::string_datatype(), false),
             ColumnSchema::new(GIT_COMMIT, ConcreteDataType::string_datatype(), false),
+            ColumnSchema::new(
+                START_TIME,
+                ConcreteDataType::timestamp_millisecond_datatype(),
+                true,
+            ),
+            ColumnSchema::new(UPTIME, ConcreteDataType::string_datatype(), true),
         ]))
     }
 
     fn builder(&self) -> InformationSchemaClusterInfoBuilder {
-        InformationSchemaClusterInfoBuilder::new(self.schema.clone(), self.catalog_manager.clone())
+        InformationSchemaClusterInfoBuilder::new(
+            self.schema.clone(),
+            self.catalog_manager.clone(),
+            self.start_time_ms,
+        )
     }
 }
 
@@ -120,17 +140,24 @@ impl InformationTable for InformationSchemaClusterInfo {
 
 struct InformationSchemaClusterInfoBuilder {
     schema: SchemaRef,
+    start_time_ms: u64,
     catalog_manager: Weak<dyn CatalogManager>,
 
     peer_ids: UInt64VectorBuilder,
     peer_types: StringVectorBuilder,
     peer_addrs: StringVectorBuilder,
     versions: StringVectorBuilder,
     git_commits: StringVectorBuilder,
+    start_times: TimestampMillisecondVectorBuilder,
+    uptimes: StringVectorBuilder,
 }
 
 impl InformationSchemaClusterInfoBuilder {
-    fn new(schema: SchemaRef, catalog_manager: Weak<dyn CatalogManager>) -> Self {
+    fn new(
+        schema: SchemaRef,
+        catalog_manager: Weak<dyn CatalogManager>,
+        start_time_ms: u64,
+    ) -> Self {
         Self {
             schema,
             catalog_manager,
@@ -139,6 +166,9 @@ impl InformationSchemaClusterInfoBuilder {
             peer_addrs: StringVectorBuilder::with_capacity(INIT_CAPACITY),
             versions: StringVectorBuilder::with_capacity(INIT_CAPACITY),
             git_commits: StringVectorBuilder::with_capacity(INIT_CAPACITY),
+            start_times: TimestampMillisecondVectorBuilder::with_capacity(INIT_CAPACITY),
+            uptimes: StringVectorBuilder::with_capacity(INIT_CAPACITY),
+            start_time_ms,
         }
     }
 
@@ -164,7 +194,10 @@ impl InformationSchemaClusterInfoBuilder {
                         last_activity_ts: -1,
                         status: NodeStatus::Standalone,
                         version: build_info.version.to_string(),
-                        git_commit: build_info.commit.to_string(),
+                        git_commit: build_info.commit_short.to_string(),
+                        // Use `self.start_time_ms` instead.
+                        // It's not precise but enough.
+                        start_time_ms: self.start_time_ms,
                     },
                 );
             }
@@ -208,6 +241,19 @@ impl InformationSchemaClusterInfoBuilder {
         self.peer_addrs.push(Some(&node_info.peer.addr));
         self.versions.push(Some(&node_info.version));
         self.git_commits.push(Some(&node_info.git_commit));
+        if node_info.start_time_ms > 0 {
+            self.start_times
+                .push(Some(TimestampMillisecond(Timestamp::new_millisecond(
+                    node_info.start_time_ms as i64,
+                ))));
+            let now = common_time::util::current_time_millis() as u64;
+            let duration_since_start = now - node_info.start_time_ms;
+            let format = humantime::format_duration(Duration::from_millis(duration_since_start));
+            self.uptimes.push(Some(format.to_string().as_str()));
+        } else {
+            self.start_times.push(None);
+            self.uptimes.push(None);
+        }
     }
 
     fn finish(&mut self) -> Result<RecordBatch> {
@@ -217,6 +263,8 @@ impl InformationSchemaClusterInfoBuilder {
             Arc::new(self.peer_addrs.finish()),
             Arc::new(self.versions.finish()),
             Arc::new(self.git_commits.finish()),
+            Arc::new(self.start_times.finish()),
+            Arc::new(self.uptimes.finish()),
         ];
         RecordBatch::new(self.schema.clone(), columns).context(CreateRecordBatchSnafu)
     }

src/catalog/src/information_schema/region_peers.rs

@@ -55,7 +55,7 @@ const INIT_CAPACITY: usize = 42;
 ///
 /// - `region_id`: the region id
 /// - `peer_id`: the region storage datanode peer id
-/// - `peer_addr`: the region storage datanode peer address
+/// - `peer_addr`: the region storage datanode gRPC peer address
 /// - `is_leader`: whether the peer is the leader
 /// - `status`: the region status, `ALIVE` or `DOWNGRADED`.
 /// - `down_seconds`: the duration of being offline, in seconds.

src/catalog/src/information_schema/runtime_metrics.rs

@@ -184,14 +184,13 @@ impl InformationSchemaMetricsBuilder {
                     .join(", "),
                 // Safety: always has a sample
                 ts.samples[0].value,
-                // TODO(dennis): fetching other peers metrics
-
                 // The peer column is always `None` for standalone
                 None,
                 "standalone",
             );
         }
 
+        // FIXME(dennis): fetching other peers metrics
         self.finish()
     }
 

src/cmd/src/frontend.rs

@@ -268,6 +268,7 @@ impl StartCommand {
         ]);
 
         let heartbeat_task = HeartbeatTask::new(
+            &opts,
             meta_client.clone(),
             opts.heartbeat.clone(),
             Arc::new(executor),

src/common/meta/src/cluster.rs

@@ -90,6 +90,8 @@ pub struct NodeInfo {
     pub version: String,
     // The node build git commit hash
     pub git_commit: String,
+    // The node star timestamp
+    pub start_time_ms: u64,
 }
 
 #[derive(Debug, Clone, Eq, Hash, PartialEq, Serialize, Deserialize)]
@@ -290,6 +292,7 @@ mod tests {
             }),
             version: "".to_string(),
             git_commit: "".to_string(),
+            start_time_ms: 1,
         };
 
         let node_info_bytes: Vec<u8> = node_info.try_into().unwrap();
@@ -306,6 +309,7 @@ mod tests {
                     leader_regions: 3,
                     follower_regions: 4,
                 }),
+                start_time_ms: 1,
                 ..
             }
         );

src/datanode/Cargo.toml

@@ -30,6 +30,7 @@ common-recordbatch.workspace = true
 common-runtime.workspace = true
 common-telemetry.workspace = true
 common-time.workspace = true
+common-version.workspace = true
 common-wal.workspace = true
 dashmap.workspace = true
 datafusion.workspace = true

src/datanode/src/heartbeat.rs

@@ -16,7 +16,7 @@ use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
 
-use api::v1::meta::{HeartbeatRequest, Peer, RegionRole, RegionStat, Role};
+use api::v1::meta::{HeartbeatRequest, NodeInfo, Peer, RegionRole, RegionStat, Role};
 use common_grpc::channel_manager::{ChannelConfig, ChannelManager};
 use common_meta::distributed_time_constants::META_KEEP_ALIVE_INTERVAL_SECS;
 use common_meta::heartbeat::handler::parse_mailbox_message::ParseMailboxMessageHandler;
@@ -43,6 +43,7 @@ use crate::region_server::RegionServer;
 pub(crate) mod handler;
 pub(crate) mod task_tracker;
 
+/// The datanode heartbeat task which sending `[HeartbeatRequest]` to Metasrv periodically in background.
 pub struct HeartbeatTask {
     node_id: u64,
     node_epoch: u64,
@@ -246,6 +247,7 @@ impl HeartbeatTask {
                         }
                     }
                     _ = &mut sleep => {
+                        let build_info = common_version::build_info();
                         let region_stats = Self::load_region_stats(&region_server_clone).await;
                         let now = Instant::now();
                         let duration_since_epoch = (now - epoch).as_millis() as u64;
@@ -254,6 +256,12 @@ impl HeartbeatTask {
                             region_stats,
                             duration_since_epoch,
                             node_epoch,
+                            info: Some(NodeInfo {
+                                version: build_info.version.to_string(),
+                                git_commit: build_info.commit_short.to_string(),
+                                // The start timestamp is the same as node_epoch currently.
+                                start_time_ms: node_epoch,
+                            }),
                             ..Default::default()
                         };
                         sleep.as_mut().reset(now + Duration::from_millis(interval));

src/frontend/Cargo.toml

@@ -33,6 +33,8 @@ common-query.workspace = true
 common-recordbatch.workspace = true
 common-runtime.workspace = true
 common-telemetry.workspace = true
+common-time.workspace = true
+common-version.workspace = true
 datanode.workspace = true
 humantime-serde.workspace = true
 lazy_static.workspace = true