feat(hydro_deploy)!: Perf works over SSH (#1313)

See documentation on how to use in
[Notion](https://www.notion.so/hydro-project/perf-Measuring-CPU-usage-6135b6ce56a94af38eeeba0a55deef9c).

Fix #1353

BREAKING CHANGE: changes some config arguments

---------

Co-authored-by: Mingwei Samuel <mingwei.samuel@gmail.com>

.gitignore

@@ -11,3 +11,4 @@ perf.data.old
 flamegraph.svg
 
 /rustc-ice-*.txt
+/*.perf.data*

hydro_deploy/core/Cargo.toml

@@ -20,7 +20,7 @@ dyn-clone = "1"
 futures = "0.3.26"
 futures-core = "0.3.26"
 hydroflow_cli_integration = { path = "../hydroflow_cli_integration", version = "^0.5.2" }
-indicatif = "0.17.6"
+indicatif = "0.17.8"
 memo-map = "0.3.2"
 nanoid = "0.4.0"
 nix = "0.26.2"

hydro_deploy/core/src/deployment.rs

@@ -31,7 +31,7 @@ impl Deployment {
         self.add_host(LocalhostHost::new)
     }
 
-    #[allow(non_snake_case)]
+    #[allow(non_snake_case, clippy::too_many_arguments)] // TODO(mingwei)
     pub fn GcpComputeEngineHost(
         &mut self,
         project: impl Into<String>,
@@ -40,9 +40,19 @@ impl Deployment {
         region: impl Into<String>,
         network: Arc<RwLock<GcpNetwork>>,
         user: Option<String>,
+        startup_script: Option<String>,
     ) -> Arc<GcpComputeEngineHost> {
         self.add_host(|id| {
-            GcpComputeEngineHost::new(id, project, machine_type, image, region, network, user)
+            GcpComputeEngineHost::new(
+                id,
+                project,
+                machine_type,
+                image,
+                region,
+                network,
+                user,
+                startup_script,
+            )
         })
     }
 
@@ -109,7 +119,7 @@ impl Deployment {
                     .collect::<Vec<_>>();
 
                 futures::stream::iter(services_future)
-                    .buffer_unordered(8)
+                    .buffer_unordered(16)
                     .try_fold((), |_, _| async { Ok(()) })
             })
             .await?;

hydro_deploy/core/src/gcp.rs

@@ -176,11 +176,13 @@ pub struct GcpComputeEngineHost {
     pub region: String,
     pub network: Arc<RwLock<GcpNetwork>>,
     pub user: Option<String>,
+    pub startup_script: Option<String>,
     pub launched: OnceLock<Arc<LaunchedComputeEngine>>,
     external_ports: Mutex<Vec<u16>>,
 }
 
 impl GcpComputeEngineHost {
+    #[allow(clippy::too_many_arguments)] // TODO(mingwei)
     pub fn new(
         id: usize,
         project: impl Into<String>,
@@ -189,6 +191,7 @@ impl GcpComputeEngineHost {
         region: impl Into<String>,
         network: Arc<RwLock<GcpNetwork>>,
         user: Option<String>,
+        startup_script: Option<String>,
     ) -> Self {
         Self {
             id,
@@ -198,6 +201,7 @@ impl GcpComputeEngineHost {
             region: region.into(),
             network,
             user,
+            startup_script,
             launched: OnceLock::new(),
             external_ports: Mutex::new(Vec::new()),
         }
@@ -412,7 +416,8 @@ impl Host for GcpComputeEngineHost {
                             ]
                         }
                     ],
-                    "network_interface": external_interfaces
+                    "network_interface": external_interfaces,
+                    "metadata_startup_script": self.startup_script,
                 }),
             );
 

hydro_deploy/core/src/hydroflow_crate/mod.rs

@@ -1,6 +1,8 @@
 use std::path::PathBuf;
 use std::sync::Arc;
 
+use perf_options::PerfOptions;
+
 use super::Host;
 use crate::ServiceBuilder;
 
@@ -10,6 +12,8 @@ pub mod ports;
 pub mod service;
 pub use service::*;
 
+pub mod perf_options;
+
 #[derive(PartialEq)]
 pub enum CrateTarget {
     Default,
@@ -24,7 +28,7 @@ pub struct HydroflowCrate {
     target: CrateTarget,
     on: Arc<dyn Host>,
     profile: Option<String>,
-    perf: Option<PathBuf>, /* If a path is provided, run perf to get CPU time and output to that path.perf.data */
+    perf: Option<PerfOptions>,
     args: Vec<String>,
     display_name: Option<String>,
 }
@@ -78,7 +82,7 @@ impl HydroflowCrate {
         self
     }
 
-    pub fn perf(mut self, perf: impl Into<PathBuf>) -> Self {
+    pub fn perf(mut self, perf: impl Into<PerfOptions>) -> Self {
         if self.perf.is_some() {
             panic!("perf path already set");
         }

hydro_deploy/core/src/hydroflow_crate/perf_options.rs

@@ -0,0 +1,7 @@
+use std::path::PathBuf;
+
+#[derive(Clone)]
+pub struct PerfOptions {
+    pub output_file: PathBuf,
+    pub frequency: u32,
+}

hydro_deploy/core/src/hydroflow_crate/service.rs

@@ -11,6 +11,7 @@ use serde::Serialize;
 use tokio::sync::{mpsc, RwLock};
 
 use super::build::{build_crate_memoized, BuildError, BuildOutput, BuildParams};
+use super::perf_options::PerfOptions;
 use super::ports::{self, HydroflowPortConfig, HydroflowSink, SourcePath};
 use crate::progress::ProgressTracker;
 use crate::{
@@ -21,7 +22,7 @@ pub struct HydroflowCrateService {
     id: usize,
     pub(super) on: Arc<dyn Host>,
     build_params: BuildParams,
-    perf: Option<PathBuf>,
+    perf: Option<PerfOptions>,
     args: Option<Vec<String>>,
     display_id: Option<String>,
     external_ports: Vec<u16>,
@@ -54,7 +55,7 @@ impl HydroflowCrateService {
         bin: Option<String>,
         example: Option<String>,
         profile: Option<String>,
-        perf: Option<PathBuf>,
+        perf: Option<PerfOptions>,
         features: Option<Vec<String>>,
         args: Option<Vec<String>>,
         display_id: Option<String>,

hydro_deploy/core/src/lib.rs

@@ -1,11 +1,11 @@
 use std::collections::HashMap;
 use std::net::SocketAddr;
-use std::path::PathBuf;
 use std::sync::Arc;
 
 use anyhow::Result;
 use async_trait::async_trait;
 use hydroflow_cli_integration::ServerBindConfig;
+use hydroflow_crate::perf_options::PerfOptions;
 
 pub mod deployment;
 pub use deployment::Deployment;
@@ -101,7 +101,7 @@ pub trait LaunchedHost: Send + Sync {
         id: String,
         binary: &BuildOutput,
         args: &[String],
-        perf: Option<PathBuf>,
+        perf: Option<PerfOptions>,
     ) -> Result<Box<dyn LaunchedBinary>>;
 
     async fn forward_port(&self, addr: &SocketAddr) -> Result<SocketAddr>;

hydro_deploy/core/src/localhost/mod.rs

@@ -1,6 +1,5 @@
 use std::collections::HashMap;
 use std::net::SocketAddr;
-use std::path::PathBuf;
 use std::sync::Arc;
 
 use anyhow::{Context, Result};
@@ -13,6 +12,7 @@ use super::{
     ResourceResult, ServerStrategy,
 };
 use crate::hydroflow_crate::build::BuildOutput;
+use crate::hydroflow_crate::perf_options::PerfOptions;
 use crate::HostStrategyGetter;
 
 pub mod launched_binary;
@@ -150,15 +150,22 @@ impl LaunchedHost for LaunchedLocalhost {
         id: String,
         binary: &BuildOutput,
         args: &[String],
-        perf: Option<PathBuf>,
+        perf: Option<PerfOptions>,
     ) -> Result<Box<dyn LaunchedBinary>> {
         let mut command = if let Some(perf) = perf {
             println!("Profiling binary with perf");
             let mut tmp = Command::new("perf");
-            tmp.args(["record", "-F", "5", "--call-graph", "dwarf,64000", "-o"])
-                .arg(&perf)
-                .arg(&binary.bin_path)
-                .args(args);
+            tmp.args([
+                "record",
+                "-F",
+                &perf.frequency.to_string(),
+                "--call-graph",
+                "dwarf,64000",
+                "-o",
+            ])
+            .arg(&perf.output_file)
+            .arg(&binary.bin_path)
+            .args(args);
             tmp
         } else {
             let mut tmp = Command::new(&binary.bin_path);