johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package kube

import (
	"context"
	"errors"
	"fmt"
	"sort"
	"sync"
	"time"

	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/supervise"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/fields"
	kruntime "k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/apimachinery/pkg/watch"
	"k8s.io/client-go/tools/cache"
	watchtools "k8s.io/client-go/tools/watch"
)

var ErrAwaitTimeout = errors.New("kube: the condition did not hold in time")

func Absent[T any](obj *T) (bool, error) { return obj == nil, nil }

func Present[T any](obj *T) (bool, error) { return obj != nil, nil }

func (r Resource) ListWatch(opts metav1.ListOptions) cache.ListerWatcher {
	return &cache.ListWatch{
		ListWithContextFunc: func(ctx context.Context, o metav1.ListOptions) (kruntime.Object, error) {
			ri, err := r.iface()
			if err != nil {
				return nil, err
			}
			o.LabelSelector, o.FieldSelector = opts.LabelSelector, opts.FieldSelector
			return ri.List(ctx, o)
		},
		WatchFuncWithContext: func(ctx context.Context, o metav1.ListOptions) (watch.Interface, error) {
			ri, err := r.iface()
			if err != nil {
				return nil, err
			}
			o.LabelSelector, o.FieldSelector = opts.LabelSelector, opts.FieldSelector
			return ri.Watch(ctx, o)
		},
	}
}

func expired(parent context.Context, err error, what string, timeout time.Duration) error {
	if err != nil && (wait.Interrupted(err) || errors.Is(err, watchtools.ErrWatchClosed)) {
		if parent.Err() != nil {
			return parent.Err()
		}
		return fmt.Errorf("%w: %s within %s", ErrAwaitTimeout, what, timeout)
	}
	return err
}

func Await[T any](ctx context.Context, r Resource, name string, timeout time.Duration, done func(obj *T) (bool, error)) (*T, error) {
	if _, err := r.iface(); err != nil {
		return nil, err
	}
	if done == nil {
		return nil, errors.New("kube: await needs a condition")
	}
	parent := ctx
	if timeout > 0 {
		var cancel context.CancelFunc
		ctx, cancel = context.WithTimeout(ctx, timeout)
		defer cancel()
	}
	lw := r.ListWatch(metav1.ListOptions{FieldSelector: fields.OneTermEqualSelector("metadata.name", name).String()})
	var mu sync.Mutex
	var last *T
	check := func(u *unstructured.Unstructured) (bool, error) {
		if u == nil {
			mu.Lock()
			last = nil
			mu.Unlock()
			return done(nil)
		}
		cur, err := Decode[T](u.Object)
		if err != nil {
			return false, nil
		}
		mu.Lock()
		last = cur
		mu.Unlock()
		return done(cur)
	}
	pre := func(store cache.Store) (bool, error) {
		for _, item := range store.List() {
			if u, ok := item.(*unstructured.Unstructured); ok && u.GetName() == name {
				return check(u)
			}
		}
		return check(nil)
	}
	_, err := watchtools.UntilWithSync(ctx, lw, &unstructured.Unstructured{}, pre, func(ev watch.Event) (bool, error) {
		switch ev.Type {
		case watch.Deleted:
			return check(nil)
		case watch.Added, watch.Modified:
			u, ok := ev.Object.(*unstructured.Unstructured)
			if !ok {
				return false, nil
			}
			return check(u)
		}
		return false, nil
	})
	mu.Lock()
	seen := last
	mu.Unlock()
	return seen, expired(parent, err, name, timeout)
}

func AwaitList[T any](ctx context.Context, r Resource, opts metav1.ListOptions, timeout time.Duration, done func(items []T) (bool, error)) ([]T, error) {
	if _, err := r.iface(); err != nil {
		return nil, err
	}
	if done == nil {
		return nil, errors.New("kube: await needs a condition")
	}
	parent := ctx
	if timeout > 0 {
		var cancel context.CancelFunc
		ctx, cancel = context.WithTimeout(ctx, timeout)
		defer cancel()
	}
	var mu sync.Mutex
	var store cache.Store
	var last []T
	check := func() (bool, error) {
		mu.Lock()
		s := store
		mu.Unlock()
		if s == nil {
			return false, nil
		}
		var objs []*unstructured.Unstructured
		for _, item := range s.List() {
			if u, ok := item.(*unstructured.Unstructured); ok {
				objs = append(objs, u)
			}
		}
		sort.Slice(objs, func(i, j int) bool {
			if objs[i].GetNamespace() != objs[j].GetNamespace() {
				return objs[i].GetNamespace() < objs[j].GetNamespace()
			}
			return objs[i].GetName() < objs[j].GetName()
		})
		items := make([]T, 0, len(objs))
		for _, u := range objs {
			v, err := Decode[T](u.Object)
			if err != nil {
				continue
			}
			items = append(items, *v)
		}
		mu.Lock()
		last = items
		mu.Unlock()
		return done(items)
	}
	pre := func(s cache.Store) (bool, error) {
		mu.Lock()
		store = s
		mu.Unlock()
		return check()
	}
	_, err := watchtools.UntilWithSync(ctx, r.ListWatch(opts), &unstructured.Unstructured{}, pre, func(watch.Event) (bool, error) {
		return check()
	})
	mu.Lock()
	seen := last
	mu.Unlock()
	return seen, expired(parent, err, "the list of "+r.gvr.Resource, timeout)
}

type ListCondition func(items []any) (bool, error)

func AwaitListWatch(ctx context.Context, lw cache.ListerWatcher, objType kruntime.Object, timeout time.Duration, done ListCondition) error {
	if lw == nil {
		return errors.New("kube: await needs a list watch")
	}
	if done == nil {
		return errors.New("kube: await needs a condition")
	}
	parent := ctx
	if timeout > 0 {
		var cancel context.CancelFunc
		ctx, cancel = context.WithTimeout(ctx, timeout)
		defer cancel()
	}
	informer := supervise.New(ctx)
	defer informer.Stop()
	ctx = informer.Context()
	changed := make(chan struct{}, 1)
	kick := func() {
		select {
		case changed <- struct{}{}:
		default:
		}
	}
	store, ctrl := cache.NewInformerWithOptions(cache.InformerOptions{ListerWatcher: lw, ObjectType: objType,
		Handler: cache.ResourceEventHandlerFuncs{
			AddFunc:    func(any) { kick() },
			UpdateFunc: func(any, any) { kick() },
			DeleteFunc: func(any) { kick() },
		}})
	informer.Go("kube: await informer", func(ctx context.Context) error {
		ctrl.RunWithContext(ctx)
		return nil
	})
	timedOut := func() error {
		if parent.Err() != nil {
			return parent.Err()
		}
		return fmt.Errorf("%w within %s", ErrAwaitTimeout, timeout)
	}
	if !cache.WaitForCacheSync(ctx.Done(), ctrl.HasSynced) {
		return timedOut()
	}
	for {
		ok, err := done(store.List())
		if err != nil {
			return err
		}
		if ok {
			return nil
		}
		select {
		case <-ctx.Done():
			return timedOut()
		case <-changed:
		}
	}
}

func AwaitSelected[T any](ctx context.Context, r Resource, labelSelector string, timeout time.Duration, done func(items []T) (bool, error)) error {
	if _, err := r.iface(); err != nil {
		return err
	}
	return AwaitListWatch(ctx, r.ListWatch(metav1.ListOptions{LabelSelector: labelSelector}), &unstructured.Unstructured{}, timeout, func(items []any) (bool, error) {
		out := make([]T, 0, len(items))
		for _, item := range items {
			u, ok := item.(*unstructured.Unstructured)
			if !ok {
				continue
			}
			v, err := Decode[T](u.Object)
			if err != nil {
				continue
			}
			out = append(out, *v)
		}
		return done(out)
	})
}

func Follow(ctx context.Context, lw cache.ListerWatcher, objType kruntime.Object, notify func(obj any)) {
	if lw == nil || notify == nil {
		return
	}
	gone := func(obj any) {
		if d, ok := obj.(cache.DeletedFinalStateUnknown); ok {
			obj = d.Obj
		}
		notify(obj)
	}
	_, ctrl := cache.NewInformerWithOptions(cache.InformerOptions{ListerWatcher: lw, ObjectType: objType,
		Handler: cache.ResourceEventHandlerFuncs{
			AddFunc:    notify,
			UpdateFunc: func(_, obj any) { notify(obj) },
			DeleteFunc: gone,
		}})
	ctrl.RunWithContext(ctx)
}

func FollowResource[T any](ctx context.Context, r Resource, opts metav1.ListOptions, notify func(obj *T, deleted bool)) {
	if notify == nil {
		return
	}
	if _, err := r.iface(); err != nil {
		return
	}
	emit := func(obj any, deleted bool) {
		if d, ok := obj.(cache.DeletedFinalStateUnknown); ok {
			obj = d.Obj
		}
		u, ok := obj.(*unstructured.Unstructured)
		if !ok {
			return
		}
		v, err := Decode[T](u.Object)
		if err != nil {
			return
		}
		notify(v, deleted)
	}
	_, ctrl := cache.NewInformerWithOptions(cache.InformerOptions{ListerWatcher: r.ListWatch(opts), ObjectType: &unstructured.Unstructured{},
		Handler: cache.ResourceEventHandlerFuncs{
			AddFunc:    func(obj any) { emit(obj, false) },
			UpdateFunc: func(_, obj any) { emit(obj, false) },
			DeleteFunc: func(obj any) { emit(obj, true) },
		}})
	ctrl.RunWithContext(ctx)
}