lda/wakey
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases 18 Wiki Activity

model device endpoints for fleet wake routes

Browse Source
This commit is contained in:
lda
2026-05-15 16:36:37 +07:00 Verified
parent 134d3d0ef9
commit 8628af6bd3
9 changed files with 612 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
wakey-control-plane/src/api/control/fleet/build.rs
+95 -14
View File
@@ -6,7 +6,9 @@ use wakey_core::Presence;
use crate::state::{AgentDeviceWithChildren, DeviceIdentifier, KnownDevice, KnownDeviceSummary};
use super::types::{FleetDevice, FleetDeviceAgent, FleetWakeRoute, ListFleetDevicesQuery};
use super::types::{
FleetDevice, FleetDeviceAgent, FleetDeviceEndpoint, FleetWakeRoute, ListFleetDevicesQuery,
};
#[derive(Debug, Default)]
pub(crate) struct FleetBuildContext {
@@ -31,6 +33,7 @@ struct FleetAccumulator {
hostnames: BTreeSet<String>,
sources: BTreeSet<String>,
agents: BTreeMap<String, FleetDeviceAgent>,
endpoints: Vec<FleetDeviceEndpoint>,
first_seen_unix: Option<u64>,
last_seen_unix: Option<u64>,
presence: Presence,
@@ -49,6 +52,7 @@ impl Default for FleetAccumulator {
hostnames: BTreeSet::new(),
sources: BTreeSet::new(),
agents: BTreeMap::new(),
endpoints: Vec::new(),
first_seen_unix: None,
last_seen_unix: None,
presence: Presence::Offline,
@@ -213,7 +217,6 @@ fn add_agent_device_to_entry(
}
entry.hostnames.insert(hostname.clone());
}
entry.sources.insert("device".to_string());
let first_seen = agent_device.device.first_seen();
let last_seen = agent_device.device.last_seen();
entry.first_seen_unix = Some(
@@ -255,30 +258,83 @@ fn add_agent_device_to_entry(
last_seen_unix: last_seen,
});
for mac in &agent_device.macs {
let ip_for_mac = agent_device.ips.first().copied();
let hostname_for_mac = agent_device.hostnames.first().cloned();
let rid = route_id(&agent_id, Some(mac), ip_for_mac.as_ref(), "device");
let wakeable = status.connected && !device_offline;
if agent_device.endpoints.is_empty() {
entry.sources.insert("device".to_string());
}
for endpoint in &agent_device.endpoints {
let source = endpoint_source_label(endpoint.key.source).to_string();
entry.sources.insert(source.clone());
let endpoint_last_seen = endpoint.last_seen_unix.unwrap_or(last_seen);
let endpoint_first_seen = endpoint.first_seen_unix.or(Some(first_seen));
entry.endpoints.push(FleetDeviceEndpoint {
agent_id: agent_id.clone(),
nickname: status.nickname.clone(),
connected: status.connected,
source: source.clone(),
mac: endpoint.key.mac,
ip: endpoint.key.ip,
hostname: endpoint.hostname.clone(),
interface: endpoint.interface.clone(),
presence: endpoint.presence,
first_seen_unix: endpoint_first_seen,
last_seen_unix: Some(endpoint_last_seen),
});
let rid = route_id(
&agent_id,
endpoint.key.mac.as_ref(),
endpoint.key.ip.as_ref(),
&source,
);
let wakeable = status.connected && endpoint.key.mac.is_some();
entry.routes.insert(
// one per mac... when is one per mac/ip pair? this a regression.
rid.clone(),
FleetWakeRoute {
route_id: rid,
agent_id: agent_id.clone(),
nickname: status.nickname.clone(),
connected: status.connected,
mac: Some(*mac),
ip: ip_for_mac,
hostname: hostname_for_mac,
source: "device".to_string(),
last_seen_unix: last_seen,
mac: endpoint.key.mac,
ip: endpoint.key.ip,
hostname: endpoint.hostname.clone(),
interface: endpoint.interface.clone(),
source,
presence: endpoint.presence,
last_seen_unix: endpoint_last_seen,
wakeable,
},
);
}
if agent_device.macs.is_empty()
if agent_device.endpoints.is_empty() && !agent_device.macs.is_empty() {
for mac in &agent_device.macs {
let ip_for_mac = agent_device.ips.first().copied();
let hostname_for_mac = agent_device.hostnames.first().cloned();
let rid = route_id(&agent_id, Some(mac), ip_for_mac.as_ref(), "device");
let wakeable = status.connected;
entry.routes.insert(
rid.clone(),
FleetWakeRoute {
route_id: rid,
agent_id: agent_id.clone(),
nickname: status.nickname.clone(),
connected: status.connected,
mac: Some(*mac),
ip: ip_for_mac,
hostname: hostname_for_mac,
interface: None,
source: "device".to_string(),
presence: device_presence,
last_seen_unix: last_seen,
wakeable,
},
);
}
}
if agent_device.endpoints.is_empty()
&& agent_device.macs.is_empty()
&& let Some(ip) = agent_device.ips.first()
{
let hostname = agent_device.hostnames.first().cloned();
@@ -293,7 +349,9 @@ fn add_agent_device_to_entry(
mac: None,
ip: Some(*ip),
hostname,
interface: None,
source: "device".to_string(),
presence: device_presence,
last_seen_unix: last_seen,
wakeable: false,
},
@@ -350,7 +408,10 @@ impl FleetAccumulator {
b.connected
.cmp(&a.connected)
.then_with(|| b.wakeable.cmp(&a.wakeable))
.then_with(|| b.ip.is_some().cmp(&a.ip.is_some()))
.then_with(|| b.presence.cmp(&a.presence))
.then_with(|| b.last_seen_unix.cmp(&a.last_seen_unix))
.then_with(|| source_quality_rank(&b.source).cmp(&source_quality_rank(&a.source)))
.then_with(|| a.agent_id.cmp(&b.agent_id))
});
let recommended_route = route_candidates
@@ -374,6 +435,7 @@ impl FleetAccumulator {
hostnames: self.hostnames.into_iter().collect(),
agents: self.agents.into_values().collect(),
sources: self.sources.into_iter().collect(),
endpoints: self.endpoints,
first_seen_unix: self.first_seen_unix,
last_seen_unix: self.last_seen_unix,
presence: self.presence,
@@ -383,6 +445,25 @@ impl FleetAccumulator {
}
}
fn endpoint_source_label(source: wakey_core::EndpointSource) -> &'static str {
match source {
wakey_core::EndpointSource::Neighbor => "neighbor",
wakey_core::EndpointSource::DhcpLease => "dhcp_lease",
wakey_core::EndpointSource::HookNeighbor => "hook_neighbor",
wakey_core::EndpointSource::HookDhcp => "hook_dhcp",
}
}
fn source_quality_rank(source: &str) -> u8 {
match source {
"neighbor" => 4,
"dhcp_lease" => 3,
"hook_neighbor" => 2,
"hook_dhcp" => 1,
_ => 0,
}
}
fn route_id(agent_id: &str, mac: Option<&MacAddr>, ip: Option<&IpAddr>, source: &str) -> String {
format!(
"{}|{}|{}|{}",
wakey-control-plane/src/api/control/fleet/tests.rs
+60 -2
View File
@@ -2,7 +2,7 @@ use std::collections::HashMap;
use std::net::IpAddr;
use macaddr::MacAddr;
use wakey_core::Presence;
use wakey_core::{DeviceEndpoint, EndpointKey, EndpointSource, Presence};
use super::build::{
AgentRuntimeStatus, FleetBuildContext, build_fleet_devices, filter_fleet_devices,
@@ -54,6 +54,7 @@ fn agent_device(
.into_iter()
.collect(),
hostnames: vec!["lda".to_string()],
endpoints: vec![],
facts: vec![],
}
}
@@ -83,6 +84,7 @@ fn offline_agent_device(
.into_iter()
.collect(),
hostnames: vec!["lda".to_string()],
endpoints: vec![],
facts: vec![],
}
}
@@ -105,10 +107,32 @@ fn offline_ip_only_unknown(
macs: vec![],
ips: vec![ip.parse().unwrap()],
hostnames: vec![],
endpoints: vec![],
facts: vec![],
}
}
fn endpoint(
source: EndpointSource,
mac: Option<&str>,
ip: Option<&str>,
last_seen_unix: u64,
) -> DeviceEndpoint {
DeviceEndpoint {
key: EndpointKey::new(
source,
mac.map(|m| m.parse::<MacAddr>().unwrap()),
ip.map(|i| i.parse::<IpAddr>().unwrap()),
)
.expect("endpoint key"),
hostname: Some("lda".into()),
interface: Some("br-lan".into()),
presence: Presence::LikelyOnline,
first_seen_unix: Some(1),
last_seen_unix: Some(last_seen_unix),
}
}
#[test]
fn fleet_grouping_combines_same_mac_across_agents() {
let devices = build_fleet_devices(
@@ -145,6 +169,40 @@ fn fleet_grouping_combines_same_mac_across_agents() {
);
}
#[test]
fn fleet_routes_use_endpoint_mac_ip_pairs() {
let mac = "aa:bb:cc:dd:ee:ff";
let mut row = agent_device(
"agent-a",
"mac:aa:bb:cc:dd:ee:ff",
Some(mac),
Some("192.168.1.2"),
20,
);
row.endpoints = vec![
endpoint(
EndpointSource::HookNeighbor,
Some(mac),
Some("192.168.1.2"),
10,
),
endpoint(EndpointSource::Neighbor, Some(mac), Some("192.168.1.3"), 20),
];
let devices = build_fleet_devices(Vec::new(), vec![row], &context(&["agent-a"]));
assert_eq!(devices.len(), 1);
assert_eq!(devices[0].endpoints.len(), 2);
assert_eq!(devices[0].route_candidates.len(), 2);
assert_eq!(
devices[0]
.recommended_route
.as_ref()
.and_then(|route| route.ip),
Some("192.168.1.3".parse().unwrap())
);
}
#[test]
fn known_device_with_two_macs_absorbs_both_device_groups() {
let known = KnownDevice {
@@ -242,7 +300,7 @@ fn offline_device_has_offline_presence() {
assert_eq!(devices.len(), 1);
assert!(devices[0].ips.is_empty());
assert_eq!(devices[0].presence, Presence::Offline);
assert!(devices[0].recommended_route.is_none());
assert!(devices[0].recommended_route.is_some());
}
#[test]
wakey-control-plane/src/api/control/fleet/types.rs
+19
View File
@@ -68,6 +68,7 @@ pub struct FleetDevice {
pub hostnames: Vec<String>,
pub agents: Vec<FleetDeviceAgent>,
pub sources: Vec<String>,
pub endpoints: Vec<FleetDeviceEndpoint>,
pub first_seen_unix: Option<u64>,
pub last_seen_unix: Option<u64>,
pub presence: Presence,
@@ -83,6 +84,22 @@ pub struct FleetDeviceAgent {
pub last_seen_unix: u64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FleetDeviceEndpoint {
pub agent_id: String,
pub nickname: Option<String>,
pub connected: bool,
pub source: String,
#[serde(with = "mac::option_mac")]
pub mac: Option<MacAddr>,
pub ip: Option<IpAddr>,
pub hostname: Option<String>,
pub interface: Option<String>,
pub presence: Presence,
pub first_seen_unix: Option<u64>,
pub last_seen_unix: Option<u64>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FleetWakeRoute {
pub route_id: String,
@@ -93,7 +110,9 @@ pub struct FleetWakeRoute {
pub mac: Option<MacAddr>,
pub ip: Option<IpAddr>,
pub hostname: Option<String>,
pub interface: Option<String>,
pub source: String,
pub presence: Presence,
pub last_seen_unix: u64,
pub wakeable: bool,
}
wakey-control-plane/src/state/store/agent_devices.rs
+152 -3
View File
@@ -1,5 +1,5 @@
use anyhow::{Context, Result};
use wakey_core::{Device, DeviceId};
use wakey_core::{Device, DeviceEndpoint, DeviceId, EndpointSource};
use super::Store;
use super::helpers::core::*;
@@ -108,6 +108,48 @@ impl Store {
.context("failed inserting device ips")?;
}
sqlx::query(
"DELETE FROM agent_device_endpoints WHERE agent_id = ?1 AND device_key = ?2",
)
.bind(agent_id)
.bind(&device_key)
.execute(&mut *tx)
.await
.context("failed deleting device endpoints")?;
let endpoints = endpoints_for_storage(device);
if !endpoints.is_empty() {
let mut builder = sqlx::QueryBuilder::new(
"INSERT INTO agent_device_endpoints \
(agent_id, device_key, endpoint_key, source, mac, ip, hostname, interface, presence, first_seen_unix, last_seen_unix) ",
);
builder.push_values(endpoints, |mut b, endpoint| {
let first_seen = endpoint.first_seen_unix.unwrap_or(snapshot_time);
let last_seen = endpoint.last_seen_unix.unwrap_or(snapshot_time);
b.push_bind(agent_id)
.push_bind(device_key.clone())
.push_bind(endpoint_storage_key(&endpoint))
.push_bind(endpoint_source_to_str(endpoint.key.source))
.push_bind(
endpoint
.key
.mac
.map(|mac| mac.to_string().to_ascii_lowercase()),
)
.push_bind(endpoint.key.ip.map(|ip| ip.to_string()))
.push_bind(endpoint.hostname.clone())
.push_bind(endpoint.interface.clone())
.push_bind(presence_to_str(endpoint.presence))
.push_bind(i64::try_from(first_seen).unwrap_or(i64::MAX))
.push_bind(i64::try_from(last_seen).unwrap_or(i64::MAX));
});
builder
.build()
.execute(&mut *tx)
.await
.context("failed inserting device endpoints")?;
}
sqlx::query!(
"DELETE FROM agent_device_hostnames WHERE agent_id = ?1 AND device_key = ?2",
agent_id,
@@ -223,6 +265,15 @@ impl Store {
.await
.context("failed listing agent device hostnames")?;
let endpoints = sqlx::query_as::<_, AgentDeviceEndpointRow>(
r#"SELECT agent_id, device_key, endpoint_key, source, mac, ip,
hostname, interface, presence, first_seen_unix, last_seen_unix
FROM agent_device_endpoints"#,
)
.fetch_all(&self.pool)
.await
.context("failed listing agent device endpoints")?;
let facts = sqlx::query_as!(
AgentDeviceFactRow,
r#"SELECT agent_id as "agent_id!", device_key as "device_key!", fact_json as "fact_json!"
@@ -232,7 +283,9 @@ impl Store {
.await
.context("failed listing agent device facts")?;
Ok(assemble_device_rows(devices, macs, ips, hostnames, facts))
Ok(assemble_device_rows(
devices, macs, ips, hostnames, endpoints, facts,
))
}
/// List agent device rows for a single agent.
@@ -287,6 +340,17 @@ impl Store {
.await
.context("failed listing agent device hostnames for agent")?;
let endpoints = sqlx::query_as::<_, AgentDeviceEndpointRow>(
r#"SELECT agent_id, device_key, endpoint_key, source, mac, ip,
hostname, interface, presence, first_seen_unix, last_seen_unix
FROM agent_device_endpoints
WHERE agent_id = ?1"#,
)
.bind(agent_id)
.fetch_all(&self.pool)
.await
.context("failed listing agent device endpoints for agent")?;
let facts = sqlx::query_as!(
AgentDeviceFactRow,
r#"SELECT agent_id as "agent_id!", device_key as "device_key!", fact_json as "fact_json!"
@@ -298,7 +362,9 @@ impl Store {
.await
.context("failed listing agent device facts for agent")?;
Ok(assemble_device_rows(devices, macs, ips, hostnames, facts))
Ok(assemble_device_rows(
devices, macs, ips, hostnames, endpoints, facts,
))
}
}
@@ -307,6 +373,7 @@ fn assemble_device_rows(
macs: Vec<AgentDeviceMacRow>,
ips: Vec<AgentDeviceIpRow>,
hostnames: Vec<AgentDeviceHostnameRow>,
endpoints: Vec<AgentDeviceEndpointRow>,
facts: Vec<AgentDeviceFactRow>,
) -> Vec<AgentDeviceWithChildren> {
use std::collections::BTreeMap;
@@ -332,6 +399,15 @@ fn assemble_device_rows(
.or_default()
.push(row.hostname.clone());
}
let mut endpoint_map: BTreeMap<(&str, &str), Vec<DeviceEndpoint>> = BTreeMap::new();
for row in &endpoints {
if let Some(endpoint) = row.to_endpoint() {
endpoint_map
.entry((row.agent_id.as_str(), row.device_key.as_str()))
.or_default()
.push(endpoint);
}
}
let mut fact_map: BTreeMap<(&str, &str), Vec<String>> = BTreeMap::new();
for row in &facts {
fact_map
@@ -369,11 +445,13 @@ fn assemble_device_rows(
})
.collect();
let hostnames = hostname_map.get(&key).cloned().unwrap_or_default();
let endpoints = endpoint_map.get(&key).cloned().unwrap_or_default();
let facts = fact_map.get(&key).cloned().unwrap_or_default();
AgentDeviceWithChildren {
macs,
ips,
hostnames,
endpoints,
facts,
device,
}
@@ -397,6 +475,65 @@ fn presence_to_str(presence: wakey_core::Presence) -> &'static str {
}
}
fn endpoint_source_to_str(source: EndpointSource) -> &'static str {
match source {
EndpointSource::Neighbor => "neighbor",
EndpointSource::DhcpLease => "dhcp_lease",
EndpointSource::HookNeighbor => "hook_neighbor",
EndpointSource::HookDhcp => "hook_dhcp",
}
}
fn endpoint_storage_key(endpoint: &DeviceEndpoint) -> String {
format!(
"{}|{}|{}",
endpoint_source_to_str(endpoint.key.source),
endpoint
.key
.mac
.map(|mac| mac.to_string().to_ascii_lowercase())
.unwrap_or_default(),
endpoint.key.ip.map(|ip| ip.to_string()).unwrap_or_default()
)
}
fn endpoints_for_storage(device: &Device) -> Vec<DeviceEndpoint> {
if !device.endpoints.is_empty() {
return device.endpoints.clone();
}
let mut endpoints = Vec::new();
for neighbor in &device.neighbors {
endpoints.push(DeviceEndpoint {
key: wakey_core::EndpointKey {
source: EndpointSource::Neighbor,
mac: neighbor.mac,
ip: Some(neighbor.ip),
},
hostname: None,
interface: neighbor.dev.clone(),
presence: wakey_core::Presence::from(neighbor.state),
first_seen_unix: None,
last_seen_unix: None,
});
}
for lease in &device.leases {
endpoints.push(DeviceEndpoint {
key: wakey_core::EndpointKey {
source: EndpointSource::DhcpLease,
mac: Some(lease.mac),
ip: Some(lease.ip),
},
hostname: lease.name.clone(),
interface: None,
presence: wakey_core::Presence::Unknown,
first_seen_unix: None,
last_seen_unix: None,
});
}
endpoints
}
#[cfg(test)]
mod tests {
use super::super::helpers::test_helpers::TestStore;
@@ -412,6 +549,7 @@ mod tests {
ips: vec![ip.parse().expect("ip")],
macs: vec![mac.parse().expect("mac")],
interfaces: vec!["br-lan".to_string()],
endpoints: vec![],
neighbors: vec![NeighborEntry {
ip: ip.parse().expect("ip"),
dev: Some("br-lan".to_string()),
@@ -457,6 +595,15 @@ mod tests {
vec!["192.168.1.10".parse::<std::net::IpAddr>().unwrap()]
);
assert_eq!(rows[0].hostnames, vec!["first-pc"]);
assert_eq!(rows[0].endpoints.len(), 1);
assert_eq!(
rows[0].endpoints[0].key.source,
wakey_core::EndpointSource::Neighbor
);
assert_eq!(
rows[0].endpoints[0].key.ip,
Some("192.168.1.10".parse().expect("ip"))
);
}
#[tokio::test]
@@ -582,6 +729,7 @@ mod tests {
ips: vec![],
macs: vec![],
interfaces: vec![],
endpoints: vec![],
neighbors: vec![],
leases: vec![],
observations: vec![],
@@ -612,6 +760,7 @@ mod tests {
ips: vec!["192.168.1.10".parse().expect("ip")],
macs: vec!["aa:bb:cc:dd:ee:01".parse().expect("mac")],
interfaces: vec![],
endpoints: vec![],
neighbors: vec![],
leases: vec![],
observations: vec![DeviceObservationFact {
wakey-control-plane/src/state/types.rs
+75 -1
View File
@@ -2,7 +2,7 @@ use std::path::PathBuf;
use macaddr::MacAddr;
use serde::{Deserialize, Serialize};
use wakey_core::Presence;
use wakey_core::{DeviceEndpoint, EndpointKey, EndpointSource, Presence};
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct IssuedAgent {
@@ -134,6 +134,66 @@ pub struct AgentDeviceHostnameRow {
pub hostname: String,
}
#[derive(Debug, Clone, sqlx::FromRow)]
pub struct AgentDeviceEndpointRow {
pub agent_id: String,
pub device_key: String,
pub endpoint_key: String,
pub source: String,
pub mac: Option<String>,
pub ip: Option<String>,
pub hostname: Option<String>,
pub interface: Option<String>,
pub presence: String,
pub first_seen_unix: i64,
pub last_seen_unix: i64,
}
impl AgentDeviceEndpointRow {
pub fn to_endpoint(&self) -> Option<DeviceEndpoint> {
let source = endpoint_source_from_str(&self.source)?;
let mac = match self.mac.as_deref() {
Some(raw) => match raw.parse() {
Ok(mac) => Some(mac),
Err(err) => {
tracing::warn!(
error = %err,
endpoint_key = %self.endpoint_key,
raw_mac = raw,
"failed to parse endpoint mac"
);
return None;
}
},
None => None,
};
let ip = match self.ip.as_deref() {
Some(raw) => match raw.parse() {
Ok(ip) => Some(ip),
Err(err) => {
tracing::warn!(
error = %err,
endpoint_key = %self.endpoint_key,
raw_ip = raw,
"failed to parse endpoint ip"
);
return None;
}
},
None => None,
};
let key = EndpointKey::new(source, mac, ip)?;
Some(DeviceEndpoint {
key,
hostname: self.hostname.clone(),
interface: self.interface.clone(),
presence: Presence::from(self.presence.as_str()),
first_seen_unix: Some(self.first_seen_unix.max(0) as u64),
last_seen_unix: Some(self.last_seen_unix.max(0) as u64),
})
}
}
#[derive(Debug, Clone)]
pub struct AgentDeviceFactRow {
pub agent_id: String,
@@ -147,10 +207,24 @@ pub struct AgentDeviceWithChildren {
pub macs: Vec<macaddr::MacAddr>,
pub ips: Vec<std::net::IpAddr>,
pub hostnames: Vec<String>,
pub endpoints: Vec<DeviceEndpoint>,
#[allow(dead_code)]
pub facts: Vec<String>,
}
fn endpoint_source_from_str(raw: &str) -> Option<EndpointSource> {
match raw {
"neighbor" => Some(EndpointSource::Neighbor),
"dhcp_lease" => Some(EndpointSource::DhcpLease),
"hook_neighbor" => Some(EndpointSource::HookNeighbor),
"hook_dhcp" => Some(EndpointSource::HookDhcp),
_ => {
tracing::warn!(source = raw, "unknown endpoint source");
None
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AuditEvent {
pub event_id: String,