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

cmd/datanode/
builder.rs
