package main import ( "context" "crypto/sha256" "encoding/base32" "encoding/json" "errors" "fmt" "log/slog" "slices" "sort" "strings" "time" apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/client-go/dynamic" "k8s.io/client-go/util/retry" ) var ( projectDeliveryGVR = schema.GroupVersionResource{Group: "delivery.quorate.dev", Version: "v1alpha1", Resource: "projectdeliveries"} workflowGVR = schema.GroupVersionResource{Group: "argoproj.io", Version: "v1alpha1", Resource: "workflows"} ) type controllerConfig struct { pollInterval time.Duration bucketInterval time.Duration requestTimeout time.Duration templateQuiet time.Duration } type controller struct { client dynamic.Interface source sourceReader now func() time.Time config controllerConfig qualCursor string } func (c *controller) run(ctx context.Context) error { ticker := time.NewTicker(c.config.pollInterval) defer ticker.Stop() for { c.reconcileAll(ctx) select { case <-ctx.Done(): return ctx.Err() case <-ticker.C: } } } func (c *controller) reconcileAll(ctx context.Context) { if err := c.releaseOrphans(ctx); err != nil { slog.Error("release orphan records", "error", err) } if err := c.reconcileTemplates(ctx); err != nil { slog.Error("reconcile templates", "error", err) } if err := c.qualifyAll(ctx); err != nil { slog.Error("reconcile qualification", "error", err) } items, err := c.client.Resource(projectDeliveryGVR).Namespace(metav1.NamespaceAll).List(ctx, metav1.ListOptions{}) if err != nil { slog.Error("list ProjectDeliveries", "error", err) return } for index := range items.Items { item := &items.Items[index] if err := c.reconcileKey(ctx, item.GetNamespace(), item.GetName()); err != nil { slog.Error("reconcile ProjectDelivery", "namespace", item.GetNamespace(), "name", item.GetName(), "error", err) c.recordError(ctx, item.GetNamespace(), item.GetName(), err) } } } func (c *controller) reconcileKey(ctx context.Context, namespace, name string) error { return retry.RetryOnConflict(retry.DefaultRetry, func() error { object, err := c.client.Resource(projectDeliveryGVR).Namespace(namespace).Get(ctx, name, metav1.GetOptions{}) if apierrors.IsNotFound(err) { return nil } if err != nil { return err } return c.reconcile(ctx, object) }) } func (c *controller) reconcile(ctx context.Context, object *unstructured.Unstructured) error { delivery, err := decodeProjectDelivery(object) if err != nil { return err } if delivery.Metadata.DeletionTimestamp != nil { return nil } if err := validateTargets(delivery.Spec.Targets); err != nil { return err } recovered, err := c.recoverBuilds(ctx, &delivery) if err != nil { return err } active, err := c.resumeAttempts(ctx, &delivery) if err != nil { return err } if active || recovered { return c.updateStatus(ctx, object, &delivery) } retried, err := c.reconcileRetry(ctx, &delivery) if err != nil { return err } var reconcileErr error switch { case retried: case delivery.Spec.Source.Continuous != nil: refs, err := c.readRefs(ctx, delivery.Spec.Source.Repository) if err != nil { return fmt.Errorf("read Forgejo refs: %w", err) } reconcileErr = c.reconcileContinuous(ctx, &delivery, refs) case delivery.Spec.Source.Point != nil: reconcileErr = c.reconcilePoint(ctx, &delivery) default: reconcileErr = errors.New("delivery source has no mode") } if reconcileErr != nil { return reconcileErr } delivery.Status.ObservedGeneration = delivery.Metadata.Generation return c.updateStatus(ctx, object, &delivery) } func (c *controller) resumeAttempts(ctx context.Context, delivery *projectDelivery) (bool, error) { active := false if status := delivery.Status.Continuous; status != nil && status.Attempt != nil && !terminalPhase(status.Attempt.Phase) { if err := c.advanceAttempt(ctx, delivery, status.Attempt); err != nil { return false, err } recordDelivered(status) delivery.Status.Message = "build attempt: " + phaseOrPending(status.Attempt.Phase) active = true } for index := range delivery.Status.Points { attempt := &delivery.Status.Points[index] if terminalPhase(attempt.Phase) { continue } if err := c.advanceAttempt(ctx, delivery, attempt); err != nil { return false, err } delivery.Status.Message = "point attempt: " + phaseOrPending(attempt.Phase) active = true } return active, nil } func (c *controller) reconcileRetry(ctx context.Context, delivery *projectDelivery) (bool, error) { request := delivery.Metadata.Annotations[retryAnnotation] if request == "" || request == delivery.Status.LastHandledRetry { return false, nil } target, continuous := retryTarget(delivery, request) if target == nil { return false, fmt.Errorf("retry Workflow %q is not a recorded attempt", request) } if otherAttemptActive(delivery, request) { return false, nil } if !terminalPhase(target.Phase) { return false, nil } successor := attemptStatus{ Ref: target.Ref, Revision: target.Revision, SourceIdentity: target.SourceIdentity, RetryOf: request, Phase: "Pending", } if err := c.advanceAttempt(ctx, delivery, &successor); err != nil { return false, err } if successor.WorkflowName == "" { delivery.Status.Message = templatePending.Error() return true, nil } if continuous { rememberAttempt(delivery.Status.Continuous, &successor) target = delivery.Status.Continuous.Attempt } else { delivery.Status.Points = append(delivery.Status.Points, successor) target = &delivery.Status.Points[len(delivery.Status.Points)-1] } delivery.Status.LastHandledRetry = request delivery.Status.Message = "retry attempt: " + phaseOrPending(target.Phase) return true, nil } func retryTarget(delivery *projectDelivery, workflow string) (*attemptStatus, bool) { if delivery.Status.Continuous != nil { attempt := delivery.Status.Continuous.Attempt if attempt != nil && attempt.WorkflowName == workflow { return attempt, true } } for index := range delivery.Status.Points { attempt := &delivery.Status.Points[index] if attempt.WorkflowName == workflow { return attempt, false } } return nil, false } func otherAttemptActive(delivery *projectDelivery, excluded string) bool { if delivery.Status.Continuous != nil { attempt := delivery.Status.Continuous.Attempt if attempt != nil && attempt.WorkflowName != excluded && !terminalPhase(attempt.Phase) { return true } } for index := range delivery.Status.Points { attempt := &delivery.Status.Points[index] if attempt.WorkflowName != excluded && !terminalPhase(attempt.Phase) { return true } } return false } func (c *controller) reconcileContinuous( ctx context.Context, delivery *projectDelivery, refs map[string]string, ) error { ref := delivery.Spec.Source.Continuous.Ref revision, found := refs[ref] if !found { return fmt.Errorf("continuous ref %q is absent", ref) } if delivery.Status.Continuous == nil { delivery.Status.Continuous = &continuousStatus{} } status := delivery.Status.Continuous sourceIdentity, err := c.sourceIdentity(ctx, delivery, revision) if err != nil { return fmt.Errorf("read continuous path identity: %w", err) } filtered := len(delivery.Spec.Source.Continuous.Paths) != 0 if consumedIdentity(status, sourceIdentity) { clearCandidate(status) delivery.Status.Message = "continuous source already attempted" return nil } if deliveredMatches(status.Delivered, sourceIdentity) || status.Delivered == nil && status.Attempt != nil && status.Attempt.Phase == "Succeeded" && attemptMatches(status.Attempt, sourceIdentity, revision, filtered) { clearCandidate(status) delivery.Status.Message = "continuous source already delivered" return nil } if status.Attempt != nil && terminalPhase(status.Attempt.Phase) && attemptMatches(status.Attempt, sourceIdentity, revision, filtered) { delivery.Status.Message = "continuous source already attempted" return nil } now := c.now().UTC() bucket := now.Truncate(c.config.bucketInterval).Format(time.RFC3339) if status.CandidateIdentity != sourceIdentity { status.CandidateRevision = revision status.CandidateIdentity = sourceIdentity status.CandidateBuckets = []string{bucket} status.CandidateSince = &now if delivery.Spec.Source.Continuous.Debounce == "" { delivery.Status.Message = "observed continuous candidate 1/3" } else { delivery.Status.Message = "observed continuous candidate" } return nil } status.CandidateRevision = revision if configured := delivery.Spec.Source.Continuous.Debounce; configured != "" { debounce, err := time.ParseDuration(configured) if err != nil || debounce <= 0 { return fmt.Errorf("invalid continuous debounce %q", configured) } if status.CandidateSince == nil { status.CandidateSince = &now delivery.Status.Message = "observed continuous candidate" return nil } if now.Before(status.CandidateSince.Add(debounce)) { delivery.Status.Message = "continuous candidate debouncing" return nil } } else { if !contains(status.CandidateBuckets, bucket) && len(status.CandidateBuckets) < 3 { status.CandidateBuckets = append(status.CandidateBuckets, bucket) } if len(status.CandidateBuckets) < 3 { delivery.Status.Message = fmt.Sprintf("observed continuous candidate %d/3", len(status.CandidateBuckets)) return nil } } latest, err := c.readRefs(ctx, delivery.Spec.Source.Repository) if err != nil { return fmt.Errorf("reread continuous ref: %w", err) } latestRevision, found := latest[ref] if !found { return fmt.Errorf("continuous ref %q is absent after debounce", ref) } latestIdentity, err := c.sourceIdentity(ctx, delivery, latestRevision) if err != nil { return fmt.Errorf("reread continuous path identity: %w", err) } if consumedIdentity(status, latestIdentity) { clearCandidate(status) delivery.Status.Message = "continuous source returned to attempted state" return nil } if deliveredMatches(status.Delivered, latestIdentity) { clearCandidate(status) delivery.Status.Message = "continuous source returned to delivered state" return nil } if latestIdentity != sourceIdentity { status.CandidateRevision = latestRevision status.CandidateIdentity = latestIdentity status.CandidateBuckets = nil status.CandidateSince = nil delivery.Status.Message = "continuous source changed before Workflow creation" return nil } revision = latestRevision status.CandidateRevision = revision attempt := &attemptStatus{ Ref: ref, Revision: revision, SourceIdentity: sourceIdentity, Phase: "Pending", } if err := c.advanceAttempt(ctx, delivery, attempt); err != nil { return err } if attempt.WorkflowName == "" { delivery.Status.Message = templatePending.Error() return nil } rememberAttempt(status, attempt) recordDelivered(status) delivery.Status.Message = "build attempt: " + phaseOrPending(attempt.Phase) return nil } func (c *controller) reconcilePoint( ctx context.Context, delivery *projectDelivery, ) error { for _, revision := range delivery.Spec.Source.Point.Revisions { if pointAttempted(delivery.Status.Points, revision) { continue } attempt := attemptStatus{ Revision: revision, Phase: "Pending", } if err := c.advanceAttempt(ctx, delivery, &attempt); err != nil { return err } if attempt.WorkflowName == "" { delivery.Status.Message = templatePending.Error() return nil } delivery.Status.Points = append(delivery.Status.Points, attempt) delivery.Status.Message = "point attempt: " + phaseOrPending(attempt.Phase) return nil } delivery.Status.Message = "point revisions already attempted" return nil } func (c *controller) advanceAttempt( ctx context.Context, delivery *projectDelivery, attempt *attemptStatus, ) error { if attempt.Input == nil { targets, err := json.Marshal(delivery.Spec.Targets) if err != nil { return err } attempt.Input = map[string]string{ "source-repository": delivery.Spec.Source.Repository, "source-ref": attempt.Ref, "source-revision": attempt.Revision, "targets": string(targets), } } name := attempt.WorkflowName if name == "" { name = attemptName(delivery, attempt) if err := c.ensureWorkflow(ctx, delivery, name, delivery.Spec.WorkflowTemplateRef.Name, "build", attempt.Input, map[string]string{sourceKey: attempt.SourceIdentity, retryKey: attempt.RetryOf}); err != nil { if errors.Is(err, templatePending) { return nil } return err } } state, err := c.workflow(ctx, delivery.Metadata.Namespace, name) if apierrors.IsNotFound(err) && attempt.WorkflowName != "" { attempt.Phase = "Error" return nil } if err != nil { return err } attempt.WorkflowName = name attempt.CreatedAt = state.CreatedAt attempt.Input = state.Input attempt.Ref = state.Input["source-ref"] attempt.Revision = state.Input["source-revision"] attempt.Phase = phaseOrPending(state.Phase) if result, found := state.Outputs["delivery-result"]; found { attempt.Result = opaqueResult(result) } observeQualBuild(delivery, attempt) return nil } func (c *controller) ensureWorkflow( ctx context.Context, delivery *projectDelivery, name, templateName, purpose string, parameters map[string]string, capture map[string]string, ) error { resource := c.client.Resource(workflowGVR).Namespace(delivery.Metadata.Namespace) if _, err := resource.Get(ctx, name, metav1.GetOptions{}); err == nil { return nil } else if !apierrors.IsNotFound(err) { return err } published, err := c.readyTemplate(ctx, delivery.Metadata.Namespace, templateName) if errors.Is(err, templatePending) { if _, getErr := resource.Get(ctx, name, metav1.GetOptions{}); getErr == nil { return nil } else if !apierrors.IsNotFound(getErr) { return getErr } } if err != nil { return err } metadata := map[string]any{ "name": name, "namespace": delivery.Metadata.Namespace, "annotations": map[string]any{templateKey: templateName}, "finalizers": []any{recordKey}, "labels": map[string]any{ "app.kubernetes.io/name": "project-delivery", "app.kubernetes.io/instance": delivery.Metadata.Name, "delivery.quorate.dev/purpose": purpose, recordKey: "pending", }, "ownerReferences": []any{map[string]any{ "apiVersion": "delivery.quorate.dev/v1alpha1", "kind": "ProjectDelivery", "name": delivery.Metadata.Name, "uid": delivery.Metadata.UID, "controller": true, "blockOwnerDeletion": false, }}, } for key, value := range capture { metadata["annotations"].(map[string]any)[key] = value } names := make([]string, 0, len(parameters)) for parameter := range parameters { names = append(names, parameter) } sort.Strings(names) arguments := make([]any, 0, len(names)) for _, parameter := range names { arguments = append(arguments, map[string]any{"name": parameter, "value": parameters[parameter]}) } workflow := &unstructured.Unstructured{Object: map[string]any{ "apiVersion": "argoproj.io/v1alpha1", "kind": "Workflow", "metadata": metadata, "spec": map[string]any{ "workflowTemplateRef": map[string]any{"name": published}, "arguments": map[string]any{"parameters": arguments}, "synchronization": map[string]any{ "mutexes": []any{map[string]any{"name": templateName}}, }, }, }} _, err = resource.Create(ctx, workflow, metav1.CreateOptions{}) if apierrors.IsAlreadyExists(err) { return nil } return err } func (c *controller) workflow(ctx context.Context, namespace, name string) (workflowState, error) { object, err := c.client.Resource(workflowGVR).Namespace(namespace).Get(ctx, name, metav1.GetOptions{}) if err != nil { return workflowState{}, err } phase, _, _ := unstructured.NestedString(object.Object, "status", "phase") outputs := map[string]string{} parameters, _, _ := unstructured.NestedSlice(object.Object, "status", "outputs", "parameters") rootParameters, _, _ := unstructured.NestedSlice(object.Object, "status", "nodes", name, "outputs", "parameters") parameters = append(parameters, rootParameters...) for _, item := range parameters { parameter, ok := item.(map[string]any) if !ok { continue } name, _ := parameter["name"].(string) value, _ := parameter["value"].(string) if name != "" { outputs[name] = value } } input := map[string]string{} arguments, _, _ := unstructured.NestedSlice(object.Object, "spec", "arguments", "parameters") for _, item := range arguments { parameter := item.(map[string]any) name, _ := parameter["name"].(string) value, _ := parameter["value"].(string) input[name] = value } var startedAt *time.Time if value, _, _ := unstructured.NestedString(object.Object, "status", "startedAt"); value != "" { timestamp, err := time.Parse(time.RFC3339, value) if err != nil { return workflowState{}, err } if !timestamp.IsZero() { startedAt = ×tamp } } ownerUID := "" for _, owner := range object.GetOwnerReferences() { if owner.Kind == "ProjectDelivery" && owner.APIVersion == "delivery.quorate.dev/v1alpha1" { ownerUID = string(owner.UID) } } createdAt := object.GetCreationTimestamp() return workflowState{Phase: phase, Outputs: outputs, Input: input, StartedAt: startedAt, OwnerUID: ownerUID, CreatedAt: &createdAt.Time}, nil } func consumedIdentity(status *continuousStatus, identity string) bool { return status.Attempt != nil && status.Attempt.History != nil && slices.Contains(status.Attempt.History.Identities, identity) } func rememberAttempt(status *continuousStatus, attempt *attemptStatus) { history := &attemptHistory{} if previous := status.Attempt; previous != nil { if previous.History != nil { history.Workflows = slices.Clone(previous.History.Workflows) history.Identities = slices.Clone(previous.History.Identities) } addHistory(history, previous) } addHistory(history, attempt) attempt.History = history status.Attempt = attempt } func addHistory(history *attemptHistory, attempt *attemptStatus) { if attempt.WorkflowName != "" && !slices.Contains(history.Workflows, attempt.WorkflowName) { history.Workflows = append(history.Workflows, attempt.WorkflowName) } if attempt.SourceIdentity != "" && !slices.Contains(history.Identities, attempt.SourceIdentity) { history.Identities = append(history.Identities, attempt.SourceIdentity) } } func (c *controller) updateStatus(ctx context.Context, object *unstructured.Unstructured, delivery *projectDelivery) error { status, err := toMap(delivery.Status) if err != nil { return err } copy := object.DeepCopy() copy.Object["status"] = status _, err = c.client.Resource(projectDeliveryGVR).Namespace(copy.GetNamespace()).UpdateStatus(ctx, copy, metav1.UpdateOptions{}) if apierrors.IsNotFound(err) { return c.releaseOrphans(ctx) } if err != nil { return err } return c.releaseRecords(ctx, delivery) } func (c *controller) recordError(ctx context.Context, namespace, name string, reconcileError error) { object, err := c.client.Resource(projectDeliveryGVR).Namespace(namespace).Get(ctx, name, metav1.GetOptions{}) if err != nil { return } delivery, err := decodeProjectDelivery(object) if err != nil { return } delivery.Status.Message = reconcileError.Error() delivery.Status.ObservedGeneration = delivery.Metadata.Generation _ = c.updateStatus(ctx, object, &delivery) } func decodeProjectDelivery(object *unstructured.Unstructured) (projectDelivery, error) { content, err := json.Marshal(object.Object) if err != nil { return projectDelivery{}, err } var delivery projectDelivery if err := json.Unmarshal(content, &delivery); err != nil { return projectDelivery{}, err } return delivery, nil } func validateTargets(targets []json.RawMessage) error { seen := make(map[string]struct{}, len(targets)) for index, target := range targets { key := string(target) if _, found := seen[key]; found { return fmt.Errorf("duplicate target at index %d", index) } seen[key] = struct{}{} } return nil } func toMap(value any) (map[string]any, error) { content, err := json.Marshal(value) if err != nil { return nil, err } var result map[string]any err = json.Unmarshal(content, &result) return result, err } func workflowName(purpose string, fields ...string) string { digest := sha256.Sum256([]byte("v1\x00" + purpose + "\x00" + strings.Join(fields, "\x00"))) encoded := base32.StdEncoding.WithPadding(base32.NoPadding).EncodeToString(digest[:20]) return "pd-" + purpose + "-" + strings.ToLower(encoded) } func attemptName(delivery *projectDelivery, attempt *attemptStatus) string { sourceMode := "" identity := attempt.Revision if attempt.Ref != "" { sourceMode = "continuous" if attempt.SourceIdentity != "" { identity = attempt.SourceIdentity } } fields := []string{delivery.Metadata.UID, sourceMode, identity} if attempt.RetryOf != "" { fields = append(fields, attempt.RetryOf) } return workflowName("build", fields...) } func (c *controller) readRefs(ctx context.Context, repository string) (map[string]string, error) { requestContext, cancel := context.WithTimeout(ctx, c.config.requestTimeout) defer cancel() return c.source.Refs(requestContext, repository) } func (c *controller) sourceIdentity( ctx context.Context, delivery *projectDelivery, revision string, ) (string, error) { continuous := delivery.Spec.Source.Continuous paths := append([]string(nil), continuous.Paths...) sort.Strings(paths) if len(paths) == 0 { return revision, nil } requestContext, cancel := context.WithTimeout(ctx, c.config.requestTimeout) defer cancel() value, err := c.source.PathIdentity(requestContext, delivery.Spec.Source.Repository, revision, paths) if err != nil { return "", err } return value, nil } func recordDelivered(status *continuousStatus) { attempt := status.Attempt if attempt == nil || attempt.Phase != "Succeeded" || attempt.SourceIdentity == "" { return } status.Delivered = &deliveredSource{ Revision: attempt.Revision, Identity: attempt.SourceIdentity, } } func deliveredMatches(delivered *deliveredSource, sourceIdentity string) bool { return delivered != nil && delivered.Identity == sourceIdentity } func attemptMatches(attempt *attemptStatus, sourceIdentity, revision string, filtered bool) bool { if attempt.SourceIdentity != "" { return attempt.SourceIdentity == sourceIdentity } return !filtered && attempt.Revision == revision } func clearCandidate(status *continuousStatus) { status.CandidateRevision = "" status.CandidateIdentity = "" status.CandidateBuckets = nil status.CandidateSince = nil } func contains(values []string, expected string) bool { for _, value := range values { if value == expected { return true } } return false } func terminalPhase(phase string) bool { return phase == "Succeeded" || phase == "Failed" || phase == "Error" } func pointAttempted(attempts []attemptStatus, revision string) bool { for _, attempt := range attempts { if attempt.Revision == revision { return true } } return false } func phaseOrPending(phase string) string { if phase == "" { return "Pending" } return phase }