diff --git a/.github/workflows/rust-vip-linux.yml b/.github/workflows/rust-vip-linux.yml new file mode 100644 index 000000000..7460014bd --- /dev/null +++ b/.github/workflows/rust-vip-linux.yml @@ -0,0 +1,68 @@ +name: Rust VIP Linux acceptance + +on: + pull_request: + paths: + - .github/workflows/rust-vip-linux.yml + - rust/crates/control-config/** + - rust/crates/tiproxy-rs/** + - tests/dataplane/integration/vip-standalone.sh + workflow_dispatch: + +permissions: + contents: read + +concurrency: + group: rust-vip-linux-${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true + +jobs: + two-node-vip: + name: Two Linux standalone VIP owners + runs-on: ubuntu-24.04 + timeout-minutes: 45 + env: + DATAPLANE_ARTIFACT_ROOT: ${{ github.workspace }}/tests/dataplane/integration/artifacts + steps: + - name: Check out source + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + - name: Select repository Go toolchain + uses: actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e # v7.0.0 + with: + go-version-file: go.mod + cache: true + - name: Install checksum-pinned TiUP + run: | + source tests/dataplane/integration/versions.env + archive="$RUNNER_TEMP/tiup-v${TIUP_VERSION}-linux-amd64.tar.gz" + install_root="$RUNNER_TEMP/tiup-v${TIUP_VERSION}" + curl --fail --location --retry 3 --proto '=https' --tlsv1.2 \ + --output "$archive" \ + "https://github.com/pingcap/tiup/releases/download/v${TIUP_VERSION}/tiup-v${TIUP_VERSION}-linux-amd64.tar.gz" + echo "$TIUP_LINUX_AMD64_SHA256 $archive" | sha256sum -c - + mkdir -p "$install_root" + tar -xzf "$archive" -C "$install_root" + "$install_root/bin/tiup" --version + echo "$install_root/bin" >>"$GITHUB_PATH" + echo "TIUP_HOME=$install_root" >>"$GITHUB_ENV" + - name: Preinstall frozen TiUP components + run: bash tests/dataplane/integration/warm-tiup-components.sh + - name: Install Linux network and Rust build prerequisites + run: | + sudo apt-get update + sudo apt-get install --yes iproute2 arping tcpdump protobuf-compiler + command -v mysql + - name: Build standalone Rust executable + run: make rust-build + - name: Prove VIP ownership and SQL continuity + run: bash tests/dataplane/integration/vip-standalone.sh + - name: Upload node and election evidence + if: always() + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + with: + name: rust-vip-linux-${{ github.run_id }} + path: tests/dataplane/integration/artifacts/vip-*/diagnostics/** + if-no-files-found: warn + retention-days: 14 diff --git a/rust/crates/control-config/src/lib.rs b/rust/crates/control-config/src/lib.rs index 00d25d817..6e64da766 100644 --- a/rust/crates/control-config/src/lib.rs +++ b/rust/crates/control-config/src/lib.rs @@ -27,7 +27,7 @@ pub use model::{ RoutingBalancePolicy, RoutingConfig, RoutingConnectionFactorConfig, RoutingFactorConfig, RoutingNamespace, RoutingRule, RoutingSelectionPolicy, ServingConfig, ServingKeepalive, ServingListener, ServingNamespace, ServingTlsConfig, SourceRevision, TopologyConfig, - TopologyRuntimeIdentity, + TopologyRuntimeIdentity, VipConfig, }; pub use module::{ ConfigModule, ConfigModuleHandle, ConfigModuleOptions, ConfigMutationError, diff --git a/rust/crates/control-config/src/model.rs b/rust/crates/control-config/src/model.rs index 9649fc1b1..c01c5169a 100644 --- a/rust/crates/control-config/src/model.rs +++ b/rust/crates/control-config/src/model.rs @@ -100,6 +100,45 @@ pub struct ConfigPersistence { pub cluster_tls: ClientTlsConfig, } +/// Inputs for the dedicated VIP election and network owner. +/// +/// The decision to start is pinned at process startup, matching Go's VIP +/// manager. A missing address/interface pair or a non-singleton backend set +/// disables VIP; neither case grants ownership of a partially configured IP. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct VipConfig { + /// Address as configured, including its optional CIDR prefix. + pub address: Arc, + /// Parsed host IP used for the dedicated election key and GARP. + pub ip: IpAddr, + /// Prefix length to compare and bind on the Linux interface. + pub prefix_len: u8, + /// Linux interface that holds the address. + pub interface: Arc, + /// The sole backend cluster's PD endpoints, which own the VIP election. + pub pd_addrs: Arc<[Arc]>, + /// TLS material for that cluster's PD connection. + pub cluster_tls: ClientTlsConfig, + /// Immediate GARP packets in each burst. + pub garp_burst_count: u64, + /// One-second follow-up GARP bursts after takeover. + pub garp_refresh_count: u64, +} + +impl VipConfig { + /// Returns the Go-compatible election prefix for this exact VIP host IP. + #[must_use] + pub fn election_name(&self) -> String { + format!("/tiproxy/vip/{}/owner", self.ip) + } + + /// Returns the canonical address and prefix for Linux address operations. + #[must_use] + pub fn cidr(&self) -> String { + format!("{}/{}", self.ip, self.prefix_len) + } +} + /// Health observer policy consumed by CP-TOPO. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub struct HealthCheckConfig { @@ -596,6 +635,46 @@ impl EffectiveConfig { }) } + /// Projects an enabled VIP only for a single backend cluster. The + /// configured address and interface must both be present. Go elects via + /// that cluster's PD endpoints, which may differ from `proxy.pd-addrs`. + /// + /// # Errors + /// + /// Returns an invalid VIP address, backend normalization error or an + /// unrepresentable GARP count. Config validation already rejects negative + /// counts and gives a zero burst count the Go-compatible value of one. + pub fn vip(&self) -> Result, ConfigError> { + if self.ha.virtual_ip.is_empty() || self.ha.interface.is_empty() { + return Ok(None); + } + let clusters = self.normalized_backend_clusters()?; + let [cluster] = clusters.as_slice() else { + return Ok(None); + }; + let (ip, prefix_len) = parse_vip_address(&self.ha.virtual_ip)?; + let garp_burst_count = + u64::try_from(self.ha.garp_burst_count).map_err(|_| ConfigError::InvalidField { + field: "ha.garp-burst-count", + class: "out_of_range", + })?; + let garp_refresh_count = + u64::try_from(self.ha.garp_refresh_count).map_err(|_| ConfigError::InvalidField { + field: "ha.garp-refresh-count", + class: "out_of_range", + })?; + Ok(Some(VipConfig { + address: Arc::from(self.ha.virtual_ip.as_str()), + ip, + prefix_len, + interface: Arc::from(self.ha.interface.as_str()), + pd_addrs: Arc::clone(&cluster.pd_addrs), + cluster_tls: client_tls(&self.security.cluster_tls), + garp_burst_count, + garp_refresh_count, + })) + } + /// Returns the complete SQL-serving projection formerly built by Go. /// /// # Errors @@ -2424,6 +2503,27 @@ fn normalize_ip_prefix(value: &str) -> Result { } } +fn parse_vip_address(value: &str) -> Result<(IpAddr, u8), ConfigError> { + let failure = || ConfigError::InvalidField { + field: "ha.virtual-ip", + class: "invalid_ip_or_cidr", + }; + let (ip, prefix_len) = if let Some((ip, prefix)) = value.split_once('/') { + ( + ip.parse::().map_err(|_| failure())?, + prefix.parse::().map_err(|_| failure())?, + ) + } else { + let ip = value.parse::().map_err(|_| failure())?; + let prefix_len = if ip.is_ipv4() { 32 } else { 128 }; + (ip, prefix_len) + }; + if prefix_len > if ip.is_ipv4() { 32 } else { 128 } { + return Err(failure()); + } + Ok((ip, prefix_len)) +} + fn path_arc(value: &str) -> Option> { (!value.is_empty()).then(|| Arc::from(Path::new(value))) } diff --git a/rust/crates/control-config/tests/source.rs b/rust/crates/control-config/tests/source.rs index ec98836d7..6e80891f5 100644 --- a/rust/crates/control-config/tests/source.rs +++ b/rust/crates/control-config/tests/source.rs @@ -670,6 +670,68 @@ pd-addrs = "routing-pd:2379" assert_eq!(persistence.pd_addrs[0].as_ref(), "owner-pd:2379"); } +#[test] +fn vip_requires_complete_ha_pair_and_one_backend_cluster() { + let load = |source: &str| { + ConfigNamespaceStore::from_toml(source.as_bytes(), None, current_dir()) + .unwrap_or_else(|error| unreachable!("VIP config: {error}")) + }; + let basic = r#" +[proxy] +pd-addrs = "pd:2379" +[ha] +virtual-ip = "192.0.2.5/24" +interface = "eth0" +garp-burst-count = 0 +garp-refresh-count = 2 +"#; + let store = load(basic); + let vip = store + .current() + .effective() + .vip() + .unwrap_or_else(|error| unreachable!("VIP projection: {error}")) + .unwrap_or_else(|| unreachable!("complete single-cluster VIP is enabled")); + assert_eq!(vip.address.as_ref(), "192.0.2.5/24"); + assert_eq!(vip.interface.as_ref(), "eth0"); + assert_eq!(vip.election_name(), "/tiproxy/vip/192.0.2.5/owner"); + assert_eq!(vip.cidr(), "192.0.2.5/24"); + assert_eq!(vip.garp_burst_count, 1); + assert_eq!(vip.garp_refresh_count, 2); + + let incomplete = load(&basic.replace("interface = \"eth0\"", "interface = \"\"")); + assert!(matches!(incomplete.current().effective().vip(), Ok(None))); + let no_pd = load(&basic.replace("pd-addrs = \"pd:2379\"", "pd-addrs = \"\"")); + assert!(matches!(no_pd.current().effective().vip(), Ok(None))); + let explicit_cluster = load(&basic.replace( + "[ha]", + "[[proxy.backend-clusters]]\nname = \"only\"\npd-addrs = \"cluster-pd:2379\"\n[ha]", + )); + let cluster_vip = explicit_cluster + .current() + .effective() + .vip() + .unwrap_or_else(|error| unreachable!("VIP projection: {error}")) + .unwrap_or_else(|| unreachable!("single explicit backend cluster enables VIP")); + assert_eq!(cluster_vip.pd_addrs[0].as_ref(), "cluster-pd:2379"); + let two_clusters = load( + &basic.replace( + "[ha]", + "[[proxy.backend-clusters]]\nname = \"a\"\npd-addrs = \"pd-a:2379\"\n[[proxy.backend-clusters]]\nname = \"b\"\npd-addrs = \"pd-b:2379\"\n[ha]", + ), + ); + assert!(matches!(two_clusters.current().effective().vip(), Ok(None))); + + let invalid_address = load(&basic.replace("192.0.2.5/24", "192.0.2.5/33")); + assert!(matches!( + invalid_address.current().effective().vip(), + Err(ConfigError::InvalidField { + field: "ha.virtual-ip", + class: "invalid_ip_or_cidr" + }) + )); +} + #[test] fn config_module_rejects_a_factory_that_disagrees_with_initial_transport() { let initial = EtcdClientConfig::new(["initial-pd:2379".to_owned()], None) diff --git a/rust/crates/tiproxy-rs/Cargo.toml b/rust/crates/tiproxy-rs/Cargo.toml index 8ea36e4b9..3995a29d9 100644 --- a/rust/crates/tiproxy-rs/Cargo.toml +++ b/rust/crates/tiproxy-rs/Cargo.toml @@ -28,7 +28,7 @@ if-addrs.workspace = true libc.workspace = true rustix.workspace = true rustls.workspace = true -tokio = { workspace = true, features = ["signal"] } +tokio = { workspace = true, features = ["process", "signal"] } tokio-rustls.workspace = true [dev-dependencies] diff --git a/rust/crates/tiproxy-rs/src/main.rs b/rust/crates/tiproxy-rs/src/main.rs index bf958f812..cf7ce3942 100644 --- a/rust/crates/tiproxy-rs/src/main.rs +++ b/rust/crates/tiproxy-rs/src/main.rs @@ -23,6 +23,8 @@ mod native_meter; mod startup; mod tls_material; mod topology_composition; +mod vip_network; +mod vip_owner; use std::env; use std::fmt::Display; @@ -78,6 +80,8 @@ use topology_composition::{ ArtifactClusterFactory, CompositeCandidateValidator, MetricServerConnectionAcceptor, TopologyCandidateValidator, interface_advertise_candidates, metric_owner_bind_address, }; +use vip_network::{LinuxNetwork, NetworkOperation}; +use vip_owner::{VipHandle, VipModule}; const VERSION: &str = env!("TIPROXY_BUILD_VERSION"); const COMMIT: &str = env!("TIPROXY_BUILD_COMMIT"); @@ -698,6 +702,50 @@ async fn run(options: Options) -> Result<(), String> { .await); } }; + // Go still owns VIP in the two-process configuration. Only the pure-Rust + // standalone process may start the new election, so the two owners cannot + // silently compete for the same address during the migration period. + let vip_plan = if options.standalone { + let vip = match config_owner.handle.source().current().effective().vip() { + Ok(vip) => vip, + Err(error) => return Err(guard.rollback(format!("project VIP config: {error}")).await), + }; + if let Some(vip) = vip { + let roots = tls_material::open_tls_roots(&config_owner.tls_roots); + let tls = match etcd_tls(&vip.cluster_tls, &roots) { + Ok(tls) => tls, + Err(error) => { + return Err(guard.rollback(format!("prepare VIP TLS: {error}")).await); + } + }; + let endpoints = vip + .pd_addrs + .iter() + .map(ToString::to_string) + .collect::>(); + let client = EtcdClientConfig::new(endpoints, tls).and_then(|client| { + client.with_timeouts( + Duration::from_secs(2), + Duration::from_secs(1), + Duration::from_secs(1), + Duration::from_secs(1), + Duration::from_secs(30), + ) + }); + match client { + Ok(client) => Some((vip, client)), + Err(error) => { + return Err(guard + .rollback(format!("prepare VIP etcd client: {error}")) + .await); + } + } + } else { + None + } + } else { + None + }; let metric_bind_address = metric_owner_bind_address(&initial_topology, &metric_advertise_host); let (topology_module, mut topology_handle) = match TopologyModule::new( Arc::new(config_owner.handle.source().clone()), @@ -897,7 +945,7 @@ async fn run(options: Options) -> Result<(), String> { serving.clone(), route_plane_handle, route_config_source, - topology_handle, + topology_handle.clone(), Arc::clone(&runtime_stats), HealthGates { lb_readiness: lb_readiness.clone(), @@ -1065,11 +1113,63 @@ async fn run(options: Options) -> Result<(), String> { .await); } + let mut vip_handle = None; + if let Some((vip, etcd)) = vip_plan { + let network = match LinuxNetwork::new(&vip) { + Ok(network) => Arc::new(network), + Err(error) => { + return Err(guard + .rollback(format!( + "initialize VIP network: {} / {}", + error.operation, error.class + )) + .await); + } + }; + // Match Go Start: clear a stale address before the process becomes + // ready or the dedicated election can be won. + if let Err(error) = network.delete_ip().await { + return Err(guard + .rollback(format!( + "clear stale VIP: {} / {}", + error.operation, error.class + )) + .await); + } + let host = metric_advertise_host.as_ref(); + let member_id = if host.contains(':') { + format!("[{host}]:{}", initial_topology.sql_port) + } else { + format!("{host}:{}", initial_topology.sql_port) + }; + let process_id = runtime_handle.lifecycle().owner_id; + let (module, handle) = match VipModule::new( + &vip, + member_id, + process_id.as_ref(), + etcd, + network, + topology_handle.owner_history(), + ) { + Ok(parts) => parts, + Err(error) => return Err(guard.rollback(error).await), + }; + if let Err(error) = guard.spawn_module(module) { + return Err(guard.rollback(format!("start VIP module: {error}")).await); + } + vip_handle = Some(handle); + } + if let Err(error) = in_process.mark_ready() { meter_ready.send_replace(Some(false)); - return Err(guard - .rollback(format!("mark in-process control runtime ready: {error}")) - .await); + let vip_close = close_vip(&mut vip_handle).await.err(); + let failure = match vip_close { + Some(vip_close) => { + format!("mark in-process control runtime ready: {error}; {vip_close}") + } + None => format!("mark in-process control runtime ready: {error}"), + }; + return Err(guard.rollback(failure).await); } meter_ready.send_replace(Some(true)); if let Some(app) = &admin_app { @@ -1117,12 +1217,13 @@ async fn run(options: Options) -> Result<(), String> { None => std::future::pending().await, } }); - let (control_result, sampler_result, serving_result, module_result) = tokio::select! { + let (control_result, sampler_result, serving_result, module_result, vip_close_result) = tokio::select! { control = &mut control_runtime => { let control = match control { Ok(result) => result.map_err(|error| error.to_string()), Err(_) => Err("control runtime supervisor panicked".to_owned()), }; + let vip_close_result = close_vip(&mut vip_handle).await; if control.is_err() { in_process.fail("legacy_bridge", "runtime_failure"); } else { @@ -1141,13 +1242,14 @@ async fn run(options: Options) -> Result<(), String> { Ok(result) => result.map_err(|error| error.to_string()), Err(_) => Err("metering sampler panicked".to_owned()), }; - (control, sampler, serving_result, Ok(())) + (control, sampler, serving_result, Ok(()), vip_close_result) } sampler = &mut metering_sampler => { let sampler = match sampler { Ok(result) => result.map_err(|error| error.to_string()), Err(_) => Err("metering sampler panicked".to_owned()), }; + let vip_close_result = close_vip(&mut vip_handle).await; if sampler.is_err() { in_process.fail("metering_sampler", "runtime_failure"); } else { @@ -1170,12 +1272,13 @@ async fn run(options: Options) -> Result<(), String> { Ok(result) => result.map_err(|error| error.to_string()), Err(_) => Err("control runtime supervisor panicked".to_owned()), }; - (control, sampler, serving_result, Ok(())) + (control, sampler, serving_result, Ok(()), vip_close_result) } () = &mut termination => { // Close readiness as soon as SIGTERM wins the select, even if an // in-flight config apply briefly delays the applied-snapshot read. lb_readiness.mark_unhealthy(); + let vip_close_result = close_vip(&mut vip_handle).await; let lb_wait = serving.graceful_wait_before_shutdown().await; let event = async { tokio::select! { @@ -1242,13 +1345,14 @@ async fn run(options: Options) -> Result<(), String> { Err(_) => Err("control runtime supervisor panicked".to_owned()), }, }; - (control, sampler, serving_result, module_result) + (control, sampler, serving_result, module_result, vip_close_result) } admin = &mut admin_exit => { let failure = match admin { Ok(()) => "control admin server exited unexpectedly".to_owned(), Err(_) => "control admin server panicked".to_owned(), }; + let vip_close_result = close_vip(&mut vip_handle).await; in_process.fail("control_admin", "runtime_failure"); let serving_result = stop_drain_and_join_sessions( &in_process, @@ -1266,7 +1370,7 @@ async fn run(options: Options) -> Result<(), String> { Ok(result) => result.map_err(|error| error.to_string()), Err(_) => Err("control runtime supervisor panicked".to_owned()), }; - (control, sampler, serving_result, Err(failure)) + (control, sampler, serving_result, Err(failure), vip_close_result) } module = modules.join_next() => { let failure = match module { @@ -1276,6 +1380,7 @@ async fn run(options: Options) -> Result<(), String> { }, None => "control module executor became empty unexpectedly".to_owned(), }; + let vip_close_result = close_vip(&mut vip_handle).await; in_process.fail("control_module", "runtime_failure"); let serving_result = stop_drain_and_join_sessions( &in_process, @@ -1293,7 +1398,7 @@ async fn run(options: Options) -> Result<(), String> { Ok(result) => result.map_err(|error| error.to_string()), Err(_) => Err("control runtime supervisor panicked".to_owned()), }; - (control, sampler, serving_result, Err(failure)) + (control, sampler, serving_result, Err(failure), vip_close_result) } }; if let Some(observer) = routing_shadow { @@ -1334,6 +1439,7 @@ async fn run(options: Options) -> Result<(), String> { prefer_control_failure(control_result, sampler_result)?; serving_result?; module_result?; + vip_close_result?; module_executor_result?; finish_result?; Ok(()) @@ -1347,6 +1453,18 @@ fn prefer_control_failure( sampler } +async fn close_vip(handle: &mut Option) -> Result<(), String> { + let Some(handle) = handle.as_mut() else { + return Ok(()); + }; + handle.pre_close().await.map_err(|error| { + format!( + "remove VIP before SQL drain: {} / {}", + error.operation, error.class + ) + }) +} + enum OrderedWaitExit { Control(Result<(), String>), Sampler(Result<(), String>), diff --git a/rust/crates/tiproxy-rs/src/vip_network.rs b/rust/crates/tiproxy-rs/src/vip_network.rs new file mode 100644 index 000000000..e4cb490ab --- /dev/null +++ b/rust/crates/tiproxy-rs/src/vip_network.rs @@ -0,0 +1,277 @@ +// Copyright 2026 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// 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. + +//! Linux virtual-IP address operations behind a replaceable owner boundary. + +use std::future::Future; +use std::net::IpAddr; +use std::pin::Pin; +use std::sync::Arc; +use std::time::Duration; + +use control_config::VipConfig; +use tokio::process::Command; + +const COMMAND_TIMEOUT: Duration = Duration::from_secs(2); + +type NetworkFuture<'a, T> = Pin> + Send + 'a>>; + +/// The VIP actor uses the same interface for a real Linux host and a fake. +pub(super) trait NetworkOperation: Send + Sync { + fn has_ip(&self) -> NetworkFuture<'_, bool>; + fn add_ip(&self) -> NetworkFuture<'_, ()>; + fn delete_ip(&self) -> NetworkFuture<'_, ()>; + fn send_arp(&self) -> NetworkFuture<'_, ()>; +} + +/// Bounded class of one failed Linux VIP operation; no command output leaks. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) struct NetworkError { + pub operation: &'static str, + pub class: &'static str, +} + +/// IP address and interface configured for this process's VIP owner. +pub(super) struct LinuxNetwork { + ip: IpAddr, + prefix_len: u8, + interface: Arc, + burst_count: u64, +} + +impl LinuxNetwork { + pub fn new(config: &VipConfig) -> Result { + if !cfg!(target_os = "linux") { + return Err(NetworkError { + operation: "initialize", + class: "unsupported_os", + }); + } + Ok(Self { + ip: config.ip, + prefix_len: config.prefix_len, + interface: Arc::clone(&config.interface), + burst_count: config.garp_burst_count, + }) + } + + fn cidr(&self) -> String { + format!("{}/{}", self.ip, self.prefix_len) + } + + async fn ip_output( + &self, + args: &[&str], + operation: &'static str, + ) -> Result { + run_command("ip", args, operation).await + } + + async fn mutate(&self, verb: &'static str) -> Result<(), NetworkError> { + let cidr = self.cidr(); + let args = ["addr", verb, cidr.as_str(), "dev", self.interface.as_ref()]; + let output = self.ip_output(&args, verb).await?; + if output.success { + return Ok(()); + } + // Go retries netlink AddrAdd/AddrDel through sudo only for EPERM. + // The `ip` CLI emits this stable English text with LC_ALL=C. + if !output.stderr.contains("Operation not permitted") + && !output.stderr.contains("Permission denied") + { + return Err(NetworkError { + operation: verb, + class: "ip_failed", + }); + } + let sudo_args = [ + "-n", + "ip", + "addr", + verb, + cidr.as_str(), + "dev", + self.interface.as_ref(), + ]; + let retry = run_command("sudo", &sudo_args, verb).await?; + retry.success.then_some(()).ok_or(NetworkError { + operation: verb, + class: "sudo_ip_failed", + }) + } +} + +impl NetworkOperation for LinuxNetwork { + fn has_ip(&self) -> NetworkFuture<'_, bool> { + Box::pin(async move { + let args = ["-o", "addr", "show", "dev", self.interface.as_ref()]; + let output = self.ip_output(&args, "list").await?; + if !output.success { + return Err(NetworkError { + operation: "list", + class: "ip_failed", + }); + } + Ok(contains_exact_address( + &output.stdout, + self.ip, + self.prefix_len, + )) + }) + } + + fn add_ip(&self) -> NetworkFuture<'_, ()> { + Box::pin(async move { + if self.has_ip().await? { + return Ok(()); + } + self.mutate("add").await + }) + } + + fn delete_ip(&self) -> NetworkFuture<'_, ()> { + Box::pin(async move { + if !self.has_ip().await? { + return Ok(()); + } + self.mutate("del").await + }) + } + + fn send_arp(&self) -> NetworkFuture<'_, ()> { + Box::pin(async move { + if !self.ip.is_ipv4() { + return Err(NetworkError { + operation: "garp", + class: "not_ipv4", + }); + } + let ip = self.ip.to_string(); + for _ in 0..self.burst_count { + let mut sent = false; + // iputils uses -I for interface and chooses the VIP as the + // source of -U announcements. Thomas Habets' arping uses -i + // and needs -S VIP as well; otherwise it announces the node + // address rather than the VIP. Try direct and sudo for each. + for interface_flag in ["-I", "-i"] { + let mut args = vec!["-c", "1", "-U", interface_flag, self.interface.as_ref()]; + if interface_flag == "-i" { + args.extend(["-S", ip.as_str()]); + } + args.push(ip.as_str()); + if run_command("arping", &args, "garp") + .await + .is_ok_and(|output| output.success) + { + sent = true; + break; + } + let mut sudo_args = vec!["-n", "arping"]; + sudo_args.extend(args); + if run_command("sudo", &sudo_args, "garp") + .await + .is_ok_and(|output| output.success) + { + sent = true; + break; + } + } + if !sent { + return Err(NetworkError { + operation: "garp", + class: "arping_failed", + }); + } + } + Ok(()) + }) + } +} + +struct Output { + success: bool, + stdout: String, + stderr: String, +} + +async fn run_command( + program: &str, + args: &[&str], + operation: &'static str, +) -> Result { + let mut command = Command::new(program); + command.args(args).env("LC_ALL", "C").kill_on_drop(true); + let output = tokio::time::timeout(COMMAND_TIMEOUT, command.output()) + .await + .map_err(|_| NetworkError { + operation, + class: "timeout", + })? + .map_err(|error| NetworkError { + operation, + class: if error.kind() == std::io::ErrorKind::NotFound { + "command_missing" + } else { + "command_io" + }, + })?; + Ok(Output { + success: output.status.success(), + stdout: String::from_utf8_lossy(&output.stdout).into_owned(), + stderr: String::from_utf8_lossy(&output.stderr).into_owned(), + }) +} + +fn contains_exact_address(output: &str, ip: IpAddr, prefix_len: u8) -> bool { + output.lines().any(|line| { + let mut words = line.split_whitespace(); + while let Some(word) = words.next() { + if word != "inet" && word != "inet6" { + continue; + } + let Some((address, prefix)) = words.next().and_then(|value| value.split_once('/')) + else { + continue; + }; + if address.parse::().ok() == Some(ip) + && prefix.parse::().ok() == Some(prefix_len) + { + return true; + } + } + false + }) +} + +#[cfg(test)] +mod tests { + use super::contains_exact_address; + use std::net::IpAddr; + + #[test] + fn exact_ip_and_prefix_match_is_strict() { + let output = "2: eth0 inet 192.0.2.5/24 brd 192.0.2.255 scope global eth0\n\ + 2: eth0 inet6 2001:db8::5/64 scope global\n"; + let v4: IpAddr = "192.0.2.5" + .parse() + .unwrap_or_else(|error| unreachable!("test IP: {error}")); + let v6: IpAddr = "2001:db8::5" + .parse() + .unwrap_or_else(|error| unreachable!("test IP: {error}")); + assert!(contains_exact_address(output, v4, 24)); + assert!(!contains_exact_address(output, v4, 32)); + assert!(contains_exact_address(output, v6, 64)); + assert!(!contains_exact_address(output, v6, 128)); + } +} diff --git a/rust/crates/tiproxy-rs/src/vip_owner.rs b/rust/crates/tiproxy-rs/src/vip_owner.rs new file mode 100644 index 000000000..1fc737b7d --- /dev/null +++ b/rust/crates/tiproxy-rs/src/vip_owner.rs @@ -0,0 +1,503 @@ +// Copyright 2026 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// 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. + +//! Dedicated, process-local VIP election with an awaited pre-close boundary. + +use std::sync::Arc; +use std::time::Duration; + +use control_config::VipConfig; +use control_etcd::{ElectionConfig, ElectionSession, ElectionState}; +use control_external::EtcdClientConfig; +use control_plane::logging::{self, Level}; +use control_plane::{ControlModule, LifecyclePhase, ModuleContext, ModuleError, ModuleFuture}; +use control_topology::ElectionOwnerHistory; +use tokio::sync::watch; + +use crate::vip_network::{NetworkError, NetworkOperation}; + +const TTL_SECONDS: i64 = 3; +const RETRY_DELAY: Duration = Duration::from_millis(500); +const PRE_CLOSE_TIMEOUT: Duration = Duration::from_secs(15); + +pub(super) struct VipModule { + network: Arc, + etcd: EtcdClientConfig, + election: ElectionConfig, + election_name: String, + refresh_count: u64, + stop: watch::Receiver, + closed: watch::Sender>>, + history: Arc, +} + +pub(super) struct VipHandle { + network: Arc, + stop: watch::Sender, + closed: watch::Receiver>>, +} + +impl VipModule { + pub fn new( + config: &VipConfig, + member_id: String, + process_id: &str, + etcd: EtcdClientConfig, + network: Arc, + history: Arc, + ) -> Result<(Self, VipHandle), String> { + let election_name = config.election_name(); + let session_key = format!("/tiproxy/vip/{}/session/{process_id}", config.ip); + let election = + ElectionConfig::new(election_name.clone(), member_id, session_key, TTL_SECONDS) + .map_err(|error| format!("validate VIP election: {error}"))?; + let (stop_tx, stop) = watch::channel(false); + let (closed, closed_rx) = watch::channel(None); + Ok(( + Self { + network: Arc::clone(&network), + etcd, + election, + election_name, + refresh_count: config.garp_refresh_count, + stop, + closed, + history, + }, + VipHandle { + network, + stop: stop_tx, + closed: closed_rx, + }, + )) + } + + async fn serve( + mut self, + mut lifecycle: watch::Receiver, + owner: control_plane::OwnerToken, + ) -> Result<(), ModuleError> { + // Go removes a stale local VIP before entering the election. Failing + // here is safer than campaigning with an address this process cannot + // remove on retirement. + self.network.delete_ip().await.map_err(network_failure)?; + // The module is spawned during guarded startup. A winner must not + // advertise this node before SQL serving has crossed the ready gate. + while lifecycle.borrow().phase != LifecyclePhase::Ready { + if *self.stop.borrow() || shutting_down(&lifecycle) { + return Ok(()); + } + tokio::select! { + _ = self.stop.changed() => return Ok(()), + changed = lifecycle.changed() => { + if changed.is_err() { + return Ok(()); + } + } + } + } + loop { + if *self.stop.borrow() || shutting_down(&lifecycle) { + return Ok(()); + } + let campaign = tokio::select! { + biased; + _ = self.stop.changed() => return Ok(()), + _ = lifecycle.changed() => { + if shutting_down(&lifecycle) { + return Ok(()); + } + continue; + }, + result = ElectionSession::campaign(owner.clone(), self.etcd.clone(), self.election.clone()) => result, + }; + let mut session = match campaign { + Ok(session) => session, + Err(error) => { + logging::emit(Level::Warn, &format!("VIP campaign retry: {error}")); + tokio::select! { + _ = self.stop.changed() => return Ok(()), + _ = lifecycle.changed() => continue, + () = tokio::time::sleep(RETRY_DELAY) => continue, + } + } + }; + let _owned = OwnedElection::new(Arc::clone(&self.history), self.election_name.clone()); + // Linux commands can outlast the 3s lease. Keep it alive while + // binding; dropping an in-flight command kills its child, and the + // retirement delete below covers a command that already finished. + let add = self.bind_while_owned(&mut session, &mut lifecycle).await; + if let Err(error) = add { + self.delete_while_owned(&mut session) + .await + .map_err(network_failure)?; + let _ = session.shutdown().await; + return Err(network_failure(error)); + } + if matches!(add, Ok(true)) { + logging::emit(Level::Info, "VIP election won and address bound"); + Box::pin(self.maintain(&mut session, &mut lifecycle)).await; + } + // This awaited delete is the retirement boundary. Never resign the + // lease while our local interface still has the VIP. Keepalive + // continues because a slow `ip`/`sudo` command can outlast TTL. + self.delete_while_owned(&mut session) + .await + .map_err(network_failure)?; + let _ = session.shutdown().await; + logging::emit(Level::Info, "VIP retired and address removed"); + if *self.stop.borrow() || shutting_down(&lifecycle) { + return Ok(()); + } + tokio::select! { + _ = self.stop.changed() => return Ok(()), + _ = lifecycle.changed() => {}, + () = tokio::time::sleep(RETRY_DELAY) => {} + } + } + } + + async fn bind_while_owned( + &mut self, + session: &mut ElectionSession, + lifecycle: &mut watch::Receiver, + ) -> Result { + let network = Arc::clone(&self.network); + let mut add = Box::pin(network.add_ip()); + let mut heartbeat = tokio::time::interval(Duration::from_secs(1)); + heartbeat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + tokio::select! { + biased; + _ = self.stop.changed() => return Ok(false), + changed = lifecycle.changed() => { + if changed.is_err() || shutting_down(lifecycle) { + return Ok(false); + } + }, + _ = heartbeat.tick() => { + if session.keep_alive().await.is_err() + || session.snapshot().state != ElectionState::Leader + { + return Ok(false); + } + }, + result = &mut add => return result.map(|()| true), + } + } + } + + async fn delete_while_owned(&self, session: &mut ElectionSession) -> Result<(), NetworkError> { + let network = Arc::clone(&self.network); + let mut delete = Box::pin(network.delete_ip()); + let mut heartbeat = tokio::time::interval_at( + tokio::time::Instant::now() + Duration::from_secs(1), + Duration::from_secs(1), + ); + heartbeat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + let mut lease_alive = session.snapshot().state == ElectionState::Leader; + loop { + tokio::select! { + biased; + result = &mut delete => return result, + _ = heartbeat.tick(), if lease_alive => { + lease_alive = session.keep_alive().await.is_ok() + && session.snapshot().state == ElectionState::Leader; + if !lease_alive { + logging::emit(Level::Warn, "VIP lease uncertain during address removal"); + } + }, + } + } + } + + async fn maintain( + &mut self, + session: &mut ElectionSession, + lifecycle: &mut watch::Receiver, + ) { + let mut heartbeat = tokio::time::interval(Duration::from_secs(1)); + heartbeat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + let network = Arc::clone(&self.network); + let mut arp = Some(network.send_arp()); + let mut first_arp = true; + let mut refresh: Option = None; + let mut remaining = self.refresh_count; + while !*self.stop.borrow() && !shutting_down(lifecycle) { + enum Event { + Stop, + Heartbeat, + Watch, + Refresh, + Arp(Result<(), NetworkError>), + } + let event = tokio::select! { + biased; + _ = self.stop.changed() => Event::Stop, + _ = lifecycle.changed() => Event::Stop, + _ = heartbeat.tick() => Event::Heartbeat, + result = async { + if let Some(arp) = arp.as_mut() { + arp.await + } else { + std::future::pending().await + } + }, if arp.is_some() => Event::Arp(result), + result = session.watch_once() => { + if result.is_err() { Event::Stop } else { Event::Watch } + }, + () = async { + if let Some(refresh) = refresh.as_mut() { + refresh.tick().await; + } else { + std::future::pending::<()>().await; + } + }, if refresh.is_some() && arp.is_none() && remaining > 0 => Event::Refresh, + }; + match event { + Event::Stop => break, + Event::Heartbeat => { + if session.keep_alive().await.is_err() { + // Uncertain ownership cannot leave an address bound + // until a remote lease may expire and a peer takes it. + break; + } + } + Event::Watch => {} + Event::Refresh => { + remaining -= 1; + arp = Some(network.send_arp()); + } + Event::Arp(result) => { + arp = None; + if let Err(error) = result { + logging::emit( + Level::Warn, + &format!( + "VIP GARP burst failed: {} / {}", + error.operation, error.class + ), + ); + if !first_arp { + remaining = 0; + } + } + if first_arp { + first_arp = false; + refresh = Some(tokio::time::interval_at( + tokio::time::Instant::now() + Duration::from_secs(1), + Duration::from_secs(1), + )); + } + } + } + if matches!( + session.snapshot().state, + ElectionState::Uncertain | ElectionState::Retired | ElectionState::Stopped + ) { + break; + } + } + } +} + +impl ControlModule for VipModule { + fn name(&self) -> &'static str { + "vip_owner" + } + + fn run(self: Box, context: ModuleContext) -> ModuleFuture { + Box::pin(async move { + let network = Arc::clone(&self.network); + let closed = self.closed.clone(); + let result = Box::pin(self.serve(context.lifecycle(), context.owner().clone())).await; + // A final delete covers an interrupted add and gives pre-close a + // direct result even if the election loop failed before it won. + let cleanup = network.delete_ip().await; + closed.send_replace(Some(cleanup)); + match (result, cleanup) { + (Err(error), _) => Err(error), + (Ok(()), Err(error)) => Err(network_failure(error)), + (Ok(()), Ok(())) => Ok(()), + } + }) + } +} + +impl VipHandle { + /// Removes the local VIP and waits for the actor to retire before the SQL + /// listener enters drain. A timed-out actor is never reported as closed: + /// it may still have an in-flight add after a direct deletion. + pub async fn pre_close(&mut self) -> Result<(), NetworkError> { + self.stop.send_replace(true); + let wait = async { + loop { + if let Some(result) = *self.closed.borrow() { + return result; + } + if self.closed.changed().await.is_err() { + return self.network.delete_ip().await; + } + } + }; + if let Ok(result) = tokio::time::timeout(PRE_CLOSE_TIMEOUT, wait).await { + match result { + Ok(()) => Ok(()), + Err(_) => self.network.delete_ip().await, + } + } else { + self.network.delete_ip().await?; + Err(NetworkError { + operation: "pre_close", + class: "actor_timeout", + }) + } + } +} + +struct OwnedElection { + history: Arc, + key: String, +} + +impl OwnedElection { + fn new(history: Arc, key: String) -> Self { + history.won(&key); + Self { history, key } + } +} + +impl Drop for OwnedElection { + fn drop(&mut self) { + self.history.retired(&self.key); + } +} + +fn shutting_down(lifecycle: &watch::Receiver) -> bool { + matches!( + lifecycle.borrow().phase, + LifecyclePhase::Quiescing + | LifecyclePhase::Draining + | LifecyclePhase::Stopping + | LifecyclePhase::Stopped + | LifecyclePhase::Failed + ) +} + +fn network_failure(error: NetworkError) -> ModuleError { + logging::emit( + Level::Error, + &format!( + "VIP network operation failed: {} / {}", + error.operation, error.class + ), + ); + ModuleError { + module: "vip_owner", + error_class: "network_failed", + } +} + +#[cfg(test)] +mod tests { + use std::future::Future; + use std::pin::Pin; + use std::sync::Arc; + use std::sync::atomic::{AtomicUsize, Ordering}; + + use tokio::sync::watch; + + use super::VipHandle; + use crate::vip_network::{NetworkError, NetworkOperation}; + + type NetworkFuture<'a, T> = Pin> + Send + 'a>>; + + #[derive(Default)] + struct FakeNetwork { + deletes: AtomicUsize, + } + + impl NetworkOperation for FakeNetwork { + fn has_ip(&self) -> NetworkFuture<'_, bool> { + Box::pin(async { Ok(false) }) + } + + fn add_ip(&self) -> NetworkFuture<'_, ()> { + Box::pin(async { Ok(()) }) + } + + fn delete_ip(&self) -> NetworkFuture<'_, ()> { + Box::pin(async { + self.deletes.fetch_add(1, Ordering::SeqCst); + Ok(()) + }) + } + + fn send_arp(&self) -> NetworkFuture<'_, ()> { + Box::pin(async { Ok(()) }) + } + } + + #[tokio::test] + async fn pre_close_waits_for_actor_retirement_before_reporting_success() { + let network = Arc::new(FakeNetwork::default()); + let (stop, mut stop_rx) = watch::channel(false); + let (closed, closed_rx) = watch::channel(None); + let mut handle = VipHandle { + network: network.clone(), + stop, + closed: closed_rx, + }; + let close = tokio::spawn(async move { handle.pre_close().await }); + assert!(stop_rx.changed().await.is_ok()); + assert!(*stop_rx.borrow()); + assert!(!close.is_finished(), "SQL drain must await VIP retirement"); + closed.send_replace(Some(Ok(()))); + assert!(matches!(close.await, Ok(Ok(())))); + assert_eq!(network.deletes.load(Ordering::SeqCst), 0); + } + + #[tokio::test] + async fn pre_close_deletes_after_actor_exits_without_an_ack() { + let network = Arc::new(FakeNetwork::default()); + let (stop, _stop_rx) = watch::channel(false); + let (closed, closed_rx) = watch::channel(None); + drop(closed); + let mut handle = VipHandle { + network: network.clone(), + stop, + closed: closed_rx, + }; + assert!(handle.pre_close().await.is_ok()); + assert_eq!(network.deletes.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn pre_close_retries_delete_after_actor_cleanup_failure() { + let network = Arc::new(FakeNetwork::default()); + let (stop, _stop_rx) = watch::channel(false); + let (closed, closed_rx) = watch::channel(Some(Err(NetworkError { + operation: "del", + class: "ip_failed", + }))); + let mut handle = VipHandle { + network: network.clone(), + stop, + closed: closed_rx, + }; + assert!(handle.pre_close().await.is_ok()); + assert_eq!(network.deletes.load(Ordering::SeqCst), 1); + drop(closed); + } +} diff --git a/tests/dataplane/integration/vip-standalone.sh b/tests/dataplane/integration/vip-standalone.sh new file mode 100644 index 000000000..7e1763369 --- /dev/null +++ b/tests/dataplane/integration/vip-standalone.sh @@ -0,0 +1,435 @@ +#!/usr/bin/env bash +# Copyright 2026 PingCAP, Inc. +# SPDX-License-Identifier: Apache-2.0 + +# Linux-only, privileged two-node VIP smoke. Each Rust process owns its own +# network namespace, so both may bind the same SQL port and only the elected +# process may bind the shared VIP on its eth0. The bridge gives both processes +# a real PD/TiDB and gives the host a route to query the elected VIP. +set -euo pipefail + +if [[ $(uname -s) != Linux ]]; then + echo 'VIP acceptance requires Linux network namespaces' >&2 + exit 2 +fi +for command in ip arping tcpdump tiup mysql curl; do + command -v "$command" >/dev/null || { echo "missing $command" >&2; exit 2; } +done + +script_dir=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) +repo_root=$(cd "$script_dir/../../.." && pwd) +source "$script_dir/versions.env" +rust_binary="$repo_root/rust/target/debug/tiproxy-rs" +[[ -x $rust_binary ]] || { echo "missing $rust_binary" >&2; exit 2; } + +run_id=${GITHUB_RUN_ID:-local-$$} +run_dir=${DATAPLANE_ARTIFACT_ROOT:-$script_dir/artifacts}/vip-$run_id +mkdir -p "$run_dir/diagnostics" +run_dir=$(cd "$run_dir" && pwd) +tag=vip-$run_id +bridge=br-vip-test +ns_a=vip-a-$run_id +ns_b=vip-b-$run_id +link_id=${run_id: -6} +host_a=vha-$link_id +host_b=vhb-$link_id +host_ip=198.19.99.1 +vip_ip=198.19.99.100 +node_a_ip=198.19.99.11 +node_b_ip=198.19.99.12 +offset=13000 +pd_port=$((2379 + offset)) +tidb_port=$((4000 + offset)) +tiup_pid= +monitor_pid= +capture_pid= + +owned_rust_pid() { + local node=$1 pid=$2 + [[ $pid =~ ^[0-9]+$ ]] || return 1 + sudo ps -ww -p "$pid" -o args= 2>/dev/null | + grep -Fq -- "$run_dir/$node.toml" +} + +cleanup() { + local status=$? + trap - EXIT + set +e + [[ -n $monitor_pid ]] && kill "$monitor_pid" 2>/dev/null + [[ -n $capture_pid ]] && kill "$capture_pid" 2>/dev/null + [[ -s $run_dir/diagnostics/overlap.log ]] && status=1 + for node in a b; do + pidfile=$run_dir/$node.pid + if [[ -f $pidfile ]]; then + pid=$(cat "$pidfile") + owned_rust_pid "$node" "$pid" && sudo kill -INT "$pid" 2>/dev/null + fi + done + sleep 2 + for node in a b; do + pidfile=$run_dir/$node.pid + if [[ -f $pidfile ]]; then + pid=$(cat "$pidfile") + owned_rust_pid "$node" "$pid" && sudo kill -KILL "$pid" 2>/dev/null + fi + done + if [[ -n $tiup_pid ]]; then + tiup clean "$tag" >>"$run_dir/diagnostics/cleanup.log" 2>&1 + kill -INT "$tiup_pid" 2>/dev/null + fi + sudo ip netns del "$ns_a" 2>/dev/null + sudo ip netns del "$ns_b" 2>/dev/null + sudo ip link del "$bridge" 2>/dev/null + echo "exit=$status" >>"$run_dir/diagnostics/result.txt" + exit "$status" +} +trap cleanup EXIT + +sudo ip link add "$bridge" type bridge +sudo ip addr add "$host_ip/24" dev "$bridge" +sudo ip link set "$bridge" up + +attach_node() { + local namespace=$1 host_link=$2 address=$3 mac_suffix=$4 peer=peer0 + sudo ip link add "$host_link" type veth peer name "$peer" + sudo ip link set "$peer" netns "$namespace" + # GitHub's runner assigned the same generated MAC to both peers in the + # first acceptance runs. A shared MAC lets the bridge learn the wrong port + # for the VIP even though exactly one namespace owns its IP address. + sudo ip link set "$host_link" address "02:00:00:99:10:$mac_suffix" + sudo ip link set "$host_link" master "$bridge" + sudo ip link set "$host_link" up + sudo ip -n "$namespace" link set "$peer" name eth0 + sudo ip -n "$namespace" link set eth0 address "02:00:00:99:20:$mac_suffix" + sudo ip -n "$namespace" link set eth0 up + sudo ip -n "$namespace" addr add "$address/24" dev eth0 + sudo ip -n "$namespace" route add default via "$host_ip" +} + +sudo ip netns add "$ns_a" +sudo ip netns add "$ns_b" +sudo ip -n "$ns_a" link set lo up +sudo ip -n "$ns_b" link set lo up +attach_node "$ns_a" "$host_a" "$node_a_ip" 11 +attach_node "$ns_b" "$host_b" "$node_b_ip" 12 + +# The backend playground is deliberately on the host's bridge address so a +# namespace's 127.0.0.1 can never accidentally bypass the veth network. +cat >"$run_dir/tidb.toml" <<'CONFIG' +[labels] +keyspace = "vip-ks" +CONFIG +tiup "playground:v${TIUP_VERSION}" "$TIDB_VERSION" --tag "$tag" \ + --without-monitor --host "$host_ip" --port-offset "$offset" \ + --pd 1 --kv 1 --db 1 --db.config "$run_dir/tidb.toml" \ + --tiflash 0 --tiproxy 0 \ + >"$run_dir/diagnostics/tiup.log" 2>&1 & +tiup_pid=$! + +wait_mysql() { + local host=$1 port=$2 label=$3 + local result + for _ in {1..120}; do + if result=$(mysql --protocol=TCP --connect-timeout=1 --ssl-mode=DISABLED \ + -h "$host" -P "$port" -u root --batch --skip-column-names \ + -e 'SELECT 1' 2>&1) && [[ $result == 1 ]]; then + echo "$(date +%s.%N) $label SELECT 1 PASS" >>"$run_dir/diagnostics/events.log" + return 0 + fi + printf '%s\n' "$result" >"$run_dir/diagnostics/$label-mysql-last-error.txt" + kill -0 "$tiup_pid" 2>/dev/null || break + sleep 1 + done + if declare -F snapshot >/dev/null; then + snapshot "$label-failed" + fi + echo "$label SQL query failed" >&2 + return 1 +} +wait_mysql "$host_ip" "$tidb_port" backend + +# The runner must prove each network namespace can reach the same backend +# before a failed VIP query is attributed to routing in the Rust processes. +for namespace in "$ns_a" "$ns_b"; do + if ! sudo ip netns exec "$namespace" mysql --protocol=TCP \ + --connect-timeout=2 --ssl-mode=DISABLED -h "$host_ip" -P "$tidb_port" \ + -u root --batch --skip-column-names -e 'SELECT 1' \ + >"$run_dir/diagnostics/$namespace-backend-preflight.txt" 2>&1; then + echo "$namespace cannot query TiDB from its network namespace" >&2 + exit 1 + fi +done + +# Standalone metering requires a nonempty backend keyspace before forwarding +# SQL. TiDB publishes this label in its classic /topology/tidb info record. +ctl=${TIUP_HOME:-$HOME/.tiup}/components/ctl/$TIDB_VERSION/etcdctl +[[ -x $ctl ]] || { echo "missing $ctl" >&2; exit 2; } +topology="$run_dir/diagnostics/backend-topology.json" +registered=0 +for _ in {1..30}; do + ETCDCTL_API=3 "$ctl" --endpoints "http://$host_ip:$pd_port" \ + get '/topology/tidb/' --prefix -w json >"$topology" + if python3 - "$topology" <<'PY' +import base64, json, sys +with open(sys.argv[1], encoding="utf-8") as source: + entries = json.load(source).get("kvs", []) +for entry in entries: + key = base64.b64decode(entry["key"]) + if not key.endswith(b"/info"): + continue + info = json.loads(base64.b64decode(entry["value"])) + if info.get("labels", {}).get("keyspace") == "vip-ks": + sys.exit(0) +sys.exit(1) +PY + then + registered=1 + break + fi + sleep 1 +done +[[ $registered == 1 ]] || { echo 'TiDB keyspace label missing from PD topology' >&2; exit 1; } + +# Keep passive on-wire ARP evidence separate from SQL continuity. In +# particular, some arping versions report failure for a sent unsolicited +# packet because no reply is expected; an exit code alone cannot prove GARP. +arping -V >"$run_dir/diagnostics/arping-version.txt" 2>&1 || true +sudo timeout 40s tcpdump -e -l -nn -tt -i "$bridge" arp \ + >"$run_dir/diagnostics/arp-packets.txt" \ + 2>"$run_dir/diagnostics/tcpdump-stderr.txt" & +capture_pid=$! +sleep 1 +kill -0 "$capture_pid" + +write_config() { + local node=$1 address=$2 + mkdir -p "$run_dir/$node-work" + cat >"$run_dir/$node.toml" <"$1"; shift; exec "$@"' \ + sh "$run_dir/$node.pid" \ + "$rust_binary" --standalone \ + --config "$run_dir/$node.toml" --health-port 8080 \ + >"$run_dir/diagnostics/$node.log" 2>&1 & +} +has_vip() { + local namespace=$1 + sudo ip -n "$namespace" -o addr show dev eth0 2>/dev/null | + grep -Fq "inet $vip_ip/24 " +} + +monitor_overlap() { + local a_bound b_bound + while :; do + a_bound=0 + b_bound=0 + has_vip "$ns_a" && a_bound=1 + has_vip "$ns_b" && b_bound=1 + if ((a_bound + b_bound > 1)); then + echo "$(date +%s.%N) overlap" >>"$run_dir/diagnostics/overlap.log" + return 1 + fi + sleep 0.1 + done +} + +assert_no_overlap() { + if [[ -s $run_dir/diagnostics/overlap.log ]]; then + echo 'two VIP holders observed by background sampler' >&2 + return 1 + fi +} + +garp_count() { + local node=$1 mac + if [[ $node == a ]]; then + mac=02:00:00:99:20:11 + else + mac=02:00:00:99:20:12 + fi + awk -v vip="$vip_ip" -v mac="$mac" ' + index($0, mac) && index($0, "who-has " vip) && index($0, "tell " vip) { count++ } + END { print count + 0 } + ' "$run_dir/diagnostics/arp-packets.txt" +} + +wait_garp() { + local label=$1 node=$2 previous=$3 count + for _ in {1..30}; do + count=$(garp_count "$node") + if ((count > previous)); then + echo "$(date +%s.%N) $label $node GARP observed" \ + >>"$run_dir/diagnostics/events.log" + return 0 + fi + sleep 0.2 + done + echo "$label: no VIP-source GARP from $node observed on the bridge" >&2 + return 1 +} + +monitor_overlap & +monitor_pid=$! +start_node a "$ns_a" +start_node b "$ns_b" + +wait_ready() { + local node=$1 namespace=$2 + for _ in {1..60}; do + if sudo ip netns exec "$namespace" curl --fail --silent --max-time 1 \ + 'http://127.0.0.1:8080/health' \ + >"$run_dir/diagnostics/$node-health-ready.json"; then + echo "$(date +%s.%N) $node ready" >>"$run_dir/diagnostics/events.log" + return 0 + fi + if [[ -f $run_dir/$node.pid ]]; then + sudo kill -0 "$(cat "$run_dir/$node.pid")" 2>/dev/null || break + fi + sleep 1 + done + echo "$node Rust process did not become ready" >&2 + return 1 +} +wait_ready a "$ns_a" +wait_ready b "$ns_b" + +snapshot() { + local label=$1 + sudo ip -j -n "$ns_a" addr show >"$run_dir/diagnostics/$label-a-addr.json" + sudo ip -j -n "$ns_b" addr show >"$run_dir/diagnostics/$label-b-addr.json" + sudo ip netns exec "$ns_a" curl --silent --max-time 2 \ + 'http://127.0.0.1:8080/health' \ + >"$run_dir/diagnostics/$label-a-health.json" || true + sudo ip netns exec "$ns_b" curl --silent --max-time 2 \ + 'http://127.0.0.1:8080/health' \ + >"$run_dir/diagnostics/$label-b-health.json" || true + local ctl=${TIUP_HOME:-$HOME/.tiup}/components/ctl/$TIDB_VERSION/etcdctl + if [[ -x $ctl ]]; then + ETCDCTL_API=3 "$ctl" --endpoints "http://$host_ip:$pd_port" \ + get "/tiproxy/vip/$vip_ip/owner" --prefix -w json \ + >"$run_dir/diagnostics/$label-election.json" 2>&1 || true + ETCDCTL_API=3 "$ctl" --endpoints "http://$host_ip:$pd_port" \ + get '/topology/tidb/' --prefix -w json \ + >"$run_dir/diagnostics/$label-tidb-topology.json" 2>&1 || true + fi +} + +wait_single_owner() { + local label=$1 expected=${2:-either} a_bound b_bound + for _ in {1..45}; do + a_bound=0 + b_bound=0 + has_vip "$ns_a" && a_bound=1 + has_vip "$ns_b" && b_bound=1 + if ((a_bound + b_bound == 1)); then + if [[ $expected == either || ($expected == a && $a_bound == 1) || + ($expected == b && $b_bound == 1) ]]; then + snapshot "$label" + echo "$(date +%s.%N) $label owner=$([[ $a_bound == 1 ]] && echo a || echo b)" \ + >>"$run_dir/diagnostics/events.log" + [[ $a_bound == 1 ]] && echo a || echo b + return 0 + fi + fi + if ((a_bound + b_bound > 1)); then + snapshot "$label-overlap" + echo 'two VIP holders observed' >&2 + return 1 + fi + sleep 1 + done + snapshot "$label-timeout" + echo "single VIP owner not observed at $label" >&2 + return 1 +} + +owner=$(wait_single_owner initial) +assert_no_overlap +wait_garp initial "$owner" 0 +wait_mysql "$vip_ip" 6000 initial-vip +if [[ $owner == a ]]; then + former=a + new=b +else + former=b + new=a +fi +former_pid=$(cat "$run_dir/$former.pid") +before_garp=$(garp_count "$new") +sudo kill -INT "$former_pid" +wait_single_owner after-controlled-close "$new" >/dev/null +assert_no_overlap +wait_garp after-controlled-close "$new" "$before_garp" +wait_mysql "$vip_ip" 6000 controlled-failover +for _ in {1..30}; do + sudo kill -0 "$former_pid" 2>/dev/null || break + sleep 1 +done +if sudo kill -0 "$former_pid" 2>/dev/null; then + echo 'former VIP owner did not exit after SIGINT' >&2 + exit 1 +fi + +# A restarted former owner removes any stale local address before campaigning. +if [[ $former == a ]]; then + start_node a "$ns_a" + wait_ready a "$ns_a" +else + start_node b "$ns_b" + wait_ready b "$ns_b" +fi +wait_single_owner after-restart "$new" >/dev/null +assert_no_overlap +wait_mysql "$vip_ip" 6000 restart + +# Model a whole-node crash: terminate its process and remove its network link. +# Process-only SIGKILL leaves an IP on a live Linux interface and is separately +# documented as a remaining Go/Rust operational fencing gap. +before_garp=$(garp_count "$former") +sudo kill -KILL "$(cat "$run_dir/$new.pid")" +if [[ $new == a ]]; then + sudo ip link del "$host_a" +else + sudo ip link del "$host_b" +fi +wait_single_owner after-node-link-loss "$former" >/dev/null +assert_no_overlap +wait_garp after-node-link-loss "$former" "$before_garp" +wait_mysql "$vip_ip" 6000 node-failover +kill "$monitor_pid" 2>/dev/null || true +wait "$monitor_pid" 2>/dev/null || true +monitor_pid= +kill "$capture_pid" 2>/dev/null || true +wait "$capture_pid" 2>/dev/null || true +capture_pid= +assert_no_overlap +echo 'PASS: one VIP holder and SQL continuity through controlled close, restart and node/link loss'