[ACTP] orchestrate par-control tasks

pkg/privateactionrunner/par-control/BUILD.bazel

@@ -27,6 +27,7 @@ rust_library(
         "src/jwt.rs",
         "src/lib.rs",
         "src/opms.rs",
+        "src/orchestrator.rs",
         "src/procmgr.rs",
         "src/proto.rs",
         "src/test_support.rs",
@@ -91,6 +92,7 @@ rust_test(
     target_compatible_with = _LINUX_OR_WINDOWS,
     deps = [
         "@crates//:tempfile",
+        "@crates//:tokio",
         "@crates//:tokio-stream",
     ],
 )

pkg/privateactionrunner/par-control/Cargo.toml

@@ -37,6 +37,8 @@ tokio = { workspace = true, features = [
   "net",
   "rt-multi-thread",
   "signal",
+  "sync",
+  "time",
 ] }
 tokio-rustls.workspace = true
 tokio-stream = { workspace = true, features = ["net"] }

pkg/privateactionrunner/par-control/src/bins/par-control.rs

@@ -6,7 +6,13 @@
 use anyhow::Result;
 use clap::Parser;
 use par_control::bootstrap;
+use par_control::executor::ExecutorDispatcher;
+use par_control::jwt::{Es256Signer, JwtSigner};
+use par_control::opms::{HttpOpms, HttpOpmsConfig};
+use par_control::orchestrator::{Orchestrator, Params};
+use par_control::procmgr::ProcmgrLifecycle;
 use std::process::ExitCode;
+use std::sync::Arc;
 
 #[derive(Parser)]
 #[command(name = "par-control", about = "Private Action Runner control plane")]
@@ -41,6 +47,7 @@ async fn run() -> Result<()> {
     // The logger's level filter is immutable after initialization, so it starts
     // at Trace and the configured level is applied with set_max_level once
     // bootstrap reports it. Until then, bootstrap's own logs are not filtered.
+    // A logging failure does not prevent the runner from starting.
     if let Err(error) = dd_agent_log::init(dd_agent_log::LogConfig {
         logger_name: "PAR-CONTROL",
         level: log::Level::Trace,
@@ -62,11 +69,52 @@ async fn run() -> Result<()> {
         return Ok(());
     }
 
-    let _config = bootstrapped.into_config()?;
+    let config = bootstrapped.into_config()?;
 
-    log::info!("par-control started");
-    shutdown_signal().await;
-    log::info!("par-control is exiting");
+    let signer: Arc<dyn JwtSigner> = Arc::new(Es256Signer::new(
+        config.identity.org_id,
+        config.identity.runner_id.clone(),
+        &config.identity.private_key,
+    )?);
+
+    let opms = Arc::new(HttpOpms::new(
+        config.opms_base_url.clone(),
+        signer,
+        HttpOpmsConfig {
+            runner_version: config.runner_version.clone(),
+            modes: config.modes.clone(),
+            timeout: config.opms_request_timeout,
+            proxy: config.proxy.clone(),
+            tls: config.tls.clone(),
+            extra_headers: config.opms_extra_headers.clone(),
+        },
+    )?);
+    let lifecycle = Arc::new(ProcmgrLifecycle::new(
+        &config.procmgr_socket,
+        config.executor_process_name.clone(),
+    ));
+    // The IPC certificate is loaded lazily because the executor may create it.
+    let dispatcher = Arc::new(ExecutorDispatcher::new(
+        &config.executor_socket,
+        Some(&config.ipc_cert_file),
+    ));
+
+    let params = Params::from_config(&config);
+    let orchestrator = Orchestrator::new(opms, lifecycle, dispatcher, params);
+
+    log::info!(
+        "par-control starting: version={} urn={} opms={} executor_socket={} procmgr_socket={} ipc_cert={}",
+        config.runner_version,
+        config.identity.urn,
+        config.opms_base_url,
+        config.executor_socket.display(),
+        config.procmgr_socket.display(),
+        config.ipc_cert_file.display(),
+    );
+
+    orchestrator.run(shutdown_signal()).await;
+    log::info!("par-control stopped");
+    log::logger().flush();
     Ok(())
 }
 

pkg/privateactionrunner/par-control/src/lib.rs

@@ -9,6 +9,7 @@ pub mod executor;
 pub mod identity;
 pub mod jwt;
 pub mod opms;
+pub mod orchestrator;
 pub mod procmgr;
 pub mod proto;
 pub mod tls;