johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package controller

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"sync"
	"time"

	"k8s.io/apimachinery/pkg/api/meta"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

	policyv1alpha1 "github.com/johnandersen777/socialweb-computer-kcp/api/policy/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/contract/sandbox"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/opt"
	policyengine "github.com/johnandersen777/socialweb-computer-kcp/internal/policyengine/status"
)

type EngineFactory func(ctx context.Context, engine *policyv1alpha1.PolicyEngine) (*Engine, error)

type managedEngine struct {
	engine *Engine
	hash   string
	fence  *sandbox.Fence
}

type EngineManager struct {
	mu      sync.Mutex
	factory EngineFactory
	engines map[string]*managedEngine
	failed  map[string]string
}

func NewEngineManager(factory EngineFactory) *EngineManager {
	return &EngineManager{factory: factory, engines: make(map[string]*managedEngine), failed: make(map[string]string)}
}

func NewDefaultEngineFactory(base Config) EngineFactory {
	return func(ctx context.Context, engineCR *policyv1alpha1.PolicyEngine) (*Engine, error) {
		config := EngineConfigFromCR(engineCR, base)
		var spec *policyv1alpha1.PolicySandboxConfig
		if engineCR != nil {
			spec = engineCR.Spec.Sandbox
		}
		switch {
		case base.OpenSandbox != nil:
			runner, err := base.OpenSandbox(ctx, spec)
			if err != nil {
				return nil, fmt.Errorf("opening the %s policy sandbox: %w", spec.ProviderName(), err)
			}
			config.Sandbox = runner
		case spec.ProviderName() != policyv1alpha1.PolicySandboxProviderDeno:
			return nil, fmt.Errorf("the %s policy sandbox is not available on this host", spec.ProviderName())
		}
		return NewEngine(config), nil
	}
}

func probeFence(ctx context.Context, runner sandbox.Runner) *sandbox.Fence {
	prober, ok := runner.(sandbox.Prober)
	if !ok {
		return nil
	}
	if !runner.Available() {
		return &sandbox.Fence{Reason: policyv1alpha1.PolicyEngineFencedNotProbed, Message: fmt.Sprintf("the %s sandbox runtime is unavailable", runner.Kind())}
	}
	fence, err := prober.Probe(ctx)
	if err != nil {
		return &sandbox.Fence{Reason: policyv1alpha1.PolicyEngineFencedProbeFailed, Message: err.Error()}
	}
	if fence.Reason == "" {
		fence.Reason = policyv1alpha1.PolicyEngineFencedProbeFailed
		if fence.Fenced {
			fence.Reason = policyv1alpha1.PolicyEngineFencedProbePassed
		}
	}
	return &fence
}

func (m *EngineManager) ReconcilePolicyEngine(ctx context.Context, key string, engineCR *policyv1alpha1.PolicyEngine) (EngineReconcileState, error) {
	if engineCR == nil {
		return EngineReconcileState{}, fmt.Errorf("policy engine is required")
	}
	if engineCR.DeletionTimestamp != nil {
		m.stop(ctx, key)
		state := EngineReconcileState{Phase: policyv1alpha1.PolicyEnginePending}
		meta.SetStatusCondition(&state.Conditions, metav1.Condition{
			Type:    policyengine.ConditionReady,
			Status:  metav1.ConditionFalse,
			Reason:  "Terminating",
			Message: "the policy engine is being deleted",
		})
		return state, nil
	}
	if engineCR.Spec.Suspend {
		m.stop(ctx, key)
		state := EngineReconcileState{Phase: policyv1alpha1.PolicyEnginePending}
		meta.SetStatusCondition(&state.Conditions, metav1.Condition{
			Type:    policyengine.ConditionSuspended,
			Status:  metav1.ConditionTrue,
			Reason:  "Suspended",
			Message: "the policy engine is suspended",
		})
		return state, nil
	}
	spec, err := json.Marshal(engineCR.Spec)
	if err != nil {
		return EngineReconcileState{}, err
	}
	hash := string(spec)
	m.mu.Lock()
	managed := m.engines[key]
	if managed == nil || managed.hash != hash {
		m.mu.Unlock()
		m.stop(ctx, key)
		if m.factory == nil {
			return EngineReconcileState{Phase: policyv1alpha1.PolicyEngineFailed}, fmt.Errorf("policy engine factory is not configured")
		}
		engine, factoryErr := m.factory(ctx, engineCR)
		if factoryErr != nil {
			m.mu.Lock()
			m.failed[key] = factoryErr.Error()
			m.mu.Unlock()
			state := EngineReconcileState{Phase: policyv1alpha1.PolicyEngineFailed}
			meta.SetStatusCondition(&state.Conditions, metav1.Condition{
				Type:    policyengine.ConditionReady,
				Status:  metav1.ConditionFalse,
				Reason:  "CreateFailed",
				Message: factoryErr.Error(),
			})
			return state, factoryErr
		}
		if engine == nil {
			return EngineReconcileState{Phase: policyv1alpha1.PolicyEngineFailed}, fmt.Errorf("policy engine factory returned nil")
		}
		if startErr := engine.Start(context.WithoutCancel(ctx)); startErr != nil {
			return EngineReconcileState{Phase: policyv1alpha1.PolicyEngineFailed}, startErr
		}
		fence := probeFence(ctx, engine.Sandbox())
		m.mu.Lock()
		delete(m.failed, key)
		m.engines[key] = &managedEngine{engine: engine, hash: hash, fence: fence}
		managed = m.engines[key]
		stored := *managed
		m.mu.Unlock()
		return reconcileState(&stored), nil
	}
	stored := *managed
	m.mu.Unlock()
	return reconcileState(&stored), nil
}

func reconcileState(managed *managedEngine) EngineReconcileState {
	state := EngineReconcileState{ActiveRuns: managed.engine.ActiveRuns()}
	if fence := managed.fence; fence != nil {
		status := metav1.ConditionFalse
		if fence.Fenced {
			status = metav1.ConditionTrue
		}
		meta.SetStatusCondition(&state.Conditions, metav1.Condition{
			Type:    policyv1alpha1.PolicyEngineConditionFenced,
			Status:  status,
			Reason:  fence.Reason,
			Message: fence.Message,
		})
	}
	if managed.engine.Ready() {
		state.Phase = policyv1alpha1.PolicyEngineRunning
		state.Ready = true
		meta.SetStatusCondition(&state.Conditions, metav1.Condition{
			Type:    policyengine.ConditionReady,
			Status:  metav1.ConditionTrue,
			Reason:  "Ready",
			Message: "the in-process policy engine is ready",
		})
		return state
	}
	state.Phase = policyv1alpha1.PolicyEnginePending
	notReady := metav1.Condition{
		Type:    policyengine.ConditionReady,
		Status:  metav1.ConditionFalse,
		Reason:  "RuntimeUnavailable",
		Message: "the embedded or system deno runtime is unavailable",
	}
	if fence := managed.fence; fence != nil {
		notReady.Message = fmt.Sprintf("the %s sandbox runtime is unavailable", managed.engine.Sandbox().Kind())
		if !fence.Fenced {
			notReady.Reason = fence.Reason
			notReady.Message = fence.Message
		}
	}
	meta.SetStatusCondition(&state.Conditions, notReady)
	return state
}

func (m *EngineManager) EngineFor(name string) (*Engine, bool) {
	m.mu.Lock()
	defer m.mu.Unlock()
	managed, ok := m.engines[name]
	if !ok {
		return nil, false
	}
	return managed.engine, true
}

func (m *EngineManager) Only(prefix string) (*Engine, bool) {
	m.mu.Lock()
	defer m.mu.Unlock()
	var found *Engine
	for key, managed := range m.engines {
		if !strings.HasPrefix(key, prefix) {
			continue
		}
		if found != nil {
			return nil, false
		}
		found = managed.engine
	}
	return found, found != nil
}

func (m *EngineManager) Shutdown(ctx context.Context) error {
	m.mu.Lock()
	engines := make([]*Engine, 0, len(m.engines))
	for name, managed := range m.engines {
		engines = append(engines, managed.engine)
		delete(m.engines, name)
	}
	m.mu.Unlock()
	var errs []error
	for _, engine := range engines {
		if err := engine.Shutdown(ctx); err != nil {
			errs = append(errs, err)
		}
	}
	return errors.Join(errs...)
}

func (m *EngineManager) Delete(ctx context.Context, name string) error {
	m.mu.Lock()
	managed := m.engines[name]
	delete(m.engines, name)
	delete(m.failed, name)
	m.mu.Unlock()
	if managed == nil {
		return nil
	}
	shutdownCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
	defer cancel()
	return managed.engine.Shutdown(shutdownCtx)
}

func (m *EngineManager) NotReady(key string) (string, bool) {
	m.mu.Lock()
	defer m.mu.Unlock()
	if message, ok := m.failed[key]; ok {
		return "the policy engine could not be created: " + message, true
	}
	if managed, ok := m.engines[key]; ok {
		return managed.notReady()
	}
	return "", false
}

func (m *EngineManager) OnlyNotReady(prefix string) (string, bool) {
	m.mu.Lock()
	keys := map[string]struct{}{}
	for key := range m.engines {
		if strings.HasPrefix(key, prefix) {
			keys[key] = struct{}{}
		}
	}
	for key := range m.failed {
		if strings.HasPrefix(key, prefix) {
			keys[key] = struct{}{}
		}
	}
	m.mu.Unlock()
	if len(keys) != 1 {
		return "", false
	}
	for key := range keys {
		return m.NotReady(key)
	}
	return "", false
}

func (managed *managedEngine) notReady() (string, bool) {
	runner := managed.engine.Sandbox()
	switch {
	case managed.engine.Ready():
		return "", false
	case runner == nil:
		return "the policy engine has no sandbox runner", true
	case !runner.Available():
		return fmt.Sprintf("the %s policy sandbox runtime is unavailable", runner.Kind()), true
	case managed.fence != nil && !managed.fence.Fenced:
		return fmt.Sprintf("the %s policy sandbox is not fenced (%s): %s", runner.Kind(), managed.fence.Reason, managed.fence.Message), true
	default:
		return fmt.Sprintf("the %s policy sandbox is not fenced", runner.Kind()), true
	}
}

func (m *EngineManager) stop(ctx context.Context, name string) {
	m.mu.Lock()
	managed := m.engines[name]
	delete(m.engines, name)
	delete(m.failed, name)
	m.mu.Unlock()
	if managed == nil {
		return
	}
	shutdownCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
	defer cancel()
	managed.engine.Shutdown(shutdownCtx)
}

func EngineConfigFromCR(engineCR *policyv1alpha1.PolicyEngine, base Config) Config {
	config := base
	if engineCR == nil {
		return config
	}
	limits := engineCR.Spec.Limits
	if limits.MaxConcurrentRuns > 0 {
		config.Limits.MaxConcurrentRuns = int(limits.MaxConcurrentRuns)
	}
	if limits.QueueDepth > 0 {
		config.Limits.QueueDepth = int(limits.QueueDepth)
	}
	if limits.ActiveDeadlineSeconds != nil && *limits.ActiveDeadlineSeconds > 0 {
		config.Limits.ActiveDeadline = time.Duration(*limits.ActiveDeadlineSeconds) * time.Second
	}
	if limits.MaxConsoleBytes > 0 {
		config.Limits.MaxConsoleBytes = limits.MaxConsoleBytes
	}
	if limits.MaxOutputBytes > 0 {
		config.Limits.MaxOutputBytes = limits.MaxOutputBytes
	}
	if limits.MaxCacheEntries > 0 {
		config.Limits.MaxCacheEntries = int(limits.MaxCacheEntries)
	}
	if limits.MaxCacheBytes > 0 {
		config.Limits.MaxCacheBytes = limits.MaxCacheBytes
	}
	if engineCR.Spec.Trust.TTLSeconds != nil && *engineCR.Spec.Trust.TTLSeconds > 0 {
		config.TrustTTL = time.Duration(*engineCR.Spec.Trust.TTLSeconds) * time.Second
	}
	cacheConfig := engineCR.Spec.Cache
	if cacheConfig.VerdictTTLSeconds != nil && *cacheConfig.VerdictTTLSeconds > 0 {
		config.VerdictTTL = time.Duration(*cacheConfig.VerdictTTLSeconds) * time.Second
	}
	if cacheConfig.ScopeTTLSeconds != nil && *cacheConfig.ScopeTTLSeconds > 0 {
		config.ScopeTTL = time.Duration(*cacheConfig.ScopeTTLSeconds) * time.Second
	}
	if cacheConfig.RecordTTLSeconds != nil && *cacheConfig.RecordTTLSeconds > 0 {
		config.RecordTTL = time.Duration(*cacheConfig.RecordTTLSeconds) * time.Second
	}
	config.VerdictTTL = opt.Positive(config.VerdictTTL, 30*time.Second)
	config.ScopeTTL = opt.Positive(config.ScopeTTL, 5*time.Minute)
	config.NegativeScopeTTL = min(opt.Positive(config.NegativeScopeTTL, 30*time.Second), config.ScopeTTL)
	config.TrustTTL = opt.Positive(config.TrustTTL, 5*time.Minute)
	config.DisableActions = false
	return config
}

func EngineKey(cluster, name string) string {
	return cluster + "/" + name
}