use std::collections::hash_map::DefaultHasher;
use std::hash::{Hash, Hasher};
use std::sync::Arc;
use std::time::Duration;
use crate::proto::udb::core::authz::entity::v1::{
self as authz_entity_pb, CanaryScopeKind, CanaryState,
};
pub const CANARY_EVAL_INTERVAL: Duration = Duration::from_secs(5);
#[derive(Debug, Clone, PartialEq)]
pub enum CanaryScope {
Nodes(Vec<String>),
Tenants(Vec<String>),
Percent(u8),
}
impl CanaryScope {
pub fn from_row(kind: CanaryScopeKind, values: &[String]) -> CanaryScope {
match kind {
CanaryScopeKind::Node => CanaryScope::Nodes(clean(values)),
CanaryScopeKind::Tenant => CanaryScope::Tenants(clean(values)),
CanaryScopeKind::Percent => {
let pct = values
.first()
.and_then(|v| v.trim().parse::<i64>().ok())
.unwrap_or(0)
.clamp(0, 100) as u8;
CanaryScope::Percent(pct)
}
CanaryScopeKind::Unspecified => CanaryScope::Percent(0),
}
}
pub fn node_in_scope(&self, node_id: &str) -> bool {
match self {
CanaryScope::Nodes(ids) => ids.iter().any(|id| id == node_id),
CanaryScope::Tenants(_) => false,
CanaryScope::Percent(pct) => in_percent_bucket(node_id, *pct),
}
}
#[cfg(test)]
pub fn tenant_in_scope(&self, tenant_id: &str) -> bool {
match self {
CanaryScope::Tenants(ids) => ids.iter().any(|id| id == tenant_id),
CanaryScope::Nodes(_) => false,
CanaryScope::Percent(pct) => in_percent_bucket(tenant_id, *pct),
}
}
pub fn kind(&self) -> CanaryScopeKind {
match self {
CanaryScope::Nodes(_) => CanaryScopeKind::Node,
CanaryScope::Tenants(_) => CanaryScopeKind::Tenant,
CanaryScope::Percent(_) => CanaryScopeKind::Percent,
}
}
pub fn values(&self) -> Vec<String> {
match self {
CanaryScope::Nodes(ids) | CanaryScope::Tenants(ids) => ids.clone(),
CanaryScope::Percent(pct) => vec![pct.to_string()],
}
}
}
fn clean(values: &[String]) -> Vec<String> {
values
.iter()
.map(|v| v.trim().to_string())
.filter(|v| !v.is_empty())
.collect()
}
fn in_percent_bucket(id: &str, percent: u8) -> bool {
if percent == 0 {
return false;
}
if percent >= 100 {
return true;
}
let mut h = DefaultHasher::new();
id.hash(&mut h);
let bucket = (h.finish() % 100) as u8;
bucket < percent
}
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct CanarySignal {
pub value: f64,
pub samples: i64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CanaryVerdict {
Rollback,
PromoteEligible,
Pause,
Hold,
}
#[derive(Debug, Clone, Copy)]
pub struct CanaryPolicy {
pub success_window_secs: i64,
pub metric_threshold: f64,
pub min_samples: i64,
}
impl CanaryPolicy {
pub fn from_canary(c: &authz_entity_pb::PolicyCanary) -> CanaryPolicy {
CanaryPolicy {
success_window_secs: c.success_window_secs.max(0),
metric_threshold: if c.metric_threshold.is_finite() && c.metric_threshold >= 0.0 {
c.metric_threshold
} else {
0.0
},
min_samples: c.min_samples.max(1),
}
}
}
pub fn evaluate_canary(
policy: &CanaryPolicy,
elapsed_secs: i64,
signal: CanarySignal,
) -> CanaryVerdict {
let conclusive = signal.samples >= policy.min_samples;
let breached = signal.value > policy.metric_threshold;
if breached && conclusive {
return CanaryVerdict::Rollback;
}
if !conclusive {
return if signal.samples <= 0 {
CanaryVerdict::Hold
} else {
CanaryVerdict::Pause
};
}
if elapsed_secs >= policy.success_window_secs {
CanaryVerdict::PromoteEligible
} else {
CanaryVerdict::Hold
}
}
pub fn promote_eligible(c: &authz_entity_pb::PolicyCanary, now_unix: i64) -> bool {
if c.state != CanaryState::Active as i32 {
return false;
}
now_unix.saturating_sub(canary_started_at_unix(c)) >= c.success_window_secs.max(0)
}
pub fn window_remaining_secs(c: &authz_entity_pb::PolicyCanary, now_unix: i64) -> i64 {
let elapsed = now_unix.saturating_sub(canary_started_at_unix(c));
(c.success_window_secs.max(0) - elapsed).max(0)
}
pub fn canary_started_at_unix(c: &authz_entity_pb::PolicyCanary) -> i64 {
c.started_at.as_ref().map(|t| t.seconds).unwrap_or(0)
}
#[async_trait::async_trait]
pub trait CanaryMetricSource: Send + Sync {
async fn read(&self, canary: &authz_entity_pb::PolicyCanary, window_secs: i64) -> CanarySignal;
}
#[derive(Debug, Default, Clone)]
pub struct NoSignalSource;
#[async_trait::async_trait]
impl CanaryMetricSource for NoSignalSource {
async fn read(
&self,
_canary: &authz_entity_pb::PolicyCanary,
_window_secs: i64,
) -> CanarySignal {
CanarySignal {
value: 0.0,
samples: 0,
}
}
}
#[async_trait::async_trait]
pub trait CanaryExecutor: Send + Sync {
async fn list_active_canaries(&self) -> Vec<authz_entity_pb::PolicyCanary>;
async fn auto_rollback(&self, canary: &authz_entity_pb::PolicyCanary, reason: &str);
async fn pause(&self, canary: &authz_entity_pb::PolicyCanary, reason: &str);
}
pub async fn evaluate_one(
canary: &authz_entity_pb::PolicyCanary,
now_unix: i64,
metrics_src: &dyn CanaryMetricSource,
executor: &dyn CanaryExecutor,
recorder: &Arc<dyn crate::metrics::MetricsRecorder>,
) -> CanaryVerdict {
let policy = CanaryPolicy::from_canary(canary);
let elapsed = now_unix.saturating_sub(canary_started_at_unix(canary));
let signal = metrics_src.read(canary, policy.success_window_secs).await;
let verdict = evaluate_canary(&policy, elapsed, signal);
let breached = matches!(verdict, CanaryVerdict::Rollback);
recorder.record_canary_evaluation(breached);
match verdict {
CanaryVerdict::Rollback => {
let reason = format!(
"canary metric breach: value {:.4} > threshold {:.4} ({} samples)",
signal.value, policy.metric_threshold, signal.samples
);
recorder.inc_canary_auto_rollback(&reason);
executor.auto_rollback(canary, &reason).await;
}
CanaryVerdict::Pause => {
let reason = format!(
"canary signal inconclusive: {} samples < {} required",
signal.samples, policy.min_samples
);
executor.pause(canary, &reason).await;
}
CanaryVerdict::PromoteEligible | CanaryVerdict::Hold => {}
}
verdict
}
pub fn spawn_canary_evaluator(
executor: Arc<dyn CanaryExecutor>,
metrics_src: Arc<dyn CanaryMetricSource>,
recorder: Arc<dyn crate::metrics::MetricsRecorder>,
interval: Duration,
now_fn: Arc<dyn Fn() -> i64 + Send + Sync>,
) -> tokio::task::JoinHandle<()> {
let interval = if interval.is_zero() {
CANARY_EVAL_INTERVAL
} else {
interval
};
tokio::spawn(async move {
let mut ticker = tokio::time::interval(interval);
ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
loop {
ticker.tick().await;
let canaries = executor.list_active_canaries().await;
recorder.set_canary_active(!canaries.is_empty());
let now = (now_fn)();
for c in &canaries {
evaluate_one(c, now, metrics_src.as_ref(), executor.as_ref(), &recorder).await;
}
}
})
}
#[cfg(test)]
mod tests {
use super::*;
fn signal(value: f64, samples: i64) -> CanarySignal {
CanarySignal { value, samples }
}
fn policy(window: i64, threshold: f64, min_samples: i64) -> CanaryPolicy {
CanaryPolicy {
success_window_secs: window,
metric_threshold: threshold,
min_samples,
}
}
#[test]
fn breach_in_window_rolls_back() {
let p = policy(300, 0.05, 10);
let v = evaluate_canary(&p, 10, signal(0.20, 50));
assert_eq!(v, CanaryVerdict::Rollback);
}
#[test]
fn breach_rolls_back_at_exact_threshold_is_not_a_breach() {
let p = policy(300, 0.05, 10);
let v = evaluate_canary(&p, 10, signal(0.05, 50));
assert_eq!(v, CanaryVerdict::Hold);
}
#[test]
fn healthy_after_window_is_promote_eligible() {
let p = policy(300, 0.05, 10);
let v = evaluate_canary(&p, 300, signal(0.01, 100));
assert_eq!(v, CanaryVerdict::PromoteEligible);
}
#[test]
fn healthy_inside_window_holds() {
let p = policy(300, 0.05, 10);
let v = evaluate_canary(&p, 120, signal(0.01, 100));
assert_eq!(v, CanaryVerdict::Hold);
}
#[test]
fn insufficient_samples_pauses() {
let p = policy(300, 0.05, 100);
let v = evaluate_canary(&p, 999, signal(0.0, 3));
assert_eq!(v, CanaryVerdict::Pause);
}
#[test]
fn insufficient_samples_even_with_high_value_pauses_not_rollback() {
let p = policy(300, 0.05, 100);
let v = evaluate_canary(&p, 10, signal(0.9, 2));
assert_eq!(v, CanaryVerdict::Pause);
}
#[test]
fn zero_samples_holds_not_pauses() {
let p = policy(300, 0.05, 1);
let v = evaluate_canary(&p, 400, signal(0.0, 0));
assert_eq!(v, CanaryVerdict::Hold);
}
#[test]
fn node_scope_membership() {
let s = CanaryScope::from_row(
CanaryScopeKind::Node,
&["n1".into(), "n2".into(), " ".into()],
);
assert!(s.node_in_scope("n1"));
assert!(s.node_in_scope("n2"));
assert!(!s.node_in_scope("n3"));
assert!(!s.tenant_in_scope("n1"));
}
#[test]
fn tenant_scope_membership() {
let s = CanaryScope::from_row(CanaryScopeKind::Tenant, &["t1".into(), "t2".into()]);
assert!(s.tenant_in_scope("t1"));
assert!(!s.tenant_in_scope("t9"));
assert!(!s.node_in_scope("t1"));
}
#[test]
fn percent_zero_includes_nobody_hundred_includes_everybody() {
let none = CanaryScope::from_row(CanaryScopeKind::Percent, &["0".into()]);
let all = CanaryScope::from_row(CanaryScopeKind::Percent, &["100".into()]);
for id in ["a", "b", "node-xyz", "tenant-42"] {
assert!(!none.node_in_scope(id), "0% must exclude {id}");
assert!(all.node_in_scope(id), "100% must include {id}");
assert!(all.tenant_in_scope(id));
}
}
#[test]
fn percent_bucket_is_sticky_and_monotonic() {
let ten = CanaryScope::from_row(CanaryScopeKind::Percent, &["10".into()]);
let fifty = CanaryScope::from_row(CanaryScopeKind::Percent, &["50".into()]);
let mut included_at_ten = 0usize;
for i in 0..1000 {
let id = format!("node-{i}");
if ten.node_in_scope(&id) {
included_at_ten += 1;
assert!(fifty.node_in_scope(&id), "{id} in 10% must remain in 50%");
}
}
assert!(
(40..=160).contains(&included_at_ten),
"expected ~100 in-scope at 10%, got {included_at_ten}"
);
}
#[test]
fn percent_out_of_range_clamps() {
let over = CanaryScope::from_row(CanaryScopeKind::Percent, &["250".into()]);
assert_eq!(over, CanaryScope::Percent(100));
let neg = CanaryScope::from_row(CanaryScopeKind::Percent, &["-5".into()]);
assert_eq!(neg, CanaryScope::Percent(0));
let garbage = CanaryScope::from_row(CanaryScopeKind::Percent, &["abc".into()]);
assert_eq!(garbage, CanaryScope::Percent(0));
}
#[test]
fn unspecified_scope_fails_closed() {
let s = CanaryScope::from_row(CanaryScopeKind::Unspecified, &["n1".into()]);
assert_eq!(s, CanaryScope::Percent(0));
assert!(!s.node_in_scope("n1"));
}
#[test]
fn scope_roundtrips_kind_and_values() {
let nodes = CanaryScope::Nodes(vec!["n1".into(), "n2".into()]);
assert_eq!(nodes.kind(), CanaryScopeKind::Node);
assert_eq!(nodes.values(), vec!["n1".to_string(), "n2".to_string()]);
let pct = CanaryScope::Percent(25);
assert_eq!(pct.kind(), CanaryScopeKind::Percent);
assert_eq!(pct.values(), vec!["25".to_string()]);
}
}