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 => catalog_guard.update_table(table),
302                            _ => panic!("receive an unsupported notify {:?}", resp),
303                        },
304                        PbObjectInfo::Source(source) => match resp.operation() {
305                            Operation::Add => catalog_guard.create_source(source),
306                            Operation::Delete => catalog_guard.drop_source(
307                                source.database_id,
308                                source.schema_id,
309                                source.id,
310                            ),
311                            Operation::Update => catalog_guard.update_source(source),
312                            _ => panic!("receive an unsupported notify {:?}", resp),
313                        },
314                        PbObjectInfo::Sink(sink) => match resp.operation() {
315                            Operation::Add => catalog_guard.create_sink(sink),
316                            Operation::Delete => {
317                                catalog_guard.drop_sink(sink.database_id, sink.schema_id, sink.id)
318                            }
319                            Operation::Update => catalog_guard.update_sink(sink),
320                            _ => panic!("receive an unsupported notify {:?}", resp),
321                        },
322                        PbObjectInfo::Subscription(subscription) => match resp.operation() {
323                            Operation::Add => catalog_guard.create_subscription(subscription),
324                            Operation::Delete => catalog_guard.drop_subscription(
325                                subscription.database_id,
326                                subscription.schema_id,
327                                subscription.id,
328                            ),
329                            Operation::Update => catalog_guard.update_subscription(subscription),
330                            _ => panic!("receive an unsupported notify {:?}", resp),
331                        },
332                        PbObjectInfo::Index(index) => match resp.operation() {
333                            Operation::Add => catalog_guard.create_index(index),
334                            Operation::Delete => catalog_guard.drop_index(
335                                index.database_id,
336                                index.schema_id,
337                                index.id,
338                            ),
339                            Operation::Update => catalog_guard.update_index(index),
340                            _ => panic!("receive an unsupported notify {:?}", resp),
341                        },
342                        PbObjectInfo::View(view) => match resp.operation() {
343                            Operation::Add => catalog_guard.create_view(view),
344                            Operation::Delete => {
345                                catalog_guard.drop_view(view.database_id, view.schema_id, view.id)
346                            }
347                            Operation::Update => catalog_guard.update_view(view),
348                            _ => panic!("receive an unsupported notify {:?}", resp),
349                        },
350                        ObjectInfo::Function(function) => match resp.operation() {
351                            Operation::Add => catalog_guard.create_function(function),
352                            Operation::Delete => catalog_guard.drop_function(
353                                function.database_id,
354                                function.schema_id,
355                                function.id,
356                            ),
357                            Operation::Update => catalog_guard.update_function(function),
358                            _ => panic!("receive an unsupported notify {:?}", resp),
359                        },
360                        ObjectInfo::Connection(connection) => match resp.operation() {
361                            Operation::Add => catalog_guard.create_connection(connection),
362                            Operation::Delete => catalog_guard.drop_connection(
363                                connection.database_id,
364                                connection.schema_id,
365                                connection.id,
366                            ),
367                            Operation::Update => catalog_guard.update_connection(connection),
368                            _ => panic!("receive an unsupported notify {:?}", resp),
369                        },
370                        ObjectInfo::Secret(secret) => {
371                            let mut secret = secret.clone();
372                            // The secret value should not be revealed to users. So mask it in the frontend catalog.
373                            secret.value =
374                                "SECRET VALUE SHOULD NOT BE REVEALED".as_bytes().to_vec();
375                            match resp.operation() {
376                                Operation::Add => catalog_guard.create_secret(&secret),
377                                Operation::Delete => catalog_guard.drop_secret(
378                                    secret.database_id,
379                                    secret.schema_id,
380                                    secret.id,
381                                ),
382                                Operation::Update => catalog_guard.update_secret(&secret),
383                                _ => panic!("receive an unsupported notify {:?}", resp),
384                            }
385                        }
386                    }
387                }
388            }
389            Info::Function(function) => match resp.operation() {
390                Operation::Add => catalog_guard.create_function(function),
391                Operation::Delete => catalog_guard.drop_function(
392                    function.database_id,
393                    function.schema_id,
394                    function.id,
395                ),
396                Operation::Update => catalog_guard.update_function(function),
397                _ => panic!("receive an unsupported notify {:?}", resp),
398            },
399            Info::Connection(connection) => match resp.operation() {
400                Operation::Add => catalog_guard.create_connection(connection),
401                Operation::Delete => catalog_guard.drop_connection(
402                    connection.database_id,
403                    connection.schema_id,
404                    connection.id,
405                ),
406                Operation::Update => catalog_guard.update_connection(connection),
407                _ => panic!("receive an unsupported notify {:?}", resp),
408            },
409            Info::Secret(secret) => {
410                let mut secret = secret.clone();
411                // The secret value should not be revealed to users. So mask it in the frontend catalog.
412                secret.value = "SECRET VALUE SHOULD NOT BE REVEALED".as_bytes().to_vec();
413                match resp.operation() {
414                    Operation::Add => catalog_guard.create_secret(&secret),
415                    Operation::Delete => {
416                        catalog_guard.drop_secret(secret.database_id, secret.schema_id, secret.id)
417                    }
418                    Operation::Update => catalog_guard.update_secret(&secret),
419                    _ => panic!("receive an unsupported notify {:?}", resp),
420                }
421            }
422            _ => unreachable!(),
423        }
424        assert!(
425            resp.version > self.version,
426            "resp version={:?}, current version={:?}",
427            resp.version,
428            self.version
429        );
430        self.version = resp.version;
431        self.catalog_updated_tx.send(resp.version).unwrap();
432    }
433
434    fn handle_user_notification(&mut self, resp: SubscribeResponse) {
435        let Some(info) = resp.info.as_ref() else {
436            return;
437        };
438
439        let mut user_guard = self.user_info_manager.write();
440        match info {
441            Info::User(user) => match resp.operation() {
442                Operation::Add => user_guard.create_user(user.clone()),
443                Operation::Delete => user_guard.drop_user(user.id),
444                Operation::Update => user_guard.update_user(user.clone()),
445                _ => panic!("receive an unsupported notify {:?}", resp),
446            },
447            _ => unreachable!(),
448        }
449        assert!(
450            resp.version > self.version,
451            "resp version={:?}, current version={:?}",
452            resp.version,
453            self.version
454        );
455        self.version = resp.version;
456        self.catalog_updated_tx.send(resp.version).unwrap();
457    }
458
459    fn handle_fragment_mapping_notification(&mut self, resp: SubscribeResponse) {
460        let Some(info) = resp.info.as_ref() else {
461            return;
462        };
463        match info {
464            Info::StreamingWorkerSlotMapping(streaming_worker_slot_mapping) => {
465                let fragment_id = streaming_worker_slot_mapping.fragment_id;
466                let mapping = || {
467                    WorkerSlotMapping::from_protobuf(
468                        streaming_worker_slot_mapping.mapping.as_ref().unwrap(),
469                    )
470                };
471
472                match resp.operation() {
473                    Operation::Add => {
474                        self.worker_node_manager
475                            .insert_streaming_fragment_mapping(fragment_id, mapping());
476                    }
477                    Operation::Delete => {
478                        self.worker_node_manager
479                            .remove_streaming_fragment_mapping(&fragment_id);
480                    }
481                    Operation::Update => {
482                        self.worker_node_manager
483                            .update_streaming_fragment_mapping(fragment_id, mapping());
484                    }
485                    _ => panic!("receive an unsupported notify {:?}", resp),
486                }
487            }
488            _ => unreachable!(),
489        }
490    }
491
492    fn handle_fragment_serving_mapping_notification(
493        &mut self,
494        mappings: Vec<FragmentWorkerSlotMapping>,
495        op: Operation,
496    ) {
497        match op {
498            Operation::Add | Operation::Update => {
499                self.worker_node_manager
500                    .upsert_serving_fragment_mapping(convert_worker_slot_mapping(&mappings));
501            }
502            Operation::Delete => self.worker_node_manager.remove_serving_fragment_mapping(
503                mappings
504                    .into_iter()
505                    .map(|m| m.fragment_id)
506                    .collect_vec()
507                    .as_slice(),
508            ),
509            Operation::Snapshot => {
510                self.worker_node_manager
511                    .set_serving_fragment_mapping(convert_worker_slot_mapping(&mappings));
512            }
513            _ => panic!("receive an unsupported notify {:?}", op),
514        }
515    }
516
517    /// Update max committed epoch in `HummockSnapshotManager`.
518    fn handle_hummock_snapshot_notification(&self, deltas: HummockVersionDeltas) {
519        self.hummock_snapshot_manager.update(deltas);
520    }
521
522    fn handle_secret_notification(&mut self, resp: SubscribeResponse) {
523        let resp_op = resp.operation();
524        let Some(Info::Secret(secret)) = resp.info else {
525            unreachable!();
526        };
527        match resp_op {
528            Operation::Add => {
529                LocalSecretManager::global().add_secret(secret.id, secret.value);
530            }
531            Operation::Delete => {
532                LocalSecretManager::global().remove_secret(secret.id);
533            }
534            Operation::Update => {
535                LocalSecretManager::global().update_secret(secret.id, secret.value);
536            }
537            _ => {
538                panic!("invalid notification operation: {resp_op:?}");
539            }
540        }
541    }
542
543    /// `update_worker_node_manager` is called in `start` method.
544    /// It calls `add_worker_node` and `remove_worker_node` of `WorkerNodeManager`.
545    fn update_worker_node_manager(&self, operation: Operation, node: WorkerNode) {
546        tracing::debug!(
547            "Update worker nodes, operation: {:?}, node: {:?}",
548            operation,
549            node
550        );
551
552        match operation {
553            Operation::Add => self.worker_node_manager.add_worker_node(node),
554            Operation::Delete => self.worker_node_manager.remove_worker_node(node),
555            _ => (),
556        }
557    }
558}
559
560fn convert_worker_slot_mapping(
561    worker_slot_mappings: &[FragmentWorkerSlotMapping],
562) -> HashMap<FragmentId, WorkerSlotMapping> {
563    worker_slot_mappings
564        .iter()
565        .map(
566            |FragmentWorkerSlotMapping {
567                 fragment_id,
568                 mapping,
569             }| {
570                let mapping = WorkerSlotMapping::from_protobuf(mapping.as_ref().unwrap());
571                (*fragment_id, mapping)
572            },
573        )
574        .collect()
575}
risingwave_frontend/observer/observer_manager.rs

risingwave_frontend/observer/
observer_manager.rs
