feat:支持配置无感解密

Author: chuntaojun

.DS_Store

@@ -33,7 +33,7 @@ tokio-stream = {version = "0.1.16"}
 tower = {version = "0.4.13"}
 
 # gRPC dep
-futures = "0.3.30"
+futures = {version = "0.3.30"}
 once_cell = {version = "1.19.0"}
 prost = {version = "0.12.4"}
 prost-build = {version = "0.12.4"}
@@ -43,6 +43,14 @@ tonic = {version = "0.11.0"}
 # logging
 tracing = {version = "0.1.36"}
 
+# crypto
+aes = {version = "0.7.4"}
+base64 = {version = "0.22.1"}
+block-modes = {version = "0.8.1"}
+hex = {version = "0.4.3"}
+rand = {version = "0.8.4"}
+rsa = {version = "0.9.6"}
+
 [dev-dependencies]
 tracing-subscriber = {version = "0.3", features = ["default"]}
 

Cargo.toml

@@ -24,9 +24,9 @@ use crate::{
 };
 
 use super::req::{
-    DeleteConfigFileRequest, GetConfigFileRequest, GetConfigGroupRequest, PublishConfigFileRequest,
-    UpdateConfigFileRequest, WatchConfigFileRequest, WatchConfigFileResponse,
-    WatchConfigGroupRequest, WatchConfigGroupResponse,
+    CreateConfigFileRequest, DeleteConfigFileRequest, GetConfigFileRequest, GetConfigGroupRequest,
+    PublishConfigFileRequest, UpdateConfigFileRequest, WatchConfigFileRequest,
+    WatchConfigFileResponse, WatchConfigGroupRequest, WatchConfigGroupResponse,
 };
 
 /// new_config_file_api
@@ -57,9 +57,9 @@ where
 {
     async fn get_config_file(&self, req: GetConfigFileRequest) -> Result<ConfigFile, PolarisError>;
 
-    async fn update_config_file(&self, req: UpdateConfigFileRequest) -> Result<bool, PolarisError>;
+    async fn create_config_file(&self, req: CreateConfigFileRequest) -> Result<bool, PolarisError>;
 
-    async fn delete_config_file(&self, req: DeleteConfigFileRequest) -> Result<bool, PolarisError>;
+    async fn update_config_file(&self, req: UpdateConfigFileRequest) -> Result<bool, PolarisError>;
 
     async fn publish_config_file(
         &self,

src/config/api.rs

@@ -13,73 +13,150 @@
 // CONDITIONS OF ANY KIND, either express or implied. See the License for the
 // specific language governing permissions and limitations under the License.
 
-use std::sync::Arc;
+use std::{
+    collections::HashMap,
+    sync::{
+        atomic::{AtomicBool, Ordering},
+        Arc,
+    },
+};
+
+use tokio::sync::RwLock;
 
 use crate::core::{
     context::SDKContext,
-    model::{config::ConfigFile, error::PolarisError},
+    model::{
+        cache::{EventType, ServerEvent},
+        config::{ConfigFile, ConfigFileChangeEvent},
+        error::PolarisError,
+    },
+    plugin::cache::{Action, ResourceListener},
 };
 
 use super::{
     api::{ConfigFileAPI, ConfigGroupAPI},
     req::{
-        DeleteConfigFileRequest, GetConfigFileRequest, GetConfigGroupRequest,
+        self, CreateConfigFileRequest, GetConfigFileRequest, GetConfigGroupRequest,
         PublishConfigFileRequest, UpdateConfigFileRequest, WatchConfigFileRequest,
         WatchConfigFileResponse, WatchConfigGroupRequest, WatchConfigGroupResponse,
     },
 };
 
+struct ConfigFileWatcher {
+    req: WatchConfigFileRequest,
+}
+
+struct ConfigFileResourceListener {
+    // watchers: namespace#service -> ConfigFileWatcher
+    watchers: Arc<RwLock<HashMap<String, Vec<ConfigFileWatcher>>>>,
+}
+
+#[async_trait::async_trait]
+impl ResourceListener for ConfigFileResourceListener {
+    // 处理事件
+    async fn on_event(&self, action: Action, val: ServerEvent) {
+        let event_key = val.event_key;
+        let mut watch_key = event_key.namespace.clone();
+        let group = event_key.filter.get("group");
+        let file = event_key.filter.get("file");
+        watch_key.push_str("#");
+        watch_key.push_str(group.unwrap().as_str());
+        watch_key.push_str("#");
+        watch_key.push_str(file.unwrap().as_str());
+
+        let watchers = self.watchers.read().await;
+        if let Some(watchers) = watchers.get(&watch_key) {
+            let cfg_cache_opt = val.value.to_config_file();
+            match cfg_cache_opt {
+                Some(cfg_cache_val) => {
+                    for watcher in watchers {
+                        (watcher.req.call_back)(ConfigFileChangeEvent {
+                            config_file: cfg_cache_val.to_config_file(),
+                        })
+                    }
+                }
+                None => {
+                    // do nothing
+                }
+            }
+        }
+    }
+
+    // 获取监听的key
+    fn watch_key(&self) -> EventType {
+        EventType::ConfigFile
+    }
+}
+
 /// DefaultConfigFileAPI
 pub struct DefaultConfigFileAPI {
     context: Arc<SDKContext>,
     manage_sdk: bool,
+    // watchers: namespace#service -> ConfigFileWatcher
+    watchers: Arc<ConfigFileResourceListener>,
+    //
+    register_resource_watcher: AtomicBool,
 }
 
 impl DefaultConfigFileAPI {
     pub fn new(context: Arc<SDKContext>, manage_sdk: bool) -> Self {
         Self {
             context,
             manage_sdk: manage_sdk,
+            watchers: Arc::new(ConfigFileResourceListener {
+                watchers: Arc::new(RwLock::new(HashMap::new())),
+            }),
+            register_resource_watcher: AtomicBool::new(false),
         }
     }
 }
 
 #[async_trait::async_trait]
 impl ConfigFileAPI for DefaultConfigFileAPI {
-    async fn get_config_file(
-        &self,
-        req: GetConfigFileRequest,
-    ) -> Result<ConfigFile, PolarisError> {
-        let engine = self.context.get_engine();
-        engine.get_config_file(req).await
+    async fn get_config_file(&self, req: GetConfigFileRequest) -> Result<ConfigFile, PolarisError> {
+        self.context.get_engine().get_config_file(req).await
     }
 
-    async fn update_config_file(
-        &self,
-        _req: UpdateConfigFileRequest,
-    ) -> Result<bool, PolarisError> {
-        Ok(true)
+    async fn create_config_file(&self, req: CreateConfigFileRequest) -> Result<bool, PolarisError> {
+        self.context.get_engine().create_config_file(req).await
     }
 
-    async fn delete_config_file(
-        &self,
-        _req: DeleteConfigFileRequest,
-    ) -> Result<bool, PolarisError> {
-        Ok(true)
+    async fn update_config_file(&self, req: UpdateConfigFileRequest) -> Result<bool, PolarisError> {
+        self.context.get_engine().update_config_file(req).await
     }
 
     async fn publish_config_file(
         &self,
-        _req: PublishConfigFileRequest,
+        req: PublishConfigFileRequest,
     ) -> Result<bool, PolarisError> {
-        Ok(true)
+        self.context.get_engine().publish_config_file(req).await
     }
 
     async fn watch_config_file(
         &self,
-        _req: WatchConfigFileRequest,
+        req: WatchConfigFileRequest,
     ) -> Result<WatchConfigFileResponse, PolarisError> {
-        todo!()
+        if self
+            .register_resource_watcher
+            .compare_exchange(false, true, Ordering::Relaxed, Ordering::SeqCst)
+            .is_ok()
+        {
+            // 延迟注册资源监听器
+            self.context
+                .get_engine()
+                .register_resource_listener(self.watchers.clone())
+                .await;
+        }
+
+        let mut watchers = self.watchers.watchers.write().await;
+
+        let watch_key = req.get_key();
+        let items = watchers
+            .entry(watch_key.clone())
+            .or_insert_with(|| Vec::new());
+
+        items.push(ConfigFileWatcher { req });
+        Ok(WatchConfigFileResponse {})
     }
 }
 

src/config/default.rs

@@ -13,9 +13,13 @@
 // CONDITIONS OF ANY KIND, either express or implied. See the License for the
 // specific language governing permissions and limitations under the License.
 
-use std::{collections::HashMap, time::Duration};
+use std::{collections::HashMap, sync::Arc, time::Duration};
 
-use crate::core::model::config::{ConfigFile, ConfigFileRequest};
+use crate::core::model::{
+    config::{ConfigFile, ConfigFileChangeEvent, ConfigFileRequest, ConfigReleaseRequest},
+    naming::ServiceInstancesChangeEvent,
+    pb::lib::ConfigFileRelease,
+};
 
 #[derive(Clone, Debug)]
 pub struct GetConfigFileRequest {
@@ -28,6 +32,7 @@ pub struct GetConfigFileRequest {
 #[derive(Clone, Debug)]
 pub struct CreateConfigFileRequest {
     pub flow_id: String,
+    pub timeout: Duration,
     pub file: ConfigFile,
 }
 
@@ -45,16 +50,108 @@ impl CreateConfigFileRequest {
 }
 
 #[derive(Clone, Debug)]
-pub struct UpdateConfigFileRequest {}
+pub struct UpdateConfigFileRequest {
+    pub flow_id: String,
+    pub timeout: Duration,
+    pub file: ConfigFile,
+}
 
-#[derive(Clone, Debug)]
-pub struct DeleteConfigFileRequest {}
+impl UpdateConfigFileRequest {
+    pub fn to_config_request(&self) -> ConfigFileRequest {
+        let mut flow_id = self.flow_id.clone();
+        if flow_id.is_empty() {
+            flow_id = uuid::Uuid::new_v4().to_string();
+        }
+        ConfigFileRequest {
+            flow_id: flow_id,
+            config_file: self.file.clone(),
+        }
+    }
+}
 
 #[derive(Clone, Debug)]
-pub struct PublishConfigFileRequest {}
+pub struct DeleteConfigFileRequest {
+    pub flow_id: String,
+    pub namespace: String,
+    pub group: String,
+    pub file: String,
+    pub timeout: Duration,
+}
+
+impl DeleteConfigFileRequest {
+    pub fn to_config_request(&self) -> ConfigFileRequest {
+        let mut flow_id = self.flow_id.clone();
+        if flow_id.is_empty() {
+            flow_id = uuid::Uuid::new_v4().to_string();
+        }
+        let mut file = ConfigFile::default();
+        file.namespace = self.namespace.clone();
+        file.group = self.group.clone();
+        file.name = self.file.clone();
+        ConfigFileRequest {
+            flow_id: flow_id,
+            config_file: file,
+        }
+    }
+}
 
 #[derive(Clone, Debug)]
-pub struct WatchConfigFileRequest {}
+pub struct PublishConfigFileRequest {
+    pub flow_id: String,
+    pub namespace: String,
+    pub group: String,
+    pub file: String,
+    pub release_name: String,
+    pub md5: String,
+    pub timeout: Duration,
+}
+
+impl PublishConfigFileRequest {
+    pub fn to_config_request(&self) -> ConfigReleaseRequest {
+        let mut flow_id = self.flow_id.clone();
+        if flow_id.is_empty() {
+            flow_id = uuid::Uuid::new_v4().to_string();
+        }
+        ConfigReleaseRequest {
+            flow_id: flow_id,
+            config_file: ConfigFileRelease {
+                id: None,
+                name: Some(self.release_name.clone()),
+                namespace: Some(self.namespace.clone()),
+                group: Some(self.group.clone()),
+                file_name: Some(self.file.clone()),
+                content: None,
+                comment: None,
+                md5: Some(self.md5.clone()),
+                version: None,
+                create_time: None,
+                create_by: None,
+                modify_time: None,
+                modify_by: None,
+                tags: vec![],
+                active: None,
+                format: None,
+                release_description: None,
+                release_type: None,
+                beta_labels: vec![],
+            },
+        }
+    }
+}
+
+#[derive(Clone)]
+pub struct WatchConfigFileRequest {
+    pub namespace: String,
+    pub group: String,
+    pub file: String,
+    pub call_back: Arc<dyn Fn(ConfigFileChangeEvent) + Send + Sync>,
+}
+
+impl WatchConfigFileRequest {
+    pub fn get_key(&self) -> String {
+        format!("{}#{}#{}", self.namespace, self.group, self.file)
+    }
+}
 
 pub struct WatchConfigFileResponse {}
 

src/config/req.rs

@@ -1,23 +1,34 @@
 // Tencent is pleased to support the open source community by making Polaris available.
-// 
+//
 // Copyright (C) 2019 THL A29 Limited, a Tencent company. All rights reserved.
-// 
+//
 // Licensed under the BSD 3-Clause License (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
-// 
+//
 // https://opensource.org/licenses/BSD-3-Clause
-// 
+//
 // Unless required by applicable law or agreed to in writing, software distributed
 // under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
 // CONDITIONS OF ANY KIND, either express or implied. See the License for the
 // specific language governing permissions and limitations under the License.
 
+use std::collections::HashMap;
+
 use serde::Deserialize;
 
 #[derive(Deserialize, Debug)]
 #[serde(rename_all = "camelCase", deny_unknown_fields)]
 pub struct ConfigFileConfig {
     pub properties_value_cache_size: u32,
     pub properties_value_expire_time: u32,
+    pub config_filter: ConfigFilter,
+}
+
+#[derive(Deserialize, Debug)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+pub struct ConfigFilter {
+    pub enable: bool,
+    pub chain: Vec<String>,
+    pub plugin: HashMap<String, serde_yaml::Value>,
 }