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        operator::insert::admit_row_insert_batches(&mut prepared)
84            .await
85            .map_err(BoxedError::new)
86            .context(servers::error::ExecuteGrpcQuerySnafu)?;
87
88        let mut outputs = Vec::with_capacity(prepared.len());
89        for (ctx, log) in prepared {
90            outputs.push(self.handle_log_inserts(log, ctx).await);
91        }
92        Ok(outputs)
93    }
94
95    fn check_pipeline_query_permission(&self, query_ctx: &QueryContextRef) -> ServerResult<()> {
96        self.check_permission(query_ctx, PermissionReq::Action(PIPELINE_QUERY))
97    }
98
99    async fn get_pipeline(
100        &self,
101        name: &str,
102        version: PipelineVersion,
103        query_ctx: QueryContextRef,
104    ) -> ServerResult<Arc<Pipeline>> {
105        self.pipeline_operator
106            .get_pipeline(query_ctx, name, version)
107            .await
108            .context(PipelineSnafu)
109    }
110
111    async fn insert_pipeline(
112        &self,
113        name: &str,
114        content_type: &str,
115        pipeline: &str,
116        query_ctx: QueryContextRef,
117    ) -> ServerResult<PipelineInfo> {
118        self.check_permission(&query_ctx, PermissionReq::Action(PIPELINE_INSERT))?;
119        self.pipeline_operator
120            .insert_pipeline(name, content_type, pipeline, query_ctx)
121            .await
122            .context(PipelineSnafu)
123    }
124
125    async fn delete_pipeline(
126        &self,
127        name: &str,
128        version: PipelineVersion,
129        ctx: QueryContextRef,
130    ) -> ServerResult<Option<()>> {
131        self.check_permission(&ctx, PermissionReq::Action(PIPELINE_DELETE))?;
132        self.pipeline_operator
133            .delete_pipeline(name, version, ctx)
134            .await
135            .context(PipelineSnafu)
136    }
137
138    async fn get_table(
139        &self,
140        table: &str,
141        query_ctx: &QueryContext,
142    ) -> std::result::Result<Option<Arc<Table>>, catalog::error::Error> {
143        let catalog = query_ctx.current_catalog();
144        let schema = query_ctx.current_schema();
145        self.catalog_manager
146            .table(catalog, &schema, table, None)
147            .await
148    }
149
150    fn build_pipeline(&self, pipeline: &str) -> ServerResult<Pipeline> {
151        PipelineOperator::build_pipeline(pipeline).context(PipelineSnafu)
152    }
153
154    async fn get_pipeline_str(
155        &self,
156        name: &str,
157        version: PipelineVersion,
158        query_ctx: QueryContextRef,
159    ) -> ServerResult<(String, TimestampNanosecond)> {
160        self.check_permission(&query_ctx, PermissionReq::Action(PIPELINE_QUERY))?;
161        self.pipeline_operator
162            .get_pipeline_str(name, version, query_ctx)
163            .await
164            .context(PipelineSnafu)
165    }
166}
167
168impl Instance {
169    pub async fn handle_log_inserts(
170        &self,
171        log: RowInsertRequests,
172        ctx: QueryContextRef,
173    ) -> ServerResult<Output> {
174        self.inserter
175            .handle_log_inserts(log, ctx, self.statement_executor.as_ref())
176            .await
177            .map_err(BoxedError::new)
178            .context(ExecuteGrpcRequestSnafu)
179    }
180
181    pub async fn handle_trace_inserts(
182        &self,
183        rows: RowInsertRequests,
184        ctx: QueryContextRef,
185    ) -> ServerResult<Output> {
186        self.inserter
187            .handle_trace_inserts(rows, ctx, self.statement_executor.as_ref())
188            .await
189            .map_err(BoxedError::new)
190            .context(ExecuteGrpcRequestSnafu)
191    }
192}
frontend/instance/log_handler.rs

frontend/instance/
log_handler.rs
