diff options
| -rw-r--r-- | src/ci_ctl.rs | 1 | ||||
| -rw-r--r-- | src/ci_driver.rs | 21 | ||||
| -rw-r--r-- | src/ci_runner.rs | 10 | ||||
| -rw-r--r-- | src/dbctx.rs | 93 | ||||
| -rw-r--r-- | src/main.rs | 1 | ||||
| -rw-r--r-- | src/protocol.rs | 6 | ||||
| -rw-r--r-- | src/sql.rs | 27 |
7 files changed, 136 insertions, 23 deletions
diff --git a/src/ci_ctl.rs b/src/ci_ctl.rs index 15feca9..74e6e55 100644 --- a/src/ci_ctl.rs +++ b/src/ci_ctl.rs @@ -4,6 +4,7 @@ mod sql; mod dbctx; mod notifier; mod io; +mod protocol; use dbctx::DbCtx; use notifier::NotifierConfig; diff --git a/src/ci_driver.rs b/src/ci_driver.rs index c6bef14..d8e60e5 100644 --- a/src/ci_driver.rs +++ b/src/ci_driver.rs @@ -109,12 +109,12 @@ async fn activate_run(dbctx: Arc<DbCtx>, run: &PendingRun, clients: &mut mpsc::R } }; - let run_host = client_job.client.name.clone(); + let host_id = client_job.client.host_id; let connection = dbctx.conn.lock().unwrap(); connection.execute( - "update runs set started_time=?1, run_host=?2, state=1, artifacts_path=?3, build_token=?4 where id=?5", - (now as u64, run_host, format!("{}", artifacts.display()), &client_job.client.build_token, run.id) + "update runs set started_time=?1, host_id=?2, state=1, artifacts_path=?3, build_token=?4 where id=?5", + (now as u64, host_id, format!("{}", artifacts.display()), &client_job.client.build_token, run.id) ) .expect("can update"); std::mem::drop(connection); @@ -129,15 +129,11 @@ async fn activate_run(dbctx: Arc<DbCtx>, run: &PendingRun, clients: &mut mpsc::R struct RunnerClient { tx: mpsc::Sender<Result<String, String>>, rx: BodyStream, - name: String, + host_id: u32, build_token: String, accepted_sources: Option<Vec<String>>, } -fn random_name() -> String { - "random name".to_string() -} - fn token_for_job() -> String { let mut data = [0u8; 32]; std::fs::File::open("/dev/urandom") @@ -250,13 +246,12 @@ impl ClientJob { } impl RunnerClient { - async fn new(sender: mpsc::Sender<Result<String, String>>, resp: BodyStream, accepted_sources: Option<Vec<String>>) -> Result<Self, String> { - let name = random_name(); + async fn new(sender: mpsc::Sender<Result<String, String>>, resp: BodyStream, accepted_sources: Option<Vec<String>>, host_id: u32) -> Result<Self, String> { let token = token_for_job(); let client = RunnerClient { tx: sender, rx: resp, - name, + host_id, build_token: token, accepted_sources, }; @@ -455,7 +450,9 @@ async fn handle_next_job(State(ctx): State<(Arc<DbCtx>, mpsc::Sender<RunnerClien eprintln!("client identifies itself as {:?}", host_info); - let client = match RunnerClient::new(tx_sender, job_resp, accepted_pushers).await { + let host_info_id = ctx.0.id_for_host(&host_info).expect("can get a host info id"); + + let client = match RunnerClient::new(tx_sender, job_resp, accepted_pushers, host_info_id).await { Ok(v) => v, Err(e) => { eprintln!("unable to register client"); diff --git a/src/ci_runner.rs b/src/ci_runner.rs index 11d8566..6356dcb 100644 --- a/src/ci_runner.rs +++ b/src/ci_runner.rs @@ -498,11 +498,14 @@ mod host_info { fn collect_cpu_info() -> CpuInfo { let cpu_lines: Vec<String> = std::fs::read_to_string("/proc/cpuinfo").unwrap().split("\n").map(|line| line.to_string()).collect(); let model_names: Vec<&String> = cpu_lines.iter().filter(|line| line.starts_with("model name")).collect(); - let model = model_names.first().expect("can get model name").to_string().split(":").last().unwrap().trim().to_string(); + let model_name = model_names.first().expect("can get model name").to_string().split(":").last().unwrap().trim().to_string(); let cores = model_names.len() as u32; + let vendor_id = cpu_lines.iter().find(|line| line.starts_with("vendor_id")).expect("vendor_id line is present").split(":").last().unwrap().trim().to_string(); + let family = cpu_lines.iter().find(|line| line.starts_with("vendor_id")).expect("vendor_id line is present").split(":").last().unwrap().trim().to_string(); + let model = cpu_lines.iter().find(|line| line.starts_with("model\t")).expect("vendor_id line is present").split(":").last().unwrap().trim().to_string(); let microcode = cpu_lines.iter().find(|line| line.starts_with("microcode")).expect("microcode line is present").split(":").last().unwrap().trim().to_string(); - CpuInfo { model, microcode, cores } + CpuInfo { model_name, microcode, cores, vendor_id, family, model } } fn collect_mem_info() -> MemoryInfo { @@ -536,10 +539,11 @@ mod host_info { pub fn collect_host_info() -> HostInfo { let cpu_info = collect_cpu_info(); let memory_info = collect_mem_info(); -// let hostname = hostname(); + let hostname = hostname(); let env_info = collect_env_info(); HostInfo { + hostname, cpu_info, memory_info, env_info, diff --git a/src/dbctx.rs b/src/dbctx.rs index 3eb68bd..0b1b7ad 100644 --- a/src/dbctx.rs +++ b/src/dbctx.rs @@ -56,7 +56,7 @@ pub struct Run { pub job_id: u64, pub artifacts_path: Option<String>, pub state: sql::RunState, - pub run_host: Option<String>, + pub host_id: Option<u64>, pub create_time: u64, pub start_time: Option<u64>, pub complete_time: Option<u64>, @@ -126,6 +126,8 @@ impl DbCtx { conn.execute(sql::CREATE_REPO_NAME_INDEX, ()).unwrap(); conn.execute(sql::CREATE_REMOTES_TABLE, ()).unwrap(); conn.execute(sql::CREATE_REMOTES_INDEX, ()).unwrap(); + conn.execute(sql::CREATE_RUNS_TABLE, ()).unwrap(); + conn.execute(sql::CREATE_HOSTS_TABLE, ()).unwrap(); Ok(()) } @@ -551,6 +553,95 @@ impl DbCtx { Ok(remotes) } + /// try to find a host close to `host_info`, but maybe not an exact match. + /// + /// specifically, we'll ignore microcode and family/os - enough that measurements ought to be + /// comparable but maybe not perfectly so. + pub fn find_id_like_host(&self, host_info: &crate::protocol::HostInfo) -> Result<Option<u32>, String> { + self.conn.lock() + .unwrap() + .query_row( + "select id from hosts where \ + ( \ + hostname=?1, cpu_vendor=?2, cpu_model_name=?3, cpu_family=?4, \ + cpu_model=?5, cpu_cores=?6, mem_total=?7, \ + arch=?10 \ + );", + ( + &host_info.hostname, + &host_info.cpu_info.vendor_id, + &host_info.cpu_info.model_name, + &host_info.cpu_info.family, + &host_info.cpu_info.model, + &host_info.cpu_info.cores, + &host_info.memory_info.total, + &host_info.env_info.arch, + ), + |row| { row.get(0) } + ) + .map_err(|e| e.to_string()) + } + + /// get an id for the host described by `host_info`. this may create a new record if no such + /// host exists. + pub fn id_for_host(&self, host_info: &crate::protocol::HostInfo) -> Result<u32, String> { + let conn = self.conn.lock().unwrap(); + + conn + .execute( + "insert or ignore into hosts \ + (\ + hostname, cpu_vendor_id, cpu_model_name, cpu_family, \ + cpu_model, cpu_microcode, cpu_cores, mem_total, \ + arch, family, os\ + ) values (\ + ?1, ?2, ?3, ?4, \ + ?5, ?6, ?7, ?8, \ + ?9, ?10, ?11 \ + );", + ( + &host_info.hostname, + &host_info.cpu_info.vendor_id, + &host_info.cpu_info.model_name, + &host_info.cpu_info.family, + &host_info.cpu_info.model, + &host_info.cpu_info.microcode, + &host_info.cpu_info.cores, + &host_info.memory_info.total, + &host_info.env_info.arch, + &host_info.env_info.family, + &host_info.env_info.os, + ) + ) + .expect("can insert"); + + self.conn.lock() + .unwrap() + .query_row( + "select id from hosts where \ + ( \ + hostname=?1, cpu_vendor=?2, cpu_model_name=?3, cpu_family=?4, \ + cpu_model=?5, cpu_microcode=?6, cpu_cores=?7, mem_total=?8, \ + arch=?9, family=?10, os=?11 \ + );", + ( + &host_info.hostname, + &host_info.cpu_info.vendor_id, + &host_info.cpu_info.model_name, + &host_info.cpu_info.family, + &host_info.cpu_info.model, + &host_info.cpu_info.microcode, + &host_info.cpu_info.cores, + &host_info.memory_info.total, + &host_info.env_info.arch, + &host_info.env_info.family, + &host_info.env_info.os, + ), + |row| { row.get(0) } + ) + .map_err(|e| e.to_string()) + } + pub fn last_run_for_job(&self, job_id: u64) -> Result<Option<Run>, String> { let conn = self.conn.lock().unwrap(); diff --git a/src/main.rs b/src/main.rs index d83826b..0ca0765 100644 --- a/src/main.rs +++ b/src/main.rs @@ -34,6 +34,7 @@ mod io; mod sql; mod notifier; mod dbctx; +mod protocol; use sql::RunState; diff --git a/src/protocol.rs b/src/protocol.rs index 78c7303..9f79a7c 100644 --- a/src/protocol.rs +++ b/src/protocol.rs @@ -73,6 +73,7 @@ impl TaskInfo { #[derive(Serialize, Deserialize, Debug, Clone)] pub struct HostInfo { + pub hostname: String, pub cpu_info: CpuInfo, pub memory_info: MemoryInfo, pub env_info: EnvInfo, @@ -80,9 +81,12 @@ pub struct HostInfo { #[derive(Serialize, Deserialize, Debug, Clone)] pub struct CpuInfo { - pub model: String, + pub model_name: String, pub microcode: String, pub cores: u32, + pub vendor_id: String, + pub family: String, + pub model: String, } #[derive(Serialize, Deserialize, Debug, Clone)] @@ -36,14 +36,14 @@ impl TryFrom<u8> for RunState { } pub(crate) fn row2run(row: &rusqlite::Row) -> Run { - let (id, job_id, artifacts_path, state, run_host, build_token, create_time, start_time, complete_time, run_timeout, build_result, final_text) = row.try_into().unwrap(); + let (id, job_id, artifacts_path, state, host_id, build_token, create_time, start_time, complete_time, run_timeout, build_result, final_text) = row.try_into().unwrap(); let state: u8 = state; Run { id, job_id, artifacts_path, state: state.try_into().unwrap(), - run_host, + host_id, create_time, start_time, complete_time, @@ -101,12 +101,12 @@ pub const CREATE_ARTIFACTS_TABLE: &'static str = "\ created_time INTEGER, completed_time INTEGER);"; -pub const CREATE_RUN_TABLE: &'static str = "\ +pub const CREATE_RUNS_TABLE: &'static str = "\ CREATE TABLE IF NOT EXISTS runs (id INTEGER PRIMARY KEY AUTOINCREMENT, job_id INTEGER, artifacts_path TEXT, state INTEGER NOT NULL, - run_host TEXT, + host_id INTEGER, build_token TEXT, created_time INTEGER, started_time INTEGER, @@ -115,6 +115,21 @@ pub const CREATE_RUN_TABLE: &'static str = "\ build_result INTEGER, final_status TEXT);"; +pub const CREATE_HOSTS_TABLE: &'static str = "\ + CREATE TABLE IF NOT EXISTS hosts (id INTEGER PRIMARY KEY AUTOINCREMENT, + hostname TEXT, + cpu_vendor TEXT, + cpu_model_name TEXT, + cpu_family TEXT, + cpu_model TEXT, + cpu_microcode TEXT, + cpu_cores INTEGER, + mem_total TEXT, + arch TEXT, + family TEXT, + os TEXT, + UNIQUE(hostname, cpu_vendor, cpu_model_name, cpu_family, cpu_model, cpu_microcode, cpu_cores, mem_total, arch, family, os));"; + pub const CREATE_REMOTES_INDEX: &'static str = "\ CREATE INDEX IF NOT EXISTS 'repo_to_remote' ON remotes(repo_id);"; @@ -129,7 +144,7 @@ pub const ACTIVE_RUNS: &'static str = "\ job_id, artifacts_path, state, - run_host, + host_id, build_token, created_time, started_time, @@ -170,7 +185,7 @@ pub const LAST_RUN_FOR_JOB: &'static str = "\ job_id, artifacts_path, state, - run_host, + host_id, build_token, created_time, started_time, |