package providermain import ( "cmp" "errors" "fmt" "log/slog" "os" "path/filepath" "strconv" "strings" "time" "github.com/johnandersen777/socialweb-computer-kcp/internal/eso" "github.com/johnandersen777/socialweb-computer-kcp/internal/foldmain" "github.com/johnandersen777/socialweb-computer-kcp/internal/hostprocess" "github.com/johnandersen777/socialweb-computer-kcp/internal/kcpclient" "github.com/johnandersen777/socialweb-computer-kcp/internal/kit/logx" "github.com/johnandersen777/socialweb-computer-kcp/internal/portallocator" "github.com/johnandersen777/socialweb-computer-kcp/internal/scale" ) const ( envKCPHost = "SOCIALWEB_KCP_HOST" envKCPKubeconfig = "SOCIALWEB_KCP_KUBECONFIG" envMountContext = "SOCIALWEB_CLUSTERPROVIDER_MOUNT_CONTEXT" envGrant = "SOCIALWEB_CLUSTERPROVIDER_GRANT" envLeasePath = "SOCIALWEB_CLUSTERPROVIDER_LEASE_PATH" envJoin = "SOCIALWEB_CLUSTERPROVIDER_JOIN" envAccount = "SOCIALWEB_CLUSTERPROVIDER_ACCOUNT" envDatastore = "SOCIALWEB_CLUSTERPROVIDER_DATASTORE" envStateDir = "SOCIALWEB_CLUSTERPROVIDER_STATE_DIR" envVaultStateDir = "SOCIALWEB_CLUSTERPROVIDER_VAULT_STATE_DIR" envUnitPrefix = "SOCIALWEB_CLUSTERPROVIDER_UNIT_PREFIX" envHostProcessWorkspace = "SOCIALWEB_CLUSTERPROVIDER_HOSTPROCESS_WORKSPACE" envDatastoreHasRow = "SOCIALWEB_CLUSTERPROVIDER_DATASTORE_HAS_ROW" envAdvertiseAddress = "SOCIALWEB_CLUSTERPROVIDER_ADVERTISE_ADDRESS" envStoreEndpoint = "SOCIALWEB_CLUSTERPROVIDER_STORE_ENDPOINT" envAuditPolicy = "SOCIALWEB_CLUSTERPROVIDER_AUDIT_POLICY" envMarketAdvertiseAddress = "SOCIALWEB_CLUSTERPROVIDER_MARKET_ADVERTISE_ADDRESS" envMarketDenyBidderDids = "SOCIALWEB_CLUSTERPROVIDER_MARKET_DENY_BIDDER_DIDS" envMarketReleaseTimeout = "SOCIALWEB_CLUSTERPROVIDER_MARKET_RELEASE_TIMEOUT" envScaleHold = "SOCIALWEB_CLUSTERPROVIDER_SCALE_HOLD" envNodeJoinGrace = "SOCIALWEB_CLUSTERPROVIDER_NODE_JOIN_GRACE" envScaleMaxNodes = "SOCIALWEB_CLUSTERPROVIDER_SCALE_MAX_NODES" envScalePodsPerNode = "SOCIALWEB_CLUSTERPROVIDER_SCALE_PODS_PER_NODE" EnvScaleHold = envScaleHold EnvScaleMaxNodes = envScaleMaxNodes EnvScalePodsPerNode = envScalePodsPerNode envHost = "SOCIALWEB_CLUSTERPROVIDER_HOST" envCloudControllerManager = "SOCIALWEB_CLUSTERPROVIDER_CLOUD_CONTROLLER_MANAGER" envDumbpipe = "SOCIALWEB_CLUSTERPROVIDER_DUMBPIPE" envLogLevel = "SOCIALWEB_CLUSTERPROVIDER_LOG_LEVEL" envESOManifestDir = "SOCIALWEB_CLUSTERPROVIDER_ESO_MANIFEST_DIR" envESONamespace = "SOCIALWEB_CLUSTERPROVIDER_ESO_NAMESPACE" envExternalSecrets = "SOCIALWEB_CLUSTERPROVIDER_EXTERNAL_SECRETS" envESOBundle = "SOCIALWEB_CLUSTERPROVIDER_ESO_BUNDLE" envK3s = "SOCIALWEB_CLUSTERPROVIDER_K3S" envKine = "SOCIALWEB_CLUSTERPROVIDER_KINE" envBao = "SOCIALWEB_CLUSTERPROVIDER_BAO" envPgDump = "SOCIALWEB_CLUSTERPROVIDER_PG_DUMP" envNodeClaimAgent = "SOCIALWEB_GATEWAY_NODECLAIM_AGENT" ) const DefaultESOManifestDir = "/opt/socialweb/deploy/eso" const DefaultGrantPath = "deploy/tenant-grant.yaml" const DefaultUnitPrefix = "socialweb-" const DefaultReclaimGrace = 5 * time.Minute type Config struct { KCPHost string KCPKubeconfig string MountContext string GrantPath string LeasePath string Join string Account string Datastore string StateDir string VaultStateDir string UnitPrefix string HostProcessWorkspace string AdvertiseAddress string MarketAdvertiseAddress string MarketDenyBidderDids []string MarketReleaseTimeout time.Duration StoreEndpoint string AuditPolicyPath string Host string CloudControllerManager string CloudControllerManagerExec []string FrontExec []string HostnameExec []string Dumbpipe string ESOManifestDir string ESONamespace string ExternalSecrets string ESOBundle string K3s string Kine string Bao string PgDump string NodeClaimAgent *kcpclient.ServiceAccountRef DatastoreHasRow bool ScaleHold time.Duration NodeJoinGrace time.Duration ScaleMaxNodes int ScalePodsPerNode int LogLevel slog.Level } func loadFor(get func(string) string, inv foldmain.Invocation) (Config, error) { cfg := Config{ KCPHost: strings.TrimSpace(get(envKCPHost)), KCPKubeconfig: strings.TrimSpace(get(envKCPKubeconfig)), MountContext: strings.TrimSpace(get(envMountContext)), LeasePath: strings.TrimSpace(get(envLeasePath)), Join: strings.TrimSpace(get(envJoin)), Account: strings.TrimSpace(get(envAccount)), Datastore: strings.TrimSpace(get(envDatastore)), StateDir: strings.TrimSpace(get(envStateDir)), VaultStateDir: strings.TrimSpace(get(envVaultStateDir)), UnitPrefix: strings.TrimSpace(get(envUnitPrefix)), HostProcessWorkspace: strings.TrimSpace(get(envHostProcessWorkspace)), AdvertiseAddress: strings.TrimSpace(get(envAdvertiseAddress)), MarketAdvertiseAddress: strings.TrimSpace(get(envMarketAdvertiseAddress)), StoreEndpoint: strings.TrimSpace(get(envStoreEndpoint)), AuditPolicyPath: strings.TrimSpace(get(envAuditPolicy)), Host: strings.TrimSpace(get(envHost)), CloudControllerManager: strings.TrimSpace(get(envCloudControllerManager)), Dumbpipe: strings.TrimSpace(get(envDumbpipe)), ESOManifestDir: strings.TrimSpace(get(envESOManifestDir)), ESONamespace: strings.TrimSpace(get(envESONamespace)), ExternalSecrets: strings.TrimSpace(get(envExternalSecrets)), ESOBundle: strings.TrimSpace(get(envESOBundle)), K3s: strings.TrimSpace(get(envK3s)), Kine: strings.TrimSpace(get(envKine)), Bao: strings.TrimSpace(get(envBao)), PgDump: strings.TrimSpace(get(envPgDump)), GrantPath: strings.TrimSpace(get(envGrant)), LogLevel: slog.LevelInfo, ScaleMaxNodes: scale.DefaultMaxNodes, ScalePodsPerNode: scale.DefaultPodsPerNode, } cfg.GrantPath = cmp.Or(cfg.GrantPath, DefaultGrantPath) cfg.UnitPrefix = cmp.Or(cfg.UnitPrefix, DefaultUnitPrefix) cfg.HostProcessWorkspace = cmp.Or(cfg.HostProcessWorkspace, hostprocess.DefaultWorkspace) cfg.MountContext = cmp.Or(cfg.MountContext, kcpclient.ShardAdminContext) cfg.CloudControllerManagerExec = inv.Reexec("internal", "cloud-controller-manager") cfg.FrontExec = inv.Reexec("internal", "apiserver-front") cfg.HostnameExec = inv.Reexec("internal", "ns-exec") if cfg.CloudControllerManager != "" { cfg.CloudControllerManagerExec = []string{cfg.CloudControllerManager} } cfg.ESOManifestDir = cmp.Or(cfg.ESOManifestDir, DefaultESOManifestDir) cfg.ESONamespace = cmp.Or(cfg.ESONamespace, eso.DefaultControllerNamespace) cfg.MarketAdvertiseAddress = cmp.Or(cfg.MarketAdvertiseAddress, portallocator.MarketAdvertiseAddress) cfg.MarketDenyBidderDids = splitList(get(envMarketDenyBidderDids)) switch { case cfg.StoreEndpoint != "" && cfg.Datastore != "" && cfg.StoreEndpoint != cfg.Datastore: return Config{}, fmt.Errorf( "%s is %q and %s is %q. They name ONE thing -- the store kine writes this cluster's "+ "objects into -- and two answers to one question are not a preference to be "+ "resolved silently: whichever lost would be a cluster whose objects went somewhere "+ "nobody is looking", envStoreEndpoint, cfg.StoreEndpoint, envDatastore, cfg.Datastore) case cfg.StoreEndpoint == "": cfg.StoreEndpoint = cfg.Datastore } if raw := strings.TrimSpace(get(envScaleHold)); raw != "" { d, err := time.ParseDuration(raw) if err != nil { return Config{}, fmt.Errorf("%s is not a duration: %w", envScaleHold, err) } if d < 0 { return Config{}, fmt.Errorf("%s must not be negative", envScaleHold) } d = cmp.Or(d, scale.NoHold) cfg.ScaleHold = d } if raw := strings.TrimSpace(get(envNodeJoinGrace)); raw != "" { d, err := time.ParseDuration(raw) if err != nil { return Config{}, fmt.Errorf("%s is not a duration: %w", envNodeJoinGrace, err) } if d <= 0 { return Config{}, fmt.Errorf("%s must be positive", envNodeJoinGrace) } cfg.NodeJoinGrace = d } if raw := strings.TrimSpace(get(envScaleMaxNodes)); raw != "" { n, err := strconv.Atoi(raw) if err != nil { return Config{}, fmt.Errorf("%s is not a whole number: %w", envScaleMaxNodes, err) } if n < 1 { return Config{}, fmt.Errorf( "%s is %d, and a ceiling below one node is not a smaller cluster: it is a cluster "+ "that can never place the work it has intent for. Leave it unset for the default "+ "of %d", envScaleMaxNodes, n, scale.DefaultMaxNodes) } cfg.ScaleMaxNodes = n } if raw := strings.TrimSpace(get(envScalePodsPerNode)); raw != "" { n, err := strconv.Atoi(raw) if err != nil { return Config{}, fmt.Errorf("%s is not a whole number: %w", envScalePodsPerNode, err) } if n < 1 { return Config{}, fmt.Errorf( "%s is %d, and a node that holds fewer than one pod is not a capacity to divide by. "+ "Leave it unset for the default of %d", envScalePodsPerNode, n, scale.DefaultPodsPerNode) } cfg.ScalePodsPerNode = n } for _, req := range []struct{ name, value string }{ {envKCPHost, cfg.KCPHost}, {envKCPKubeconfig, cfg.KCPKubeconfig}, {envLeasePath, cfg.LeasePath}, } { if req.value == "" { return Config{}, fmt.Errorf("%s is not set", req.name) } } if cfg.Join != "" { raw := strings.TrimSpace(get(envDatastoreHasRow)) if raw == "" { return Config{}, fmt.Errorf( "%s is not set. Whether the datastore already holds a bootstrap row is the "+ "caller's OBSERVATION and has no default: a wrong \"false\" makes k3s exit "+ "at boot with \"bootstrap data already found and encrypted with different "+ "token\" (ADR 0023)", envDatastoreHasRow) } has, err := parseBool(raw) if err != nil { return Config{}, fmt.Errorf("%s: %w", envDatastoreHasRow, err) } cfg.DatastoreHasRow = has if cfg.StateDir == "" { return Config{}, fmt.Errorf( "%s is not set. It holds this cluster's node records, its instance token, its "+ "audit log and its identity bundle, and without it a gateway that restarts "+ "can neither find nor reap the agent the last one started -- so it is required "+ "whether or not %s names an external store", envStateDir, envStoreEndpoint) } if !filepath.IsAbs(cfg.StateDir) { return Config{}, fmt.Errorf( "%s is %q, which is relative. A node's identity IS the absolute socket path under "+ "this directory -- that string is what a running VMM is recognised by, what a "+ "reap compares against, and what a record written by an earlier gateway names. "+ "A relative path makes that identity depend on the working directory the gateway "+ "was started from, so the same node started from two directories yields two "+ "identities, and the second one is invisible to the check that exists to stop two "+ "VMMs running behind one node name", envStateDir, cfg.StateDir) } if err := cfg.checkMarketPlacement(); err != nil { return Config{}, err } if len(cfg.CloudControllerManagerExec) == 0 { return Config{}, fmt.Errorf( "%s is not set. Every joined cluster's process group carries the cloud controller "+ "manager, which is what stamps a Node's providerID and the labels an autoscaler "+ "groups by, and nothing else names that binary: a join that accepted an empty one "+ "would lease, config and spawn everything else and silently drop the component "+ "that gives a joined node its cloud identity", envCloudControllerManager) } if len(cfg.FrontExec) == 0 { return Config{}, errors.New( "the cluster provider is not running from the socialweb-computer binary, so no executable " + "carries `internal apiserver-front`: every joined cluster's apiserver port is owned by that " + "front, and k3s sits behind it on a slot port") } if cfg.ExternalSecrets == "" || cfg.ESOBundle == "" { return Config{}, fmt.Errorf( "%s and %s must both be set. Every joined cluster's process group carries the External "+ "Secrets controller and its webhook beside the control plane: the first names the "+ "pinned external-secrets binary, the second the pinned upstream bundle whose CRDs, "+ "ClusterRoles and webhook registration the group installs into the tenant before "+ "the pair starts. A join that accepted either empty would spawn a pair with nothing "+ "to run or a tenant serving none of its kinds", envExternalSecrets, envESOBundle) } for env, p := range cfg.pinnedExecutables() { if p != "" && !filepath.IsAbs(p) { return Config{}, fmt.Errorf("%s is %q, which is relative; it names a pinned executable by its absolute path, never by PATH", env, p) } } } var err error if cfg.NodeClaimAgent, err = kcpclient.ParseServiceAccountRef(strings.TrimSpace(get(envNodeClaimAgent))); err != nil { return Config{}, fmt.Errorf("%s: %w", envNodeClaimAgent, err) } if cfg.MarketReleaseTimeout, err = optionalDuration(get, envMarketReleaseTimeout); err != nil { return Config{}, err } if cfg.LogLevel, err = logx.ParseLevel(get(envLogLevel)); err != nil { return Config{}, fmt.Errorf("%s: %w", envLogLevel, err) } return cfg, nil } const JoinAllClusters = "*" func (c Config) ObserveOnly() bool { return c.Join == "" } func (c Config) AllClusters() bool { return c.Join == JoinAllClusters } func (c *Config) checkMarketPlacement() error { if c.AdvertiseAddress != "" { return fmt.Errorf( "%s is %q. A node is a VM rented from the compute market, and the address it dials its "+ "cluster's apiserver on is put on the guest's OWN loopback and reverse-forwarded to "+ "this host by the market's door. That address comes from %s, not from here, so a "+ "value in this variable cannot be in force: it is refused rather than accepted and "+ "ignored, because an operator who set one and saw no complaint would believe it was "+ "carrying it", envAdvertiseAddress, c.AdvertiseAddress, envMarketAdvertiseAddress) } if err := c.checkHost(); err != nil { return err } if err := portallocator.AssertMarketAdvertiseAddress(c.MarketAdvertiseAddress); err != nil { return fmt.Errorf("%s: %w", envMarketAdvertiseAddress, err) } c.AdvertiseAddress = c.MarketAdvertiseAddress return nil } func (c Config) checkHost() error { host, err := os.Hostname() if err != nil { return fmt.Errorf( "%s is %q and this host's own name cannot be read (%w), so the comparison that "+ "keeps a gateway from placing nodes on a machine it is not cannot be made. An "+ "uncompared host is worse than an absent one: it would be accepted as a match", envHost, c.Host, err) } switch { case c.Host == "": return fmt.Errorf( "%s is not set. A gateway that places nodes has to be told which host it is, so that "+ "the answer can be checked against os.Hostname() (%q); nothing here may assume, "+ "because the assumption is accepted on the one machine where it is wrong", envHost, host) case c.Host != host: return fmt.Errorf( "%s is %q and this host calls itself %q. The mismatch is refused rather than "+ "reconciled: a gateway told it is a host it is not would place nodes on the wrong "+ "machine -- the node records and the tenant registrations would both name a "+ "cluster running somewhere else, and every later pass would read that "+ "other machine's state as its own", envHost, c.Host, host) } return nil } func splitList(raw string) []string { out := []string{} for part := range strings.SplitSeq(raw, ",") { if trimmed := strings.TrimSpace(part); trimmed != "" { out = append(out, trimmed) } } return out } func parseBool(raw string) (bool, error) { switch strings.ToLower(raw) { case "true", "1", "yes": return true, nil case "false", "0", "no": return false, nil } return false, fmt.Errorf("%q is not one of true, false", raw) } func optionalDuration(get func(string) string, name string) (time.Duration, error) { raw := strings.TrimSpace(get(name)) if raw == "" { return 0, nil } d, err := time.ParseDuration(raw) if err != nil { return 0, fmt.Errorf("%s: %q is not a duration", name, raw) } if d <= 0 { return 0, fmt.Errorf("%s: %q must be positive", name, raw) } return d, nil }