frontend/
frontend.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 common_base::readable_size::ReadableSize;
18use common_config::config::Configurable;
19use common_event_recorder::EventRecorderOptions;
20use common_memory_manager::OnExhaustedPolicy;
21use common_options::datanode::DatanodeClientOptions;
22use common_options::memory::MemoryOptions;
23use common_telemetry::logging::{LoggingOptions, SlowQueryOptions, TracingOptions};
24use meta_client::MetaClientOptions;
25use pipeline::PipelineOptions;
26use query::options::QueryOptions;
27use serde::{Deserialize, Serialize};
28use servers::grpc::GrpcOptions;
29use servers::http::HttpOptions;
30use servers::server::ServerHandlers;
31use snafu::ResultExt;
32
33use crate::error;
34use crate::error::Result;
35use crate::heartbeat::HeartbeatTask;
36use crate::instance::Instance;
37use crate::service_config::{
38    InfluxdbOptions, JaegerOptions, MysqlOptions, OpentsdbOptions, OtlpOptions,
39    PendingRowsBatcherOptions, PostgresOptions, PromStoreOptions,
40};
41
42#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
43#[serde(default)]
44pub struct FrontendOptions {
45    pub node_id: Option<String>,
46    pub default_timezone: Option<String>,
47    pub default_column_prefix: Option<String>,
48    /// Server-side global switch for auto table creation on write.
49    /// Acts as an upper bound: when `false`, missing tables are never auto-created
50    /// even if a request sets the `auto_create_table` hint to `true`. When `true`
51    /// (default), the per-request hint still applies. Default: `true`.
52    pub auto_create_table: bool,
53    /// Maximum total memory for all concurrent write request bodies and messages (HTTP, gRPC, Flight).
54    /// Set to 0 to disable the limit. Default: "0" (unlimited)
55    pub max_in_flight_write_bytes: ReadableSize,
56    /// Policy when write bytes quota is exhausted.
57    /// Options: "wait" (default, 10s), "wait(<duration>)", "fail"
58    pub write_bytes_exhausted_policy: OnExhaustedPolicy,
59    pub http: HttpOptions,
60    pub grpc: GrpcOptions,
61    /// The internal gRPC options for the frontend service.
62    /// it provide the same service as the public gRPC service, just only for internal use.
63    pub internal_grpc: Option<GrpcOptions>,
64    pub mysql: MysqlOptions,
65    pub postgres: PostgresOptions,
66    pub opentsdb: OpentsdbOptions,
67    pub influxdb: InfluxdbOptions,
68    /// Shared experimental ordinary-table batching; independent of Prom batching.
69    pub pending_rows_batcher: PendingRowsBatcherOptions,
70    pub prom_store: PromStoreOptions,
71    pub jaeger: JaegerOptions,
72    pub otlp: OtlpOptions,
73    pub meta_client: Option<MetaClientOptions>,
74    pub logging: LoggingOptions,
75    pub datanode: DatanodeClientOptions,
76    pub user_provider: Option<String>,
77    pub tracing: TracingOptions,
78    pub query: QueryOptions,
79    pub slow_query: SlowQueryOptions,
80    pub memory: MemoryOptions,
81    /// The pipeline options.
82    pub pipeline: PipelineOptions,
83    /// The event recorder options.
84    pub event_recorder: EventRecorderOptions,
85    /// Environment variable keys to read and report in heartbeat messages.
86    pub heartbeat_env_vars: Vec<String>,
87}
88
89impl Default for FrontendOptions {
90    fn default() -> Self {
91        Self {
92            node_id: None,
93            default_timezone: None,
94            default_column_prefix: None,
95            auto_create_table: true,
96            max_in_flight_write_bytes: ReadableSize(0),
97            write_bytes_exhausted_policy: OnExhaustedPolicy::default(),
98            http: HttpOptions::default(),
99            grpc: GrpcOptions::default(),
100            internal_grpc: None,
101            mysql: MysqlOptions::default(),
102            postgres: PostgresOptions::default(),
103            opentsdb: OpentsdbOptions::default(),
104            influxdb: InfluxdbOptions::default(),
105            pending_rows_batcher: PendingRowsBatcherOptions::default(),
106            jaeger: JaegerOptions::default(),
107            prom_store: PromStoreOptions::default(),
108            otlp: OtlpOptions::default(),
109            meta_client: None,
110            logging: LoggingOptions::default(),
111            datanode: DatanodeClientOptions::default(),
112            user_provider: None,
113            tracing: TracingOptions::default(),
114            query: QueryOptions::default(),
115            slow_query: SlowQueryOptions::default(),
116            memory: MemoryOptions::default(),
117            pipeline: PipelineOptions::default(),
118            event_recorder: EventRecorderOptions::default(),
119            heartbeat_env_vars: vec![],
120        }
121    }
122}
123
124impl Configurable for FrontendOptions {
125    fn env_list_keys() -> Option<&'static [&'static str]> {
126        Some(&[
127            "heartbeat_env_vars",
128            "meta_client.metasrv_addrs",
129            "event_recorder.event_types",
130            "pending_rows_batcher.protocols",
131        ])
132    }
133}
134
135/// The [`Frontend`] struct is the main entry point for the frontend service
136/// which contains server handlers, frontend instance and some background tasks.
137pub struct Frontend {
138    pub instance: Arc<Instance>,
139    pub servers: ServerHandlers,
140    pub heartbeat_task: Option<HeartbeatTask>,
141}
142
143impl Frontend {
144    pub async fn start(&mut self) -> Result<()> {
145        if let Some(t) = &self.heartbeat_task
146            && let Err(error) = t.start().await
147        {
148            t.shutdown().await;
149            return Err(error);
150        }
151
152        if let Err(source) = self.servers.start_all().await {
153            if let Some(t) = &self.heartbeat_task {
154                t.shutdown().await;
155            }
156            return Err(source).context(error::StartServerSnafu);
157        }
158        Ok(())
159    }
160
161    pub async fn shutdown(&mut self) -> Result<()> {
162        if let Some(t) = &self.heartbeat_task {
163            t.shutdown().await;
164        }
165        self.servers
166            .shutdown_all()
167            .await
168            .context(error::ShutdownServerSnafu)
169    }
170
171    pub fn server_handlers(&self) -> &ServerHandlers {
172        &self.servers
173    }
174}
175
176#[cfg(test)]
177mod tests {
178    use std::any::Any;
179    use std::net::SocketAddr;
180    use std::pin::Pin;
181    use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
182    use std::time::Duration;
183
184    use api::v1::meta::heartbeat_server::HeartbeatServer;
185    use api::v1::meta::mailbox_message::Payload;
186    use api::v1::meta::{
187        AskLeaderRequest, AskLeaderResponse, HeartbeatRequest, HeartbeatResponse, MailboxMessage,
188        Peer, ResponseHeader, Role, heartbeat_server,
189    };
190    use async_trait::async_trait;
191    use client::{Client, Database};
192    use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
193    use common_error::ext::ErrorExt;
194    use common_error::from_header_to_err_code_msg;
195    use common_error::status_code::StatusCode;
196    use common_grpc::channel_manager::ChannelManager;
197    use common_meta::heartbeat::handler::HandlerGroupExecutor;
198    use common_meta::heartbeat::handler::parse_mailbox_message::ParseMailboxMessageHandler;
199    use common_meta::heartbeat::handler::suspend::SuspendHandler;
200    use common_meta::instruction::Instruction;
201    use common_stat::ResourceStatImpl;
202    use futures::Stream;
203    use meta_client::MetaClientRef;
204    use meta_client::client::MetaClientBuilder;
205    use servers::grpc::{FlightCompression, GRPC_SERVER};
206    use servers::http::HTTP_SERVER;
207    use servers::http::result::greptime_result_v1::GreptimedbV1Response;
208    use servers::server::Server;
209    use tokio::sync::mpsc;
210    use tonic::codec::CompressionEncoding;
211    use tonic::codegen::tokio_stream::StreamExt;
212    use tonic::codegen::tokio_stream::wrappers::ReceiverStream;
213    use tonic::{Request, Response, Status, Streaming};
214
215    use crate::frontend::*;
216    use crate::heartbeat::{
217        FrontendHeartbeatExtension, FrontendHeartbeatExtensionResult, FrontendHeartbeatExtensions,
218    };
219    use crate::instance::builder::FrontendBuilder;
220    use crate::server::Services;
221
222    type GrpcStream<T> =
223        Pin<Box<dyn Stream<Item = std::result::Result<T, Status>> + Send + Sync + 'static>>;
224
225    #[test]
226    fn test_batcher_protocols_from_env() {
227        temp_env::with_vars(
228            [(
229                "FRONTEND_BATCHER_TEST__PENDING_ROWS_BATCHER__PROTOCOLS",
230                Some("influxdb,http_sql"),
231            )],
232            || {
233                let options =
234                    FrontendOptions::load_layered_options(None, "FRONTEND_BATCHER_TEST").unwrap();
235                assert_eq!(
236                    options.pending_rows_batcher.protocols,
237                    vec![
238                        servers::http::BatchingProtocol::Influxdb,
239                        servers::http::BatchingProtocol::HttpSql
240                    ]
241                );
242            },
243        );
244    }
245
246    #[test]
247    fn test_protocol_pending_rows_batcher_config() {
248        let defaults: FrontendOptions = toml::from_str("").unwrap();
249        assert!(
250            !defaults
251                .pending_rows_batcher
252                .pending_rows_batching_enabled()
253        );
254        let options: FrontendOptions = toml::from_str(
255            r#"
256[pending_rows_batcher]
257protocols = ["influxdb", "http_sql"]
258pending_rows_flush_interval = "5ms"
259max_batch_rows = 25
260"#,
261        )
262        .unwrap();
263        assert_eq!(options.pending_rows_batcher.max_batch_rows, 25);
264        assert_eq!(options.pending_rows_batcher.protocols.len(), 2);
265        assert!(options.pending_rows_batcher.pending_rows_batching_enabled());
266        let serialized = toml::to_string(&options).unwrap();
267        let parsed: FrontendOptions = toml::from_str(&serialized).unwrap();
268        assert_eq!(options.influxdb, parsed.influxdb);
269        assert_eq!(options.opentsdb, parsed.opentsdb);
270        assert_eq!(options.pending_rows_batcher, parsed.pending_rows_batcher);
271    }
272
273    #[test]
274    fn test_toml() {
275        let opts = FrontendOptions::default();
276        let toml_string = toml::to_string(&opts).unwrap();
277        assert!(toml_string.contains("experimental_enable_exponential_histogram = false"));
278        let parsed: FrontendOptions = toml::from_str(&toml_string).unwrap();
279        assert_eq!(parsed.otlp, opts.otlp);
280        assert_eq!(parsed.influxdb, opts.influxdb);
281        assert_eq!(parsed.opentsdb, opts.opentsdb);
282    }
283
284    #[test]
285    fn test_http_api_server_defaults_on_when_absent() {
286        // When `[http]` is not present in the config, the dedicated API server is
287        // disabled by default (its port/host defaults are still 4006 / 127.0.0.1).
288        let parsed: FrontendOptions = toml::from_str("").unwrap();
289        assert!(!parsed.http.enable_api_server);
290        assert_eq!(parsed.http.api_server_addr, "127.0.0.1:4006");
291    }
292
293    struct SuspendableHeartbeatServer {
294        suspend: Arc<AtomicBool>,
295        fail_heartbeat: bool,
296    }
297
298    struct FailingServer;
299
300    struct ShutdownTrackingExtension {
301        shutdown_calls: AtomicUsize,
302    }
303
304    #[async_trait]
305    impl FrontendHeartbeatExtension for ShutdownTrackingExtension {
306        fn name(&self) -> &str {
307            "shutdown-tracking"
308        }
309
310        async fn shutdown(&self) -> FrontendHeartbeatExtensionResult<()> {
311            self.shutdown_calls.fetch_add(1, Ordering::AcqRel);
312            Ok(())
313        }
314    }
315
316    #[async_trait]
317    impl Server for FailingServer {
318        async fn shutdown(&self) -> servers::error::Result<()> {
319            Ok(())
320        }
321
322        async fn start(&mut self, _listening: SocketAddr) -> servers::error::Result<()> {
323            Err(servers::error::Error::Internal {
324                err_msg: "mock server start failure".to_string(),
325            })
326        }
327
328        fn name(&self) -> &str {
329            "FAILING_SERVER"
330        }
331
332        fn as_any(&self) -> &dyn Any {
333            self
334        }
335    }
336
337    #[async_trait]
338    impl heartbeat_server::Heartbeat for SuspendableHeartbeatServer {
339        type HeartbeatStream = GrpcStream<HeartbeatResponse>;
340
341        async fn heartbeat(
342            &self,
343            request: Request<Streaming<HeartbeatRequest>>,
344        ) -> std::result::Result<Response<Self::HeartbeatStream>, Status> {
345            if self.fail_heartbeat {
346                return Err(Status::unavailable("mock initial heartbeat failure"));
347            }
348
349            let (tx, rx) = mpsc::channel(4);
350
351            common_runtime::spawn_global({
352                let mut requests = request.into_inner();
353                let suspend = self.suspend.clone();
354                async move {
355                    // Make the heartbeat interval short in unit tests to reduce the waiting time.
356                    // Only the handshake response needs to populate it (as metasrv does).
357                    let heartbeat_interval_ms = Duration::from_millis(200).as_millis() as u64;
358                    let mut is_handshake = true;
359                    while let Some(request) = requests.next().await {
360                        if let Err(e) = request {
361                            let _ = tx.send(Err(e)).await;
362                            return;
363                        }
364
365                        let mailbox_message =
366                            suspend.load(Ordering::Relaxed).then(|| MailboxMessage {
367                                payload: Some(Payload::Json(
368                                    serde_json::to_string(&Instruction::Suspend).unwrap(),
369                                )),
370                                ..Default::default()
371                            });
372                        let heartbeat_config =
373                            is_handshake.then_some(api::v1::meta::HeartbeatConfig {
374                                heartbeat_interval_ms,
375                                retry_interval_ms: heartbeat_interval_ms,
376                                gc_enabled: false,
377                            });
378                        is_handshake = false;
379                        let response = HeartbeatResponse {
380                            header: Some(ResponseHeader::success()),
381                            mailbox_message,
382                            heartbeat_config,
383                            ..Default::default()
384                        };
385
386                        let _ = tx.send(Ok(response)).await;
387                    }
388                }
389            });
390
391            Ok(Response::new(Box::pin(ReceiverStream::new(rx))))
392        }
393
394        async fn ask_leader(
395            &self,
396            _: Request<AskLeaderRequest>,
397        ) -> std::result::Result<Response<AskLeaderResponse>, Status> {
398            Ok(Response::new(AskLeaderResponse {
399                header: Some(ResponseHeader::success()),
400                leader: Some(Peer {
401                    addr: "localhost:0".to_string(),
402                    ..Default::default()
403                }),
404            }))
405        }
406    }
407
408    async fn create_meta_client(
409        options: &MetaClientOptions,
410        heartbeat_server: Arc<SuspendableHeartbeatServer>,
411    ) -> MetaClientRef {
412        let (client, server) = tokio::io::duplex(1024);
413
414        // create the heartbeat server:
415        common_runtime::spawn_global(async move {
416            let mut router = tonic::transport::Server::builder();
417            let router = router.add_service(
418                HeartbeatServer::from_arc(heartbeat_server)
419                    .accept_compressed(CompressionEncoding::Zstd)
420                    .send_compressed(CompressionEncoding::Zstd),
421            );
422            router
423                .serve_with_incoming(futures::stream::iter([Ok::<_, std::io::Error>(server)]))
424                .await
425        });
426
427        // Move client to an option so we can _move_ the inner value
428        // on the first attempt to connect. All other attempts will fail.
429        let mut client = Some(client);
430        let connector = tower::service_fn(move |_| {
431            let client = client.take();
432            async move {
433                if let Some(client) = client {
434                    Ok(hyper_util::rt::TokioIo::new(client))
435                } else {
436                    Err(std::io::Error::other("client already taken"))
437                }
438            }
439        });
440        let manager = ChannelManager::new();
441        manager
442            .reset_with_connector("localhost:0", connector)
443            .unwrap();
444
445        // create the heartbeat client:
446        let mut client = MetaClientBuilder::new(0, Role::Frontend)
447            .enable_heartbeat()
448            .heartbeat_channel_manager(manager)
449            .build();
450        client.start(&options.metasrv_addrs).await.unwrap();
451        Arc::new(client)
452    }
453
454    async fn create_frontend(
455        options: &FrontendOptions,
456        meta_client: MetaClientRef,
457    ) -> Result<Frontend> {
458        let instance = Arc::new(
459            FrontendBuilder::new_test(options, meta_client.clone())
460                .try_build()
461                .await?,
462        );
463
464        let servers =
465            Services::new(options.clone(), instance.clone(), Default::default()).build()?;
466
467        let executor = Arc::new(HandlerGroupExecutor::new(vec![
468            Arc::new(ParseMailboxMessageHandler),
469            Arc::new(SuspendHandler::new(instance.suspend_state())),
470        ]));
471        let heartbeat_task = Some(HeartbeatTask::new(
472            instance.frontend_peer_addr().to_string(),
473            options,
474            meta_client,
475            executor,
476            Arc::new(ResourceStatImpl::default()),
477        ));
478
479        let mut frontend = Frontend {
480            instance,
481            servers,
482            heartbeat_task,
483        };
484        frontend.start().await?;
485        Ok(frontend)
486    }
487
488    #[tokio::test]
489    async fn test_server_start_failure_shuts_down_heartbeat() {
490        let meta_client_options = MetaClientOptions {
491            metasrv_addrs: vec!["localhost:0".to_string()],
492            ..Default::default()
493        };
494        let options = FrontendOptions {
495            meta_client: Some(meta_client_options.clone()),
496            ..Default::default()
497        };
498        let heartbeat_server = Arc::new(SuspendableHeartbeatServer {
499            suspend: Arc::new(AtomicBool::new(false)),
500            fail_heartbeat: false,
501        });
502        let meta_client = create_meta_client(&meta_client_options, heartbeat_server).await;
503        let instance = Arc::new(
504            FrontendBuilder::new_test(&options, meta_client.clone())
505                .try_build()
506                .await
507                .unwrap(),
508        );
509        let heartbeat_task = HeartbeatTask::new(
510            instance.frontend_peer_addr().to_string(),
511            &options,
512            meta_client,
513            Arc::new(HandlerGroupExecutor::new(vec![])),
514            Arc::new(ResourceStatImpl::default()),
515        );
516        let heartbeat_probe = heartbeat_task.clone();
517        let servers = ServerHandlers::default();
518        servers.insert((Box::new(FailingServer), "127.0.0.1:0".parse().unwrap()));
519        let mut frontend = Frontend {
520            instance,
521            servers,
522            heartbeat_task: Some(heartbeat_task),
523        };
524
525        assert!(frontend.start().await.is_err());
526        assert!(heartbeat_probe.is_shutdown());
527    }
528
529    #[tokio::test]
530    async fn test_heartbeat_start_failure_shuts_down_extensions() {
531        let meta_client_options = MetaClientOptions {
532            metasrv_addrs: vec!["localhost:0".to_string()],
533            ..Default::default()
534        };
535        let options = FrontendOptions {
536            meta_client: Some(meta_client_options.clone()),
537            ..Default::default()
538        };
539        let heartbeat_server = Arc::new(SuspendableHeartbeatServer {
540            suspend: Arc::new(AtomicBool::new(false)),
541            fail_heartbeat: true,
542        });
543        let meta_client = create_meta_client(&meta_client_options, heartbeat_server).await;
544        let instance = Arc::new(
545            FrontendBuilder::new_test(&options, meta_client.clone())
546                .try_build()
547                .await
548                .unwrap(),
549        );
550        let extension = Arc::new(ShutdownTrackingExtension {
551            shutdown_calls: AtomicUsize::new(0),
552        });
553        let extensions = FrontendHeartbeatExtensions::default();
554        assert!(extensions.register(extension.clone()));
555        let heartbeat_task = HeartbeatTask::new(
556            instance.frontend_peer_addr().to_string(),
557            &options,
558            meta_client,
559            Arc::new(HandlerGroupExecutor::new(vec![])),
560            Arc::new(ResourceStatImpl::default()),
561        )
562        .with_extensions(extensions);
563        let heartbeat_probe = heartbeat_task.clone();
564        let mut frontend = Frontend {
565            instance,
566            servers: ServerHandlers::default(),
567            heartbeat_task: Some(heartbeat_task),
568        };
569
570        assert!(frontend.start().await.is_err());
571        assert!(heartbeat_probe.is_shutdown());
572        assert_eq!(extension.shutdown_calls.load(Ordering::Acquire), 1);
573    }
574
575    async fn verify_suspend_state_by_http(
576        frontend: &Frontend,
577        expected: std::result::Result<&str, (StatusCode, &str)>,
578    ) {
579        let addr = frontend.server_handlers().addr(HTTP_SERVER).unwrap();
580        let response = reqwest::get(format!("http://{}/v1/sql?sql=SELECT 1", addr))
581            .await
582            .unwrap();
583
584        let headers = response.headers();
585        let response = if let Some((code, error)) = from_header_to_err_code_msg(headers) {
586            Err((code, error))
587        } else {
588            Ok(response.text().await.unwrap())
589        };
590
591        match (response, expected) {
592            (Ok(response), Ok(expected)) => {
593                let response: GreptimedbV1Response = serde_json::from_str(&response).unwrap();
594                let response = serde_json::to_string(response.output()).unwrap();
595                assert_eq!(&response, expected);
596            }
597            (Err(actual), Err(expected)) => assert_eq!(actual, expected),
598            _ => unreachable!(),
599        }
600    }
601
602    async fn verify_suspend_state_by_grpc(
603        frontend: &Frontend,
604        expected: std::result::Result<&str, (StatusCode, &str)>,
605    ) {
606        let addr = frontend.server_handlers().addr(GRPC_SERVER).unwrap();
607        let client = Client::with_urls([addr.to_string()]);
608        let client = Database::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, client);
609        let response = client.sql("SELECT 1").await;
610
611        match (response, expected) {
612            (Ok(response), Ok(expected)) => {
613                let response = response.data.pretty_print().await;
614                assert_eq!(&response, expected.trim());
615            }
616            (Err(actual), Err(expected)) => {
617                assert_eq!(actual.status_code(), expected.0);
618                assert_eq!(actual.output_msg(), expected.1);
619            }
620            _ => unreachable!(),
621        }
622    }
623
624    async fn wait_for_suspend_state(frontend: &Frontend, expected: bool) {
625        let check = || frontend.instance.is_suspended() == expected;
626        if check() {
627            return;
628        }
629
630        tokio::time::timeout(Duration::from_secs(5), async move {
631            while !check() {
632                tokio::time::sleep(Duration::from_millis(20)).await;
633            }
634        })
635        .await
636        .unwrap();
637    }
638
639    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
640    async fn test_suspend_frontend() -> Result<()> {
641        common_telemetry::init_default_ut_logging();
642
643        let meta_client_options = MetaClientOptions {
644            metasrv_addrs: vec!["localhost:0".to_string()],
645            ..Default::default()
646        };
647        let options = FrontendOptions {
648            http: HttpOptions {
649                addr: "127.0.0.1:0".to_string(),
650                ..Default::default()
651            },
652            grpc: GrpcOptions {
653                bind_addr: "127.0.0.1:0".to_string(),
654                flight_compression: FlightCompression::None,
655                ..Default::default()
656            },
657            mysql: MysqlOptions {
658                enable: false,
659                ..Default::default()
660            },
661            postgres: PostgresOptions {
662                enable: false,
663                ..Default::default()
664            },
665            meta_client: Some(meta_client_options.clone()),
666            ..Default::default()
667        };
668
669        let server = Arc::new(SuspendableHeartbeatServer {
670            suspend: Arc::new(AtomicBool::new(false)),
671            fail_heartbeat: false,
672        });
673        let meta_client = create_meta_client(&meta_client_options, server.clone()).await;
674        let frontend = create_frontend(&options, meta_client).await?;
675
676        // initial state: not suspend:
677        assert!(!frontend.instance.is_suspended());
678        verify_suspend_state_by_http(&frontend, Ok(r#"[{"records":{"schema":{"column_schemas":[{"name":"Int64(1)","data_type":"Int64"}]},"rows":[[1]],"total_rows":1}}]"#)).await;
679        verify_suspend_state_by_grpc(
680            &frontend,
681            Ok(r#"
682+----------+
683| Int64(1) |
684+----------+
685| 1        |
686+----------+"#),
687        )
688        .await;
689
690        // make heartbeat server returned "suspend" instruction,
691        server.suspend.store(true, Ordering::Relaxed);
692        wait_for_suspend_state(&frontend, true).await;
693        // ... then the frontend is suspended:
694        assert!(frontend.instance.is_suspended());
695        verify_suspend_state_by_http(
696            &frontend,
697            Err((
698                StatusCode::Suspended,
699                "error: Service suspended, execution_time_ms: 0",
700            )),
701        )
702        .await;
703        verify_suspend_state_by_grpc(&frontend, Err((StatusCode::Suspended, "Service suspended")))
704            .await;
705
706        // make heartbeat server NOT returned "suspend" instruction,
707        server.suspend.store(false, Ordering::Relaxed);
708        wait_for_suspend_state(&frontend, false).await;
709        // ... then frontend's suspend state is cleared:
710        assert!(!frontend.instance.is_suspended());
711        verify_suspend_state_by_http(&frontend, Ok(r#"[{"records":{"schema":{"column_schemas":[{"name":"Int64(1)","data_type":"Int64"}]},"rows":[[1]],"total_rows":1}}]"#)).await;
712        verify_suspend_state_by_grpc(
713            &frontend,
714            Ok(r#"
715+----------+
716| Int64(1) |
717+----------+
718| 1        |
719+----------+"#),
720        )
721        .await;
722        Ok(())
723    }
724}
frontend/frontend.rs

frontend/
frontend.rs
