rrdnsd/

core.rs

// rrdnsd - monitoring and failover for Round Robin DNS records
// Copyright 2024-2025 Federico Ceratto <federico@debian.org>
// Released under AGPLv3

use anyhow::anyhow;
use log::{debug, error, info, warn};
use statsd::client::Client as Statsd;
use std::collections::HashMap;
// TODO: tokio exit?
use std::time::SystemTime;

use crate::*;
/// Receives:
///  - an incoming datapoint from a probe
///  - a request to generate the dashboard page
///  - a Change action from the RPC API
///
/// Triggers DNS failover / RR change if needed.
async fn updater_handle_one_update(
    statsd: &Statsd,
    conf: &Conf,
    upwrap: UpWrap,
    service_statuses: &mut FqdnToStatuses,
    chan_outbox_tx: &OutboxSend,
) -> Result<()> {
    match upwrap {
        // Respond to a request from the HTTP service to generate the HTML dashboard
        UpWrap::Fetch { chan_dash_tx } => {
            updater_fetch(statsd, service_statuses, chan_dash_tx);
            return Ok(());
        }
        // Handle a RPC API Change request
        UpWrap::APIChangeRequestWrap {
            request,
            resp_chan_tx,
        } => {
            // updater_handle_api_change_request(request, resp_chan_tx);
            debug!("[updater] processing {request:?}");

            let results = request
                .actions
                .iter()
                .map(
                    |action| match handle_change(conf, service_statuses, action, chan_outbox_tx) {
                        Ok(()) => APIChangeResult::Ok,
                        Err(e) => APIChangeResult::Error {
                            error: e.to_string(),
                        },
                    },
                )
                .collect();

            let resp = APIChangeResponse { results };
            if resp_chan_tx.send(resp).is_err() {
                error!("[updater] failed to send response");
            }
            return Ok(());
        }

        // Receive a monitoring datapoint
        UpWrap::Update { update } => {
            // debug!("[updater] doing update")
            // statsd metric `received_update.cnt` - internal
            statsd.incr("received_update.cnt");
            let msg = update; //.context("Empty update")?;

            debug!("[updater] received a datapoint for {:?}", msg.ipaddr);
            let endpoint_ipaddr = msg.ipaddr;

            let srv = service_statuses
                .get_mut(&msg.fqdn)
                .context(format!("Unknown FQDN in quorum msg '{}'", msg.fqdn))?;

            let end = srv
                .endpoint_statuses
                .get_mut(&endpoint_ipaddr)
                .context("Unknown endpoind ipaddr in quorum msg")?;

            // Store received datapoint
            let change = ingest_datapoint(end, &msg);
            if !change {
                return Ok(());
            }

            // Endopoint status change detected: Update DNS
            // statsd metric `status_change.cnt` - Endpoint status change count
            statsd.incr("status_change.cnt");
            info!(
                "[updater] Service '{}' endpoint {} changed to {:?}",
                &msg.fqdn, endpoint_ipaddr, end.computed_status
            );
            handle_endpoint_change(conf, srv, endpoint_ipaddr, &msg.fqdn).await;
        }
    }
    Ok(())
}

// HTTP API - handle change request
fn handle_change(
    conf: &Conf,
    service_statuses: &mut FqdnToStatuses,
    action: &APIRPCUpdateAction,
    chan_outbox_tx: &OutboxSend,
) -> Result<()> {
    match action {
        // Create new service with unique FQDN
        APIRPCUpdateAction::CreateService {
            fqdn,
            healthcheck,
            ttl,
            zone,
            probe_interval_ms,
        } => {
            if service_statuses.contains_key(fqdn) {
                bail!("Service with FQDN {fqdn} already exists")
            }
            let probe_target_uri = healthcheck.replace("{}", fqdn);
            let probe_interval = Duration::from_millis(*probe_interval_ms);

            service_statuses.insert(
                fqdn.to_string(),
                ServiceStatus {
                    dns_ttl: *ttl,
                    dns_zone: zone.clone(),
                    endpoint_statuses: HashMap::new(),
                    probe_interval,
                    probe_target_url: probe_target_uri,
                },
            );

            Ok(())
        }

        // Delete service by FQDN
        APIRPCUpdateAction::DeleteService { fqdn } => {
            if service_statuses.remove(fqdn).is_some() {
                Ok(())
            } else {
                bail!("Service with FQDN {fqdn} not found when deleting");
            }
        }

        // Add a new ipaddr to an existing service
        APIRPCUpdateAction::AddIpAddr { fqdn, ipaddr } => {
            let service = service_statuses
                .get_mut(fqdn)
                .ok_or_else(|| anyhow!("Service with FQDN {fqdn} not found when adding ipaddr"))?;

            let endpoint_ipaddr = ipaddr;

            if service.endpoint_statuses.contains_key(endpoint_ipaddr) {
                bail!("{ipaddr} is already listed for '{fqdn}'");
            }

            let cancel_probe = CancellationToken::new();
            let probe_target_uri = service.probe_target_url.clone();
            info!("[updater] Adding {ipaddr} for {fqdn} and starting probe for {probe_target_uri}");

            let node_to_ens = conf
                .parsed_nodes
                .iter()
                .map(|&node| {
                    (
                        node,
                        Enst {
                            status: Status::Unknown,
                            last_update_time: SystemTime::now(),
                        },
                    )
                })
                .collect();

            let es = EndpointStatus {
                node_to_ens,
                computed_status_time: SystemTime::now(),
                computed_status: Status::Unknown,
                probe_cancel: cancel_probe.clone(),
            };

            service.endpoint_statuses.insert(*endpoint_ipaddr, es);

            // spawn dedicated probe worker
            let local_node = conf.parsed_local_node.unwrap();
            let prober = probe(
                service.probe_interval,
                local_node,
                *endpoint_ipaddr,
                probe_target_uri,
                fqdn.to_string(),
                chan_outbox_tx.clone(),
                cancel_probe,
            );
            tokio::spawn(prober);

            Ok(())
        }

        // Delete an existing ipaddr from an existing service
        APIRPCUpdateAction::DeleteIpAddr { fqdn, ipaddr } => {
            let svc = service_statuses.get_mut(fqdn).ok_or_else(|| {
                anyhow!("Service with FQDN {fqdn} not found when deleting ipaddr")
            })?;

            let endpoint_status = svc
                .endpoint_statuses
                .remove(ipaddr)
                .ok_or_else(|| anyhow!("{ipaddr} not found for '{fqdn}'"))?;

            endpoint_status.probe_cancel.cancel(); // shut down probe
            Ok(())
        }
    }
}

/// Runs the updater actor. See `updater_handle_one_update`
pub async fn updater(
    statsd: Statsd,
    conf: Conf,
    mut service_statuses: FqdnToStatuses,
    mut chan_update_rx: UpdateRec,
    chan_outbox_tx: OutboxSend,
) {
    info!("[updater] started");
    while let Some(upwrap) = chan_update_rx.recv().await {
        let res = updater_handle_one_update(
            &statsd,
            &conf,
            upwrap,
            &mut service_statuses,
            &chan_outbox_tx,
        )
        .await;
        if let Err(err) = &res {
            warn!("[updater_handle_one_update] {err}");
        }
    }
    info!("[updater] exiting");
    drop(chan_update_rx);
}

/// Updater responds to the fetch message with the service and endpoint status
fn updater_fetch(
    statsd: &Statsd,
    service_statuses: &FqdnToStatuses,
    chan_statusclone_tx: ServiceStatusCloneSend,
) {
    // statsd metric `fetch.cnt` - internal
    statsd.incr("fetch.cnt");

    let payload: FqdnToStatuses = service_statuses.clone();
    if chan_statusclone_tx.send(payload).is_err() {
        error!("[updater] failed to send response");
    }
}