Skip to content

feat(huntsman)!: Replace endpoint's IP address with general hostname. #401

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
sitaowang1998 merged 7 commits into from
Jul 17, 2026
+102 −107
Merged

feat(huntsman)!: Replace endpoint's IP address with general hostname. #401

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
eaadd32
Add host type alias
sitaowang1998 Jul 16, 2026
8e4456f
Replace host in core
sitaowang1998 Jul 16, 2026
27b7e7a
Update proto serde
sitaowang1998 Jul 16, 2026
233e43b
Update scheduler
sitaowang1998 Jul 16, 2026
9d695cb
Update db
sitaowang1998 Jul 16, 2026
016bf89
Update lock
sitaowang1998 Jul 16, 2026
c8d01f9
Merge branch 'main' into hostname
sitaowang1998 Jul 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

6 changes: 3 additions & 3 deletions components/spider-core/src/types/scheduler.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
use std::net::IpAddr;
use spider_utils::config::Host;

use crate::types::id::JobId;
use crate::types::id::ResourceGroupId;
Expand All @@ -8,10 +8,10 @@ use crate::types::id::TaskAssignmentId;
use crate::types::id::TaskId;

/// The currently registered scheduler endpoint.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RegisteredScheduler {
pub id: SchedulerId,
pub ip_address: IpAddr,
pub host: Host,
pub port: u16,
}

Expand Down
1 change: 1 addition & 0 deletions components/spider-proto-rust/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ path = "src/lib.rs"
[dependencies]
prost = "0.14.4"
spider-core = { path = "../spider-core" }
spider-utils = { path = "../spider-utils" }
thiserror = "2.0.18"
tonic = "0.14.6"
tonic-prost = "0.14.6"
Expand Down
4 changes: 2 additions & 2 deletions components/spider-proto-rust/src/generated/storage.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -181,7 +181,7 @@ pub struct UpdateExecutionManagerHeartbeatResponse {
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
pub struct RegisterSchedulerRequest {
#[prost(string, tag = "1")]
pub ip_address: ::prost::alloc::string::String,
pub host: ::prost::alloc::string::String,
#[prost(uint32, tag = "2")]
pub port: u32,
}
Expand All @@ -202,7 +202,7 @@ pub struct Scheduler {
#[prost(uint64, tag = "1")]
pub scheduler_id: u64,
#[prost(string, tag = "2")]
pub ip_address: ::prost::alloc::string::String,
pub host: ::prost::alloc::string::String,
#[prost(uint32, tag = "3")]
pub port: u32,
}
Expand Down
12 changes: 6 additions & 6 deletions components/spider-proto-rust/src/scheduler_registration.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,34 +8,34 @@ impl From<RegisteredScheduler> for storage::Scheduler {
fn from(scheduler: RegisteredScheduler) -> Self {
Self {
scheduler_id: scheduler.id.get(),
ip_address: scheduler.ip_address.to_string(),
host: scheduler.host.into_inner(),
port: u32::from(scheduler.port),
}
}
}

#[cfg(test)]
mod tests {
use std::net::IpAddr;

use spider_core::types::id::SchedulerId;
use spider_core::types::scheduler::RegisteredScheduler;
use spider_utils::config::Host;

use crate::storage;

#[test]
fn registered_scheduler_to_protocol_carries_id_ip_and_port() {
fn test_registered_scheduler_to_protocol_carries_id_host_and_port() {
const SCHEDULER_ID: SchedulerId = SchedulerId::from(42);
const PORT: u16 = 5678;

let scheduler = storage::Scheduler::from(RegisteredScheduler {
id: SCHEDULER_ID,
ip_address: IpAddr::V4("127.0.0.1".parse().expect("valid IP")),
host: Host::new("spider-scheduler".to_owned())
.expect("scheduler host should not be empty"),
port: PORT,
});

assert_eq!(scheduler.scheduler_id, SCHEDULER_ID.get());
assert_eq!(scheduler.ip_address, "127.0.0.1");
assert_eq!(scheduler.host, "spider-scheduler");
assert_eq!(scheduler.port, u32::from(PORT));
}
}
16 changes: 6 additions & 10 deletions components/spider-proto-rust/src/unpack/storage.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ use spider_core::types::id::ResourceGroupId;
use spider_core::types::id::SessionId;
use spider_core::types::id::TaskId;
use spider_core::types::id::TaskInstanceId;
use spider_utils::config::Host;
use tonic::Code;

use crate::storage::AddResourceGroupRequest;
Expand Down Expand Up @@ -185,21 +186,16 @@ impl RequestUnpack for ExecutionManagerIdRequest {
}
}

/// Unpacks [`RegisterSchedulerRequest`] into a tuple containing:
///
/// * The scheduler IP address.
/// * The scheduler port.
/// Unpacks [`RegisterSchedulerRequest`] into the scheduler host and port.
impl RequestUnpack for RegisterSchedulerRequest {
type Unpacked = (IpAddr, u16);
type Unpacked = (Host, u16);

fn unpack(self) -> Result<Self::Unpacked, UnpackError> {
let ip_address = self
.ip_address
.parse::<IpAddr>()
.map_err(|error| invalid_argument(format!("invalid IP address: {error}")))?;

@20001020ycx 20001020ycx Jul 17, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So what if the input is an IP, and the IP is invalid, would deleting the check cause any regression

@sitaowang1998 sitaowang1998 Jul 17, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There is no good way to tell if it is an invalid IP or just not an IP address. I think the best way is to leave it to the other functions mentioned in #401 (comment) to handle the failure.

@20001020ycx 20001020ycx Jul 17, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

really? isnt that you are doing this in the deleted line?

Say user inputs 256.256.256.256, this is not a valid IP right? We can easily parse it (of course though some lib) and let the user knows

@sitaowang1998 sitaowang1998 Jul 17, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Previously, it can only be an IPv4, Thus, we know for sure if it is not a valid IPv4 address, it is invalid. The AddrParseError is not specific enough to inform us if it is a invalid IP address out of range, or just something totally does not look like an IP address.

let host = Host::new(self.host)

@20001020ycx 20001020ycx Jul 17, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What if DNS failed to resolve the URL. This failure is often seen in the networking, and we should be able to deal with the gracefully

@sitaowang1998 sitaowang1998 Jul 17, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If the DNS name is syntactically wrong, then in EndpointConfig::endpoint function, the Endpoint::from_shared will return an error, and an error will be logged in the respecting grpc_server.rs and the binary exits with an error.

If the DNS name is syntactically correct, and not an existing address that is connectable, then the gRPC client's connect will return an error, an error will be logged and the binary exits with an error.

I am not sure if these are graceful enough in your opinion.

@20001020ycx 20001020ycx Jul 17, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Its good that you are thinking about this, but maybe its my problem that I dont really understand what exactly is this "error" you are talking about.

can you tell me say if I input an invalid ip or user name, what would the user see?

I am okay if you think what you presented to user can clearly show what is the root cause.

@sitaowang1998 sitaowang1998 Jul 17, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For example, if user set an invalid storage hostname in scheduler config, scheduler will fail with "Failed to parse storage endpoint." or "Failed to connect to storage gRPC service.". It also includes the message from underlying library.

.map_err(|_| invalid_argument("host must not be empty".to_owned()))?;
let port = u16::try_from(self.port)
.map_err(|_| invalid_argument(format!("port does not fit in `u16`: {}", self.port)))?;
Ok((ip_address, port))
Ok((host, port))
}
}

Expand Down
4 changes: 2 additions & 2 deletions components/spider-proto/storage/storage.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -174,7 +174,7 @@ message UpdateExecutionManagerHeartbeatResponse {
}

message RegisterSchedulerRequest {
string ip_address = 1;
string host = 1;
uint32 port = 2;
}

Expand All @@ -189,7 +189,7 @@ message RegisterSchedulerResponse {

message Scheduler {
uint64 scheduler_id = 1;
string ip_address = 2;
string host = 2;
uint32 port = 3;
}

Expand Down
2 changes: 1 addition & 1 deletion components/spider-scheduler/src/bin/grpc_server.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
let cli = Cli::parse();
let server_config = ServerConfig::from_yaml_file(&cli.config)
.inspect_err(|error| tracing::error!(error = % error, "Failed to load configuration."))?;
let listen_addr = SocketAddr::new(server_config.runtime.host, server_config.runtime.port);
let listen_addr = SocketAddr::new(server_config.host, server_config.port);

let storage_endpoint = server_config.storage_endpoint.endpoint().inspect_err(
|error| tracing::error!(error = % error, "Failed to parse storage endpoint."),
Expand Down
9 changes: 8 additions & 1 deletion components/spider-scheduler/src/config.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
//! Scheduler service configuration.

use std::net::IpAddr;
use std::num::NonZeroUsize;

use serde::Deserialize;
Expand All @@ -14,6 +15,12 @@ use crate::storage_client::SchedulerStorageClient;
/// Top-level configuration for the scheduler gRPC server.
#[derive(Clone, Debug, Deserialize)]
pub struct ServerConfig {
/// The IP address the gRPC server listens on.
pub host: IpAddr,

/// The port the gRPC server listens on.
pub port: u16,

/// The storage service gRPC endpoint.
pub storage_endpoint: EndpointConfig,

Expand All @@ -22,7 +29,7 @@ pub struct ServerConfig {
/// Must be greater than zero.
pub connection_pool_size: NonZeroUsize,

/// The scheduler runtime configuration (also supplies the gRPC listen host/port).
/// The scheduler runtime configuration.
pub runtime: RuntimeConfig,
}

Expand Down
2 changes: 1 addition & 1 deletion components/spider-scheduler/src/core_impl/round_robin/tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -149,7 +149,7 @@ impl MockStorageClient {
impl SchedulerStorageClient for MockStorageClient {
async fn register(
&self,
_ip_address: std::net::IpAddr,
_host: spider_utils::config::Host,
_port: u16,
) -> Result<SchedulerId, StorageClientError> {
Ok(SchedulerId::from(0))
Expand Down
33 changes: 17 additions & 16 deletions components/spider-scheduler/src/runtime.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ use std::time::Duration;

use serde::Deserialize;
use spider_core::types::id::SessionId;
use spider_utils::config::EndpointConfig;
use tokio_util::sync::CancellationToken;

use crate::config::SchedulerConfig;
Expand All @@ -34,11 +35,8 @@ pub struct RuntimeConfig {
#[serde(default)]
pub em_registry: ExecutionManagerRegistryConfig,

/// The IP address this scheduler advertises to the storage service during registration.
pub host: std::net::IpAddr,

/// The port this scheduler advertises to the storage service during registration.
pub port: u16,
/// The scheduler endpoint advertised to the storage service during registration.
pub advertised_endpoint: EndpointConfig,

/// The maximum time, in seconds, to wait for background tasks to stop during shutdown.
#[serde(default = "default_stop_timeout_sec")]
Expand Down Expand Up @@ -122,17 +120,18 @@ pub async fn create_runtime<SchedulerStorageClientType: SchedulerStorageClient +
),
SchedulerRuntimeError,
> {
let cancellation_token = CancellationToken::new();

let scheduler_id = storage_client.register(config.host, config.port).await?;
tracing::info!(scheduler_id = % scheduler_id, "Scheduler registered with storage.");

let RuntimeConfig {
scheduler: scheduler_config,
em_registry: execution_manager_registry_config,
advertised_endpoint,
stop_timeout_sec,
..
} = config;
let cancellation_token = CancellationToken::new();

let scheduler_id = storage_client
.register(advertised_endpoint.host, advertised_endpoint.port)
.await?;
tracing::info!(scheduler_id = % scheduler_id, "Scheduler registered with storage.");
let dispatch_queue_capacity = scheduler_config.dispatch_queue_capacity();

let (reschedule_queue_sender, reschedule_queue_receiver) =
Expand Down Expand Up @@ -189,15 +188,14 @@ const fn default_stop_timeout_sec() -> u64 {

#[cfg(test)]
mod tests {
use std::net::IpAddr;
use std::net::Ipv4Addr;
use std::num::NonZeroU64;
use std::num::NonZeroUsize;

use async_trait::async_trait;
use spider_core::job::JobState;
use spider_core::types::id::JobId;
use spider_core::types::id::SchedulerId;
use spider_utils::config::Host;

use super::*;
use crate::core_impl::RoundRobinConfig;
Expand All @@ -217,7 +215,7 @@ mod tests {
impl SchedulerStorageClient for MockStorageClient {
async fn register(
&self,
_ip_address: IpAddr,
_host: Host,
_port: u16,
) -> Result<SchedulerId, StorageClientError> {
Ok(SchedulerId::from(SCHEDULER_ID))
Expand Down Expand Up @@ -268,8 +266,11 @@ mod tests {
finalizing_job_expiration_timeout_sec: 60,
}),
em_registry: ExecutionManagerRegistryConfig::default(),
host: IpAddr::V4(Ipv4Addr::LOCALHOST),
port: 0,
advertised_endpoint: EndpointConfig {
host: Host::new("scheduler.example.com".to_owned())
.expect("advertised host should not be empty"),
port: 50052,
},
stop_timeout_sec,
}
}
Expand Down
9 changes: 3 additions & 6 deletions components/spider-scheduler/src/storage_client/grpc.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ use spider_proto_rust::storage::InboundQueueServiceClient;
use spider_proto_rust::storage::JobOrchestrationServiceClient;
use spider_proto_rust::storage::SchedulerRegistrationServiceClient;
use spider_proto_rust::storage::{self};
use spider_utils::config::Host;
use spider_utils::grpc::client::ConnectionPool;
use tonic::Code;
use tonic::Status;
Expand Down Expand Up @@ -74,13 +75,9 @@ impl GrpcSchedulerStorageClient {

#[async_trait]
impl SchedulerStorageClient for GrpcSchedulerStorageClient {
async fn register(
&self,
ip_address: std::net::IpAddr,
port: u16,
) -> Result<SchedulerId, StorageClientError> {
async fn register(&self, host: Host, port: u16) -> Result<SchedulerId, StorageClientError> {
let request = storage::RegisterSchedulerRequest {
ip_address: ip_address.to_string(),
host: host.into_inner(),
port: u32::from(port),
};
let response = self
Expand Down
9 changes: 3 additions & 6 deletions components/spider-scheduler/src/storage_client/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ use spider_core::job::JobState;
use spider_core::types::id::JobId;
use spider_core::types::id::SchedulerId;
use spider_core::types::id::SessionId;
use spider_utils::config::Host;

use crate::error::StorageClientError;
use crate::types::InboundEntry;
Expand All @@ -27,7 +28,7 @@ pub trait SchedulerStorageClient: Send + Sync + Clone {
///
/// # Parameters
///
/// * `ip_address` - The IP address this scheduler is reachable on.
/// * `host` - The host this scheduler is reachable on.
/// * `port` - The port this scheduler is reachable on.
///
/// # Returns
Expand All @@ -41,11 +42,7 @@ pub trait SchedulerStorageClient: Send + Sync + Clone {
/// * [`StorageClientError::Server`] if the storage service returns an error.
/// * [`StorageClientError::Transport`] if the storage transport fails or returns malformed
/// data.
async fn register(
&self,
ip_address: std::net::IpAddr,
port: u16,
) -> Result<SchedulerId, StorageClientError>;
async fn register(&self, host: Host, port: u16) -> Result<SchedulerId, StorageClientError>;

/// Polls the regular-task lane of the storage-owned inbound queue for ready tasks.
///
Expand Down
Loading
Loading