feat: Add initial version prover_autoscaler

core/lib/config/Cargo.toml

@@ -23,6 +23,7 @@ anyhow.workspace = true
 rand.workspace = true
 secrecy.workspace = true
 serde = { workspace = true, features = ["derive"] }
+time = { workspace = true, features = ["serde-human-readable"] }
 
 [dev-dependencies]
 serde_json.workspace = true

core/lib/config/src/configs/mod.rs

@@ -60,6 +60,7 @@ pub mod house_keeper;
 pub mod object_store;
 pub mod observability;
 pub mod proof_data_handler;
+pub mod prover_autoscaler;
 pub mod prover_job_monitor;
 pub mod pruning;
 pub mod secrets;

core/lib/config/src/configs/prover_autoscaler.rs

@@ -0,0 +1,65 @@
+use serde::Deserialize;
+use time::Duration;
+
+/// Config used for running ProverAutoscaler (both Scaler and Agent).
+#[derive(Debug, Clone, PartialEq, Deserialize)]
+pub struct GeneralConfig {
+    /// Amount of time ProverJobMonitor will wait all it's tasks to finish.
+    // TODO: find a way to use #[serde(with = "humantime_serde")] with time::Duration.
+    #[serde(default = "GeneralConfig::default_graceful_shutdown_timeout")]
+    pub graceful_shutdown_timeout: Duration,
+    pub agent_config: Option<ProverAutoscalerAgentConfig>,
+    pub scaler_config: Option<ProverAutoscalerScalerConfig>,
+}
+
+#[derive(Debug, Clone, PartialEq, Deserialize)]
+pub struct ProverAutoscalerAgentConfig {
+    /// Port for prometheus metrics connection.
+    pub prometheus_port: u16,
+    /// HTTP port for Scaler to connect to.
+    pub http_port: u16,
+    /// List of namespaces to watch.
+    #[serde(default = "ProverAutoscalerAgentConfig::default_namespaces")]
+    pub namespaces: Vec<String>,
+    /// Watched cluster name. Also can be set via flag.
+    pub cluster_name: Option<String>,
+}
+
+#[derive(Debug, Clone, PartialEq, Deserialize)]
+pub struct ProverAutoscalerScalerConfig {
+    /// Port for prometheus metrics connection.
+    pub prometheus_port: u16,
+    /// The interval between runs for GPU Prover Archiver.
+    #[serde(default = "ProverAutoscalerScalerConfig::default_scaler_run_interval")]
+    pub scaler_run_interval: Duration,
+    /// URL to get queue reports from.
+    #[serde(default = "ProverAutoscalerScalerConfig::default_prover_job_monitor_url")]
+    pub prover_job_monitor_url: String,
+    /// List of ProverAutoscaler Agents to get data from.
+    pub agents: Vec<String>,
+}
+
+impl GeneralConfig {
+    /// Default graceful shutdown timeout -- 5 seconds
+    pub fn default_graceful_shutdown_timeout() -> Duration {
+        Duration::seconds(5)
+    }
+}
+
+impl ProverAutoscalerAgentConfig {
+    pub fn default_namespaces() -> Vec<String> {
+        vec!["prover-blue".to_string(), "prover-red".to_string()]
+    }
+}
+
+impl ProverAutoscalerScalerConfig {
+    /// Default scaler_run_interval -- 10s
+    pub fn default_scaler_run_interval() -> Duration {
+        Duration::seconds(10)
+    }
+
+    /// Default prover_job_monitor_url -- cluster local URL
+    pub fn default_prover_job_monitor_url() -> String {
+        "http://prover-job-monitor.stage2.svc.cluster.local:3074/queue_report".to_string()
+    }
+}

core/lib/protobuf_config/Cargo.toml

@@ -26,6 +26,7 @@ rand.workspace = true
 hex.workspace = true
 secrecy.workspace = true
 tracing.workspace = true
+time.workspace = true
 
 [build-dependencies]
 zksync_protobuf_build.workspace = true

core/lib/protobuf_config/src/lib.rs

@@ -32,6 +32,7 @@ mod snapshots_creator;
 mod da_client;
 mod external_price_api_client;
 mod external_proof_integration_api;
+mod prover_autoscaler;
 mod prover_job_monitor;
 mod snapshot_recovery;
 #[cfg(test)]

core/lib/protobuf_config/src/proto/config/prover_autoscaler.proto

@@ -0,0 +1,25 @@
+syntax = "proto3";
+
+package zksync.config.prover_autoscaler;
+
+import "zksync/std.proto";
+
+message GeneralConfig {
+  optional std.Duration graceful_shutdown_timeout = 1; // optional
+  optional ProverAutoscalerAgentConfig agent_config = 2; // optional
+  optional ProverAutoscalerScalerConfig scaler_config = 3; // optional
+}
+
+message ProverAutoscalerAgentConfig {
+  optional uint32 prometheus_port = 1; // required
+  optional uint32 http_port = 2; // required
+  repeated string namespaces = 3; // optional
+  optional string cluster_name = 4; // optional
+}
+
+message ProverAutoscalerScalerConfig {
+  optional uint32 prometheus_port = 1; // required
+  optional std.Duration scaler_run_interval = 2; // optional
+  optional string prover_job_monitor_url = 3; // required
+  repeated string agents = 4; // required at least one
+}

core/lib/protobuf_config/src/prover_autoscaler.rs

@@ -0,0 +1,78 @@
+use anyhow::Context as _;
+use zksync_config::configs;
+use zksync_protobuf::{read_optional, repr::ProtoRepr, required, ProtoFmt};
+
+use crate::{proto::prover_autoscaler as proto, read_optional_repr};
+
+impl ProtoRepr for proto::GeneralConfig {
+    type Type = configs::prover_autoscaler::GeneralConfig;
+    fn read(&self) -> anyhow::Result<Self::Type> {
+        Ok(Self::Type {
+            graceful_shutdown_timeout: read_optional(&self.graceful_shutdown_timeout)
+                .context("graceful_shutdown_timeout")?
+                .unwrap_or(Self::Type::default_graceful_shutdown_timeout()),
+            agent_config: read_optional_repr(&self.agent_config),
+            scaler_config: read_optional_repr(&self.scaler_config),
+        })
+    }
+
+    fn build(this: &Self::Type) -> Self {
+        Self {
+            graceful_shutdown_timeout: Some(ProtoFmt::build(&this.graceful_shutdown_timeout)),
+            agent_config: this.agent_config.as_ref().map(ProtoRepr::build),
+            scaler_config: this.scaler_config.as_ref().map(ProtoRepr::build),
+        }
+    }
+}
+
+impl ProtoRepr for proto::ProverAutoscalerAgentConfig {
+    type Type = configs::prover_autoscaler::ProverAutoscalerAgentConfig;
+    fn read(&self) -> anyhow::Result<Self::Type> {
+        Ok(Self::Type {
+            prometheus_port: required(&self.prometheus_port)
+                .and_then(|x| Ok((*x).try_into()?))
+                .context("prometheus_port")?,
+            http_port: required(&self.http_port)
+                .and_then(|x| Ok((*x).try_into()?))
+                .context("http_port")?,
+            namespaces: self.namespaces.to_vec(),
+            cluster_name: Some("".to_string()),
+        })
+    }
+
+    fn build(this: &Self::Type) -> Self {
+        Self {
+            prometheus_port: Some(this.prometheus_port.into()),
+            http_port: Some(this.http_port.into()),
+            namespaces: this.namespaces.clone(),
+            cluster_name: this.cluster_name.clone(),
+        }
+    }
+}
+
+impl ProtoRepr for proto::ProverAutoscalerScalerConfig {
+    type Type = configs::prover_autoscaler::ProverAutoscalerScalerConfig;
+    fn read(&self) -> anyhow::Result<Self::Type> {
+        Ok(Self::Type {
+            prometheus_port: required(&self.prometheus_port)
+                .and_then(|x| Ok((*x).try_into()?))
+                .context("prometheus_port")?,
+            scaler_run_interval: read_optional(&self.scaler_run_interval)
+                .context("scaler_run_interval")?
+                .unwrap_or(Self::Type::default_scaler_run_interval()),
+            prover_job_monitor_url: required(&self.prover_job_monitor_url)
+                .context("prover_job_monitor_url")?
+                .clone(),
+            agents: self.agents.to_vec(),
+        })
+    }
+
+    fn build(this: &Self::Type) -> Self {
+        Self {
+            prometheus_port: Some(this.prometheus_port.into()),
+            scaler_run_interval: Some(ProtoFmt::build(&this.scaler_run_interval)),
+            prover_job_monitor_url: Some(this.prover_job_monitor_url.clone()),
+            agents: this.agents.clone(),
+        }
+    }
+}