Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412package config
import ( "context" "fmt" "math" "time"
"github.com/bluesky-social/indigo/atproto/syntax" "github.com/sethvargo/go-envconfig" "tangled.org/core/spindle/quota" "tangled.org/core/xrpc/serviceauth")
type Server struct { ListenAddr string `env:"LISTEN_ADDR, default=0.0.0.0:6555"` MetricsListenAddr string `env:"METRICS_LISTEN_ADDR, default=127.0.0.1:9091"` DBPath string `env:"DB_PATH, default=spindle.db"` RepoDir string `env:"REPO_DIR, default=repos"` Hostname string `env:"HOSTNAME, required"` JetstreamEndpoint string `env:"JETSTREAM_ENDPOINT, default=wss://jetstream1.us-west.bsky.network/subscribe"` Tap Tap `env:",prefix=TAP_"` 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"` Secrets Secrets `env:",prefix=SECRETS_"` LogDir string `env:"LOG_DIR, default=/var/log/spindle"` QueueSize int `env:"QUEUE_SIZE, default=100"` MaxJobCount int `env:"MAX_JOB_COUNT, default=2"` // max number of pipelines that run at a time DockerSocket string `env:"DOCKER_SOCKET"` // path to a docker socket to expose to workflow containers}
type Tap struct { Embed bool `env:"EMBED, default=true"` Url string `env:"URL, default=http://[::1]:2480"` Bind string `env:"BIND, default=[::1]:2480"` DBPath string `env:"DB_PATH, default=tap.db"` RelayUrl string `env:"RELAY_URL, default=https://bsky.network"` AdminPassword string `env:"ADMIN_PASSWORD"`}
func (s Server) Did() syntax.DID { return serviceauth.DidWeb(s.Hostname)}
type Secrets struct { Provider string `env:"PROVIDER, default=sqlite"` OpenBao OpenBaoConfig `env:",prefix=OPENBAO_"`}
type OpenBaoConfig struct { ProxyAddr string `env:"PROXY_ADDR, default=http://127.0.0.1:8200"` Mount string `env:"MOUNT, default=spindle"`}
type NixeryPipelines struct { Nixery string `env:"NIXERY, default=nixery.tangled.sh"` WorkflowTimeout string `env:"WORKFLOW_TIMEOUT, default=5m"` MaxJobMemoryMB int64 `env:"MAX_JOB_MEMORY_MB, default=6144"` // per-container memory limit in MiB (default 6 GiB) MaxConcurrentWorkflows int `env:"MAX_CONCURRENT_WORKFLOWS, default=8"` // max number of workflow containers running at once (memory cap)}
type ArtifactStoreDisk struct { Dir string `env:"DIR"`}
type ArtifactStoreS3 struct { Bucket string `env:"BUCKET"` Region string `env:"REGION, default=us-east-1"`}
type ArtifactStores struct { Disk ArtifactStoreDisk `env:",prefix=DISK_"` S3 ArtifactStoreS3 `env:",prefix=S3_"`}
type LegacyS3 struct { LogBucket string `env:"LOG_BUCKET"`}
type Cache struct { Backend string `env:"BACKEND"` // "disk" or "s3" DiskDir string `env:"DISK_DIR"` S3Bucket string `env:"S3_BUCKET"` S3Prefix string `env:"S3_PREFIX, default=spindle/cache"` StoreID string `env:"STORE_ID"` Retention time.Duration `env:"RETENTION, default=720h"` PruneInterval time.Duration `env:"PRUNE_INTERVAL, default=1h"` MaxBytesPerOwner int64 `env:"MAX_BYTES_PER_OWNER, default=10737418240"` MaxEntriesPerOwner int64 `env:"MAX_ENTRIES_PER_OWNER, default=1000"`}
type MicroVMPipelines struct { ImageDir string `env:"IMAGE_DIR"` OverlayDir string `env:"OVERLAY_DIR"` // where microVM temporary disks will live DefaultImage string `env:"DEFAULT_IMAGE, default=nixos-x86_64"` AgentPort uint32 `env:"AGENT_PORT, default=10240"` EnableKVM bool `env:"ENABLE_KVM, default=true"` WorkflowTimeout string `env:"WORKFLOW_TIMEOUT, default=5m"`
MaxTotalMemoryMiB int64 `env:"MAX_TOTAL_MEMORY_MIB, default=0"` MaxTotalVCPUs int64 `env:"MAX_TOTAL_VCPUS, default=0"` MaxTotalDiskMiB int64 `env:"MAX_TOTAL_DISK_MIB, default=0"`
MaxWorkflowMemoryMiB int64 `env:"MAX_WORKFLOW_MEMORY_MIB, default=0"` MaxWorkflowVCPUs int64 `env:"MAX_WORKFLOW_VCPUS, default=0"` MaxWorkflowDiskMiB int64 `env:"MAX_WORKFLOW_DISK_MIB, default=0"`
AgingThreshold time.Duration `env:"AGING_THRESHOLD, default=30s"`
DebugSSH DebugSSH `env:",prefix=DEBUG_SSH_"`
EnableCgroups bool `env:"ENABLE_CGROUPS, default=false"` CgroupParent string `env:"CGROUP_PARENT, default=self"` CgroupPidsMax int64 `env:"CGROUP_PIDS_MAX, default=4096"` CgroupSwapMaxMiB *int64 `env:"CGROUP_SWAP_MAX_MIB"` // cpu.max quota as a percentage of one core. 0 caps each vm at its // configured vcpu count, negative disables the limit CgroupCPUMaxPercent int64 `env:"CGROUP_CPU_MAX_PERCENT, default=0"` // io.weight for workflow cgroups (1-10000, kernel default 100). 0 leaves // io unlimited CgroupIOWeight uint64 `env:"CGROUP_IO_WEIGHT, default=0"` // memory.min that will get assigned to the supervisor (spindle itself) cgroup CgroupSupervisorMemoryMinMiB int64 `env:"CGROUP_SUPERVISOR_MEMORY_MIN_MIB, default=512"`}
type DebugSSH struct { Enabled bool `env:"ENABLED, default=false"` ListenAddr string `env:"LISTEN_ADDR, default=0.0.0.0:2222"` JumpHost string `env:"JUMP_HOST"` Host string `env:"HOST"` // path to private key; if empty, spindle will generate one next to the db HostKeyPath string `env:"HOST_KEY_PATH"` // how long to keep a failed wf alive after failure, for sshing in GracePeriod time.Duration `env:"GRACE_PERIOD, default=5m"`}
type NixCache struct { ReadURLs []string `env:"READ_URLS"` TrustedPublicKeys []string `env:"TRUSTED_PUBLIC_KEYS"` UploadURL string `env:"UPLOAD_URL"` RequireSignedUploads bool `env:"REQUIRE_SIGNED_UPLOADS, default=false"` MaxStagedMiB int64 `env:"MAX_STAGED_MIB, default=5120"` MaxStagedObjects int `env:"MAX_STAGED_OBJECTS, default=1024"`}
type Quota struct { User struct { CacheStorageMiB int64 `env:"CACHE_STORAGE_MIB, default=0"` Workflows int64 `env:"WORKFLOWS, default=0"` VCPUs int64 `env:"VCPUS, default=0"` MemoryMiB int64 `env:"MEMORY_MIB, default=0"` DiskMiB int64 `env:"DISK_MIB, default=0"` Webhooks int64 `env:"WEBHOOKS, default=0"` } `env:",prefix=USER_"` Repo struct { CacheStorageMiB int64 `env:"CACHE_STORAGE_MIB, default=0"` Workflows int64 `env:"WORKFLOWS, default=0"` VCPUs int64 `env:"VCPUS, default=0"` MemoryMiB int64 `env:"MEMORY_MIB, default=0"` DiskMiB int64 `env:"DISK_MIB, default=0"` Webhooks int64 `env:"WEBHOOKS, default=0"` } `env:",prefix=REPO_"`}
type quotaEnvironment struct { Quota Quota `env:",prefix=SPINDLE_QUOTA_"`}
// governs how spindle places and runs jobstype Role string
const ( RoleStandalone Role = "standalone" RoleMill Role = "mill" RoleExecutor Role = "executor")
// fields are selectively active depending on the roletype Mill struct { URL string `env:"URL"` // mill websocket endpoint dialled by the executor SharedSecret string `env:"SHARED_SECRET"` // the executor's token for dialing the mill MaxPending int `env:"MAX_PENDING, default=100"` // mill pending job queue limit ReconnectGrace time.Duration `env:"RECONNECT_GRACE, default=45s"` // reconnect window before leases are failed CancelAckTimeout time.Duration `env:"CANCEL_ACK_TIMEOUT, default=10s"` CancelTeardownTimeout time.Duration `env:"CANCEL_TEARDOWN_TIMEOUT, default=10m"` DrainTimeout time.Duration `env:"DRAIN_TIMEOUT, default=20m"` Seats int `env:"SEATS, default=4"` // executor seats advertised to the mill Labels []string `env:"LABELS"` // executor capability labels ArtifactStore string `env:"ARTIFACT_STORE"` // store shared by mill and its executors JumpListenAddr string `env:"JUMP_LISTEN_ADDR"` JumpHostKeyPath string `env:"JUMP_HOST_KEY_PATH"` DebugExecutorPort uint32 `env:"DEBUG_EXECUTOR_PORT, default=2223"` MaxJumpConnections int `env:"MAX_JUMP_CONNECTIONS, default=128"`}type Tracing struct { Endpoint string `env:"ENDPOINT"` ServiceName string `env:"SERVICE_NAME, default=spindle"` SampleRatio float64 `env:"SAMPLE_RATIO, default=1"` Insecure bool `env:"INSECURE, default=false"`}
type Logging struct { Endpoint string `env:"ENDPOINT"` ServiceName string `env:"SERVICE_NAME, default=spindle"` Insecure bool `env:"INSECURE, default=false"`}
type Schedule struct { MinimumInterval time.Duration `env:"MINIMUM_INTERVAL, default=5m"` Concurrency int `env:"CONCURRENCY, default=4"`}
type Config struct { Role Role `env:"SPINDLE_ROLE, default=standalone"` Server Server `env:",prefix=SPINDLE_SERVER_"` NixeryPipelines NixeryPipelines `env:",prefix=SPINDLE_NIXERY_PIPELINES_"` MicroVMPipelines MicroVMPipelines `env:",prefix=SPINDLE_MICROVM_PIPELINES_"` NixCache NixCache `env:",prefix=SPINDLE_NIX_CACHE_"` Quota Quota `env:",prefix=SPINDLE_QUOTA_"` ArtifactStores ArtifactStores `env:",prefix=SPINDLE_ARTIFACT_STORES_"` LegacyS3 LegacyS3 `env:",prefix=SPINDLE_S3_"` Mill Mill `env:",prefix=SPINDLE_MILL_"` Tracing Tracing `env:",prefix=SPINDLE_TRACING_"` Logging Logging `env:",prefix=SPINDLE_LOGGING_"` Cache Cache `env:",prefix=SPINDLE_CACHE_"` Schedule Schedule `env:",prefix=SPINDLE_SCHEDULE_"`}
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) } case RoleExecutor: if c.Mill.URL == "" { return fmt.Errorf("SPINDLE_ROLE=executor requires SPINDLE_MILL_URL (the mill to dial)") } if c.Mill.SharedSecret == "" { return fmt.Errorf("SPINDLE_ROLE=executor requires SPINDLE_MILL_SHARED_SECRET (its executor token)") } default: return fmt.Errorf("unknown SPINDLE_ROLE %q (want standalone, mill, or executor)", c.Role) } if c.Mill.DrainTimeout <= 0 { return fmt.Errorf("SPINDLE_MILL_DRAIN_TIMEOUT must be greater than zero") } if c.Schedule.MinimumInterval < time.Minute || c.Schedule.MinimumInterval%time.Minute != 0 { return fmt.Errorf("SPINDLE_SCHEDULE_MINIMUM_INTERVAL must be a positive whole number of minutes") } if c.Schedule.Concurrency < 1 { return fmt.Errorf("SPINDLE_SCHEDULE_CONCURRENCY must be greater than zero") } if c.NixCache.MaxStagedMiB <= 0 || c.NixCache.MaxStagedMiB > math.MaxInt64>>20 { return fmt.Errorf("SPINDLE_NIX_CACHE_MAX_STAGED_MIB must be greater than zero and fit in bytes") } if c.NixCache.MaxStagedObjects <= 0 { return fmt.Errorf("SPINDLE_NIX_CACHE_MAX_STAGED_OBJECTS must be greater than zero") } if c.Mill.JumpListenAddr != "" { if c.Role != RoleMill { return fmt.Errorf("SPINDLE_MILL_JUMP_LISTEN_ADDR requires SPINDLE_ROLE=mill") } if c.Mill.JumpHostKeyPath == "" { return fmt.Errorf("SPINDLE_MILL_JUMP_LISTEN_ADDR requires SPINDLE_MILL_JUMP_HOST_KEY_PATH") } if c.Mill.MaxJumpConnections <= 0 { return fmt.Errorf("SPINDLE_MILL_MAX_JUMP_CONNECTIONS must be greater than zero") } } if math.IsNaN(c.Tracing.SampleRatio) || c.Tracing.SampleRatio < 0 || c.Tracing.SampleRatio > 1 { return fmt.Errorf("SPINDLE_TRACING_SAMPLE_RATIO must be between 0 and 1 inclusive, got %f", c.Tracing.SampleRatio) } if c.Tracing.Endpoint != "" && c.Tracing.ServiceName == "" { return fmt.Errorf("SPINDLE_TRACING_SERVICE_NAME must not be empty when tracing is enabled") } if c.Logging.Endpoint != "" && c.Logging.ServiceName == "" { return fmt.Errorf("SPINDLE_LOGGING_SERVICE_NAME must not be empty when logging is enabled") } if c.Quota.User.CacheStorageMiB < 0 { return fmt.Errorf("user cache storage limit cannot be negative, got %d", c.Quota.User.CacheStorageMiB) } if c.Quota.User.Workflows < 0 { return fmt.Errorf("user workflows limit cannot be negative, got %d", c.Quota.User.Workflows) } if c.Quota.User.VCPUs < 0 { return fmt.Errorf("user vcpus limit cannot be negative, got %d", c.Quota.User.VCPUs) } if c.Quota.User.MemoryMiB < 0 { return fmt.Errorf("user memory limit cannot be negative, got %d", c.Quota.User.MemoryMiB) } if c.Quota.User.DiskMiB < 0 { return fmt.Errorf("user disk limit cannot be negative, got %d", c.Quota.User.DiskMiB) }
if c.Quota.Repo.CacheStorageMiB < 0 { return fmt.Errorf("repo cache storage limit cannot be negative, got %d", c.Quota.Repo.CacheStorageMiB) } if c.Quota.Repo.Workflows < 0 { return fmt.Errorf("repo workflows limit cannot be negative, got %d", c.Quota.Repo.Workflows) } if c.Quota.Repo.VCPUs < 0 { return fmt.Errorf("repo vcpus limit cannot be negative, got %d", c.Quota.Repo.VCPUs) } if c.Quota.Repo.MemoryMiB < 0 { return fmt.Errorf("repo memory limit cannot be negative, got %d", c.Quota.Repo.MemoryMiB) } if c.Quota.Repo.DiskMiB < 0 { return fmt.Errorf("repo disk limit cannot be negative, got %d", c.Quota.Repo.DiskMiB) }
if c.Quota.User.CacheStorageMiB > 8796093022207 { return fmt.Errorf("user cache storage limit %d MiB would overflow bytes", c.Quota.User.CacheStorageMiB) } if c.Quota.Repo.CacheStorageMiB > 8796093022207 { return fmt.Errorf("repo cache storage limit %d MiB would overflow bytes", c.Quota.Repo.CacheStorageMiB) } return nil}
func (c *Config) ToDefaults() (quota.Defaults, error) { mibToBytes := func(mib int64) (int64, error) { if mib < 0 { return 0, fmt.Errorf("negative limit: %d", mib) } if mib == 0 { return 0, nil } if mib > 8796093022207 { return 0, fmt.Errorf("limit %d MiB would overflow bytes", mib) } return mib * 1024 * 1024, nil }
limits := []struct { scope string resource string limit int64 }{ {scope: "user", resource: quota.ResourceWorkflows, limit: c.Quota.User.Workflows}, {scope: "user", resource: quota.ResourceVCPUs, limit: c.Quota.User.VCPUs}, {scope: "user", resource: quota.ResourceMemoryMiB, limit: c.Quota.User.MemoryMiB}, {scope: "user", resource: quota.ResourceDiskMiB, limit: c.Quota.User.DiskMiB}, {scope: "repo", resource: quota.ResourceWorkflows, limit: c.Quota.Repo.Workflows}, {scope: "repo", resource: quota.ResourceWebhooks, limit: c.Quota.Repo.Webhooks}, {scope: "user", resource: quota.ResourceWebhooks, limit: c.Quota.User.Webhooks}, {scope: "repo", resource: quota.ResourceVCPUs, limit: c.Quota.Repo.VCPUs}, {scope: "repo", resource: quota.ResourceMemoryMiB, limit: c.Quota.Repo.MemoryMiB}, {scope: "repo", resource: quota.ResourceDiskMiB, limit: c.Quota.Repo.DiskMiB}, } for _, item := range limits { if item.limit < 0 { return nil, fmt.Errorf("%s %s limit cannot be negative, got %d", item.scope, item.resource, item.limit) } }
userCacheBytes, err := mibToBytes(c.Quota.User.CacheStorageMiB) if err != nil { return nil, err } repoCacheBytes, err := mibToBytes(c.Quota.Repo.CacheStorageMiB) if err != nil { return nil, err }
return quota.Defaults{ quota.ScopeUser: quota.Resources{ quota.ResourceCacheStorageBytes: userCacheBytes, quota.ResourceWorkflows: c.Quota.User.Workflows, quota.ResourceVCPUs: c.Quota.User.VCPUs, quota.ResourceMemoryMiB: c.Quota.User.MemoryMiB, quota.ResourceDiskMiB: c.Quota.User.DiskMiB, quota.ResourceWebhooks: c.Quota.User.Webhooks, }, quota.ScopeRepo: quota.Resources{ quota.ResourceCacheStorageBytes: repoCacheBytes, quota.ResourceWorkflows: c.Quota.Repo.Workflows, quota.ResourceVCPUs: c.Quota.Repo.VCPUs, quota.ResourceMemoryMiB: c.Quota.Repo.MemoryMiB, quota.ResourceDiskMiB: c.Quota.Repo.DiskMiB, quota.ResourceWebhooks: c.Quota.Repo.Webhooks, }, }, nil}
func LoadQuotaDefaults(ctx context.Context) (quota.Defaults, error) { var env quotaEnvironment if err := envconfig.Process(ctx, &env); err != nil { return nil, err } cfg := Config{Quota: env.Quota} return cfg.ToDefaults()}
func Load(ctx context.Context) (*Config, error) { var cfg Config err := envconfig.Process(ctx, &cfg) if err != nil { return nil, err } if err := cfg.validate(); err != nil { return nil, err }
return &cfg, nil}