risingwave_frontend/observer/
observer_manager.rs

1// Copyright 2022 RisingWave Labs
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::collections::HashMap;
16use std::sync::Arc;
17
18use itertools::Itertools;
19use parking_lot::RwLock;
20use risingwave_batch::worker_manager::worker_node_manager::WorkerNodeManagerRef;
21use risingwave_common::catalog::CatalogVersion;
22use risingwave_common::hash::WorkerSlotMapping;
23use risingwave_common::license::LicenseManager;
24use risingwave_common::secret::LocalSecretManager;
25use risingwave_common::session_config::SessionConfig;
26use risingwave_common::system_param::local_manager::LocalSystemParamsManagerRef;
27use risingwave_common_service::ObserverState;
28use risingwave_hummock_sdk::FrontendHummockVersion;
29use risingwave_pb::common::WorkerNode;
30use risingwave_pb::hummock::{HummockVersionDeltas, HummockVersionStats};
31use risingwave_pb::meta::object::{ObjectInfo, PbObjectInfo};
32use risingwave_pb::meta::subscribe_response::{Info, Operation};
33use risingwave_pb::meta::{FragmentWorkerSlotMapping, MetaSnapshot, SubscribeResponse};
34use risingwave_rpc_client::ComputeClientPoolRef;
35use tokio::sync::watch::Sender;
36
37use crate::catalog::FragmentId;
38use crate::catalog::root_catalog::Catalog;
39use crate::scheduler::HummockSnapshotManagerRef;
40use crate::user::user_manager::UserInfoManager;
41
42pub struct FrontendObserverNode {
43    worker_node_manager: WorkerNodeManagerRef,
44    version: CatalogVersion,
45    catalog_updated_tx: Sender<CatalogVersion>,
46    // When both locks are needed, always acquire `catalog` before `user_info_manager`.
47    catalog: Arc<RwLock<Catalog>>,
48    user_info_manager: Arc<RwLock<UserInfoManager>>,
49    hummock_snapshot_manager: HummockSnapshotManagerRef,
50    system_params_manager: LocalSystemParamsManagerRef,
51    session_params: Arc<RwLock<SessionConfig>>,
52    compute_client_pool: ComputeClientPoolRef,
53}
54
55impl ObserverState for FrontendObserverNode {
56    fn subscribe_type() -> risingwave_pb::meta::SubscribeType {
57        risingwave_pb::meta::SubscribeType::Frontend
58    }
59
60    fn handle_notification(&mut self, resp: SubscribeResponse) {
61        let Some(info) = resp.info.as_ref() else {
62            return;
63        };
64
65        // TODO: this clone can be avoided
66        match info.to_owned() {
67            Info::Database(_)
68            | Info::Schema(_)
69            | Info::ObjectGroup(_)
70            | Info::Function(_)
71            | Info::Connection(_) => {
72                self.handle_catalog_notification(resp);
73            }
74            Info::Secret(_) => {
75                self.handle_catalog_notification(resp.clone());
76                self.handle_secret_notification(resp);
77            }
78            Info::Node(node) => {
79                self.update_worker_node_manager(resp.operation(), node);
80            }
81            Info::User(_) => {
82                self.handle_user_notification(resp);
83            }
84            Info::Snapshot(_) => {
85                panic!(
86                    "receiving a snapshot in the middle is unsupported now {:?}",
87                    resp
88                )
89            }
90            Info::HummockVersionDeltas(deltas) => {
91                self.handle_hummock_snapshot_notification(deltas);
92            }
93            Info::MetaBackupManifestId(_) => {
94                panic!("frontend node should not receive MetaBackupManifestId");
95            }
96            Info::HummockWriteLimits(_) => {
97                panic!("frontend node should not receive HummockWriteLimits");
98            }
99            Info::SystemParams(p) => {
100                self.system_params_manager.try_set_params(p);
101            }
102            Info::SessionParam(p) => {
103                self.session_params
104                    .write()
105                    .set(&p.param, p.value().to_owned(), &mut ())
106                    .unwrap();
107            }
108            Info::HummockStats(stats) => {
109                self.handle_table_stats_notification(stats);
110            }
111            Info::StreamingWorkerSlotMapping(_) => self.handle_fragment_mapping_notification(resp),
112            Info::ServingWorkerSlotMappings(m) => {
113                self.handle_fragment_serving_mapping_notification(m.mappings, resp.operation())
114            }
115            Info::Recovery(_) => {
116                self.compute_client_pool.invalidate_all();
117            }
118            Info::ClusterResource(resource) => {
119                LicenseManager::get().update_cluster_resource(resource);
120            }
121            Info::TableRefillRuntimeConfig(_) => {
122                panic!("frontend node should not receive TableRefillRuntimeConfig");
123            }
124        }
125    }
126
127    fn handle_initialization_notification(&mut self, resp: SubscribeResponse) {
128        let mut catalog_guard = self.catalog.write();
129        let mut user_guard = self.user_info_manager.write();
130        catalog_guard.clear();
131        user_guard.clear();
132
133        let Some(Info::Snapshot(snapshot)) = resp.info else {
134            unreachable!();
135        };
136        let MetaSnapshot {
137            databases,
138            schemas,
139            sources,
140            sinks,
141            tables,
142            indexes,
143            views,
144            subscriptions,
145            functions,
146            connections,
147            users,
148            nodes,
149            hummock_version,
150            meta_backup_manifest_id: _,
151            hummock_write_limits: _,
152            streaming_worker_slot_mappings,
153            serving_worker_slot_mappings,
154            session_params,
155            version,
156            secrets,
157            cluster_resource,
158            object_dependencies,
159            table_refill_runtime_config: _,
160        } = snapshot;
161
162        for db in databases {
163            catalog_guard.create_database(&db)
164        }
165        for schema in schemas {
166            catalog_guard.create_schema(&schema)
167        }
168        for source in sources {
169            catalog_guard.create_source(&source)
170        }
171        for sink in sinks {
172            catalog_guard.create_sink(&sink)
173        }
174        for subscription in subscriptions {
175            catalog_guard.create_subscription(&subscription)
176        }
177        for table in tables {
178            catalog_guard.create_table(&table)
179        }
180        for index in indexes {
181            catalog_guard.create_index(&index)
182        }
183        for view in views {
184            catalog_guard.create_view(&view)
185        }
186        for function in functions {
187            catalog_guard.create_function(&function)
188        }
189        for connection in connections {
190            catalog_guard.create_connection(&connection)
191        }
192        for secret in &secrets {
193            catalog_guard.create_secret(secret)
194        }
195        catalog_guard.set_object_dependencies(object_dependencies);
196        for user in users {
197            user_guard.create_user(user)
198        }
199
200        self.worker_node_manager.refresh(
201            nodes,
202            convert_worker_slot_mapping(&streaming_worker_slot_mappings),
203            convert_worker_slot_mapping(&serving_worker_slot_mappings),
204        );
205        self.hummock_snapshot_manager
206            .init(FrontendHummockVersion::from_protobuf(
207                hummock_version.unwrap(),
208            ));
209
210        let snapshot_version = version.unwrap();
211        self.version = snapshot_version.catalog_version;
212        self.catalog_updated_tx
213            .send(snapshot_version.catalog_version)
214            .unwrap();
215        *self.session_params.write() =
216            serde_json::from_str(&session_params.unwrap().params).unwrap();
217        LocalSecretManager::global().init_secrets(secrets);
218        LicenseManager::get().update_cluster_resource(cluster_resource.unwrap());
219    }
220}
221
222impl FrontendObserverNode {
223    pub fn new(
224        worker_node_manager: WorkerNodeManagerRef,
225        catalog: Arc<RwLock<Catalog>>,
226        catalog_updated_tx: Sender<CatalogVersion>,
227        user_info_manager: Arc<RwLock<UserInfoManager>>,
228        hummock_snapshot_manager: HummockSnapshotManagerRef,
229        system_params_manager: LocalSystemParamsManagerRef,
230        session_params: Arc<RwLock<SessionConfig>>,
231        compute_client_pool: ComputeClientPoolRef,
232    ) -> Self {
233        Self {
234            version: 0,
235            worker_node_manager,
236            catalog,
237            catalog_updated_tx,
238            user_info_manager,
239            hummock_snapshot_manager,
240            system_params_manager,
241            session_params,
242            compute_client_pool,
243        }
244    }
245
246    fn handle_table_stats_notification(&mut self, table_stats: HummockVersionStats) {
247        let mut catalog_guard = self.catalog.write();
248        catalog_guard.set_table_stats(table_stats);
249    }
250
251    fn handle_catalog_notification(&mut self, resp: SubscribeResponse) {
252        let Some(info) = resp.info.as_ref() else {
253            return;
254        };
255        tracing::trace!(op = ?resp.operation(), ?info, "handle catalog notification");
256
257        let mut catalog_guard = self.catalog.write();
258        match info {
259            Info::Database(database) => match resp.operation() {
260                Operation::Add => catalog_guard.create_database(database),
261                Operation::Delete => catalog_guard.drop_database(database.id),
262                Operation::Update => catalog_guard.update_database(database),
263                _ => panic!("receive an unsupported notify {:?}", resp),
264            },
265            Info::Schema(schema) => match resp.operation() {
266                Operation::Add => catalog_guard.create_schema(schema),
267                Operation::Delete => catalog_guard.drop_schema(schema.database_id, schema.id),
268                Operation::Update => catalog_guard.update_schema(schema),
269                _ => panic!("receive an unsupported notify {:?}", resp),
270            },
271            Info::ObjectGroup(object_group) => {
272                if !object_group.dependencies.is_empty() {
273                    catalog_guard.insert_object_dependencies(object_group.dependencies.clone());
274                }
275                for object in &object_group.objects {
276                    let Some(obj) = object.object_info.as_ref() else {
277                        continue;
278                    };
279                    match obj {
280                        ObjectInfo::Database(db) => match resp.operation() {
281                            Operation::Add => catalog_guard.create_database(db),
282                            Operation::Delete => catalog_guard.drop_database(db.id),
283                            Operation::Update => catalog_guard.update_database(db),
284                            _ => panic!("receive an unsupported notify {:?}", resp),
285                        },
286                        ObjectInfo::Schema(schema) => match resp.operation() {
287                            Operation::Add => catalog_guard.create_schema(schema),
288                            Operation::Delete => {
289                                catalog_guard.drop_schema(schema.database_id, schema.id)
290                            }
291                            Operation::Update => catalog_guard.update_schema(schema),
292                            _ => panic!("receive an unsupported notify {:?}", resp),
293                        },
294                        PbObjectInfo::Table(table) => match resp.operation() {
295                            Operation::Add => catalog_guard.create_table(table),
296                            Operation::Delete => catalog_guard.drop_table(
297                                table.database_id,
298                                table.schema_id,
299                                table.id,
300                            ),
301                            Operation::Update => {
302                                let old_fragment_id = catalog_guard
303                                    .get_any_table_by_id(table.id)
304                                    .unwrap()
305                                    .fragment_id;
306                                catalog_guard.update_table(table);
307                                if old_fragment_id != table.fragment_id {
308                                    // FIXME: the frontend node delete its fragment for the update
309                                    // operation by itself.
310                                    self.worker_node_manager
311                                        .remove_streaming_fragment_mapping(&old_fragment_id);
312                                }
313                            }
314                            _ => panic!("receive an unsupported notify {:?}", resp),
315                        },
316                        PbObjectInfo::Source(source) => match resp.operation() {
317                            Operation::Add => catalog_guard.create_source(source),
318                            Operation::Delete => catalog_guard.drop_source(
319                                source.database_id,
320                                source.schema_id,
321                                source.id,
322                            ),
323                            Operation::Update => catalog_guard.update_source(source),
324                            _ => panic!("receive an unsupported notify {:?}", resp),
325                        },
326                        PbObjectInfo::Sink(sink) => match resp.operation() {
327                            Operation::Add => catalog_guard.create_sink(sink),
328                            Operation::Delete => {
329                                catalog_guard.drop_sink(sink.database_id, sink.schema_id, sink.id)
330                            }
331                            Operation::Update => catalog_guard.update_sink(sink),
332                            _ => panic!("receive an unsupported notify {:?}", resp),
333                        },
334                        PbObjectInfo::Subscription(subscription) => match resp.operation() {
335                            Operation::Add => catalog_guard.create_subscription(subscription),
336                            Operation::Delete => catalog_guard.drop_subscription(
337                                subscription.database_id,
338                                subscription.schema_id,
339                                subscription.id,
340                            ),
341                            Operation::Update => catalog_guard.update_subscription(subscription),
342                            _ => panic!("receive an unsupported notify {:?}", resp),
343                        },
344                        PbObjectInfo::Index(index) => match resp.operation() {
345                            Operation::Add => catalog_guard.create_index(index),
346                            Operation::Delete => catalog_guard.drop_index(
347                                index.database_id,
348                                index.schema_id,
349                                index.id,
350                            ),
351                            Operation::Update => catalog_guard.update_index(index),
352                            _ => panic!("receive an unsupported notify {:?}", resp),
353                        },
354                        PbObjectInfo::View(view) => match resp.operation() {
355                            Operation::Add => catalog_guard.create_view(view),
356                            Operation::Delete => {
357                                catalog_guard.drop_view(view.database_id, view.schema_id, view.id)
358                            }
359                            Operation::Update => catalog_guard.update_view(view),
360                            _ => panic!("receive an unsupported notify {:?}", resp),
361                        },
362                        ObjectInfo::Function(function) => match resp.operation() {
363                            Operation::Add => catalog_guard.create_function(function),
364                            Operation::Delete => catalog_guard.drop_function(
365                                function.database_id,
366                                function.schema_id,
367                                function.id,
368                            ),
369                            Operation::Update => catalog_guard.update_function(function),
370                            _ => panic!("receive an unsupported notify {:?}", resp),
371                        },
372                        ObjectInfo::Connection(connection) => match resp.operation() {
373                            Operation::Add => catalog_guard.create_connection(connection),
374                            Operation::Delete => catalog_guard.drop_connection(
375                                connection.database_id,
376                                connection.schema_id,
377                                connection.id,
378                            ),
379                            Operation::Update => catalog_guard.update_connection(connection),
380                            _ => panic!("receive an unsupported notify {:?}", resp),
381                        },
382                        ObjectInfo::Secret(secret) => {
383                            let mut secret = secret.clone();
384                            // The secret value should not be revealed to users. So mask it in the frontend catalog.
385                            secret.value =
386                                "SECRET VALUE SHOULD NOT BE REVEALED".as_bytes().to_vec();
387                            match resp.operation() {
388                                Operation::Add => catalog_guard.create_secret(&secret),
389                                Operation::Delete => catalog_guard.drop_secret(
390                                    secret.database_id,
391                                    secret.schema_id,
392                                    secret.id,
393                                ),
394                                Operation::Update => catalog_guard.update_secret(&secret),
395                                _ => panic!("receive an unsupported notify {:?}", resp),
396                            }
397                        }
398                    }
399                }
400            }
401            Info::Function(function) => match resp.operation() {
402                Operation::Add => catalog_guard.create_function(function),
403                Operation::Delete => catalog_guard.drop_function(
404                    function.database_id,
405                    function.schema_id,
406                    function.id,
407                ),
408                Operation::Update => catalog_guard.update_function(function),
409                _ => panic!("receive an unsupported notify {:?}", resp),
410            },
411            Info::Connection(connection) => match resp.operation() {
412                Operation::Add => catalog_guard.create_connection(connection),
413                Operation::Delete => catalog_guard.drop_connection(
414                    connection.database_id,
415                    connection.schema_id,
416                    connection.id,
417                ),
418                Operation::Update => catalog_guard.update_connection(connection),
419                _ => panic!("receive an unsupported notify {:?}", resp),
420            },
421            Info::Secret(secret) => {
422                let mut secret = secret.clone();
423                // The secret value should not be revealed to users. So mask it in the frontend catalog.
424                secret.value = "SECRET VALUE SHOULD NOT BE REVEALED".as_bytes().to_vec();
425                match resp.operation() {
426                    Operation::Add => catalog_guard.create_secret(&secret),
427                    Operation::Delete => {
428                        catalog_guard.drop_secret(secret.database_id, secret.schema_id, secret.id)
429                    }
430                    Operation::Update => catalog_guard.update_secret(&secret),
431                    _ => panic!("receive an unsupported notify {:?}", resp),
432                }
433            }
434            _ => unreachable!(),
435        }
436        assert!(
437            resp.version > self.version,
438            "resp version={:?}, current version={:?}",
439            resp.version,
440            self.version
441        );
442        self.version = resp.version;
443        self.catalog_updated_tx.send(resp.version).unwrap();
444    }
445
446    fn handle_user_notification(&mut self, resp: SubscribeResponse) {
447        let Some(info) = resp.info.as_ref() else {
448            return;
449        };
450
451        let mut user_guard = self.user_info_manager.write();
452        match info {
453            Info::User(user) => match resp.operation() {
454                Operation::Add => user_guard.create_user(user.clone()),
455                Operation::Delete => user_guard.drop_user(user.id),
456                Operation::Update => user_guard.update_user(user.clone()),
457                _ => panic!("receive an unsupported notify {:?}", resp),
458            },
459            _ => unreachable!(),
460        }
461        assert!(
462            resp.version > self.version,
463            "resp version={:?}, current version={:?}",
464            resp.version,
465            self.version
466        );
467        self.version = resp.version;
468        self.catalog_updated_tx.send(resp.version).unwrap();
469    }
470
471    fn handle_fragment_mapping_notification(&mut self, resp: SubscribeResponse) {
472        let Some(info) = resp.info.as_ref() else {
473            return;
474        };
475        match info {
476            Info::StreamingWorkerSlotMapping(streaming_worker_slot_mapping) => {
477                let fragment_id = streaming_worker_slot_mapping.fragment_id;
478                let mapping = || {
479                    WorkerSlotMapping::from_protobuf(
480                        streaming_worker_slot_mapping.mapping.as_ref().unwrap(),
481                    )
482                };
483
484                match resp.operation() {
485                    Operation::Add => {
486                        self.worker_node_manager
487                            .insert_streaming_fragment_mapping(fragment_id, mapping());
488                    }
489                    Operation::Delete => {
490                        self.worker_node_manager
491                            .remove_streaming_fragment_mapping(&fragment_id);
492                    }
493                    Operation::Update => {
494                        self.worker_node_manager
495                            .update_streaming_fragment_mapping(fragment_id, mapping());
496                    }
497                    _ => panic!("receive an unsupported notify {:?}", resp),
498                }
499            }
500            _ => unreachable!(),
501        }
502    }
503
504    fn handle_fragment_serving_mapping_notification(
505        &mut self,
506        mappings: Vec<FragmentWorkerSlotMapping>,
507        op: Operation,
508    ) {
509        match op {
510            Operation::Add | Operation::Update => {
511                self.worker_node_manager
512                    .upsert_serving_fragment_mapping(convert_worker_slot_mapping(&mappings));
513            }
514            Operation::Delete => self.worker_node_manager.remove_serving_fragment_mapping(
515                mappings
516                    .into_iter()
517                    .map(|m| m.fragment_id)
518                    .collect_vec()
519                    .as_slice(),
520            ),
521            Operation::Snapshot => {
522                self.worker_node_manager
523                    .set_serving_fragment_mapping(convert_worker_slot_mapping(&mappings));
524            }
525            _ => panic!("receive an unsupported notify {:?}", op),
526        }
527    }
528
529    /// Update max committed epoch in `HummockSnapshotManager`.
530    fn handle_hummock_snapshot_notification(&self, deltas: HummockVersionDeltas) {
531        self.hummock_snapshot_manager.update(deltas);
532    }
533
534    fn handle_secret_notification(&mut self, resp: SubscribeResponse) {
535        let resp_op = resp.operation();
536        let Some(Info::Secret(secret)) = resp.info else {
537            unreachable!();
538        };
539        match resp_op {
540            Operation::Add => {
541                LocalSecretManager::global().add_secret(secret.id, secret.value);
542            }
543            Operation::Delete => {
544                LocalSecretManager::global().remove_secret(secret.id);
545            }
546            Operation::Update => {
547                LocalSecretManager::global().update_secret(secret.id, secret.value);
548            }
549            _ => {
550                panic!("invalid notification operation: {resp_op:?}");
551            }
552        }
553    }
554
555    /// `update_worker_node_manager` is called in `start` method.
556    /// It calls `add_worker_node` and `remove_worker_node` of `WorkerNodeManager`.
557    fn update_worker_node_manager(&self, operation: Operation, node: WorkerNode) {
558        tracing::debug!(
559            "Update worker nodes, operation: {:?}, node: {:?}",
560            operation,
561            node
562        );
563
564        match operation {
565            Operation::Add => self.worker_node_manager.add_worker_node(node),
566            Operation::Delete => self.worker_node_manager.remove_worker_node(node),
567            _ => (),
568        }
569    }
570}
571
572fn convert_worker_slot_mapping(
573    worker_slot_mappings: &[FragmentWorkerSlotMapping],
574) -> HashMap<FragmentId, WorkerSlotMapping> {
575    worker_slot_mappings
576        .iter()
577        .map(
578            |FragmentWorkerSlotMapping {
579                 fragment_id,
580                 mapping,
581             }| {
582                let mapping = WorkerSlotMapping::from_protobuf(mapping.as_ref().unwrap());
583                (*fragment_id, mapping)
584            },
585        )
586        .collect()
587}
risingwave_frontend/observer/observer_manager.rs

risingwave_frontend/observer/
observer_manager.rs
