package policyaction import ( "bytes" "context" "encoding/json" "errors" "fmt" "io" "strings" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" policyv1alpha1 "github.com/johnandersen777/socialweb-computer-kcp/api/policy/v1alpha1" "github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1" "github.com/johnandersen777/socialweb-computer-kcp/internal/computevm" commonv1 "github.com/johnandersen777/socialweb-computer-kcp/api/common/v1alpha1" "github.com/johnandersen777/socialweb-computer-kcp/internal/kcplibs/common/ref" "github.com/johnandersen777/socialweb-computer-kcp/internal/kcplibs/impl/kcpstore" "github.com/johnandersen777/socialweb-computer-kcp/internal/policyengine/controller" ) const ContractFieldManager = "socialweb.computer/policyaction/contract" var ( ErrNoStore = errors.New("policyaction: a kcp store is required") ErrNoSpec = errors.New("policyaction: the ComputeVM action spec is required") ) type ComputeVMSpec struct { Workspace string `json:"workspace,omitempty"` Name string `json:"name"` Owner string `json:"owner"` Contract *v1alpha1.ComputeVMContract `json:"contract,omitempty"` VM v1alpha1.ComputeVMShape `json:"vm"` Requester v1alpha1.ComputeVMRequester `json:"requester"` Vault ComputeVMVaultRef `json:"vault"` ReuseStale bool `json:"reuseStale,omitempty"` StopGraceSeconds int32 `json:"stopGraceSeconds,omitempty"` GuestLogin *v1alpha1.ComputeVMGuestLogin `json:"guestLogin,omitempty"` Accept *ComputeVMAcceptRef `json:"accept,omitempty"` } type ComputeVMVaultRef struct { Server string `json:"server"` Mount string `json:"mount"` } type ComputeVMAcceptRef struct { Uri string `json:"uri"` Cid string `json:"cid"` } type ComputeVMHandler struct { Store *kcpstore.Store Workspace string } func NewComputeVMHandler(store *kcpstore.Store, workspace string) (*ComputeVMHandler, error) { if store == nil { return nil, ErrNoStore } return &ComputeVMHandler{Store: store, Workspace: strings.TrimSpace(workspace)}, nil } func (h *ComputeVMHandler) Kind() string { return controller.TriggerActionComputeVM } func (h *ComputeVMHandler) Dispatch(ctx context.Context, spec json.RawMessage, _ controller.Outcome) error { if h == nil || h.Store == nil { return ErrNoStore } decoded, err := decodeComputeVMSpec(spec) if err != nil { return err } workspace := strings.TrimSpace(decoded.Workspace) if workspace == "" { workspace = strings.TrimSpace(h.Workspace) } if err := validateWorkspace(workspace); err != nil { return err } name := strings.TrimSpace(decoded.Name) if name == "" { return errors.New("policyaction: the ComputeVM action spec names no name") } owner := strings.TrimSpace(decoded.Owner) if owner == "" { return errors.New("policyaction: the ComputeVM action spec names no owner") } input := computevm.VMInput{ Name: name, Handle: owner, VM: decoded.VM, RequesterDID: strings.TrimSpace(decoded.Requester.Did), Vault: computevm.GuestVaultRef{Server: strings.TrimSpace(decoded.Vault.Server), Mount: strings.TrimSpace(decoded.Vault.Mount)}, ReuseStale: decoded.ReuseStale, StopGraceSeconds: decoded.StopGraceSeconds, GuestLogin: decoded.GuestLogin, Contract: decoded.Contract, } if decoded.Accept != nil { input.Accept = &computevm.AcceptRef{URI: strings.TrimSpace(decoded.Accept.Uri), CID: strings.TrimSpace(decoded.Accept.Cid)} } if err := computevm.ValidateVMInput(input); err != nil { return fmt.Errorf("policyaction: the ComputeVM action spec: %w", err) } vm := &v1alpha1.ComputeVM{ TypeMeta: metav1.TypeMeta{APIVersion: v1alpha1.GroupVersion.String(), Kind: "ComputeVM"}, ObjectMeta: metav1.ObjectMeta{Name: input.Name}, Spec: v1alpha1.ComputeVMSpec{ BackendRef: computevm.BackendRef(input.Name), VM: input.VM, Owner: v1alpha1.ComputeVMOwner{Handle: input.Handle}, UserData: v1alpha1.ComputeVMUserData{Vault: computevm.UserDataRef(input.Name, input.Vault)}, Requester: v1alpha1.ComputeVMRequester{Did: input.RequesterDID, Plc: strings.TrimSpace(decoded.Requester.Plc)}, }, } if input.Accept != nil && input.Accept.Key() != "" { vm.Spec.Accept = &commonv1.StrongRef{URI: input.Accept.URI, CID: input.Accept.CID} } if input.Contract != nil { contract := *input.Contract vm.Spec.Contract = &contract } if input.ReuseStale { vm.Spec.Image = &v1alpha1.ComputeVMImageRequest{ReuseStale: true} } if input.StopGraceSeconds > 0 { vm.Spec.StopGraceSeconds = input.StopGraceSeconds } if input.GuestLogin != nil { login := *input.GuestLogin vm.Spec.GuestLogin = &login } return h.ensure(ctx, workspace, vm) } func (h *ComputeVMHandler) DispatchComputeVM(ctx context.Context, action policyv1alpha1.PolicyAction, outcome controller.Outcome) error { return h.Dispatch(ctx, action.Spec, outcome) } func (h *ComputeVMHandler) ensure(ctx context.Context, workspace string, vm *v1alpha1.ComputeVM) error { resource := kcpstore.OfClusterScoped[v1alpha1.ComputeVM](h.Store, computevm.GVR) target := ref.New(workspace, "", vm.Name) existing, err := resource.Get(ctx, target) if err == nil { return ensureExisting(ctx, resource, target, existing, vm) } if !kcpstore.IsNotFound(err) { return fmt.Errorf("policyaction: read ComputeVM %s in %s: %w", vm.Name, workspace, err) } if err := resource.Create(ctx, workspace, vm); err != nil { return fmt.Errorf("policyaction: create ComputeVM %s in %s: %w", vm.Name, workspace, err) } return nil } func ensureExisting(ctx context.Context, resource *kcpstore.Resource[v1alpha1.ComputeVM], target ref.Ref, existing, wanted *v1alpha1.ComputeVM) error { if existing.DeletionTimestamp != nil { return fmt.Errorf("policyaction: ComputeVM %s in %s is terminating", existing.Name, target.LogicalCluster) } if existing.Spec.Owner.Handle != wanted.Spec.Owner.Handle { return fmt.Errorf("policyaction: ComputeVM %s in %s belongs to owner %q, not %q", existing.Name, target.LogicalCluster, existing.Spec.Owner.Handle, wanted.Spec.Owner.Handle) } if wanted.Spec.Contract == nil { return nil } if existing.Spec.Contract != nil { if *existing.Spec.Contract == *wanted.Spec.Contract { return nil } return fmt.Errorf("policyaction: ComputeVM %s in %s already carries a different contract", existing.Name, target.LogicalCluster) } body, err := json.Marshal(map[string]any{"apiVersion": v1alpha1.GroupVersion.String(), "kind": "ComputeVM", "metadata": map[string]any{"name": existing.Name, "uid": string(existing.UID)}, "spec": map[string]any{"contract": wanted.Spec.Contract}}) if err != nil { return fmt.Errorf("policyaction: encode the contract of ComputeVM %s: %w", existing.Name, err) } if err := resource.Apply(ctx, target, body, ContractFieldManager); err != nil { return fmt.Errorf("policyaction: set the contract of ComputeVM %s in %s: %w", existing.Name, target.LogicalCluster, err) } return nil } func decodeComputeVMSpec(raw json.RawMessage) (ComputeVMSpec, error) { if len(bytes.TrimSpace(raw)) == 0 { return ComputeVMSpec{}, ErrNoSpec } if bytes.Equal(bytes.TrimSpace(raw), []byte("null")) { return ComputeVMSpec{}, ErrNoSpec } decoder := json.NewDecoder(bytes.NewReader(raw)) decoder.DisallowUnknownFields() var spec ComputeVMSpec if err := decoder.Decode(&spec); err != nil { return ComputeVMSpec{}, fmt.Errorf("policyaction: decode the ComputeVM action spec: %w", err) } var extra any if err := decoder.Decode(&extra); err != io.EOF { if err == nil { return ComputeVMSpec{}, errors.New("policyaction: the ComputeVM action spec holds more than one JSON value") } return ComputeVMSpec{}, fmt.Errorf("policyaction: decode the ComputeVM action spec: %w", err) } return spec, nil } func validateWorkspace(workspace string) error { switch { case workspace == "": return errors.New("policyaction: the ComputeVM action names no target workspace") case workspace == "." || workspace == "..": return fmt.Errorf("policyaction: %q is not a target workspace", workspace) case strings.ContainsAny(workspace, "/\\"): return fmt.Errorf("policyaction: %q is not a target workspace", workspace) } for _, r := range workspace { if r <= ' ' || r == 127 { return fmt.Errorf("policyaction: %q is not a target workspace", workspace) } } return nil }