frontend/instance/
logs.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::ops::Deref;
16
17use auth::{LOG_QUERY, PermissionReq, PermissionTableTarget, PermissionTableTargets};
18use client::Output;
19use common_error::ext::BoxedError;
20use log_query::LogQuery;
21use server_error::Result as ServerResult;
22use servers::error::{self as server_error, AuthSnafu, ExecuteQuerySnafu};
23use servers::interceptor::{LogQueryInterceptor, LogQueryInterceptorRef};
24use servers::query_handler::LogQueryHandler;
25use session::context::{QueryContext, QueryContextRef};
26use snafu::ResultExt;
27use tonic::async_trait;
28
29use crate::instance::{Instance, map_query_output};
30
31#[async_trait]
32impl LogQueryHandler for Instance {
33    async fn query(&self, mut request: LogQuery, ctx: QueryContextRef) -> ServerResult<Output> {
34        let interceptor = self
35            .plugins
36            .get::<LogQueryInterceptorRef<server_error::Error>>();
37        let targets = PermissionTableTargets::resolved(vec![PermissionTableTarget::new(
38            &request.table.catalog_name,
39            &request.table.schema_name,
40            &request.table.table_name,
41        )]);
42        let targets = self.resolve_query_permission_targets(targets, &ctx).await?;
43
44        self.check_table_permission(&ctx, PermissionReq::Action(LOG_QUERY), targets)
45            .context(AuthSnafu)?;
46
47        interceptor.as_ref().pre_query(&request, ctx.clone())?;
48
49        request
50            .time_filter
51            .canonicalize()
52            .map_err(BoxedError::new)
53            .context(ExecuteQuerySnafu)?;
54
55        let plan = self
56            .query_engine
57            .planner()
58            .plan_logs_query(request, ctx.clone())
59            .await
60            .map_err(BoxedError::new)
61            .context(ExecuteQuerySnafu)?;
62
63        let output = self
64            .statement_executor
65            .exec_plan(plan, ctx.clone())
66            .await
67            .map_err(BoxedError::new)
68            .context(ExecuteQuerySnafu)?;
69
70        let output = map_query_output(output)
71            .map_err(BoxedError::new)
72            .context(ExecuteQuerySnafu)?;
73        Ok(interceptor.as_ref().post_query(output, ctx.clone())?)
74    }
75
76    fn catalog_manager(&self, _ctx: &QueryContext) -> ServerResult<&dyn catalog::CatalogManager> {
77        Ok(self.catalog_manager.deref())
78    }
79}
frontend/instance/logs.rs

frontend/instance/
logs.rs
