335 lines
10 KiB
Go
335 lines
10 KiB
Go
|
package main
|
||
|
|
|
||
|
|
import (
|
||
|
|
"context"
|
||
|
|
"crypto/sha256"
|
||
|
|
"encoding/base32"
|
||
|
|
"encoding/json"
|
||
|
|
"errors"
|
||
|
|
"fmt"
|
||
|
|
"reflect"
|
||
|
|
"slices"
|
||
|
|
"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"
|
||
|
|
"sigs.k8s.io/yaml"
|
||
|
|
)
|
||
|
|
|
||
|
|
var (
|
||
|
|
templateGVR = schema.GroupVersionResource{Group: "argoproj.io", Version: "v1alpha1", Resource: "workflowtemplates"}
|
||
|
|
configMapGVR = schema.GroupVersionResource{Version: "v1", Resource: "configmaps"}
|
||
|
|
templatePending = errors.New("workflow template update is pending")
|
||
|
|
)
|
||
|
|
|
||
|
|
const (
|
||
|
|
templateKey = "delivery.quorate.dev/template"
|
||
|
|
candidateKey = "delivery.quorate.dev/candidate"
|
||
|
|
observedKey = "delivery.quorate.dev/observed"
|
||
|
|
publishedKey = "delivery.quorate.dev/published"
|
||
|
|
)
|
||
|
|
|
||
|
|
type templateBundle struct {
|
||
|
|
name string
|
||
|
|
spec map[string]any
|
||
|
|
configs []*unstructured.Unstructured
|
||
|
|
}
|
||
|
|
|
||
|
|
type podPatch struct {
|
||
|
|
parent map[string]any
|
||
|
|
body any
|
||
|
|
}
|
||
|
|
|
||
|
|
func (c *controller) reconcileTemplates(ctx context.Context) error {
|
||
|
|
items, err := c.client.Resource(projectDeliveryGVR).Namespace(metav1.NamespaceAll).List(ctx, metav1.ListOptions{})
|
||
|
|
if err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
seen := make(map[string]bool)
|
||
|
|
var failures []error
|
||
|
|
for _, item := range items.Items {
|
||
|
|
delivery, err := decodeProjectDelivery(&item)
|
||
|
|
if err != nil {
|
||
|
|
failures = append(failures, err)
|
||
|
|
continue
|
||
|
|
}
|
||
|
|
names := []string{delivery.Spec.WorkflowTemplateRef.Name}
|
||
|
|
if delivery.Spec.Qualification != nil {
|
||
|
|
names = append(names, delivery.Spec.Qualification.WorkflowTemplateRef.Name)
|
||
|
|
}
|
||
|
|
for _, name := range names {
|
||
|
|
key := item.GetNamespace() + "/" + name
|
||
|
|
if seen[key] {
|
||
|
|
continue
|
||
|
|
}
|
||
|
|
seen[key] = true
|
||
|
|
if _, err := c.readyTemplate(ctx, item.GetNamespace(), name); err != nil && !errors.Is(err, templatePending) {
|
||
|
|
failures = append(failures, fmt.Errorf("template %s: %w", key, err))
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
templates, err := c.client.Resource(templateGVR).Namespace(metav1.NamespaceAll).List(ctx, metav1.ListOptions{})
|
||
|
|
if err != nil {
|
||
|
|
return errors.Join(append(failures, err)...)
|
||
|
|
}
|
||
|
|
for _, template := range templates.Items {
|
||
|
|
annotations := template.GetAnnotations()
|
||
|
|
if annotations[candidateKey] == "" || annotations[candidateKey] == annotations[publishedKey] {
|
||
|
|
continue
|
||
|
|
}
|
||
|
|
key := template.GetNamespace() + "/" + template.GetName()
|
||
|
|
if seen[key] {
|
||
|
|
continue
|
||
|
|
}
|
||
|
|
if _, err := c.readyTemplate(ctx, template.GetNamespace(), template.GetName()); err != nil && !errors.Is(err, templatePending) {
|
||
|
|
failures = append(failures, fmt.Errorf("template %s: %w", key, err))
|
||
|
|
}
|
||
|
|
}
|
||
|
|
return errors.Join(failures...)
|
||
|
|
}
|
||
|
|
|
||
|
|
func (c *controller) readyTemplate(ctx context.Context, namespace, name string) (string, error) {
|
||
|
|
resource := c.client.Resource(templateGVR).Namespace(namespace)
|
||
|
|
template, err := resource.Get(ctx, name, metav1.GetOptions{})
|
||
|
|
if err != nil {
|
||
|
|
return "", err
|
||
|
|
}
|
||
|
|
bundle, err := c.captureBundle(ctx, template)
|
||
|
|
if err != nil {
|
||
|
|
return "", err
|
||
|
|
}
|
||
|
|
wanted := bundle.name
|
||
|
|
annotations := template.GetAnnotations()
|
||
|
|
if annotations == nil {
|
||
|
|
annotations = make(map[string]string)
|
||
|
|
}
|
||
|
|
now := c.now().UTC()
|
||
|
|
if annotations[candidateKey] != wanted {
|
||
|
|
annotations[candidateKey] = wanted
|
||
|
|
annotations[observedKey] = now.Format(time.RFC3339Nano)
|
||
|
|
template.SetAnnotations(annotations)
|
||
|
|
template, err = resource.Update(ctx, template, metav1.UpdateOptions{})
|
||
|
|
if err != nil {
|
||
|
|
return "", err
|
||
|
|
}
|
||
|
|
}
|
||
|
|
since, err := time.Parse(time.RFC3339Nano, annotations[observedKey])
|
||
|
|
if err != nil {
|
||
|
|
return "", fmt.Errorf("template observation: %w", err)
|
||
|
|
}
|
||
|
|
if now.Before(since.Add(c.config.templateQuiet)) {
|
||
|
|
return "", templatePending
|
||
|
|
}
|
||
|
|
if annotations[publishedKey] == wanted {
|
||
|
|
return wanted, nil
|
||
|
|
}
|
||
|
|
active, err := c.templateActive(ctx, namespace, name)
|
||
|
|
if err != nil {
|
||
|
|
return "", err
|
||
|
|
}
|
||
|
|
if active {
|
||
|
|
return "", templatePending
|
||
|
|
}
|
||
|
|
snapshot := &unstructured.Unstructured{Object: map[string]any{
|
||
|
|
"apiVersion": "argoproj.io/v1alpha1", "kind": "WorkflowTemplate", "spec": bundle.spec,
|
||
|
|
}}
|
||
|
|
snapshot.SetName(wanted)
|
||
|
|
snapshot.SetNamespace(namespace)
|
||
|
|
snapshot.SetLabels(template.GetLabels())
|
||
|
|
snapshot.SetAnnotations(map[string]string{templateKey: name})
|
||
|
|
created, err := resource.Create(ctx, snapshot, metav1.CreateOptions{})
|
||
|
|
if apierrors.IsAlreadyExists(err) {
|
||
|
|
created, err = resource.Get(ctx, wanted, metav1.GetOptions{})
|
||
|
|
}
|
||
|
|
if err != nil {
|
||
|
|
return "", err
|
||
|
|
}
|
||
|
|
if !reflect.DeepEqual(created.Object["spec"], bundle.spec) || created.GetAnnotations()[templateKey] != name {
|
||
|
|
return "", fmt.Errorf("template snapshot %q has different content", wanted)
|
||
|
|
}
|
||
|
|
for _, config := range bundle.configs {
|
||
|
|
config.SetOwnerReferences([]metav1.OwnerReference{{
|
||
|
|
APIVersion: "argoproj.io/v1alpha1", Kind: "WorkflowTemplate", Name: wanted, UID: created.GetUID(),
|
||
|
|
}})
|
||
|
|
_, err := c.client.Resource(configMapGVR).Namespace(namespace).Create(ctx, config, metav1.CreateOptions{})
|
||
|
|
if apierrors.IsAlreadyExists(err) {
|
||
|
|
existing, getErr := c.client.Resource(configMapGVR).Namespace(namespace).Get(ctx, config.GetName(), metav1.GetOptions{})
|
||
|
|
if getErr != nil {
|
||
|
|
return "", getErr
|
||
|
|
}
|
||
|
|
if existing.Object["immutable"] != true || !reflect.DeepEqual(existing.Object["data"], config.Object["data"]) || !reflect.DeepEqual(existing.Object["binaryData"], config.Object["binaryData"]) {
|
||
|
|
return "", fmt.Errorf("action snapshot %q has different content", config.GetName())
|
||
|
|
}
|
||
|
|
} else if err != nil {
|
||
|
|
return "", err
|
||
|
|
}
|
||
|
|
}
|
||
|
|
// Resource-version comparison prevents publishing over a newer desired template.
|
||
|
|
annotations[publishedKey] = wanted
|
||
|
|
template.SetAnnotations(annotations)
|
||
|
|
if _, err := resource.Update(ctx, template, metav1.UpdateOptions{}); err != nil {
|
||
|
|
return "", err
|
||
|
|
}
|
||
|
|
return wanted, nil
|
||
|
|
}
|
||
|
|
|
||
|
|
func (c *controller) templateActive(ctx context.Context, namespace, name string) (bool, error) {
|
||
|
|
workflows, err := c.client.Resource(workflowGVR).Namespace(namespace).List(ctx, metav1.ListOptions{})
|
||
|
|
if err != nil {
|
||
|
|
return false, err
|
||
|
|
}
|
||
|
|
for _, workflow := range workflows.Items {
|
||
|
|
phase, _, err := unstructured.NestedString(workflow.Object, "status", "phase")
|
||
|
|
if err != nil {
|
||
|
|
return false, err
|
||
|
|
}
|
||
|
|
if terminalPhase(phase) {
|
||
|
|
continue
|
||
|
|
}
|
||
|
|
ref, _, err := unstructured.NestedString(workflow.Object, "spec", "workflowTemplateRef", "name")
|
||
|
|
if err != nil {
|
||
|
|
return false, err
|
||
|
|
}
|
||
|
|
if ref == name || workflow.GetAnnotations()[templateKey] == name {
|
||
|
|
return true, nil
|
||
|
|
}
|
||
|
|
if ref == "" {
|
||
|
|
continue
|
||
|
|
}
|
||
|
|
bound, err := c.client.Resource(templateGVR).Namespace(namespace).Get(ctx, ref, metav1.GetOptions{})
|
||
|
|
if apierrors.IsNotFound(err) {
|
||
|
|
continue
|
||
|
|
}
|
||
|
|
if err != nil {
|
||
|
|
return false, err
|
||
|
|
}
|
||
|
|
if bound.GetAnnotations()[templateKey] == name {
|
||
|
|
return true, nil
|
||
|
|
}
|
||
|
|
}
|
||
|
|
return false, nil
|
||
|
|
}
|
||
|
|
|
||
|
|
func (c *controller) captureBundle(ctx context.Context, template *unstructured.Unstructured) (templateBundle, error) {
|
||
|
|
spec, _, err := unstructured.NestedMap(template.Object, "spec")
|
||
|
|
if err != nil {
|
||
|
|
return templateBundle{}, err
|
||
|
|
}
|
||
|
|
patches, err := expandPatches(spec)
|
||
|
|
if err != nil {
|
||
|
|
return templateBundle{}, err
|
||
|
|
}
|
||
|
|
refs := make(map[string][]map[string]any)
|
||
|
|
configRefs(spec, refs)
|
||
|
|
names := make([]string, 0, len(refs))
|
||
|
|
for name := range refs {
|
||
|
|
names = append(names, name)
|
||
|
|
}
|
||
|
|
slices.Sort(names)
|
||
|
|
bundle := templateBundle{spec: spec}
|
||
|
|
for _, name := range names {
|
||
|
|
config, err := c.client.Resource(configMapGVR).Namespace(template.GetNamespace()).Get(ctx, name, metav1.GetOptions{})
|
||
|
|
if err != nil {
|
||
|
|
return templateBundle{}, err
|
||
|
|
}
|
||
|
|
if config.GetLabels()["app.kubernetes.io/name"] != "project-delivery" {
|
||
|
|
continue
|
||
|
|
}
|
||
|
|
body := map[string]any{}
|
||
|
|
for _, field := range []string{"data", "binaryData"} {
|
||
|
|
if value, found := config.Object[field]; found {
|
||
|
|
body[field] = value
|
||
|
|
}
|
||
|
|
}
|
||
|
|
snapshot := &unstructured.Unstructured{Object: body}
|
||
|
|
snapshot.SetAPIVersion("v1")
|
||
|
|
snapshot.SetKind("ConfigMap")
|
||
|
|
snapshot.SetNamespace(template.GetNamespace())
|
||
|
|
snapshot.SetName(name)
|
||
|
|
snapshot.SetLabels(config.GetLabels())
|
||
|
|
snapshot.Object["immutable"] = true
|
||
|
|
bundle.configs = append(bundle.configs, snapshot)
|
||
|
|
}
|
||
|
|
data, err := json.Marshal(map[string]any{"name": template.GetName(), "spec": spec, "configs": bundle.configs})
|
||
|
|
if err != nil {
|
||
|
|
return templateBundle{}, err
|
||
|
|
}
|
||
|
|
bundle.name = contentName("tpl-", data)
|
||
|
|
for _, config := range bundle.configs {
|
||
|
|
name := config.GetName()
|
||
|
|
config.SetName(contentName("act-", []byte(bundle.name+"/"+name)))
|
||
|
|
for _, ref := range refs[name] {
|
||
|
|
ref["name"] = config.GetName()
|
||
|
|
}
|
||
|
|
}
|
||
|
|
for _, patch := range patches {
|
||
|
|
data, err := json.Marshal(patch.body)
|
||
|
|
if err != nil {
|
||
|
|
return templateBundle{}, err
|
||
|
|
}
|
||
|
|
patch.parent["podSpecPatch"] = string(data)
|
||
|
|
}
|
||
|
|
return bundle, nil
|
||
|
|
}
|
||
|
|
|
||
|
|
func expandPatches(value any) ([]podPatch, error) {
|
||
|
|
var patches []podPatch
|
||
|
|
switch value := value.(type) {
|
||
|
|
case map[string]any:
|
||
|
|
if text, ok := value["podSpecPatch"].(string); ok {
|
||
|
|
data, err := yaml.YAMLToJSON([]byte(text))
|
||
|
|
if err != nil {
|
||
|
|
return nil, fmt.Errorf("parse podSpecPatch: %w", err)
|
||
|
|
}
|
||
|
|
var body any
|
||
|
|
if err := json.Unmarshal(data, &body); err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
value["podSpecPatch"] = body
|
||
|
|
patches = append(patches, podPatch{parent: value, body: body})
|
||
|
|
}
|
||
|
|
for _, child := range value {
|
||
|
|
nested, err := expandPatches(child)
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
patches = append(patches, nested...)
|
||
|
|
}
|
||
|
|
case []any:
|
||
|
|
for _, child := range value {
|
||
|
|
nested, err := expandPatches(child)
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
patches = append(patches, nested...)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
return patches, nil
|
||
|
|
}
|
||
|
|
|
||
|
|
func configRefs(value any, refs map[string][]map[string]any) {
|
||
|
|
switch value := value.(type) {
|
||
|
|
case map[string]any:
|
||
|
|
for key, child := range value {
|
||
|
|
if key == "configMap" || key == "configMapRef" || key == "configMapKeyRef" {
|
||
|
|
if ref, ok := child.(map[string]any); ok {
|
||
|
|
if name, ok := ref["name"].(string); ok {
|
||
|
|
refs[name] = append(refs[name], ref)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
configRefs(child, refs)
|
||
|
|
}
|
||
|
|
case []any:
|
||
|
|
for _, child := range value {
|
||
|
|
configRefs(child, refs)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
func contentName(prefix string, data []byte) string {
|
||
|
|
digest := sha256.Sum256(data)
|
||
|
|
return prefix + base32.NewEncoding("abcdefghijklmnopqrstuvwxyz234567").WithPadding(base32.NoPadding).EncodeToString(digest[:])[:16]
|
||
|
|
}
|