server, agent: allow jobs to identify factory used for worker

Author: jclulow

agent/src/control/mod.rs

@@ -13,7 +13,7 @@ use tokio::{
     net::UnixStream,
 };
 
-use protocol::{Decoder, Message, Payload};
+use protocol::{Decoder, FactoryInfo, Message, Payload};
 
 pub(crate) mod protocol;
 pub(crate) mod server;
@@ -79,6 +79,7 @@ pub async fn main() -> Result<()> {
     l.cmd("store", "access the job store", cmd!(cmd_store))?;
     l.cmd("address", "manage IP addresses for this job", cmd!(cmd_address))?;
     l.cmd("process", "manage background processes", cmd!(cmd_process))?;
+    l.cmd("factory", "factory information for this worker", cmd!(cmd_factory))?;
     l.hcmd("eng", "for working on and testing buildomat", cmd!(cmd_eng))?;
 
     sel!(l).run().await
@@ -491,3 +492,89 @@ async fn cmd_process_start(mut l: Level<Stuff>) -> Result<()> {
         }
     }
 }
+
+async fn factory_info(s: &mut Stuff) -> Result<FactoryInfo> {
+    let mout =
+        Message { id: s.ids.next().unwrap(), payload: Payload::FactoryInfo };
+
+    match s.send_and_recv(&mout).await {
+        Ok(min) => {
+            match min.payload {
+                Payload::Error(e) => {
+                    /*
+                     * This request is purely local to the agent, so an
+                     * error is not something we should retry indefinitely.
+                     */
+                    bail!("could not get factory info: {e}");
+                }
+                Payload::FactoryInfoResult(fi) => Ok(fi),
+                other => bail!("unexpected response: {other:?}"),
+            }
+        }
+        Err(e) => {
+            /*
+             * Requests to the agent are relatively simple and over a UNIX
+             * socket; they should not fail.  This implies something has
+             * gone seriously wrong and it is unlikely that it will be fixed
+             * without intervention.  Don't retry.
+             */
+            bail!("could not talk to the agent: {e}");
+        }
+    }
+}
+
+async fn cmd_factory(mut l: Level<Stuff>) -> Result<()> {
+    l.context_mut().connect().await?;
+
+    l.cmd(
+        "name",
+        "print the name of the factory that produced this worker",
+        cmd!(cmd_factory_name),
+    )?;
+    l.cmd(
+        "id",
+        "print the unique ID of the factory that produced this worker",
+        cmd!(cmd_factory_id),
+    )?;
+    l.cmd(
+        "private",
+        "print the factory-specific identifier for the underlying resource",
+        cmd!(cmd_factory_private),
+    )?;
+
+    sel!(l).run().await
+}
+
+async fn cmd_factory_id(mut l: Level<Stuff>) -> Result<()> {
+    no_args!(l);
+
+    let fi = factory_info(l.context_mut()).await?;
+
+    println!("{}", fi.id);
+
+    Ok(())
+}
+
+async fn cmd_factory_name(mut l: Level<Stuff>) -> Result<()> {
+    no_args!(l);
+
+    let fi = factory_info(l.context_mut()).await?;
+
+    println!("{}", fi.name);
+
+    Ok(())
+}
+
+async fn cmd_factory_private(mut l: Level<Stuff>) -> Result<()> {
+    no_args!(l);
+
+    let fi = factory_info(l.context_mut()).await?;
+
+    let Some(fp) = &fi.private else {
+        bail!("factory private info not available");
+    };
+
+    println!("{fp}");
+
+    Ok(())
+}

agent/src/control/protocol.rs

@@ -9,6 +9,13 @@ use buildomat_types::metadata;
 use bytes::{Buf, BufMut, BytesMut};
 use serde::{Deserialize, Serialize};
 
+#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq)]
+pub struct FactoryInfo {
+    pub id: String,
+    pub name: String,
+    pub private: Option<String>,
+}
+
 #[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq)]
 pub struct StoreEntry {
     pub name: String,
@@ -38,6 +45,9 @@ pub enum Payload {
         uid: u32,
         gid: u32,
     },
+
+    FactoryInfo,
+    FactoryInfoResult(FactoryInfo),
 }
 
 #[derive(Clone, Debug, Serialize, Deserialize)]

agent/src/control/server.rs

@@ -204,7 +204,8 @@ async fn handle_client_turn(
                 Payload::StoreGet(..)
                 | Payload::StorePut(..)
                 | Payload::MetadataAddresses
-                | Payload::ProcessStart { .. } => {
+                | Payload::ProcessStart { .. }
+                | Payload::FactoryInfo => {
                     /*
                      * These are requests from the control program.  Pass them
                      * on to the main loop.

agent/src/main.rs

@@ -36,7 +36,7 @@ mod exec;
 mod shadow;
 mod upload;
 
-use control::protocol::Payload;
+use control::protocol::{FactoryInfo, Payload};
 use exec::ExitDetails;
 
 const CONFIG_PATH: &str = "/opt/buildomat/etc/agent.json";
@@ -1201,6 +1201,7 @@ async fn cmd_run(mut l: Level<()>) -> Result<()> {
     let mut bgprocs = exec::BackgroundProcesses::new();
 
     let mut metadata: Option<metadata::FactoryMetadata> = None;
+    let mut factory: Option<WorkerPingFactoryInfo> = None;
     let mut set_root_password = false;
     let mut set_root_keys = false;
     let mut dump_device_configured = false;
@@ -1295,6 +1296,10 @@ async fn cmd_run(mut l: Level<()>) -> Result<()> {
                         continue;
                     }
 
+                    if let Some(fi) = &p.factory_info {
+                        factory = Some(fi.clone());
+                    }
+
                     if let Some(md) = &p.factory_metadata {
                         /*
                          * If the factory has given us a post-job diagnostic
@@ -1431,6 +1436,17 @@ async fn cmd_run(mut l: Level<()>) -> Result<()> {
                         Err(e) => Payload::Error(e.to_string()),
                     }
                 }
+                Payload::FactoryInfo => {
+                    if let Some(f) = &factory {
+                        Payload::FactoryInfoResult(FactoryInfo {
+                            id: f.id.to_string(),
+                            name: f.name.to_string(),
+                            private: f.private.clone(),
+                        })
+                    } else {
+                        Payload::Error("factory info not available".into())
+                    }
+                }
                 _ => Payload::Error("unexpected message type".to_string()),
             };
 

client/openapi.json

@@ -2767,6 +2767,10 @@
                 "nullable": true,
                 "type": "string"
               },
+              "rpool_disable_sync": {
+                "nullable": true,
+                "type": "boolean"
+              },
               "v": {
                 "type": "string",
                 "enum": [
@@ -3929,6 +3933,25 @@
           "value"
         ]
       },
+      "WorkerPingFactoryInfo": {
+        "type": "object",
+        "properties": {
+          "id": {
+            "type": "string"
+          },
+          "name": {
+            "type": "string"
+          },
+          "private": {
+            "nullable": true,
+            "type": "string"
+          }
+        },
+        "required": [
+          "id",
+          "name"
+        ]
+      },
       "WorkerPingInput": {
         "type": "object",
         "properties": {
@@ -4006,6 +4029,14 @@
       "WorkerPingResult": {
         "type": "object",
         "properties": {
+          "factory_info": {
+            "nullable": true,
+            "allOf": [
+              {
+                "$ref": "#/components/schemas/WorkerPingFactoryInfo"
+              }
+            ]
+          },
           "factory_metadata": {
             "nullable": true,
             "allOf": [
@@ -4127,4 +4158,4 @@
       }
     }
   }
-}
+}

server/src/api/worker.rs

@@ -118,11 +118,19 @@ pub(crate) struct WorkerPingJob {
     inputs: Vec<WorkerPingInput>,
 }
 
+#[derive(Serialize, JsonSchema)]
+pub(crate) struct WorkerPingFactoryInfo {
+    id: String,
+    name: String,
+    private: Option<String>,
+}
+
 #[derive(Serialize, JsonSchema)]
 pub(crate) struct WorkerPingResult {
     poweroff: bool,
     job: Option<WorkerPingJob>,
     factory_metadata: Option<metadata::FactoryMetadata>,
+    factory_info: Option<WorkerPingFactoryInfo>,
 }
 
 #[endpoint {
@@ -143,6 +151,8 @@ pub(crate) async fn worker_ping(
 
     let factory_metadata = w.factory_metadata().or_500()?;
 
+    let factory = c.db.factory(w.factory()).or_500()?;
+
     let job = if w.wait_for_flush {
         /*
          * The factory may have event records (e.g., boot time console logs or
@@ -217,7 +227,16 @@ pub(crate) async fn worker_ping(
         w.recycle || w.deleted
     };
 
-    let res = WorkerPingResult { poweroff, job, factory_metadata };
+    let res = WorkerPingResult {
+        poweroff,
+        job,
+        factory_metadata,
+        factory_info: Some(WorkerPingFactoryInfo {
+            id: factory.id.to_string(),
+            name: factory.name,
+            private: w.factory_private.clone(),
+        }),
+    };
 
     Ok(HttpResponseOk(res))
 }