hydra-evaluator: Rewrite in Rust

Cargo.toml

@@ -1,6 +1,11 @@
 [workspace]
 
-members  = [ "subprojects/hydra-builder", "subprojects/hydra-queue-runner", "subprojects/crates/*" ]
+members = [
+    "subprojects/hydra-builder",
+    "subprojects/hydra-evaluator",
+    "subprojects/hydra-queue-runner",
+    "subprojects/crates/*",
+]
 resolver = "2"
 
 [workspace.package]

flake.nix

@@ -83,6 +83,9 @@
           hydra-builder = self'.callPackage ./subprojects/hydra-builder/package.nix {
             inherit nixComponents;
           };
+          hydra-evaluator = self'.callPackage ./subprojects/hydra-evaluator/package.nix {
+            inherit nixComponents;
+          };
         });
       mkHydraBuilder =
         { pkgs, nixComponents }:
@@ -137,6 +140,7 @@
           hydra-linters
           hydra-queue-runner
           hydra-builder
+          hydra-evaluator
           ;
       };
 
@@ -280,6 +284,7 @@
               hydra-linters
               hydra-queue-runner
               hydra-builder
+              hydra-evaluator
               ;
             foreman = pkgs.callPackage ./packaging/foreman/package.nix {
               foreman-src = foreman;

nixos-modules/default.nix

@@ -7,6 +7,9 @@ rec {
       _file = ./default.nix;
       imports = [ ./web-app.nix ];
       services.hydra-dev.package = lib.mkDefault flakePackages.${pkgs.stdenv.hostPlatform.system}.hydra;
+      services.hydra-dev.evaluatorExecutable = lib.mkDefault "${
+        flakePackages.${pkgs.stdenv.hostPlatform.system}.hydra-evaluator
+      }/bin/hydra-evaluator";
     };
 
   postgresql = ./postgresql.nix;

nixos-modules/web-app.nix

@@ -86,6 +86,11 @@ in
         description = "The Hydra package.";
       };
 
+      evaluatorExecutable = mkOption {
+        type = types.path;
+        description = "Path to the hydra-evaluator executable.";
+      };
+
       hydraURL = mkOption {
         type = types.str;
         description = ''
@@ -300,18 +305,21 @@ in
       ];
       path = with pkgs; [
         hostname-debian
+        # Because hydra-evaluator calls `hydra-eval-jobset`. If we
+        # move that perl script into rust, then we can get rid of
+        # this.
         cfg.package
       ];
       environment = env // {
         HYDRA_DBI = "${env.HYDRA_DBI};application_name=hydra-evaluator";
       };
       serviceConfig = {
         ExecStart = escapeShellArgs [
-          "@${cfg.package}/bin/hydra-evaluator"
+          "@${cfg.evaluatorExecutable}"
           "hydra-evaluator"
         ];
         ExecStopPost = escapeShellArgs [
-          "${cfg.package}/bin/hydra-evaluator"
+          "${cfg.evaluatorExecutable}"
           "--unlock"
         ];
         User = "hydra";

packaging/dev-shell.nix

@@ -7,6 +7,7 @@
   hydra-linters,
   hydra-queue-runner,
   hydra-builder,
+  hydra-evaluator,
   foreman,
 }:
 
@@ -21,6 +22,7 @@ let
     hydra-linters
     hydra-queue-runner
     hydra-builder
+    hydra-evaluator
   ];
 
   # Collect and deduplicate build inputs from all components,

subprojects/crates/db/src/lib.rs

@@ -37,6 +37,23 @@ impl Database {
         })
     }
 
+    pub async fn new_with_options(
+        options: sqlx::postgres::PgConnectOptions,
+        max_connections: u32,
+    ) -> Result<Self, Error> {
+        Ok(Self {
+            pool: sqlx::postgres::PgPoolOptions::new()
+                .max_connections(max_connections)
+                .connect_with(options)
+                .await?,
+        })
+    }
+
+    #[must_use]
+    pub fn pool(&self) -> &sqlx::PgPool {
+        &self.pool
+    }
+
     pub async fn get(&self) -> Result<Connection, Error> {
         let conn = self.pool.acquire().await?;
         Ok(Connection::new(conn))

subprojects/hydra-builder/package.nix

@@ -38,11 +38,11 @@ rustPlatform.buildRustPackage {
     };
   };
 
-  # The source fileset above intentionally excludes hydra-queue-runner,
-  # so drop it from the workspace members to keep cargo from trying to
-  # load its (absent) manifest.
+  # Drop the other Rust binary crates from the workspace; their sources
+  # are excluded from the fileset above, so cargo would otherwise fail
+  # trying to load their (absent) manifests.
   postPatch = ''
-    sed -i 's|"subprojects/hydra-queue-runner", ||' Cargo.toml
+    sed -i '/hydra-builder/!{/"subprojects\/hydra-/d;}' Cargo.toml
   '';
 
   buildAndTestSubdir = "subprojects/hydra-builder";

subprojects/hydra-evaluator/Cargo.toml

@@ -0,0 +1,19 @@
+[package]
+name                   = "hydra-evaluator"
+version.workspace      = true
+edition                = "2024"
+license                = "GPL-3.0"
+rust-version.workspace = true
+
+[dependencies]
+anyhow.workspace       = true
+clap                   = { workspace = true, features = [ "derive" ] }
+fs-err.workspace       = true
+futures.workspace      = true
+sqlx                   = { workspace = true, features = [ "runtime-tokio", "tls-rustls-ring-webpki", "postgres" ] }
+tokio                  = { workspace = true, features = [ "full" ] }
+tokio-stream.workspace = true
+tracing.workspace      = true
+
+db            = { path = "../crates/db" }
+hydra-tracing = { path = "../crates/tracing" }

subprojects/hydra-evaluator/package.nix

@@ -0,0 +1,64 @@
+{
+  lib,
+  version,
+
+  rustPlatform,
+
+  nixComponents,
+  protobuf,
+  pkg-config,
+  rust-jemalloc-sys,
+}:
+
+rustPlatform.buildRustPackage {
+  pname = "hydra-evaluator";
+  inherit version;
+
+  src = lib.fileset.toSource {
+    root = ../..;
+    fileset = lib.fileset.unions [
+      ../../Cargo.toml
+      ../../Cargo.lock
+      ../../.cargo
+      ../../.sqlx
+      ../../subprojects/hydra-evaluator/Cargo.toml
+      ../../subprojects/hydra-evaluator/src
+      ../../subprojects/crates
+      # For unit tests which want to spin up a fresh database
+      ../../subprojects/hydra/sql/hydra.sql
+      ../../subprojects/proto
+    ];
+  };
+
+  cargoLock = {
+    lockFile = ../../Cargo.lock;
+    outputHashes = {
+      "harmonia-store-core-0.0.0-alpha.0" = "sha256-g7JJGrjWnnzBtxtxLaqL/wKehPBZAHh8C7U7ALYW6o0=";
+    };
+  };
+
+  # Drop the other Rust binary crates from the workspace; their sources
+  # are excluded from the fileset above, so cargo would otherwise fail
+  # trying to load their (absent) manifests.
+  postPatch = ''
+    sed -i '/hydra-evaluator/!{/"subprojects\/hydra-/d;}' Cargo.toml
+  '';
+
+  buildAndTestSubdir = "subprojects/hydra-evaluator";
+
+  nativeBuildInputs = [
+    pkg-config
+    protobuf
+  ];
+
+  buildInputs = [
+    nixComponents.nix-main
+    protobuf
+    rust-jemalloc-sys
+  ];
+
+  # FIXME: get these passing in a prod build
+  doCheck = false;
+
+  meta.description = "Hydra evaluator (Rust)";
+}

subprojects/hydra-evaluator/src/config.rs

@@ -0,0 +1,137 @@
+use std::collections::HashMap;
+
+use anyhow::Context as _;
+use sqlx::postgres::{PgConnectOptions, PgSslMode};
+
+#[derive(Debug)]
+pub(crate) struct HydraConfig {
+    options: HashMap<String, String>,
+}
+
+impl HydraConfig {
+    pub(crate) fn load() -> Self {
+        let mut options = HashMap::new();
+
+        let path = match std::env::var("HYDRA_CONFIG") {
+            Ok(p) if !p.is_empty() => p,
+            _ => return Self { options },
+        };
+
+        let contents = match fs_err::read_to_string(&path) {
+            Ok(c) => c,
+            Err(e) => {
+                tracing::warn!("could not read HYDRA_CONFIG at {path}: {e}");
+                return Self { options };
+            }
+        };
+
+        for line in contents.lines() {
+            // Strip comments
+            let line = match line.find('#') {
+                Some(pos) => &line[..pos],
+                None => line,
+            };
+            let line = line.trim();
+
+            let Some(eq) = line.find('=') else {
+                continue;
+            };
+
+            let key = line[..eq].trim();
+            let value = line[eq + 1..].trim();
+
+            if key.is_empty() {
+                continue;
+            }
+
+            options.insert(key.to_owned(), value.to_owned());
+        }
+
+        Self { options }
+    }
+
+    pub(crate) fn get_int(&self, key: &str, default: u64) -> u64 {
+        self.options
+            .get(key)
+            .and_then(|v| v.parse().ok())
+            .unwrap_or(default)
+    }
+}
+
+/// Parse a `HYDRA_DBI` environment variable into `PgConnectOptions`.
+///
+/// Accepts strings like `dbi:Pg:dbname=hydra;host=localhost;port=5432`.
+pub(crate) fn parse_hydra_dbi() -> anyhow::Result<PgConnectOptions> {
+    let dbi = std::env::var("HYDRA_DBI").unwrap_or_else(|_| "dbi:Pg:dbname=hydra;".to_owned());
+    parse_dbi(&dbi)
+}
+
+fn parse_dbi(dbi: &str) -> anyhow::Result<PgConnectOptions> {
+    let params = dbi
+        .strip_prefix("dbi:Pg:")
+        .or_else(|| dbi.strip_prefix("DBI:Pg:"))
+        .context("$HYDRA_DBI does not denote a PostgreSQL database")?;
+
+    let mut opts = PgConnectOptions::new();
+
+    for pair in params.split(';').filter(|s| !s.is_empty()) {
+        let (key, value) = pair
+            .split_once('=')
+            .with_context(|| format!("invalid DBI parameter: {pair}"))?;
+        match key.trim() {
+            "dbname" => opts = opts.database(value.trim()),
+            "host" => opts = opts.host(value.trim()),
+            "port" => {
+                opts = opts.port(
+                    value
+                        .trim()
+                        .parse()
+                        .with_context(|| format!("invalid port: {value}"))?,
+                );
+            }
+            "user" => opts = opts.username(value.trim()),
+            "password" => opts = opts.password(value.trim()),
+            "application_name" => opts = opts.application_name(value.trim()),
+            // The C++ evaluator used libpq which understood these
+            // natively; without explicit handling TLS-required
+            // deployments would fail.
+            "sslmode" => {
+                let mode = match value.trim() {
+                    "disable" => PgSslMode::Disable,
+                    "allow" => PgSslMode::Allow,
+                    "prefer" => PgSslMode::Prefer,
+                    "require" => PgSslMode::Require,
+                    "verify-ca" => PgSslMode::VerifyCa,
+                    "verify-full" => PgSslMode::VerifyFull,
+                    v => anyhow::bail!("invalid sslmode: {v}"),
+                };
+                opts = opts.ssl_mode(mode);
+            }
+            "sslrootcert" => opts = opts.ssl_root_cert(value.trim()),
+            "sslcert" => opts = opts.ssl_client_cert(value.trim()),
+            "sslkey" => opts = opts.ssl_client_key(value.trim()),
+            // Warn rather than bail on unknown parameters, to avoid
+            // breaking on libpq keywords we haven't mapped yet.
+            other => {
+                tracing::warn!("ignoring unsupported DBI parameter: {other}");
+            }
+        }
+    }
+
+    Ok(opts)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parse_simple_dbi() {
+        parse_dbi("dbi:Pg:dbname=hydra;host=localhost;port=5432").unwrap();
+    }
+
+    #[test]
+    fn parse_dbi_uppercase() {
+        parse_dbi("DBI:Pg:dbname=testdb;").unwrap();
+    }
+}