common_meta/ddl/
drop_flow.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
15mod metadata;
16
17use api::v1::flow::{DropRequest, FlowRequest, flow_request};
18use async_trait::async_trait;
19use common_catalog::format_full_flow_name;
20use common_error::ext::ErrorExt;
21use common_error::status_code::StatusCode;
22use common_procedure::error::{FromJsonSnafu, ToJsonSnafu};
23use common_procedure::{
24    Context as ProcedureContext, EventContext, EventTrigger, LockKey, Procedure,
25    Result as ProcedureResult, Status,
26};
27use common_telemetry::info;
28use futures::future::join_all;
29use serde::{Deserialize, Serialize};
30use snafu::{ResultExt, ensure};
31use strum::AsRefStr;
32
33use crate::cache_invalidator::Context;
34use crate::ddl::DdlContext;
35use crate::ddl::event::flow::{DROP_FLOW_EVENT_TYPE, FlowDdlEvent};
36use crate::ddl::utils::{add_peer_context_if_needed, map_to_procedure_error};
37use crate::error::{self, Result};
38use crate::flow_name::FlowName;
39use crate::instruction::{CacheIdent, DropFlow};
40use crate::key::flow::flow_info::FlowInfoValue;
41use crate::key::flow::flow_route::FlowRouteValue;
42use crate::lock_key::{CatalogLock, FlowLock};
43use crate::metrics;
44use crate::rpc::ddl::DropFlowTask;
45
46/// The procedure for dropping a flow.
47pub struct DropFlowProcedure {
48    /// The context of procedure runtime.
49    pub(crate) context: DdlContext,
50    /// The serializable data.
51    pub(crate) data: DropFlowData,
52}
53
54impl DropFlowProcedure {
55    pub const TYPE_NAME: &'static str = "metasrv-procedure::DropFlow";
56
57    pub fn new(task: DropFlowTask, context: DdlContext) -> Self {
58        Self {
59            context,
60            data: DropFlowData {
61                state: DropFlowState::Prepare,
62                task,
63                flow_info_value: None,
64                flow_route_values: vec![],
65            },
66        }
67    }
68
69    pub fn from_json(json: &str, context: DdlContext) -> ProcedureResult<Self> {
70        let data: DropFlowData = serde_json::from_str(json).context(FromJsonSnafu)?;
71
72        Ok(Self { context, data })
73    }
74
75    /// Checks whether flow exists.
76    /// - Early returns if flow not exists and `drop_if_exists` is `true`.
77    /// - Throws an error if flow not exists and `drop_if_exists` is `false`.
78    pub(crate) async fn on_prepare(&mut self) -> Result<Status> {
79        let catalog_name = &self.data.task.catalog_name;
80        let flow_name = &self.data.task.flow_name;
81        let exists = self
82            .context
83            .flow_metadata_manager
84            .flow_name_manager()
85            .exists(catalog_name, flow_name)
86            .await?;
87
88        if !exists && self.data.task.drop_if_exists {
89            return Ok(Status::done());
90        }
91
92        ensure!(
93            exists,
94            error::FlowNotFoundSnafu {
95                flow_name: format_full_flow_name(catalog_name, flow_name)
96            }
97        );
98
99        self.fill_flow_metadata().await?;
100        self.data.state = DropFlowState::DeleteMetadata;
101        Ok(Status::executing(true))
102    }
103
104    async fn on_flownode_drop_flows(&self) -> Result<Status> {
105        // Safety: checked
106        let flownode_ids = &self.data.flow_info_value.as_ref().unwrap().flownode_ids;
107        let flow_id = self.data.task.flow_id;
108        let mut drop_flow_tasks = Vec::with_capacity(flownode_ids.len());
109
110        for FlowRouteValue { peer } in &self.data.flow_route_values {
111            let requester = self.context.node_manager.flownode(peer).await;
112            let request = FlowRequest {
113                body: Some(flow_request::Body::Drop(DropRequest {
114                    flow_id: Some(api::v1::FlowId { id: flow_id }),
115                })),
116                ..Default::default()
117            };
118
119            drop_flow_tasks.push(async move {
120                if let Err(err) = requester.handle(request).await
121                    && err.status_code() != StatusCode::FlowNotFound
122                {
123                    return Err(add_peer_context_if_needed(peer.clone())(err));
124                }
125                Ok(())
126            });
127        }
128
129        join_all(drop_flow_tasks)
130            .await
131            .into_iter()
132            .collect::<Result<Vec<_>>>()?;
133
134        Ok(Status::done())
135    }
136
137    async fn on_delete_metadata(&mut self) -> Result<Status> {
138        let flow_id = self.data.task.flow_id;
139        self.context
140            .flow_metadata_manager
141            .destroy_flow_metadata(
142                flow_id,
143                // Safety: checked
144                self.data.flow_info_value.as_ref().unwrap(),
145            )
146            .await?;
147        info!("Deleted flow metadata for flow {flow_id}");
148        self.data.state = DropFlowState::InvalidateFlowCache;
149        Ok(Status::executing(true))
150    }
151
152    async fn on_broadcast(&mut self) -> Result<Status> {
153        let flow_id = self.data.task.flow_id;
154        let ctx = Context {
155            subject: Some("Invalidate flow cache by dropping flow".to_string()),
156        };
157        let flow_info_value = self.data.flow_info_value.as_ref().unwrap();
158
159        let flow_part2nodes = flow_info_value
160            .flownode_ids()
161            .clone()
162            .into_iter()
163            .collect::<Vec<_>>();
164
165        self.context
166            .cache_invalidator
167            .invalidate(
168                &ctx,
169                &[
170                    CacheIdent::FlowId(flow_id),
171                    CacheIdent::FlowName(FlowName {
172                        catalog_name: flow_info_value.catalog_name.clone(),
173                        flow_name: flow_info_value.flow_name.clone(),
174                    }),
175                    CacheIdent::DropFlow(DropFlow {
176                        flow_id,
177                        source_table_ids: flow_info_value.source_table_ids.clone(),
178                        flow_part2node_id: flow_part2nodes,
179                    }),
180                ],
181            )
182            .await?;
183        self.data.state = DropFlowState::DropFlows;
184        Ok(Status::executing(true))
185    }
186}
187
188#[async_trait]
189impl Procedure for DropFlowProcedure {
190    fn type_name(&self) -> &str {
191        Self::TYPE_NAME
192    }
193
194    async fn execute(&mut self, _ctx: &ProcedureContext) -> ProcedureResult<Status> {
195        let state = &self.data.state;
196        let _timer = metrics::METRIC_META_PROCEDURE_DROP_FLOW
197            .with_label_values(&[state.as_ref()])
198            .start_timer();
199
200        match self.data.state {
201            DropFlowState::Prepare => self.on_prepare().await,
202            DropFlowState::DeleteMetadata => self.on_delete_metadata().await,
203            DropFlowState::InvalidateFlowCache => self.on_broadcast().await,
204            DropFlowState::DropFlows => self.on_flownode_drop_flows().await,
205        }
206        .map_err(map_to_procedure_error)
207    }
208
209    fn dump(&self) -> ProcedureResult<String> {
210        serde_json::to_string(&self.data).context(ToJsonSnafu)
211    }
212
213    fn lock_key(&self) -> LockKey {
214        let catalog_name = &self.data.task.catalog_name;
215        let flow_id = self.data.task.flow_id;
216
217        let lock_key = vec![
218            CatalogLock::Read(catalog_name).into(),
219            FlowLock::Write(flow_id).into(),
220        ];
221
222        LockKey::new(lock_key)
223    }
224
225    fn event(&self, ctx: &EventContext<'_>) -> Option<Box<dyn common_event_recorder::Event>> {
226        if !ctx.event_type_filter.allows(DROP_FLOW_EVENT_TYPE) {
227            return None;
228        }
229
230        let event = match &ctx.trigger {
231            EventTrigger::Submitted => FlowDdlEvent::drop_submitted(
232                &self.data.task.catalog_name,
233                &self.data.task.flow_name,
234                self.data.task.flow_id,
235                self.data.task.drop_if_exists,
236            ),
237            _ => FlowDdlEvent::drop_lifecycle(
238                &self.data.task.catalog_name,
239                &self.data.task.flow_name,
240                self.data.task.flow_id,
241            ),
242        };
243
244        Some(Box::new(event))
245    }
246}
247
248/// The serializable data
249#[derive(Debug, Serialize, Deserialize)]
250pub(crate) struct DropFlowData {
251    state: DropFlowState,
252    task: DropFlowTask,
253    pub(crate) flow_info_value: Option<FlowInfoValue>,
254    pub(crate) flow_route_values: Vec<FlowRouteValue>,
255}
256
257/// The state of drop flow
258#[derive(Debug, Serialize, Deserialize, AsRefStr, PartialEq)]
259enum DropFlowState {
260    /// Prepares to drop the flow
261    Prepare,
262    /// Deletes metadata
263    DeleteMetadata,
264    /// Invalidate flow cache
265    InvalidateFlowCache,
266    /// Drop flows on flownode
267    DropFlows,
268    // TODO(weny): support to rollback
269}
common_meta/ddl/drop_flow.rs

common_meta/ddl/
drop_flow.rs
