aboutsummaryrefslogtreecommitdiff
path: root/src/rpc
diff options
context:
space:
mode:
Diffstat (limited to 'src/rpc')
-rw-r--r--src/rpc/Cargo.toml3
-rw-r--r--src/rpc/consul.rs2
-rw-r--r--src/rpc/kubernetes.rs2
-rw-r--r--src/rpc/rpc_helper.rs22
-rw-r--r--src/rpc/system.rs32
5 files changed, 30 insertions, 31 deletions
diff --git a/src/rpc/Cargo.toml b/src/rpc/Cargo.toml
index 9f223715..5d5750c4 100644
--- a/src/rpc/Cargo.toml
+++ b/src/rpc/Cargo.toml
@@ -17,6 +17,7 @@ path = "lib.rs"
format_table.workspace = true
garage_db.workspace = true
garage_util.workspace = true
+garage_net.workspace = true
arc-swap.workspace = true
bytes.workspace = true
@@ -49,8 +50,6 @@ tokio.workspace = true
tokio-stream.workspace = true
opentelemetry.workspace = true
-netapp.workspace = true
-
[features]
kubernetes-discovery = [ "kube", "k8s-openapi", "schemars" ]
consul-discovery = [ "reqwest", "err-derive" ]
diff --git a/src/rpc/consul.rs b/src/rpc/consul.rs
index 71fd71b0..f088bf3f 100644
--- a/src/rpc/consul.rs
+++ b/src/rpc/consul.rs
@@ -6,7 +6,7 @@ use std::net::{IpAddr, SocketAddr};
use err_derive::Error;
use serde::{Deserialize, Serialize};
-use netapp::NodeID;
+use garage_net::NodeID;
use garage_util::config::ConsulDiscoveryAPI;
use garage_util::config::ConsulDiscoveryConfig;
diff --git a/src/rpc/kubernetes.rs b/src/rpc/kubernetes.rs
index 63c6567d..85254bb5 100644
--- a/src/rpc/kubernetes.rs
+++ b/src/rpc/kubernetes.rs
@@ -10,7 +10,7 @@ use k8s_openapi::apiextensions_apiserver::pkg::apis::apiextensions::v1::CustomRe
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
-use netapp::NodeID;
+use garage_net::NodeID;
use garage_util::config::KubernetesDiscoveryConfig;
diff --git a/src/rpc/rpc_helper.rs b/src/rpc/rpc_helper.rs
index e59c372a..c46e577f 100644
--- a/src/rpc/rpc_helper.rs
+++ b/src/rpc/rpc_helper.rs
@@ -14,13 +14,13 @@ use opentelemetry::{
Context,
};
-pub use netapp::endpoint::{Endpoint, EndpointHandler, StreamingEndpointHandler};
-pub use netapp::message::{
+pub use garage_net::endpoint::{Endpoint, EndpointHandler, StreamingEndpointHandler};
+pub use garage_net::message::{
IntoReq, Message as Rpc, OrderTag, Req, RequestPriority, Resp, PRIO_BACKGROUND, PRIO_HIGH,
PRIO_NORMAL, PRIO_SECONDARY,
};
-use netapp::peering::fullmesh::FullMeshPeeringStrategy;
-pub use netapp::{self, NetApp, NodeID};
+use garage_net::peering::PeeringManager;
+pub use garage_net::{self, NetApp, NodeID};
use garage_util::data::*;
use garage_util::error::Error;
@@ -90,7 +90,7 @@ pub struct RpcHelper(Arc<RpcHelperInner>);
struct RpcHelperInner {
our_node_id: Uuid,
- fullmesh: Arc<FullMeshPeeringStrategy>,
+ peering: Arc<PeeringManager>,
ring: watch::Receiver<Arc<Ring>>,
metrics: RpcMetrics,
rpc_timeout: Duration,
@@ -99,7 +99,7 @@ struct RpcHelperInner {
impl RpcHelper {
pub(crate) fn new(
our_node_id: Uuid,
- fullmesh: Arc<FullMeshPeeringStrategy>,
+ peering: Arc<PeeringManager>,
ring: watch::Receiver<Arc<Ring>>,
rpc_timeout: Option<Duration>,
) -> Self {
@@ -107,7 +107,7 @@ impl RpcHelper {
Self(Arc::new(RpcHelperInner {
our_node_id,
- fullmesh,
+ peering,
ring,
metrics,
rpc_timeout: rpc_timeout.unwrap_or(DEFAULT_TIMEOUT),
@@ -183,7 +183,7 @@ impl RpcHelper {
N: IntoReq<M>,
H: StreamingEndpointHandler<M>,
{
- let msg = msg.into_req().map_err(netapp::error::Error::from)?;
+ let msg = msg.into_req().map_err(garage_net::error::Error::from)?;
let resps = join_all(
to.iter()
@@ -210,7 +210,7 @@ impl RpcHelper {
{
let to = self
.0
- .fullmesh
+ .peering
.get_peer_list()
.iter()
.map(|p| p.id.into())
@@ -274,7 +274,7 @@ impl RpcHelper {
H: StreamingEndpointHandler<M> + 'static,
S: Send + 'static,
{
- let msg = msg.into_req().map_err(netapp::error::Error::from)?;
+ let msg = msg.into_req().map_err(garage_net::error::Error::from)?;
// Build future for each request
// They are not started now: they are added below in a FuturesUnordered
@@ -391,7 +391,7 @@ impl RpcHelper {
pub fn request_order(&self, nodes: &[Uuid]) -> Vec<Uuid> {
// Retrieve some status variables that we will use to sort requests
- let peer_list = self.0.fullmesh.get_peer_list();
+ let peer_list = self.0.peering.get_peer_list();
let ring: Arc<Ring> = self.0.ring.borrow().clone();
let our_zone = match ring.layout.node_role(&self.0.our_node_id) {
Some(pc) => &pc.zone,
diff --git a/src/rpc/system.rs b/src/rpc/system.rs
index 4cec369b..de44e656 100644
--- a/src/rpc/system.rs
+++ b/src/rpc/system.rs
@@ -16,11 +16,11 @@ use tokio::select;
use tokio::sync::watch;
use tokio::sync::Mutex;
-use netapp::endpoint::{Endpoint, EndpointHandler};
-use netapp::message::*;
-use netapp::peering::fullmesh::FullMeshPeeringStrategy;
-use netapp::util::parse_and_resolve_peer_addr_async;
-use netapp::{NetApp, NetworkKey, NodeID, NodeKey};
+use garage_net::endpoint::{Endpoint, EndpointHandler};
+use garage_net::message::*;
+use garage_net::peering::PeeringManager;
+use garage_net::util::parse_and_resolve_peer_addr_async;
+use garage_net::{NetApp, NetworkKey, NodeID, NodeKey};
#[cfg(feature = "kubernetes-discovery")]
use garage_util::config::KubernetesDiscoveryConfig;
@@ -92,7 +92,7 @@ pub struct System {
node_status: RwLock<HashMap<Uuid, (u64, NodeStatus)>>,
pub netapp: Arc<NetApp>,
- fullmesh: Arc<FullMeshPeeringStrategy>,
+ peering: Arc<PeeringManager>,
pub rpc: RpcHelper,
system_endpoint: Arc<Endpoint<SystemRpc, System>>,
@@ -326,9 +326,9 @@ impl System {
}
let netapp = NetApp::new(GARAGE_VERSION_TAG, network_key, node_key);
- let fullmesh = FullMeshPeeringStrategy::new(netapp.clone(), vec![], rpc_public_addr);
+ let peering = PeeringManager::new(netapp.clone(), vec![], rpc_public_addr);
if let Some(ping_timeout) = config.rpc_ping_timeout_msec {
- fullmesh.set_ping_timeout_millis(ping_timeout);
+ peering.set_ping_timeout_millis(ping_timeout);
}
let system_endpoint = netapp.endpoint(SYSTEM_RPC_PATH.into());
@@ -358,10 +358,10 @@ impl System {
local_status: ArcSwap::new(Arc::new(local_status)),
node_status: RwLock::new(HashMap::new()),
netapp: netapp.clone(),
- fullmesh: fullmesh.clone(),
+ peering: peering.clone(),
rpc: RpcHelper::new(
netapp.id.into(),
- fullmesh,
+ peering,
ring.clone(),
config.rpc_timeout_msec.map(Duration::from_millis),
),
@@ -393,7 +393,7 @@ impl System {
self.netapp
.clone()
.listen(self.rpc_listen_addr, None, must_exit.clone()),
- self.fullmesh.clone().run(must_exit.clone()),
+ self.peering.clone().run(must_exit.clone()),
self.discovery_loop(must_exit.clone()),
self.status_exchange_loop(must_exit.clone()),
);
@@ -405,7 +405,7 @@ impl System {
pub fn get_known_nodes(&self) -> Vec<KnownNodeInfo> {
let node_status = self.node_status.read().unwrap();
let known_nodes = self
- .fullmesh
+ .peering
.get_peer_list()
.iter()
.map(|n| KnownNodeInfo {
@@ -726,10 +726,10 @@ impl System {
async fn discovery_loop(self: &Arc<Self>, mut stop_signal: watch::Receiver<bool>) {
while !*stop_signal.borrow() {
let not_configured = self.ring.borrow().layout.check().is_err();
- let no_peers = self.fullmesh.get_peer_list().len() < self.replication_factor;
+ let no_peers = self.peering.get_peer_list().len() < self.replication_factor;
let expected_n_nodes = self.ring.borrow().layout.num_nodes();
let bad_peers = self
- .fullmesh
+ .peering
.get_peer_list()
.iter()
.filter(|p| p.is_up())
@@ -811,7 +811,7 @@ impl System {
// Prepare new peer list to save to file
// It is a vec of tuples (node ID as Uuid, node SocketAddr)
let mut peer_list = self
- .fullmesh
+ .peering
.get_peer_list()
.iter()
.map(|n| (n.id.into(), n.addr))
@@ -985,7 +985,7 @@ async fn resolve_peers(peers: &[String]) -> Vec<(NodeID, SocketAddr)> {
fn connect_error_message(
addr: SocketAddr,
pubkey: ed25519::PublicKey,
- e: netapp::error::Error,
+ e: garage_net::error::Error,
) -> String {
format!("Error establishing RPC connection to remote node: {}@{}.\nThis can happen if the remote node is not reachable on the network, but also if the two nodes are not configured with the same rpc_secret.\n{}", hex::encode(pubkey), addr, e)
}
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-10 08:51:41 +0000