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

rm sled migration, sqlite migration

Browse Source
This commit is contained in:
lda
2026-05-03 16:22:43 +07:00 Verified
parent 7e8e1fca23
commit b573e9f27f
15 changed files with 149 additions and 510 deletions
Download Patch File Download Diff File Expand all files Collapse all files
wakey-control-plane/src/cli.rs
+5 -6
View File
@@ -43,8 +43,8 @@ pub enum Command {
RevokeAgent(RevokeAgentArgs),
/// Print state backend stats.
StateStats(StateStatsArgs),
/// Import a legacy sled state directory into a SQLite state file.
ImportSledState(ImportSledStateArgs),
/// Migrate a legacy sqlite state into a new SQLite state file.
MigrateSqliteState(MigrateSqliteStateArgs),
/// Send SIGHUP to an already-running daemon.
Reload(ReloadArgs),
}
@@ -253,14 +253,13 @@ pub struct StateStatsArgs {
pub target: AdminTargetArgs,
}
// not the greatest thing ive used, this and the migration code gets deleted shortly anyways
#[derive(Args)]
pub struct ImportSledStateArgs {
pub struct MigrateSqliteStateArgs {
#[arg(long, visible_alias = "from")]
pub from_sled_state: PathBuf,
pub from_sqlite_db: PathBuf,
#[arg(long, visible_alias = "to")]
pub to_state_file: PathBuf,
pub to_sqlite_db: PathBuf,
#[arg(long)]
pub force: bool,
wakey-control-plane/src/main.rs
+2 -2
View File
@@ -61,9 +61,9 @@ async fn main() -> Result<()> {
tracing::init(cli.verbose, &config::TelemetryConfig::default())?;
runtime::state_stats(args).await
}
Command::ImportSledState(args) => {
Command::MigrateSqliteState(args) => {
tracing::init(cli.verbose, &config::TelemetryConfig::default())?;
runtime::import_sled_state(args).await
runtime::migrate_sqlite_state(args).await
}
Command::Reload(args) => {
tracing::init(cli.verbose, &config::TelemetryConfig::default())?;
wakey-control-plane/src/runtime/admin.rs
+7 -6
View File
@@ -4,7 +4,7 @@ use anyhow::{Context, Result};
use crate::api;
use crate::cli::{
ImportSledStateArgs, IssueEnrollTokenArgs, ListEnrollTokensArgs, RevokeAgentArgs,
IssueEnrollTokenArgs, ListEnrollTokensArgs, MigrateSqliteStateArgs, RevokeAgentArgs,
RevokeEnrollTokenArgs, StateStatsArgs,
};
use crate::config;
@@ -277,10 +277,11 @@ pub async fn state_stats(args: StateStatsArgs) -> Result<()> {
Ok(())
}
pub async fn import_sled_state(args: ImportSledStateArgs) -> Result<()> {
state::Store::import_sled_state(&args.from_sled_state, &args.to_state_file, args.force).await?;
println!("from_sled_state={}", args.from_sled_state.display());
println!("to_state_file={}", args.to_state_file.display());
println!("imported=true");
pub async fn migrate_sqlite_state(args: MigrateSqliteStateArgs) -> Result<()> {
state::Store::migrate_sqlite_state(&args.from_sqlite_db, &args.to_sqlite_db, args.force)
.await?;
println!("from_sqlite_db={}", args.from_sqlite_db.display());
println!("to_sqlite_db={}", args.to_sqlite_db.display());
println!("migrated=true");
Ok(())
}
wakey-control-plane/src/runtime/mod.rs
+1 -1
View File
@@ -27,7 +27,7 @@ mod admin;
mod process;
pub use admin::revoke_agent;
pub use admin::{
import_sled_state, issue_enroll_token, list_enroll_tokens, revoke_enroll_token, state_stats,
issue_enroll_token, list_enroll_tokens, migrate_sqlite_state, revoke_enroll_token, state_stats,
};
pub use process::reload_daemon;
use process::{remove_pid_file, write_pid_file};
wakey-control-plane/src/state/store.rs
+1 -63
View File
@@ -29,11 +29,10 @@ mod db;
mod devices;
mod enrollment;
mod helpers;
mod import_sled;
mod migrate_sqlite;
use helpers::alerts_audit::*;
use helpers::core::*;
use helpers::legacy::*;
use helpers::rows::*;
#[cfg(test)]
@@ -588,65 +587,4 @@ mod tests {
let _ = fs::remove_dir_all(&dir);
}
#[tokio::test]
async fn import_sled_state_copies_core_records() {
let dir =
std::env::temp_dir().join(format!("wakey-cp-store-test-{}", uuid::Uuid::new_v4()));
let sled_path = dir.join("legacy-state.db");
let sqlite_path = dir.join("state.sqlite3");
fs::create_dir_all(&dir).expect("dir should exist");
let legacy = sled::open(&sled_path).expect("legacy db should open");
let meta = legacy.open_tree("meta").expect("meta should open");
meta.insert(
super::SCHEMA_VERSION_KEY.as_bytes(),
&super::SCHEMA_VERSION.to_le_bytes(),
)
.expect("schema should insert");
let enroll = legacy
.open_tree("enroll_tokens")
.expect("enroll tree should open");
enroll
.insert(b"enr-import-test", &(i64::MAX as u64).to_le_bytes())
.expect("token should insert");
let agents = legacy.open_tree("agents").expect("agents should open");
agents
.insert(b"agent-import", b"tok-import")
.expect("agent should insert");
let agent_meta = legacy.open_tree("agent_meta").expect("meta should open");
agent_meta
.insert(b"agent-import", b"imported-router")
.expect("nickname should insert");
legacy.flush().expect("legacy flush should succeed");
drop(agent_meta);
drop(agents);
drop(enroll);
drop(meta);
drop(legacy);
Store::import_sled_state(&sled_path, &sqlite_path, false)
.await
.expect("import should succeed");
let store = Store::load_or_init(&sqlite_path, Vec::new(), Duration::from_secs(60))
.await
.expect("sqlite should load");
assert!(store.verify_agent_token("agent-import", "tok-import").await);
let agents = store.list_agents_with_nicknames().await;
assert!(agents.iter().any(|(id, nickname)| {
id == "agent-import" && nickname.as_deref() == Some("imported-router")
}));
let tokens = store
.list_enroll_tokens()
.await
.expect("tokens should list");
assert!(
tokens
.iter()
.any(|token| token.enroll_token == "enr-import-test")
);
let _ = fs::remove_dir_all(&dir);
}
}
wakey-control-plane/src/state/store/db.rs
+1 -2
View File
@@ -8,9 +8,8 @@ impl Store {
) -> Result<Self> {
if path.is_dir() {
anyhow::bail!(
"state_file {} is a directory, which looks like a legacy sled store; run `wakey-control-plane import-sled-state --from-sled-state {} --to-state-file <sqlite-file>` and update state_file",
"state_file {} is a directory, which looks like a legacy sled store; sled state is no longer supported",
path.display(),
path.display()
);
}
wakey-control-plane/src/state/store/helpers.rs
-1
View File
@@ -2,7 +2,6 @@ use super::*;
pub(super) mod alerts_audit;
pub(super) mod core;
pub(super) mod legacy;
pub(super) mod rows;
#[cfg(test)]
pub(crate) mod test_helpers;
wakey-control-plane/src/state/store/helpers/alerts_audit.rs
-93
View File
@@ -1,38 +1,5 @@
use super::*;
pub async fn insert_audit_event(pool: &SqlitePool, key: &str, event: &AuditEvent) -> Result<()> {
let metadata_json =
serde_json::to_string(&event.metadata).context("failed to encode audit metadata")?;
let ts_unix = i64::try_from(event.ts_unix).context("audit timestamp overflow")?;
let latency_ms = event
.latency_ms
.map(i64::try_from)
.transpose()
.context("audit latency overflow")?;
sqlx::query!(
"INSERT OR REPLACE INTO audit_events
(event_key, event_id, ts_unix, actor_type, actor_id, agent_id, request_id,
event_type, outcome, latency_ms, message, metadata_json)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)",
key,
event.event_id,
ts_unix,
event.actor_type,
event.actor_id,
event.agent_id,
event.request_id,
event.event_type,
event.outcome,
latency_ms,
event.message,
metadata_json
)
.execute(pool)
.await
.context("failed persisting audit event")?;
Ok(())
}
pub async fn insert_active_alert(
tx: &mut Transaction<'_, Sqlite>,
alert: &AlertState,
@@ -66,36 +33,6 @@ pub async fn insert_active_alert(
Ok(())
}
pub async fn insert_active_alert_pool(pool: &SqlitePool, alert: &AlertState) -> Result<()> {
let metadata_json =
serde_json::to_string(&alert.metadata).context("failed to encode active alert metadata")?;
let alert_value = i64::try_from(alert.value).context("active alert value overflow")?;
let alert_threshold =
i64::try_from(alert.threshold).context("active alert threshold overflow")?;
let last_seen_unix =
i64::try_from(alert.last_seen_unix).context("active alert timestamp overflow")?;
sqlx::query!(
"INSERT OR REPLACE INTO active_alerts
(alert_id, kind, severity, status, agent_id, message, value, threshold,
last_seen_unix, metadata_json)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
alert.alert_id,
alert.kind,
alert.severity,
alert.status,
alert.agent_id,
alert.message,
alert_value,
alert_threshold,
last_seen_unix,
metadata_json
)
.execute(pool)
.await
.context("failed writing active alert snapshot")?;
Ok(())
}
pub async fn insert_alert_transition(
tx: &mut Transaction<'_, Sqlite>,
key: &str,
@@ -126,36 +63,6 @@ pub async fn insert_alert_transition(
Ok(())
}
pub async fn insert_alert_transition_pool(
pool: &SqlitePool,
key: &str,
transition: &AlertTransition,
) -> Result<()> {
let metadata_json = serde_json::to_string(&transition.metadata)
.context("failed to encode alert transition metadata")?;
let ts_unix = i64::try_from(transition.ts_unix).context("alert timestamp overflow")?;
sqlx::query!(
"INSERT OR REPLACE INTO alert_transitions
(transition_key, transition_id, ts_unix, alert_id, kind, agent_id,
from_status, to_status, message, metadata_json)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
key,
transition.transition_id,
ts_unix,
transition.alert_id,
transition.kind,
transition.agent_id,
transition.from_status,
transition.to_status,
transition.message,
metadata_json
)
.execute(pool)
.await
.context("failed persisting alert transition")?;
Ok(())
}
pub fn audit_event_from_row(row: sqlx::sqlite::SqliteRow) -> Result<AuditEvent> {
let ts_unix: i64 = row.try_get("ts_unix")?;
let latency_ms: Option<i64> = row.try_get("latency_ms")?;
wakey-control-plane/src/state/store/helpers/core.rs
-9
View File
@@ -29,15 +29,6 @@ pub fn now_unix() -> u64 {
.unwrap_or(0)
}
pub fn decode_expiry(raw: &[u8]) -> Result<u64> {
if raw.len() != 8 {
anyhow::bail!("invalid token expiry length {}", raw.len());
}
let mut arr = [0u8; 8];
arr.copy_from_slice(raw);
Ok(u64::from_le_bytes(arr))
}
pub fn decode_schema(raw: &[u8]) -> Result<u32> {
if raw.len() != 4 {
anyhow::bail!("invalid schema version length {}", raw.len());
wakey-control-plane/src/state/store/helpers/legacy.rs
-143
View File
@@ -1,143 +0,0 @@
use super::*;
pub async fn import_tree_raw(
pool: &SqlitePool,
legacy: &sled::Db,
sled_tree: &str,
sql_table: &str,
key_col: &str,
value_col: &str,
) -> Result<()> {
let tree = legacy
.open_tree(sled_tree)
.with_context(|| format!("failed to open legacy {sled_tree} tree"))?;
let sql =
format!("INSERT OR REPLACE INTO {sql_table} ({key_col}, {value_col}) VALUES (?1, ?2)");
for item in tree.iter() {
let (key, value) =
item.with_context(|| format!("failed reading legacy {sled_tree} tree"))?;
let key = String::from_utf8(key.to_vec())
.with_context(|| format!("legacy {sled_tree} key is not utf-8"))?;
sqlx::query(&sql)
.bind(key)
.bind(value.to_vec())
.execute(pool)
.await
.with_context(|| format!("failed importing legacy {sled_tree} row"))?;
}
Ok(())
}
pub async fn import_enroll_tokens(pool: &SqlitePool, legacy: &sled::Db) -> Result<()> {
let tree = legacy
.open_tree("enroll_tokens")
.context("failed to open legacy enroll_tokens tree")?;
for item in tree.iter() {
let (token, expiry) = item.context("failed reading legacy enroll token")?;
let token =
String::from_utf8(token.to_vec()).context("legacy enroll token key is not utf-8")?;
let expires_at_unix = i64::try_from(decode_expiry(expiry.as_ref())?)
.context("legacy token expiry overflow")?;
sqlx::query!(
"INSERT OR REPLACE INTO enroll_tokens (token, expires_at_unix) VALUES (?1, ?2)",
token,
expires_at_unix
)
.execute(pool)
.await
.context("failed importing legacy enroll token")?;
}
Ok(())
}
pub async fn import_agents(pool: &SqlitePool, legacy: &sled::Db) -> Result<()> {
let tree = legacy
.open_tree("agents")
.context("failed to open legacy agents tree")?;
for item in tree.iter() {
let (agent_id, agent_token) = item.context("failed reading legacy agent")?;
let agent_id =
String::from_utf8(agent_id.to_vec()).context("legacy agent id is not utf-8")?;
let agent_token =
String::from_utf8(agent_token.to_vec()).context("legacy agent token is not utf-8")?;
sqlx::query!(
"INSERT OR REPLACE INTO agents (agent_id, agent_token) VALUES (?1, ?2)",
agent_id,
agent_token
)
.execute(pool)
.await
.context("failed importing legacy agent")?;
}
Ok(())
}
pub async fn import_agent_meta(pool: &SqlitePool, legacy: &sled::Db) -> Result<()> {
let tree = legacy
.open_tree("agent_meta")
.context("failed to open legacy agent_meta tree")?;
for item in tree.iter() {
let (agent_id, nickname) = item.context("failed reading legacy agent metadata")?;
let agent_id =
String::from_utf8(agent_id.to_vec()).context("legacy agent id is not utf-8")?;
let nickname =
String::from_utf8(nickname.to_vec()).context("legacy nickname is not utf-8")?;
sqlx::query!(
"INSERT OR REPLACE INTO agent_meta (agent_id, nickname) VALUES (?1, ?2)",
agent_id,
nickname
)
.execute(pool)
.await
.context("failed importing legacy agent metadata")?;
}
Ok(())
}
pub async fn import_audit_events(pool: &SqlitePool, legacy: &sled::Db) -> Result<()> {
let tree = legacy
.open_tree("audit_events")
.context("failed to open legacy audit_events tree")?;
for item in tree.iter() {
let (key, value) = item.context("failed reading legacy audit event")?;
let key = String::from_utf8(key.to_vec()).context("legacy audit key is not utf-8")?;
let event: AuditEvent =
serde_json::from_slice(value.as_ref()).context("failed decoding legacy audit event")?;
insert_audit_event(pool, &key, &event)
.await
.context("failed importing legacy audit event")?;
}
Ok(())
}
pub async fn import_active_alerts(pool: &SqlitePool, legacy: &sled::Db) -> Result<()> {
let tree = legacy
.open_tree("active_alerts")
.context("failed to open legacy active_alerts tree")?;
for item in tree.iter() {
let (_, value) = item.context("failed reading legacy active alert")?;
let alert: AlertState = serde_json::from_slice(value.as_ref())
.context("failed decoding legacy active alert")?;
insert_active_alert_pool(pool, &alert)
.await
.context("failed importing legacy active alert")?;
}
Ok(())
}
pub async fn import_alert_transitions(pool: &SqlitePool, legacy: &sled::Db) -> Result<()> {
let tree = legacy
.open_tree("alert_transitions")
.context("failed to open legacy alert_transitions tree")?;
for item in tree.iter() {
let (key, value) = item.context("failed reading legacy alert transition")?;
let key =
String::from_utf8(key.to_vec()).context("legacy alert transition key is not utf-8")?;
let transition: AlertTransition = serde_json::from_slice(value.as_ref())
.context("failed decoding legacy alert transition")?;
insert_alert_transition_pool(pool, &key, &transition)
.await
.context("failed importing legacy alert transition")?;
}
Ok(())
}
wakey-control-plane/src/state/store/import_sled.rs
-61
View File
@@ -1,61 +0,0 @@
use super::*;
impl Store {
pub async fn import_sled_state(
from_sled_state: &Path,
to_state_file: &Path,
force: bool,
) -> Result<()> {
if !from_sled_state.is_dir() {
anyhow::bail!(
"legacy sled state {} does not exist or is not a directory",
from_sled_state.display()
);
}
if to_state_file.is_dir() {
anyhow::bail!(
"target state file {} is a directory",
to_state_file.display()
);
}
if to_state_file.exists()
&& to_state_file
.metadata()
.with_context(|| format!("failed to stat {}", to_state_file.display()))?
.len()
> 0
{
if !force {
anyhow::bail!(
"target SQLite state file {} already exists and is non-empty; re-run with --force to overwrite",
to_state_file.display()
);
}
std::fs::remove_file(to_state_file)
.with_context(|| format!("failed to remove {}", to_state_file.display()))?;
}
let legacy = sled::open(from_sled_state).with_context(|| {
format!(
"failed to open legacy sled state {}",
from_sled_state.display()
)
})?;
let store = Store::load_or_init(to_state_file, Vec::new(), Duration::from_secs(1)).await?;
import_tree_raw(&store.pool, &legacy, "meta", "meta", "key", "value").await?;
import_enroll_tokens(&store.pool, &legacy).await?;
import_agents(&store.pool, &legacy).await?;
import_agent_meta(&store.pool, &legacy).await?;
import_audit_events(&store.pool, &legacy).await?;
import_active_alerts(&store.pool, &legacy).await?;
import_alert_transitions(&store.pool, &legacy).await?;
info!(
from = %from_sled_state.display(),
to = %to_state_file.display(),
"imported legacy sled state into SQLite"
);
Ok(())
}
}
wakey-control-plane/src/state/store/migrate_sqlite.rs
+112
View File
@@ -0,0 +1,112 @@
use super::*;
impl Store {
pub async fn migrate_sqlite_state(
from_sqlite_state: &Path,
to_state_file: &Path,
force: bool,
) -> Result<()> {
if !from_sqlite_state.is_file() {
anyhow::bail!(
"legacy sqlite state {} does not exist or is not a file",
from_sqlite_state.display()
);
}
if to_state_file.is_dir() {
anyhow::bail!(
"target state file {} is a directory",
to_state_file.display()
);
}
let is_same_file = match (
std::fs::canonicalize(from_sqlite_state),
std::fs::canonicalize(to_state_file),
) {
(Ok(from_canon), Ok(to_canon)) => from_canon == to_canon,
_ => from_sqlite_state == to_state_file,
};
let actual_from_path = if is_same_file {
let mut bak = to_state_file.to_path_buf();
let mut file_name = bak.file_name().unwrap_or_default().to_os_string();
file_name.push(".bak");
bak.set_file_name(file_name);
if bak.exists() {
if !force {
anyhow::bail!(
"backup file {} already exists; re-run with --force to overwrite",
bak.display()
);
}
std::fs::remove_file(&bak).with_context(|| {
format!("failed to remove existing backup {}", bak.display())
})?;
}
std::fs::rename(from_sqlite_state, &bak)
.with_context(|| "failed to rename legacy state for in-place migration")?;
bak
} else {
if to_state_file.exists()
&& to_state_file
.metadata()
.with_context(|| format!("failed to stat {}", to_state_file.display()))?
.len()
> 0
{
if !force {
anyhow::bail!(
"target SQLite state file {} already exists and is non-empty; re-run with --force to overwrite",
to_state_file.display()
);
}
std::fs::remove_file(to_state_file)
.with_context(|| format!("failed to remove {}", to_state_file.display()))?;
}
from_sqlite_state.to_path_buf()
};
let store = Store::load_or_init(to_state_file, Vec::new(), Duration::from_secs(1)).await?;
let from_path_str = actual_from_path
.to_str()
.ok_or_else(|| anyhow::anyhow!("invalid path"))?;
sqlx::query(&format!("ATTACH DATABASE '{}' AS legacy", from_path_str))
.execute(&store.pool)
.await
.with_context(|| "failed to attach legacy database")?;
let tables = [
"meta",
"enroll_tokens",
"agents",
"agent_meta",
"known_devices",
"device_identifiers",
"audit_events",
"active_alerts",
"alert_transitions",
];
for table in tables {
let q = format!("INSERT INTO {} SELECT * FROM legacy.{}", table, table);
sqlx::query(&q)
.execute(&store.pool)
.await
.with_context(|| format!("failed to migrate table {}", table))?;
}
sqlx::query("DETACH DATABASE legacy")
.execute(&store.pool)
.await?;
info!(
from = %from_sqlite_state.display(),
to = %to_state_file.display(),
"migrated legacy sqlite state into new sqlite state"
);
Ok(())
}
}