Add admin rpc to return sync service state (#151)

This commit is contained in:
Bo QIU 2024-08-06 14:01:57 +08:00 committed by GitHub
parent 891e00fa80
commit 6ade66c086
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 46 additions and 4 deletions

View File

@ -2,7 +2,7 @@ use crate::types::{LocationInfo, NetworkInfo, PeerInfo};
use jsonrpsee::core::RpcResult; use jsonrpsee::core::RpcResult;
use jsonrpsee::proc_macros::rpc; use jsonrpsee::proc_macros::rpc;
use std::collections::HashMap; use std::collections::HashMap;
use sync::FileSyncInfo; use sync::{FileSyncInfo, SyncServiceState};
#[rpc(server, client, namespace = "admin")] #[rpc(server, client, namespace = "admin")]
pub trait Rpc { pub trait Rpc {
@ -27,6 +27,9 @@ pub trait Rpc {
#[method(name = "terminateSync")] #[method(name = "terminateSync")]
async fn terminate_sync(&self, tx_seq: u64) -> RpcResult<bool>; async fn terminate_sync(&self, tx_seq: u64) -> RpcResult<bool>;
#[method(name = "getSyncServiceState")]
async fn get_sync_service_state(&self) -> RpcResult<SyncServiceState>;
#[method(name = "getSyncStatus")] #[method(name = "getSyncStatus")]
async fn get_sync_status(&self, tx_seq: u64) -> RpcResult<String>; async fn get_sync_status(&self, tx_seq: u64) -> RpcResult<String>;

View File

@ -8,7 +8,7 @@ use network::{multiaddr::Protocol, Multiaddr};
use std::collections::HashMap; use std::collections::HashMap;
use std::net::IpAddr; use std::net::IpAddr;
use storage::config::all_shards_available; use storage::config::all_shards_available;
use sync::{FileSyncInfo, SyncRequest, SyncResponse}; use sync::{FileSyncInfo, SyncRequest, SyncResponse, SyncServiceState};
use task_executor::ShutdownReason; use task_executor::ShutdownReason;
pub struct RpcServerImpl { pub struct RpcServerImpl {
@ -119,6 +119,17 @@ impl RpcServer for RpcServerImpl {
} }
} }
async fn get_sync_service_state(&self) -> RpcResult<SyncServiceState> {
info!("admin_getSyncServiceState()");
let response = self.ctx.request_sync(SyncRequest::SyncState).await?;
match response {
SyncResponse::SyncState { state } => Ok(state),
_ => Err(error::internal_error("unexpected response type")),
}
}
#[tracing::instrument(skip(self), err)] #[tracing::instrument(skip(self), err)]
async fn get_sync_status(&self, tx_seq: u64) -> RpcResult<String> { async fn get_sync_status(&self, tx_seq: u64) -> RpcResult<String> {
info!("admin_getSyncStatus({tx_seq})"); info!("admin_getSyncStatus({tx_seq})");

View File

@ -7,9 +7,10 @@ mod controllers;
mod service; mod service;
pub mod test_util; pub mod test_util;
use auto_sync::{batcher_random::RandomBatcherState, batcher_serial::SerialBatcherState};
pub use controllers::FileSyncInfo; pub use controllers::FileSyncInfo;
use duration_str::deserialize_duration; use duration_str::deserialize_duration;
use serde::Deserialize; use serde::{Deserialize, Serialize};
pub use service::{SyncMessage, SyncReceiver, SyncRequest, SyncResponse, SyncSender, SyncService}; pub use service::{SyncMessage, SyncReceiver, SyncRequest, SyncResponse, SyncSender, SyncService};
use std::{ use std::{
fmt::Debug, fmt::Debug,
@ -64,3 +65,11 @@ impl InstantWrapper {
self.0.elapsed() self.0.elapsed()
} }
} }
#[derive(Debug, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct SyncServiceState {
pub num_syncing: usize,
pub auto_sync_serial: Option<SerialBatcherState>,
pub auto_sync_random: Option<RandomBatcherState>,
}

View File

@ -4,7 +4,7 @@ use crate::controllers::{
FailureReason, FileSyncGoal, FileSyncInfo, SerialSyncController, SyncState, FailureReason, FileSyncGoal, FileSyncInfo, SerialSyncController, SyncState,
MAX_CHUNKS_TO_REQUEST, MAX_CHUNKS_TO_REQUEST,
}; };
use crate::Config; use crate::{Config, SyncServiceState};
use anyhow::{bail, Result}; use anyhow::{bail, Result};
use file_location_cache::FileLocationCache; use file_location_cache::FileLocationCache;
use libp2p::swarm::DialError; use libp2p::swarm::DialError;
@ -74,6 +74,7 @@ pub enum SyncMessage {
#[derive(Debug)] #[derive(Debug)]
pub enum SyncRequest { pub enum SyncRequest {
SyncState,
SyncStatus { SyncStatus {
tx_seq: u64, tx_seq: u64,
}, },
@ -99,6 +100,7 @@ pub enum SyncRequest {
#[derive(Debug)] #[derive(Debug)]
pub enum SyncResponse { pub enum SyncResponse {
SyncState { state: SyncServiceState },
SyncStatus { status: Option<SyncState> }, SyncStatus { status: Option<SyncState> },
SyncFile { err: String }, SyncFile { err: String },
FileSyncInfo { result: HashMap<u64, FileSyncInfo> }, FileSyncInfo { result: HashMap<u64, FileSyncInfo> },
@ -278,6 +280,23 @@ impl SyncService {
sender: channel::ResponseSender<SyncResponse>, sender: channel::ResponseSender<SyncResponse>,
) { ) {
match req { match req {
SyncRequest::SyncState => {
let state = match &self.auto_sync_manager {
Some(manager) => SyncServiceState {
num_syncing: self.controllers.len(),
auto_sync_serial: Some(manager.serial.get_state().await),
auto_sync_random: manager.random.get_state().await.ok(),
},
None => SyncServiceState {
num_syncing: self.controllers.len(),
auto_sync_serial: None,
auto_sync_random: None,
},
};
let _ = sender.send(SyncResponse::SyncState { state });
}
SyncRequest::SyncStatus { tx_seq } => { SyncRequest::SyncStatus { tx_seq } => {
let status = self let status = self
.controllers .controllers