frontend/instance/
log_handler.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 api::v1::RowInsertRequests;
18use async_trait::async_trait;
19use auth::{
20    LOG_WRITE, PIPELINE_DELETE, PIPELINE_INSERT, PIPELINE_QUERY, PermissionChecker,
21    PermissionCheckerRef, PermissionReq,
22};
23use client::Output;
24use common_error::ext::BoxedError;
25use datatypes::timestamp::TimestampNanosecond;
26use pipeline::pipeline_operator::PipelineOperator;
27use pipeline::{Pipeline, PipelineInfo, PipelineVersion};
28use servers::error::{
29    AuthSnafu, Error as ServerError, ExecuteGrpcRequestSnafu, PipelineSnafu, Result as ServerResult,
30};
31use servers::interceptor::{LogIngestInterceptor, LogIngestInterceptorRef};
32use servers::query_handler::PipelineHandler;
33use session::context::{QueryContext, QueryContextRef};
34use snafu::ResultExt;
35use table::Table;
36
37use crate::instance::Instance;
38
39impl Instance {
40    async fn prepare_log_insert(
41        &self,
42        log: RowInsertRequests,
43        ctx: QueryContextRef,
44    ) -> ServerResult<RowInsertRequests> {
45        self.plugins
46            .get::<PermissionCheckerRef>()
47            .as_ref()
48            .check_permission(ctx.current_user(), PermissionReq::Action(LOG_WRITE))
49            .context(AuthSnafu)?;
50
51        let log = self
52            .plugins
53            .get::<LogIngestInterceptorRef<ServerError>>()
54            .as_ref()
55            .pre_ingest(log, ctx.clone())?;
56
57        self.check_row_insert_permission(&log, &ctx, PermissionReq::Action(LOG_WRITE))
58            .context(AuthSnafu)?;
59
60        Ok(log)
61    }
62}
63
64#[async_trait]
65impl PipelineHandler for Instance {
66    async fn insert(&self, log: RowInsertRequests, ctx: QueryContextRef) -> ServerResult<Output> {
67        let log = self.prepare_log_insert(log, ctx.clone()).await?;
68        self.handle_log_inserts(log, Arc::new(ctx.fork())).await
69    }
70
71    async fn insert_all(
72        &self,
73        inputs: Vec<(QueryContextRef, RowInsertRequests)>,
74    ) -> ServerResult<Vec<ServerResult<Output>>> {
75        let mut prepared = Vec::with_capacity(inputs.len());
76        for (ctx, log) in inputs {
77            let log = self.prepare_log_insert(log, ctx.clone()).await?;
78            // Detach from context clones retained by pre-ingest hooks so the
79            // checked schema cannot change before this batch is written.
80            prepared.push((Arc::new(ctx.fork()), log));
81        }
82
83        let mut outputs = Vec::with_capacity(prepared.len());
84        for (ctx, log) in prepared {
85            outputs.push(self.handle_log_inserts(log, ctx).await);
86        }
87        Ok(outputs)
88    }
89
90    fn check_pipeline_query_permission(&self, query_ctx: &QueryContextRef) -> ServerResult<()> {
91        self.check_permission(query_ctx, PermissionReq::Action(PIPELINE_QUERY))
92    }
93
94    async fn get_pipeline(
95        &self,
96        name: &str,
97        version: PipelineVersion,
98        query_ctx: QueryContextRef,
99    ) -> ServerResult<Arc<Pipeline>> {
100        self.pipeline_operator
101            .get_pipeline(query_ctx, name, version)
102            .await
103            .context(PipelineSnafu)
104    }
105
106    async fn insert_pipeline(
107        &self,
108        name: &str,
109        content_type: &str,
110        pipeline: &str,
111        query_ctx: QueryContextRef,
112    ) -> ServerResult<PipelineInfo> {
113        self.check_permission(&query_ctx, PermissionReq::Action(PIPELINE_INSERT))?;
114        self.pipeline_operator
115            .insert_pipeline(name, content_type, pipeline, query_ctx)
116            .await
117            .context(PipelineSnafu)
118    }
119
120    async fn delete_pipeline(
121        &self,
122        name: &str,
123        version: PipelineVersion,
124        ctx: QueryContextRef,
125    ) -> ServerResult<Option<()>> {
126        self.check_permission(&ctx, PermissionReq::Action(PIPELINE_DELETE))?;
127        self.pipeline_operator
128            .delete_pipeline(name, version, ctx)
129            .await
130            .context(PipelineSnafu)
131    }
132
133    async fn get_table(
134        &self,
135        table: &str,
136        query_ctx: &QueryContext,
137    ) -> std::result::Result<Option<Arc<Table>>, catalog::error::Error> {
138        let catalog = query_ctx.current_catalog();
139        let schema = query_ctx.current_schema();
140        self.catalog_manager
141            .table(catalog, &schema, table, None)
142            .await
143    }
144
145    fn build_pipeline(&self, pipeline: &str) -> ServerResult<Pipeline> {
146        PipelineOperator::build_pipeline(pipeline).context(PipelineSnafu)
147    }
148
149    async fn get_pipeline_str(
150        &self,
151        name: &str,
152        version: PipelineVersion,
153        query_ctx: QueryContextRef,
154    ) -> ServerResult<(String, TimestampNanosecond)> {
155        self.check_permission(&query_ctx, PermissionReq::Action(PIPELINE_QUERY))?;
156        self.pipeline_operator
157            .get_pipeline_str(name, version, query_ctx)
158            .await
159            .context(PipelineSnafu)
160    }
161}
162
163impl Instance {
164    pub async fn handle_log_inserts(
165        &self,
166        log: RowInsertRequests,
167        ctx: QueryContextRef,
168    ) -> ServerResult<Output> {
169        self.inserter
170            .handle_log_inserts(log, ctx, self.statement_executor.as_ref())
171            .await
172            .map_err(BoxedError::new)
173            .context(ExecuteGrpcRequestSnafu)
174    }
175
176    pub async fn handle_trace_inserts(
177        &self,
178        rows: RowInsertRequests,
179        ctx: QueryContextRef,
180    ) -> ServerResult<Output> {
181        self.inserter
182            .handle_trace_inserts(rows, ctx, self.statement_executor.as_ref())
183            .await
184            .map_err(BoxedError::new)
185            .context(ExecuteGrpcRequestSnafu)
186    }
187}
frontend/instance/log_handler.rs

frontend/instance/
log_handler.rs
