cmd/datanode/
builder.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 cache::build_datanode_cache_registry;
18use catalog::kvbackend::MetaKvBackend;
19use common_base::Plugins;
20use common_meta::cache::LayeredCacheRegistryBuilder;
21use common_telemetry::info;
22use common_version::{short_version, verbose_version};
23use datanode::datanode::DatanodeBuilder;
24use datanode::service::DatanodeServiceBuilder;
25use meta_client::MetaClientType;
26use snafu::{OptionExt, ResultExt};
27use tracing_appender::non_blocking::WorkerGuard;
28
29use crate::datanode::{DatanodeOptions, Instance, APP_NAME};
30use crate::error::{MetaClientInitSnafu, MissingConfigSnafu, Result, StartDatanodeSnafu};
31use crate::{create_resource_limit_metrics, log_versions, maybe_activate_heap_profile};
32
33/// Builder for Datanode instance.
34pub struct InstanceBuilder {
35    guard: Vec<WorkerGuard>,
36    opts: DatanodeOptions,
37    datanode_builder: DatanodeBuilder,
38}
39
40impl InstanceBuilder {
41    /// Try to create a new [InstanceBuilder], and do some initialization work like allocating
42    /// runtime resources, setting up global logging and plugins, etc.
43    pub async fn try_new_with_init(
44        mut opts: DatanodeOptions,
45        mut plugins: Plugins,
46    ) -> Result<Self> {
47        let guard = Self::init(&mut opts, &mut plugins).await?;
48
49        let datanode_builder = Self::datanode_builder(&opts, plugins).await?;
50
51        Ok(Self {
52            guard,
53            opts,
54            datanode_builder,
55        })
56    }
57
58    async fn init(opts: &mut DatanodeOptions, plugins: &mut Plugins) -> Result<Vec<WorkerGuard>> {
59        common_runtime::init_global_runtimes(&opts.runtime);
60
61        let dn_opts = &mut opts.component;
62        let guard = common_telemetry::init_global_logging(
63            APP_NAME,
64            &dn_opts.logging,
65            &dn_opts.tracing,
66            dn_opts.node_id.map(|x| x.to_string()),
67            None,
68        );
69
70        log_versions(verbose_version(), short_version(), APP_NAME);
71        maybe_activate_heap_profile(&dn_opts.memory);
72        create_resource_limit_metrics(APP_NAME);
73
74        plugins::setup_datanode_plugins(plugins, &opts.plugins, dn_opts)
75            .await
76            .context(StartDatanodeSnafu)?;
77
78        dn_opts.grpc.detect_server_addr();
79
80        info!("Initialized Datanode instance with {:#?}", opts);
81        Ok(guard)
82    }
83
84    async fn datanode_builder(opts: &DatanodeOptions, plugins: Plugins) -> Result<DatanodeBuilder> {
85        let dn_opts = &opts.component;
86
87        let member_id = dn_opts
88            .node_id
89            .context(MissingConfigSnafu { msg: "'node_id'" })?;
90        let meta_client_options = dn_opts.meta_client.as_ref().context(MissingConfigSnafu {
91            msg: "meta client options",
92        })?;
93        let client = meta_client::create_meta_client(
94            MetaClientType::Datanode { member_id },
95            meta_client_options,
96            Some(&plugins),
97            None,
98        )
99        .await
100        .context(MetaClientInitSnafu)?;
101
102        let backend = Arc::new(MetaKvBackend {
103            client: client.clone(),
104        });
105        let mut builder = DatanodeBuilder::new(dn_opts.clone(), plugins.clone(), backend.clone());
106
107        let registry = Arc::new(
108            LayeredCacheRegistryBuilder::default()
109                .add_cache_registry(build_datanode_cache_registry(backend))
110                .build(),
111        );
112        builder
113            .with_cache_registry(registry)
114            .with_meta_client(client.clone());
115        Ok(builder)
116    }
117
118    /// Get the mutable builder for Datanode, in case you want to change some fields before the
119    /// final construction.
120    pub fn mut_datanode_builder(&mut self) -> &mut DatanodeBuilder {
121        &mut self.datanode_builder
122    }
123
124    /// Try to build the Datanode instance.
125    pub async fn build(self) -> Result<Instance> {
126        let mut datanode = self
127            .datanode_builder
128            .build()
129            .await
130            .context(StartDatanodeSnafu)?;
131
132        let services = DatanodeServiceBuilder::new(&self.opts.component)
133            .with_default_grpc_server(&datanode.region_server())
134            .enable_http_service()
135            .build()
136            .context(StartDatanodeSnafu)?;
137        datanode.setup_services(services);
138
139        Ok(Instance::new(datanode, self.guard))
140    }
141}
cmd/datanode/builder.rs

cmd/datanode/
builder.rs
