frontend/
server.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::net::SocketAddr;
16use std::sync::Arc;
17
18use auth::UserProviderRef;
19use common_base::Plugins;
20use common_config::Configurable;
21use servers::error::Error as ServerError;
22use servers::grpc::builder::GrpcServerBuilder;
23use servers::grpc::greptime_handler::GreptimeRequestHandler;
24use servers::grpc::{GrpcOptions, GrpcServer, GrpcServerConfig};
25use servers::http::event::LogValidatorRef;
26use servers::http::{HttpServer, HttpServerBuilder};
27use servers::interceptor::LogIngestInterceptorRef;
28use servers::metrics_handler::MetricsHandler;
29use servers::mysql::server::{MysqlServer, MysqlSpawnConfig, MysqlSpawnRef};
30use servers::otel_arrow::OtelArrowServiceHandler;
31use servers::postgres::PostgresServer;
32use servers::query_handler::grpc::ServerGrpcQueryHandlerAdapter;
33use servers::query_handler::sql::ServerSqlQueryHandlerAdapter;
34use servers::server::{Server, ServerHandlers};
35use servers::tls::{maybe_watch_tls_config, ReloadableTlsServerConfig};
36use snafu::ResultExt;
37
38use crate::error::{self, Result, StartServerSnafu, TomlFormatSnafu};
39use crate::frontend::FrontendOptions;
40use crate::instance::Instance;
41
42pub struct Services<T>
43where
44    T: Into<FrontendOptions> + Configurable + Clone,
45{
46    opts: T,
47    instance: Arc<Instance>,
48    grpc_server_builder: Option<GrpcServerBuilder>,
49    http_server_builder: Option<HttpServerBuilder>,
50    plugins: Plugins,
51}
52
53impl<T> Services<T>
54where
55    T: Into<FrontendOptions> + Configurable + Clone,
56{
57    pub fn new(opts: T, instance: Arc<Instance>, plugins: Plugins) -> Self {
58        Self {
59            opts,
60            instance,
61            grpc_server_builder: None,
62            http_server_builder: None,
63            plugins,
64        }
65    }
66
67    pub fn grpc_server_builder(&self, opts: &GrpcOptions) -> Result<GrpcServerBuilder> {
68        let grpc_config = GrpcServerConfig {
69            max_recv_message_size: opts.max_recv_message_size.as_bytes() as usize,
70            max_send_message_size: opts.max_send_message_size.as_bytes() as usize,
71            tls: opts.tls.clone(),
72        };
73        let builder = GrpcServerBuilder::new(grpc_config, common_runtime::global_runtime())
74            .with_tls_config(opts.tls.clone())
75            .context(error::InvalidTlsConfigSnafu)?;
76        Ok(builder)
77    }
78
79    pub fn http_server_builder(&self, opts: &FrontendOptions) -> HttpServerBuilder {
80        let mut builder = HttpServerBuilder::new(opts.http.clone())
81            .with_sql_handler(ServerSqlQueryHandlerAdapter::arc(self.instance.clone()));
82
83        let validator = self.plugins.get::<LogValidatorRef>();
84        let ingest_interceptor = self.plugins.get::<LogIngestInterceptorRef<ServerError>>();
85        builder =
86            builder.with_log_ingest_handler(self.instance.clone(), validator, ingest_interceptor);
87        builder = builder.with_logs_handler(self.instance.clone());
88
89        if let Some(user_provider) = self.plugins.get::<UserProviderRef>() {
90            builder = builder.with_user_provider(user_provider);
91        }
92
93        if opts.opentsdb.enable {
94            builder = builder.with_opentsdb_handler(self.instance.clone());
95        }
96
97        if opts.influxdb.enable {
98            builder = builder.with_influxdb_handler(self.instance.clone());
99        }
100
101        if opts.prom_store.enable {
102            builder = builder
103                .with_prom_handler(
104                    self.instance.clone(),
105                    Some(self.instance.clone()),
106                    opts.prom_store.with_metric_engine,
107                    opts.http.is_strict_mode,
108                )
109                .with_prometheus_handler(self.instance.clone());
110        }
111
112        if opts.otlp.enable {
113            builder = builder.with_otlp_handler(self.instance.clone());
114        }
115
116        if opts.jaeger.enable {
117            builder = builder.with_jaeger_handler(self.instance.clone());
118        }
119
120        builder
121    }
122
123    pub fn with_grpc_server_builder(self, builder: GrpcServerBuilder) -> Self {
124        Self {
125            grpc_server_builder: Some(builder),
126            ..self
127        }
128    }
129
130    pub fn with_http_server_builder(self, builder: HttpServerBuilder) -> Self {
131        Self {
132            http_server_builder: Some(builder),
133            ..self
134        }
135    }
136
137    fn build_grpc_server(&mut self, opts: &FrontendOptions) -> Result<GrpcServer> {
138        let builder = if let Some(builder) = self.grpc_server_builder.take() {
139            builder
140        } else {
141            self.grpc_server_builder(&opts.grpc)?
142        };
143
144        let user_provider = self.plugins.get::<UserProviderRef>();
145
146        // Determine whether it is Standalone or Distributed mode based on whether the meta client is configured.
147        let runtime = if opts.meta_client.is_none() {
148            Some(builder.runtime().clone())
149        } else {
150            None
151        };
152
153        let greptime_request_handler = GreptimeRequestHandler::new(
154            ServerGrpcQueryHandlerAdapter::arc(self.instance.clone()),
155            user_provider.clone(),
156            runtime,
157        );
158
159        let grpc_server = builder
160            .database_handler(greptime_request_handler.clone())
161            .prometheus_handler(self.instance.clone(), user_provider.clone())
162            .otel_arrow_handler(OtelArrowServiceHandler(self.instance.clone()))
163            .flight_handler(Arc::new(greptime_request_handler))
164            .build();
165        Ok(grpc_server)
166    }
167
168    fn build_http_server(&mut self, opts: &FrontendOptions, toml: String) -> Result<HttpServer> {
169        let builder = if let Some(builder) = self.http_server_builder.take() {
170            builder
171        } else {
172            self.http_server_builder(opts)
173        };
174
175        let http_server = builder
176            .with_metrics_handler(MetricsHandler)
177            .with_plugins(self.plugins.clone())
178            .with_greptime_config_options(toml)
179            .build();
180        Ok(http_server)
181    }
182
183    pub fn build(mut self) -> Result<ServerHandlers> {
184        let opts = self.opts.clone();
185        let instance = self.instance.clone();
186
187        let toml = opts.to_toml().context(TomlFormatSnafu)?;
188        let opts: FrontendOptions = opts.into();
189
190        let handlers = ServerHandlers::default();
191
192        let user_provider = self.plugins.get::<UserProviderRef>();
193
194        {
195            // Always init GRPC server
196            let grpc_addr = parse_addr(&opts.grpc.bind_addr)?;
197            let grpc_server = self.build_grpc_server(&opts)?;
198            handlers.insert((Box::new(grpc_server), grpc_addr));
199        }
200
201        {
202            // Always init HTTP server
203            let http_options = &opts.http;
204            let http_addr = parse_addr(&http_options.addr)?;
205            let http_server = self.build_http_server(&opts, toml)?;
206            handlers.insert((Box::new(http_server), http_addr));
207        }
208
209        if opts.mysql.enable {
210            // Init MySQL server
211            let opts = &opts.mysql;
212            let mysql_addr = parse_addr(&opts.addr)?;
213
214            let tls_server_config = Arc::new(
215                ReloadableTlsServerConfig::try_new(opts.tls.clone()).context(StartServerSnafu)?,
216            );
217
218            // will not watch if watch is disabled in tls option
219            maybe_watch_tls_config(tls_server_config.clone()).context(StartServerSnafu)?;
220
221            let mysql_server = MysqlServer::create_server(
222                common_runtime::global_runtime(),
223                Arc::new(MysqlSpawnRef::new(
224                    ServerSqlQueryHandlerAdapter::arc(instance.clone()),
225                    user_provider.clone(),
226                )),
227                Arc::new(MysqlSpawnConfig::new(
228                    opts.tls.should_force_tls(),
229                    tls_server_config,
230                    opts.keep_alive.as_secs(),
231                    opts.reject_no_database.unwrap_or(false),
232                )),
233            );
234            handlers.insert((mysql_server, mysql_addr));
235        }
236
237        if opts.postgres.enable {
238            // Init PosgresSQL Server
239            let opts = &opts.postgres;
240            let pg_addr = parse_addr(&opts.addr)?;
241
242            let tls_server_config = Arc::new(
243                ReloadableTlsServerConfig::try_new(opts.tls.clone()).context(StartServerSnafu)?,
244            );
245
246            maybe_watch_tls_config(tls_server_config.clone()).context(StartServerSnafu)?;
247
248            let pg_server = Box::new(PostgresServer::new(
249                ServerSqlQueryHandlerAdapter::arc(instance.clone()),
250                opts.tls.should_force_tls(),
251                tls_server_config,
252                opts.keep_alive.as_secs(),
253                common_runtime::global_runtime(),
254                user_provider.clone(),
255            )) as Box<dyn Server>;
256
257            handlers.insert((pg_server, pg_addr));
258        }
259
260        Ok(handlers)
261    }
262}
263
264fn parse_addr(addr: &str) -> Result<SocketAddr> {
265    addr.parse().context(error::ParseAddrSnafu { addr })
266}
frontend/server.rs

frontend/
server.rs
