package nodeclaim import ( "cmp" "context" "crypto/sha256" "encoding/hex" "encoding/json" "errors" "fmt" "log/slog" "os" "path" "reflect" "regexp" "strings" "sync" commonv1 "github.com/johnandersen777/socialweb-computer-kcp/api/common/v1alpha1" "github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1" contract "github.com/johnandersen777/socialweb-computer-kcp/internal/contract/nodeclaim" "github.com/johnandersen777/socialweb-computer-kcp/internal/kit/opt" "github.com/johnandersen777/socialweb-computer-kcp/internal/nodemarket" ) const ( InfraNamePrefix = "nodes-" EntryNamePrefix = "node-" NodeSecretsDir = "/etc/socialweb/node" OpenBaoTicketPath = NodeSecretsDir + "/openbao-ticket" OpenBaoCAPath = NodeSecretsDir + "/openbao-ca.pem" NodeLinkEnvPath = NodeSecretsDir + "/apiserver.env" OpenBaoCARelativeToClusterDir = "openbao/tls/ca.crt" DefaultVMReadyTimeoutSec = 120 bundleKey = "bundle" ) var ( Arches = []string{"x86_64", "aarch64"} ipv4AddressPattern = regexp.MustCompile(`^([0-9]{1,3}\.){3}[0-9]{1,3}$`) portNumberPattern = regexp.MustCompile(`^[0-9]{1,5}$`) dumbpipeTicketPattern = regexp.MustCompile(`^[a-z0-9]+$`) ) var _ contract.Mapper = (*Mapper)(nil) type SecretEntry struct { Path string `json:"path"` Value string `json:"value"` } type Policy struct { Name string Args map[string]any } type Pin struct { DownloadBase string Sha256 map[string]string } type Config struct { VaultMount string VMReadyTimeoutSec int Policy *Policy UserDataBase string Meshd *Pin NodeAgent *Pin } type Deps struct { ReadFile func(string) (string, error) CAFileOf func(v1alpha1.NodeClaimSpec) string MeshdBase func(context.Context) (string, bool) NodeAgentBase func(context.Context) (string, bool) Log *slog.Logger } type Mapper struct { cfg Config deps Deps mu sync.Mutex baseWaiting bool } func New(cfg Config, deps Deps) (*Mapper, error) { cfg.VaultMount = strings.Trim(cfg.VaultMount, "/") if cfg.VaultMount == "" { return nil, errors.New("nodeclaim: the node market needs a vault mount") } cfg.VMReadyTimeoutSec = opt.Positive(cfg.VMReadyTimeoutSec, DefaultVMReadyTimeoutSec) if err := checkPin("meshd", cfg.Meshd, deps.MeshdBase != nil); err != nil { return nil, err } if err := checkPin("node agent", cfg.NodeAgent, deps.NodeAgentBase != nil); err != nil { return nil, err } if deps.CAFileOf == nil { deps.CAFileOf = DefaultOpenBaoCAFile } if deps.ReadFile == nil { deps.ReadFile = func(p string) (string, error) { raw, err := os.ReadFile(p) return string(raw), err } } if deps.Log == nil { deps.Log = slog.New(slog.DiscardHandler) } if cfg.NodeAgent == nil { deps.Log.Info("node_agent_not_offered", "reason", "the downloads carry no sha256-pinned socialweb-node build") } return &Mapper{cfg: cfg, deps: deps}, nil } func checkPin(what string, pin *Pin, resolved bool) error { if pin == nil { return nil } if pin.DownloadBase == "" && !resolved { return fmt.Errorf("nodeclaim: the %s pin has no download base", what) } for _, arch := range Arches { if pin.Sha256[arch] == "" { return fmt.Errorf("nodeclaim: the %s pin names no %s build", what, arch) } } return nil } func (m *Mapper) InfraName(claim contract.Claim) string { return InfraNamePrefix + digest(claim.Workspace+"/"+claim.Object.Spec.ClusterRef.Name, 20) } func (m *Mapper) EntryName(claim contract.Claim) string { return EntryNamePrefix + digest(cmp.Or(string(claim.Object.UID), claim.Key()), 12) } func digest(basis string, n int) string { sum := sha256.Sum256([]byte(basis)) return hex.EncodeToString(sum[:])[:n] } func (m *Mapper) BundleRef(name string) commonv1.SecretRef { return commonv1.SecretRef{Mount: m.cfg.VaultMount, Path: "contracts/" + name + "/bundle"} } func (m *Mapper) Wants(claim contract.Claim) (bool, bool) { phase := claim.Object.Status.Phase if phase == v1alpha1.NodeClaimAdopting || ((phase == "" || phase == v1alpha1.NodeClaimPending) && claim.Object.Spec.AdoptJoinedNode) { return true, true } if phase == "" || phase == v1alpha1.NodeClaimPending || phase == v1alpha1.NodeClaimBidding || phase == v1alpha1.NodeClaimProvisioning { return true, false } return false, false } func (m *Mapper) State(claim contract.Claim) v1alpha1.InfraContractState { phase := claim.Object.Status.Phase switch { case !claim.Object.DeletionTimestamp.IsZero(), phase == v1alpha1.NodeClaimFailed, phase == v1alpha1.NodeClaimLost, phase == v1alpha1.NodeClaimReleased: return v1alpha1.InfraContractReleased case phase == "", phase == v1alpha1.NodeClaimPending: return v1alpha1.InfraContractStaged } return v1alpha1.InfraContractRequested } func (m *Mapper) Bases(ctx context.Context, adopt bool) (contract.Bases, bool) { if adopt { return contract.Bases{}, true } mesh, ok := m.base(ctx, m.cfg.Meshd, m.deps.MeshdBase) if !ok { return contract.Bases{}, false } agent, ok := m.base(ctx, m.cfg.NodeAgent, m.deps.NodeAgentBase) if !ok { return contract.Bases{}, false } return contract.Bases{Meshd: mesh, NodeAgent: agent}, true } func (m *Mapper) base(ctx context.Context, pin *Pin, resolve func(context.Context) (string, bool)) (string, bool) { if pin == nil { return "", true } if resolve == nil { return pin.DownloadBase, true } resolved, ok := resolve(ctx) m.mu.Lock() defer m.mu.Unlock() if !ok { if !m.baseWaiting { m.baseWaiting = true m.deps.Log.Info("node_claim_download_base_waiting") } return "", false } if m.baseWaiting { m.baseWaiting = false m.deps.Log.Info("node_claim_download_base_resolved", "base", resolved) } return resolved, true } func (m *Mapper) LinkNames(claim contract.Claim) (string, string) { cluster := claim.Object.Spec.ClusterRef.Name return nodemarket.OpenBaoLinkName(cluster), nodemarket.ApiserverLinkName(cluster) } func (m *Mapper) Bundle(claim contract.Claim, links contract.Links) (map[string]any, error) { spec := claim.Object.Spec link := links.OpenBao if link.Err != nil { return nil, link.Err } if link.Status == nil || link.Status.Ticket == "" { message := fmt.Sprintf("the cluster's OpenBao link %s carries no ticket yet", link.Name) if link.Status != nil && link.Status.Message != "" { message += " (" + link.Status.Message + ")" } return nil, errors.New(message) } openBaoCA, err := m.deps.ReadFile(m.deps.CAFileOf(spec)) if err != nil { return nil, err } bundle := []SecretEntry{ {Path: OpenBaoTicketPath, Value: link.Status.Ticket}, {Path: OpenBaoCAPath, Value: openBaoCA}, } api := links.Apiserver linkEnv := "" if api.Err == nil && api.Status != nil && api.Status.Phase == v1alpha1.ClusterLinkListening && api.Status.Ticket != "" { linkEnv = nodeLinkEnv(spec.NodeName, spec.AdvertiseAddress, spec.ApiserverPort, api.Status.Ticket) } if linkEnv != "" { bundle = append(bundle, SecretEntry{Path: NodeLinkEnvPath, Value: linkEnv}) m.deps.Log.Info("node_claim_apiserver_link_delivered", "claim", claim.Key(), "link", api.Name) } else { phase := "" ticketBytes := 0 if api.Status != nil { phase = string(api.Status.Phase) ticketBytes = len(api.Status.Ticket) } var why any if api.Err != nil { why = api.Err } m.deps.Log.Info("node_claim_apiserver_link_withheld", "claim", claim.Key(), "link", api.Name, "phase", phase, "ticketBytes", ticketBytes, "error", why) } raw, err := json.Marshal(bundle) if err != nil { return nil, err } return map[string]any{bundleKey: string(raw)}, nil } func (m *Mapper) Entry(claim contract.Claim, adopt bool, bases contract.Bases) (v1alpha1.InfraContract, v1alpha1.InfraContractNode) { spec := claim.Object.Spec entry := v1alpha1.InfraContract{ComputeContractTemplate: v1alpha1.ComputeContractTemplate{VMName: spec.NodeName, VMReadyTimeoutSec: m.cfg.VMReadyTimeoutSec}} node := v1alpha1.InfraContractNode{OwnerDid: spec.OwnerDid, NodeClaimRef: claim.Ref()} if len(spec.DenyBidderDids) > 0 { entry.DenyBidderDids = append([]string(nil), spec.DenyBidderDids...) } if m.cfg.UserDataBase != "" { entry.UserData = m.cfg.UserDataBase } entry.Policy = v1alpha1.RequesterDefaultPolicy if m.cfg.Policy != nil { entry.Policy = cmp.Or(strings.TrimSpace(m.cfg.Policy.Name), v1alpha1.RequesterDefaultPolicy) args := m.cfg.Policy.Args if args == nil { args = map[string]any{} } if raw, err := json.Marshal(args); err == nil { entry.PolicyArgs = string(raw) } } if adopt { node.Adopt = true return entry, node } join := &v1alpha1.ComputeContractNodeJoin{ClusterRef: spec.ClusterRef} if m.cfg.Meshd != nil { mesh := &v1alpha1.ComputeContractMeshd{DownloadBase: bases.Meshd} for _, arch := range Arches { mesh.Pins = append(mesh.Pins, v1alpha1.ComputeContractMeshdPin{Arch: arch, Sha256: m.cfg.Meshd.Sha256[arch]}) } join.Meshd = mesh } if m.cfg.NodeAgent != nil { agent := &v1alpha1.ComputeContractNodeAgent{DownloadBase: bases.NodeAgent} for _, arch := range Arches { agent.Pins = append(agent.Pins, v1alpha1.ComputeContractMeshdPin{Arch: arch, Sha256: m.cfg.NodeAgent.Sha256[arch]}) } join.NodeAgent = agent } node.NodeJoin = join return entry, node } func (m *Mapper) Requested(status v1alpha1.NodeClaimStatus, adopt bool) v1alpha1.NodeClaimStatus { next := statusCopy(status) if adopt { next.Phase = v1alpha1.NodeClaimAdopting next.Message = "a joined Node with no claim: matching it to its receipt and vm.hostKeys record" } else { next.Phase = v1alpha1.NodeClaimBidding next.Message = "requesting a node from the compute market" } return next } func (m *Mapper) Failed(status v1alpha1.NodeClaimStatus, cause error) v1alpha1.NodeClaimStatus { next := statusCopy(status) next.Phase = v1alpha1.NodeClaimFailed next.Message = "the contract failed: " + cause.Error() return next } func (m *Mapper) Status(claim contract.Claim, cc v1alpha1.ComputeContract) *v1alpha1.NodeClaimStatus { current := claim.Object.Status phase := current.Phase if phase == v1alpha1.NodeClaimFailed || phase == v1alpha1.NodeClaimReleased || phase == v1alpha1.NodeClaimLost { return nil } status := cc.Status if status.Phase == "" { return nil } state := status.Contract adopt := cc.Spec.Adopt next := statusCopy(current) if state != nil { var copyOf v1alpha1.NodeClaimContract state.DeepCopyInto(©Of) if copyOf.Receipt != nil { copyOf.ReceiptURI = cmp.Or(copyOf.ReceiptURI, copyOf.Receipt.URI) copyOf.ReceiptCid = cmp.Or(copyOf.ReceiptCid, copyOf.Receipt.CID) } next.Contract = copyOf if state.VMAddress != "" { next.Fqdn = state.VMAddress } } if status.Grant != nil { grant := *status.Grant next.Grant = &grant } inFlight := phase == "" || phase == v1alpha1.NodeClaimPending || phase == v1alpha1.NodeClaimBidding || phase == v1alpha1.NodeClaimProvisioning || phase == v1alpha1.NodeClaimAdopting switch status.Phase { case v1alpha1.ComputeContractPending: case v1alpha1.ComputeContractRequesting: if inFlight && !adopt { if state != nil && state.Accept != nil { next.Phase = v1alpha1.NodeClaimProvisioning next.Message = "a bid was accepted; the VM is booting" } else { next.Phase = v1alpha1.NodeClaimBidding next.Message = "requesting a node from the compute market" } } case v1alpha1.ComputeContractHeld: switch { case adopt && (phase == v1alpha1.NodeClaimAdopting || inFlight): next.Phase = v1alpha1.NodeClaimRunning next.Message = cmp.Or(status.Message, "adopted") case status.SecretsFetchedAt != "" && inFlight: next.Phase = v1alpha1.NodeClaimJoining next.Message = "the node fetched its join secrets and is joining its cluster" case status.SecretsFetchedAt == "" && inFlight: next.Phase = v1alpha1.NodeClaimProvisioning next.Message = "a bid was accepted; the VM is booting" } case v1alpha1.ComputeContractReleasing: case v1alpha1.ComputeContractReleased: if state != nil && state.Termination != nil { next.Phase = v1alpha1.NodeClaimLost next.Message = LostMessage(state) } else { next.Phase = v1alpha1.NodeClaimReleased next.Message = cmp.Or(status.Message, "the contract was released") } case v1alpha1.ComputeContractFailed: next.Phase = v1alpha1.NodeClaimFailed next.Message = cmp.Or(status.Message, "the contract failed") } if !adopt && next.Phase == v1alpha1.NodeClaimJoining && next.JoinedAt != "" { next.Phase = v1alpha1.NodeClaimRunning next.Message = "the node joined its cluster and the cluster reports it" } if reflect.DeepEqual(current, next) { return nil } return &next } func LostMessage(state *v1alpha1.NodeClaimContract) string { t := state.Termination if t == nil { return "the node's guest is gone" } exit := "" if t.ExitCode != nil { exit = fmt.Sprintf(", exit code %d", *t.ExitCode) } return fmt.Sprintf("the bidder reported the guest terminated (%s%s) at %s; the node is "+ "released without a vm.delete and replaced through a new claim and RFP", t.Reason, exit, t.ObservedAt) } func (m *Mapper) Settle(claim contract.Claim, cc *v1alpha1.ComputeContract) contract.Settlement { finished := cc == nil || cc.Status.Phase == v1alpha1.ComputeContractReleased || cc.Status.Phase == v1alpha1.ComputeContractFailed phase := claim.Object.Status.Phase abandoned := phase == v1alpha1.NodeClaimFailed || phase == v1alpha1.NodeClaimLost if !finished && abandoned && cc != nil && cc.Status.Phase == v1alpha1.ComputeContractReleasing { next := statusCopy(claim.Object.Status) next.Phase = v1alpha1.NodeClaimFailed next.Message = "the claim was deleted; the release of its VM is still retrying on ComputeContract " + cc.Name return contract.Settlement{Status: &next, Done: true} } if !finished { return contract.Settlement{} } var state *v1alpha1.NodeClaimContract if cc != nil { state = cc.Status.Contract } confirmed := cc == nil || cc.Status.Phase == v1alpha1.ComputeContractReleased || state == nil || state.Receipt == nil next := statusCopy(claim.Object.Status) if confirmed { next.Phase = v1alpha1.NodeClaimReleased next.Message = "the claim was deleted and the node's VM released" } else { next.Phase = v1alpha1.NodeClaimFailed next.Message = "the claim was deleted but the node's VM could not be confirmed released" } if state != nil { var copyOf v1alpha1.NodeClaimContract state.DeepCopyInto(©Of) next.Contract = copyOf } return contract.Settlement{Status: &next, RemoveBundle: true, Done: true} } func statusCopy(status v1alpha1.NodeClaimStatus) v1alpha1.NodeClaimStatus { var out v1alpha1.NodeClaimStatus status.DeepCopyInto(&out) return out } func DefaultOpenBaoCAFile(spec v1alpha1.NodeClaimSpec) string { return path.Join(path.Dir(spec.TokenFile), OpenBaoCARelativeToClusterDir) } func PublishedOpenBaoCAFile(caDir string) func(v1alpha1.NodeClaimSpec) string { caDir = strings.TrimRight(caDir, "/") return func(spec v1alpha1.NodeClaimSpec) string { return caDir + "/" + path.Base(path.Dir(spec.TokenFile)) + ".crt" } } func nodeLinkEnv(node, advertiseAddress string, apiserverPort int, ticket string) string { if !dumbpipeTicketPattern.MatchString(ticket) { return "" } if !ipv4AddressPattern.MatchString(advertiseAddress) { return "" } if !portNumberPattern.MatchString(fmt.Sprintf("%d", apiserverPort)) { return "" } return fmt.Sprintf("NODE_NAME=%s\nADVERTISE_ADDRESS=%s\nAPISERVER_PORT=%d\nAPISERVER_TICKET=%s\n", node, advertiseAddress, apiserverPort, ticket) }