健康检查与后台服务

pub trait HealthCheck {
    /// Check the given backend.
    ///
    /// `Ok(())`` if the check passes, otherwise the check fails.
    ///
    /// A health-check pass drops this future to cancel the check on shutdown,
    /// so an implementation must not rely on running to completion once polled.
    async fn check(&self, target: &Backend) -> Result<()>;

    /// Called when the health changes for a [Backend].
    async fn health_status_change(&self, _target: &Backend, _healthy: bool) {}

    /// Called when a detailed [Backend] summary is needed.
    fn backend_summary(&self, target: &Backend) -> String {
        format!("{target:?}")
    }

    /// This function defines how many *consecutive* checks should flip the health of a backend.
    // ...
    fn health_threshold(&self, success: bool) -> usize;
}

impl Default for Health {
    fn default() -> Self {
        Health(Arc::new(ArcSwap::new(Arc::new(HealthInner {
            healthy: true, // TODO: allow to start with unhealthy
            consecutive_counter: 0,
        }))))
    }
}

        loop {
            if *shutdown.borrow() {
                return;
            }

            if schedule.update_is_due(now) {
                // TODO: log err
                match race_shutdown(&mut shutdown, self.update()).await {
                    Raced::ShuttingDown => return,
                    Raced::Completed(_) => schedule.update_finished_at(now),
                }
            }

            // After the first update, discovery and selection setup will be
            // done, so dependent services can start receiving traffic.
            if let Some(ready) = ready_opt.take() {
                ServiceReadyNotifier::notify_ready(ready)
            }

            if schedule.health_check_is_due(now) {
                let health_check = self.backends.run_health_check(self.parallel_health_check);
                match race_shutdown(&mut shutdown, health_check).await {
                    Raced::ShuttingDown => return,
                    Raced::Completed(()) => schedule.health_check_finished_at(now),
                }
            }

            if schedule.is_idle() {
                return;
            }
            let to_wake = schedule.next(now);
            match race_shutdown(&mut shutdown, tokio::time::sleep_until(to_wake.into())).await {
                Raced::ShuttingDown => return,
                Raced::Completed(()) => {}
            }
            now = Instant::now();
        }

// a TCP check that takes 300 ms, like one to a far-away backend
struct SlowTcp(Box<TcpHealthCheck>);

#[async_trait]
impl HealthCheck for SlowTcp {
  async fn check(&self, target: &Backend) -> Result<()> {
    tokio::time::sleep(Duration::from_millis(300)).await;
    self.0.check(target).await
  }
  fn health_threshold(&self, success: bool) -> usize {
    self.0.health_threshold(success)
  }
}

500869 t=2265ms killed upstream B
503596 t=5015ms restarted upstream B
t=    0.. 2241ms  x89   200 A|B
t= 2265.. 2265ms  x1    502 
t= 2312.. 2312ms  x1    200 A
t= 2336.. 2336ms  x1    502 
...
t= 2979.. 2979ms  x1    200 A
t= 3003.. 3003ms  x1    502 
t= 3026.. 6000ms  x121  200 A
t= 6024.. 7999ms  x80   200 A|B

497572 proxy up hc=tcp freq=1000ms retry=false
501585 WARN pingora_load_balancing Backend { addr: Inet(127.0.0.1:18702), weight: 1, ext: {} } becomes unhealthy,  ConnectRefused context: Fail to connect to BasicPeer { _address: Inet(127.0.0.1:18702
504594 INFO pingora_load_balancing Backend { addr: Inet(127.0.0.1:18702), weight: 1, ext: {} } becomes healthy
--- error lines: 16
500869 ERROR pingora_proxy Fail to proxy: Upstream ConnectRefused context: Fail to connect to addr: 127.0.0.1:18702, scheme: HTTP cause:  context: Fail to connect to 127.0.0.1:18702 cause: Connection 
...

async fn race_shutdown<F: Future>(shutdown: &mut ShutdownWatch, work: F) -> Raced<F::Output> {
    tokio::pin!(work);
    loop {
        tokio::select! {
            biased;
            change = shutdown_changed(&mut *shutdown) => match change {
                WatchChange::ShuttingDown => return Raced::ShuttingDown,
                WatchChange::Spurious => continue,
                // Nothing can signal shutdown any more, so stop racing
                // altogether and let the work finish.
                WatchChange::Closed => break,
            },
            output = &mut work => return Raced::Completed(output),
        }
    }
    Raced::Completed(work.await)
}

631028 check #1 starts
631229 check #2 starts
631431 check #3 starts
632075 INFO SIGTERM received, gracefully exiting
632076 INFO Broadcasting graceful shutdown
...
632076 INFO Graceful shutdown: grace period 2s starts
632076 check #3 dropped
632076 INFO service 'BG lb' exited.
634081 INFO Graceful shutdown: grace period ends
...
634081 INFO All runtimes exited, exiting now

719906 check #1 starts
720108 check #2 starts
720311 check #3 starts
720937 INFO SIGTERM received, gracefully exiting
...
720937 INFO Graceful shutdown: grace period 2s starts
722938 INFO Graceful shutdown: grace period ends
722938 INFO Waiting for runtimes to exit!
722938 INFO Waiting for service runtime BG lb to exit
722938 check #3 dropped
723942 INFO All runtimes exited, exiting now

t=    0..    0ms  x1    200 A
t=    6..    6ms  x1    502 
t=    9..    9ms  x1    200 A
t=   12..   12ms  x1    502 
...
t=  583..  583ms  x1    200 A
t=  587..  587ms  x1    502 
t=  590.. 1499ms  x291  200 A