Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055{ config, lib, pkgs, ...}: let cfg = config.services.tangled.spindle;in with lib; { options = let processOptions = { package = mkOption { type = types.package; description = "Package to use for the spindle"; };
server = { listenAddr = mkOption { type = types.str; default = "0.0.0.0:6555"; description = "Address to listen on"; };
metricsListenAddr = mkOption { type = types.str; default = "127.0.0.1:9091"; description = "Address for the metrics server to listen on"; };
dbPath = mkOption { type = types.path; default = "/var/lib/spindle/spindle.db"; description = "Path to the database file"; };
repoDir = mkOption { type = types.path; default = "/var/lib/spindle/repos"; description = "Path where synced git repositories live"; };
hostname = mkOption { type = types.str; example = "my.spindle.com"; description = "Hostname for the server (required)"; };
plcUrl = mkOption { type = types.str; default = "https://plc.directory"; description = "atproto PLC directory"; };
jetstreamEndpoint = mkOption { type = types.str; default = "wss://jetstream1.us-west.bsky.network/subscribe"; description = "Jetstream endpoint to subscribe to"; };
dev = mkOption { type = types.bool; default = false; description = "Enable development mode (disables signature verification)"; };
owner = mkOption { type = types.str; example = "did:plc:qfpnj4og54vl56wngdriaxug"; description = "DID of owner (required)"; };
maxJobCount = mkOption { type = types.int; default = 2; example = 5; description = "Maximum number of concurrent jobs to run"; };
queueSize = mkOption { type = types.int; default = 100; example = 100; description = "Maximum number of jobs queue up"; };
secrets = { provider = mkOption { type = types.str; default = "sqlite"; description = "Backend to use for secret management, valid options are 'sqlite', and 'openbao'."; };
openbao = { proxyAddr = mkOption { type = types.str; default = "http://127.0.0.1:8200"; description = "Address of the OpenBAO proxy server"; }; mount = mkOption { type = types.str; default = "spindle"; description = "Mount path in OpenBAO to read secrets from"; }; }; };
tap = { embed = mkOption { type = types.bool; default = true; description = "Run an embedded tap inside the spindle process"; };
url = mkOption { type = types.str; default = "http://[::1]:2480"; description = "URL the spindle's tap client dials"; };
bind = mkOption { type = types.str; default = "[::1]:2480"; description = "Loopback address the embedded tap server listens on"; };
dbPath = mkOption { type = types.path; default = "/var/lib/spindle/tap.db"; description = "Path to the embedded tap sqlite database"; };
relayUrl = mkOption { type = types.str; default = "https://bsky.network"; description = "Relay used by the embedded tap firehose"; }; }; };
artifactStores = { disk.dir = mkOption { type = types.path; default = "/var/log/spindle"; description = "Root directory for disk artifacts"; };
s3.bucket = mkOption { type = types.str; default = "tangled-logs"; description = "S3 bucket for artifacts"; };
s3.region = mkOption { type = types.str; default = "us-east-1"; description = "AWS region for the artifact bucket"; }; };
pipelines = { workflowTimeout = mkOption { type = types.str; default = "5m"; description = "Timeout for a whole workflow, covering the wait for a concurrency slot, setup, and every step in it"; };
nixery = { nixery = mkOption { type = types.str; default = "nixery.tangled.sh"; # note: this is *not* on tangled.org yet description = "Nixery instance to use"; };
maxJobMemoryMb = mkOption { type = types.int; default = 6144; description = "Memory limit per nixery workflow container in MiB (default 6 GiB)"; }; maxConcurrentWorkflows = mkOption { type = types.int; default = 8; description = "Maximum number of nixery workflows running simultaneously. Zero disables this limit."; }; };
microvm = { enableKVM = mkOption { type = types.bool; default = true; description = "Enable KVM hardware acceleration"; };
imageDir = mkOption { type = types.str; default = "/var/lib/spindle/images"; description = "Directory containing microVM image spec JSONs or image spec directories"; }; overlayDir = mkOption { type = types.str; default = "/tmp"; description = "Directory to store microVM temporary overlay files"; }; defaultImage = mkOption { type = types.str; default = "nixos"; description = "Default microVM image spec to use if none is specified in workflow"; }; agentPort = mkOption { type = types.port; default = 10240; description = "Host vsock port the microVM agent connects back to"; };
limits = { total = { memoryMiB = mkOption { type = types.int; default = 0; description = "Maximum declared guest memory in MiB allowed across all running microVM workflows. Zero disables this limit."; }; vcpus = mkOption { type = types.int; default = 0; description = "Maximum declared vCPUs allowed across all running microVM workflows. Zero disables this limit."; }; diskMiB = mkOption { type = types.int; default = 0; description = "Maximum declared disk in MiB allowed across all running microVM workflows. Zero disables this limit."; }; };
workflow = { memoryMiB = mkOption { type = types.int; default = 0; description = "Maximum declared guest memory in MiB allowed for a single microVM workflow. Zero disables this limit."; }; vcpus = mkOption { type = types.int; default = 0; description = "Maximum declared vCPUs allowed for a single microVM workflow. Zero disables this limit."; }; diskMiB = mkOption { type = types.int; default = 0; description = "Maximum declared disk in MiB allowed for a single microVM workflow. Zero disables this limit."; }; }; };
cgroup = { enable = mkOption { type = types.bool; default = false; description = "Enable cgroup v2 containment for microVM processes."; }; parent = mkOption { type = types.str; default = "self"; description = "Parent cgroup for microVM workflow cgroups. Use 'self' to resolve the spindle service cgroup."; }; pidsMax = mkOption { type = types.int; default = 4096; description = "Maximum number of processes allowed in each microVM workflow cgroup."; }; swapMaxMiB = mkOption { type = types.int; default = 0; description = "Maximum swap in MiB allowed in each microVM workflow cgroup. Zero disables swap."; }; supervisorMinMiB = mkOption { type = types.int; default = 512; description = '' Amount of memory in MiB that will be protected by the cgroup for the spindle (allowing it to not get OOMed first.) ''; }; };
debugSsh = { enable = mkOption { type = types.bool; default = false; description = '' Enable the debug ssh server that lets authorized users ssh into a failed microVM to debug it. ''; }; listenAddr = mkOption { type = types.str; default = "0.0.0.0:2222"; example = "0.0.0.0:2225"; description = "Address for the debug ssh server to listen on."; }; host = mkOption { type = types.str; default = ""; example = "127.0.0.1"; description = "Host reached from the SSH jump host."; }; jumpHost = mkOption { type = types.str; default = ""; example = "spindle.example.com"; description = "SSH jump host used in the printed debug command."; }; hostKeyPath = mkOption { type = with types; nullOr path; default = null; example = "/var/lib/spindle/debug_ssh_host_key"; description = '' Path to the ssh host key for the debug server. If null, one is generated once and persisted next to the spindle db. ''; }; gracePeriod = mkOption { type = types.str; default = "5m"; description = '' How long a failed workflow's microVM is kept alive for the user to ssh in. ''; }; }; };
nixCache = { readUrls = mkOption { type = types.listOf types.str; default = []; example = ["http://ncps.internal:8501" "ssh-ng://user@my-awesome-cache"]; description = "Nix binary cache URLs the Spindle guest should read from."; };
trustedPublicKeys = mkOption { type = types.listOf types.str; default = []; example = ["internal-1:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="]; description = "Public keys trusted for the configured Nix binary caches."; };
uploadUrl = mkOption { type = types.str; default = ""; example = "local"; description = "Optional cache upload URL used by live cache import paths."; };
requireSignedUploads = mkOption { type = types.bool; default = false; description = "Require uploaded narinfos to have a signature from a configured trusted public key."; }; }; };
tracing = { endpoint = mkOption { type = types.str; default = ""; description = "OpenTelemetry collector endpoint (empty disables tracing)"; };
serviceName = mkOption { type = types.str; default = "spindle"; description = "OpenTelemetry service name"; };
sampleRatio = mkOption { type = types.number; default = 1.0; description = "Fraction of traces to sample"; };
insecure = mkOption { type = types.bool; default = false; description = "Use plaintext OTLP/HTTP"; }; };
logging = { format = mkOption { type = types.enum ["text" "json"]; default = "text"; description = "Local structured log format"; };
endpoint = mkOption { type = types.str; default = ""; description = "OpenTelemetry log collector endpoint (empty disables export)"; };
serviceName = mkOption { type = types.str; default = "spindle"; description = "OpenTelemetry log service name"; };
insecure = mkOption { type = types.bool; default = false; description = "Use plaintext OTLP/HTTP for logs"; }; };
quota = { user = { cacheStorageMiB = mkOption { type = types.int; default = 0; description = "Default cache storage limit for users in MiB. Zero is unlimited."; }; workflows = mkOption { type = types.int; default = 0; description = "Default concurrent workflows limit for users. Zero is unlimited."; }; vCPUs = mkOption { type = types.int; default = 0; description = "Default vCPU limit for users. Zero is unlimited."; }; memoryMiB = mkOption { type = types.int; default = 0; description = "Default memory limit for users in MiB. Zero is unlimited."; }; diskMiB = mkOption { type = types.int; default = 0; description = "Default disk limit for users in MiB. Zero is unlimited."; }; }; repo = { cacheStorageMiB = mkOption { type = types.int; default = 0; description = "Default cache storage limit for repositories in MiB. Zero is unlimited."; }; workflows = mkOption { type = types.int; default = 0; description = "Default concurrent workflows limit for repositories. Zero is unlimited."; }; vCPUs = mkOption { type = types.int; default = 0; description = "Default vCPU limit for repositories. Zero is unlimited."; }; memoryMiB = mkOption { type = types.int; default = 0; description = "Default memory limit for repositories in MiB. Zero is unlimited."; }; diskMiB = mkOption { type = types.int; default = 0; description = "Default disk limit for repositories in MiB. Zero is unlimited."; }; }; };
environmentFile = mkOption { type = with types; nullOr path; default = null; example = "/etc/spindle.env"; description = '' Additional environment file as defined in {manpage}`systemd.exec(5)`.
Sensitive secrets such as {env}`AWS_SECRET_ACCESS_KEY`, {env}`AWS_ACCESS_KEY_ID`, {env}`AWS_REGION` may be passed to the service without making them world readable in the nix store. ''; }; };
localServiceType = name: types.submodule ({config, ...}: { options = processOptions // { generateToken = mkOption { type = types.bool; default = false; description = "Generate the token file once at a persistent path."; };
millUrl = mkOption { type = types.str; default = "ws://127.0.0.1:6555/mill"; description = "Mill URL used by this executor."; };
seats = mkOption { type = types.ints.positive; default = 4; description = "Jobs this executor may run at once."; };
stateDirectory = mkOption { type = types.str; default = "spindle-executor-${name}"; description = "State directory 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; tracing = mkDefault cfg.tracing; logging = mkDefault cfg.logging; quota = mkDefault cfg.quota; } { 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 ""; # a colocated executor cannot share the mill's metrics port; opt in per executor server.metricsListenAddr = mkOverride 900 ""; pipelines.microvm.debugSsh.listenAddr = mkOverride 900 "127.0.0.1:${toString cfg.mill.debugExecutorPort}"; } ]; });
executorType = types.submodule ({name, ...}: { options = { 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."; };
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 // { enable = mkOption { type = types.bool; default = false; description = "Enable a tangled spindle"; };
role = mkOption { 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"]; default = "s3"; description = "Artifact store shared by the mill and its executors."; };
drainTimeout = mkOption { type = types.ints.positive; default = 1200; description = "Seconds an executor waits for running jobs to finish before stopping."; };
executors = mkOption { type = types.attrsOf executorType; default = {}; description = "Executors allowed to connect to this mill."; };
jumpListenAddr = mkOption { type = types.str; default = ""; example = "0.0.0.0:22"; description = "Address for the mill's restricted debug SSH jump server."; };
jumpHostKeyPath = mkOption { type = with types; nullOr path; default = null; example = "/var/lib/spindle/debug_jump_host_key"; description = "Path to the debug SSH jump server host key."; };
debugExecutorPort = mkOption { type = types.port; default = 2223; description = "Private debug SSH port shared by executors."; };
maxJumpConnections = mkOption { type = types.ints.positive; default = 128; description = "Maximum concurrent connections to the mill's debug SSH jump server."; }; }; }; }; config = let deps = [ pkgs.git pkgs.qemu pkgs.e2fsprogs pkgs.slirp4netns pkgs.iproute2 pkgs.util-linux config.nix.package ]; 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; metricsServices = filterAttrs (_: service: service.server.metricsListenAddr != "") localServices; localMetricsListenAddrs = mapAttrsToList (_: service: service.server.metricsListenAddr) metricsServices; 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}" "SPINDLE_SERVER_METRICS_LISTEN_ADDR=${instance.server.metricsListenAddr}" "SPINDLE_SERVER_DB_PATH=${instance.server.dbPath}" "SPINDLE_SERVER_REPO_DIR=${instance.server.repoDir}" "SPINDLE_SERVER_HOSTNAME=${instance.server.hostname}" "SPINDLE_SERVER_PLC_URL=${instance.server.plcUrl}" "SPINDLE_SERVER_JETSTREAM_ENDPOINT=${instance.server.jetstreamEndpoint}" "SPINDLE_SERVER_DEV=${lib.boolToString instance.server.dev}" "SPINDLE_SERVER_OWNER=${instance.server.owner}" "SPINDLE_SERVER_MAX_JOB_COUNT=${toString instance.server.maxJobCount}" "SPINDLE_SERVER_QUEUE_SIZE=${toString instance.server.queueSize}" "SPINDLE_SERVER_SECRETS_PROVIDER=${instance.server.secrets.provider}" "SPINDLE_SERVER_SECRETS_OPENBAO_PROXY_ADDR=${instance.server.secrets.openbao.proxyAddr}" "SPINDLE_SERVER_SECRETS_OPENBAO_MOUNT=${instance.server.secrets.openbao.mount}" "SPINDLE_SERVER_TAP_EMBED=${lib.boolToString instance.server.tap.embed}" "SPINDLE_SERVER_TAP_URL=${instance.server.tap.url}" "SPINDLE_SERVER_TAP_BIND=${instance.server.tap.bind}" "SPINDLE_SERVER_TAP_DB_PATH=${instance.server.tap.dbPath}" "SPINDLE_SERVER_TAP_RELAY_URL=${instance.server.tap.relayUrl}" "SPINDLE_NIXERY_PIPELINES_NIXERY=${instance.pipelines.nixery.nixery}" "SPINDLE_NIXERY_PIPELINES_WORKFLOW_TIMEOUT=${instance.pipelines.workflowTimeout}" "SPINDLE_NIXERY_PIPELINES_MAX_JOB_MEMORY_MB=${toString instance.pipelines.nixery.maxJobMemoryMb}" "SPINDLE_NIXERY_PIPELINES_MAX_CONCURRENT_WORKFLOWS=${toString instance.pipelines.nixery.maxConcurrentWorkflows}" "SPINDLE_MICROVM_PIPELINES_IMAGE_DIR=${instance.pipelines.microvm.imageDir}" "SPINDLE_MICROVM_PIPELINES_OVERLAY_DIR=${instance.pipelines.microvm.overlayDir}" "SPINDLE_MICROVM_PIPELINES_DEFAULT_IMAGE=${instance.pipelines.microvm.defaultImage}" "SPINDLE_MICROVM_PIPELINES_AGENT_PORT=${toString instance.pipelines.microvm.agentPort}" "SPINDLE_MICROVM_PIPELINES_ENABLE_KVM=${lib.boolToString instance.pipelines.microvm.enableKVM}" "SPINDLE_MICROVM_PIPELINES_WORKFLOW_TIMEOUT=${instance.pipelines.workflowTimeout}" "SPINDLE_MICROVM_PIPELINES_MAX_TOTAL_MEMORY_MIB=${toString instance.pipelines.microvm.limits.total.memoryMiB}" "SPINDLE_MICROVM_PIPELINES_MAX_TOTAL_VCPUS=${toString instance.pipelines.microvm.limits.total.vcpus}" "SPINDLE_MICROVM_PIPELINES_MAX_TOTAL_DISK_MIB=${toString instance.pipelines.microvm.limits.total.diskMiB}" "SPINDLE_MICROVM_PIPELINES_MAX_WORKFLOW_MEMORY_MIB=${toString instance.pipelines.microvm.limits.workflow.memoryMiB}" "SPINDLE_MICROVM_PIPELINES_MAX_WORKFLOW_VCPUS=${toString instance.pipelines.microvm.limits.workflow.vcpus}" "SPINDLE_MICROVM_PIPELINES_MAX_WORKFLOW_DISK_MIB=${toString instance.pipelines.microvm.limits.workflow.diskMiB}" "SPINDLE_MICROVM_PIPELINES_ENABLE_CGROUPS=${lib.boolToString instance.pipelines.microvm.cgroup.enable}" "SPINDLE_MICROVM_PIPELINES_CGROUP_PARENT=${instance.pipelines.microvm.cgroup.parent}" "SPINDLE_MICROVM_PIPELINES_CGROUP_PIDS_MAX=${toString instance.pipelines.microvm.cgroup.pidsMax}" "SPINDLE_MICROVM_PIPELINES_CGROUP_SWAP_MAX_MIB=${toString instance.pipelines.microvm.cgroup.swapMaxMiB}" "SPINDLE_MICROVM_PIPELINES_CGROUP_SUPERVISOR_MEMORY_MIN_MIB=${toString instance.pipelines.microvm.cgroup.supervisorMinMiB}" "SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_ENABLED=${lib.boolToString instance.pipelines.microvm.debugSsh.enable}" "SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_LISTEN_ADDR=${instance.pipelines.microvm.debugSsh.listenAddr}" "SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_HOST=${instance.pipelines.microvm.debugSsh.host}" "SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_JUMP_HOST=${instance.pipelines.microvm.debugSsh.jumpHost}" "SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_HOST_KEY_PATH=${optionalString (instance.pipelines.microvm.debugSsh.hostKeyPath != null) (toString instance.pipelines.microvm.debugSsh.hostKeyPath)}" "SPINDLE_MICROVM_PIPELINES_DEBUG_SSH_GRACE_PERIOD=${instance.pipelines.microvm.debugSsh.gracePeriod}" "SPINDLE_NIX_CACHE_READ_URLS=${concatStringsSep "," instance.pipelines.nixCache.readUrls}" "SPINDLE_NIX_CACHE_TRUSTED_PUBLIC_KEYS=${concatStringsSep "," instance.pipelines.nixCache.trustedPublicKeys}" "SPINDLE_NIX_CACHE_UPLOAD_URL=${instance.pipelines.nixCache.uploadUrl}" "SPINDLE_NIX_CACHE_REQUIRE_SIGNED_UPLOADS=${lib.boolToString instance.pipelines.nixCache.requireSignedUploads}" "SPINDLE_QUOTA_USER_CACHE_STORAGE_MIB=${toString instance.quota.user.cacheStorageMiB}" "SPINDLE_QUOTA_USER_WORKFLOWS=${toString instance.quota.user.workflows}" "SPINDLE_QUOTA_USER_VCPUS=${toString instance.quota.user.vCPUs}" "SPINDLE_QUOTA_USER_MEMORY_MIB=${toString instance.quota.user.memoryMiB}" "SPINDLE_QUOTA_USER_DISK_MIB=${toString instance.quota.user.diskMiB}" "SPINDLE_QUOTA_REPO_CACHE_STORAGE_MIB=${toString instance.quota.repo.cacheStorageMiB}" "SPINDLE_QUOTA_REPO_WORKFLOWS=${toString instance.quota.repo.workflows}" "SPINDLE_QUOTA_REPO_VCPUS=${toString instance.quota.repo.vCPUs}" "SPINDLE_QUOTA_REPO_MEMORY_MIB=${toString instance.quota.repo.memoryMiB}" "SPINDLE_QUOTA_REPO_DISK_MIB=${toString instance.quota.repo.diskMiB}" "SPINDLE_ARTIFACT_STORES_DISK_DIR=${instance.artifactStores.disk.dir}" "SPINDLE_ARTIFACT_STORES_S3_BUCKET=${instance.artifactStores.s3.bucket}" "SPINDLE_ARTIFACT_STORES_S3_REGION=${instance.artifactStores.s3.region}" "SPINDLE_MILL_ARTIFACT_STORE=${cfg.mill.artifactStore}" "SPINDLE_MILL_DRAIN_TIMEOUT=${toString cfg.mill.drainTimeout}s" "SPINDLE_TRACING_ENDPOINT=${instance.tracing.endpoint}" "SPINDLE_TRACING_SERVICE_NAME=${instance.tracing.serviceName}" "SPINDLE_TRACING_SAMPLE_RATIO=${toString instance.tracing.sampleRatio}" "SPINDLE_TRACING_INSECURE=${lib.boolToString instance.tracing.insecure}" "SPINDLE_LOG_FORMAT=${instance.logging.format}" "SPINDLE_LOGGING_ENDPOINT=${instance.logging.endpoint}" "SPINDLE_LOGGING_SERVICE_NAME=${instance.logging.serviceName}" "SPINDLE_LOGGING_INSECURE=${lib.boolToString instance.logging.insecure}" ];
connectionEnvironment = executor: [ "SPINDLE_MILL_URL=${executor.millUrl}" "SPINDLE_MILL_SEATS=${toString executor.seats}" "SPINDLE_MILL_LABELS=${concatStringsSep "," executor.labels}" ];
mainEnvironment = processEnvironment cfg ++ [ "SPINDLE_ROLE=${cfg.role}" "SPINDLE_MILL_JUMP_LISTEN_ADDR=${cfg.mill.jumpListenAddr}" "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}" ] ++ optionals isDedicatedExecutor (connectionEnvironment cfg.executor);
localExecutorEnvironment = executor: processEnvironment executor.localService ++ [ "SPINDLE_ROLE=executor" "SPINDLE_SERVER_LOG_DIR=/var/log/${executor.localService.stateDirectory}" ] ++ connectionEnvironment { inherit (executor) labels; inherit (executor.localService) millUrl seats; };
registrationCommands = concatStringsSep "\n" (mapAttrsToList ( name: executor: let tokenFile = escapeShellArg executor.tokenFile; tokenDir = escapeShellArg (builtins.dirOf executor.tokenFile); tokenTemplate = escapeShellArg "${executor.tokenFile}.tmp.XXXXXX"; generateCommand = escapeShellArgs [ "${cfg.package}/bin/spindle" "mill" "executor" "token" "generate" "--db" (toString cfg.server.dbPath) ]; registerCommand = escapeShellArgs ( [ "${cfg.package}/bin/spindle" "mill" "executor" "add" "--db" (toString cfg.server.dbPath) "--token-file" executor.tokenFile ] ++ concatMap (label: ["--label" label]) executor.labels ++ [name] ); emptyGeneratedToken = escapeShellArg "generated token for ${name} is empty"; missingToken = escapeShellArg "token file for ${name} is missing or empty: ${executor.tokenFile}"; in '' ${optionalString (executor.localService != null && executor.localService.generateToken) '' if [ ! -e ${tokenFile} ]; then if [ ! -d ${tokenDir} ]; then ${pkgs.coreutils}/bin/install -d -m 0700 ${tokenDir} fi umask 0077 tmp="$(${pkgs.coreutils}/bin/mktemp ${tokenTemplate})" trap '${pkgs.coreutils}/bin/rm -f "$tmp"' EXIT ${generateCommand} > "$tmp" if [ ! -s "$tmp" ]; then printf '%s\n' ${emptyGeneratedToken} >&2 exit 1 fi ${pkgs.coreutils}/bin/chmod 0600 "$tmp" ${pkgs.coreutils}/bin/mv "$tmp" ${tokenFile} trap - EXIT fi ''} if [ ! -f ${tokenFile} ] || [ ! -s ${tokenFile} ]; then printf '%s\n' ${missingToken} >&2 exit 1 fi ${registerCommand} '' ) cfg.mill.executors);
executorServices = mapAttrs' ( name: executor: let service = executor.localService; in nameValuePair (executorServiceName name) { description = "spindle mill executor ${name}"; after = [ "network.target" "docker.service" "spindle.service" "spindle-mill-executors.service" ]; requires = [ "docker.service" "spindle-mill-executors.service" ]; partOf = ["spindle.service"]; wantedBy = ["multi-user.target"]; path = deps; serviceConfig = { LogsDirectory = service.stateDirectory; 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 = executorExecStart (executorServiceName name) service.package; TimeoutStopSec = cfg.mill.drainTimeout + 510; Restart = "always"; }; } ) localExecutors; in mkIf cfg.enable { assertions = mapAttrsToList (name: executor: { assertion = !hasPrefix "/nix/store/" executor.tokenFile; message = "services.tangled.spindle.mill.executors.${name}.tokenFile must not point into the Nix store"; }) cfg.mill.executors ++ mapAttrsToList (name: executor: { assertion = hasPrefix "/" executor.tokenFile; message = "services.tangled.spindle.mill.executors.${name}.tokenFile must be an absolute path"; }) cfg.mill.executors ++ mapAttrsToList (name: executor: { assertion = executor.localService == null || !executor.localService.generateToken || (!hasPrefix "/run/" executor.tokenFile && !hasPrefix "/tmp/" executor.tokenFile); message = "services.tangled.spindle.mill.executors.${name}.tokenFile must persist when localService.generateToken is enabled"; }) cfg.mill.executors ++ [ { 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 = 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 localStateDirectories == length (unique localStateDirectories); message = "services.tangled.spindle.mill.executors must use different localService state directories"; } { assertion = !elem "spindle" localStateDirectories; message = "services.tangled.spindle.mill.executors must not use the main spindle's state directory"; } { 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) localDbPaths; message = "services.tangled.spindle.mill.executors must not use the main spindle's server.dbPath"; } { 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) localRepoDirs; message = "services.tangled.spindle.mill.executors must not use the main spindle's server.repoDir"; } { assertion = length localDebugSshListenAddrs == length (unique localDebugSshListenAddrs); message = "services.tangled.spindle.mill.executors must use different localService.pipelines.microvm.debugSsh.listenAddr values"; } { assertion = length localMetricsListenAddrs == length (unique localMetricsListenAddrs); message = "services.tangled.spindle.mill.executors must use different localService.server.metricsListenAddr values"; } { assertion = !elem cfg.server.metricsListenAddr localMetricsListenAddrs; message = "services.tangled.spindle.mill.executors must not use the main spindle's server.metricsListenAddr"; } { assertion = if cfg.mill.artifactStore == "disk" then all (service: toString service.artifactStores.disk.dir == toString cfg.artifactStores.disk.dir) (attrValues localServices) else all ( service: service.artifactStores.s3.bucket == cfg.artifactStores.s3.bucket && service.artifactStores.s3.region == cfg.artifactStores.s3.region ) (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 "$@" '') ];
virtualisation.docker.enable = true;
boot.kernelModules = ["vhost" "vsock" "vhost_vsock"];
systemd.services = { spindle-mill-executors = mkIf hasRegistrations { description = "register spindle mill executors"; before = ["spindle.service"] ++ mapAttrsToList (name: _: "${executorServiceName name}.service") localExecutors; wantedBy = ["multi-user.target"]; serviceConfig = { Type = "oneshot"; RemainAfterExit = true; StateDirectory = "spindle"; }; script = registrationCommands; };
spindle = { description = "spindle service"; after = [ "network.target" "docker.service" ] ++ optional hasRegistrations "spindle-mill-executors.service"; requires = optional hasRegistrations "spindle-mill-executors.service"; wantedBy = ["multi-user.target"]; path = deps; serviceConfig = { LogsDirectory = "spindle"; StateDirectory = "spindle"; Delegate = cfg.pipelines.microvm.cgroup.enable; EnvironmentFile = mkIf (cfg.environmentFile != null) cfg.environmentFile; Environment = mainEnvironment; 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"; RestartMode = mkIf (localExecutors != {}) "direct"; }; }; } // executorServices; }; }