risingwave_compute/rpc/service/
config_service.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::sync::Arc;
16
17use foyer::HybridCache;
18use risingwave_batch::task::BatchManager;
19use risingwave_common::error::tonic::ToTonicStatus;
20use risingwave_hummock_sdk::HummockSstableObjectId;
21use risingwave_pb::compute::config_service_server::ConfigService;
22use risingwave_pb::compute::{
23    ResizeCacheRequest, ResizeCacheResponse, ShowConfigRequest, ShowConfigResponse,
24};
25use risingwave_storage::hummock::{Block, Sstable, SstableBlockIndex};
26use risingwave_stream::task::LocalStreamManager;
27use thiserror_ext::AsReport;
28use tonic::{Code, Request, Response, Status};
29
30pub struct ConfigServiceImpl {
31    batch_mgr: Arc<BatchManager>,
32    stream_mgr: LocalStreamManager,
33    meta_cache: Option<HybridCache<HummockSstableObjectId, Box<Sstable>>>,
34    block_cache: Option<HybridCache<SstableBlockIndex, Box<Block>>>,
35}
36
37#[async_trait::async_trait]
38impl ConfigService for ConfigServiceImpl {
39    async fn show_config(
40        &self,
41        _request: Request<ShowConfigRequest>,
42    ) -> Result<Response<ShowConfigResponse>, Status> {
43        let batch_config = serde_json::to_string(self.batch_mgr.config())
44            .map_err(|e| e.to_status(Code::Internal, "compute"))?;
45        // TODO(config): show overridden config for specific job
46        let stream_config = serde_json::to_string(&self.stream_mgr.env.global_config())
47            .map_err(|e| e.to_status(Code::Internal, "compute"))?;
48
49        let show_config_response = ShowConfigResponse {
50            batch_config,
51            stream_config,
52        };
53        Ok(Response::new(show_config_response))
54    }
55
56    async fn resize_cache(
57        &self,
58        request: Request<ResizeCacheRequest>,
59    ) -> Result<Response<ResizeCacheResponse>, Status> {
60        let req = request.into_inner();
61
62        // A zero capacity means "do not resize" for backward compatibility. The clear flags
63        // independently trigger HybridCache::clear() and never change the configured capacity.
64        if let Some(meta_cache) = &self.meta_cache {
65            if req.clear_meta_cache {
66                meta_cache
67                    .clear()
68                    .await
69                    .map_err(|e| Status::internal(e.to_report_string()))?;
70                tracing::info!("clear meta file cache");
71            }
72
73            if req.meta_cache_capacity > 0 {
74                match meta_cache.memory().resize(req.meta_cache_capacity as _) {
75                    Ok(_) => tracing::info!(
76                        "resize meta cache capacity to {:?}",
77                        req.meta_cache_capacity
78                    ),
79                    Err(e) => return Err(Status::internal(e.to_report_string())),
80                }
81            }
82        }
83
84        if let Some(block_cache) = &self.block_cache {
85            if req.clear_data_cache {
86                block_cache
87                    .clear()
88                    .await
89                    .map_err(|e| Status::internal(e.to_report_string()))?;
90                tracing::info!("clear data file cache");
91            }
92
93            if req.data_cache_capacity > 0 {
94                match block_cache.memory().resize(req.data_cache_capacity as _) {
95                    Ok(_) => tracing::info!(
96                        "resize data cache capacity to {:?}",
97                        req.data_cache_capacity
98                    ),
99                    Err(e) => return Err(Status::internal(e.to_report_string())),
100                }
101            }
102        }
103
104        Ok(Response::new(ResizeCacheResponse {}))
105    }
106}
107
108impl ConfigServiceImpl {
109    pub fn new(
110        batch_mgr: Arc<BatchManager>,
111        stream_mgr: LocalStreamManager,
112        meta_cache: Option<HybridCache<HummockSstableObjectId, Box<Sstable>>>,
113        block_cache: Option<HybridCache<SstableBlockIndex, Box<Block>>>,
114    ) -> Self {
115        Self {
116            batch_mgr,
117            stream_mgr,
118            meta_cache,
119            block_cache,
120        }
121    }
122}
risingwave_compute/rpc/service/config_service.rs

risingwave_compute/rpc/service/
config_service.rs
