diff --git a/nix/modules/spindle.nix b/nix/modules/spindle.nix index 949cb57e..28561f4d 100644 --- a/nix/modules/spindle.nix +++ b/nix/modules/spindle.nix @@ -359,73 +359,106 @@ in }; }; - executorType = types.submodule ({ - name, - config, - ... - }: { - options = - processOptions - // { - enable = mkOption { - type = types.bool; - default = false; - description = "Run this executor on the same host."; - }; + localServiceType = name: + types.submodule ({config, ...}: { + options = + processOptions + // { + generateToken = mkOption { + type = types.bool; + default = false; + description = "Generate the token file once at a persistent path."; + }; - tokenFile = mkOption { - type = types.str; - description = "File containing this executor's token."; - }; + millUrl = mkOption { + type = types.str; + default = "ws://127.0.0.1:6555/mill"; + description = "Mill URL used by this executor."; + }; - generateToken = mkOption { - type = types.bool; - default = false; - description = "Generate the token file once at a persistent path."; - }; + seats = mkOption { + type = types.ints.positive; + default = 4; + description = "Jobs this executor may run at once."; + }; - labels = mkOption { - type = types.listOf types.str; - default = []; - description = "Labels this executor may use."; + stateDirectory = mkOption { + type = types.str; + default = "spindle-executor-${name}"; + description = "State directory used by this executor."; + }; }; - url = mkOption { - type = types.str; - default = "ws://127.0.0.1:6555/mill"; - description = "Mill URL used by this executor."; - }; + config = mkMerge [ + { + package = mkDefault cfg.package; + server = mkDefault cfg.server; + artifactStores = mkDefault cfg.artifactStores; + pipelines = mkDefault cfg.pipelines; + environmentFile = mkDefault cfg.environmentFile; + } + { + server.listenAddr = mkOverride 900 "127.0.0.1:0"; + server.dbPath = mkOverride 900 "/var/lib/${config.stateDirectory}/spindle.db"; + server.repoDir = mkOverride 900 "/var/lib/${config.stateDirectory}/repos"; + server.hostname = mkOverride 900 name; + server.owner = mkOverride 900 ""; + pipelines.microvm.debugSsh.listenAddr = mkOverride 900 "127.0.0.1:${toString cfg.mill.debugExecutorPort}"; + } + ]; + }); - seats = mkOption { - type = types.ints.positive; - default = 4; - description = "Jobs this executor may run at once."; - }; + executorType = types.submodule ({name, ...}: { + options = { + tokenFile = mkOption { + type = types.str; + description = "File containing this executor's token."; + }; - stateDirectory = mkOption { - type = types.str; - default = "spindle-executor-${name}"; - description = "State directory used by this executor."; - }; + labels = mkOption { + type = types.listOf types.str; + default = []; + description = "Labels this executor may use."; }; - config = mkMerge [ - { - package = mkDefault cfg.package; - server = mkDefault cfg.server; - artifactStores = mkDefault cfg.artifactStores; - pipelines = mkDefault cfg.pipelines; - environmentFile = mkDefault cfg.environmentFile; - } - { - server.listenAddr = mkOverride 900 "127.0.0.1:0"; - server.dbPath = mkOverride 900 "/var/lib/${config.stateDirectory}/spindle.db"; - server.repoDir = mkOverride 900 "/var/lib/${config.stateDirectory}/repos"; - server.hostname = mkOverride 900 name; - pipelines.microvm.debugSsh.listenAddr = mkOverride 900 "127.0.0.1:${toString cfg.mill.debugExecutorPort}"; - } - ]; + localService = mkOption { + type = with types; nullOr (localServiceType name); + default = null; + description = "Service running this executor on the mill host."; + }; + }; }); + + dedicatedExecutorType = types.submodule { + options = { + name = mkOption { + type = types.str; + description = "Name registered for this executor."; + }; + + millUrl = mkOption { + type = types.str; + description = "Mill URL used by this executor."; + }; + + tokenFile = mkOption { + type = types.str; + description = "File containing this executor's token."; + }; + + labels = mkOption { + type = types.listOf types.str; + default = []; + description = "Labels this executor may use."; + }; + + seats = mkOption { + type = types.ints.positive; + default = 4; + description = "Jobs this executor may run at once."; + }; + }; + }; in { services.tangled.spindle = processOptions @@ -437,11 +470,17 @@ in }; role = mkOption { - type = types.enum ["standalone" "mill"]; + type = types.enum ["standalone" "mill" "executor"]; default = "standalone"; description = "How the main spindle runs."; }; + executor = mkOption { + type = with types; nullOr dedicatedExecutorType; + default = null; + description = "Settings used when the main spindle runs as an executor."; + }; + mill = { artifactStore = mkOption { type = types.enum ["disk" "s3"]; @@ -499,16 +538,29 @@ in pkgs.util-linux config.nix.package ]; - enabledExecutors = filterAttrs (_: executor: executor.enable) cfg.mill.executors; - enabledAgentPorts = mapAttrsToList (_: executor: executor.pipelines.microvm.agentPort) enabledExecutors; - enabledStateDirectories = mapAttrsToList (_: executor: executor.stateDirectory) enabledExecutors; - enabledDbPaths = mapAttrsToList (_: executor: toString executor.server.dbPath) enabledExecutors; - enabledRepoDirs = mapAttrsToList (_: executor: toString executor.server.repoDir) enabledExecutors; - debugExecutors = filterAttrs (_: executor: executor.pipelines.microvm.debugSsh.enable) enabledExecutors; - enabledDebugSshListenAddrs = - mapAttrsToList (_: executor: executor.pipelines.microvm.debugSsh.listenAddr) debugExecutors; - hasExecutors = cfg.mill.executors != {}; + localExecutors = filterAttrs (_: executor: executor.localService != null) cfg.mill.executors; + localServices = mapAttrs (_: executor: executor.localService) localExecutors; + localAgentPorts = mapAttrsToList (_: service: service.pipelines.microvm.agentPort) localServices; + localStateDirectories = mapAttrsToList (_: service: service.stateDirectory) localServices; + localDbPaths = mapAttrsToList (_: service: toString service.server.dbPath) localServices; + localRepoDirs = mapAttrsToList (_: service: toString service.server.repoDir) localServices; + debugServices = filterAttrs (_: service: service.pipelines.microvm.debugSsh.enable) localServices; + localDebugSshListenAddrs = + mapAttrsToList (_: service: service.pipelines.microvm.debugSsh.listenAddr) debugServices; + hasRegistrations = cfg.role == "mill" && cfg.mill.executors != {}; + isDedicatedExecutor = cfg.role == "executor" && cfg.executor != null; executorServiceName = name: "spindle-executor-${name}"; + executorExecStart = name: package: + pkgs.writeShellScript name '' + export SPINDLE_MILL_SHARED_SECRET="$(${pkgs.coreutils}/bin/cat "$CREDENTIALS_DIRECTORY/mill-token")" + exec ${pkgs.coreutils}/bin/env \ + SPINDLE_ROLE=executor \ + SPINDLE_MILL_ARTIFACT_STORE=${cfg.mill.artifactStore} \ + SPINDLE_MILL_DRAIN_TIMEOUT=${toString cfg.mill.drainTimeout}s \ + SPINDLE_MILL_JUMP_LISTEN_ADDR= \ + SPINDLE_MILL_JUMP_HOST_KEY_PATH= \ + ${package}/bin/spindle + ''; processEnvironment = instance: [ "SPINDLE_SERVER_LISTEN_ADDR=${instance.server.listenAddr}" @@ -566,6 +618,12 @@ in "SPINDLE_MILL_DRAIN_TIMEOUT=${toString cfg.mill.drainTimeout}s" ]; + connectionEnvironment = executor: [ + "SPINDLE_MILL_URL=${executor.millUrl}" + "SPINDLE_MILL_SEATS=${toString executor.seats}" + "SPINDLE_MILL_LABELS=${concatStringsSep "," executor.labels}" + ]; + mainEnvironment = processEnvironment cfg ++ [ @@ -574,16 +632,16 @@ in "SPINDLE_MILL_JUMP_HOST_KEY_PATH=${optionalString (cfg.mill.jumpHostKeyPath != null) (toString cfg.mill.jumpHostKeyPath)}" "SPINDLE_MILL_DEBUG_EXECUTOR_PORT=${toString cfg.mill.debugExecutorPort}" "SPINDLE_MILL_MAX_JUMP_CONNECTIONS=${toString cfg.mill.maxJumpConnections}" - ]; - - executorEnvironment = executor: - processEnvironment executor - ++ [ - "SPINDLE_ROLE=executor" - "SPINDLE_MILL_URL=${executor.url}" - "SPINDLE_MILL_SEATS=${toString executor.seats}" - "SPINDLE_MILL_LABELS=${concatStringsSep "," executor.labels}" - ]; + ] + ++ optionals isDedicatedExecutor (connectionEnvironment cfg.executor); + + localExecutorEnvironment = executor: + processEnvironment executor.localService + ++ ["SPINDLE_ROLE=executor"] + ++ connectionEnvironment { + inherit (executor) labels; + inherit (executor.localService) millUrl seats; + }; registrationCommands = concatStringsSep "\n" (mapAttrsToList ( name: executor: let @@ -616,7 +674,7 @@ in emptyGeneratedToken = escapeShellArg "generated token for ${name} is empty"; missingToken = escapeShellArg "token file for ${name} is missing or empty: ${executor.tokenFile}"; in '' - ${optionalString executor.generateToken '' + ${optionalString (executor.localService != null && executor.localService.generateToken) '' if [ ! -e ${tokenFile} ]; then if [ ! -d ${tokenDir} ]; then ${pkgs.coreutils}/bin/install -d -m 0700 ${tokenDir} @@ -645,7 +703,9 @@ in executorServices = mapAttrs' ( - name: executor: + name: executor: let + service = executor.localService; + in nameValuePair (executorServiceName name) { description = "spindle mill executor ${name}"; after = [ @@ -662,27 +722,18 @@ in path = deps; serviceConfig = { LogsDirectory = "spindle"; - StateDirectory = executor.stateDirectory; - Delegate = executor.pipelines.microvm.cgroup.enable; - EnvironmentFile = mkIf (executor.environmentFile != null) executor.environmentFile; - Environment = executorEnvironment executor; + StateDirectory = service.stateDirectory; + Delegate = service.pipelines.microvm.cgroup.enable; + EnvironmentFile = mkIf (service.environmentFile != null) service.environmentFile; + Environment = localExecutorEnvironment executor; LoadCredential = "mill-token:${executor.tokenFile}"; - ExecStart = pkgs.writeShellScript (executorServiceName name) '' - export SPINDLE_MILL_SHARED_SECRET="$(${pkgs.coreutils}/bin/cat "$CREDENTIALS_DIRECTORY/mill-token")" - exec ${pkgs.coreutils}/bin/env \ - SPINDLE_ROLE=executor \ - SPINDLE_MILL_ARTIFACT_STORE=${cfg.mill.artifactStore} \ - SPINDLE_MILL_DRAIN_TIMEOUT=${toString cfg.mill.drainTimeout}s \ - SPINDLE_MILL_JUMP_LISTEN_ADDR= \ - SPINDLE_MILL_JUMP_HOST_KEY_PATH= \ - ${executor.package}/bin/spindle - ''; + ExecStart = executorExecStart (executorServiceName name) service.package; TimeoutStopSec = cfg.mill.drainTimeout + 510; Restart = "always"; }; } ) - enabledExecutors; + localExecutors; in mkIf cfg.enable { assertions = @@ -698,69 +749,94 @@ in cfg.mill.executors ++ mapAttrsToList (name: executor: { assertion = - !executor.generateToken + executor.localService + == null + || !executor.localService.generateToken || (!hasPrefix "/run/" executor.tokenFile && !hasPrefix "/tmp/" executor.tokenFile); - message = "services.tangled.spindle.mill.executors.${name}.tokenFile must persist across reboots when generateToken is enabled"; + message = "services.tangled.spindle.mill.executors.${name}.tokenFile must persist when localService.generateToken is enabled"; }) cfg.mill.executors ++ [ { - assertion = length enabledAgentPorts == length (unique enabledAgentPorts); - message = "services.tangled.spindle.mill.executors must use different pipelines.microvm.agentPort values"; + assertion = (cfg.role == "executor") == (cfg.executor != null); + message = "services.tangled.spindle.executor must be set exactly when role is executor"; + } + { + assertion = cfg.mill.executors == {} || cfg.role == "mill"; + message = "services.tangled.spindle.mill.executors requires role mill"; + } + { + assertion = cfg.executor == null || !hasPrefix "/nix/store/" cfg.executor.tokenFile; + message = "services.tangled.spindle.executor.tokenFile must not point into the Nix store"; + } + { + assertion = cfg.executor == null || hasPrefix "/" cfg.executor.tokenFile; + message = "services.tangled.spindle.executor.tokenFile must be an absolute path"; } { - assertion = all (port: port != cfg.pipelines.microvm.agentPort) enabledAgentPorts; + assertion = length localAgentPorts == length (unique localAgentPorts); + message = "services.tangled.spindle.mill.executors must use different localService.pipelines.microvm.agentPort values"; + } + { + assertion = all (port: port != cfg.pipelines.microvm.agentPort) localAgentPorts; message = "services.tangled.spindle.mill.executors must not use the main spindle's pipelines.microvm.agentPort"; } { - assertion = length enabledStateDirectories == length (unique enabledStateDirectories); - message = "services.tangled.spindle.mill.executors must use different state directories"; + assertion = length localStateDirectories == length (unique localStateDirectories); + message = "services.tangled.spindle.mill.executors must use different localService state directories"; } { - assertion = !elem "spindle" enabledStateDirectories; + assertion = !elem "spindle" localStateDirectories; message = "services.tangled.spindle.mill.executors must not use the main spindle's state directory"; } { - assertion = length enabledDbPaths == length (unique enabledDbPaths); - message = "services.tangled.spindle.mill.executors must use different server.dbPath values"; + assertion = length localDbPaths == length (unique localDbPaths); + message = "services.tangled.spindle.mill.executors must use different localService.server.dbPath values"; } { - assertion = !elem (toString cfg.server.dbPath) enabledDbPaths; + assertion = !elem (toString cfg.server.dbPath) localDbPaths; message = "services.tangled.spindle.mill.executors must not use the main spindle's server.dbPath"; } { - assertion = length enabledRepoDirs == length (unique enabledRepoDirs); - message = "services.tangled.spindle.mill.executors must use different server.repoDir values"; + assertion = length localRepoDirs == length (unique localRepoDirs); + message = "services.tangled.spindle.mill.executors must use different localService.server.repoDir values"; } { - assertion = !elem (toString cfg.server.repoDir) enabledRepoDirs; + assertion = !elem (toString cfg.server.repoDir) localRepoDirs; message = "services.tangled.spindle.mill.executors must not use the main spindle's server.repoDir"; } { - assertion = length enabledDebugSshListenAddrs == length (unique enabledDebugSshListenAddrs); - message = "services.tangled.spindle.mill.executors must use different pipelines.microvm.debugSsh.listenAddr values"; + assertion = length localDebugSshListenAddrs == length (unique localDebugSshListenAddrs); + message = "services.tangled.spindle.mill.executors must use different localService.pipelines.microvm.debugSsh.listenAddr values"; } { assertion = if cfg.mill.artifactStore == "disk" - then all (executor: toString executor.artifactStores.disk.dir == toString cfg.artifactStores.disk.dir) (attrValues enabledExecutors) + then all (service: toString service.artifactStores.disk.dir == toString cfg.artifactStores.disk.dir) (attrValues localServices) else all ( - executor: - executor.artifactStores.s3.bucket + service: + service.artifactStores.s3.bucket == cfg.artifactStores.s3.bucket - && executor.artifactStores.s3.region == cfg.artifactStores.s3.region + && service.artifactStores.s3.region == cfg.artifactStores.s3.region ) - (attrValues enabledExecutors); + (attrValues localServices); message = "services.tangled.spindle.mill.executors must use the main spindle's configured artifact store"; } ]; + services.tangled.spindle.server = mkIf isDedicatedExecutor { + listenAddr = mkDefault "127.0.0.1:0"; + hostname = mkForce cfg.executor.name; + owner = mkForce ""; + }; + environment.systemPackages = [ (pkgs.writeShellScriptBin "spindle" '' export PATH="${lib.makeBinPath deps}:$PATH" ${lib.optionalString (cfg.environmentFile != null) "set -a; source ${cfg.environmentFile}; set +a"} ${lib.concatMapStringsSep "\n" (entry: "export ${entry}") mainEnvironment} + ${lib.optionalString isDedicatedExecutor "export SPINDLE_MILL_SHARED_SECRET=\"$(${pkgs.coreutils}/bin/cat ${escapeShellArg cfg.executor.tokenFile})\""} exec ${cfg.package}/bin/spindle "$@" '') ]; @@ -769,11 +845,11 @@ in systemd.services = { - spindle-mill-executors = mkIf hasExecutors { + spindle-mill-executors = mkIf hasRegistrations { description = "register spindle mill executors"; before = ["spindle.service"] - ++ mapAttrsToList (name: _: "${executorServiceName name}.service") enabledExecutors; + ++ mapAttrsToList (name: _: "${executorServiceName name}.service") localExecutors; wantedBy = ["multi-user.target"]; serviceConfig = { Type = "oneshot"; @@ -790,8 +866,8 @@ in "network.target" "docker.service" ] - ++ optional hasExecutors "spindle-mill-executors.service"; - requires = optional hasExecutors "spindle-mill-executors.service"; + ++ optional hasRegistrations "spindle-mill-executors.service"; + requires = optional hasRegistrations "spindle-mill-executors.service"; wantedBy = ["multi-user.target"]; path = deps; serviceConfig = { @@ -800,7 +876,11 @@ in Delegate = cfg.pipelines.microvm.cgroup.enable; EnvironmentFile = mkIf (cfg.environmentFile != null) cfg.environmentFile; Environment = mainEnvironment; - ExecStart = "${pkgs.coreutils}/bin/env SPINDLE_ROLE=${cfg.role} SPINDLE_MILL_ARTIFACT_STORE=${cfg.mill.artifactStore} SPINDLE_MILL_DRAIN_TIMEOUT=${toString cfg.mill.drainTimeout}s ${cfg.package}/bin/spindle"; + LoadCredential = mkIf isDedicatedExecutor "mill-token:${cfg.executor.tokenFile}"; + ExecStart = + if isDedicatedExecutor + then executorExecStart "spindle-executor" cfg.package + else "${pkgs.coreutils}/bin/env SPINDLE_ROLE=${cfg.role} SPINDLE_MILL_ARTIFACT_STORE=${cfg.mill.artifactStore} SPINDLE_MILL_DRAIN_TIMEOUT=${toString cfg.mill.drainTimeout}s ${cfg.package}/bin/spindle"; TimeoutStopSec = cfg.mill.drainTimeout + 510; Restart = "always"; }; diff --git a/spindle/config/config.go b/spindle/config/config.go index b41646af..1db8298b 100644 --- a/spindle/config/config.go +++ b/spindle/config/config.go @@ -20,7 +20,7 @@ type Server struct { PlcUrl string `env:"PLC_URL, default=https://plc.directory"` Dev bool `env:"DEV, default=false"` DevExtraHosts []string `env:"DEV_EXTRA_HOSTS"` - Owner string `env:"OWNER, required"` + Owner string `env:"OWNER"` Secrets Secrets `env:",prefix=SECRETS_"` LogDir string `env:"LOG_DIR, default=/var/log/spindle"` QueueSize int `env:"QUEUE_SIZE, default=100"` @@ -166,6 +166,9 @@ type Config struct { func (c *Config) validate() error { switch c.Role { case RoleStandalone, RoleMill: + if c.Server.Owner == "" { + return fmt.Errorf("SPINDLE_SERVER_OWNER is required when SPINDLE_ROLE=%s", c.Role) + } if c.Mill.URL != "" { return fmt.Errorf("SPINDLE_MILL_URL is set but SPINDLE_ROLE=%s; only an executor dials a mill", c.Role) } diff --git a/spindle/config/config_test.go b/spindle/config/config_test.go index cc700dc1..396bdb23 100644 --- a/spindle/config/config_test.go +++ b/spindle/config/config_test.go @@ -23,6 +23,35 @@ func TestLoadAllowsUnconfiguredMicroVMEngine(t *testing.T) { } } +func TestLoadRequiresOwnerForCoordinatorRoles(t *testing.T) { + t.Setenv("SPINDLE_SERVER_HOSTNAME", "spindle.example.com") + t.Setenv("SPINDLE_SERVER_OWNER", "") + + for _, role := range []string{"standalone", "mill"} { + t.Run(role, func(t *testing.T) { + t.Setenv("SPINDLE_ROLE", role) + if _, err := Load(context.Background()); err == nil { + t.Fatalf("Load accepted SPINDLE_ROLE=%s without an owner", role) + } + }) + } +} + +func TestLoadAllowsExecutorWithoutOwner(t *testing.T) { + t.Setenv("SPINDLE_ROLE", "executor") + t.Setenv("SPINDLE_SERVER_HOSTNAME", "worker-1") + t.Setenv("SPINDLE_MILL_URL", "wss://spindle.example.com/mill") + t.Setenv("SPINDLE_MILL_SHARED_SECRET", "secret") + + cfg, err := Load(context.Background()) + if err != nil { + t.Fatal(err) + } + if cfg.Server.Owner != "" { + t.Fatalf("owner = %q, want empty", cfg.Server.Owner) + } +} + func TestLoadRejectsNonPositiveDrainTimeout(t *testing.T) { t.Setenv("SPINDLE_SERVER_HOSTNAME", "spindle.example.com") t.Setenv("SPINDLE_SERVER_OWNER", "did:web:spindle.example.com")