1use std::net::SocketAddr;
16use std::sync::Arc;
17use std::time::Duration;
18
19use auth::UserProviderRef;
20use axum::extract::{Request, State};
21use axum::middleware::Next;
22use axum::response::IntoResponse;
23use common_base::Plugins;
24use common_config::Configurable;
25use common_telemetry::{info, warn};
26use meta_client::MetaClientOptions;
27use servers::error::Error as ServerError;
28use servers::grpc::builder::GrpcServerBuilder;
29use servers::grpc::flight::FlightCraftRef;
30use servers::grpc::frontend_grpc_handler::FrontendGrpcHandler;
31use servers::grpc::greptime_handler::GreptimeRequestHandler;
32use servers::grpc::{GrpcOptions, GrpcServer};
33use servers::http::event::LogValidatorRef;
34use servers::http::result::error_result::ErrorResponse;
35use servers::http::utils::router::RouterConfigurator;
36use servers::http::{HttpOptions, HttpServer, HttpServerBuilder};
37use servers::interceptor::LogIngestInterceptorRef;
38use servers::metrics_handler::MetricsHandler;
39use servers::mysql::server::{MysqlServer, MysqlSpawnConfig, MysqlSpawnRef};
40use servers::otel_arrow::OtelArrowServiceHandler;
41use servers::pending_rows_batcher::{PendingRowsBatcher, pending_rows_batch_sync_enabled};
42use servers::postgres::PostgresServer;
43use servers::request_memory_limiter::ServerMemoryLimiter;
44use servers::server::{Server, ServerHandlers};
45use servers::tls::{ReloadableTlsServerConfig, maybe_watch_server_tls_config};
46use snafu::ResultExt;
47use tonic::Status;
48
49use crate::error::{self, Result, StartServerSnafu, TomlFormatSnafu};
50use crate::frontend::FrontendOptions;
51use crate::instance::Instance;
52
53pub struct Services<T>
54where
55 T: Into<FrontendOptions> + Configurable + Clone,
56{
57 opts: T,
58 instance: Arc<Instance>,
59 grpc_server_builder: Option<GrpcServerBuilder>,
60 http_server_builder: Option<HttpServerBuilder>,
61 plugins: Plugins,
62 flight_handler: Option<FlightCraftRef>,
63 internal_flight_handler: Option<FlightCraftRef>,
64 pub server_memory_limiter: ServerMemoryLimiter,
65}
66
67impl<T> Services<T>
68where
69 T: Into<FrontendOptions> + Configurable + Clone,
70{
71 pub fn new(opts: T, instance: Arc<Instance>, plugins: Plugins) -> Self {
72 let feopts = opts.clone().into();
73 let server_memory_limiter = ServerMemoryLimiter::new(
75 feopts.max_in_flight_write_bytes.as_bytes(),
76 feopts.write_bytes_exhausted_policy,
77 );
78
79 Self {
80 opts,
81 instance,
82 grpc_server_builder: None,
83 http_server_builder: None,
84 plugins,
85 flight_handler: None,
86 internal_flight_handler: None,
87 server_memory_limiter,
88 }
89 }
90
91 pub fn grpc_server_builder(
92 &self,
93 opts: &GrpcOptions,
94 request_memory_limiter: ServerMemoryLimiter,
95 ) -> Result<GrpcServerBuilder> {
96 let builder = GrpcServerBuilder::new(opts.as_config(), common_runtime::global_runtime())
97 .with_memory_limiter(request_memory_limiter)
98 .with_tls_config(opts.tls.clone())
99 .context(error::InvalidTlsConfigSnafu)?;
100 Ok(builder)
101 }
102
103 pub fn http_server_builder(
104 &self,
105 opts: &FrontendOptions,
106 request_memory_limiter: ServerMemoryLimiter,
107 ) -> HttpServerBuilder {
108 let mut builder = HttpServerBuilder::new(effective_http_options(opts))
109 .with_memory_limiter(request_memory_limiter)
110 .with_sql_handler(self.instance.clone());
111
112 let validator = self.plugins.get::<LogValidatorRef>();
113 let ingest_interceptor = self.plugins.get::<LogIngestInterceptorRef<ServerError>>();
114 builder =
115 builder.with_log_ingest_handler(self.instance.clone(), validator, ingest_interceptor);
116 builder = builder.with_logs_handler(self.instance.clone());
117
118 if let Some(user_provider) = self.plugins.get::<UserProviderRef>() {
119 builder = builder.with_user_provider(user_provider);
120 }
121
122 if opts.opentsdb.enable {
123 builder = builder.with_opentsdb_handler(self.instance.clone());
124 }
125
126 if opts.influxdb.enable {
127 builder = builder.with_influxdb_handler(self.instance.clone());
128 }
129
130 if opts.prom_store.enable {
131 let pending_rows_batcher = if opts.prom_store.with_metric_engine {
132 PendingRowsBatcher::try_new(
133 self.instance.partition_manager().clone(),
134 self.instance.node_manager().clone(),
135 self.instance.catalog_manager().clone(),
136 self.instance.table_flownode_set_cache().clone(),
137 opts.prom_store.with_metric_engine,
138 self.instance.clone(),
139 opts.prom_store.pending_rows_flush_interval,
140 opts.prom_store.max_batch_rows,
141 opts.prom_store.max_concurrent_flushes,
142 opts.prom_store.worker_channel_capacity,
143 opts.prom_store.max_inflight_requests,
144 opts.prom_store.flow_notification_queue_capacity,
145 )
146 } else {
147 None
148 };
149 builder = builder
150 .with_prom_handler(
151 self.instance.clone(),
152 Some(self.instance.clone()),
153 opts.prom_store.with_metric_engine,
154 opts.prom_store.prom_validation_mode,
155 opts.prom_store
156 .experimental_enable_prometheus_native_histogram,
157 pending_rows_batcher,
158 )
159 .with_prometheus_handler(self.instance.clone());
160 }
161
162 if opts.otlp.enable {
163 builder = builder.with_otlp_handler(
164 self.instance.clone(),
165 opts.prom_store.with_metric_engine,
166 opts.otlp.experimental_enable_exponential_histogram,
167 );
168 }
169
170 if opts.jaeger.enable {
171 builder = builder.with_jaeger_handler(self.instance.clone());
172 }
173
174 builder = builder.with_dashboard_handler(self.instance.clone());
175
176 if let Some(configurator) = self.plugins.get::<RouterConfigurator>() {
177 info!("Adding extra router from plugins");
178 builder = builder.with_extra_router(configurator.router());
179 }
180
181 builder.add_layer(axum::middleware::from_fn_with_state(
182 self.instance.clone(),
183 async move |State(state): State<Arc<Instance>>, request: Request, next: Next| {
184 if state.is_suspended() {
185 return ErrorResponse::from_error(servers::error::SuspendedSnafu.build())
186 .into_response();
187 }
188 next.run(request).await
189 },
190 ))
191 }
192
193 pub fn with_grpc_server_builder(self, builder: GrpcServerBuilder) -> Self {
194 Self {
195 grpc_server_builder: Some(builder),
196 ..self
197 }
198 }
199
200 pub fn with_http_server_builder(self, builder: HttpServerBuilder) -> Self {
201 Self {
202 http_server_builder: Some(builder),
203 ..self
204 }
205 }
206
207 pub fn with_flight_handler(self, flight_handler: FlightCraftRef) -> Self {
208 Self {
209 flight_handler: Some(flight_handler),
210 ..self
211 }
212 }
213
214 pub fn with_internal_flight_handler(self, flight_handler: FlightCraftRef) -> Self {
215 Self {
216 internal_flight_handler: Some(flight_handler),
217 ..self
218 }
219 }
220
221 fn build_grpc_server(
222 &mut self,
223 grpc: &GrpcOptions,
224 meta_client: &Option<MetaClientOptions>,
225 name: Option<String>,
226 external: bool,
227 request_memory_limiter: ServerMemoryLimiter,
228 ) -> Result<GrpcServer> {
229 let builder = if let Some(builder) = self.grpc_server_builder.take() {
230 builder
231 } else {
232 self.grpc_server_builder(grpc, request_memory_limiter)?
233 };
234
235 let user_provider = if external {
236 self.plugins.get::<UserProviderRef>()
237 } else {
238 None
240 };
241
242 let runtime = if meta_client.is_none() {
244 Some(builder.runtime().clone())
245 } else {
246 None
247 };
248
249 let greptime_request_handler = GreptimeRequestHandler::new(
250 self.instance.clone(),
251 user_provider.clone(),
252 runtime,
253 grpc.flight_compression,
254 );
255
256 let default_flight_handler = Arc::new(greptime_request_handler.clone()) as FlightCraftRef;
257 let flight_handler = if external {
258 self.flight_handler
259 .clone()
260 .unwrap_or(default_flight_handler)
261 } else {
262 self.internal_flight_handler
263 .clone()
264 .unwrap_or(default_flight_handler)
265 };
266
267 let grpc_server = builder
268 .name(name)
269 .database_handler(greptime_request_handler.clone())
270 .prometheus_handler(self.instance.clone(), user_provider.clone())
271 .otel_arrow_handler(OtelArrowServiceHandler::new(
272 self.instance.clone(),
273 user_provider.clone(),
274 ))
275 .flight_handler(flight_handler)
276 .add_layer(axum::middleware::from_fn_with_state(
277 self.instance.clone(),
278 async move |State(state): State<Arc<Instance>>, request: Request, next: Next| {
279 if state.is_suspended() {
280 let status = Status::from(servers::error::SuspendedSnafu.build());
281 return status.into_http();
282 }
283 next.run(request).await
284 },
285 ));
286
287 let grpc_server = if !external {
288 let frontend_grpc_handler =
289 FrontendGrpcHandler::new(self.instance.process_manager().clone());
290 grpc_server.frontend_grpc_handler(frontend_grpc_handler)
291 } else {
292 grpc_server
293 }
294 .build();
295
296 Ok(grpc_server)
297 }
298
299 fn build_http_server(
300 &mut self,
301 opts: &FrontendOptions,
302 toml: String,
303 request_memory_limiter: ServerMemoryLimiter,
304 ) -> Result<(HttpServer, Option<HttpServer>)> {
305 let builder = if let Some(builder) = self.http_server_builder.take() {
306 builder
307 } else {
308 self.http_server_builder(opts, request_memory_limiter)
309 };
310
311 let (internal, api) = builder
315 .with_metrics_handler(MetricsHandler)
316 .with_greptime_config_options(toml)
317 .build_servers();
318 Ok((internal, api))
319 }
320
321 pub fn build(mut self) -> Result<ServerHandlers> {
322 let opts = self.opts.clone();
323 let instance = self.instance.clone();
324
325 let toml = opts.to_toml().context(TomlFormatSnafu)?;
326 let opts: FrontendOptions = opts.into();
327
328 let handlers = ServerHandlers::default();
329
330 let user_provider = self.plugins.get::<UserProviderRef>();
331
332 {
333 let grpc_addr = parse_addr(&opts.grpc.bind_addr)?;
335 let grpc_server = self.build_grpc_server(
336 &opts.grpc,
337 &opts.meta_client,
338 None,
339 true,
340 self.server_memory_limiter.clone(),
341 )?;
342 handlers.insert((Box::new(grpc_server), grpc_addr));
343 }
344
345 if let Some(internal_grpc) = &opts.internal_grpc {
346 let grpc_addr = parse_addr(&internal_grpc.bind_addr)?;
348 let grpc_server = self.build_grpc_server(
349 internal_grpc,
350 &opts.meta_client,
351 Some("INTERNAL_GRPC_SERVER".to_string()),
352 false,
353 self.server_memory_limiter.clone(),
354 )?;
355 handlers.insert((Box::new(grpc_server), grpc_addr));
356 }
357
358 {
359 let http_options = &opts.http;
362 let http_addr = parse_addr(&http_options.addr)?;
363 let (http_server, http_api_server) =
364 self.build_http_server(&opts, toml, self.server_memory_limiter.clone())?;
365 handlers.insert((Box::new(http_server), http_addr));
366
367 if let Some(http_api_server) = http_api_server {
368 let http_api_addr = parse_addr(&http_options.api_server_addr)?;
369 info!("HTTP API server is enabled at {}", http_api_addr);
370 handlers.insert((Box::new(http_api_server), http_api_addr));
371 }
372 }
373
374 if opts.mysql.enable {
375 let opts = &opts.mysql;
377 let mysql_addr = parse_addr(&opts.addr)?;
378
379 let tls_server_config = Arc::new(
380 ReloadableTlsServerConfig::try_new(opts.tls.clone()).context(StartServerSnafu)?,
381 );
382
383 maybe_watch_server_tls_config(tls_server_config.clone()).context(StartServerSnafu)?;
385
386 let mysql_server = MysqlServer::create_server(
387 common_runtime::global_runtime(),
388 Arc::new(MysqlSpawnRef::new(instance.clone(), user_provider.clone())),
389 Arc::new(MysqlSpawnConfig::new(
390 opts.tls.should_force_tls(),
391 tls_server_config,
392 opts.keep_alive.as_secs(),
393 opts.reject_no_database.unwrap_or(false),
394 opts.prepared_stmt_cache_size,
395 )),
396 Some(instance.process_manager().clone()),
397 );
398 handlers.insert((mysql_server, mysql_addr));
399 }
400
401 if opts.postgres.enable {
402 let opts = &opts.postgres;
404 let pg_addr = parse_addr(&opts.addr)?;
405
406 let tls_server_config = Arc::new(
407 ReloadableTlsServerConfig::try_new(opts.tls.clone()).context(StartServerSnafu)?,
408 );
409
410 maybe_watch_server_tls_config(tls_server_config.clone()).context(StartServerSnafu)?;
411
412 let pg_server = Box::new(PostgresServer::new(
413 instance.clone(),
414 opts.tls.should_force_tls(),
415 tls_server_config,
416 opts.keep_alive.as_secs(),
417 common_runtime::global_runtime(),
418 user_provider.clone(),
419 Some(self.instance.process_manager().clone()),
420 )) as Box<dyn Server>;
421
422 handlers.insert((pg_server, pg_addr));
423 }
424
425 Ok(handlers)
426 }
427}
428
429fn effective_http_options(opts: &FrontendOptions) -> HttpOptions {
430 effective_http_options_with_sync(opts, pending_rows_batch_sync_enabled())
431}
432
433fn effective_http_options_with_sync(opts: &FrontendOptions, batch_sync: bool) -> HttpOptions {
434 let mut http = opts.http.clone();
435 let flush_interval = opts.prom_store.pending_rows_flush_interval;
436 let fallback_timeout = flush_interval.saturating_add(Duration::from_secs(1));
437 if !opts.prom_store.pending_rows_batching_enabled()
441 || !batch_sync
442 || http.timeout.is_zero()
443 || http.timeout > fallback_timeout
444 {
445 return http;
446 }
447
448 let configured_timeout = http.timeout;
449 http.timeout = fallback_timeout;
450 warn!(
451 ?configured_timeout,
452 ?flush_interval,
453 ?fallback_timeout,
454 "HTTP request timeout is not longer than the pending-row timeout fallback; using the fallback"
455 );
456 http
457}
458
459fn parse_addr(addr: &str) -> Result<SocketAddr> {
460 addr.parse().context(error::ParseAddrSnafu { addr })
461}
462
463#[cfg(test)]
464mod tests {
465 use std::sync::Arc;
466 use std::sync::atomic::{AtomicUsize, Ordering};
467 use std::time::Duration;
468
469 use api::v1::HealthCheckRequest;
470 use api::v1::health_check_client::HealthCheckClient;
471 use api::v1::meta::Role;
472 use arrow_flight::{FlightData, PutResult, Ticket};
473 use async_trait::async_trait;
474 use auth::{UserProviderRef, static_user_provider_from_option};
475 use client::{Client, Database};
476 use meta_client::client::MetaClientBuilder;
477 use servers::grpc::GRPC_SERVER;
478 use servers::grpc::flight::{FlightCraft, FlightCraftRef, TonicStream};
479 use tonic::{Code, Request, Response, Status, Streaming};
480
481 use super::*;
482 use crate::instance::builder::FrontendBuilder;
483
484 struct CountingFlightCraft {
485 inner: FlightCraftRef,
486 do_get_calls: AtomicUsize,
487 do_put_calls: AtomicUsize,
488 }
489
490 #[async_trait]
491 impl FlightCraft for CountingFlightCraft {
492 async fn do_get(
493 &self,
494 request: Request<Ticket>,
495 ) -> std::result::Result<Response<TonicStream<FlightData>>, Status> {
496 self.do_get_calls.fetch_add(1, Ordering::SeqCst);
497 self.inner.do_get(request).await
498 }
499
500 async fn do_put(
501 &self,
502 request: Request<Streaming<FlightData>>,
503 ) -> std::result::Result<Response<TonicStream<PutResult>>, Status> {
504 self.do_put_calls.fetch_add(1, Ordering::SeqCst);
505 self.inner.do_put(request).await
506 }
507 }
508
509 #[test]
510 fn test_effective_http_timeout_for_pending_rows() {
511 let cases = [
512 ("disabled timeout", 0, 5000, true, true, 0),
513 ("disabled prom store", 1000, 5000, false, true, 1000),
514 ("disabled metric engine", 1000, 5000, true, false, 1000),
515 ("disabled batching", 1000, 0, true, true, 1000),
516 ("timeout below flush interval", 4000, 5000, true, true, 6000),
517 (
518 "timeout equals flush interval",
519 5000,
520 5000,
521 true,
522 true,
523 6000,
524 ),
525 ("timeout below fallback", 5500, 5000, true, true, 6000),
526 ("timeout equals fallback", 6000, 5000, true, true, 6000),
527 ("timeout above fallback", 7000, 5000, true, true, 7000),
528 ];
529
530 for (name, timeout, flush_interval, enable, with_metric_engine, expected) in cases {
531 let mut opts = FrontendOptions::default();
532 opts.http.timeout = Duration::from_millis(timeout);
533 opts.prom_store.pending_rows_flush_interval = Duration::from_millis(flush_interval);
534 opts.prom_store.enable = enable;
535 opts.prom_store.with_metric_engine = with_metric_engine;
536
537 assert_eq!(
538 Duration::from_millis(expected),
539 effective_http_options_with_sync(&opts, true).timeout,
540 "{name}"
541 );
542 }
543 }
544
545 #[test]
546 fn test_effective_http_timeout_skips_fallback_in_async_batch_mode() {
547 let mut opts = FrontendOptions::default();
551 opts.http.timeout = Duration::from_millis(1000);
552 opts.prom_store.pending_rows_flush_interval = Duration::from_millis(5000);
553
554 assert_eq!(
555 Duration::from_millis(1000),
556 effective_http_options_with_sync(&opts, false).timeout,
557 );
558 assert_eq!(
559 Duration::from_millis(6000),
560 effective_http_options_with_sync(&opts, true).timeout,
561 );
562 }
563
564 #[test]
565 fn test_effective_http_timeout_skips_fallback_when_batcher_disabled() {
566 type KnobMutator = fn(&mut FrontendOptions);
570 let cases: [(&str, KnobMutator); 4] = [
571 ("zero max_batch_rows", |opts| {
572 opts.prom_store.max_batch_rows = 0
573 }),
574 ("zero max_concurrent_flushes", |opts| {
575 opts.prom_store.max_concurrent_flushes = 0
576 }),
577 ("zero worker_channel_capacity", |opts| {
578 opts.prom_store.worker_channel_capacity = 0
579 }),
580 ("zero max_inflight_requests", |opts| {
581 opts.prom_store.max_inflight_requests = 0
582 }),
583 ];
584
585 for (name, disable_batcher) in cases {
586 let mut opts = FrontendOptions::default();
587 opts.http.timeout = Duration::from_millis(1000);
588 opts.prom_store.pending_rows_flush_interval = Duration::from_millis(5000);
589 disable_batcher(&mut opts);
590
591 assert_eq!(
592 Duration::from_millis(1000),
593 effective_http_options_with_sync(&opts, true).timeout,
594 "{name}"
595 );
596 }
597 }
598
599 #[tokio::test]
600 async fn test_database_sql_authentication_differs_between_public_and_internal_grpc() {
601 let options = FrontendOptions {
602 http: HttpOptions {
603 addr: "127.0.0.1:0".to_string(),
604 ..Default::default()
605 },
606 grpc: GrpcOptions::default().with_bind_addr("127.0.0.1:0"),
607 internal_grpc: Some(GrpcOptions::default().with_bind_addr("127.0.0.1:0")),
608 mysql: crate::service_config::MysqlOptions {
609 enable: false,
610 ..Default::default()
611 },
612 postgres: crate::service_config::PostgresOptions {
613 enable: false,
614 ..Default::default()
615 },
616 ..Default::default()
617 };
618 let meta_client = Arc::new(
619 MetaClientBuilder::new(0, Role::Frontend)
620 .enable_procedure()
621 .build(),
622 );
623 let instance = Arc::new(
624 FrontendBuilder::new_test(&options, meta_client)
625 .try_build()
626 .await
627 .unwrap(),
628 );
629 let plugins = Plugins::new();
630 let provider =
631 static_user_provider_from_option("static_user_provider:cmd:greptime=greptime").unwrap();
632 plugins.insert::<UserProviderRef>(Arc::new(provider));
633 let public_flight_handler = Arc::new(GreptimeRequestHandler::new(
634 instance.clone(),
635 plugins.get::<UserProviderRef>(),
636 None,
637 options.grpc.flight_compression,
638 )) as FlightCraftRef;
639 let internal_flight_handler = Arc::new(CountingFlightCraft {
640 inner: Arc::new(GreptimeRequestHandler::new(
641 instance.clone(),
642 None,
643 None,
644 options.grpc.flight_compression,
645 )),
646 do_get_calls: AtomicUsize::new(0),
647 do_put_calls: AtomicUsize::new(0),
648 });
649 let internal_flight_handler_ref = internal_flight_handler.clone() as FlightCraftRef;
650 let mut services = Services::new(options, instance, plugins)
651 .with_flight_handler(public_flight_handler)
652 .with_internal_flight_handler(internal_flight_handler_ref)
653 .build()
654 .unwrap();
655
656 services.start_all().await.unwrap();
657 let public_addr = services.addr(GRPC_SERVER).unwrap();
658 let internal_addr = services.addr("INTERNAL_GRPC_SERVER").unwrap();
659 let public_database = Database::new(
660 "greptime",
661 "public",
662 Client::with_urls([public_addr.to_string()]),
663 );
664 let internal_database = Database::new(
665 "greptime",
666 "public",
667 Client::with_urls([internal_addr.to_string()]),
668 );
669
670 let internal_result = internal_database.sql("SELECT 1").await;
671 let put_result = internal_database
672 .do_put(Box::pin(futures::stream::empty()))
673 .await;
674 let public_result = public_database.sql("SELECT 1").await;
675
676 services.shutdown_all().await.unwrap();
677
678 assert!(internal_result.is_ok());
679 assert!(put_result.is_ok());
680 assert_eq!(
681 1,
682 internal_flight_handler.do_get_calls.load(Ordering::SeqCst)
683 );
684 assert_eq!(
685 1,
686 internal_flight_handler.do_put_calls.load(Ordering::SeqCst)
687 );
688 assert_eq!(
689 Some(Code::Unauthenticated),
690 public_result
691 .as_ref()
692 .err()
693 .and_then(|err| err.tonic_code())
694 );
695 }
696
697 #[tokio::test]
698 async fn test_services_builder_health_check_is_reachable() {
699 let options = FrontendOptions {
701 http: HttpOptions {
702 addr: "127.0.0.1:0".to_string(),
703 ..Default::default()
704 },
705 grpc: GrpcOptions::default().with_bind_addr("127.0.0.1:0"),
706 mysql: crate::service_config::MysqlOptions {
707 enable: false,
708 ..Default::default()
709 },
710 postgres: crate::service_config::PostgresOptions {
711 enable: false,
712 ..Default::default()
713 },
714 ..Default::default()
715 };
716 let meta_client = Arc::new(
717 MetaClientBuilder::new(0, Role::Frontend)
718 .enable_procedure()
719 .build(),
720 );
721 let instance = Arc::new(
722 FrontendBuilder::new_test(&options, meta_client)
723 .try_build()
724 .await
725 .unwrap(),
726 );
727 let mut services = Services::new(options, instance, Default::default())
728 .build()
729 .unwrap();
730
731 services.start_all().await.unwrap();
733 let addr = services.addr(GRPC_SERVER).unwrap();
734 let health_check = HealthCheckClient::connect(format!("http://{addr}"))
735 .await
736 .unwrap()
737 .health_check(HealthCheckRequest {})
738 .await;
739 services.shutdown_all().await.unwrap();
740
741 assert!(health_check.is_ok());
743 }
744}