gitdata.ai/gitks
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore(infra): add gRPC layer, update protobufs, remove immediate module

Browse Source 
- Add gRPC service modules: auth, channel, channel settings, member,
  permission
- Update protobuf definitions and generated code
- Remove immediate/ real-time module (superseded by IM service)
- Update etcd discovery and registration
- Update cache, error, config, and build infrastructure
- Add ADR documentation
- Update OpenAPI spec
This commit is contained in:
zhenyi
2026-06-10 18:49:42 +08:00
parent 9eb77ab98b
commit 1000f8a80d
57 changed files with 22524 additions and 2703 deletions
Download Patch File Download Diff File Expand all files Collapse all files
etcd/discovery.rs
+149 -24
View File
@@ -5,18 +5,151 @@ use uuid::Uuid;
use crate::error::{AppError, AppResult};
use crate::pb::{EmailClient, RepoClient};
use super::types::ServiceInstance;
use super::{EtcdRegistry, EtcdRegistryInner};
use super::types::{GitksPeerInfo, ServiceInstance};
use super::{EtcdRegistry, EtcdRegistryInner, storage_name_to_uuid};
/// etcd prefix where gitks nodes register themselves (gitks::cluster::ClusterManager).
const GITKS_NODES_PREFIX: &str = "/gitks/nodes/";
impl EtcdRegistry {
pub async fn start_discovery(&self) -> AppResult<()> {
self.load_initial("git").await?;
self.load_initial("mail").await?;
self.spawn_watch("git");
self.spawn_watch("mail");
// Discover gitks nodes from gitks's own etcd prefix.
self.load_gitks_nodes().await?;
self.spawn_gitks_watch();
// Discover mail services from appks's service prefix.
if self.email_client.is_none() {
self.load_initial("mail").await?;
self.spawn_watch("mail");
}
Ok(())
}
// Section: gitks node discovery (from /gitks/nodes/)
async fn load_gitks_nodes(&self) -> AppResult<()> {
let resp = {
let mut client = self.inner.client.lock().await;
client
.get(GITKS_NODES_PREFIX, Some(GetOptions::new().with_prefix()))
.await
.map_err(|e| {
AppError::Config(format!("etcd get {GITKS_NODES_PREFIX} failed: {e}"))
})?
};
for kv in resp.kvs() {
let key = kv.key_str().unwrap_or_default();
let value = kv.value_str().unwrap_or_default();
if let Ok(peer) = serde_json::from_str::<GitksPeerInfo>(value) {
Self::upsert_gitks_node(&self.inner, key, &peer);
} else {
tracing::warn!(key = key, "failed to parse gitks peer info from etcd");
}
}
tracing::info!(
prefix = GITKS_NODES_PREFIX,
count = self.inner.git_nodes.len(),
"gitks node discovery complete"
);
Ok(())
}
fn spawn_gitks_watch(&self) {
let inner = self.inner.clone();
tokio::spawn(async move {
loop {
match Self::gitks_watch_loop(&inner).await {
Ok(()) => break,
Err(e) => {
tracing::warn!(error = %e, "gitks etcd watch disconnected, retrying in 3s");
tokio::time::sleep(std::time::Duration::from_secs(3)).await;
}
}
}
});
}
async fn gitks_watch_loop(inner: &EtcdRegistryInner) -> AppResult<()> {
let mut stream = {
let mut client = inner.client.lock().await;
client
.watch(GITKS_NODES_PREFIX, Some(WatchOptions::new().with_prefix()))
.await
.map_err(|e| {
AppError::Config(format!("etcd watch {GITKS_NODES_PREFIX} failed: {e}"))
})?
};
while let Some(resp) = stream.next().await {
let resp =
resp.map_err(|e| AppError::Config(format!("gitks watch stream error: {e}")))?;
for event in resp.events() {
let Some(kv) = event.kv() else { continue };
let key = kv.key_str().unwrap_or_default();
match event.event_type() {
etcd_client::EventType::Put => {
let value = kv.value_str().unwrap_or_default();
if let Ok(peer) = serde_json::from_str::<GitksPeerInfo>(value) {
Self::upsert_gitks_node(inner, key, &peer);
tracing::info!(
storage_name = %peer.storage_name,
grpc_addr = %peer.grpc_addr,
"gitks node upserted"
);
}
}
etcd_client::EventType::Delete => {
let storage_name = key.strip_prefix(GITKS_NODES_PREFIX).unwrap_or(&key);
let node_id = storage_name_to_uuid(storage_name);
inner.git_nodes.remove(&node_id);
tracing::info!(storage_name = storage_name, "gitks node removed");
}
}
}
}
Ok(())
}
fn upsert_gitks_node(inner: &EtcdRegistryInner, key: &str, peer: &GitksPeerInfo) {
let node_id = storage_name_to_uuid(&peer.storage_name);
if peer.grpc_addr.is_empty() {
tracing::warn!(
storage_name = %peer.storage_name,
key = key,
"gitks peer has empty grpc_addr, skipping"
);
return;
}
match RepoClient::lazy_connect(&peer.grpc_addr) {
Ok(client) => {
inner.git_nodes.insert(node_id, client);
tracing::debug!(
storage_name = %peer.storage_name,
node_id = %node_id,
grpc_addr = %peer.grpc_addr,
"gitks node connected"
);
}
Err(e) => {
tracing::error!(
storage_name = %peer.storage_name,
grpc_addr = %peer.grpc_addr,
error = %e,
"gitks node connect failed"
);
}
}
}
// Section: mail service discovery (from appks's own etcd prefix)
async fn load_initial(&self, service: &str) -> AppResult<()> {
let prefix = self.service_prefix(service);
let resp = {
@@ -38,7 +171,7 @@ impl EtcdRegistry {
tracing::info!(
service = service,
prefix = prefix.as_str(),
"etcd initial discovery complete"
"etcd mail discovery complete"
);
Ok(())
}
@@ -66,7 +199,7 @@ impl EtcdRegistry {
}
async fn watch_loop(inner: &EtcdRegistryInner, prefix: &str, service: &str) -> AppResult<()> {
let (mut watcher, mut stream) = {
let mut stream = {
let mut client = inner.client.lock().await;
client
.watch(prefix, Some(WatchOptions::new().with_prefix()))
@@ -74,8 +207,6 @@ impl EtcdRegistry {
.map_err(|e| AppError::Config(format!("etcd watch {prefix} failed: {e}")))?
};
let _keep = &mut watcher;
while let Some(resp) = stream.next().await {
let resp =
resp.map_err(|e| AppError::Config(format!("etcd watch stream error: {e}")))?;
@@ -89,12 +220,12 @@ impl EtcdRegistry {
let value = kv.value_str().unwrap_or_default();
if let Ok(instance) = serde_json::from_str::<ServiceInstance>(value) {
Self::upsert_instance(inner, service, key, &instance);
tracing::info!(service = service, key = key, "etcd service upserted");
tracing::info!(service = service, key = key, "mail service upserted");
}
}
etcd_client::EventType::Delete => {
Self::remove_instance(inner, service, key);
tracing::info!(service = service, key = key, "etcd service removed");
tracing::info!(service = service, key = key, "mail service removed");
}
}
}
@@ -123,20 +254,17 @@ impl EtcdRegistry {
let addr = instance.addr.clone();
match service {
"git" => match RepoClient::lazy_connect(&addr) {
Ok(client) => {
inner.git_nodes.insert(node_id, client);
}
Err(e) => {
tracing::error!(key = key, addr = addr.as_str(), error = %e, "git client connect failed");
}
},
"mail" => match EmailClient::lazy_connect(&addr) {
Ok(client) => {
inner.mail_nodes.insert(node_id, client);
}
Err(e) => {
tracing::error!(key = key, addr = addr.as_str(), error = %e, "mail client connect failed");
tracing::error!(
key = key,
addr = addr.as_str(),
error = %e,
"mail client connect failed"
);
}
},
_ => {}
@@ -148,9 +276,6 @@ impl EtcdRegistry {
return;
};
match service {
"git" => {
inner.git_nodes.remove(&node_id);
}
"mail" => {
inner.mail_nodes.remove(&node_id);
}