johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package controller

import (
	"bytes"
	"context"
	"encoding/json"
	"reflect"
	"sort"
	"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/policyengine/models"
	policyengine "github.com/johnandersen777/socialweb-computer-kcp/internal/policyengine/status"
)

const (
	DefaultStatusTimeout = 15 * time.Second

	ConditionRunning = "Running"
)

func StartingStatus(run *policyv1alpha1.PolicyWorkflowRun, now time.Time) policyv1alpha1.PolicyWorkflowRunStatus {
	restarts := run.Status.Restarts
	starting := *run.Status.DeepCopy()
	starting.Phase = policyv1alpha1.PolicyWorkflowRunRunning
	starting.Restarts = restarts
	if starting.StartTime == nil {
		started := metav1.NewTime(now)
		starting.StartTime = &started
	}
	conditions := append([]metav1.Condition(nil), starting.Conditions...)
	if run.Status.Phase == policyv1alpha1.PolicyWorkflowRunRunning {
		restarts++
		starting.Restarts = restarts
		meta.SetStatusCondition(&conditions, metav1.Condition{
			Type:    policyengine.ConditionRestarted,
			Status:  metav1.ConditionTrue,
			Reason:  "Restarted",
			Message: "the policy run is executing again",
		})
	}
	meta.SetStatusCondition(&conditions, metav1.Condition{
		Type:               ConditionRunning,
		Status:             metav1.ConditionTrue,
		Reason:             "Executing",
		Message:            "the policy run is executing",
		ObservedGeneration: run.Generation,
	})
	starting.Conditions = conditions
	return starting
}

func FinalStatus(current *policyv1alpha1.PolicyWorkflowRun, outcome Outcome, now time.Time) policyv1alpha1.PolicyWorkflowRunStatus {
	status := RunStatusFromOutcome(outcome)
	status.Restarts = max(status.Restarts, current.Status.Restarts)
	switch {
	case current.Spec.Suspend:
		suspended := RunStatusFromOutcome(cancelledOutcome(current, policyengine.ConditionSuspended, "Suspended", "the run is suspended"))
		status = preserveResult(suspended, status)
	case current.Spec.Cancel:
		cancelled := RunStatusFromOutcome(cancelledOutcome(current, policyengine.ConditionCancelled, "Cancelled", "the run was cancelled"))
		status = preserveResult(cancelled, status)
	}
	if status.RunID == "" {
		status.RunID = current.Status.RunID
	}
	if status.StartTime == nil {
		status.StartTime = current.Status.StartTime
	}
	if status.StartTime == nil {
		started := metav1.NewTime(now)
		status.StartTime = &started
	}
	if status.CompletionTime == nil {
		completed := metav1.NewTime(now)
		status.CompletionTime = &completed
	}
	if status.Phase == "" {
		status.Phase = policyv1alpha1.PolicyWorkflowRunFailed
		status.ExitStatus = policyv1alpha1.PolicyExitStatusFailure
	}
	return status
}

func CancelledStatus(run *policyv1alpha1.PolicyWorkflowRun) policyv1alpha1.PolicyWorkflowRunStatus {
	return RunStatusFromOutcome(cancelledOutcome(run, policyengine.ConditionCancelled, "Cancelled", "the run was cancelled"))
}

func SuspendedStatus(run *policyv1alpha1.PolicyWorkflowRun) policyv1alpha1.PolicyWorkflowRunStatus {
	return RunStatusFromOutcome(cancelledOutcome(run, policyengine.ConditionSuspended, "Suspended", "the run is suspended"))
}

func DeadlineStatus(run *policyv1alpha1.PolicyWorkflowRun) policyv1alpha1.PolicyWorkflowRunStatus {
	return notStartedStatus(run, "DeadlineExceeded", "the policy run exceeded its deadline")
}

func EngineNotReadyStatus(run *policyv1alpha1.PolicyWorkflowRun, message string) policyv1alpha1.PolicyWorkflowRunStatus {
	return notStartedStatus(run, "EngineNotReady", message)
}

func notStartedStatus(run *policyv1alpha1.PolicyWorkflowRun, reason, message string) policyv1alpha1.PolicyWorkflowRunStatus {
	outcome := failedOutcome(nil)
	outcome.Status.Annotations["error"] = []models.Annotation{{AnnotationLevel: models.AnnotationError, Message: message, Title: message}}
	outcome.Status.Conditions = nil
	meta.SetStatusCondition(&outcome.Status.Conditions, metav1.Condition{
		Type:    policyengine.ConditionFailed,
		Status:  metav1.ConditionTrue,
		Reason:  reason,
		Message: message,
	})
	status := RunStatusFromOutcome(outcome)
	status.RunID = run.Status.RunID
	status.Restarts = run.Status.Restarts
	status.StartTime = run.Status.StartTime
	return status
}

func PendingStatus(run *policyv1alpha1.PolicyWorkflowRun, reason, message string) policyv1alpha1.PolicyWorkflowRunStatus {
	value := *run.Status.DeepCopy()
	value.Phase = policyv1alpha1.PolicyWorkflowRunPending
	conditions := append([]metav1.Condition(nil), value.Conditions...)
	meta.SetStatusCondition(&conditions, metav1.Condition{
		Type:               policyengine.ConditionReady,
		Status:             metav1.ConditionFalse,
		Reason:             reason,
		Message:            message,
		ObservedGeneration: run.Generation,
	})
	value.Conditions = conditions
	return value
}

func preserveResult(status, result policyv1alpha1.PolicyWorkflowRunStatus) policyv1alpha1.PolicyWorkflowRunStatus {
	if result.RunID != "" {
		status.RunID = result.RunID
	}
	if result.Outputs != nil {
		status.Outputs = result.Outputs
	}
	if result.Cache != nil {
		status.Cache = result.Cache
	}
	if result.ConsoleOutput != "" {
		status.ConsoleOutput = result.ConsoleOutput
	}
	if result.Annotations != nil {
		status.Annotations = result.Annotations
	}
	return status
}

func RunDeadline(run *policyv1alpha1.PolicyWorkflowRun) (time.Time, bool) {
	if run.Spec.ActiveDeadlineSeconds == nil || *run.Spec.ActiveDeadlineSeconds <= 0 {
		return time.Time{}, false
	}
	start := run.CreationTimestamp.Time
	if run.Status.StartTime != nil {
		start = run.Status.StartTime.Time
	}
	return start.Add(time.Duration(*run.Spec.ActiveDeadlineSeconds) * time.Second), true
}

func RunTTL(run *policyv1alpha1.PolicyWorkflowRun, workflow *policyv1alpha1.PolicyWorkflow, fallback time.Duration) (time.Duration, bool) {
	if run.Spec.TTLSecondsAfterFinished != nil {
		if *run.Spec.TTLSecondsAfterFinished < 0 {
			return 0, false
		}
		return time.Duration(*run.Spec.TTLSecondsAfterFinished) * time.Second, true
	}
	if workflow != nil && workflow.Spec.TTLSecondsAfterFinished != nil {
		if *workflow.Spec.TTLSecondsAfterFinished < 0 {
			return 0, false
		}
		return time.Duration(*workflow.Spec.TTLSecondsAfterFinished) * time.Second, true
	}
	if fallback > 0 {
		return fallback, true
	}
	return 0, false
}

func RunExpiry(run *policyv1alpha1.PolicyWorkflowRun, ttl time.Duration) time.Time {
	completed := run.Status.CompletionTime
	if completed == nil {
		completed = run.Status.StartTime
	}
	if completed == nil {
		created := run.CreationTimestamp
		completed = &created
	}
	return completed.Add(ttl)
}

func Finished(run *policyv1alpha1.PolicyWorkflowRun) bool {
	return run.Status.Phase == policyv1alpha1.PolicyWorkflowRunSucceeded || run.Status.Phase == policyv1alpha1.PolicyWorkflowRunFailed
}

func (r *TriggerReconciler) DispatchRuns(ctx context.Context, trigger *policyv1alpha1.PolicyTrigger, runs []*policyv1alpha1.PolicyWorkflowRun,
	write func(TriggerReconcileState) error) error {
	finished := make([]*policyv1alpha1.PolicyWorkflowRun, 0, len(runs))
	for _, run := range runs {
		if Finished(run) {
			finished = append(finished, run)
		}
	}
	sort.SliceStable(finished, func(i, j int) bool {
		left := RunCompletion(finished[i])
		right := RunCompletion(finished[j])
		if left.Equal(right) {
			return finished[i].Name < finished[j].Name
		}
		return left.Before(right)
	})
	var lastErr error
	for _, run := range finished {
		completion := metav1.NewTime(RunCompletion(run))
		if !TriggerMatchesRun(trigger, run) {
			continue
		}
		if trigger.Status.LastRun != nil && !completion.After(trigger.Status.LastRun.Time) {
			continue
		}
		state, err := r.Handle(ctx, trigger, OutcomeFromRun(run))
		if err != nil {
			lastErr = err
			continue
		}
		state.LastRun = &completion
		state.NextRun = trigger.Status.NextRun
		conditions := append([]metav1.Condition(nil), trigger.Status.Conditions...)
		for _, condition := range state.Conditions {
			meta.SetStatusCondition(&conditions, condition)
		}
		state.Conditions = conditions
		if err := write(state); err != nil {
			return err
		}
		trigger.Status.LastRun = state.LastRun
		trigger.Status.NextRun = state.NextRun
		trigger.Status.Active = state.Active
		trigger.Status.Conditions = state.Conditions
	}
	return lastErr
}

func TriggerMatchesRun(trigger *policyv1alpha1.PolicyTrigger, run *policyv1alpha1.PolicyWorkflowRun) bool {
	if trigger.DeletionTimestamp != nil || trigger.Spec.Suspend {
		return false
	}
	if !trigger.CreationTimestamp.IsZero() && RunCompletion(run).Before(trigger.CreationTimestamp.Time) {
		return false
	}
	if trigger.Spec.EngineRef != "" && run.Spec.EngineRef != "" && trigger.Spec.EngineRef != run.Spec.EngineRef {
		return false
	}
	if trigger.Spec.WorkflowRef != "" {
		if run.Spec.WorkflowRef != trigger.Spec.WorkflowRef {
			return false
		}
	} else if len(bytes.TrimSpace(trigger.Spec.Workflow)) > 0 {
		if len(bytes.TrimSpace(run.Spec.Workflow)) == 0 {
			return false
		}
		left, leftErr := normalizeWorkflow(trigger.Spec.Workflow)
		right, rightErr := normalizeWorkflow(run.Spec.Workflow)
		if leftErr != nil || rightErr != nil || !bytes.Equal(left, right) {
			return false
		}
	}
	return MatchesOutputs(outputsFromAPI(run.Status.Outputs), trigger.Spec.Match)
}

func OutcomeFromRun(run *policyv1alpha1.PolicyWorkflowRun) Outcome {
	status := policyengine.Status{
		Phase:         policyengine.Phase(run.Status.Phase),
		RunID:         run.Status.RunID,
		ExitStatus:    models.ExitStatus(run.Status.ExitStatus),
		Outputs:       outputsFromAPI(run.Status.Outputs),
		Cache:         CacheFromAPI(run.Status.Cache),
		ConsoleOutput: run.Status.ConsoleOutput,
		Annotations:   annotationsFromAPI(run.Status.Annotations),
		Restarts:      run.Status.Restarts,
		StartedAt:     run.Status.StartTime,
		CompletedAt:   run.Status.CompletionTime,
		Conditions:    append([]metav1.Condition(nil), run.Status.Conditions...),
	}
	return Outcome{
		Result: models.Result{
			RunID:         status.RunID,
			Status:        models.ExecutionComplete,
			ExitStatus:    status.ExitStatus,
			Outputs:       status.Outputs,
			Cache:         status.Cache,
			Annotations:   status.Annotations,
			ConsoleOutput: status.ConsoleOutput,
		},
		Status: status,
	}
}

func outputsFromAPI(values map[string]json.RawMessage) models.Outputs {
	out := models.Outputs{}
	for key, raw := range values {
		var value any
		if err := json.Unmarshal(raw, &value); err != nil {
			continue
		}
		out[key] = value
	}
	return out
}

func annotationsFromAPI(values map[string]json.RawMessage) models.Annotations {
	out := models.Annotations{}
	for key, raw := range values {
		var value any
		if err := json.Unmarshal(raw, &value); err != nil {
			continue
		}
		if object, ok := value.(map[string]any); ok && len(object) == 1 {
			if items, exists := object["items"]; exists {
				out[key] = items
				continue
			}
			if single, exists := object["value"]; exists {
				out[key] = single
				continue
			}
		}
		out[key] = value
	}
	return out
}

func RunCompletion(run *policyv1alpha1.PolicyWorkflowRun) time.Time {
	if run.Status.CompletionTime != nil {
		return run.Status.CompletionTime.Time
	}
	if run.Status.StartTime != nil {
		return run.Status.StartTime.Time
	}
	return run.CreationTimestamp.Time
}

func EngineStatusEqual(current policyv1alpha1.PolicyEngineStatus, state EngineReconcileState) bool {
	return current.Phase == state.Phase && current.Ready == state.Ready && current.ActiveRuns == state.ActiveRuns && conditionsMatch(current.Conditions, state.Conditions)
}

func WorkflowStatusEqual(current policyv1alpha1.PolicyWorkflowStatus, state WorkflowReconcileState, generation int64) bool {
	return current.Phase == state.Phase && current.ObservedGeneration == generation && conditionsMatch(current.Conditions, state.Conditions)
}

func TriggerStatusEqual(current policyv1alpha1.PolicyTriggerStatus, state TriggerReconcileState) bool {
	return timePtrEqual(current.LastRun, state.LastRun) && timePtrEqual(current.NextRun, state.NextRun) && current.Active == state.Active && conditionsMatch(current.Conditions, state.Conditions)
}

func RunStatusEqual(current, next policyv1alpha1.PolicyWorkflowRunStatus) bool {
	if current.Phase != next.Phase || current.RunID != next.RunID || current.ExitStatus != next.ExitStatus || current.Restarts != next.Restarts {
		return false
	}
	if current.ConsoleOutput != next.ConsoleOutput {
		return false
	}
	if !timePtrEqual(current.StartTime, next.StartTime) || !timePtrEqual(current.CompletionTime, next.CompletionTime) {
		return false
	}
	if !reflect.DeepEqual(current.Outputs, next.Outputs) || !reflect.DeepEqual(current.Cache, next.Cache) || !reflect.DeepEqual(current.Annotations, next.Annotations) {
		return false
	}
	return conditionsMatch(current.Conditions, next.Conditions)
}

func conditionsMatch(left, right []metav1.Condition) bool {
	if len(left) != len(right) {
		return false
	}
	normalize := func(values []metav1.Condition) []metav1.Condition {
		out := make([]metav1.Condition, len(values))
		copy(out, values)
		for index := range out {
			out[index].LastTransitionTime = metav1.Time{}
		}
		sort.SliceStable(out, func(i, j int) bool {
			return out[i].Type < out[j].Type
		})
		return out
	}
	return reflect.DeepEqual(normalize(left), normalize(right))
}

func timePtrEqual(left, right *metav1.Time) bool {
	if left == nil || right == nil {
		return left == nil && right == nil
	}
	return left.Time.Equal(right.Time)
}