standalone/
information_extension.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::sync::Arc;
16
17use catalog::information_schema::{DatanodeInspectRequest, InformationExtension};
18use client::SendableRecordBatchStream;
19use client::api::v1::meta::RegionRole;
20use common_error::ext::BoxedError;
21use common_meta::cluster::{NodeInfo, NodeStatus};
22use common_meta::datanode::RegionStat;
23use common_meta::key::flow::flow_state::FlowStat;
24use common_meta::peer::Peer;
25use common_procedure::{ProcedureInfo, ProcedureManagerRef};
26use common_query::request::QueryRequest;
27use common_stat::{ResourceStatImpl, ResourceStatRef};
28use datanode::region_server::RegionServer;
29use flow::FlowDualEngineRef;
30use snafu::ResultExt;
31use store_api::storage::RegionId;
32use tokio::sync::RwLock;
33
34pub struct StandaloneInformationExtension {
35    region_server: RegionServer,
36    procedure_manager: ProcedureManagerRef,
37    start_time_ms: u64,
38    flow_engine: RwLock<Option<FlowDualEngineRef>>,
39    resource_stat: ResourceStatRef,
40}
41
42impl StandaloneInformationExtension {
43    pub fn new(region_server: RegionServer, procedure_manager: ProcedureManagerRef) -> Self {
44        let mut resource_stat = ResourceStatImpl::default();
45        resource_stat.start_collect_cpu_usage();
46        Self {
47            region_server,
48            procedure_manager,
49            start_time_ms: common_time::util::current_time_millis() as u64,
50            flow_engine: RwLock::new(None),
51            resource_stat: Arc::new(resource_stat),
52        }
53    }
54
55    /// Set the flow engine for the standalone instance.
56    pub async fn set_flow_engine(&self, flow_engine: FlowDualEngineRef) {
57        let mut guard = self.flow_engine.write().await;
58        *guard = Some(flow_engine);
59    }
60}
61
62#[async_trait::async_trait]
63impl InformationExtension for StandaloneInformationExtension {
64    type Error = catalog::error::Error;
65
66    async fn nodes(&self) -> std::result::Result<Vec<NodeInfo>, Self::Error> {
67        let build_info = common_version::build_info();
68        let node_info = NodeInfo {
69            // For the standalone:
70            // - id always 0
71            // - empty string for peer_addr
72            peer: Peer {
73                id: 0,
74                addr: "".to_string(),
75            },
76            last_activity_ts: -1,
77            status: NodeStatus::Standalone,
78            version: build_info.version.to_string(),
79            git_commit: build_info.commit_short.to_string(),
80            // Use `self.start_time_ms` instead.
81            // It's not precise but enough.
82            start_time_ms: self.start_time_ms,
83            total_cpu_millicores: self.resource_stat.get_total_cpu_millicores(),
84            total_memory_bytes: self.resource_stat.get_total_memory_bytes(),
85            cpu_usage_millicores: self.resource_stat.get_cpu_usage_millicores(),
86            memory_usage_bytes: self.resource_stat.get_memory_usage_bytes(),
87            hostname: hostname::get()
88                .unwrap_or_default()
89                .to_string_lossy()
90                .to_string(),
91            env_vars: Default::default(),
92            schedulable: None,
93        };
94        Ok(vec![node_info])
95    }
96
97    async fn procedures(&self) -> std::result::Result<Vec<(String, ProcedureInfo)>, Self::Error> {
98        self.procedure_manager
99            .list_procedures()
100            .await
101            .map_err(BoxedError::new)
102            .map(|procedures| {
103                procedures
104                    .into_iter()
105                    .map(|procedure| {
106                        let status = procedure.state.as_str_name().to_string();
107                        (status, procedure)
108                    })
109                    .collect::<Vec<_>>()
110            })
111            .context(catalog::error::ListProceduresSnafu)
112    }
113
114    async fn region_stats(&self) -> std::result::Result<Vec<RegionStat>, Self::Error> {
115        let stats = self
116            .region_server
117            .reportable_regions()
118            .into_iter()
119            .map(|stat| {
120                let region_stat = self
121                    .region_server
122                    .region_statistic(stat.region_id)
123                    .unwrap_or_default();
124                RegionStat {
125                    id: stat.region_id,
126                    rcus: 0,
127                    wcus: 0,
128                    approximate_bytes: region_stat.estimated_disk_size(),
129                    engine: stat.engine,
130                    role: RegionRole::from(stat.role).into(),
131                    num_rows: region_stat.num_rows,
132                    memtable_size: region_stat.memtable_size,
133                    manifest_size: region_stat.manifest_size,
134                    sst_size: region_stat.sst_size,
135                    sst_num: region_stat.sst_num,
136                    index_size: region_stat.index_size,
137                    region_manifest: region_stat.manifest.into(),
138                    data_topic_latest_entry_id: region_stat.data_topic_latest_entry_id,
139                    metadata_topic_latest_entry_id: region_stat.metadata_topic_latest_entry_id,
140                    written_bytes: region_stat.written_bytes,
141                    ingested_parquet_bytes: region_stat.ingested_parquet_bytes,
142                    query_cpu_time: region_stat.query_cpu_time,
143                    query_scanned_bytes: region_stat.query_scanned_bytes,
144                    min_timestamp: region_stat.min_timestamp,
145                    max_timestamp: region_stat.max_timestamp,
146                }
147            })
148            .collect::<Vec<_>>();
149        Ok(stats)
150    }
151
152    async fn flow_stats(&self) -> std::result::Result<Option<FlowStat>, Self::Error> {
153        Ok(Some(
154            self.flow_engine
155                .read()
156                .await
157                .as_ref()
158                .unwrap()
159                .gen_state_report()
160                .await,
161        ))
162    }
163
164    async fn inspect_datanode(
165        &self,
166        request: DatanodeInspectRequest,
167    ) -> std::result::Result<SendableRecordBatchStream, Self::Error> {
168        let req = QueryRequest {
169            plan: request
170                .build_plan()
171                .context(catalog::error::DatafusionSnafu)?,
172            region_id: RegionId::default(),
173            header: None,
174        };
175
176        self.region_server
177            .handle_read(req)
178            .await
179            .map_err(BoxedError::new)
180            .context(catalog::error::InternalSnafu)
181    }
182}
standalone/information_extension.rs

standalone/
information_extension.rs
